feat(twitter): t.co 替换 + yt-dlp 进度心跳 + access_key 全脱敏

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>
This commit is contained in:
2026-05-13 18:03:57 +08:00
co-authored by Claude Opus 4.7
parent 85b615d9ab
commit 64e382897b
3 changed files with 67 additions and 7 deletions
+37 -2
View File
@@ -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])
+28 -5
View File
@@ -168,14 +168,34 @@ def _full_text(result: dict, legacy: dict) -> str:
return str(legacy.get("full_text") or "").strip()
def _strip_media_tco(text: str, legacy: dict) -> str:
"""去掉正文末尾 Twitter 自动附加的图片/视频 t.co 短链。"""
_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, "")
return text.strip()
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]:
@@ -598,7 +618,8 @@ class XTwitterScraper:
if self.until and created > self.until:
return
text = _strip_media_tco(_full_text(result, legacy), legacy)
raw_text = _full_text(result, legacy)
text = _expand_tco(raw_text, legacy)
# 引用推文
is_quote = bool(legacy.get("is_quote_status"))
@@ -610,7 +631,7 @@ class XTwitterScraper:
or _safe(quoted, "core", "user_results", "result", "core", "screen_name")
or "?"
)
q_text = _strip_media_tco(_full_text(quoted, q_legacy), q_legacy)
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()
@@ -682,6 +703,8 @@ class XTwitterScraper:
"favorite_count": favorite,
"retweet_count": retweet,
"reply_count": reply_c,
# 保留 Twitter 原始正文(含 t.co 短链),便于追溯
"original_full_text": raw_text,
},
)
if media_kind == "photo":
+2
View File
@@ -141,9 +141,11 @@ 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
pref = _key_prefix(job)