# mipu_requests — 从 requests 迁移指南 ## 快速对比 ```python # ── requests ────────────────────────────────────── import requests s = requests.Session() s.proxies = {"https": "http://proxy:8080"} resp = s.get("https://example.com/page") resp.raise_for_status() data = resp.json() # ── mipu_requests ───────────────────────────────── import mipu_requests s = mipu_requests.session( base_url="https://api-eu.mipuyun.com", client_key="your_client_key", client_secret="your_client_secret", agent="JQ", ) s.proxies = "res_mipu_bookxWdyg" resp = s.get("https://example.com/page") resp.raise_for_status() data = resp.json() ``` ## 安装 ```bash pip install mipu-requests ``` ## 创建 Session ```python import mipu_requests # Agent 模式 — 通过指定 agent 走 /unlocker/agent 端点 s = mipu_requests.session( base_url="https://api-eu.mipuyun.com", client_key="your_key", client_secret="your_secret", agent="JQ", ) # Request 模式 — 走 /unlocker/request 端点(无需 agent) s = mipu_requests.session( base_url="https://api-eu.mipuyun.com", client_key="your_key", client_secret="your_secret", mode="request", ) ``` ## 设置代理 与 `requests` 唯一**必须修改**的地方: ```python # requests — 传 dict s.proxies = {"http": "http://proxy:8080", "https": "http://proxy:8080"} # mipu_requests — 传字符串(米普 proxy 标识) s.proxies = "res_mipu_bookxWdyg" # 单次请求覆盖代理 resp = s.get(url, proxy="another_proxy") ``` ## 请求方法 完全一致,无需改动: ```python resp = s.get(url, params={"key": "value"}, headers={...}) resp = s.post(url, data={"key": "value"}, headers={...}) resp = s.post(url, json={"key": "value"}, headers={...}) resp = s.put(url, data=body, headers={...}) resp = s.delete(url, headers={...}) ``` ## Response 对象 属性和方法与 `requests.Response` 一致: ```python resp.status_code # int HTTP 状态码 resp.text # str 响应体文本 resp.content # bytes 响应体字节 resp.headers # dict 响应头 resp.ok # bool status_code 在 200-399 resp.url # str 请求 URL resp.encoding # str 编码(默认 utf-8) resp.cookies # dict 空 dict(兼容属性) resp.reason # str 状态码原因短语 resp.json() # 解析 JSON resp.raise_for_status() # >=400 抛异常 resp.iter_content(chunk_size) # 按块迭代 ``` ## 异常处理 ```python try: resp = s.get(url) resp.raise_for_status() except mipu_requests.HTTPError as e: print(f"目标站错误: {e.response.status_code}") except mipu_requests.APIError as e: print(f"API 错误: {e}") except mipu_requests.RequestException as e: print(f"请求异常: {e}") ``` ## 异步模式 默认开启异步轮询防止长连接断连: ```python # 默认异步 s = mipu_requests.session(...) # 关闭异步 s = mipu_requests.session(..., async_mode=False) # 自定义参数 s = mipu_requests.session( ..., poll_interval=3, poll_max=20, submit_timeout=60, poll_timeout=(5, 30), ) ``` ## 完整迁移示例 ```python # ── 迁移前 (requests) ──────────────────────────── import requests s = requests.Session() s.headers.update({"user-agent": "Mozilla/5.0"}) s.proxies = {"https": "http://proxy:8080"} resp = s.get("https://example.com/api", params={"page": 1}) resp.raise_for_status() data = resp.json() # ── 迁移后 (mipu_requests) ─────────────────────── import mipu_requests s = mipu_requests.session( base_url="https://api-eu.mipuyun.com", client_key="your_key", client_secret="your_secret", agent="JQ", ) s.headers = {"user-agent": "Mozilla/5.0"} s.proxies = "res_mipu_bookxWdyg" resp = s.get("https://example.com/api", params={"page": 1}) resp.raise_for_status() data = resp.json() ``` **迁移只需 2 步:** 1. 构造 `mipu_requests.session()` 时传入 `base_url`/`client_key`/`client_secret`/`agent` 2. `s.proxies = {...}` → `s.proxies = "proxy_name"` 其余代码无需改动。