app/api_keys.py

"""网关 API Key 管理：多 key 库（JSON 文件存储）+ 生成/校验。

- 文件：``<dir>/api_keys.json``（dir 通常 = settings.log_dir，HF 持久化在 /data/logs）。
- 兼容：config 的 ``[gateway].api_key``（gateway_api_key）自动并入校验集合，
  面板生成的 key 与它任一匹配即放行 —— 老客户端不用换 key。
- 线程安全：写盘用 flock 原子替换；内存缓存由 app.state.api_keys 持有，
  admin 修改后同步（见 app/admin.py 的 /admin/apikeys 端点）。
"""
from __future__ import annotations

import json
import secrets
import time
from pathlib import Path
from typing import Any

from app._compat import flock_ex

KEY_LEN = 32  # token_urlsafe 字节数 → 43 字符


def generate_key() -> str:
    """生成随机 API key（url-safe，无需转义）。"""
    return "an-" + secrets.token_urlsafe(KEY_LEN)


def _path(dir_path: Path) -> Path:
    return dir_path / "api_keys.json"


def load_keys(dir_path: Path) -> list[dict[str, Any]]:
    """读 api_keys.json；文件不存在/损坏 → 空列表。"""
    p = _path(dir_path)
    if not p.is_file():
        return []
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
        if isinstance(data, list):
            return [k for k in data if isinstance(k, dict) and k.get("key")]
    except (ValueError, OSError):  # noqa: BLE001
        pass
    return []


def save_keys(dir_path: Path, keys: list[dict[str, Any]]) -> None:
    """原子写 api_keys.json（flock + tmp 替换）。"""
    dir_path.mkdir(parents=True, exist_ok=True)
    p = _path(dir_path)
    tmp = p.with_suffix(".json.tmp")
    payload = json.dumps(keys, ensure_ascii=False, indent=2)
    with tmp.open("w", encoding="utf-8") as f:
        f.write(payload)
        f.write("\n")
        flock_ex(f.fileno())
    tmp.replace(p)


def add_key(dir_path: Path, name: str) -> dict[str, Any]:
    """生成并追加一个新 key，写盘。name 为空给默认名。"""
    entry = {
        "name": (name or "").strip() or f"key-{int(time.time())}",
        "key": generate_key(),
        "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "last_used": 0.0,
    }
    keys = load_keys(dir_path)
    keys.append(entry)
    save_keys(dir_path, keys)
    return entry


def delete_key(dir_path: Path, key: str) -> bool:
    """删除指定 key（按完整 key 值匹配）；删除成功返回 True。"""
    keys = load_keys(dir_path)
    keep = [k for k in keys if k.get("key") != key]
    if len(keep) == len(keys):
        return False
    save_keys(dir_path, keep)
    return True


def is_valid_key(provided: str | None, keys: list[dict[str, Any]], legacy: str = "") -> bool:
    """校验：面板 key 库任一项匹配，或等于 config 兼容 key。"""
    if not provided:
        return False
    if legacy and provided == legacy:
        return True
    return any(k.get("key") == provided for k in keys)