ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

Python爬取Instagram照片视频:从JSON解析到并发下载实战

2026/9/13 9:39:12 拓冰建站 浏览量
Python爬取Instagram照片视频:从JSON解析到并发下载实战 简介一套聚焦 Instagram 博主照片与视频抓取的 Python 爬虫项目资源面向有一定 Python 基础、希望学习社交平台数据采集的开发者。项目围绕 requests、BeautifulSoup、Selenium 三种典型工具展开覆盖模拟登录、动态内容加载、JSON 数据解析、文件保存与异常处理等核心环节可帮助读者打通从请求发送到媒体落盘的完整流程。包体共 4 个文件含 2 个 Python 脚本、1 个 README 说明文档和 1 个 .gitattributes 配置项压缩包仅 5KB结构紧凑便于直接查看源码与理解分工。已有 656 人学习浏览适合通过小型实战项目快速了解 Instagram 抓取的合规边界、反爬应对思路及异常恢复策略。1. 从一张截图到一个 Python 脚本Instagram 博主素材备份的第一步很多人误以为抓 Instagram 博主的照片视频直接拿wget -r镜像个人主页就行结果只会拿到一堆空壳 HTML。Instagram 的页面早已改为 React 渲染真实数据全部放在一串名为_sharedData的 JSON 里照片视频走独立 CDN还带签名参数和版本裁剪。本文要做的就是把这套结构拆开用 Python 的 requests 拿到 JSON用线程池并发拉取文件最后整理出一份可以直接落地的Instagram_crawler脚本思路。适合想把喜欢的博主作品归档、给运营整理竞品素材、或者做数据集训练的 Python 读者。默认你已经装好 Python 3.9 以上版本。2. Requests JSON 解析Instagram 公开相册照片视频的最小管道2.1 Instagram 个人主页不是静态 HTML是内嵌 JSON打开任意公开博主的个人主页右键查看源代码你会看到页面里塞满了script typetext/javascriptwindow._sharedData {...}/script。用户名、粉丝数、帖子数、照片视频的 CDN 地址全部在这一串 JSON 里。爬虫的起点不是解析 HTML 标签而是先把_sharedData抠出来。另一个变化是https://www.instagram.com/{username}/?__a1这类匿名 JSON 接口已经被关闭现在返回 404 或重定向到登录页。所以眼下最稳的链路是抓桌面端 HTML → 找_sharedData→ 解析 JSON → 取媒体 URL。这也是Instagram_crawler内部最常见的处理顺序。2.2 Python 环境安装与依赖清单建议用虚拟环境隔离避免污染系统 Pythonmkdir instagram_crawler cd instagram_crawler python3 -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install requests2.32.3 beautifulsoup44.12.3 lxml5.2.1三个库的分工requests发 HTTP 请求带上 Cookie、Headers 和参数beautifulsoup4lxml定位script标签并抽取 JSON 字符串比纯re正则稳不装selenium。这里没有动态点击需求纯 HTTP 就能拿全部媒体信息引入浏览器反而会触发更严格的风控。2.3 第一个有效脚本提取博主主页里的 JSON 数据import json import requests from bs4 import BeautifulSoup HEADERS { User-Agent: ( Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 ), Accept-Language: en-US,en;q0.9, } def fetch_shared_data(username: str) - dict: url fhttps://www.instagram.com/{username}/ resp requests.get(url, headersHEADERS, timeout15) resp.raise_for_status() soup BeautifulSoup(resp.text, lxml) # 找到包含 window._sharedData 的 script 标签 script_tag soup.find( script, textlambda t: t and window._sharedData in t ) if not script_tag: raise RuntimeError( f未找到 _sharedData用户名 {username} 可能不存在或页面结构已变 ) raw script_tag.string json_str raw.split(window._sharedData , 1)[1].rsplit(;/script, 1)[0] return json.loads(json_str) if __name__ __main__: data fetch_shared_data(instagram) print(data.keys())这段代码的关键是split与rsplit的配合先按window._sharedData 切出 JSON 开头再从尾部切掉;/script避开正则转义问题。拿到data之后媒体列表的位置在data[entry_data][ProfilePage][0][graphql][user][edge_owner_to_timeline_media][edges]每一条edge[node]是一个帖子。2.4 从 node 里解析照片和视频 URLdef extract_media(node: dict) - list: media_list [] if node.get(is_video): # 视频把 video_versions 按分辨率排序拿最大版本 vv node[video_versions] vv_sorted sorted(vv, keylambda x: x[height] * x[width], reverseTrue) media_list.append({ type: video, url: vv_sorted[0][url], width: vv_sorted[0][width], height: vv_sorted[0][height], }) else: # 图片image_versions2.candidates 里有多个尺寸取面积最大的 iv node[image_versions2][candidates] best max(iv, keylambda x: x[width] * x[height]) media_list.append({ type: image, url: best[url], width: best[width], height: best[height], }) # 多图帖carousel_media 里还有后续图片/视频要递归展开 if node.get(__typename) XDTGraphSidecar: for child in node.get(carousel_media, []): media_list.extend(extract_media(child)) return media_list注意__typename的枚举值在不同时期不一样XDTGraphImage、XDTGraphVideo、XDTGraphSidecar如果抓到的结果里是GraphImage把XDT前缀去掉即可。对多图帖做递归解析是因为轮播帖的node只保留第一张图其余全在carousel_media里不递归会丢素材。2.5 分页参数 max_id 与完整抓取循环个人主页第一页只显示最新 12 个帖子。要抓全部历史必须用end_cursor翻页def fetch_all_media(username: str, max_pages: int 10): after None results [] for page in range(max_pages): page_data fetch_graphql_page(username, after) # 伪代码见下方说明 edges page_data[data][user][edge_owner_to_timeline_media][edges] if not edges: break for edge in edges: results.append(extract_media(edge[node])) page_info page_data[data][user][edge_owner_to_timeline_media][page_info] after page_info.get(end_cursor) if not page_info.get(has_next_page): break return results这段代码里fetch_graphql_page是示意完整实现需要带着variables参数请求 GraphQL 端点。分页的核心是维护end_cursor它在每次响应的page_info里返回下一次请求把它塞回after参数。如果has_next_page为False说明已经翻完。字段名位置含义edge_owner_to_timeline_mediaProfilePage JSON 下graphql.user帖子分页容器edges[].node上述容器的子节点单个帖子的全部元数据page_info.end_cursor容器末尾下一次翻页游标page_info.has_next_page容器末尾是否还有下一页carousel_medianode 子节点多图帖中的后续媒体操作提示第一次运行时先只跑一页确认输出是https://scontent-*.cdninstagram.com/开头的外链。如果拿到blob:或data:开头的内容说明解析层级错了要回node与image_versions2的嵌套关系里排查。3. Instagram 登录态 Cookie 处理把低清缩略图升级为原图原视频3.1 为什么匿名抓取拿不到最高清素材第 2 章拿到的图片 URL 虽然能下载但分辨率通常被压到 1080px。博主手机上传原图是 1440px 甚至 4000px匿名接口不提供。视频同样受限匿名响应里的video_versions往往只有height720一条登录后能拿到多个版本。原因是 Instagram 会按请求者的登录状态动态裁剪媒体版本列表。登录后客户端能拿到更完整的video_versions数组所以只取url字段可能拿的不是最大版本。我的做法是把整个video_versions按height排序优先选择高度最大的 URL 下载第 2 章的extract_media已经做了这件事。3.2 用 requests.Session 维持登录态不推荐在脚本里模拟用户名密码登录Instagram 对登录接口有行为检测密码错误几次就可能触发checkpoint_required。常见做法是在浏览器里手动登录一次从 DevTools 里复制 Cookie 请求头写进脚本。import requests COOKIE_STR ( sessionid123456%3Aabc123; csrftokenXXXX; ds_user_id123456; ig_didABC-DEF; ) session requests.Session() session.headers.update({ User-Agent: ( Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Mobile/15E148 Safari/604.1 ), x-ig-app-id: 936619743392459, cookie: COOKIE_STR, })ig_did是设备指纹缺失时部分接口会拒答。x-ig-app-id是移动端接口的凭证桌面端网页不加也能跑加上后行为更接近 App。Cookie 有效期一般为 1 到 7 天失效时表现为请求返回 200但 JSON 里user是null。这时候直接去浏览器重新复制 Cookie 即可。3.3 Cookie 持久化不用每次跑脚本都手动复制把 Cookie 存到本地 JSON 文件检查保存时间超过 5 天就提示重新复制import json import os from datetime import datetime, timedelta COOKIE_FILE cookie.json def save_cookie(cookie_str: str): with open(COOKIE_FILE, w, encodingutf-8) as f: json.dump( {cookie: cookie_str, saved_at: datetime.now().isoformat()}, f, ensure_asciiFalse, indent2, ) def load_cookie() - str: if not os.path.exists(COOKIE_FILE): raise FileNotFoundError(缺少 cookie.json请先在浏览器登录后复制 Cookie) with open(COOKIE_FILE, encodingutf-8) as f: payload json.load(f) saved_at datetime.fromisoformat(payload[saved_at]) if datetime.now() - saved_at timedelta(days5): print(警告Cookie 已超过 5 天可能失效) return payload[cookie]这套持久化机制把「登录一次、长期使用」落实到文件层。实际操作里我一般把 Cookie 保存与加载封装成独立模块爬虫主程序只依赖load_cookie()换账号时自动重新加载。3.4 私密账号与关注限制的边界私密博主的主页在没有被关注时返回 JSON 里user字段存在但edge_owner_to_timeline_media为空。此时继续翻页只会换来限流。如果确实需要抓私密账号素材必须用已关注该账号的 Cookie并且抓取时加上?hlen参数避免非英文语言环境下的字段歧义。账号状态匿名请求返回登录请求返回公开账号edges有值edges有值媒体版本更多私密账号未关注edges为空数组edges为空数组私密账号已关注403 或 checkpointedges有值可抓取已注销404404有个容易踩坑的细节私密账号即便你已关注用桌面端 UA 请求也可能拿到空数据因为桌面端默认不带关注关系上下文。换成移动端 UA 通常能正常返回这是服务端对不同客户端实现的差异化返回。4. Instagram 素材并发下载与断点续传ThreadPoolExecutor Range 请求4.1 串行下载为什么慢假设一个博主有 3000 个帖子其中三分之一是视频平均每个视频 5MB。单线程for循环下载算上每个请求的握手和响应时间要跑 1 到 2 个小时。并发下载能把这个时间压缩到 10 到 20 分钟这对 Python 爬虫来说是很常见的性能需求。另一个问题是网络抖动单个大文件下载到一半断连整个文件就得重来。断点续传不是 Instagram 提供的功能而是 HTTP 协议本身支持的Range请求头指定从哪个字节开始拿服务器返回206 Partial Content。4.2 ThreadPoolExecutor 并发下载器import os from concurrent.futures import ThreadPoolExecutor, as_completed import requests SAVE_ROOT downloads def download_one(item: dict, username: str, session: requests.Session) - str: url item[url] media_type item[type] code item.get(code, unknown) idx item.get(idx, 0) ext .mp4 if media_type video else .jpg filepath os.path.join(SAVE_ROOT, username, media_type, f{code}_{idx}{ext}) os.makedirs(os.path.dirname(filepath), exist_okTrue) # 已存在且大小大于 0 则跳过避免重复下载 if os.path.exists(filepath) and os.path.getsize(filepath) 0: return filepath temp_path filepath .part downloaded os.path.getsize(temp_path) if os.path.exists(temp_path) else 0 headers {Range: fbytes{downloaded}-} with session.get(url, headersheaders, streamTrue, timeout30) as resp: if resp.status_code 206: mode ab # 追加写从断点续传 elif resp.status_code 200: mode wb # 服务器不支持 Range覆盖写 else: raise RuntimeError(f下载失败 {resp.status_code}: {url}) with open(temp_path, mode) as f: for chunk in resp.iter_content(chunk_size1024 * 256): if chunk: f.write(chunk) os.rename(temp_path, filepath) return filepath def parallel_download(media_items: list, username: str, session: requests.Session, workers: int 8): tasks [] with ThreadPoolExecutor(max_workersworkers) as pool: for idx, item in enumerate(media_items): item dict(item) item[idx] idx tasks.append(pool.submit(download_one, item, username, session)) done_count 0 for future in as_completed(tasks): try: future.result() except Exception as exc: print(f任务失败: {exc}) continue done_count 1 if done_count % 50 0: print(f已完成 {done_count}/{len(media_items)})要点说明mode ab是追加写mode wb是覆盖写。服务器不支持 Range 时回退到全量下载避免文件错乱临时文件名加.part后缀防止下载中断后主进程误判文件完整ThreadPoolExecutor(max_workers8)控制并发度不需要额外引 gevent 或 asyncioiter_content的chunk_size是 256KB太小会频繁磁盘写太大会让进度感知延迟。4.3 参数调优并发数、超时与重试的关系参数建议值过大后果过小后果max_workers8~12触发 CDN 限流大量 429下载速度上不去timeout30s无响应连接等太久大文件容易误判超时chunk_size256KB进度不实时磁盘写频繁重试次数3 次拉长失败队列单次抖动就丢文件重试间隔2s 随机抖动总耗时长频繁重试加剧封锁核心经验是不要用同一个超时值应对所有文件。视频文件建议把timeout调到 60s图片保持 30s。如果博主历史帖子太多可以在parallel_download外层再包一层分批次循环每批次 500 个任务批次之间sleep(5)把长任务的整体节奏放慢降低触发风控的概率。4.4 目录结构与落盘规范downloads/ └── natgeo/ ├── image/ │ ├── CwXAabc123_0.jpg │ └── CwXAabc123_1.jpg └── video/ └── CwXAabc123_0.mp4code是帖子的短码全局唯一_0、_1表示多图帖的第几张。目录按类型分开方便后续做媒体库索引。脚本里os.makedirs(..., exist_okTrue)保证多级目录自动创建。这里有一个容易忽略的点Instagram 的 CDN URL 链接有效期大约 2 小时。你在第 2 章解析完 URL再执行第 4 章下载中间如果隔太久链接会失效并返回 404。所以解析和下载要放在同一个进程内不要拆成两个独立步骤保存成文本再下载。如果必须存文本就应把签名参数完整保留并且尽快消费。5. Instagram 限流与风控被 429 和 checkpoint 拦截之后的应对5.1 认识返回码200 不等于成功爬虫运行中最容易误导人的状态码就是 200。Instagram 在触发风控时经常返回 200但 JSON 里是{require_login: true}或者checkpoint提示。代码里不能只判断resp.status_code还要检查响应体的关键字def is_login_required(resp: requests.Response) - bool: if resp.status_code 200: text resp.text if login_required in text or checkpoint_required in text: return True return resp.status_code in (401, 403)状态码实际含义处理策略200HTML/JSON 正常返回进一步检查内容是否含登录要求206部分内容断点续传成功继续写入临时文件301/302重定向通常到登录页检查 Cookie 是否失效429请求过于频繁冷却等待指数退避403禁止访问停止当前批次换 UA 或冷却404用户或 URL 不存在跳过记录到日志5.2 请求频率控制每用户最小间隔Instagram 风控是按请求频率建模的单个账号的请求间隔不能太快。稳妥做法是每 500 个请求之间至少停顿 60 秒把整体速率控制在 8 个/秒以内。对单个用户主页的翻页请求间隔要更大建议每页之间sleep(2)。import random import time last_request_time 0.0 def polite_delay(min_interval: float 2.0): global last_request_time elapsed time.time() - last_request_time if elapsed min_interval: time.sleep(min_interval - elapsed random.uniform(0.3, 1.5)) last_request_time time.time()random.uniform(0.3, 1.5)的抖动是打破固定节律的关键。固定间隔反而是风控模型最容易识别的特征加了随机抖动后请求时间分布更接近真人浏览行为。我一般会在每次翻页、每次下载前都调用一次polite_delay这样即便某个模块中途抛异常频率控制也不会失效。5.3 遇到 checkpoint 时的处理路径checkpoint_required表示账号被临时验证这时继续请求只会不断弹验证码。三个可执行路径冷却停止脚本收到 checkpoint 后立即停止保留当前游标等待 10 到 15 分钟再用同 Cookie 重试人工过验证在浏览器里打开 Instagram确认「这是你本人吗」完成后重新复制 Cookie切换备用 Cookie一个 Cookie 被限流后换另一个已登录账号的 Cookie。通常先把第 1 条路径写进代码让脚本自动冷却冷却后如果还是 checkpoint再抛异常交给外层处理。不要设计成无限重试那只会让账号的验证级别从「验证码」升级到「封禁」。补充一个常见理解偏差throttled与checkpoint不是一回事。throttled是纯频率控制过几分钟自己恢复checkpoint是账号级安全验证必须人工介入。如果代码把两者混为一谈会导致本该自动恢复的任务也被挂起或者本该停下的任务还在无效刷新。5.4 UA 多样性与会话保持同一个 Cookie 配多个不同 UA 不会明显提升成功率反而会因设备指纹不一致触发风控。正确的做法是一个 Cookie 绑定一个 UA在Session初始化时固定下来。爬虫要模拟的是一次完整会话而不是一堆碎片化请求。如果是分布式场景不同机器上跑同一个账号的 Cookie 会导致单账号多设备登录触发登录验证。要遵循「一个账号配一台机器」的约束或者在不同机器上各配一个账号。这里也是Instagram_crawler在团队场景下要考虑的横向扩展问题与其并发抢一个账号的配额不如把博主列表按账号拆开各机器只处理属于自己的那批用户名互不干扰。6. Instagram 归档完整性校验与增量更新SHA-256 确保素材可用6.1 SHA-256 校验断点续传后的隐患第 4 章的断点续传逻辑中文件通过.part临时文件追加写入最后改名。如果网络在最后一次写入后异常断开os.rename不会执行临时文件停留在磁盘上。更隐蔽的情况是文件写完了但实际缺字节文件大小正常但解码不了。因此验证文件完整性不能只看大小import hashlib def sha256_of(filepath: str, chunk_size: int 1024 * 1024) - str: h hashlib.sha256() with open(filepath, rb) as f: for chunk in iter(lambda: f.read(chunk_size), b): h.update(chunk) return h.hexdigest()每次下载完调用一次sha256_of(filepath)与摘要文件里记录的哈希比对。比对失败就删除文件并重新下载。这个操作在单个文件 5MB 时耗时不到 0.1 秒成本可以忽略。6.2 用摘要文件记录归档状态把每次下载的元数据写入manifest.json下次运行时对照它决定哪些跳过import json from datetime import datetime MANIFEST_PATH downloads/manifest.json def load_manifest() - dict: try: with open(MANIFEST_PATH, encodingutf-8) as f: return json.load(f) except FileNotFoundError: return {files: {}} def mark_downloaded(manifest: dict, filepath: str, sha256: str, code: str): manifest[files][filepath] { sha256: sha256, code: code, downloaded_at: datetime.now().isoformat(), } def save_manifest(manifest: dict): with open(MANIFEST_PATH, w, encodingutf-8) as f: json.dump(manifest, f, ensure_asciiFalse, indent2)增量更新时先逐行遍历博主前 N 个帖子的shortcode和 manifest 里的已下载 code 取差集只对新帖子的媒体发起下载。这就是增量爬取的核心逻辑不是每次重爬全站而是只拉新增内容。场景判断方式动作文件已存在 哈希一致manifest 中有相同 code跳过文件已存在 哈希不一致manifest 哈希不同删除重下文件不存在manifest 无记录全量下载.part临时文件残留文件存在但仍是.part删除后重下6.3 最后的验收命令与代码健壮性在downloads目录下执行find . -name *.mp4 -o -name *.jpg | wc -l du -sh downloads/这两个命令分别给出文件总数和总大小。如果文件总数与第 4 章解析出的媒体列表数量一致说明这次抓取基本完整。更严格的做法是写一个校验脚本重新计算所有文件的 SHA-256和manifest.json里记录的比对不一致的列出清单。Instagram 的页面结构和接口签名不定期变化一旦代码跑不通优先检查_sharedData的字段名前缀XDT是否出现、Cookie 是否过期、以及video_versions数组长度是否退化到 1。把这三点做成独立的检查函数你的Instagram_crawler就不太容易在博主更新页面结构的第二天报废。本文还有配套的精品资源点击获取