csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

批量购买

Browse Source
This commit is contained in:
break
2026-07-15 12:00:05 +08:00
parent 2e130b98f5
commit 1a9db9328e
3 changed files with 603 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
scripts/batch_package_purchase/README.md Normal file
View File

@@ -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 人工确认后单独重跑。

23
scripts/batch_package_purchase/assets.example.csv Normal file
View File

@@ -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
1 identifier
2 8986032445201075412,
3 8986032445201075413,
4 8986032445201075414,
5 8986032445201075416,
6 8986032445201075417,
7 8986032445201075418,
8 8986032445201075419,
9 8986032445201075420,
10 8986032445201075421,
11 8986032445201075422,
12 8986032445201075423,
13 8986032445201075425,
14 8986032445201075426,
15 8986032445201075428,
16 8986032445201075430,
17 8986032445201075433,
18 8986032445201075434,
19 8986032445201075441,
20 8986032445201075444,
21 8986032445201075447,
22 8986032445201075448,
23 8986032445201075451

492
scripts/batch_package_purchase/batch_purchase.py Executable file
View File

@@ -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())