diff --git a/scripts/batch_package_purchase/README.md b/scripts/batch_package_purchase/README.md new file mode 100644 index 0000000..751b240 --- /dev/null +++ b/scripts/batch_package_purchase/README.md @@ -0,0 +1,88 @@ +# 批量购买套餐脚本 + +该脚本读取只有一列的 CSV，逐条调用 `POST /api/admin/orders`，为一批资产购买同一个套餐。 + +脚本仅使用 Python 标准库，不需要安装依赖。默认只预演，必须增加 `--execute` 才会真实下单。 + +## CSV 格式 + +首行表头可选，每行填写一个 ICCID 或虚拟号： + +```csv +identifier +89860000000000000001 +VIRTUAL000001 +``` + +脚本会在请求前检查空文件、多列和重复资产。发现重复资产时整批终止，避免同一资产被重复购买。 + +## 预演 + +```bash +python3 scripts/batch_package_purchase/batch_purchase.py \ + --base-url https://cmp-api.example.com \ + --csv scripts/batch_package_purchase/assets.example.csv \ + --package-id 123 +``` + +预演只校验 CSV 并展示请求示例，不需要 Token，也不会调用接口。 + +## 使用已有 Token 执行 + +推荐通过环境变量传递 Token，避免进入命令历史： + +```bash +JUNHONG_ADMIN_TOKEN='<后台Access Token>' \ +python3 scripts/batch_package_purchase/batch_purchase.py \ + --base-url https://cmp-api.example.com \ + --csv /path/to/assets.csv \ + --package-id 123 \ + --execute +``` + +## 使用账号自动登录后执行 + +未提供 Token 时，可以使用后台账号调用 `/api/admin/login` 自动获取 Token： + +```bash +JUNHONG_ADMIN_USERNAME='<后台账号>' \ +JUNHONG_ADMIN_PASSWORD='<后台密码>' \ +python3 scripts/batch_package_purchase/batch_purchase.py \ + --base-url https://cmp-api.example.com \ + --csv /path/to/assets.csv \ + --package-id 123 \ + --execute +``` + +也可以使用 `--token`、`--username`、`--password` 参数。密码优先通过环境变量或交互输入，避免保存在 Shell 历史中。 + +## 结果文件 + +默认在输入 CSV 同目录生成： + +```text +原文件名_购买结果_YYYYMMDD_HHMMSS.csv +``` + +结果包含资产标识、成功或失败状态、HTTP 状态码、业务错误码、错误消息、订单 ID、订单号和订单金额。每处理一条都会立即刷新文件，中途中断时已完成的结果不会丢失。 + +为避免覆盖历史结果，`--output` 指定的文件已经存在时脚本会直接报错。 + +可通过 `--output` 指定结果路径： + +```bash +python3 scripts/batch_package_purchase/batch_purchase.py \ + --base-url https://cmp-api.example.com \ + --csv /path/to/assets.csv \ + --package-id 123 \ + --output /path/to/result.csv \ + --execute +``` + +## 其他参数 + +- `--timeout`：单次请求超时秒数，默认 `30`。 +- `--interval`：每次下单后的等待秒数，默认 `0.2`。 +- `--execute`：显式开启真实下单。 + +脚本固定使用 `wallet` 钱包支付，每个资产单独创建一个订单。脚本不会自动重试 POST 请求，避免服务端已成功但客户端未收到响应时重复下单。失败项可根据结果 CSV 人工确认后单独重跑。 diff --git a/scripts/batch_package_purchase/assets.example.csv b/scripts/batch_package_purchase/assets.example.csv new file mode 100644 index 0000000..3587686 --- /dev/null +++ b/scripts/batch_package_purchase/assets.example.csv @@ -0,0 +1,23 @@ +identifier +8986032445201075412, +8986032445201075413, +8986032445201075414, +8986032445201075416, +8986032445201075417, +8986032445201075418, +8986032445201075419, +8986032445201075420, +8986032445201075421, +8986032445201075422, +8986032445201075423, +8986032445201075425, +8986032445201075426, +8986032445201075428, +8986032445201075430, +8986032445201075433, +8986032445201075434, +8986032445201075441, +8986032445201075444, +8986032445201075447, +8986032445201075448, +8986032445201075451 diff --git a/scripts/batch_package_purchase/batch_purchase.py b/scripts/batch_package_purchase/batch_purchase.py new file mode 100755 index 0000000..042e90b --- /dev/null +++ b/scripts/batch_package_purchase/batch_purchase.py @@ -0,0 +1,492 @@ +#!/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())