admin/content-harvester
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(twitter): 单条推文 URL + cookie 清旧 + log_tail 放宽

Browse Source 
P2 #5 cookie 清旧:run() 开头先 clear_cookies x.com/twitter.com 域,再 add_cookies。
  之前 persistent profile 累积旧 cookie,换账号时新 auth_token 可能被旧的串扰。
P2 #6 单条推文 URL:parse_x_url 识别 /status/{id} 和 /i/web/status/{id};单条推文
  模式跳过原创/时间窗过滤(用户要的就是那一条),text-only 也保留;TweetDetail/
  TweetResultByRestId GraphQL 响应解析。
P2 #7 log_tail 放宽:to_dict 返回末尾 80→500 行,内部上限 800→2000;100 条 tweet
  采集再也不会丢前面滚动/敏感页处理那段日志。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
huyue
2026-05-13 17:52:04 +08:00
co-authored by Claude Opus 4.7
parent 2ae3d9b802
commit 85b615d9ab
2 changed files with 151 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/scraper_twitter.py
+147 -53
View File
@@ -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
+4 -3
View File
@@ -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)