Claude 3.5动态推理压缩机制解析:中间层归零原理与工程实践
2026/6/9 7:25:53
不碰网页 HTML,只扒干净漂亮的 JSON:一招走遍天下接口
返回结构化 JSON,省去 DOM 解析烦恼
带宽小,速率快,反爬成本远低于 HTML 页面
很多站点 SPA 本身就靠公开 API 渲染,不违法也不违规
参数可复用,一套代码能“薅”多端(Web、H5、App)
刷新页面,按 Size 降序,优先挑json/api路径
全局搜token,sign,t=(时间戳)、appKey,快速锁定鉴权参数
mitmproxy + 夜神模拟器,安装系统证书后,启动 App 下滑刷新,拿到带Content-Type: application/json的请求
小技巧:把接口
curl直接复制到 Postman,参数一个不落,先跑通再编码
| 名称 | 工具 | 作用 |
|---|---|---|
| 参数含义 | Postman 环境变量 | 去繁就简,删掉无关头 |
| 签名逻辑 | Chrome 断点 / 解包 | 定位sign = md5(stringA + key) |
| 时间戳 | Pythontime.time()*1000 | 保证t与服务器误差 <30 s |
| Token 有效期 | 响应头Expires/ JWT payload | 提前 5 分钟刷新 |
目标:按城市编码获取 7 日天气,落库 MySQL,每日 06:00 自动跑
GET https://api.weather.com/v1/forecast/daily 参数: cityCode=101010100 (北京) lang=zh unit=m key=YOUR_KEY (免费申请)pip install requests pydantic sqlalchemy pymysql python-dotenv schedulepython 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)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 | 多云 | +------+------------+----------+----------+--------+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()def get_token() -> str: r = requests.post("https://x.com/api/refresh", json={"refresh": REFRESH}) return r.json()["data"]["access_token"]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()只采“公开”接口——需要破解加密、拆 App 壳的,先让法务评估
遵守 robots.txt与《反不正当竞争法》,控制频率(建议 ≤ 1 QPS)
不碰个人信息——手机号、身份证、地址一律不落盘
设置黑名单——对方返回Retry-After或403立即停爬并告警
标注数据来源,对内对外报表都加“本数据由××接口抓取,仅供参考”
| 异常 | 根因 | 修复 |
|---|---|---|
| 401 Unauthorized | token 过期 | 自动刷新后再发 |
| 460 “非法签名” | 参数顺序/大小写错误 | 打印待签字符串逐字符对比 |
空 JSON{} | 限流返回兜底体 | 指数退避重试 |
| SSLCertVerificationError | 公司抓包证书冲突 | verify=False+ 警告过滤 |
先 Postman 跑通 → 2. Python 封装requests→ 3. Pydantic 做模型 → 4. SQLAlchemy 落库 → 5. 调度 + 重试 + 监控
一条链下来,代码 <150 行,就能让“别人的接口”变成你报表里的实时数据。
如遇任何疑问或有进一步的需求,请随时与我私信或者评论联系。