493 lines
17 KiB
Python
Executable File
493 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""批量购买套餐脚本：读取单列 CSV，逐资产调用后台订单接口购买同一套餐。
|
||
|
||
默认只执行预演。只有显式传入 --execute 时，才会调用 POST /api/admin/orders。
|
||
脚本仅使用 Python 标准库，不需要安装第三方依赖。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
from getpass import getpass
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.request import Request, urlopen
|
||
|
||
|
||
ORDER_PATH = "/api/admin/orders"
|
||
LOGIN_PATH = "/api/admin/login"
|
||
AUTH_ERROR_CODES = {1002, 1003, 1004}
|
||
HEADER_NAMES = {
|
||
"iccid",
|
||
"virtual_no",
|
||
"identifier",
|
||
"资产标识",
|
||
"虚拟号",
|
||
"iccid/虚拟号",
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AssetInput:
|
||
"""保存 CSV 中的资产标识及原始行号。"""
|
||
|
||
line_no: int
|
||
identifier: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class HTTPResult:
|
||
"""保存一次 HTTP 请求的响应信息。"""
|
||
|
||
status: int
|
||
body: dict[str, Any] | None
|
||
raw_body: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PurchaseOutcome:
|
||
"""保存单个资产的下单结果及结果文件行。"""
|
||
|
||
row: dict[str, object]
|
||
success: bool
|
||
http_status: int | None
|
||
code: int | str | None
|
||
message: str
|
||
order_no: str
|
||
|
||
|
||
class RequestFailedError(Exception):
|
||
"""表示请求尚未获得可解析的 HTTP 响应。"""
|
||
|
||
|
||
class AdminAPIClient:
|
||
"""调用后台认证和订单接口的轻量客户端。"""
|
||
|
||
def __init__(self, base_url: str, timeout: float) -> None:
|
||
self.base_url = base_url.rstrip("/")
|
||
self.timeout = timeout
|
||
|
||
def login(self, username: str, password: str) -> str:
|
||
"""使用后台账号登录并返回 Access Token。"""
|
||
result = self._post_json(
|
||
LOGIN_PATH,
|
||
{"username": username, "password": password, "device": "web"},
|
||
token=None,
|
||
)
|
||
code = response_code(result.body)
|
||
if not is_success(result.status, code):
|
||
raise RequestFailedError(
|
||
f"登录失败：HTTP {result.status}，code={display_value(code)}，"
|
||
f"msg={response_message(result.body, result.raw_body)}"
|
||
)
|
||
|
||
data = result.body.get("data") if result.body else None
|
||
token = data.get("access_token") if isinstance(data, dict) else None
|
||
if not isinstance(token, str) or not token.strip():
|
||
raise RequestFailedError("登录响应中缺少 data.access_token")
|
||
return token.strip()
|
||
|
||
def create_order(self, token: str, identifier: str, package_id: int) -> HTTPResult:
|
||
"""为单个资产创建钱包套餐订单。"""
|
||
return self._post_json(
|
||
ORDER_PATH,
|
||
{
|
||
"identifier": identifier,
|
||
"package_ids": [package_id],
|
||
"payment_method": "wallet",
|
||
},
|
||
token=token,
|
||
)
|
||
|
||
def _post_json(self, path: str, payload: dict[str, Any], token: str | None) -> HTTPResult:
|
||
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||
headers = {
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json",
|
||
"User-Agent": "junhong-batch-package-purchase/1.0",
|
||
}
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
|
||
request = Request(
|
||
url=self.base_url + path,
|
||
data=body,
|
||
headers=headers,
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urlopen(request, timeout=self.timeout) as response:
|
||
raw_body = response.read().decode("utf-8", errors="replace")
|
||
return HTTPResult(
|
||
status=response.status,
|
||
body=parse_json_object(raw_body),
|
||
raw_body=raw_body,
|
||
)
|
||
except HTTPError as exc:
|
||
raw_body = exc.read().decode("utf-8", errors="replace")
|
||
return HTTPResult(
|
||
status=exc.code,
|
||
body=parse_json_object(raw_body),
|
||
raw_body=raw_body,
|
||
)
|
||
except (URLError, TimeoutError, OSError) as exc:
|
||
raise RequestFailedError(f"请求失败：{exc}") from exc
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
"""解析命令行参数。"""
|
||
parser = argparse.ArgumentParser(
|
||
description="读取单列 CSV，逐资产调用 /api/admin/orders 购买同一套餐",
|
||
)
|
||
parser.add_argument(
|
||
"--base-url",
|
||
default=os.getenv("JUNHONG_ADMIN_BASE_URL", ""),
|
||
help="接口 Base URL，例如 https://cmp-api.example.com；也可使用 JUNHONG_ADMIN_BASE_URL",
|
||
)
|
||
parser.add_argument("--csv", required=True, help="单列资产 CSV 文件路径，首行可有表头")
|
||
parser.add_argument("--package-id", required=True, type=int, help="本批资产统一购买的套餐 ID")
|
||
parser.add_argument(
|
||
"--token",
|
||
default=os.getenv("JUNHONG_ADMIN_TOKEN", ""),
|
||
help="后台 Access Token；也可使用 JUNHONG_ADMIN_TOKEN",
|
||
)
|
||
parser.add_argument(
|
||
"--username",
|
||
default=os.getenv("JUNHONG_ADMIN_USERNAME", ""),
|
||
help="未提供 Token 时用于自动登录；也可使用 JUNHONG_ADMIN_USERNAME",
|
||
)
|
||
parser.add_argument(
|
||
"--password",
|
||
default=os.getenv("JUNHONG_ADMIN_PASSWORD", ""),
|
||
help="后台登录密码；建议使用 JUNHONG_ADMIN_PASSWORD，避免进入命令历史",
|
||
)
|
||
parser.add_argument("--output", default="", help="结果 CSV 路径；默认输出到输入文件同目录")
|
||
parser.add_argument("--timeout", type=float, default=30.0, help="单次请求超时秒数（默认 30）")
|
||
parser.add_argument("--interval", type=float, default=0.2, help="每次下单后的间隔秒数（默认 0.2）")
|
||
parser.add_argument(
|
||
"--execute",
|
||
action="store_true",
|
||
help="真实调用接口下单；不传时只校验 CSV 并预览请求",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def load_assets(csv_path: Path) -> list[AssetInput]:
|
||
"""读取单列 CSV，识别可选表头，并在下单前拦截重复资产。"""
|
||
if not csv_path.exists():
|
||
raise ValueError(f"找不到 CSV 文件：{csv_path}")
|
||
if not csv_path.is_file():
|
||
raise ValueError(f"CSV 路径不是文件：{csv_path}")
|
||
|
||
assets: list[AssetInput] = []
|
||
first_line_by_identifier: dict[str, int] = {}
|
||
errors: list[str] = []
|
||
first_nonempty_seen = False
|
||
|
||
with csv_path.open("r", encoding="utf-8-sig", newline="") as file:
|
||
reader = csv.reader(file)
|
||
for line_no, row in enumerate(reader, start=1):
|
||
values = [value.strip() for value in row if value.strip()]
|
||
if not values:
|
||
continue
|
||
if len(values) != 1:
|
||
errors.append(f"第 {line_no} 行必须只有一列，实际读取到 {len(values)} 个非空值")
|
||
continue
|
||
|
||
identifier = values[0]
|
||
if not first_nonempty_seen:
|
||
first_nonempty_seen = True
|
||
if identifier.lower() in HEADER_NAMES:
|
||
continue
|
||
|
||
if identifier in first_line_by_identifier:
|
||
errors.append(
|
||
f"第 {line_no} 行与第 {first_line_by_identifier[identifier]} 行重复：{identifier}"
|
||
)
|
||
continue
|
||
|
||
first_line_by_identifier[identifier] = line_no
|
||
assets.append(AssetInput(line_no=line_no, identifier=identifier))
|
||
|
||
if errors:
|
||
preview = "\n".join(f" - {error}" for error in errors[:20])
|
||
if len(errors) > 20:
|
||
preview += f"\n - 其余 {len(errors) - 20} 个错误已省略"
|
||
raise ValueError(f"CSV 校验失败，请修正后重试：\n{preview}")
|
||
if not assets:
|
||
raise ValueError("CSV 中没有有效的 ICCID 或虚拟号")
|
||
return assets
|
||
|
||
|
||
def parse_json_object(raw_body: str) -> dict[str, Any] | None:
|
||
"""尝试把响应正文解析为 JSON 对象。"""
|
||
if not raw_body.strip():
|
||
return None
|
||
try:
|
||
value = json.loads(raw_body)
|
||
except json.JSONDecodeError:
|
||
return None
|
||
return value if isinstance(value, dict) else None
|
||
|
||
|
||
def response_code(body: dict[str, Any] | None) -> int | str | None:
|
||
"""读取统一响应中的业务错误码。"""
|
||
if not body:
|
||
return None
|
||
code = body.get("code")
|
||
if isinstance(code, bool):
|
||
return int(code)
|
||
if isinstance(code, int):
|
||
return code
|
||
if isinstance(code, str):
|
||
stripped = code.strip()
|
||
return int(stripped) if stripped.isdigit() else stripped
|
||
return None
|
||
|
||
|
||
def response_message(body: dict[str, Any] | None, raw_body: str) -> str:
|
||
"""读取统一响应消息，非 JSON 响应则保留截断后的正文。"""
|
||
if body:
|
||
message = body.get("msg", body.get("message", ""))
|
||
if message is not None and str(message).strip():
|
||
return str(message).strip()
|
||
text = raw_body.strip().replace("\r", " ").replace("\n", " ")
|
||
return text[:500] if text else "接口未返回错误信息"
|
||
|
||
|
||
def is_success(http_status: int, code: int | str | None) -> bool:
|
||
"""同时校验 HTTP 状态码和业务响应码。"""
|
||
return 200 <= http_status < 300 and str(code) == "0"
|
||
|
||
|
||
def display_value(value: object) -> str:
|
||
"""把可能为空的字段转为适合日志和 CSV 的文本。"""
|
||
return "" if value is None else str(value)
|
||
|
||
|
||
def resolve_output_path(input_path: Path, output_arg: str) -> Path:
|
||
"""生成本批次的结果文件路径。"""
|
||
if output_arg.strip():
|
||
return Path(output_arg).expanduser().resolve()
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
return input_path.with_name(f"{input_path.stem}_购买结果_{timestamp}.csv")
|
||
|
||
|
||
def resolve_token(args: argparse.Namespace, client: AdminAPIClient) -> str:
|
||
"""优先使用现有 Token，否则使用后台账号自动登录。"""
|
||
token = args.token.strip()
|
||
if token:
|
||
return token
|
||
|
||
username = args.username.strip()
|
||
if not username:
|
||
raise ValueError(
|
||
"真实执行需要 --token，或同时提供 --username/--password；"
|
||
"也可通过 JUNHONG_ADMIN_TOKEN 等环境变量配置"
|
||
)
|
||
|
||
password = args.password
|
||
if not password and sys.stdin.isatty():
|
||
password = getpass("请输入后台登录密码：")
|
||
if not password:
|
||
raise ValueError("使用账号登录时必须提供密码")
|
||
|
||
print(f"正在使用后台账号 {username!r} 获取 Access Token...")
|
||
return client.login(username, password)
|
||
|
||
|
||
def preview(assets: list[AssetInput], base_url: str, package_id: int) -> int:
|
||
"""输出预演信息，不发送任何 HTTP 请求。"""
|
||
print("预演完成：未发送任何 HTTP 请求。")
|
||
print(f"接口地址：{base_url.rstrip('/')}{ORDER_PATH}")
|
||
print(f"资产数量：{len(assets)}")
|
||
print(f"套餐 ID：{package_id}")
|
||
print("支付方式：wallet")
|
||
print("请求示例：")
|
||
for asset in assets[:5]:
|
||
payload = {
|
||
"identifier": asset.identifier,
|
||
"package_ids": [package_id],
|
||
"payment_method": "wallet",
|
||
}
|
||
print(f" 第 {asset.line_no} 行：{json.dumps(payload, ensure_ascii=False)}")
|
||
if len(assets) > 5:
|
||
print(f" 其余 {len(assets) - 5} 条已省略")
|
||
print("确认参数无误后，增加 --execute 才会真实购买。")
|
||
return 0
|
||
|
||
|
||
def purchase_asset(
|
||
client: AdminAPIClient,
|
||
token: str,
|
||
asset: AssetInput,
|
||
package_id: int,
|
||
) -> PurchaseOutcome:
|
||
"""调用一次订单接口，并转换为统一的结果记录。"""
|
||
try:
|
||
result = client.create_order(token, asset.identifier, package_id)
|
||
code = response_code(result.body)
|
||
message = response_message(result.body, result.raw_body)
|
||
data = result.body.get("data") if result.body else None
|
||
order_data = data if isinstance(data, dict) else {}
|
||
success = is_success(result.status, code)
|
||
order_no = display_value(order_data.get("order_no"))
|
||
return PurchaseOutcome(
|
||
row={
|
||
"line_no": asset.line_no,
|
||
"identifier": asset.identifier,
|
||
"status": "成功" if success else "失败",
|
||
"http_status": result.status,
|
||
"code": display_value(code),
|
||
"msg": message,
|
||
"order_id": display_value(order_data.get("id")),
|
||
"order_no": order_no,
|
||
"total_amount": display_value(order_data.get("total_amount")),
|
||
},
|
||
success=success,
|
||
http_status=result.status,
|
||
code=code,
|
||
message=message,
|
||
order_no=order_no,
|
||
)
|
||
except RequestFailedError as exc:
|
||
message = str(exc)
|
||
return PurchaseOutcome(
|
||
row={
|
||
"line_no": asset.line_no,
|
||
"identifier": asset.identifier,
|
||
"status": "失败",
|
||
"http_status": "",
|
||
"code": "",
|
||
"msg": message,
|
||
"order_id": "",
|
||
"order_no": "",
|
||
"total_amount": "",
|
||
},
|
||
success=False,
|
||
http_status=None,
|
||
code=None,
|
||
message=message,
|
||
order_no="",
|
||
)
|
||
|
||
|
||
def execute(
|
||
assets: list[AssetInput],
|
||
client: AdminAPIClient,
|
||
token: str,
|
||
package_id: int,
|
||
output_path: Path,
|
||
interval: float,
|
||
) -> int:
|
||
"""顺序执行下单，并把每条结果立即写入 CSV。"""
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
success_count = 0
|
||
failed_count = 0
|
||
total = len(assets)
|
||
|
||
with output_path.open("w", encoding="utf-8-sig", newline="") as file:
|
||
writer = csv.DictWriter(
|
||
file,
|
||
fieldnames=[
|
||
"line_no",
|
||
"identifier",
|
||
"status",
|
||
"http_status",
|
||
"code",
|
||
"msg",
|
||
"order_id",
|
||
"order_no",
|
||
"total_amount",
|
||
],
|
||
)
|
||
writer.writeheader()
|
||
file.flush()
|
||
|
||
for index, asset in enumerate(assets, start=1):
|
||
outcome = purchase_asset(client, token, asset, package_id)
|
||
writer.writerow(outcome.row)
|
||
if outcome.success:
|
||
success_count += 1
|
||
print(f"[{index}/{total}] 成功：{asset.identifier}，订单号={outcome.order_no}")
|
||
else:
|
||
failed_count += 1
|
||
print(
|
||
f"[{index}/{total}] 失败：{asset.identifier}，"
|
||
f"HTTP={display_value(outcome.http_status)}，"
|
||
f"code={display_value(outcome.code)}，msg={outcome.message}",
|
||
file=sys.stderr,
|
||
)
|
||
|
||
file.flush()
|
||
|
||
if outcome.http_status == 401 or outcome.code in AUTH_ERROR_CODES:
|
||
print("认证已失效，停止后续下单；已处理结果已保存。", file=sys.stderr)
|
||
break
|
||
if interval > 0 and index < total:
|
||
time.sleep(interval)
|
||
|
||
print()
|
||
print(f"执行结束：成功 {success_count} 条，失败 {failed_count} 条。")
|
||
print(f"结果文件：{output_path}")
|
||
return 0 if failed_count == 0 and success_count == total else 2
|
||
|
||
|
||
def main() -> int:
|
||
"""校验参数，执行预演或真实批量下单。"""
|
||
args = parse_args()
|
||
try:
|
||
base_url = args.base_url.strip()
|
||
if not base_url:
|
||
raise ValueError("必须通过 --base-url 或 JUNHONG_ADMIN_BASE_URL 配置接口地址")
|
||
if not base_url.startswith(("http://", "https://")):
|
||
raise ValueError("base-url 必须以 http:// 或 https:// 开头")
|
||
if args.package_id <= 0:
|
||
raise ValueError("package-id 必须大于 0")
|
||
if args.timeout <= 0:
|
||
raise ValueError("timeout 必须大于 0")
|
||
if args.interval < 0:
|
||
raise ValueError("interval 不能小于 0")
|
||
|
||
input_path = Path(args.csv).expanduser().resolve()
|
||
assets = load_assets(input_path)
|
||
if not args.execute:
|
||
return preview(assets, base_url, args.package_id)
|
||
|
||
output_path = resolve_output_path(input_path, args.output)
|
||
if output_path == input_path:
|
||
raise ValueError("结果文件不能与输入 CSV 使用同一路径")
|
||
if output_path.exists():
|
||
raise ValueError(f"结果文件已存在，请更换 --output 路径：{output_path}")
|
||
client = AdminAPIClient(base_url, args.timeout)
|
||
token = resolve_token(args, client)
|
||
|
||
print(f"即将真实下单：资产 {len(assets)} 个，套餐 ID={args.package_id}，支付方式=wallet")
|
||
print(f"接口地址：{base_url.rstrip('/')}{ORDER_PATH}")
|
||
print(f"结果文件：{output_path}")
|
||
return execute(
|
||
assets=assets,
|
||
client=client,
|
||
token=token,
|
||
package_id=args.package_id,
|
||
output_path=output_path,
|
||
interval=args.interval,
|
||
)
|
||
except (ValueError, RequestFailedError) as exc:
|
||
print(f"错误：{exc}", file=sys.stderr)
|
||
return 1
|
||
except KeyboardInterrupt:
|
||
print("\n用户中断执行；已写入的结果会保留。", file=sys.stderr)
|
||
return 130
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|