Merge branch 'hy-dev' into dev

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,555 @@
+"""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")
+_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 extract_handle(url: str) -> Optional[str]:
+    """https://x.com/elonmusk?foo=1 -> 'elonmusk';带 /status/ 等子路径或保留路径则返回 None。"""
+    try:
+        path = urlparse(url).path.strip("/")
+    except Exception:
+        return None
+    if not path:
+        return None
+    parts = path.split("/")
+    handle = parts[0]
+    # 保留路径 / 非用户页
+    if handle.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
+    if not re.fullmatch(r"[A-Za-z0-9_]{1,15}", handle):
+        return None
+    return handle
+
+
+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()
+
+
+def _strip_media_tco(text: str, legacy: dict) -> str:
+    """去掉正文末尾 Twitter 自动附加的图片/视频 t.co 短链。"""
+    media = _safe(legacy, "entities", "media") or []
+    for m in media:
+        u = m.get("url")
+        if u:
+            text = text.replace(u, "")
+    return text.strip()
+
+
+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 = ""
+
+    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 = extract_handle(series_url)
+        if not handle:
+            raise ValueError(f"无法从链接解析出 Twitter 用户名(应形如 https://x.com/<用户名>): {series_url}")
+        self._handle = handle
+
+        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:
+            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)
+
+            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 i, it in enumerate(items, 1):
+            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)} 条)")
+        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 _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
+            # 先用已收集的响应数粗判,够了就停
+            before_bodies = len(self._bodies)
+            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._bodies) == before_bodies:
+                idle += 1
+                if idle >= self.idle_scrolls:
+                    self.log(f"连续 {idle} 次滚动无新数据,判定到底,停止")
+                    break
+            else:
+                idle = 0
+            if n % 10 == 0:
+                self.log(f"滚动 {n} 次,已收集原创推文 {len(self._items)} 条")
+
+    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):
+        # data.user.result.timeline_v2.timeline.instructions 或 .timeline.timeline.instructions
+        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
+            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
+
+        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 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 self.since and created < self.since:
+                return
+            if self.until and created > self.until:
+                return
+
+        text = _strip_media_tco(_full_text(result, legacy), 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 = _strip_media_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"
+        else:
+            media_kind = "text"
+
+        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,
+            },
+        )
+        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")
@@ -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
@@ -85,6 +90,13 @@ def _log(job: Job, msg: str) -> None:
     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 +114,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
@@ -129,7 +145,8 @@ async def _process_video_item(job: Job, item, idx: int, total: int, vid: str, it
         CONCURRENT_FRAGMENTS, MAX_VIDEO_MB,
     )
     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 +157,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 +173,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 +195,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 +230,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 +243,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 +262,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 +320,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 +331,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))