diff --git a/README.md b/README.md index d7d908d..83cf9d4 100644 --- a/README.md +++ b/README.md @@ -253,6 +253,7 @@ default: - **批量同步**：卡状态、实名状态、流量使用情况 - **批量购买套餐脚本**：支持从单列 CSV 读取 ICCID/虚拟号，逐资产调用后台订单接口购买统一套餐，提供预演、重复拦截和逐条结果落盘能力。详见 [使用说明](scripts/batch_package_purchase/README.md) 与 [功能总结](docs/批量购买套餐脚本/功能总结.md) - **批量换货脚本**：支持从双列 CSV 读取新旧资产映射，固定执行直接换货和全量数据迁移，提供预演、批次冲突拦截和逐条结果落盘能力。详见 [使用说明](scripts/batch_exchange/README.md) 与 [功能总结](docs/批量换货脚本/功能总结.md) +- **批量回收设备脚本**：支持从单列 CSV 读取 IMEI/虚拟号，复用已部署的设备查询与回收接口，提供整批预检查、默认预演和逐设备结果记录。详见 [使用说明](scripts/batch_device_recall/README.md) 与 [功能总结](docs/批量回收设备脚本/功能总结.md) - **轮询系统**：IoT 卡实名状态、流量使用、套餐余额的定时轮询检查；支持配置化轮询策略、动态并发控制、告警系统、数据清理和手动触发功能；详见 [轮询系统文档](docs/polling-system/README.md) - **套餐系统升级**：完整的套餐生命周期管理，支持主套餐排队激活、加油包绑定主套餐、囤货待实名激活、流量按优先级扣减、自然月/按天有效期计算、日/月/年流量重置、客户端流量查询和套餐流量详单；详见 [套餐系统升级文档](docs/package-system-upgrade/) - **套餐生效条件覆盖与购买快照**：支持代理分配覆盖套餐生效条件，并将生效条件、周期类型和购买时长固化到套餐使用记录；激活与排队接续只消费购买快照。详见 [功能总结](docs/ur55-package-expiry-base/功能总结.md) diff --git a/docs/批量回收设备脚本/功能总结.md b/docs/批量回收设备脚本/功能总结.md new file mode 100644 index 0000000..9c6137d --- /dev/null +++ b/docs/批量回收设备脚本/功能总结.md @@ -0,0 +1,21 @@ +# 批量回收设备脚本功能总结 + +## 用途 + +生产环境尚未部署设备异步 CSV 批量任务时，使用 `scripts/batch_device_recall/batch_device_recall.py` 读取单列 IMEI/虚拟号 CSV，并调用已经部署的后台设备接口完成回收。 + +## 安全边界 + +- 默认仅预演，显式传入 `--execute` 才会请求接口。 +- 真实执行前通过 `GET /api/admin/devices` 精确解析全部标识；任何设备未找到、匹配不唯一或重复引用都会整批停止。 +- 设备解析完成后按接口上限每批最多 100 台调用 `POST /api/admin/devices/recall`。 +- 不直接修改数据库，继续复用既有权限校验、设备与绑定卡归属同步、分配记录、审计日志和缓存失效逻辑。 +- 回收请求不自动重试，避免服务端成功但客户端未收到响应时产生误判。 + +## 验证 + +运行： + +```bash +python3 -m unittest scripts/batch_device_recall/test_batch_device_recall.py +``` diff --git a/scripts/batch_device_recall/README.md b/scripts/batch_device_recall/README.md new file mode 100644 index 0000000..7c95c8d --- /dev/null +++ b/scripts/batch_device_recall/README.md @@ -0,0 +1,44 @@ +# 批量回收设备脚本 + +该脚本读取单列 CSV，通过生产环境已有的设备列表和回收接口处理设备，不依赖尚未部署的异步 CSV 批量任务。 + +默认只预演，不发送 HTTP 请求。只有增加 `--execute` 才会先解析全部设备，再按每批最多 100 台调用 `POST /api/admin/devices/recall`。接口会同步设备及其绑定卡的归属，并保留分配记录和审计日志。 + +## CSV 格式 + +首行表头可选，每行填写一个设备 IMEI 或虚拟号： + +```csv +device_identifier +868120000000001 +VIRTUAL000001 +``` + +脚本会拦截多列、空值、重复标识、未找到设备、模糊匹配和同一设备被 IMEI、虚拟号重复引用的情况。全部设备预检查通过后才会发送回收请求。 + +回收目标沿用登录账号的既有接口权限：平台账号回收到平台库存，代理账号只能从直属下级回收到自己的店铺。 + +## 预演 + +```bash +python3 scripts/batch_device_recall/batch_device_recall.py \ + --base-url https://cmp-api.example.com \ + --csv scripts/batch_device_recall/devices.example.csv +``` + +## 真实执行 + +推荐通过环境变量传递 Token： + +```bash +JUNHONG_ADMIN_TOKEN='<后台Access Token>' \ +python3 scripts/batch_device_recall/batch_device_recall.py \ + --base-url https://cmp-api.example.com \ + --csv /path/to/devices.csv \ + --remark '生产环境人工批量回收' \ + --execute +``` + +未提供 Token 时，也可使用 `JUNHONG_ADMIN_USERNAME` 和 `JUNHONG_ADMIN_PASSWORD` 自动登录。 + +结果默认写入输入文件同目录的 `原文件名_回收结果_YYYYMMDD_HHMMSS.csv`。脚本不会自动重试回收请求；网络异常时应先核对设备归属，再决定是否重跑失败项。 diff --git a/scripts/batch_device_recall/__pycache__/batch_device_recall.cpython-313.pyc b/scripts/batch_device_recall/__pycache__/batch_device_recall.cpython-313.pyc new file mode 100644 index 0000000..35995fc Binary files /dev/null and b/scripts/batch_device_recall/__pycache__/batch_device_recall.cpython-313.pyc differ diff --git a/scripts/batch_device_recall/__pycache__/test_batch_device_recall.cpython-313.pyc b/scripts/batch_device_recall/__pycache__/test_batch_device_recall.cpython-313.pyc new file mode 100644 index 0000000..aff3962 Binary files /dev/null and b/scripts/batch_device_recall/__pycache__/test_batch_device_recall.cpython-313.pyc differ diff --git a/scripts/batch_device_recall/batch_device_recall.py b/scripts/batch_device_recall/batch_device_recall.py new file mode 100644 index 0000000..697761e --- /dev/null +++ b/scripts/batch_device_recall/batch_device_recall.py @@ -0,0 +1,574 @@ +#!/usr/bin/env python3 +"""批量回收设备脚本：读取单列 CSV，通过现有后台接口回收设备。 + +默认只执行预演。只有显式传入 --execute 时，才会查询设备并调用 +POST /api/admin/devices/recall。脚本仅使用 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.parse import urlencode +from urllib.request import Request, urlopen + + +DEVICE_LIST_PATH = "/api/admin/devices" +DEVICE_RECALL_PATH = "/api/admin/devices/recall" +LOGIN_PATH = "/api/admin/login" +AUTH_ERROR_CODES = {1002, 1003, 1004} +MAX_RECALL_BATCH_SIZE = 100 +HEADER_NAMES = { + "device_identifier", + "identifier", + "virtual_no", + "imei", + "设备标识", + "虚拟号", +} + + +@dataclass(frozen=True) +class DeviceInput: + """保存 CSV 中的设备标识及原始行号。""" + + line_no: int + identifier: str + + +@dataclass(frozen=True) +class ResolvedDevice: + """保存接口解析出的设备信息。""" + + source: DeviceInput + device_id: int + virtual_no: str + imei: str + + +@dataclass(frozen=True) +class HTTPResult: + """保存一次 HTTP 请求的响应信息。""" + + status: int + body: dict[str, Any] | None + raw_body: 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._request_json( + "POST", + 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 find_devices(self, token: str, identifier: str) -> HTTPResult: + """使用现有设备列表接口按关键字查询候选设备。""" + query = urlencode({"keyword": identifier, "page": 1, "page_size": 100}) + return self._request_json("GET", f"{DEVICE_LIST_PATH}?{query}", None, token) + + def recall_devices( + self, + token: str, + devices: list[ResolvedDevice], + remark: str, + ) -> HTTPResult: + """调用现有接口回收一批设备。""" + return self._request_json( + "POST", + DEVICE_RECALL_PATH, + {"device_ids": [device.device_id for device in devices], "remark": remark}, + token, + ) + + def _request_json( + self, + method: str, + path: str, + payload: dict[str, Any] | None, + token: str | None, + ) -> HTTPResult: + body = None + if payload is not None: + body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + headers = {"Accept": "application/json", "User-Agent": "junhong-batch-device-recall/1.0"} + if payload is not None: + headers["Content-Type"] = "application/json" + if token: + headers["Authorization"] = f"Bearer {token}" + request = Request(self.base_url + path, data=body, headers=headers, method=method) + try: + with urlopen(request, timeout=self.timeout) as response: + raw_body = response.read().decode("utf-8", errors="replace") + return HTTPResult(response.status, parse_json_object(raw_body), raw_body) + except HTTPError as exc: + raw_body = exc.read().decode("utf-8", errors="replace") + return HTTPResult(exc.code, parse_json_object(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/devices/recall 批量回收设备", + ) + parser.add_argument( + "--base-url", + default=os.getenv("JUNHONG_ADMIN_BASE_URL", ""), + help="接口 Base URL；也可使用 JUNHONG_ADMIN_BASE_URL", + ) + parser.add_argument("--csv", required=True, help="单列设备 CSV 路径，首行可有表头") + parser.add_argument( + "--remark", + default="生产环境 CSV 批量回收设备", + help="回收备注", + ) + 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 时用于自动登录", + ) + parser.add_argument( + "--password", + default=os.getenv("JUNHONG_ADMIN_PASSWORD", ""), + help="后台登录密码；建议使用环境变量", + ) + 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_devices(csv_path: Path) -> list[DeviceInput]: + """读取单列 CSV，并在调用接口前拦截重复标识。""" + if not csv_path.exists(): + raise ValueError(f"找不到 CSV 文件：{csv_path}") + if not csv_path.is_file(): + raise ValueError(f"CSV 路径不是文件：{csv_path}") + + devices: list[DeviceInput] = [] + 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: + for line_no, row in enumerate(csv.reader(file), start=1): + if not any(value.strip() for value in row): + continue + if len(row) != 1: + errors.append(f"第 {line_no} 行必须正好有一列，实际读取到 {len(row)} 列") + continue + identifier = row[0].strip() + if not first_nonempty_seen: + first_nonempty_seen = True + if identifier.lower() in HEADER_NAMES: + continue + if not identifier: + errors.append(f"第 {line_no} 行设备标识不能为空") + continue + if len(identifier) > 100: + errors.append(f"第 {line_no} 行设备标识不能超过 100 个字符") + 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 + devices.append(DeviceInput(line_no, identifier)) + + if errors: + raise ValueError(format_errors("CSV 校验失败，请修正后重试", errors)) + if not devices: + raise ValueError("CSV 中没有有效的 IMEI 或虚拟号") + return devices + + +def resolve_devices( + client: AdminAPIClient, + token: str, + inputs: list[DeviceInput], +) -> list[ResolvedDevice]: + """在任何回收请求前精确解析全部标识，确保整批输入可用。""" + resolved: list[ResolvedDevice] = [] + first_line_by_device_id: dict[int, int] = {} + errors: list[str] = [] + for index, item in enumerate(inputs, start=1): + result = client.find_devices(token, item.identifier) + code = response_code(result.body) + if not is_success(result.status, code): + errors.append( + f"第 {item.line_no} 行查询失败：HTTP {result.status}，code={display_value(code)}，" + f"msg={response_message(result.body, result.raw_body)}" + ) + if result.status == 401 or code in AUTH_ERROR_CODES: + break + continue + matches = exact_device_matches(result.body, item.identifier) + if len(matches) != 1: + reason = "未找到设备" if not matches else "同时精确匹配多个设备" + errors.append(f"第 {item.line_no} 行{reason}：{item.identifier}") + continue + device = matches[0] + device_id = parse_positive_int(device.get("id")) + if device_id is None: + errors.append(f"第 {item.line_no} 行设备响应缺少有效 ID：{item.identifier}") + continue + if device_id in first_line_by_device_id: + errors.append( + f"第 {item.line_no} 行与第 {first_line_by_device_id[device_id]} 行指向同一设备：" + f"{item.identifier}" + ) + continue + first_line_by_device_id[device_id] = item.line_no + resolved.append( + ResolvedDevice( + source=item, + device_id=device_id, + virtual_no=display_value(device.get("virtual_no")), + imei=display_value(device.get("imei")), + ) + ) + print(f"[{index}/{len(inputs)}] 已解析：{item.identifier} -> 设备ID {device_id}") + + if errors: + raise ValueError(format_errors("设备预检查失败，未发送任何回收请求", errors)) + return resolved + + +def exact_device_matches(body: dict[str, Any] | None, identifier: str) -> list[dict[str, Any]]: + """从模糊查询结果中保留 IMEI 或虚拟号精确匹配项。""" + data = body.get("data") if body else None + items = data.get("items") if isinstance(data, dict) else None + if not isinstance(items, list): + return [] + return [ + item + for item in items + if isinstance(item, dict) + and (item.get("virtual_no") == identifier or item.get("imei") == identifier) + ] + + +def recall_batch_rows( + batch: list[ResolvedDevice], + result: HTTPResult, +) -> tuple[list[dict[str, object]], bool]: + """把一次批量回收响应转换为逐设备结果行。""" + code = response_code(result.body) + message = response_message(result.body, result.raw_body) + if not is_success(result.status, code): + return [result_row(device, False, result.status, code, message) for device in batch], False + + data = result.body.get("data") if result.body else None + if not isinstance(data, dict): + message = "接口返回成功但缺少回收结果，请人工核对" + return [result_row(device, False, result.status, code, message) for device in batch], False + failed_items = data.get("failed_items") + failed_by_id: dict[int, str] = {} + if isinstance(failed_items, list): + for item in failed_items: + if not isinstance(item, dict): + continue + device_id = parse_positive_int(item.get("device_id")) + if device_id is not None: + failed_by_id[device_id] = display_value(item.get("reason")) or "回收失败" + success_count = parse_nonnegative_int(data.get("success_count")) + fail_count = parse_nonnegative_int(data.get("fail_count")) + if ( + success_count is None + or fail_count is None + or success_count + fail_count != len(batch) + or fail_count != len(failed_by_id) + ): + message = "接口回收统计与请求数量不一致，请人工核对" + return [result_row(device, False, result.status, code, message) for device in batch], False + rows = [ + result_row( + device, + device.device_id not in failed_by_id, + result.status, + code, + failed_by_id.get(device.device_id, message), + ) + for device in batch + ] + return rows, not failed_by_id + + +def result_row( + device: ResolvedDevice, + success: bool, + http_status: int | str, + code: int | str | None, + message: str, +) -> dict[str, object]: + """构造结果 CSV 的单行内容。""" + return { + "line_no": device.source.line_no, + "identifier": device.source.identifier, + "device_id": device.device_id, + "virtual_no": device.virtual_no, + "imei": device.imei, + "status": "成功" if success else "失败", + "http_status": http_status, + "code": display_value(code), + "msg": message, + } + + +def execute( + devices: list[ResolvedDevice], + client: AdminAPIClient, + token: str, + remark: str, + output_path: Path, + interval: float, +) -> int: + """按接口上限分批回收，并把每批结果立即写入 CSV。""" + output_path.parent.mkdir(parents=True, exist_ok=True) + success_count = 0 + failed_count = 0 + with output_path.open("w", encoding="utf-8-sig", newline="") as file: + writer = csv.DictWriter( + file, + fieldnames=[ + "line_no", "identifier", "device_id", "virtual_no", "imei", + "status", "http_status", "code", "msg", + ], + ) + writer.writeheader() + file.flush() + batches = list(chunks(devices, MAX_RECALL_BATCH_SIZE)) + for batch_index, batch in enumerate(batches, start=1): + try: + result = client.recall_devices(token, batch, remark) + rows, batch_success = recall_batch_rows(batch, result) + except RequestFailedError as exc: + rows = [result_row(device, False, "", None, str(exc)) for device in batch] + batch_success = False + result = None + for row in rows: + writer.writerow(row) + if row["status"] == "成功": + success_count += 1 + else: + failed_count += 1 + file.flush() + print( + f"[{batch_index}/{len(batches)}] 本批 {len(batch)} 台：" + f"成功 {sum(row['status'] == '成功' for row in rows)}，" + f"失败 {sum(row['status'] != '成功' for row in rows)}" + ) + if result is not None: + code = response_code(result.body) + if result.status == 401 or code in AUTH_ERROR_CODES: + print("认证已失效，停止后续回收；已处理结果已保存。", file=sys.stderr) + break + if not batch_success: + print("本批存在失败，请根据结果文件人工核对。", file=sys.stderr) + if interval > 0 and batch_index < len(batches): + time.sleep(interval) + print(f"执行结束：成功 {success_count} 条，失败 {failed_count} 条。") + print(f"结果文件：{output_path}") + return 0 if success_count == len(devices) and failed_count == 0 else 2 + + +def preview(devices: list[DeviceInput], base_url: str, remark: str) -> int: + """输出预演信息，不发送 HTTP 请求。""" + print("预演完成：未发送任何 HTTP 请求。") + print(f"设备数量：{len(devices)}") + print(f"查询接口：{base_url.rstrip('/')}{DEVICE_LIST_PATH}?keyword=<设备标识>") + print(f"回收接口：{base_url.rstrip('/')}{DEVICE_RECALL_PATH}") + print(f"回收备注：{remark}") + print("标识示例：") + for device in devices[:5]: + print(f" 第 {device.line_no} 行：{device.identifier}") + if len(devices) > 5: + print(f" 其余 {len(devices) - 5} 条已省略") + print("增加 --execute 后，脚本会先精确解析全部设备，再开始分批回收。") + return 0 + + +def chunks(devices: list[ResolvedDevice], size: int): + """按固定大小切分设备列表。""" + for start in range(0, len(devices), size): + yield devices[start:start + size] + + +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: + """读取统一响应消息。""" + 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 parse_positive_int(value: object) -> int | None: + """解析正整数。""" + parsed = parse_nonnegative_int(value) + return parsed if parsed is not None and parsed > 0 else None + + +def parse_nonnegative_int(value: object) -> int | None: + """解析非负整数，拒绝布尔值。""" + if isinstance(value, bool): + return None + if isinstance(value, int) and value >= 0: + return value + if isinstance(value, str) and value.strip().isdigit(): + return int(value.strip()) + return None + + +def display_value(value: object) -> str: + """把可能为空的字段转为文本。""" + return "" if value is None else str(value) + + +def format_errors(title: str, errors: list[str]) -> str: + """截断并格式化批量错误。""" + preview = "\n".join(f" - {error}" for error in errors[:20]) + if len(errors) > 20: + preview += f"\n - 其余 {len(errors) - 20} 个错误已省略" + return f"{title}：\n{preview}" + + +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，否则使用后台账号自动登录。""" + if args.token.strip(): + return args.token.strip() + if not args.username.strip(): + raise ValueError("真实执行需要 --token，或同时提供 --username/--password") + password = args.password + if not password and sys.stdin.isatty(): + password = getpass("请输入后台登录密码：") + if not password: + raise ValueError("使用账号登录时必须提供密码") + print(f"正在使用后台账号 {args.username.strip()!r} 获取 Access Token...") + return client.login(args.username.strip(), password) + + +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:// 开头") + remark = args.remark.strip() + if not remark or len(remark) > 500: + raise ValueError("remark 必须为 1 至 500 个字符") + if args.timeout <= 0 or args.interval < 0: + raise ValueError("timeout 必须大于 0，interval 不能小于 0") + input_path = Path(args.csv).expanduser().resolve() + inputs = load_devices(input_path) + if not args.execute: + return preview(inputs, base_url, remark) + + 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) + resolved = resolve_devices(client, token, inputs) + print(f"预检查通过：{len(resolved)} 台设备；即将按每批最多 100 台真实回收。") + return execute(resolved, client, token, remark, output_path, 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()) diff --git a/scripts/batch_device_recall/devices.example.csv b/scripts/batch_device_recall/devices.example.csv new file mode 100644 index 0000000..a484d36 --- /dev/null +++ b/scripts/batch_device_recall/devices.example.csv @@ -0,0 +1,61 @@ +device_identifier +862639073065981 +862639073536379 +862639073876858 +862639071874939 +862639073762900 +862639071854352 +862639073461925 +862639073996755 +862639073027965 +862639073337596 +862639073775027 +862639071857918 +862639073005409 +862639075961740 +862639073877062 +862639073940258 +862639073863336 +862639071971818 +862639073329528 +862639073015960 +862639071942918 +862639073787832 +862639073918049 +862639075954604 +862639071986337 +862639073831101 +862639073397517 +862639075961781 +862639073934822 +862639073981799 +862639071857900 +862639073785372 +862639073017438 +862639073778799 +862639073369672 +862639071974481 +862639071920955 +862639073062491 +862639073867253 +862639071912374 +862639071854402 +862639073538466 +862639073885289 +862639073394191 +862639073903652 +862639073949309 +862639071932281 +862639075954364 +862639073955231 +862639073061121 +862639073003941 +862639073502140 +862639073966170 +862639073476576 +862639073775076 +862639073383434 +862639073488316 +862639073948756 +862639073368856 +862639073966527 diff --git a/scripts/batch_device_recall/devices.example_回收结果_20260728_110203.csv b/scripts/batch_device_recall/devices.example_回收结果_20260728_110203.csv new file mode 100644 index 0000000..b1e0c5d --- /dev/null +++ b/scripts/batch_device_recall/devices.example_回收结果_20260728_110203.csv @@ -0,0 +1,61 @@ +line_no,identifier,device_id,virtual_no,imei,status,http_status,code,msg +2,862639073065981,237,862639073065981,862639073065981,成功,200,0,success +3,862639073536379,219,862639073536379,862639073536379,成功,200,0,success +4,862639073876858,249,862639073876858,862639073876858,成功,200,0,success +5,862639071874939,183,862639071874939,862639071874939,成功,200,0,success +6,862639073762900,239,862639073762900,862639073762900,成功,200,0,success +7,862639071854352,178,862639071854352,862639071854352,成功,200,0,success +8,862639073461925,210,862639073461925,862639073461925,成功,200,0,success +9,862639073996755,172,862639073996755,862639073996755,成功,200,0,success +10,862639073027965,228,862639073027965,862639073027965,成功,200,0,success +11,862639073337596,194,862639073337596,862639073337596,成功,200,0,success +12,862639073775027,241,862639073775027,862639073775027,成功,200,0,success +13,862639071857918,181,862639071857918,862639071857918,成功,200,0,success +14,862639073005409,225,862639073005409,862639073005409,成功,200,0,success +15,862639075961740,176,862639075961740,862639075961740,成功,200,0,success +16,862639073877062,250,862639073877062,862639073877062,成功,200,0,success +17,862639073940258,161,862639073940258,862639073940258,成功,200,0,success +18,862639073863336,247,862639073863336,862639073863336,成功,200,0,success +19,862639071971818,191,862639071971818,862639071971818,成功,200,0,success +20,862639073329528,192,862639073329528,862639073329528,成功,200,0,success +21,862639073015960,226,862639073015960,862639073015960,成功,200,0,success +22,862639071942918,190,862639071942918,862639071942918,成功,200,0,success +23,862639073787832,245,862639073787832,862639073787832,成功,200,0,success +24,862639073918049,257,862639073918049,862639073918049,成功,200,0,success +25,862639075954604,175,862639075954604,862639075954604,成功,200,0,success +26,862639071986337,222,862639071986337,862639071986337,成功,200,0,success +27,862639073831101,246,862639073831101,862639073831101,成功,200,0,success +28,862639073397517,205,862639073397517,862639073397517,成功,200,0,success +29,862639075961781,177,862639075961781,862639075961781,成功,200,0,success +30,862639073934822,259,862639073934822,862639073934822,成功,200,0,success +31,862639073981799,170,862639073981799,862639073981799,成功,200,0,success +32,862639071857900,180,862639071857900,862639071857900,成功,200,0,success +33,862639073785372,244,862639073785372,862639073785372,成功,200,0,success +34,862639073017438,227,862639073017438,862639073017438,成功,200,0,success +35,862639073778799,243,862639073778799,862639073778799,成功,200,0,success +36,862639073369672,202,862639073369672,862639073369672,成功,200,0,success +37,862639071974481,221,862639071974481,862639071974481,成功,200,0,success +38,862639071920955,188,862639071920955,862639071920955,成功,200,0,success +39,862639073062491,236,862639073062491,862639073062491,成功,200,0,success +40,862639073867253,248,862639073867253,862639073867253,成功,200,0,success +41,862639071912374,186,862639071912374,862639071912374,成功,200,0,success +42,862639071854402,179,862639071854402,862639071854402,成功,200,0,success +43,862639073538466,220,862639073538466,862639073538466,成功,200,0,success +44,862639073885289,251,862639073885289,862639073885289,成功,200,0,success +45,862639073394191,204,862639073394191,862639073394191,成功,200,0,success +46,862639073903652,254,862639073903652,862639073903652,成功,200,0,success +47,862639073949309,164,862639073949309,862639073949309,成功,200,0,success +48,862639071932281,189,862639071932281,862639071932281,成功,200,0,success +49,862639075954364,174,862639075954364,862639075954364,成功,200,0,success +50,862639073955231,167,862639073955231,862639073955231,成功,200,0,success +51,862639073061121,235,862639073061121,862639073061121,成功,200,0,success +52,862639073003941,224,862639073003941,862639073003941,成功,200,0,success +53,862639073502140,213,862639073502140,862639073502140,成功,200,0,success +54,862639073966170,168,862639073966170,862639073966170,成功,200,0,success +55,862639073476576,211,862639073476576,862639073476576,成功,200,0,success +56,862639073775076,242,862639073775076,862639073775076,成功,200,0,success +57,862639073383434,203,862639073383434,862639073383434,成功,200,0,success +58,862639073488316,212,862639073488316,862639073488316,成功,200,0,success +59,862639073948756,163,862639073948756,862639073948756,成功,200,0,success +60,862639073368856,201,862639073368856,862639073368856,成功,200,0,success +61,862639073966527,169,862639073966527,862639073966527,成功,200,0,success diff --git a/scripts/batch_device_recall/test_batch_device_recall.py b/scripts/batch_device_recall/test_batch_device_recall.py new file mode 100644 index 0000000..b7c8a6f --- /dev/null +++ b/scripts/batch_device_recall/test_batch_device_recall.py @@ -0,0 +1,127 @@ +"""批量回收设备脚本测试。""" +from __future__ import annotations + +import importlib.util +import sys +import tempfile +import unittest +from pathlib import Path + + +SCRIPT_PATH = Path(__file__).with_name("batch_device_recall.py") +SPEC = importlib.util.spec_from_file_location("batch_device_recall", SCRIPT_PATH) +assert SPEC is not None and SPEC.loader is not None +batch_device_recall = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = batch_device_recall +SPEC.loader.exec_module(batch_device_recall) + + +class LoadDevicesTest(unittest.TestCase): + """验证单列 CSV 输入。""" + + def write_csv(self, content: str) -> Path: + """创建临时 CSV。""" + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + path = Path(directory.name) / "devices.csv" + path.write_text(content, encoding="utf-8") + return path + + def test_loads_header_and_rows(self) -> None: + """支持表头并保留行号。""" + path = self.write_csv("device_identifier\nIMEI001\nVIRTUAL001\n") + + devices = batch_device_recall.load_devices(path) + + self.assertEqual( + devices, + [ + batch_device_recall.DeviceInput(2, "IMEI001"), + batch_device_recall.DeviceInput(3, "VIRTUAL001"), + ], + ) + + def test_rejects_duplicate_identifier(self) -> None: + """同一标识不能重复回收。""" + path = self.write_csv("IMEI001\nIMEI001\n") + + with self.assertRaisesRegex(ValueError, "重复"): + batch_device_recall.load_devices(path) + + +class ExactDeviceMatchesTest(unittest.TestCase): + """验证模糊查询结果必须再次精确匹配。""" + + def test_keeps_only_exact_virtual_no_or_imei(self) -> None: + """排除仅包含关键字的候选设备。""" + body = { + "data": { + "items": [ + {"id": 1, "virtual_no": "VIRTUAL001", "imei": "IMEI001"}, + {"id": 2, "virtual_no": "VIRTUAL001-X", "imei": "IMEI002"}, + ] + } + } + + matches = batch_device_recall.exact_device_matches(body, "VIRTUAL001") + + self.assertEqual([item["id"] for item in matches], [1]) + + +class RecallBatchRowsTest(unittest.TestCase): + """验证批量接口结果映射。""" + + def test_maps_failed_items_to_source_rows(self) -> None: + """接口失败明细应对应回原 CSV 标识。""" + devices = [ + batch_device_recall.ResolvedDevice( + batch_device_recall.DeviceInput(2, "IMEI001"), 1, "V001", "IMEI001" + ), + batch_device_recall.ResolvedDevice( + batch_device_recall.DeviceInput(3, "IMEI002"), 2, "V002", "IMEI002" + ), + ] + result = batch_device_recall.HTTPResult( + 200, + { + "code": 0, + "msg": "success", + "data": { + "success_count": 1, + "fail_count": 1, + "failed_items": [{"device_id": 2, "reason": "设备已在平台库存中"}], + }, + }, + "", + ) + + rows, success = batch_device_recall.recall_batch_rows(devices, result) + + self.assertFalse(success) + self.assertEqual([row["status"] for row in rows], ["成功", "失败"]) + self.assertEqual(rows[1]["msg"], "设备已在平台库存中") + + def test_rejects_incomplete_failed_items(self) -> None: + """失败数量没有对应明细时不能把设备误记为成功。""" + device = batch_device_recall.ResolvedDevice( + batch_device_recall.DeviceInput(2, "IMEI001"), 1, "V001", "IMEI001" + ) + result = batch_device_recall.HTTPResult( + 200, + { + "code": 0, + "msg": "success", + "data": {"success_count": 0, "fail_count": 1, "failed_items": []}, + }, + "", + ) + + rows, success = batch_device_recall.recall_batch_rows([device], result) + + self.assertFalse(success) + self.assertEqual(rows[0]["status"], "失败") + self.assertIn("统计", rows[0]["msg"]) + + +if __name__ == "__main__": + unittest.main()