# -*- coding: utf-8 -*- """700SU 新静态住宅IP客户接口示例。Python 3.9+，仅标准库。 直接运行只登录并查询余额；创建订单、支付和续费需调用方显式执行。 示例不进行写请求自动重试，不提供服务端没有实现的幂等保证。 """ import getpass import json import os import socket import time import urllib.error import urllib.parse import urllib.request from decimal import Decimal from pathlib import Path class APIError(RuntimeError): def __init__(self, status, code, data=None): self.status, self.code, self.data = status, code, data super().__init__(f"HTTP {status}: {code}") class OutcomeUnknown(RuntimeError): """写请求结果不确定。保存原记录、查询原单，不可直接重复提交。""" class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None class Client: def __init__(self, token=None, timeout=90): self.base_url = "https://700su.com" self.token = token self.timeout = timeout self.opener = urllib.request.build_opener(NoRedirect()) def request(self, method, path, body=None, authenticated=True): headers = {"Accept": "application/json"} if authenticated: if not self.token: raise ValueError("请先登录或提供700SU客户Token") headers["Authorization"] = "Bearer " + self.token data = None if body is not None: headers["Content-Type"] = "application/json" data = json.dumps(body, ensure_ascii=False).encode("utf-8") req = urllib.request.Request(self.base_url + path, data=data, headers=headers, method=method) # 不为任何请求设置自动重试，尤其不能自动重发创建／支付操作。 try: with self.opener.open(req, timeout=self.timeout) as res: raw = res.read() except urllib.error.HTTPError as e: try: payload = json.loads(e.read()) except (ValueError, UnicodeDecodeError): payload = {} if method != "GET" and e.code >= 500: raise OutcomeUnknown("写请求返回5xx；先查原订单／批次，不能直接重发") from None raise APIError(e.code, payload.get("error", "http_error"), payload) from None except (urllib.error.URLError, TimeoutError, socket.timeout, OSError): if method != "GET": raise OutcomeUnknown("写请求连接中断或超时；先核对是否已创建或已扣款") from None raise APIError(0, "network_error") from None try: return json.loads(raw) except (ValueError, UnicodeDecodeError): if method != "GET": raise OutcomeUnknown("写请求响应无法解析；先核对原单，不能重复提交") from None raise APIError(0, "invalid_json") from None def login(self, email, password): result = self.request("POST", "/api/auth/login", {"email": email, "password": password}, False) self.token = result["token"] return result["user"] def account(self): return self.request("GET", "/api/users/me") def catalog(self): return self.request("GET", "/api/public/catalog", authenticated=False) def account_catalog(self): return self.request("GET", "/api/inventory/availability") def cidrs(self, product): fields = {k: product[k] for k in ("country", "city", "line", "ip_type")} return self.request("GET", "/api/static/available-cidrs?" + urllib.parse.urlencode(fields), authenticated=False) @staticmethod def _record(path, record, exclusive=False): flags = os.O_WRONLY | os.O_CREAT | (os.O_EXCL if exclusive else os.O_TRUNC) fd = os.open(str(path), flags, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(record, f, ensure_ascii=False, indent=2) f.flush() os.fsync(f.fileno()) def create_order_once(self, product, state_file, quantity=1, protocol="socks5", cidr_id=None): """每次业务购买使用一个持久化记录文件；文件存在即拒绝再创建。 文件不能代替服务端幂等键。异常时保留文件并人工核对；不要删除后盲目重跑。 多服务器接入需由调用方以数据库唯一键协调业务请求。 """ if type(quantity) is not int or not 1 <= quantity <= 500: raise ValueError("quantity必须是1–500的整数") if protocol not in ("http", "socks5"): raise ValueError("protocol只能为http或socks5") if product.get("ip_type") != "static_residential": raise ValueError("此示例仅用于目录中的静态住宅商品") body = {k: product[k] for k in ("country", "city", "line", "ip_type")} body.update(quantity=quantity, cycle_days=30, protocol=protocol) if cidr_id is not None: body["selected_cidr_id"] = cidr_id path = Path(state_file) record = {"stage": "creation_started", "request": body} self._record(path, record, exclusive=True) order = self.request("POST", "/api/orders", body) if not isinstance(order, dict) or not order.get("id") or not order.get("order_no"): raise OutcomeUnknown("创建响应缺少订单标识，请保留记录并核对订单列表") record.update(stage="created", order_id=order["id"], order_no=order["order_no"], amount_cny=order["amount_cny"]) self._record(path, record) return order def get_order(self, order_no): return self.request("GET", "/api/orders/" + urllib.parse.quote(str(order_no), safe=""))["order"] def orders(self): return self.request("GET", "/api/users/me/orders") @staticmethod def _check_budget(amount, max_amount_cny): amount, limit = Decimal(str(amount)), Decimal(str(max_amount_cny)) if not amount.is_finite() or not limit.is_finite() or amount <= 0 or amount > limit: raise ValueError("订单金额无效或超过明确设定的费用上限，停止支付") def pay_balance(self, order_no, max_amount_cny): order = self.get_order(order_no) if order["payment_status"] == "paid": return {"already": True, "order": order} if order["payment_status"] != "pending" or order["order_status"] != "pending" or order.get("quote_expired"): raise ValueError("当前订单不可支付，先核对订单状态及报价") self._check_budget(order["amount_cny"], max_amount_cny) return self.request("POST", "/api/orders/" + urllib.parse.quote(str(order_no), safe="") + "/pay-balance") def delivered_ips(self, order_id): return self.request("GET", f"/api/orders/id/{int(order_id)}/ips") def wait_delivery(self, order_no, wait_seconds=120): deadline = time.monotonic() + wait_seconds while True: order = self.get_order(order_no) if order.get("delivery_review_required"): raise RuntimeError("原订单需要核对，请保存订单号联系客服，不要另建单") if order["payment_status"] != "paid": raise RuntimeError("订单尚未付款，请核对原单") if order["order_status"] == "fulfilled": data = self.delivered_ips(order["id"]) ips = data.get("ips", []) if len(ips) != int(order["quantity"]): raise RuntimeError("交付数量不匹配，请核对原订单") return ips if order["order_status"] != "provisioning": raise RuntimeError("订单状态需要核对，请保存原单") if time.monotonic() >= deadline: raise TimeoutError("等待已结束但订单仍需查询；不得自动重新购买") time.sleep(min(5, max(0, deadline - time.monotonic()))) def identify_renewal(self, ip_item): body = {k: ip_item[k] for k in ("port", "username", "password")} body["ip"] = ip_item.get("connection_host") or ip_item["ip"] return self.request("POST", "/api/ip-renewals/identify", body) def create_renewal_once(self, ip_ids, state_file): if not ip_ids or len(ip_ids) > 500 or any(type(i) is not int or i <= 0 for i in ip_ids): raise ValueError("ip_ids需要1–500个有效整数ID") body = {"ip_ids": ip_ids, "cycle_days": 30} path = Path(state_file) self._record(path, {"stage": "creation_started", "request": body}, exclusive=True) result = self.request("POST", "/api/ip-renewals/orders", body) self._record(path, {"stage": "created", "batch_no": result["batch_no"], "total_cny": result["total_cny"], "request": body}) return result def get_renewal(self, batch_no): return self.request("GET", "/api/ip-renewals/orders/" + urllib.parse.quote(str(batch_no), safe="")) def pay_renewal_balance(self, batch_no, max_amount_cny): batch = self.get_renewal(batch_no) if batch["payment_status"] == "paid": return {"already": True, **batch} if not batch.get("can_pay"): raise ValueError("当前续费批次不可支付，请核对原批次") self._check_budget(batch["total_cny"], max_amount_cny) return self.request("POST", "/api/ip-renewals/orders/" + urllib.parse.quote(str(batch_no), safe="") + "/pay-balance") if __name__ == "__main__": client = Client() try: client.login(input("700SU邮箱：").strip(), getpass.getpass("700SU密码（不回显）：")) account = client.account() print("登录成功。当前余额：", account["balance_cny"], "元") print("本示例未创建订单、未扣款。购买和续费需在自己的代码中显式调用对应方法。") except APIError as e: print("请求未成功：", e) except OutcomeUnknown as e: print(e)