Function import_tokens_from_directory_api

app/admin/api.py:706–780

从本地目录导入 token 文件。

(request: Request)

@router.post("/tokens/import-directory", dependencies=[Depends(require_auth)])
async def import_tokens_from_directory_api(request: Request):
    """从本地目录导入 token 文件。"""
    from app.core.config import settings
    from app.services.token_automation import run_directory_import

    form_data = await request.form()
    source_dir = str(
        form_data.get("source_dir")
        or settings.TOKEN_AUTO_IMPORT_SOURCE_DIR
        or ""
    ).strip()
    try:
        source_dir = _validate_directory_path(source_dir)
    except ValueError as exc:
        return _build_alert(
            str(exc),
            title="导入失败！",
            level="error",
            status_code=400,
        )

    try:
        summary = await run_directory_import(
            source_dir,
            provider=DEFAULT_TOKEN_NAMESPACE,
            validate=True,
        )
    except (FileNotFoundError, NotADirectoryError) as exc:
        return _build_alert(
            str(exc),
            title="导入失败！",
            level="error",
            status_code=400,
        )
    except RuntimeError as exc:
        return _build_alert(
            str(exc),
            title="导入稍后重试",
            level="warning",
            status_code=409,
        )
    except Exception as exc:
        logger.exception(f"❌ 本地目录导入 Token 失败: {exc}")
        return _build_alert(
            f"目录扫描或入库异常: {exc}",
            title="导入失败！",
            level="error",
            status_code=500,
        )

    if summary.imported_count > 0:
        title = "导入成功！" if summary.failed_count == 0 else "导入完成！"
        detail = (
            f"目录 {summary.source_dir} 共扫描 {summary.scanned_files} 个文件，"
            f"成功导入 {summary.imported_count} 个 Token，"
            f"重复 {summary.duplicate_count} 个，"
            f"无效 JSON {summary.invalid_json_count} 个，"
            f"缺少 token {summary.missing_token_count} 个，"