Merge branch 'hy-dev' into dev

P3 #16 t.co 处理(关键):主站审核 LLM 看到 t.co 短链会判成"联系方式/广告"导致
  Twitter 推文大量被拒(NASA 15 条只入 11 条因为这个)。改:
    - 媒体 t.co(entities.media[].url):直接删
    - 外链 t.co(entities.urls[].url):替换成 expanded_url(真实目的地址),
      LLM 看到 nasa.gov/blog/xxx 就不再误判
    - 正则兜底删任何残留的 https://t.co/xxx
  原始 full_text 存进 extra_data.original_full_text 备查。
P3 #14 yt-dlp 进度心跳:download_video 加 on_progress 回调,jobs.py 传 _log 进去。
  yt-dlp 的 progress_hooks 每跨 10% 或 5 秒打一行(节流),exec_log 里能看到
  「↓ 35% 12.3MB/35.0MB @ 1024KB/s」,大视频不再发冷。
P3 #15 MinIO access_key 全脱敏(从前4位+*** → 全 ***)。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

core/media.py

@@ -54,7 +54,7 @@ class MinioConfig:
         return bool(self.endpoint and self.access_key and self.secret_key and self.bucket)
 
     def safe_repr(self) -> str:
-        return f"MinioConfig(endpoint={self.endpoint!r}, bucket={self.bucket!r}, access_key={self.access_key[:4]+'***' if self.access_key else ''!r}, secret_key=***)"
+        return f"MinioConfig(endpoint={self.endpoint!r}, bucket={self.bucket!r}, access_key=***, secret_key=***)"
 
 
 def _endpoint_host_secure(endpoint: str) -> tuple[str, bool]:
@@ -147,15 +147,49 @@ def head_ok(url: str, referer: str = "", timeout: int = 15) -> bool:
 
 
 def download_video(media_url: str, referer: str, dest_dir: Path, video_id: str,
-                   concurrent_fragments: int = 8, max_mb: int = 0) -> Path:
+                   concurrent_fragments: int = 8, max_mb: int = 0,
+                   on_progress=None) -> Path:
     """用 yt-dlp 下载 m3u8/mp4 → dest_dir/<video_id>.mp4(尽量)。返回最终文件路径。
 
     max_mb > 0 时,下完发现超过则删文件并抛 ValueError("video too large: ...")。
+
+    on_progress: 可选回调 callable(msg:str),用来把 yt-dlp 的下载心跳回写到调用方
+    的日志里(否则大视频会有几分钟外面看起来"卡死"的体验)。节流:每跨过 10% 或
+    5 秒打一行,二者满足其一即可。
     """
+    import time as _t
     import yt_dlp
 
     dest_dir.mkdir(parents=True, exist_ok=True)
     outtmpl = str(dest_dir / f"{video_id}.%(ext)s")
+
+    _state = {"last_pct": -10, "last_t": 0.0}
+    def _hook(d):
+        if not on_progress:
+            return
+        try:
+            status = d.get("status")
+            if status == "downloading":
+                total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
+                done = d.get("downloaded_bytes") or 0
+                speed = d.get("speed") or 0
+                now = _t.time()
+                pct = int(done * 100 / total) if total else -1
+                # 每跨过 10% 或 5 秒报一次
+                if (pct >= _state["last_pct"] + 10) or (now - _state["last_t"] >= 5):
+                    _state["last_pct"] = pct if pct >= 0 else _state["last_pct"]
+                    _state["last_t"] = now
+                    pct_s = f"{pct}%" if pct >= 0 else "?"
+                    on_progress(
+                        f"  ↓ {pct_s} "
+                        f"{(done/1024/1024):.1f}MB/{(total/1024/1024):.1f}MB"
+                        f" @ {(speed/1024):.0f}KB/s"
+                    )
+            elif status == "finished":
+                on_progress(f"  ↓ 下载完成,准备 mux/写盘 …")
+        except Exception:  # noqa: BLE001
+            pass
+
     opts = {
         "outtmpl": outtmpl,
         "format": "best",
@@ -168,6 +202,7 @@ def download_video(media_url: str, referer: str, dest_dir: Path, video_id: str,
         "retries": 3,
         "fragment_retries": 3,
         "noplaylist": True,
+        "progress_hooks": [_hook],
     }
     with yt_dlp.YoutubeDL(opts) as ydl:
         ydl.download([media_url])

core/models.py

@@ -16,8 +16,10 @@ class ScrapedItem:
     title: str
     # 来源站点名,默认 "xchina"(可被 invoke 载荷里的 platform_name 覆盖)
     source_name: str = "xchina"
-    # 资源类型: "video"(视频) | "photo"(图集)
+    # 资源类型: "video"(视频) | "photo"(图集) | "text"(纯文本,如无媒体的推文)
     media_kind: str = "video"
+    # 采集器标识,写入 extra_data.collector(默认 xchina;twitter scraper 设为 "twitter")
+    collector: str = "xchina"
     # 封面图 URL(上传 MinIO 后改为 MinIO 地址,否则是 xchina 图床地址)
     cover_image: str | None = None
     # 视频 URL(media_kind=video 时;上传 MinIO 后改为 MinIO 地址,否则是 xchina 限时签名 m3u8)
@@ -41,7 +43,7 @@ class ScrapedItem:
         extra_data: dict[str, Any] = {
             "referer": self.source_url,
             "page_no": self.page_no,
-            "collector": "xchina",
+            "collector": self.collector or "xchina",
         }
         if self.video_id:
             extra_data["video_id"] = self.video_id
@@ -67,16 +69,31 @@ class ScrapedItem:
                 "video_url": None,
                 "extra_data": extra_data,
             }
+        if self.media_kind == "video":
+            return {
+                "source_type": "video",
+                "source_name": self.source_name,
+                "source_url": self.source_url,
+                "content_type": "video",
+                "title": self.title,
+                "cover_image": self.cover_image,
+                "summary": (self.summary or None),
+                "content": content,
+                "video_url": self.video_url,
+                "extra_data": extra_data,
+            }
+        # 纯文本(无媒体):SourceTypeEnum=article, ContentTypeEnum=article
         return {
-            "source_type": "video",
+            "source_type": "article",
             "source_name": self.source_name,
             "source_url": self.source_url,
-            "content_type": "video",
+            "content_type": "article",
             "title": self.title,
             "cover_image": self.cover_image,
             "summary": (self.summary or None),
             "content": content,
-            "video_url": self.video_url,
+            "images": list(self.images) if self.images else None,
+            "video_url": None,
             "extra_data": extra_data,
         }
 

core/scraper_twitter.py

@@ -0,0 +1,718 @@
+"""Twitter / X 用户主页采集 —— 抓某个账号的原创推文列表。
+
+输入是一个用户主页链接(https://x.com/{handle} 或 https://twitter.com/{handle}),
+输出每条原创推文一个 ScrapedItem(含文本 + 图片/视频地址,媒体由 jobs.py 统一下载转存 MinIO)。
+
+实现思路:不做 DOM 抓取(timeline 是虚拟列表,滚动即从 DOM 移除),而是监听页面
+发出的 GraphQL XHR 响应(UserTweets / UserTweetsAndReplies),解析其 JSON。滚动到底
+会触发下一批 cursor 分页请求,循环直到收满 max_items / 命中 since / 多次滚动无新增。
+
+登录态:调用方把 cookie 字符串("auth_token=xxx; ct0=yyy")透传进来,导航前注入到
+BrowserContext;Twitter 自家 JS 会用 ct0 自动带上 x-csrf-token,无需手动设。
+
+过滤规则:只保留账号自己发的原创推文 —— 跳过转推(legacy.retweeted_status_result)
+和回复(legacy.in_reply_to_status_id_str 指向他人);引用推文(quote)保留,被引用
+推文的文本拼进 content。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import re
+import time
+from datetime import datetime, timezone
+from typing import Any, Callable, Optional
+from urllib.parse import urlparse
+
+from core.models import ScrapedItem, ScrapeResult
+from core.scraper import ScrapeCancelled
+
+# 命中这些 GraphQL operation 的响应才解析
+_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"
+_MONTHS = {
+    "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6,
+    "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12,
+}
+# "Wed Oct 10 20:19:24 +0000 2018" —— 不依赖 locale 的手写解析
+_TW_DATE_RE = re.compile(
+    r"^[A-Za-z]{3}\s+([A-Za-z]{3})\s+(\d{1,2})\s+(\d{2}):(\d{2}):(\d{2})\s+([+-]\d{4})\s+(\d{4})$"
+)
+
+
+def is_twitter_url(url: str) -> bool:
+    try:
+        host = (urlparse(url).hostname or "").lower()
+    except Exception:
+        return False
+    return any(host == h or host.endswith("." + h) for h in _TWITTER_HOSTS)
+
+
+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:
+        parsed = urlparse(url)
+        host = (parsed.hostname or "").lower()
+        path = parsed.path.strip("/")
+    except Exception:
+        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, 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, None
+    handle = parts[0]
+    if not re.fullmatch(r"[A-Za-z0-9_]{1,15}", 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]:
+    """把 "auth_token=xxx; ct0=yyy" 或 [{name,value},...] 解析成 playwright add_cookies 入参。
+
+    会为 .x.com 和 .twitter.com 各加一份。
+    """
+    pairs: list[tuple[str, str]] = []
+    if isinstance(raw, list):
+        for c in raw:
+            if isinstance(c, dict) and c.get("name"):
+                pairs.append((str(c["name"]).strip(), str(c.get("value", "")).strip()))
+    elif isinstance(raw, dict):
+        for k, v in raw.items():
+            pairs.append((str(k).strip(), str(v).strip()))
+    elif isinstance(raw, str):
+        for seg in raw.replace("\n", ";").split(";"):
+            seg = seg.strip()
+            if not seg or "=" not in seg:
+                continue
+            k, v = seg.split("=", 1)
+            pairs.append((k.strip(), v.strip()))
+    out: list[dict] = []
+    for name, value in pairs:
+        if not name or not value:
+            continue
+        for domain in (".x.com", ".twitter.com"):
+            out.append({"name": name, "value": value, "domain": domain, "path": "/"})
+    return out
+
+
+def _safe(d: Any, *path, default=None):
+    cur = d
+    for k in path:
+        if isinstance(k, int):
+            if not isinstance(cur, list) or k >= len(cur):
+                return default
+            cur = cur[k]
+        else:
+            if not isinstance(cur, dict) or k not in cur:
+                return default
+            cur = cur[k]
+    return cur
+
+
+def _tweet_result_node(node: dict) -> Optional[dict]:
+    """tweet_results.result 可能是 Tweet / TweetWithVisibilityResults(实际推文在 .tweet)。"""
+    if not isinstance(node, dict):
+        return None
+    if node.get("__typename") == "TweetWithVisibilityResults":
+        return node.get("tweet") if isinstance(node.get("tweet"), dict) else None
+    return node
+
+
+def _full_text(result: dict, legacy: dict) -> str:
+    note = _safe(result, "note_tweet", "note_tweet_results", "result", "text")
+    if isinstance(note, str) and note.strip():
+        return note
+    return str(legacy.get("full_text") or "").strip()
+
+
+_TCO_RE = re.compile(r"https?://t\.co/\w+")
+
+
+def _expand_tco(text: str, legacy: dict) -> str:
+    """处理 Twitter 给所有外链/媒体生成的 t.co 短链:
+    - 媒体 t.co(entities.media[].url):直接删掉(图/视频已在 images/video_url 里)
+    - 外链 t.co(entities.urls[].url):替换成 expanded_url(真实目的地址),
+      让主站审核 LLM 看到的是真实域名而不是不透明的 t.co/xxx;否则容易被
+      误判成"联系方式/广告"导致整条被拒。
+    - 兜底用正则把任何残留的 https://t.co/xxx 也删掉。
+    """
+    media = _safe(legacy, "entities", "media") or []
+    for m in media:
+        u = m.get("url")
+        if u:
+            text = text.replace(u, "")
+    urls = _safe(legacy, "entities", "urls") or []
+    for u in urls:
+        short = u.get("url")
+        expanded = u.get("expanded_url") or u.get("display_url") or ""
+        if short:
+            text = text.replace(short, expanded)
+    text = _TCO_RE.sub("", text)
+    return re.sub(r"[ \t]+\n", "\n", text).strip()
+
+
+# 兼容旧名
+_strip_media_tco = _expand_tco
+
+
+def _pick_video_url(media: dict) -> Optional[str]:
+    variants = _safe(media, "video_info", "variants") or []
+    mp4 = [v for v in variants if v.get("content_type") == "video/mp4" and v.get("url")]
+    if not mp4:
+        # animated_gif 的 variant 没有 bitrate
+        any_v = [v for v in variants if v.get("url")]
+        return any_v[0]["url"] if any_v else None
+    mp4.sort(key=lambda v: v.get("bitrate") or 0, reverse=True)
+    return mp4[0]["url"]
+
+
+def _parse_created_at(legacy: dict) -> Optional[datetime]:
+    raw = legacy.get("created_at")
+    if not raw:
+        return None
+    raw = str(raw).strip()
+    m = _TW_DATE_RE.match(raw)
+    if m:
+        mon = _MONTHS.get(m.group(1).lower())
+        if mon:
+            try:
+                day, hh, mm, ss = int(m.group(2)), int(m.group(3)), int(m.group(4)), int(m.group(5))
+                year = int(m.group(7))
+                tzs = m.group(6)
+                offset = int(tzs[1:3]) * 60 + int(tzs[3:5])
+                if tzs[0] == "-":
+                    offset = -offset
+                from datetime import timedelta
+                tz = timezone(timedelta(minutes=offset))
+                return datetime(year, mon, day, hh, mm, ss, tzinfo=tz)
+            except Exception:
+                pass
+    try:
+        return datetime.strptime(raw, _TW_DATE_FMT)
+    except Exception:
+        return None
+
+
+def _date_only(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%d")
+
+
+def _parse_bound(raw: Any) -> Optional[datetime]:
+    if not raw:
+        return None
+    s = str(raw).strip()
+    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S"):
+        try:
+            return datetime.strptime(s, fmt).replace(tzinfo=timezone.utc)
+        except Exception:
+            continue
+    return None
+
+
+class XTwitterScraper:
+    """与 XChinaScraper 同接口:run(url, start_page, end_page) -> ScrapeResult。
+
+    start_page / end_page 对推特无意义,忽略。
+    """
+
+    def __init__(
+        self,
+        context,
+        is_cancelled: Callable[[], bool] = lambda: False,
+        log: Callable[[str], None] = lambda _m: None,
+        cookies: Any = None,
+        since: Any = None,
+        until: Any = None,
+        max_items: int = 100,
+        max_scroll: int = 200,
+        idle_scrolls: int = 6,
+        overall_timeout: float = 1800.0,
+    ) -> None:
+        self.context = context
+        self.is_cancelled = is_cancelled
+        self.log = log
+        self.cookies = cookies
+        self.since = _parse_bound(since)
+        self.until = _parse_bound(until)
+        self.max_items = max(1, int(max_items or 100))
+        self.max_scroll = max_scroll
+        self.idle_scrolls = idle_scrolls
+        self.overall_timeout = overall_timeout
+        # tweet_id -> ScrapedItem(已通过过滤的原创推文)
+        self._items: dict[str, ScrapedItem] = {}
+        # 已处理过的 tweet_id(含被过滤掉的),避免重复解析
+        self._seen: set[str] = set()
+        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():
+            raise ScrapeCancelled()
+
+    async def run(self, series_url: str, start_page: int = 1, end_page: int = 1) -> ScrapeResult:
+        result = ScrapeResult()
+        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:
+            try:
+                await self.context.add_cookies(cookie_objs)
+                names = sorted({c["name"] for c in cookie_objs})
+                self.log(f"已注入 {len(names)} 个 cookie: {', '.join(names)}")
+            except Exception as exc:  # noqa: BLE001
+                self.log(f"注入 cookie 失败(继续,可能拿不到完整 timeline): {exc}")
+        else:
+            self.log("未提供 cookie,将以未登录态访问(timeline 多半为空或受限)")
+
+        page = await self.context.new_page()
+        page.on("response", lambda r: asyncio.create_task(self._on_response(r)))
+        deadline = time.time() + self.overall_timeout
+        try:
+            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
+            self.log(f"采集过程中出错: {type(exc).__name__}: {exc}")
+            result.item_errors.append({"source_url": series_url, "reason": f"{type(exc).__name__}: {exc}"})
+        finally:
+            try:
+                await page.close()
+            except Exception:  # noqa: BLE001
+                pass
+
+        # 解析所有收集到的响应体(响应在事件回调里只暂存原始文本)
+        for body in self._bodies:
+            try:
+                self._consume_body(body)
+            except Exception as exc:  # noqa: BLE001
+                self.log(f"解析 GraphQL 响应失败(跳过): {exc}")
+
+        items = list(self._items.values())
+        # 按时间倒序(新→旧),截断到 max_items
+        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 it in items:
+            it.page_no = 1
+        result.resources = items
+        result.pages_scanned = 1
+        if not items and not result.item_errors:
+            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
+
+    # ------------------------------------------------------------------
+    # 响应收集 / 滚动
+    # ------------------------------------------------------------------
+
+    async def _on_response(self, response) -> None:
+        try:
+            url = response.url
+            if not any(h in url for h in _GRAPHQL_HINTS):
+                return
+            if "graphql" not in url.lower():
+                return
+            body = await response.text()
+            if body and "tweet" in body.lower():
+                self._bodies.append(body)
+        except Exception:  # noqa: BLE001
+            pass
+
+    async def _dismiss_gates(self, page) -> None:
+        """关掉 cookie 同意横幅;点掉「此账号可能含敏感内容 → 是的,查看」确认页。"""
+        # cookie 横幅:点「拒绝非必要」/「Accept all」任一即可消失
+        for sel in (
+            "button:has-text('Refuse non-essential cookies')",
+            "button:has-text('拒绝非必要 Cookie')",
+            "button:has-text('Accept all cookies')",
+            "button:has-text('接受所有 Cookie')",
+        ):
+            try:
+                btn = page.locator(sel).first
+                if await btn.count() and await btn.is_visible():
+                    await btn.click(timeout=3000)
+                    await asyncio.sleep(0.6)
+                    break
+            except Exception:  # noqa: BLE001
+                pass
+        # 敏感内容确认:多试几次(账号 UI 语言可能是英文或中文)
+        clicked = False
+        for _ in range(3):
+            for sel in (
+                "button:has-text('Yes, view profile')",
+                "button:has-text('是的，查看个人资料')",
+                "button:has-text('是的,查看个人资料')",
+                "[data-testid='empty_state_button_text']",
+                "a[href='/settings/content_you_see'] ~ button",
+            ):
+                try:
+                    btn = page.locator(sel).first
+                    if await btn.count() and await btn.is_visible():
+                        await btn.click(timeout=3000)
+                        clicked = True
+                        self.log("已点掉「敏感内容」确认页")
+                        break
+                except Exception:  # noqa: BLE001
+                    pass
+            if clicked:
+                break
+            await asyncio.sleep(1.0)
+        if clicked:
+            await self._wait_settle(page, 3.0)
+
+    async def _wait_settle(self, page, seconds: float) -> None:
+        try:
+            await page.wait_for_load_state("networkidle", timeout=int(seconds * 1000))
+        except Exception:  # noqa: BLE001
+            await asyncio.sleep(seconds)
+
+    async def _scroll_loop(self, page, deadline: float) -> None:
+        idle = 0
+        for n in range(1, self.max_scroll + 1):
+            self._check_cancel()
+            if time.time() > deadline:
+                self.log("达到总时长上限,停止滚动")
+                break
+            # 用「已处理推文数」判 idle(_seen 比 _items 鲁棒:即使所有新推文都被过滤掉
+            # 我们也知道时间线还在出新内容,不会误判到底)
+            before_seen = len(self._seen)
+            try:
+                await page.evaluate("window.scrollBy(0, document.body.scrollHeight)")
+            except Exception:  # noqa: BLE001
+                pass
+            await asyncio.sleep(2.2)
+            # 解析一遍当前已有响应,统计原创推文数 & 最旧时间(用于 since 早停)
+            self._reparse_so_far()
+            if len(self._items) >= self.max_items:
+                self.log(f"已收集 {len(self._items)} 条原创推文,达到上限 {self.max_items},停止滚动")
+                break
+            if self.since and self._oldest_seen and self._oldest_seen < self.since:
+                self.log(f"已滚动到 {_date_only(self._oldest_seen)},早于 since={_date_only(self.since)},停止滚动")
+                break
+            if len(self._seen) == before_seen:
+                idle += 1
+                if idle >= self.idle_scrolls:
+                    self.log(f"连续 {idle} 次滚动无新数据,判定到底,停止(原创 {len(self._items)} / 已处理 {len(self._seen)})")
+                    break
+            else:
+                idle = 0
+            if n % 10 == 0:
+                self.log(f"滚动 {n} 次,原创 {len(self._items)} / 已处理 {len(self._seen)}")
+
+    def _reparse_so_far(self) -> None:
+        """用目前所有响应体重新解析一遍(幂等,_seen/_items 去重)。"""
+        for body in list(self._bodies):
+            try:
+                self._consume_body(body)
+            except Exception:  # noqa: BLE001
+                pass
+
+    # ------------------------------------------------------------------
+    # GraphQL JSON 解析
+    # ------------------------------------------------------------------
+
+    def _consume_body(self, body: str) -> None:
+        data = json.loads(body)
+        for entry in self._iter_entries(data):
+            try:
+                self._handle_entry(entry)
+            except Exception:  # noqa: BLE001
+                continue
+
+    def _iter_entries(self, data: Any):
+        # 候选根: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 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):
+                continue
+            for ins in instructions:
+                itype = ins.get("type") if isinstance(ins, dict) else None
+                if itype in ("TimelineAddEntries", "TimelineReplaceEntry"):
+                    entries = ins.get("entries") or ([ins.get("entry")] if ins.get("entry") else [])
+                    for e in entries:
+                        if isinstance(e, dict):
+                            yield e
+                elif itype == "TimelinePinEntry" and isinstance(ins.get("entry"), dict):
+                    yield ins["entry"]
+
+    def _handle_entry(self, entry: dict) -> None:
+        entry_id = str(entry.get("entryId") or "")
+        if entry_id.startswith(("cursor-", "promoted-tweet")):
+            return
+        content = entry.get("content") or {}
+        etype = content.get("entryType") or content.get("__typename")
+        if etype == "TimelineTimelineItem":
+            item_content = content.get("itemContent") or {}
+            self._handle_item_content(item_content)
+        elif etype == "TimelineTimelineModule":
+            # 自串(self-thread) 等会包成 module,里面多条 item
+            for sub in content.get("items") or []:
+                ic = _safe(sub, "item", "itemContent") or {}
+                self._handle_item_content(ic)
+
+    def _handle_item_content(self, item_content: dict) -> None:
+        if not isinstance(item_content, dict):
+            return
+        if item_content.get("itemType") not in (None, "TimelineTweet"):
+            return
+        result = _tweet_result_node(_safe(item_content, "tweet_results", "result"))
+        if not isinstance(result, dict):
+            return
+        self._handle_tweet_result(result)
+
+    def _handle_tweet_result(self, result: dict) -> None:
+        tweet_id = str(result.get("rest_id") or _safe(result, "legacy", "id_str") or "")
+        if not tweet_id or tweet_id in self._seen:
+            return
+        self._seen.add(tweet_id)
+        legacy = result.get("legacy") or {}
+        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:
+                self._oldest_seen = created
+
+        # 作者 handle(新旧 API 路径都试)
+        author = (
+            _safe(result, "core", "user_results", "result", "legacy", "screen_name")
+            or _safe(result, "core", "user_results", "result", "core", "screen_name")
+            or self._handle
+        )
+
+        # —— 过滤:转推 / 他人回复 ——(单条推文模式不过滤,用户要的就是那一条)
+        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 and not self._target_tweet_id:
+            if self.since and created < self.since:
+                return
+            if self.until and created > self.until:
+                return
+
+        raw_text = _full_text(result, legacy)
+        text = _expand_tco(raw_text, legacy)
+
+        # 引用推文
+        is_quote = bool(legacy.get("is_quote_status"))
+        quoted = _tweet_result_node(_safe(result, "quoted_status_result", "result"))
+        if isinstance(quoted, dict):
+            q_legacy = quoted.get("legacy") or {}
+            q_author = (
+                _safe(quoted, "core", "user_results", "result", "legacy", "screen_name")
+                or _safe(quoted, "core", "user_results", "result", "core", "screen_name")
+                or "?"
+            )
+            q_text = _expand_tco(_full_text(quoted, q_legacy), q_legacy)
+            if q_text:
+                text = f"{text}\n\n— 引用 @{q_author}:\n{q_text}".strip()
+
+        # 媒体(extended_entities 优先)
+        media_list = _safe(legacy, "extended_entities", "media") or _safe(legacy, "entities", "media") or []
+        images: list[str] = []
+        video_url: Optional[str] = None
+        video_thumb: Optional[str] = None
+        for m in media_list:
+            mtype = m.get("type")
+            if mtype == "photo":
+                u = m.get("media_url_https")
+                if u:
+                    images.append(u + ("?name=orig" if "?" not in u else ""))
+            elif mtype in ("video", "animated_gif"):
+                vu = _pick_video_url(m)
+                if vu and not video_url:
+                    video_url = vu
+                    video_thumb = m.get("media_url_https")
+
+        if video_url:
+            media_kind = "video"
+        elif images:
+            media_kind = "photo"
+        elif self._target_tweet_id:
+            # 单条推文模式:用户明确指定了 id,即使纯文本也返回
+            media_kind = "text"
+        else:
+            # 用户主页模式:纯文本推文不要,主站只要带媒体的
+            return
+
+        title = ""
+        for line in (text or "").splitlines():
+            line = line.strip()
+            if line:
+                title = line
+                break
+        if not title:
+            title = f"@{author} 的推文 {tweet_id}"
+        title = title[:200]
+
+        favorite = legacy.get("favorite_count")
+        retweet = legacy.get("retweet_count")
+        reply_c = legacy.get("reply_count")
+        summary_bits = []
+        if any(v is not None for v in (favorite, retweet, reply_c)):
+            summary_bits.append(f"❤{favorite or 0} 🔁{retweet or 0} 💬{reply_c or 0}")
+        summary = " · ".join([s for s in [title[:80], *summary_bits] if s]) or None
+
+        source_url = f"https://x.com/{author}/status/{tweet_id}"
+        item = ScrapedItem(
+            source_url=source_url,
+            title=title,
+            source_name="twitter",
+            media_kind=media_kind,
+            collector="twitter",
+            cover_image=(video_thumb or (images[0] if images else None)),
+            video_url=video_url,
+            images=images,
+            summary=summary,
+            content=(text or title),
+            page_no=1,
+            video_id=tweet_id,
+            extra={
+                "tweet_id": tweet_id,
+                "tweet_created_at": created.astimezone(timezone.utc).isoformat() if created else None,
+                "author_handle": author,
+                "is_quote": is_quote,
+                "favorite_count": favorite,
+                "retweet_count": retweet,
+                "reply_count": reply_c,
+                # 保留 Twitter 原始正文(含 t.co 短链),便于追溯
+                "original_full_text": raw_text,
+            },
+        )
+        if media_kind == "photo":
+            item.image_total = len(images)
+        if video_url:
+            item.extra["original_video_url"] = video_url
+            if video_thumb:
+                item.extra["original_video_thumb"] = video_thumb
+        if images:
+            item.extra["original_images"] = list(images)
+        self._items[tweet_id] = item

server/app.py

@@ -25,8 +25,9 @@ logger = logging.getLogger("harvester.app")
 
 SERVICE_TOKEN = os.environ.get("HARVESTER_TOKEN", "")
 HEADLESS = os.environ.get("HEADLESS", "1") not in ("0", "false", "False")
+_DEFAULT_ALLOWED_HOSTS = "xchina.co,x.com,twitter.com,pbs.twimg.com,video.twimg.com"
 ALLOWED_HOSTS = tuple(
-    h.strip().lower() for h in os.environ.get("ALLOWED_HOSTS", "xchina.co").split(",") if h.strip()
+    h.strip().lower() for h in os.environ.get("ALLOWED_HOSTS", _DEFAULT_ALLOWED_HOSTS).split(",") if h.strip()
 )
 MAX_PAGES = int(os.environ.get("MAX_PAGES", "50"))
 DEFAULT_MAX_IMAGES = int(os.environ.get("DEFAULT_MAX_IMAGES", "100"))
@@ -69,10 +70,14 @@ async def healthz():
 async def invoke(agent_id: str, payload: dict, authorization: str | None = Header(default=None)):
     _check_auth(authorization)
 
+    def _as_int(v, default):
+        try:
+            return int(v)
+        except (TypeError, ValueError):
+            return default
+
     # payload 即主站 task.input_params:含 content_url / platform_name / source_kind / start_page / end_page / _crawl_config ...
-    source_kind = payload.get("source_kind")
-    if source_kind is not None and str(source_kind) != "website":
-        raise HTTPException(status_code=400, detail=f"unsupported source_kind: {source_kind}(本服务只处理 website)")
+    source_kind = str(payload.get("source_kind") or "").strip().lower()
 
     content_url = (payload.get("content_url") or payload.get("series_url") or payload.get("url") or "").strip()
     if not content_url:
@@ -80,19 +85,40 @@ async def invoke(agent_id: str, payload: dict, authorization: str | None = Heade
     if not _host_allowed(content_url):
         raise HTTPException(status_code=400, detail=f"host not allowed: {content_url}")
 
-    def _as_int(v, default):
-        try:
-            return int(v)
-        except (TypeError, ValueError):
-            return default
+    # ---- Twitter / X 来源:抓某账号的原创推文列表 ----
+    from core.scraper_twitter import is_twitter_url
 
+    is_twitter = source_kind in ("twitter", "x") or is_twitter_url(content_url)
+    if source_kind and source_kind not in ("website", "twitter", "x"):
+        raise HTTPException(status_code=400, detail=f"unsupported source_kind: {source_kind}")
+
+    source_name = str(payload.get("platform_name") or ("twitter" if is_twitter else "xchina"))[:100]
+    # MinIO 配置由主站经 lobster payload 下发(主站把绑定的 minio_* Skill 变量并入 payload);
+    # 缺失则降级:不上传,媒体地址保留原始来源地址
+    minio_cfg = MinioConfig.from_payload(payload)
+
+    if is_twitter:
+        max_items = _as_int(payload.get("max_items"), _as_int(payload.get("max_images"), DEFAULT_MAX_IMAGES))
+        max_items = max(1, min(max_items, MAX_IMAGES_CAP))
+        job = jobs.submit(
+            content_url, source_name, 1, 1,
+            headless=HEADLESS, minio_cfg=minio_cfg, media_kind="twitter", max_images=max_items,
+            cookies=payload.get("cookies") or payload.get("x_cookies") or payload.get("X_COOKIES"),
+            since=payload.get("since"), until=payload.get("until"),
+        )
+        logger.info(
+            "invoke agent=%s job=%s kind=twitter url=%s max_items=%s since=%s until=%s minio=%s cookies=%s",
+            agent_id, job.id, content_url, max_items, payload.get("since"), payload.get("until"),
+            "on" if minio_cfg.is_complete else "off", "yes" if job.cookies else "no",
+        )
+        return {"task_id": job.id, "status": job.status}
+
+    # ---- 内容网站(xchina 等)----
     start_page = max(1, _as_int(payload.get("start_page"), 1))
     end_page = max(start_page, _as_int(payload.get("end_page"), start_page))
     if end_page - start_page + 1 > MAX_PAGES:
         end_page = start_page + MAX_PAGES - 1
 
-    source_name = str(payload.get("platform_name") or "xchina")[:100]
-
     # 媒体类型:可由 payload.media_kind 显式指定(video/photo);否则按 content_url 自动判断;
     # 再否则按 agent_id 兜底(含 photo → photo)。
     media_kind = str(payload.get("media_kind") or "").strip().lower()
@@ -110,10 +136,6 @@ async def invoke(agent_id: str, payload: dict, authorization: str | None = Heade
     max_images = _as_int(payload.get("max_images"), DEFAULT_MAX_IMAGES)
     max_images = max(1, min(max_images, MAX_IMAGES_CAP))
 
-    # MinIO 配置由主站经 lobster payload 下发(主站把绑定的 minio_* Skill 变量并入 payload);
-    # 缺失则降级:不上传,媒体地址保留原始来源地址
-    minio_cfg = MinioConfig.from_payload(payload)
-
     job = jobs.submit(
         content_url, source_name, start_page, end_page,
         headless=HEADLESS, minio_cfg=minio_cfg, media_kind=media_kind, max_images=max_images,

server/jobs.py

@@ -27,6 +27,8 @@ from core.media import (
     guess_content_type,
     upload_to_minio,
 )
+import re
+
 from core.scraper import ScrapeCancelled, XChinaScraper
 
 logger = logging.getLogger("harvester.jobs")
@@ -35,7 +37,7 @@ _RUN_LOCK = asyncio.Lock()
 
 TMP_ROOT = Path(os.environ.get("TMP_DIR", "/tmp")) / "content-harvester"
 CONCURRENT_FRAGMENTS = int(os.environ.get("CONCURRENT_FRAGMENTS", "8"))
-MAX_VIDEO_MB = int(os.environ.get("MAX_VIDEO_MB", "0"))  # 0 = 不限
+MAX_VIDEO_MB = int(os.environ.get("MAX_VIDEO_MB", "500"))  # 单个视频上限,0=不限;默认 500MB 避免误下载几个 G 的长视频
 
 
 @dataclass
@@ -46,8 +48,11 @@ class Job:
     start_page: int
     end_page: int
     minio_cfg: MinioConfig = field(default_factory=MinioConfig)
-    media_kind: str = "auto"          # auto | video | photo
-    max_images: int = 100             # 图集每个套图最多抓多少张(media_kind=photo 时)
+    media_kind: str = "auto"          # auto | video | photo | twitter
+    max_images: int = 100             # 图集每个套图最多抓多少张(media_kind=photo);twitter 时复用为 max_items
+    cookies: Any = None               # twitter:cookie 字符串("auth_token=...; ct0=...")或列表
+    since: Any = None                 # twitter:只采集该日期(YYYY-MM-DD)之后的推文
+    until: Any = None                 # twitter:只采集该日期之前的推文
     status: str = "running"           # running | succeeded | failed
     output: Optional[dict[str, Any]] = None
     error: Optional[str] = None
@@ -65,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:],
         }
 
 
@@ -80,11 +86,18 @@ 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)
 
 
+def _key_prefix(job: Job) -> str:
+    """MinIO object key 前缀,按来源名取(xchina / twitter / …),无效则回退 xchina。"""
+    raw = (job.source_name or "").strip().lower()
+    slug = re.sub(r"[^a-z0-9]+", "-", raw).strip("-")
+    return slug or "xchina"
+
+
 def _bytes_h(n: int) -> str:
     f = float(n)
     for unit in ("B", "KB", "MB", "GB"):
@@ -102,11 +115,15 @@ async def _process_uploads(job: Job, result, tmp_dir: Path) -> None:
             raise ScrapeCancelled()
         vid = item.video_id or f"item{idx}"
         item_dir = tmp_dir / vid
+        mk = getattr(item, "media_kind", "video")
         try:
-            if getattr(item, "media_kind", "video") == "photo":
+            if mk == "photo":
                 await _process_photo_item(job, item, idx, total, item_dir)
-            else:
+            elif mk == "video" or item.video_url:
                 await _process_video_item(job, item, idx, total, vid, item_dir)
+            else:
+                # 纯文本条目,无媒体需要下载
+                continue
         except ScrapeCancelled:
             raise
         except Exception as exc:  # noqa: BLE001
@@ -124,12 +141,15 @@ async def _process_video_item(job: Job, item, idx: int, total: int, vid: str, it
         _log(job, f"[{idx}/{total}] {item.title}: 无视频地址,跳过上传")
         return
     _log(job, f"[{idx}/{total}] {item.title}: 下载视频 …")
+    prefix = f"[{idx}/{total}] {item.title}"
     mp4_path = await asyncio.to_thread(
         download_video, item.video_url, item.source_url, item_dir, vid,
         CONCURRENT_FRAGMENTS, MAX_VIDEO_MB,
+        lambda m: _log(job, f"{prefix}:{m}"),
     )
     size = mp4_path.stat().st_size
-    video_key = f"xchina/{vid}{mp4_path.suffix.lower() or '.mp4'}"
+    pref = _key_prefix(job)
+    video_key = f"{pref}/{vid}{mp4_path.suffix.lower() or '.mp4'}"
     _log(job, f"[{idx}/{total}] {item.title}: 视频 {_bytes_h(size)},上传 MinIO {video_key} …")
     video_public = await asyncio.to_thread(upload_to_minio, mp4_path, video_key, "video/mp4", cfg)
     item.extra["original_m3u8"] = item.video_url
@@ -140,7 +160,7 @@ async def _process_video_item(job: Job, item, idx: int, total: int, vid: str, it
     if item.cover_image:
         cover_path = await asyncio.to_thread(download_cover, item.cover_image, item.source_url, item_dir, vid)
         if cover_path is not None:
-            cover_key = f"xchina/{vid}{cover_path.suffix.lower() or '.jpg'}"
+            cover_key = f"{pref}/{vid}{cover_path.suffix.lower() or '.jpg'}"
             try:
                 cover_public = await asyncio.to_thread(
                     upload_to_minio, cover_path, cover_key, guess_content_type(cover_path, "image/jpeg"), cfg,
@@ -156,6 +176,7 @@ async def _process_video_item(job: Job, item, idx: int, total: int, vid: str, it
 async def _process_photo_item(job: Job, item, idx: int, total: int, item_dir: Path) -> None:
     """逐张下载图集图片 → 上传 MinIO → 把 item.images 换成 MinIO 地址(边下边传边删,不囤盘)。"""
     cfg = job.minio_cfg
+    pref = _key_prefix(job)
     detail_id = item.video_id or f"item{idx}"
     src_images = list(item.images or [])
     if not src_images:
@@ -177,7 +198,7 @@ async def _process_photo_item(job: Job, item, idx: int, total: int, item_dir: Pa
             fail_cnt += 1
             continue
         ext = path.suffix.lower() or ".webp"
-        key = f"xchina/{detail_id}/{i:04d}{ext}"
+        key = f"{pref}/{detail_id}/{i:04d}{ext}"
         try:
             public = await asyncio.to_thread(
                 upload_to_minio, path, key, guess_content_type(path, "image/webp"), cfg,
@@ -212,8 +233,11 @@ async def _process_photo_item(job: Job, item, idx: int, total: int, item_dir: Pa
 
 def _resolve_media_kind(job: Job) -> str:
     kind = (job.media_kind or "auto").lower()
-    if kind in ("video", "photo"):
+    if kind in ("video", "photo", "twitter"):
         return kind
+    from core.scraper_twitter import is_twitter_url
+    if is_twitter_url(job.series_url):
+        return "twitter"
     from core.scraper_photos import is_photo_detail_url, is_photo_series_url
     url = job.series_url
     if is_photo_series_url(url) or is_photo_detail_url(url):
@@ -222,6 +246,12 @@ def _resolve_media_kind(job: Job) -> str:
 
 
 def _make_scraper(kind: str, ctx, job: Job):
+    if kind == "twitter":
+        from core.scraper_twitter import XTwitterScraper
+        return XTwitterScraper(
+            ctx, is_cancelled=lambda: job._cancelled, log=lambda m: _log(job, m),
+            cookies=job.cookies, since=job.since, until=job.until, max_items=job.max_images,
+        )
     if kind == "photo":
         from core.scraper_photos import XChinaPhotoScraper
         return XChinaPhotoScraper(
@@ -235,8 +265,17 @@ async def _run(job: Job, headless: bool) -> None:
     try:
         async with _RUN_LOCK:
             kind = _resolve_media_kind(job)
-            extra = f", max_images={job.max_images}" if kind == "photo" else ""
-            _log(job, f"开始采集[{kind}]: {job.series_url} (page {job.start_page}-{job.end_page}{extra})")
+            if kind == "photo":
+                _log(job, f"开始采集[photo]: {job.series_url} (page {job.start_page}-{job.end_page}, max_images={job.max_images})")
+            elif kind == "twitter":
+                win = []
+                if job.since:
+                    win.append(f"since={job.since}")
+                if job.until:
+                    win.append(f"until={job.until}")
+                _log(job, f"开始采集[twitter]: {job.series_url} (max_items={job.max_images}{', ' + ', '.join(win) if win else ''})")
+            else:
+                _log(job, f"开始采集[{kind}]: {job.series_url} (page {job.start_page}-{job.end_page})")
             if job.minio_cfg.is_complete:
                 _log(job, f"启用 MinIO 上传: {job.minio_cfg.safe_repr()}")
             else:
@@ -284,7 +323,8 @@ async def _run(job: Job, headless: bool) -> None:
 
 def submit(series_url: str, source_name: str, start_page: int, end_page: int,
            headless: bool = True, minio_cfg: Optional[MinioConfig] = None,
-           media_kind: str = "auto", max_images: int = 100) -> Job:
+           media_kind: str = "auto", max_images: int = 100,
+           cookies: Any = None, since: Any = None, until: Any = None) -> Job:
     job = Job(
         id=uuid.uuid4().hex,
         series_url=series_url,
@@ -294,6 +334,9 @@ def submit(series_url: str, source_name: str, start_page: int, end_page: int,
         minio_cfg=minio_cfg or MinioConfig(),
         media_kind=(media_kind or "auto"),
         max_images=max(1, int(max_images or 100)),
+        cookies=cookies,
+        since=since,
+        until=until,
     )
     _JOBS[job.id] = job
     job._task = asyncio.create_task(_run(job, headless))