129 lines
4.4 KiB
Python
129 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import mimetypes
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
|
|
API_BASE = os.getenv("HUIHUI_API_BASE", "http://192.168.1.188:8000/api").rstrip("/")
|
|
AVATAR_DIR = Path(os.getenv("HUIHUI_AVATAR_DIR", str(Path.home() / "Downloads" / "头像")))
|
|
ACCOUNT_START = os.getenv("HUIHUI_ACCOUNT_START", "13721560041")
|
|
ACCOUNT_END = os.getenv("HUIHUI_ACCOUNT_END", "13721560199")
|
|
POLL_SECONDS = int(os.getenv("HUIHUI_POLL_SECONDS", "10"))
|
|
TIMEOUT_SECONDS = int(os.getenv("HUIHUI_TIMEOUT_SECONDS", "5400"))
|
|
|
|
|
|
def load_avatar_files(expected_count: int) -> list[Path]:
|
|
files = sorted(path for path in AVATAR_DIR.iterdir() if path.is_file())
|
|
if len(files) < expected_count:
|
|
raise RuntimeError(f"头像文件不足: 需要 {expected_count} 个，实际只有 {len(files)} 个")
|
|
return files[:expected_count]
|
|
|
|
|
|
def fetch_target_users(client: httpx.Client) -> list[dict]:
|
|
users: list[dict] = []
|
|
page = 1
|
|
page_size = 100
|
|
|
|
while True:
|
|
resp = client.get(
|
|
f"{API_BASE}/users",
|
|
params={"page": page, "page_size": page_size, "keyword": "1372156"},
|
|
)
|
|
resp.raise_for_status()
|
|
payload = resp.json()["data"]
|
|
items = payload["items"]
|
|
users.extend(items)
|
|
if page * page_size >= payload["total"] or not items:
|
|
break
|
|
page += 1
|
|
|
|
targets = [u for u in users if ACCOUNT_START <= u["account"] <= ACCOUNT_END]
|
|
targets.sort(key=lambda item: item["account"])
|
|
return targets
|
|
|
|
|
|
def upload_avatar(client: httpx.Client, user: dict, avatar_path: Path) -> str:
|
|
mime = mimetypes.guess_type(str(avatar_path))[0] or "application/octet-stream"
|
|
with avatar_path.open("rb") as fp:
|
|
resp = client.post(
|
|
f"{API_BASE}/users/{user['id']}/upload-avatar",
|
|
params={"sync_to_platform": "true"},
|
|
files={"file": (avatar_path.name, fp, mime)},
|
|
timeout=120,
|
|
)
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
if payload.get("code") not in (0, 200):
|
|
raise RuntimeError(payload.get("message") or f"上传失败: {payload}")
|
|
return payload["data"]["avatar_url"]
|
|
|
|
|
|
def main() -> int:
|
|
started_at = time.time()
|
|
processed: set[str] = set()
|
|
|
|
with httpx.Client(timeout=30) as client:
|
|
initial_targets = fetch_target_users(client)
|
|
if not initial_targets:
|
|
raise RuntimeError("没有找到目标手机号范围内的用户")
|
|
|
|
avatar_files = load_avatar_files(len(initial_targets))
|
|
avatar_map = {
|
|
user["account"]: avatar_files[idx]
|
|
for idx, user in enumerate(initial_targets)
|
|
}
|
|
|
|
print(
|
|
f"准备处理 {len(initial_targets)} 个用户，头像目录 {AVATAR_DIR}，接口 {API_BASE}",
|
|
flush=True,
|
|
)
|
|
|
|
while len(processed) < len(initial_targets):
|
|
if time.time() - started_at > TIMEOUT_SECONDS:
|
|
pending = sorted(set(avatar_map) - processed)
|
|
print(f"超时，仍有 {len(pending)} 个用户未完成: {', '.join(pending[:20])}", flush=True)
|
|
return 2
|
|
|
|
targets = fetch_target_users(client)
|
|
status_summary: dict[int, int] = {}
|
|
for user in targets:
|
|
status_summary[user["status"]] = status_summary.get(user["status"], 0) + 1
|
|
|
|
for user in targets:
|
|
account = user["account"]
|
|
if account in processed:
|
|
continue
|
|
if user["status"] != 2:
|
|
continue
|
|
|
|
avatar_url = upload_avatar(client, user, avatar_map[account])
|
|
processed.add(account)
|
|
print(
|
|
f"[{len(processed):03d}/{len(targets)}] {account} -> {avatar_map[account].name} -> {avatar_url}",
|
|
flush=True,
|
|
)
|
|
|
|
if len(processed) < len(initial_targets):
|
|
print(
|
|
f"等待登录中: 已完成 {len(processed)}/{len(initial_targets)}，状态分布 {status_summary}",
|
|
flush=True,
|
|
)
|
|
time.sleep(POLL_SECONDS)
|
|
|
|
print("全部头像替换完成", flush=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except KeyboardInterrupt:
|
|
print("已中断", file=sys.stderr)
|
|
raise SystemExit(130)
|