1
GET/POST https://api.haozhuma.com/sms/?api=<接口名>&<参数...>

1
sequenceDiagram
2
    participant App as 业务程序
3
    participant HZM as 好芝麻平台
4
    participant Biz as 目标业务系统
5

6
    App->>HZM: api=login 换取 token，只做一次并缓存
7
    HZM-->>App: token
8
    App->>HZM: api=getPhone&sid=项目ID
9
    HZM-->>App: phone / sp / phone_gsd
10
    App->>Biz: 用该号码触发发送短信
11
    loop 每 15 秒查询一次，最长 3 分钟
12
        App->>HZM: api=getMessage&phone=号码
13
        HZM-->>App: 未收到时 yzm 为空
14
    end
15
    alt 收到验证码
16
        App->>Biz: 提交验证码完成流程
17
    else 超时仍无验证码
18
        App->>HZM: api=addBlacklist&phone=号码，然后换号重试
19
    end

1
# -*- coding: utf-8 -*-
2
"""好芝麻接码平台最小可用客户端。"""
3
import json
4
import os
5
import re
6
import time
7

8
import httpx
9

10
SERVERS = ["https://api.haozhuma.com", "https://api.haozhuyun.com"]
11
ISP = {1: "移动", 5: "联通", 9: "电信", 14: "广电", 16: "虚商"}
12
CONFIG_FILE = "hzm_config.json"
13

14

15
class HzmError(Exception):
16
    pass
17

18

19
class HzmClient:
20
    def __init__(self, config_file: str = CONFIG_FILE, timeout: float = 20.0):
21
        self.config_file = config_file
22
        cfg = {}
23
        if os.path.exists(config_file):
24
            try:
25
                cfg = json.load(open(config_file, encoding="utf-8"))
26
            except Exception:
27
                cfg = {}
28
        self.user = (cfg.get("user") or os.environ.get("HZM_USER") or "").strip()
29
        self.password = (cfg.get("pass") or os.environ.get("HZM_PASS") or "").strip()
30
        self.sid = str(cfg.get("sid") or os.environ.get("HZM_SID") or "").strip()
31
        self.server = (cfg.get("server") or SERVERS[0]).rstrip("/")
32
        self._token = (cfg.get("token") or "").strip()
33
        self.http = httpx.Client(timeout=timeout)
34

35
    # ---------- 底层 ----------
36
    @staticmethod
37
    def _ok(code) -> bool:
38
        # 各接口成功码不统一，0 和 200 都算成功
39
        return str(code) in ("0", "200")
40

41
    def _save_token(self, token: str) -> None:
42
        self._token = token
43
        cfg = {}
44
        if os.path.exists(self.config_file):
45
            try:
46
                cfg = json.load(open(self.config_file, encoding="utf-8"))
47
            except Exception:
48
                cfg = {}
49
        cfg["token"] = token
50
        json.dump(cfg, open(self.config_file, "w", encoding="utf-8"),
51
                  ensure_ascii=False, indent=2)
52

53
    def _call(self, api: str, params: dict) -> dict:
54
        params = {"api": api, **params}
55
        last_err = None
56
        # 主线路失败自动切备用线路
57
        for base in [self.server] + [s for s in SERVERS if s.rstrip("/") != self.server]:
58
            try:
59
                r = self.http.get(base + "/sms/", params=params,
60
                                  headers={"User-Agent": "hzm-client/1.0"})
61
                data = r.json()
62
                self.server = base.rstrip("/")
63
                return data
64
            except Exception as e:
65
                last_err = e
66
        raise HzmError(f"请求失败({api}): {last_err}")
67

68
    # ---------- 基础接口 ----------
69
    def login(self, force: bool = False) -> str:
70
        """token 是固定值，缓存复用，不要每次取号都登录。"""
71
        if self._token and not force:
72
            return self._token
73
        if not self.user or not self.password:
74
            raise HzmError("缺少凭据：请配置 user/pass 或设置 HZM_USER/HZM_PASS")
75
        r = self._call("login", {"user": self.user, "pass": self.password})
76
        if not self._ok(r.get("code")):
77
            raise HzmError(f"登录失败: {r.get('msg')}")
78
        self._save_token(r["token"])
79
        return self._token
80

81
    def get_summary(self) -> dict:
82
        r = self._call("getSummary", {"token": self.login()})
83
        if not self._ok(r.get("code")):
84
            raise HzmError(f"查询余额失败: {r.get('msg')}")
85
        return r
86

87
    def get_phone(self, isp: int | None = None, exclude: str | None = None) -> dict:
88
        p = {"token": self.login(), "sid": self.sid}
89
        if isp:
90
            p["isp"] = isp
91
        if exclude:
92
            p["exclude"] = exclude          # 多号段用 | 连接，如 "1380|1580"
93
        r = self._call("getPhone", p)
94
        if not self._ok(r.get("code")):
95
            raise HzmError(f"取号失败: {r.get('msg')}")
96
        return r
97

98
    def occupy(self, phone: str) -> dict:
99
        """复用旧号：必须先占用，否则读码会提示没有权限。"""
100
        r = self._call("getPhone",
101
                       {"token": self.login(), "sid": self.sid, "phone": phone})
102
        if not self._ok(r.get("code")):
103
            raise HzmError(f"占用号码 {phone} 失败: {r.get('msg')}")
104
        return r
105

106
    def get_message(self, phone: str) -> dict | None:
107
        r = self._call("getMessage",
108
                       {"token": self.login(), "sid": self.sid, "phone": phone})
109
        return r if self._ok(r.get("code")) else None
110

111
    def release(self, phone: str) -> None:
112
        """释放：下次取号还可能再出现。"""
113
        self._call("cancelRecv",
114
                   {"token": self.login(), "sid": self.sid, "phone": phone})
115

116
    def release_all(self) -> None:
117
        self._call("cancelAllRecv", {"token": self.login()})
118

119
    def blacklist(self, phone: str) -> None:
120
        """拉黑：该号码不再分配。"""
121
        self._call("addBlacklist",
122
                   {"token": self.login(), "sid": self.sid, "phone": phone})
123

124
    # ---------- 高层封装 ----------
125
    @staticmethod
126
    def extract_yzm(resp: dict) -> str:
127
        yzm = str(resp.get("yzm") or "").strip()
128
        if yzm:
129
            return yzm
130
        sms = resp.get("sms") or ""
131
        m = re.search(r"\b(\d{4,8})\b", sms) or re.search(r"(\d{4,8})", sms)
132
        return m.group(1) if m else ""
133

134
    def wait_code(self, phone: str, timeout: float = 180.0,
135
                  interval: float = 15.0) -> str | None:
136
        """每 15 秒读一次，最长 3 分钟。"""
137
        deadline = time.time() + timeout
138
        while time.time() < deadline:
139
            resp = self.get_message(phone)
140
            yzm = self.extract_yzm(resp) if resp else ""
141
            if yzm:
142
                return yzm
143
            remain = deadline - time.time()
144
            if remain <= 0:
145
                break
146
            time.sleep(min(interval, remain))
147
        return None
148

149
    def auto_get_code(self, send_fn, max_numbers: int = 5,
150
                      code_timeout: float = 180.0, log=print):
151
        """取号 -> 触发业务短信 -> 收码；超时则拉黑换号。"""
152
        self.login()
153
        for i in range(1, max_numbers + 1):
154
            info = self.get_phone()
155
            phone = str(info["phone"])
156
            log(f"[{i}/{max_numbers}] 取号: {phone}（{info.get('sp')} {info.get('phone_gsd')}）")
157

158
            if not send_fn(phone):          # send_fn 由调用方实现：触发业务发短信
159
                log(f"发送失败，释放 {phone} 换号")
160
                self.release(phone)
161
                continue
162

163
            yzm = self.wait_code(phone, timeout=code_timeout)
164
            if yzm:
165
                return phone, yzm
166

167
            log(f"{code_timeout:.0f}s 无验证码，拉黑 {phone} 换号")
168
            self.blacklist(phone)
169
        raise HzmError(f"连续 {max_numbers} 个号码都没收到验证码，请检查 sid 是否正确")
170

171

172
if __name__ == "__main__":
173
    c = HzmClient()
174
    print("余额:", c.get_summary().get("money"))
175
    info = c.get_phone()
176
    print("测试取号:", info.get("phone"), info.get("sp"), info.get("phone_gsd"))
177
    c.release(info["phone"])
178
    print("已释放")