229 lines
9.9 KiB
Python
229 lines
9.9 KiB
Python
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()
|