1
def parse_entry(entry: str) -> str:
2
    """ip:port:user:pass -> 带账密的代理 URL；其余原样包 http://。"""
3
    entry = entry.strip()
4
    if "://" in entry:
5
        return entry
6
    parts = entry.split(":")
7
    if len(parts) == 4:
8
        ip, port, user, pwd = parts
9
        return f"http://{user}:{pwd}@{ip}:{port}"
10
    return f"http://{entry}"

1
import httpx
2

3
CHECK_URL = "https://ip.3322.net"
4

5

6
def check(proxy: str, test_url: str = CHECK_URL,
7
          timeout: float = 15.0) -> tuple[bool, str | None]:
8
    try:
9
        with httpx.Client(proxy=proxy, timeout=timeout) as c:
10
            r = c.get(test_url, headers={"User-Agent": "Mozilla/5.0"})
11
            r.raise_for_status()
12
            return True, r.text.strip()[:64]
13
    except Exception as e:
14
        return False, f"{type(e).__name__}: {e}"

1
import json
2
import os
3
import time
4

5
HISTORY_FILE = "used_exits.json"
6
FRESH_HOURS = 24.0
7

8

9
def load_history() -> dict:
10
    if os.path.exists(HISTORY_FILE):
11
        try:
12
            return json.load(open(HISTORY_FILE, encoding="utf-8"))
13
        except Exception:
14
            pass
15
    return {}
16

17

18
def recent_exits() -> set:
19
    """最近 24 小时内用过的出口 IP。"""
20
    cutoff = time.time() - FRESH_HOURS * 3600
21
    return {ip for ip, ts in load_history().items() if ts >= cutoff}
22

23

24
def mark_exit(exit_ip: str) -> None:
25
    hist = load_history()
26
    hist[exit_ip] = time.time()
27
    cutoff = time.time() - FRESH_HOURS * 3600
28
    hist = {ip: ts for ip, ts in hist.items() if ts >= cutoff}
29
    json.dump(hist, open(HISTORY_FILE, "w", encoding="utf-8"),
30
              ensure_ascii=False, indent=2)

1
def get_fresh_proxy(client, max_tries: int = 6, log=print):
2
    """提取一个出口最近未用过的可用代理，返回 (代理URL, 出口IP)。"""
3
    seen = set()
4
    for i in range(1, max_tries + 1):
5
        proxy = client.get_proxy()
6
        ok, info = client.check(proxy)
7
        if not ok:
8
            log(f"出口不可用（{info}），重新提取 ...")
9
            continue
10
        exit_ip = str(info).strip()
11
        if exit_ip in recent_exits() or exit_ip in seen:
12
            log(f"出口 {exit_ip} 最近用过（第 {i} 次），避开重取 ...")
13
            seen.add(exit_ip)
14
            continue
15
        mark_exit(exit_ip)
16
        return proxy, exit_ip
17
    raise RuntimeError(f"连续 {max_tries} 次未取到未用过的出口 IP")

1
flowchart TD
2
    A["读取配置: trade_no 与 key 或 api_link"] --> B{鉴权模式}
3
    B -->|whitelist| C[查白名单并补录本机公网 IP]
4
    B -->|fixed_link 或 userpass| D[跳过白名单]
5
    C --> E[提取代理]
6
    D --> E
7
    E --> F[走代理访问国内回显拿出口 IP]
8
    F --> G{可用且最近未用过}
9
    G -->|否| E
10
    G -->|是| H[记录出口 IP 并返回代理 URL]
11
    H --> I[业务请求携带该代理]

1
# -*- coding: utf-8 -*-
2
"""巨量HTTP 按量付费代理最小可用客户端。"""
3
import hashlib
4
import json
5
import os
6
import time
7

8
import httpx
9

10
ENDPOINT = "http://v2.api.juliangip.com/postpay/getips"
11
API_BASE = "http://v2.api.juliangip.com/postpay"
12
IP_ECHO_URLS = [
13
    "https://ip.3322.net",
14
    "https://api.ipify.org/?format=json",
15
    "http://members.3322.org/dyndns/getip",
16
]
17
CONFIG_FILE = "juliangip_config.json"
18
HISTORY_FILE = "used_exits.json"
19
FRESH_HOURS = 24.0
20

21

22
class JuliangError(Exception):
23
    pass
24

25

26
class JuliangClient:
27
    def __init__(self, config_file: str = CONFIG_FILE, timeout: float = 20.0):
28
        self.config_file = config_file
29
        cfg = {}
30
        if os.path.exists(config_file):
31
            try:
32
                cfg = json.load(open(config_file, encoding="utf-8"))
33
            except Exception:
34
                cfg = {}
35
        self.trade_no = str(cfg.get("trade_no") or os.environ.get("JLIP_TRADE_NO") or "").strip()
36
        self.key = str(cfg.get("key") or os.environ.get("JLIP_KEY") or "").strip()
37
        # fixed_link / userpass / whitelist
38
        self.auth_mode = cfg.get("auth_mode") or "fixed_link"
39
        self.api_link = (cfg.get("api_link") or "").strip()
40
        self.pt = int(cfg.get("pt", 1))              # 1=HTTP 2=SOCKS
41
        self.filter = int(cfg.get("filter", 1))      # 1=过滤当天已提取
42
        self.auto_white = int(cfg.get("auto_white", 1))
43
        self.http = httpx.Client(timeout=timeout)
44

45
    # ---------- 签名 ----------
46
    @staticmethod
47
    def sign(params: dict, key: str) -> str:
48
        raw = "&".join(f"{k}={v}" for k, v in sorted(params.items())) + f"&key={key}"
49
        return hashlib.md5(raw.encode("utf-8")).hexdigest()
50

51
    def _api_get(self, api: str, params: dict) -> dict:
52
        params = {"trade_no": self.trade_no, **params}
53
        params["sign"] = self.sign(params, self.key)
54
        try:
55
            r = self.http.get(f"{API_BASE}/{api}", params=params)
56
        except Exception as e:
57
            raise JuliangError(f"{api} 请求失败: {type(e).__name__}: {e}")
58
        try:
59
            data = r.json()
60
        except Exception:
61
            raise JuliangError(f"{api} 返回非 JSON: {r.text[:200]}")
62
        if data.get("code") != 200:
63
            raise JuliangError(f"{api} 失败: {data.get('msg')} (code={data.get('code')})")
64
        return data.get("data") or {}
65

66
    # ---------- 公网 IP 与白名单 ----------
67
    def public_ip(self, timeout: float = 10.0) -> str:
68
        """多回显源依次尝试，避免单一海外源不可达导致误判。"""
69
        last_err = None
70
        for url in IP_ECHO_URLS:
71
            try:
72
                r = self.http.get(url, timeout=timeout)
73
                txt = r.text.strip()
74
                if "format=json" in url:
75
                    txt = str((r.json() or {}).get("ip") or "").strip()
76
                if txt:
77
                    return txt.split()[0]
78
            except Exception as e:
79
                last_err = e
80
        raise JuliangError(f"查询本机公网 IP 失败: {last_err}")
81

82
    def get_whiteips(self) -> dict:
83
        return self._api_get("getwhiteip", {})
84

85
    def set_whiteips(self, ips: list) -> dict:
86
        return self._api_get("setwhiteip", {"ips": ",".join(ips)})
87

88
    def ensure_whitelist(self) -> str:
89
        my_ip = self.public_ip()
90
        info = self.get_whiteips()
91
        current = [str(x) for x in (info.get("current_white_ip") or [])]
92
        if my_ip not in current:
93
            if int(info.get("surplus_white_ip_quantity", 0)) <= 0:
94
                raise JuliangError(f"白名单名额已满，且本机 IP {my_ip} 不在其中，请先清理")
95
            self.set_whiteips([my_ip])
96
        return my_ip
97

98
    # ---------- 提取 ----------
99
    @staticmethod
100
    def parse_entry(entry: str) -> str:
101
        entry = entry.strip()
102
        if "://" in entry:
103
            return entry
104
        parts = entry.split(":")
105
        if len(parts) == 4:                          # ip:port:user:pass
106
            ip, port, user, pwd = parts
107
            return f"http://{user}:{pwd}@{ip}:{port}"
108
        return f"http://{entry}"
109

110
    def get_proxy(self, num: int = 1) -> str:
111
        if self.auth_mode == "whitelist":
112
            self.ensure_whitelist()
113

114
        # 固定提取链接：签名已内置，直接 GET
115
        if self.api_link and self.auth_mode == "fixed_link":
116
            r = self.http.get(self.api_link, timeout=20)
117
            txt = r.text.strip()
118
            if txt.startswith("{"):
119
                data = json.loads(txt)
120
                if data.get("code") != 200:
121
                    raise JuliangError(f"提取失败: {data.get('msg')}")
122
                entries = list((data.get("data") or {}).get("proxy_list") or [])
123
            else:
124
                if txt.upper().startswith("ERROR"):
125
                    raise JuliangError(f"提取失败: {txt[:200]}")
126
                entries = [l.strip() for l in txt.splitlines() if l.strip()]
127
            if not entries:
128
                raise JuliangError("提取返回为空")
129
            return self.parse_entry(entries[0])
130

131
        # 自己拼参数 + 签名
132
        params = {
133
            "auto_white": self.auto_white,
134
            "filter": self.filter,
135
            "num": num,
136
            "pt": self.pt,
137
            "result_type": "json",
138
            "split": "1",
139
            "trade_no": self.trade_no,
140
        }
141
        params["sign"] = self.sign(params, self.key)
142
        r = self.http.get(ENDPOINT, params=params)
143
        data = r.json()
144
        if data.get("code") != 200:
145
            raise JuliangError(f"提取失败: {data.get('msg')} (code={data.get('code')})")
146
        entries = list((data.get("data") or {}).get("proxy_list") or [])
147
        if not entries:
148
            raise JuliangError("提取成功但列表为空")
149
        return self.parse_entry(entries[0])
150

151
    # ---------- 验证 ----------
152
    @staticmethod
153
    def check(proxy: str, test_url: str = IP_ECHO_URLS[0],
154
              timeout: float = 15.0) -> tuple:
155
        try:
156
            with httpx.Client(proxy=proxy, timeout=timeout) as c:
157
                r = c.get(test_url, headers={"User-Agent": "Mozilla/5.0"})
158
                r.raise_for_status()
159
                return True, r.text.strip()[:64]
160
        except Exception as e:
161
            return False, f"{type(e).__name__}: {e}"
162

163
    # ---------- 出口去重 ----------
164
    def _load_history(self) -> dict:
165
        if os.path.exists(HISTORY_FILE):
166
            try:
167
                return json.load(open(HISTORY_FILE, encoding="utf-8"))
168
            except Exception:
169
                pass
170
        return {}
171

172
    def recent_exits(self) -> set:
173
        cutoff = time.time() - FRESH_HOURS * 3600
174
        return {ip for ip, ts in self._load_history().items() if ts >= cutoff}
175

176
    def mark_exit(self, exit_ip: str) -> None:
177
        hist = self._load_history()
178
        hist[exit_ip] = time.time()
179
        cutoff = time.time() - FRESH_HOURS * 3600
180
        hist = {ip: ts for ip, ts in hist.items() if ts >= cutoff}
181
        json.dump(hist, open(HISTORY_FILE, "w", encoding="utf-8"),
182
                  ensure_ascii=False, indent=2)
183

184
    def get_fresh_proxy(self, max_tries: int = 6, log=print):
185
        """提取一个出口最近未用过的可用代理。"""
186
        seen = set()
187
        for i in range(1, max_tries + 1):
188
            proxy = self.get_proxy()
189
            ok, info = self.check(proxy)
190
            if not ok:
191
                log(f"[代理] 不可用（{info}），重新提取 ...")
192
                continue
193
            exit_ip = str(info).strip()
194
            if exit_ip in self.recent_exits() or exit_ip in seen:
195
                log(f"[代理] 出口 {exit_ip} 最近用过（第 {i} 次），避开重取 ...")
196
                seen.add(exit_ip)
197
                continue
198
            self.mark_exit(exit_ip)
199
            return proxy, exit_ip
200
        raise JuliangError(f"连续 {max_tries} 次未取到未用过的出口 IP")
201

202

203
if __name__ == "__main__":
204
    jp = JuliangClient()
205
    print("鉴权模式:", jp.auth_mode)
206
    proxy, exit_ip = jp.get_fresh_proxy()
207
    print("出口 IP:", exit_ip)
208
    print("代理地址:", proxy.split("@")[-1], "（含账密）" if "@" in proxy else "")