import asyncio import re import hashlib from typing import Tuple, List, Optional from playwright.async_api import async_playwright, Browser, BrowserContext, Page from .models import PostSegment, MediaItem, RippedThread from .stitcher import build_markdown, build_plain_text, build_standalone_html class ThreadScraper: def __init__(self, headless: bool = True): self.headless = headless self._browser: Optional[Browser] = None self._playwright = None async def get_browser(self) -> Browser: if self._browser is None or not self._browser.is_connected(): if self._playwright is None: self._playwright = await async_playwright().start() self._browser = await self._playwright.chromium.launch( headless=self.headless, args=[ "--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage", "--disable-gpu", ] ) return self._browser async def close(self): if self._browser: await self._browser.close() self._browser = None if self._playwright: await self._playwright.stop() self._playwright = None async def rip(self, url: str, mode: str = "author_only", max_scrolls: int = 5) -> RippedThread: """Scrapes and rips a Threads post and its chained continuation posts.""" browser = await self.get_browser() context: BrowserContext = await browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", viewport={"width": 1280, "height": 900} ) page: Page = await context.new_page() try: # Navigate and follow redirects await page.goto(url, wait_until="networkidle", timeout=30000) canonical_url = page.url # Small delay for hydration await page.wait_for_timeout(2500) # Extract Open Graph metadata og_title = await page.evaluate("() => document.querySelector('meta[property=\"og:title\"]')?.content || ''") og_image = await page.evaluate("() => document.querySelector('meta[property=\"og:image\"]')?.content || ''") og_desc = await page.evaluate("() => document.querySelector('meta[property=\"og:description\"]')?.content || ''") # Determine original author handle from URL or title author_handle = "" author_name = "" # Check canonical URL for /@username match_handle = re.search(r"/@([a-zA-Z0-9_\.]+)", canonical_url) if match_handle: author_handle = match_handle.group(1) # Check og:title e.g. "Beth Parent McMillian (@bethmcmilliandesigns) on Threads" if og_title: m_title = re.search(r"^(.*?)\s*\(@?([a-zA-Z0-9_\.]+)\)", og_title) if m_title: author_name = m_title.group(1).strip() if not author_handle: author_handle = m_title.group(2).strip() if not author_name: author_name = author_handle or "Threads Author" # Author profile avatar author_avatar = og_image if og_image else None # Progressively scroll to load dynamic replies in the thread for _ in range(max_scrolls): await page.evaluate("window.scrollBy(0, 1200)") await page.wait_for_timeout(1000) # Query all pressable post containers containers = await page.query_selector_all("div[data-pressable-container='true']") raw_segments: List[PostSegment] = [] seen_texts = set() for idx, c in enumerate(containers): # 1. Identify post author handle user_links = await c.query_selector_all("a[href*='/@']") post_handle = None post_url = None for ul in user_links: href = await ul.get_attribute("href") if href: if "/@" in href and not post_handle: parts = href.split("/@") if len(parts) > 1: post_handle = parts[1].split("/")[0].split("?")[0] if "/post/" in href and not post_url: post_url = href.split("?")[0] # Check for "Author" badge is_author_badge = False badge = await c.query_selector("xpath=.//*[text()='Author']") if badge: is_author_badge = True # Determine if this segment belongs to the original author is_author_post = False if idx == 0: is_author_post = True if post_handle and not author_handle: author_handle = post_handle elif is_author_badge: is_author_post = True elif post_handle and author_handle and (post_handle.lower() == author_handle.lower()): is_author_post = True # If author_only mode is on and this is not by the author, skip if mode == "author_only" and not is_author_post: continue # 2. Extract text with dir='auto' dir_els = await c.query_selector_all("[dir='auto']") segment_text_parts = [] for del_el in dir_els: txt = (await del_el.inner_text()).strip() if not txt: continue # Ignore handle, author badge, bullet dots, stats if txt == post_handle or txt == "Author" or txt == "ยท" or txt == author_handle: continue # Ignore pure stat numbers (e.g., '16.2K', '5d', '2K', '711') if re.match(r"^(\d+(\.\d+)?[KMkmdh]?|\d+)$", txt): continue if txt not in segment_text_parts: segment_text_parts.append(txt) combined_text = "\n\n".join(segment_text_parts).strip() if not combined_text and not is_author_post: continue # Deduplicate identical consecutive segments text_hash = hashlib.md5(combined_text.encode("utf-8")).hexdigest() if text_hash in seen_texts and combined_text: continue seen_texts.add(text_hash) # 3. Extract media images inside post (exclude profile pictures) imgs = await c.query_selector_all("img") post_images: List[MediaItem] = [] for img in imgs: src = await img.get_attribute("src") alt = await img.get_attribute("alt") or "" if src: # Exclude small avatars or generic SVG icons if "profile_pic" in src or "scontent" in src and "206x206" in src: continue post_images.append(MediaItem(type="image", url=src, alt_text=alt)) raw_segments.append( PostSegment( index=len(raw_segments), author_handle=post_handle or author_handle, author_name=author_name if is_author_post else (post_handle or "User"), author_avatar=author_avatar if is_author_post else None, url=post_url, text=combined_text, is_author=is_author_post, images=post_images ) ) # If no segments could be extracted from containers, fallback to OG description if not raw_segments and og_desc: raw_segments.append( PostSegment( index=0, author_handle=author_handle, author_name=author_name, author_avatar=author_avatar, url=canonical_url, text=og_desc, is_author=True, images=[] ) ) # Construct stitched outputs markdown_out = build_markdown(author_name, author_handle, canonical_url, raw_segments) plain_text_out = build_plain_text(author_name, author_handle, canonical_url, raw_segments) html_out = build_standalone_html(author_name, author_handle, author_avatar, canonical_url, raw_segments) # Generate unique ID for this ripped thread thread_id = hashlib.md5(canonical_url.split("?")[0].encode("utf-8")).hexdigest()[:12] title = raw_segments[0].text.split("\n")[0] if raw_segments and raw_segments[0].text else f"Thread by @{author_handle}" if len(title) > 90: title = title[:87] + "..." total_words = sum(len(s.text.split()) for s in raw_segments) return RippedThread( id=thread_id, original_url=url, canonical_url=canonical_url, author_handle=author_handle, author_name=author_name, author_avatar=author_avatar, created_at="Recent", title=title, segments=raw_segments, stitched_markdown=markdown_out, stitched_text=plain_text_out, stitched_html=html_out, segment_count=len(raw_segments), total_words=total_words ) finally: await context.close()