当前位置: 首页 > news >正文

用 Python 把“API 接口”当数据源——从找口子到落库的全流程实战

不碰网页 HTML,只扒干净漂亮的 JSON:一招走遍天下接口


一、为什么选择“只爬 API”

  1. 返回结构化 JSON,省去 DOM 解析烦恼

  2. 带宽小,速率快,反爬成本远低于 HTML 页面

  3. 很多站点 SPA 本身就靠公开 API 渲染,不违法也不违规

  4. 参数可复用,一套代码能“薅”多端(Web、H5、App)


二、找接口:3 个“一眼定位”技巧

1. 浏览器 F12 → Network → XHR/JSON 过滤器

刷新页面,按 Size 降序,优先挑json/api路径

2. 搜索关键字

全局搜token,sign,t=(时间戳)、appKey,快速锁定鉴权参数

3. 手机抓包(App 接口)

mitmproxy + 夜神模拟器,安装系统证书后,启动 App 下滑刷新,拿到带Content-Type: application/json的请求

小技巧:把接口curl直接复制到 Postman,参数一个不落,先跑通再编码


三、接口逆向 4 件套

名称工具作用
参数含义Postman 环境变量去繁就简,删掉无关头
签名逻辑Chrome 断点 / 解包定位sign = md5(stringA + key)
时间戳Pythontime.time()*1000保证t与服务器误差 <30 s
Token 有效期响应头Expires/ JWT payload提前 5 分钟刷新

四、完整实战:某天气 API(公开免费版)

目标:按城市编码获取 7 日天气,落库 MySQL,每日 06:00 自动跑

1. 接口信息

GET https://api.weather.com/v1/forecast/daily 参数: cityCode=101010100 (北京) lang=zh unit=m key=YOUR_KEY (免费申请)

2. 依赖库

pip install requests pydantic sqlalchemy pymysql python-dotenv schedule

3. 代码(可直接python main.py跑)

# -*- coding: utf-8 -*- import os, time, requests, schedule from datetime import date from typing import List from pydantic import BaseModel, Field from sqlalchemy import create_engine, Column, String, Float, Date from sqlalchemy.orm import declarative_base, sessionmaker # ---------- 1. 配置 ---------- BASE_URL = "https://api.weather.com/v1/forecast/daily" API_KEY = os.getenv("WEATHER_KEY") # 写 .env 里 CITY_CODE = "101010100" ENGINE = create_engine("mysql+pymysql://user:pwd@localhost:3306/weather?charset=utf8mb4") Session = sessionmaker(bind=ENGINE) Base = declarative_base() # ---------- 2. ORM ---------- class Forecast(Base): __tablename__ = "t_forecast" city = Column(String(20), primary_key=True) f_date = Column(Date, primary_key=True) max_temp = Column(Float) min_temp = Column(Float) cond = Column(String(20)) # ---------- 3. 数据模型 ---------- class Day(BaseModel): date: date tmp_max: float = Field(alias="max_temp") tmp_min: float = Field(alias="min_temp") cond_txt_d: str = Field(alias="cond") class Resp(BaseModel): daily: List[Day] # ---------- 4. 抓取 ---------- def crawl(): params = {"cityCode": CITY_CODE, "lang": "zh", "unit": "m", "key": API_KEY} r = requests.get(BASE_URL, params=params, timeout=10) r.raise_for_status() return Resp.parse_obj(r.json()).daily # ---------- 5. 落库 ---------- def save(data: List[Day]): with Session() as sess: for d in data: obj = Forecast(city="北京", f_date=d.date, max_temp=d.tmp_max, min_temp=d.tmp_min, cond=d.cond_txt_d) sess.merge(obj) # 存在则更新 sess.commit() # ---------- 6. 调度 ---------- def job(): save(crawl()) print(time.strftime("%F %T"), "抓取完成") schedule.every().day.at("06:00").do(job) if __name__ == "__main__": Base.metadata.create_all(ENGINE) # 首次建表 job() # 手动跑一次 while True: schedule.run_pending() time.sleep(1)

4. 运行结果

2025-04-18 06:00:02 抓取完成 mysql> select * from t_forecast limit 2; +------+------------+----------+----------+--------+ | city | f_date | max_temp | min_temp | cond | +------+------------+----------+----------+--------+ | 北京 | 2025-04-18 | 24.0 | 11.0 | 晴 | | 北京 | 2025-04-19 | 26.0 | 13.0 | 多云 | +------+------------+----------+----------+--------+

五、进阶:带签名的“非公开”接口案例

1. 签名规则(逆向结果)

def sign(params: dict, app_secret: str) -> str: text = "&".join([f"{k}={params[k]}" for k in sorted(params)]) + f"@{app_secret}" return hashlib.md5(text.encode()).hexdigest().upper()

2. 自动刷新 Token

def get_token() -> str: r = requests.post("https://x.com/api/refresh", json={"refresh": REFRESH}) return r.json()["data"]["access_token"]

3. 重试 / 限流 / 降级

from tenacity import retry, stop_after_attempt, wait_random @retry(stop=stop_after_attempt(5), wait=wait_random(1, 3)) def fetch_with_sign(params): params["t"] = int(time.time()*1000) params["sign"] = sign(params, APP_SECRET) r = requests.get(API_URL, params=params, headers={"Authorization": f"Bearer {get_token()}"}) if r.status_code == 429: raise Exception("Rate limit") # 触发重试 r.raise_for_status() return r.json()

六、合法与合规红线

  1. 只采“公开”接口——需要破解加密、拆 App 壳的,先让法务评估

  2. 遵守 robots.txt与《反不正当竞争法》,控制频率(建议 ≤ 1 QPS)

  3. 不碰个人信息——手机号、身份证、地址一律不落盘

  4. 设置黑名单——对方返回Retry-After403立即停爬并告警

  5. 标注数据来源,对内对外报表都加“本数据由××接口抓取,仅供参考”


七、常见错误速查表

异常根因修复
401 Unauthorizedtoken 过期自动刷新后再发
460 “非法签名”参数顺序/大小写错误打印待签字符串逐字符对比
空 JSON{}限流返回兜底体指数退避重试
SSLCertVerificationError公司抓包证书冲突verify=False+ 警告过滤

八、小结:把接口当“免费数据库”的正确姿势

  1. 先 Postman 跑通 → 2. Python 封装requests→ 3. Pydantic 做模型 → 4. SQLAlchemy 落库 → 5. 调度 + 重试 + 监控
    一条链下来,代码 <150 行,就能让“别人的接口”变成你报表里的实时数据。

如遇任何疑问或有进一步的需求,请随时与我私信或者评论联系。

http://www.cnnetsun.cn/news/3935.html

相关文章:

  • Wan2.2-T2V-A14B为何成为高端广告生成平台的核心基座?
  • 内存屏障-Volatile ,示例程序
  • 高阶函数之-数据分组的思考
  • Wan2.2-T2V-5B能否理解‘慢动作’‘快进’等时间修饰词?
  • 终极CotEditor开源贡献完整指南:从入门到精通
  • 千万别信!留学生求职辅导真的管用吗?
  • ESD保护器件基础选型指南:从参数到实战
  • 系统级ESD防护设计:超越单一器件的思考
  • PHP实现电商GraphQL API的8个避坑指南(资深架构师亲授)
  • 安全加固后SSH无法远程登陆的主要原因
  • 解锁心理占星认证之路:为何新月文化的国际认证成为副业发展的首选平台?
  • FOC控制(仿真+硬件+软件)
  • 智能科学毕设简单的课题建议
  • Wan2.2-T2V-A14B如何识别并规避敏感内容生成?
  • Wan2.2-T2V-5B在智能客服视频回复系统中的集成构想
  • Wan2.2-T2V-5B支持与大语言模型联动生成脚本+视频
  • 惊呆了!用原生JS实现拖放,竟比你想象的简单100倍!
  • JavaScript Notifications API:告别alert弹窗,开启沉浸式用户体验革命!
  • Wan2.2-T2V-A14B支持昼夜交替与时间流逝效果模拟
  • WanVideo_comfy:革命性AI视频创作框架深度解析
  • 分布式系统元数据高可用:Apache Doris如何实现99.99%可用性保障
  • 超800万罚单再敲警钟:合规无小事,智慧法务成企业“必答题”
  • 计算机毕设java高校宿舍管理系统 基于Java的高校宿舍信息化管理平台设计与实现 Java技术驱动的高校宿舍管理系统开发与应用
  • 强化学习7倍速提速方法:从单核蜗牛到多核猎豹
  • Python跨平台剪贴板操作终极指南:快速实现文本复制粘贴
  • TinyMCE6处理OA系统word文档批量上传
  • 美股Tick数据获取方式汇总
  • 终极文件解压工具与二维码生成器:跨平台完整指南
  • 突破传统:WheelPicker轮盘组件如何重塑Android应用交互体验
  • Readest翻页动画系统:打造沉浸式数字阅读体验的完整指南