spider_work/deca_auto_spider/common/isms_client.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/09/09
"""爱接码 i-sms.app 接码平台客户端（改编自官方 Python SDK 示例）。

用途：账号池自动补货时，通过接码平台拿手机号 + 轮询收得卡短信验证码，实现「登录即注册」全自动。
来源：https://github.com/i-sms-app/i-sms-api-sdk-examples/tree/main/python （标准库 urllib，无第三方依赖）。

鉴权：请求头 X-API-KEY（api_key 建议从环境变量 ISMS_API_KEY 读，勿硬编码/提交公开仓库）。
调用流程：search_projects → get_number → get_sms(轮询) → release_number。
"""
import os
import json
import time
import urllib.parse
import urllib.request
import urllib.error

ISMS_BASE = "https://www.i-sms.app"     # 接码平台端点
POLL_INTERVAL_SEC = 5                    # 轮询验证码间隔（官方建议 ≥5s，过快触发 WAF）
POLL_MAX_SEC = 90                        # 轮询验证码最长等待（收不到就快换号，少浪费取号费；官方建议 60~180s）
# 必带浏览器 UA：urllib 默认 UA(Python-urllib/x.y) 会被爱接码 WAF 拦成 403（2026/09/09 实测）
_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"


class ISmsClient:
    """爱接码接码平台 API 客户端：搜项目 / 取号 / 收验证码 / 释放号 / 查余额。"""

    def __init__(self, api_key: str = None, base_url: str = ISMS_BASE, timeout: int = 20):
        """初始化客户端。

        Args:
            api_key (str, optional): 接码平台 API Key；None 时从环境变量 ISMS_API_KEY 读。Defaults to None。
            base_url (str, optional): 接口端点。Defaults to ISMS_BASE。
            timeout (int, optional): 单次请求超时秒数。Defaults to 20。

        Raises:
            ValueError: api_key 缺失（既未传参也无 ISMS_API_KEY 环境变量）时抛出。
        """
        self.api_key = api_key or os.environ.get("ISMS_API_KEY")
        if not self.api_key:
            raise ValueError("缺少接码平台 api_key（传参或设环境变量 ISMS_API_KEY）")
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    def search_projects(self, keyword: str) -> dict:
        """按关键词搜索接码项目（拿 project_id/name/token）。

        Args:
            keyword (str): 项目关键词（如 "得卡" / "DECA" / "decalive"）。

        Returns:
            dict: 平台响应 JSON，成功时 data 为项目列表，每项含 project_id/name/token。
        """
        return self._get("/api/v2/projects", {"keyword": keyword})

    def get_number(self, project_id, project_name, project_token, quantity: int = 1,
                   phone: str = None, province: str = None, carrier: str = None,
                   ascription: int = None) -> dict:
        """获取一个（或多个）可用手机号。

        Args:
            project_id: 项目 ID（来自 search_projects）。
            project_name: 项目名（来自 search_projects，需与 id/token 匹配）。
            project_token: 项目 V2 安全 Token（来自 search_projects）。
            quantity (int, optional): 取号数量（1~10）。Defaults to 1。
            phone (str, optional): 指定手机号（一般留空由平台分配）。Defaults to None。
            province (str, optional): 省份代码（见 README，留空不限）。Defaults to None。
            carrier (str, optional): 运营商代码（CMCC/CUCC/CTCC…，留空不限）。Defaults to None。
            ascription (int, optional): 卡类型 1=虚拟卡 / 2=实体卡（强风控平台建议 2）。Defaults to None。

        Returns:
            dict: 平台响应 JSON，成功时 data[i] 含 number/orderId，顶层含 balance。
        """
        return self._get("/api/v2/get_number", {
            "project_id": project_id, "project_name": project_name,
            "project_token": project_token, "quantity": quantity, "phone": phone,
            "province": province, "carrier": carrier, "ascription": ascription})

    def get_sms(self, order_id=None, phone_number=None, project_id=None) -> dict:
        """获取某订单收到的短信验证码（单次查询，未到时 success=False）。

        Args:
            order_id (optional): 取号返回的订单 ID（首选）。Defaults to None。
            phone_number (optional): 手机号（备用定位）。Defaults to None。
            project_id (optional): 项目 ID（备用定位）。Defaults to None。

        Returns:
            dict: 平台响应 JSON，成功时含 sms_code/sms_content。
        """
        return self._get("/api/v1/get_sms", {
            "order_id": order_id, "phone_number": phone_number, "project_id": project_id})

    def poll_sms(self, order_id, interval: int = POLL_INTERVAL_SEC, max_sec: int = POLL_MAX_SEC,
                 log=None) -> str | None:
        """按固定间隔轮询验证码直到拿到或超时（对 get_sms 的封装）。

        Args:
            order_id: 取号返回的订单 ID。
            interval (int, optional): 轮询间隔秒。Defaults to POLL_INTERVAL_SEC(5)。
            max_sec (int, optional): 最长等待秒。Defaults to POLL_MAX_SEC(180)。
            log (optional): 日志对象。Defaults to None。

        Returns:
            str | None: 验证码字符串；超时未收到返回 None。
        """
        deadline = time.time() + max_sec
        while time.time() < deadline:
            resp = self.get_sms(order_id=order_id)
            if resp.get("success") and resp.get("sms_code"):
                return str(resp["sms_code"])
            time.sleep(interval)
        if log:
            log.warning(f"[接码] 订单 {order_id} 轮询 {max_sec}s 未收到验证码")
        return None

    def release_number(self, order_id=None, phone_number=None, project_id=None) -> dict:
        """释放不再使用的号码（用完及时调，降资源占用）。

        Args:
            order_id (optional): 订单 ID（首选）。Defaults to None。
            phone_number (optional): 手机号。Defaults to None。
            project_id (optional): 项目 ID。Defaults to None。

        Returns:
            dict: 平台响应 JSON。
        """
        return self._get("/api/v1/release_number", {
            "order_id": order_id, "phone_number": phone_number, "project_id": project_id})

    def get_user_info(self) -> dict:
        """查询账户信息与余额。

        Returns:
            dict: 平台响应 JSON（含余额，用于补货前判断额度是否充足）。
        """
        return self._get("/api/v1/user/info")

    def _get(self, path: str, params: dict = None) -> dict:
        """内部：发 GET 请求并解析 JSON（HTTP 错误也返回带 http_status 的 JSON，不抛异常）。

        Args:
            path (str): 接口路径。
            params (dict, optional): query 参数（None/"" 值自动剔除）。Defaults to None。

        Returns:
            dict: 响应 JSON；HTTP 错误时返回 {success:False, ..., http_status:code}。
        """
        query = {k: v for k, v in (params or {}).items() if v is not None and v != ""}
        url = f"{self.base_url}{path}"
        if query:
            url = f"{url}?{urllib.parse.urlencode(query)}"
        req = urllib.request.Request(url, headers={
            "X-API-KEY": self.api_key, "User-Agent": _UA, "Accept": "application/json"})
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8")
            try:
                payload = json.loads(body)
            except json.JSONDecodeError:
                payload = {"success": False, "message": body}
            payload["http_status"] = exc.code
            return payload