Python爬取Instagram照片视频:从JSON解析到并发下载实战
2026/9/13 9:39:10 网站建设 项目流程

简介:一套聚焦 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 type="text/javascript">window._sharedData = {...}</script>。用户名、粉丝数、帖子数、照片视频的 CDN 地址,全部在这一串 JSON 里。爬虫的起点不是解析 HTML 标签,而是先把_sharedData抠出来。

另一个变化是,https://www.instagram.com/{username}/?__a=1这类匿名 JSON 接口已经被关闭,现在返回 404 或重定向到登录页。所以眼下最稳的链路是:抓桌面端 HTML → 找_sharedData→ 解析 JSON → 取媒体 URL。这也是Instagram_crawler内部最常见的处理顺序。

2.2 Python 环境安装与依赖清单

建议用虚拟环境隔离,避免污染系统 Python:

mkdir instagram_crawler && cd instagram_crawler python3 -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install requests==2.32.3 beautifulsoup4==4.12.3 lxml==5.2.1

三个库的分工:

  • requests:发 HTTP 请求,带上 Cookie、Headers 和参数;
  • beautifulsoup4+lxml:定位<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;q=0.9", } def fetch_shared_data(username: str) -> dict: url = f"https://www.instagram.com/{username}/" resp = requests.get(url, headers=HEADERS, timeout=15) resp.raise_for_status() soup = BeautifulSoup(resp.text, "lxml") # 找到包含 window._sharedData 的 script 标签 script_tag = soup.find( "script", text=lambda 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())

这段代码的关键是splitrsplit的配合:先按window._sharedData =切出 JSON 开头,再从尾部切掉;</script>,避开正则转义问题。拿到data之后,媒体列表的位置在data["entry_data"]["ProfilePage"][0]["graphql"]["user"]["edge_owner_to_timeline_media"]["edges"],每一条edge["node"]是一个帖子。

2.4 从 node 里解析照片和视频 URL

def extract_media(node: dict) -> list: media_list = [] if node.get("is_video"): # 视频:把 video_versions 按分辨率排序,拿最大版本 vv = node["video_versions"] vv_sorted = sorted(vv, key=lambda x: x["height"] * x["width"], reverse=True) 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, key=lambda 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的枚举值在不同时期不一样:XDTGraphImageXDTGraphVideoXDTGraphSidecar,如果抓到的结果里是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_pageFalse,说明已经翻完。

字段名位置含义
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:开头的内容,说明解析层级错了,要回nodeimage_versions2的嵌套关系里排查。

3. Instagram 登录态 Cookie 处理:把低清缩略图升级为原图原视频

3.1 为什么匿名抓取拿不到最高清素材

第 2 章拿到的图片 URL 虽然能下载,但分辨率通常被压到 1080px。博主手机上传原图是 1440px 甚至 4000px,匿名接口不提供。视频同样受限:匿名响应里的video_versions往往只有height=720一条,登录后能拿到多个版本。

原因是 Instagram 会按请求者的登录状态动态裁剪媒体版本列表。登录后客户端能拿到更完整的video_versions数组,所以只取url字段可能拿的不是最大版本。我的做法是把整个video_versionsheight排序,优先选择高度最大的 URL 下载,第 2 章的extract_media已经做了这件事。

3.2 用 requests.Session 维持登录态

不推荐在脚本里模拟用户名密码登录,Instagram 对登录接口有行为检测,密码错误几次就可能触发checkpoint_required。常见做法是:在浏览器里手动登录一次,从 DevTools 里复制 Cookie 请求头,写进脚本

import requests COOKIE_STR = ( "sessionid=123456%3Aabc123; " "csrftoken=XXXX; " "ds_user_id=123456; " "ig_did=ABC-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 里usernull。这时候直接去浏览器重新复制 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", encoding="utf-8") as f: json.dump( {"cookie": cookie_str, "saved_at": datetime.now().isoformat()}, f, ensure_ascii=False, indent=2, ) def load_cookie() -> str: if not os.path.exists(COOKIE_FILE): raise FileNotFoundError("缺少 cookie.json,请先在浏览器登录后复制 Cookie") with open(COOKIE_FILE, encoding="utf-8") as f: payload = json.load(f) saved_at = datetime.fromisoformat(payload["saved_at"]) if datetime.now() - saved_at > timedelta(days=5): print("警告:Cookie 已超过 5 天,可能失效") return payload["cookie"]

这套持久化机制把「登录一次、长期使用」落实到文件层。实际操作里,我一般把 Cookie 保存与加载封装成独立模块,爬虫主程序只依赖load_cookie(),换账号时自动重新加载。

3.4 私密账号与关注限制的边界

私密博主的主页在没有被关注时,返回 JSON 里user字段存在,但edge_owner_to_timeline_media为空。此时继续翻页只会换来限流。如果确实需要抓私密账号素材,必须用已关注该账号的 Cookie,并且抓取时加上?hl=en参数避免非英文语言环境下的字段歧义。

账号状态匿名请求返回登录请求返回
公开账号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_ok=True) # 已存在且大小大于 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": f"bytes={downloaded}-"} with session.get(url, headers=headers, stream=True, timeout=30) 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_size=1024 * 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_workers=workers) 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_workers=8)控制并发度,不需要额外引 gevent 或 asyncio;
  • iter_contentchunk_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.mp4

code是帖子的短码,全局唯一,_0_1表示多图帖的第几张。目录按类型分开,方便后续做媒体库索引。脚本里os.makedirs(..., exist_ok=True)保证多级目录自动创建。

这里有一个容易忽略的点: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表示账号被临时验证,这时继续请求只会不断弹验证码。三个可执行路径:

  1. 冷却停止:脚本收到 checkpoint 后立即停止,保留当前游标,等待 10 到 15 分钟再用同 Cookie 重试;
  2. 人工过验证:在浏览器里打开 Instagram,确认「这是你本人吗」,完成后重新复制 Cookie;
  3. 切换备用 Cookie:一个 Cookie 被限流后,换另一个已登录账号的 Cookie。

通常先把第 1 条路径写进代码,让脚本自动冷却,冷却后如果还是 checkpoint,再抛异常交给外层处理。不要设计成无限重试,那只会让账号的验证级别从「验证码」升级到「封禁」。

补充一个常见理解偏差:throttledcheckpoint不是一回事。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, encoding="utf-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", encoding="utf-8") as f: json.dump(manifest, f, ensure_ascii=False, indent=2)

增量更新时,先逐行遍历博主前 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就不太容易在博主更新页面结构的第二天报废。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询