Merge branch 'hy-dev' into dev

core/scraper_twitter.py

@@ -28,7 +28,14 @@ from core.models import ScrapedItem, ScrapeResult
 from core.scraper import ScrapeCancelled
 
 # 命中这些 GraphQL operation 的响应才解析
-_GRAPHQL_HINTS = ("UserTweets", "UserTweetsAndReplies", "UserMedia")
+_GRAPHQL_HINTS = (
+    "UserTweets",
+    "UserTweetsAndReplies",
+    "UserMedia",
+    # 单条推文/线程
+    "TweetDetail",
+    "TweetResultByRestId",
+)
 _TWITTER_HOSTS = ("x.com", "twitter.com", "mobile.twitter.com")
 # Twitter 的 created_at 形如 "Wed Oct 10 20:19:24 +0000 2018"
 _TW_DATE_FMT = "%a %b %d %H:%M:%S %z %Y"
@@ -50,29 +57,56 @@ def is_twitter_url(url: str) -> bool:
     return any(host == h or host.endswith("." + h) for h in _TWITTER_HOSTS)
 
 
-def extract_handle(url: str) -> Optional[str]:
-    """https://x.com/elonmusk?foo=1 -> 'elonmusk';带 /status/ 等子路径或保留路径则返回 None。"""
+def parse_x_url(url: str) -> tuple[Optional[str], Optional[str]]:
+    """解析 X / Twitter URL,返回 (handle, tweet_id)。
+
+    - https://x.com/elonmusk            → ('elonmusk', None)        用户主页
+    - https://x.com/elonmusk/with_replies → ('elonmusk', None)
+    - https://x.com/elonmusk/status/123 → ('elonmusk', '123')      单条推文
+    - https://x.com/i/web/status/123    → (None, '123')             单条推文(无 handle)
+    - 其他 / 非 twitter URL              → (None, None)
+    """
     try:
-        path = urlparse(url).path.strip("/")
+        parsed = urlparse(url)
+        host = (parsed.hostname or "").lower()
+        path = parsed.path.strip("/")
     except Exception:
-        return None
+        return None, None
+    if not host or not any(host == h or host.endswith("." + h) for h in _TWITTER_HOSTS):
+        return None, None
     if not path:
-        return None
-    parts = path.split("/")
-    handle = parts[0]
-    # 保留路径 / 非用户页
-    if handle.lower() in {
+        return None, None
+    parts = [p for p in path.split("/") if p]
+    # /i/web/status/{id}
+    if len(parts) >= 4 and parts[0].lower() == "i" and parts[2].lower() == "status":
+        if re.fullmatch(r"\d{5,30}", parts[3]):
+            return None, parts[3]
+    # 保留路径
+    if parts[0].lower() in {
         "i", "home", "search", "explore", "notifications", "messages", "settings",
         "hashtag", "compose", "intent", "share", "login", "signup", "tos", "privacy",
     }:
-        return None
-    if len(parts) > 1:
-        # 只接受 /{handle} 或 /{handle}/with_replies
-        if parts[1].lower() not in {"with_replies", "media", "highlights", ""}:
-            return None
+        return None, None
+    handle = parts[0]
     if not re.fullmatch(r"[A-Za-z0-9_]{1,15}", handle):
-        return None
-    return handle
+        return None, None
+    # /{handle}/status/{id}
+    if len(parts) >= 3 and parts[1].lower() == "status" and re.fullmatch(r"\d{5,30}", parts[2]):
+        return handle, parts[2]
+    # /{handle} 或 /{handle}/with_replies|media|highlights
+    if len(parts) == 1 or parts[1].lower() in {"with_replies", "media", "highlights", ""}:
+        return handle, None
+    return None, None
+
+
+def extract_handle(url: str) -> Optional[str]:
+    """向后兼容:仅返回用户主页 URL 的 handle。"""
+    handle, tweet_id = parse_x_url(url)
+    return handle if (handle and not tweet_id) else None
+
+
+def extract_tweet_id(url: str) -> Optional[str]:
+    return parse_x_url(url)[1]
 
 
 def _parse_cookies(raw: Any) -> list[dict]:
@@ -234,6 +268,8 @@ class XTwitterScraper:
         self._oldest_seen: Optional[datetime] = None
         self._bodies: list[str] = []
         self._handle: str = ""
+        # 单条推文模式:只保留 target_tweet_id,跳过转推/回复/纯文本过滤
+        self._target_tweet_id: Optional[str] = None
 
     def _check_cancel(self) -> None:
         if self.is_cancelled():
@@ -241,10 +277,22 @@ class XTwitterScraper:
 
     async def run(self, series_url: str, start_page: int = 1, end_page: int = 1) -> ScrapeResult:
         result = ScrapeResult()
-        handle = extract_handle(series_url)
-        if not handle:
-            raise ValueError(f"无法从链接解析出 Twitter 用户名(应形如 https://x.com/<用户名>): {series_url}")
-        self._handle = handle
+        handle, tweet_id = parse_x_url(series_url)
+        if not handle and not tweet_id:
+            raise ValueError(
+                f"无法识别 X/Twitter 链接(应形如 https://x.com/<用户名> 或 https://x.com/<用户名>/status/<id>): {series_url}"
+            )
+        self._handle = handle or ""
+        self._target_tweet_id = tweet_id
+
+        # 先清掉 profile 里残留的 x.com / twitter.com cookie,再注入这次的(避免账号串扰)
+        try:
+            await self.context.clear_cookies(domain=".x.com")
+            await self.context.clear_cookies(domain=".twitter.com")
+            await self.context.clear_cookies(domain="x.com")
+            await self.context.clear_cookies(domain="twitter.com")
+        except Exception as exc:  # noqa: BLE001
+            self.log(f"清旧 cookie 失败(继续): {exc}")
 
         cookie_objs = _parse_cookies(self.cookies)
         if cookie_objs:
@@ -261,15 +309,29 @@ class XTwitterScraper:
         page.on("response", lambda r: asyncio.create_task(self._on_response(r)))
         deadline = time.time() + self.overall_timeout
         try:
-            target = f"https://x.com/{handle}"
-            self.log(f"打开用户主页: {target}")
-            await page.goto(target, wait_until="domcontentloaded", timeout=45000)
-            # 等首批 GraphQL 响应落地
-            await self._wait_settle(page, 2.0)
-            # 关 cookie 横幅 + 点掉「敏感内容」确认页(否则 timeline 不会加载)
-            await self._dismiss_gates(page)
-
-            await self._scroll_loop(page, deadline)
+            if tweet_id:
+                target = (
+                    f"https://x.com/{handle}/status/{tweet_id}" if handle
+                    else f"https://x.com/i/web/status/{tweet_id}"
+                )
+                self.log(f"打开单条推文: {target}")
+                await page.goto(target, wait_until="domcontentloaded", timeout=45000)
+                await self._wait_settle(page, 2.0)
+                await self._dismiss_gates(page)
+                # 单条推文不需要滚动,多等几秒让 TweetDetail 响应到位
+                for _ in range(6):
+                    self._check_cancel()
+                    if any("TweetDetail" in b or "TweetResultByRestId" in b
+                           or '"tweet_results"' in b for b in self._bodies):
+                        break
+                    await asyncio.sleep(1.0)
+            else:
+                target = f"https://x.com/{handle}"
+                self.log(f"打开用户主页: {target}")
+                await page.goto(target, wait_until="domcontentloaded", timeout=45000)
+                await self._wait_settle(page, 2.0)
+                await self._dismiss_gates(page)
+                await self._scroll_loop(page, deadline)
         except ScrapeCancelled:
             raise
         except Exception as exc:  # noqa: BLE001
@@ -293,16 +355,21 @@ class XTwitterScraper:
         items.sort(key=lambda it: it.extra.get("tweet_created_at") or "", reverse=True)
         if len(items) > self.max_items:
             items = items[: self.max_items]
-        for i, it in enumerate(items, 1):
+        for it in items:
             it.page_no = 1
         result.resources = items
         result.pages_scanned = 1
         if not items and not result.item_errors:
-            result.item_errors.append({
-                "source_url": series_url,
-                "reason": "未抓到任何原创推文(可能是 cookie 失效 / 账号受保护 / 该账号无原创推文)",
-            })
-        self.log(f"抓取完成: 原创推文 {len(items)} 条(已处理推文 {len(self._seen)} 条)")
+            reason = (
+                "未抓到该推文(链接是否正确?账号是否受保护?cookie 是否有效?)"
+                if tweet_id
+                else "未抓到任何原创推文(可能是 cookie 失效 / 账号受保护 / 该账号无原创推文)"
+            )
+            result.item_errors.append({"source_url": series_url, "reason": reason})
+        if tweet_id:
+            self.log(f"抓取完成: 单条推文 {len(items)} 条(已处理 {len(self._seen)} 条)")
+        else:
+            self.log(f"抓取完成: 原创推文 {len(items)} 条(已处理推文 {len(self._seen)} 条)")
         return result
 
     # ------------------------------------------------------------------
@@ -424,14 +491,33 @@ class XTwitterScraper:
                 continue
 
     def _iter_entries(self, data: Any):
-        # data.user.result.timeline_v2.timeline.instructions 或 .timeline.timeline.instructions
+        # 候选根:UserTweets/UserTweetsAndReplies → data.user.result.timeline_v2|timeline
+        #         TweetDetail → data.threaded_conversation_with_injections_v2
+        roots: list[dict] = []
         user_result = _safe(data, "data", "user", "result")
-        if not isinstance(user_result, dict):
-            return
-        for tkey in ("timeline_v2", "timeline"):
-            tl = user_result.get(tkey)
-            if not isinstance(tl, dict):
-                continue
+        if isinstance(user_result, dict):
+            for tkey in ("timeline_v2", "timeline"):
+                tl = user_result.get(tkey)
+                if isinstance(tl, dict):
+                    roots.append(tl)
+        tc = _safe(data, "data", "threaded_conversation_with_injections_v2")
+        if isinstance(tc, dict):
+            roots.append(tc)
+        # TweetResultByRestId:data.tweetResult.result(单个 tweet,不带 instructions)
+        direct = _safe(data, "data", "tweetResult", "result")
+        if isinstance(direct, dict):
+            yield {
+                "entryId": f"tweet-direct-{direct.get('rest_id', '?')}",
+                "content": {
+                    "entryType": "TimelineTimelineItem",
+                    "itemContent": {
+                        "itemType": "TimelineTweet",
+                        "tweet_results": {"result": direct},
+                    },
+                },
+            }
+
+        for tl in roots:
             inner = tl.get("timeline") if isinstance(tl.get("timeline"), dict) else tl
             instructions = inner.get("instructions") if isinstance(inner, dict) else None
             if not isinstance(instructions, list):
@@ -480,6 +566,10 @@ class XTwitterScraper:
         if not isinstance(legacy, dict):
             return
 
+        # 单条推文模式:只要 target_tweet_id 那一条,其他对话/上下文跳过
+        if self._target_tweet_id and tweet_id != self._target_tweet_id:
+            return
+
         created = _parse_created_at(legacy)
         if created is not None:
             if self._oldest_seen is None or created < self._oldest_seen:
@@ -492,16 +582,17 @@ class XTwitterScraper:
             or self._handle
         )
 
-        # —— 过滤:转推 / 他人回复 ——
-        if legacy.get("retweeted_status_result"):
-            return
-        in_reply_to = legacy.get("in_reply_to_status_id_str")
-        in_reply_user = legacy.get("in_reply_to_screen_name")
-        if in_reply_to and in_reply_user and str(in_reply_user).lower() != str(author).lower():
-            return  # 回复别人,跳过(自串保留)
+        # —— 过滤:转推 / 他人回复 ——(单条推文模式不过滤,用户要的就是那一条)
+        if not self._target_tweet_id:
+            if legacy.get("retweeted_status_result"):
+                return
+            in_reply_to = legacy.get("in_reply_to_status_id_str")
+            in_reply_user = legacy.get("in_reply_to_screen_name")
+            if in_reply_to and in_reply_user and str(in_reply_user).lower() != str(author).lower():
+                return  # 回复别人,跳过(自串保留)
 
-        # 时间窗过滤
-        if created is not None:
+        # 时间窗过滤(单条推文模式也不过滤)
+        if created is not None and not self._target_tweet_id:
             if self.since and created < self.since:
                 return
             if self.until and created > self.until:
@@ -544,8 +635,11 @@ class XTwitterScraper:
             media_kind = "video"
         elif images:
             media_kind = "photo"
+        elif self._target_tweet_id:
+            # 单条推文模式:用户明确指定了 id,即使纯文本也返回
+            media_kind = "text"
         else:
-            # 纯文本(无图无视频)推文不要,主站只要带媒体的
+            # 用户主页模式:纯文本推文不要,主站只要带媒体的
             return
 
         title = ""

server/jobs.py

@@ -70,7 +70,8 @@ class Job:
             "error": self.error,
             "created_at": self.created_at,
             "finished_at": self.finished_at,
-            "log_tail": self.log_lines[-80:],
+            # 大量采集任务每条 tweet 占 2 行日志,80 不够看到滚动/敏感页处理那段;给到 500
+            "log_tail": self.log_lines[-500:],
         }
 
 
@@ -85,8 +86,8 @@ def _log(job: Job, msg: str) -> None:
     # 注意:不要把 secret_key 等敏感值传进来
     line = f"{time.strftime('%H:%M:%S')} {msg}"
     job.log_lines.append(line)
-    if len(job.log_lines) > 800:
-        del job.log_lines[: len(job.log_lines) - 800]
+    if len(job.log_lines) > 2000:
+        del job.log_lines[: len(job.log_lines) - 2000]
     logger.info("[job %s] %s", job.id, msg)