采集 X 媒体：会出错的四个地方
采集 X 媒体：会出错的四个地方
从 X 取媒体看起来是个已解决的问题：拿到推文、读出 URL、下载文件。代码十二行，在你第一个测试账号上跑得好好的。
然后你拿几千条推跑一遍，发现四个各自独立的问题，而且没有一个会主动报错。文件互相覆盖、GIF 全变成 mp4、每张图都只有应有分辨率的四分之一，以及任务跑到 70% 挂掉时你没法续跑。
这篇只讲这四个。取数本身是最简单的部分——取某个账号的媒体那篇讲过——这篇假设响应已经在你手里，现在要正确处理它。
有个免费 demo 端点，可以先摸清响应结构，再围着它写下载逻辑。
先说清楚：图片和视频是两个不同字段
这个在别的问题之前就会先绊你一下，因为两种媒体不在同一个数组里。
{
"id": "...",
"photos": ["https://pbs.twimg.com/media/..."],
"videos": [
{
"url": "https://video.twimg.com/...mp4",
"duration_ms": 21000,
"thumbnail": "https://pbs.twimg.com/..."
}
]
}
photos 是字符串列表，videos 是对象列表。 写一个循环当它们一样处理，这是第一个 bug——遇到第一条视频推就 TypeError，这还算响亮的那种。安静的那种版本是：代码只读 photos，于是视频推上什么都没采到，而且不报错。
一条推最多四张图，或者一个视频，但不会两者都有。主机也不同：图片在 pbs.twimg.com，视频在 video.twimg.com。
1. X 上的 GIF 不是 GIF
X 把「GIF」存成无声循环播放的 mp4。整条链路里没有任何 .gif。
这件事有两重影响。如果你按扩展名存文件，你会写出一堆实际是 mp4 容器的 .gif 文件——而下游任何信任扩展名的东西都打不开它们。如果你在做展示界面，给 GIF 配上视频控制条会很怪：用户预期它自己静音循环，而不是等人点播放。
类型在 video 对象上，所以按它判断，不要按时长或文件大小猜：
def media_kind(video):
# animated_gif 本质是无声循环 mp4 —— 不是 .gif 文件
return "gif" if video.get("type") == "animated_gif" else "video"
短视频和 GIF 按长度根本分不开，这就是为什么拿 duration_ms 猜不行。
2. 文件名冲突会悄悄丢文件
最顺手的文件名是 URL 的最后一段路径。单个账号没问题，跨账号就有问题，而且这个失效是看不见的——你只是最后发现文件数比推文数少。
import re
def safe_name(tweet_id, index, url, kind):
ext = "mp4" if kind in ("video", "gif") else \
(re.search(r'\.(jpg|jpeg|png|webp)', url) or ["", "jpg"])[1]
slot = "v" if kind in ("video", "gif") else "p"
# tweet_id + 类型 + 序号, 每个媒体位唯一; 扩展名来自 kind 而不是 URL
return f"{tweet_id}_{slot}{index}.{ext}"
按推文 ID 加媒体序号做键，不要按 URL。 不同推文携带的媒体可能有相同的基础文件名，而一条四图推需要从同一个推文 ID 产出四个不同名字。
p/v 这个标记是双保险。 实际上一条推要么带图、要么带一个视频，不会两者都有，所以光靠序号就已经唯一了。这个前缀不花成本，而且在你把两个来源的媒体合并到同一个目录时仍然正确。
扩展名要从媒体类型推，而不是从 URL 推——原因就是上面的 GIF 问题：一个以 .mp4 结尾但实际是 animated GIF 的 URL 仍然该存成 .mp4，而带查询参数的图片 URL 会把粗率的 rsplit('.') 搞晕。
3. 图片尺寸写在 URL 里
X 用多个分辨率提供同一张图，你拿到哪个取决于你请求的 URL。这是可以实测的——下面是写这篇时实测的某张头像的真实字节数：
| URL 形式 | 字节 |
|---|---|
..._normal.jpg |
2,111 |
..._bigger.jpg |
3,151 |
..._400x400.jpg |
28,375 |
....jpg（无后缀） |
32,796 |
★后缀就是降采样，去掉它能拿到这个 URL 提供的最大版本★——同一张图，_normal 和无后缀之间差 15 倍。
import re
def upscale_profile_image(url: str) -> str:
# 去掉 _normal / _bigger / _400x400 这类尺寸后缀
return re.sub(r'_(normal|bigger|mini|\d+x\d+)(\.[a-z]{3,4})$', r'\2', url)
两个必须说清的前提，因为这一段是最容易吹过头的地方：
头像 URL 和推文媒体 URL 是不同路径（/profile_images/ vs /media/），行为不一定一致。上面的字节数是在头像上实测的——不要在没验证的情况下假设同一套后缀对别的路径也成立。
去掉后缀不会创造出原本没上传的分辨率。 你拿到的是已存储的最大版本，原图本来就小的话，你得到的是同样的像素、更大的文件。用返回图片的实际尺寸去确认，不要信 URL——方法在图片尺寸与裁切那篇里。
4. 可续跑，否则你会把一切重下一遍
媒体任务慢，因为文件大。这意味着它一定会被打断——而没有状态，被打断就等于从头再来。
先看清时间花在哪。 列表只是一次请求：媒体类端点一次调用返回最多 limit 条，没有游标可走。真正花几分钟到几小时的是下载，所以要存档的是下载进度。
import json, os, pathlib, requests
API, HEADERS = "https://api.socialapi.tech/v1", {"X-API-Key": "your_key"}
def save_state(state_file, done):
tmp = state_file + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump({"done": sorted(done)}, f)
os.replace(tmp, state_file) # 原子写, 和文件本身一样
def scrape_media(username, out="media", state_file="media_state.json", limit=200):
out_dir = pathlib.Path(out) / username
out_dir.mkdir(parents=True, exist_ok=True)
done = set()
if os.path.exists(state_file):
try:
with open(state_file, encoding="utf-8") as f:
done = set(json.load(f).get("done", []))
except (json.JSONDecodeError, OSError):
pass # 状态损坏: 重新开始, 不要崩
# 一次列表调用 —— 续跑时重新调一次很便宜
r = requests.get(f"{API}/user/media", params={"username": username, "limit": limit},
headers=HEADERS, timeout=60)
if r.status_code == 429:
return # 干净地停下, 以后续跑
r.raise_for_status()
posts = r.json().get("data") or []
for n, p in enumerate(posts, 1):
tid = p["id"]
for i, url in enumerate(p.get("photos") or []):
fetch_one(url, out_dir / safe_name(tid, i, url, "photo"), done)
for i, v in enumerate(p.get("videos") or []):
if v.get("url"):
kind = media_kind(v)
fetch_one(v["url"], out_dir / safe_name(tid, i, v["url"], kind), done)
if n % 10 == 0:
save_state(state_file, done) # 每 10 条帖子存档一次
save_state(state_file, done)
def fetch_one(url, path, done):
if url in done or path.exists():
return # 两道独立的跳过判断
tmp = path.with_suffix(path.suffix + ".part")
try:
with requests.get(url, stream=True, timeout=120) as resp:
resp.raise_for_status()
with open(tmp, "wb") as f:
for chunk in resp.iter_content(1 << 16):
f.write(chunk)
tmp.replace(path) # 到这一刻才算完成
done.add(url)
except requests.RequestException:
tmp.unlink(missing_ok=True) # 绝不留下截断的文件
这里有三处是必要的。
先下到 .part，再改名。 否则下载中途崩溃会留下一个截断文件，而 path.exists() 会很乐意永远跳过它。改名是原子的，所以文件要么不存在、要么是完整的。
存档下载进度，不是存档列表。 列表是一次很便宜的请求，续跑时重调一次就行。下载才是贵的部分，所以状态文件每隔几条帖子记一次已完成的 URL，和文件本身一样原子写入。
失败时删掉半截文件。 没有这一步，一次网络抖动就永久污染了那个文件——它存在，所以跳过判断通过，于是你永远不会发现它只是半张 JPEG。
这件事做不到的部分
- 受保护账号的媒体——取不到，换什么工具都一样。
- 已删除的媒体——X 把文件移除后 URL 就 404 了，即使你还留着那条推。
- 超出 X 存储版本的原图——平台上传时会重新编码，你拿到的是它的版本。
- 绕过限流批量下载——媒体 URL 在 CDN 上，但找到它们的那些 API 调用仍然计量。
关于权利：公开可见和「可以转载」是两件不同的事。 采集媒体用于分析，和把它重新发布出去，是不同性质的行为，而后者受著作权约束、与 API 条款无关。红线在哪那篇讲得更完整。
我们的 /v1/user/media 返回的就是上面这些结构，另有 /v1/user/photos 和 /v1/user/videos，想只要一种类型时不用在客户端过滤。用 Google 一键注册送 $0.20 额度、不用信用卡，够你拿上面的脚本跑一个真实账号、看看自己哪个假设会被打破。
常见问题
怎么从推特抓图片？
取推文、读 photos 数组、逐个下载 URL。真正要花心思的是之后：唯一文件名、原子写入、可续跑的状态。
为什么图片和视频分在两个字段？
因为结构不同——photos 是 URL 字符串，videos 是带 url、duration_ms、thumbnail 的对象。一个循环处理两者会在第一条视频推上抛 TypeError。
为什么我下载的 GIF 是 mp4？
因为 X 把 GIF 存成无声循环 mp4，链路里没有 .gif。看 video 的类型字段是不是 animated_gif，不要从扩展名猜。
怎么区分 GIF 和短视频？ 只能靠类型字段。时长区分不了——很多真视频比很多 GIF 还短。
怎么拿到原图而不是缩略图？
去掉 URL 里的尺寸后缀。 某张头像实测：_normal 2,111 字节 vs 无后缀 32,796 字节——同一张图，数据量差 15 倍。
去掉后缀一定能得到更高分辨率吗？ 得到的是该 URL 已存储的最大版本。原图本来就小的话，你拿到同样的像素、更大的文件。读返回图片的实际尺寸来确认，不要信 URL。
头像 URL 和推文图片 URL 行为一样吗？
它们是不同路径——/profile_images/ vs /media/——不该假设接受同样的参数。拿你手上真实的 URL 形式去测。
为什么我的文件数比推文数少？ 文件名冲突。 如果你用 URL 最后一段做名字，不同推文的媒体会互相静默覆盖。按推文 ID 加媒体序号做键。
一条推最多几张图？ 最多四张，或者一个视频，但不会两者都有。这就是文件名里需要序号的原因——一个推文 ID 能产出四个文件。
文件由哪个主机提供？
图片 pbs.twimg.com，视频 video.twimg.com。它们是 CDN，和给你 URL 的那个 API 是分开的。
该通过自己的服务器下载还是直连？ 直连 CDN。 把媒体经自己后端转发会增加带宽成本和一个故障点，而且没有收益——除非你确实需要改写或加水印。
怎么让媒体任务可续跑？ 每隔几条帖子，把已完成的 URL 原子写进状态文件。 重启时重调一次列表接口，跳过已完成或已在磁盘上的。
为什么要先下到 .part 文件？
因为下载中途崩溃会留下截断文件，而单纯的 exists() 判断会永远跳过它。写 .part 再改名意味着文件要么不存在、要么完整。
下载中途失败该怎么办？ 删掉半截文件。留着比没有更糟，因为你的跳过逻辑会把半张图当成已完成。
怎么避免重复下载已有文件？ 两道独立判断：状态文件里的已完成 URL 集合，加上文件系统存在性检查。单用任何一个都有缺口，一起用才覆盖重启和状态丢失。
能指定视频分辨率吗？ 你拿到的是 API 返回的那个 URL。转码是你这一侧的事——下载一次，需要别的规格就本地转。
视频有缩略图吗？ 有，在 video 对象上。是图片 URL，适合在不下载每个 mp4 的情况下先搭出网格界面。
怎么只取视频或只取图片？ 用专门的端点，不要在客户端过滤——这样你不用为之后扔掉的推文付费。
能下载受保护账号的媒体吗？ 不能。受保护内容不通过任何 API 提供，和你跟这个账号的关系无关。
推文被删后媒体会怎样？ URL 不再解析。存着推文对象不会让文件活着——需要就在看到时下载。
这个需要 X 开发者账号吗？ 通过第三方 API 读公开媒体不需要，你用的是服务商的 key。开发者账号是给官方 API 和写操作用的。
媒体 URL 能用多久？ 当它们不稳定来对待。有的时候就把字节下下来，不要只存 URL 等以后再取。
抓推特媒体合法吗？ 读公开数据和转载别人的照片是两件不同的事。 前者普遍被认为可以接受；后者是著作权问题，与 API 条款无关。
媒体任务里怎么处理限流？ 列表调用返回 429 就干净地停下，以后续跑。媒体文件本身走 CDN，所以被限流的不是下载。
媒体任务为什么比文本任务慢那么多？ 因为你在搬文件，不是搬 JSON。API 调用只占挂钟时间的一小部分，下载才是主体——这也是为什么可续跑在这里更重要。
媒体该存数据库还是磁盘？ 磁盘或对象存储，元数据进数据库。 把二进制当 blob 存会让备份和查询都很痛苦，而且没有真实收益。
媒体任务最能体现「按请求计费」的形状：API 调用很便宜、字节不要钱，所以你付的是「找到媒体」的钱，不是「搬运」的钱。 用 Google 注册送 $0.20 额度、不绑卡，可以拿你自己的账号清单去实测这笔账。