This commit is contained in:
574
scripts/batch_device_recall/batch_device_recall.py
Normal file
574
scripts/batch_device_recall/batch_device_recall.py
Normal file
@@ -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())
|
||||
Reference in New Issue
Block a user