""" Session — 兼容 requests.Session 的米普代理客户端 用法与 requests.Session 一致: 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.proxies = "res_mipu_bookxWdyg" resp = s.get("https://example.com") resp = s.post(url, data={"key": "val"}, headers={...}) resp.raise_for_status() data = resp.json() """ import json as _json import time from urllib.parse import urlencode import requests as _requests from .exceptions import APIError, PollTimeoutError from .response import Response class Session: """兼容 requests.Session 的米普代理客户端。 Args: base_url: 米普 API 地址 (如 "https://api-eu.mipuyun.com") client_key: 客户端 key client_secret: 客户端 secret mode: "agent" → /unlocker/agent | "request" → /unlocker/request agent: Agent 编码 (mode="agent" 时必传) async_mode: 异步轮询 (默认 True) poll_interval: 轮询间隔秒 (默认 3) poll_max: 最大轮询次数 (默认 20) submit_timeout:提交超时秒 (默认 60) poll_timeout: 轮询超时 (connect, read) (默认 (5, 30)) """ def __init__( self, base_url, client_key, client_secret, mode="agent", agent=None, async_mode=True, poll_interval=3, poll_max=20, submit_timeout=60, poll_timeout=(5, 30), ): self.base_url = base_url.rstrip("/") self.client_key = client_key self.client_secret = client_secret if mode not in ("agent", "request"): raise ValueError(f"mode 必须是 'agent' 或 'request'，收到: {mode!r}") if mode == "agent" and not agent: raise ValueError("mode='agent' 时 agent 不能为空") self.mode = mode self._api_path = "/unlocker/agent" if mode == "agent" else "/unlocker/request" self.agent = agent or None self.async_mode = async_mode self.poll_interval = poll_interval self.poll_max = poll_max self.submit_timeout = submit_timeout self.poll_timeout = poll_timeout self.proxies = None self.headers = {} self.session_id = None self._http = _requests.Session() # ── 核心请求方法 ──────────────────────────────────────────────── def request(self, method, url, *, params=None, data=None, json=None, headers=None, content_type=None, proxy=None): """发送请求（兼容 requests.Session.request）。 Args: method: HTTP 方法 url: 目标 URL params: URL 查询参数 (dict) data: 请求体 (dict → form-encoded, str → raw) json: 请求体 (dict → JSON) headers: 请求头 (dict) content_type: Content-Type 覆盖 proxy: 本次请求代理覆盖 Returns: Response """ proxy = proxy or self.proxies if not proxy: raise ValueError("未设置 proxy，请通过 session.proxies = 'xxx' 或 request(proxy='xxx') 指定") # 拼接 params target_url = url if params: sep = "&" if "?" in target_url else "?" target_url = target_url + sep + urlencode(params) # 序列化 body body = "" ct = content_type if json is not None: body = _json.dumps(json, ensure_ascii=False) ct = ct or "application/json" elif data is not None: if isinstance(data, dict): body = urlencode(data) ct = ct or "application/x-www-form-urlencoded" else: body = str(data) # 合并 headers: session 默认 + 本次传入 merged_headers = {**self.headers, **(headers or {})} # 构建 payload payload = { "url": target_url, "method": method.upper(), "proxy": proxy, "async": self.async_mode, } if self.mode == "agent": payload["agent"] = self.agent if self.session_id: payload["sessionId"] = self.session_id if body: payload["body"] = body if ct: payload["contentType"] = ct if merged_headers: payload["headers"] = merged_headers # 调用米普 API api_url = f"{self.base_url}{self._api_path}" hdrs = { "Content-Type": "application/json", "client-key": self.client_key, "client-secret": self.client_secret, } resp = self._http.post(api_url, json=payload, headers=hdrs, timeout=self.submit_timeout) api_resp = resp.json() code = api_resp.get("code") # 异步轮询 if code == 202 and self.async_mode: agent_request_id = api_resp.get("agentRequestId", "") time.sleep(2) api_resp = self._poll_result(agent_request_id) code = api_resp.get("code") # 检查错误 if code != 0: raise APIError(api_resp) # 锁定 sessionId sid = api_resp.get("sessionId", "") if sid and not self.session_id: self.session_id = sid return Response( status_code=api_resp.get("statusCode", 0), text=api_resp.get("responseBody", ""), headers=api_resp.get("responseHeaders", {}), url=target_url, raw=api_resp, ) def get(self, url, **kwargs): return self.request("GET", url, **kwargs) def post(self, url, **kwargs): return self.request("POST", url, **kwargs) def put(self, url, **kwargs): return self.request("PUT", url, **kwargs) def delete(self, url, **kwargs): return self.request("DELETE", url, **kwargs) def head(self, url, **kwargs): return self.request("HEAD", url, **kwargs) def options(self, url, **kwargs): return self.request("OPTIONS", url, **kwargs) def patch(self, url, **kwargs): return self.request("PATCH", url, **kwargs) # ── 异步轮询 ──────────────────────────────────────────────────── def _poll_result(self, agent_request_id): query_url = f"{self.base_url}/async/query/{agent_request_id}" hdrs = { "client-key": self.client_key, "client-secret": self.client_secret, "Connection": "close", } for attempt in range(1, self.poll_max + 1): try: s = _requests.Session() s.headers.update(hdrs) resp = s.get(query_url, timeout=self.poll_timeout) s.close() result = resp.json() if result.get("code") == 0: return result except _requests.exceptions.RequestException: pass except ValueError: pass time.sleep(self.poll_interval) raise PollTimeoutError(agent_request_id, self.poll_max) # ── 生命周期 ──────────────────────────────────────────────────── def close(self): self._http.close() def __enter__(self): return self def __exit__(self, *args): self.close() def __repr__(self): mode_info = f"agent={self.agent!r}" if self.mode == "agent" else "mode=request" return (f"") def session(base_url, client_key, client_secret, **kwargs): """创建 Session 实例（与 requests.session() 用法一致）。 Args: base_url: 米普 API 地址 client_key: 客户端 key client_secret: 客户端 secret **kwargs: 传递给 Session 的其他参数 (mode, agent, async_mode 等) Returns: Session """ return Session(base_url, client_key, client_secret, **kwargs)