436 lines
14 KiB
Python
436 lines
14 KiB
Python
import json
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Path, Query, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.dependencies import get_admin_user, get_db
|
|
from app.models.api.api_key import ApiKey
|
|
from app.models.api.api_key_upscale_config import ApiKeyUpscaleConfig
|
|
from app.models.api.api_usage_log import ApiUsageLog
|
|
from app.models.user import User
|
|
from app.schemas.admin_api.api_key import (
|
|
ApiKeyCallableModel,
|
|
ApiKeyCreateRequest,
|
|
ApiKeyCreateResponse,
|
|
ApiKeyListItem,
|
|
ApiKeyListOut,
|
|
ApiKeyQuotaAdjustRequest,
|
|
ApiKeyRevealResponse,
|
|
ApiKeyResponse,
|
|
ApiKeyUpdateRequest,
|
|
)
|
|
from app.schemas.admin_api.api_upscale import (
|
|
ApiUpscaleConfigData,
|
|
ApiUpscaleConfigResponse,
|
|
ApiUpscaleConfigSaveRequest,
|
|
)
|
|
from app.schemas.admin_api.api_usage import ApiUsageLogResponse, ApiUsageSummaryResponse
|
|
from app.services.api_v3 import (
|
|
key_service,
|
|
upscale_service,
|
|
usage_log_service,
|
|
)
|
|
|
|
logger = logging.getLogger("videogen")
|
|
|
|
router = APIRouter(prefix="/admin/api-keys", tags=["admin-api-keys"])
|
|
|
|
|
|
def _key_to_list_item(key: ApiKey) -> ApiKeyListItem:
|
|
"""将 ApiKey 模型转为列表项 Schema。"""
|
|
try:
|
|
callable_models = json.loads(key.callable_models) if key.callable_models else []
|
|
except (json.JSONDecodeError, TypeError):
|
|
callable_models = []
|
|
|
|
return ApiKeyListItem(
|
|
id=key.id,
|
|
company_name=key.company_name,
|
|
api_key_prefix=f"{key.api_key_prefix}****",
|
|
description=key.description,
|
|
callable_models=[ApiKeyCallableModel(**m) for m in callable_models],
|
|
quota_limit=key.quota_limit,
|
|
quota_cycle=key.quota_cycle,
|
|
quota_used=key.quota_used,
|
|
is_active=key.is_active,
|
|
valid_from=key.valid_from,
|
|
valid_until=key.valid_until,
|
|
max_concurrent_video_tasks=key.max_concurrent_video_tasks,
|
|
last_used_at=key.last_used_at,
|
|
created_at=key.created_at,
|
|
)
|
|
|
|
|
|
# === API Key CRUD ===
|
|
|
|
@router.get("", response_model=ApiKeyListOut, summary="列出 API Key")
|
|
async def list_keys(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=200),
|
|
company_name: str | None = None,
|
|
is_active: bool | None = None,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> ApiKeyListOut:
|
|
"""列出所有 API Key（分页+筛选）。"""
|
|
total, keys = await key_service.list_api_keys(
|
|
db, skip=skip, limit=limit,
|
|
company_name=company_name, is_active=is_active,
|
|
)
|
|
return ApiKeyListOut(
|
|
total=total,
|
|
items=[_key_to_list_item(k) for k in keys],
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=ApiKeyCreateResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
summary="创建 API Key",
|
|
)
|
|
async def create_key(
|
|
req: ApiKeyCreateRequest,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> ApiKeyCreateResponse:
|
|
"""创建新的 API Key。
|
|
|
|
返回包含完整明文 api_key，仅此一次。
|
|
"""
|
|
callable_models = [m.model_dump() for m in req.callable_models] if req.callable_models else []
|
|
|
|
key, raw_key = await key_service.create_api_key(
|
|
db=db,
|
|
company_name=req.company_name,
|
|
callable_models=callable_models,
|
|
quota_limit=req.quota_limit,
|
|
quota_cycle=req.quota_cycle,
|
|
valid_from=req.valid_from,
|
|
valid_until=req.valid_until,
|
|
max_concurrent_video_tasks=req.max_concurrent_video_tasks,
|
|
description=req.description,
|
|
)
|
|
await db.commit()
|
|
|
|
return ApiKeyCreateResponse(
|
|
id=key.id,
|
|
company_name=key.company_name,
|
|
api_key=raw_key,
|
|
api_key_prefix=key.api_key_prefix,
|
|
valid_until=key.valid_until,
|
|
created_at=key.created_at,
|
|
)
|
|
|
|
|
|
@router.get("/{key_id}/reveal", response_model=ApiKeyRevealResponse, summary="揭秘 API Key")
|
|
async def reveal_key(
|
|
key_id: str = Path(..., description="API Key ID"),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> ApiKeyRevealResponse:
|
|
"""揭秘 API Key（随时可获取完整明文 Key）。"""
|
|
key = await key_service.get_api_key(db, key_id)
|
|
if not key:
|
|
raise HTTPException(status_code=404, detail="API Key 不存在")
|
|
|
|
plaintext = key.decrypt_api_key()
|
|
if not plaintext:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="该 API Key 创建时未启用加密存储，无法揭秘。请重新创建 Key。",
|
|
)
|
|
|
|
return ApiKeyRevealResponse(
|
|
id=key.id,
|
|
company_name=key.company_name,
|
|
api_key=plaintext,
|
|
api_key_prefix=key.api_key_prefix,
|
|
)
|
|
|
|
|
|
@router.get("/{key_id}", response_model=ApiKeyListItem, summary="获取 API Key 详情")
|
|
async def get_key(
|
|
key_id: str = Path(..., description="API Key ID"),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> ApiKeyListItem:
|
|
"""获取单个 API Key 详情。"""
|
|
key = await key_service.get_api_key(db, key_id)
|
|
if not key:
|
|
raise HTTPException(status_code=404, detail="API Key 不存在")
|
|
return _key_to_list_item(key)
|
|
|
|
|
|
@router.put("/{key_id}", response_model=ApiKeyListItem, summary="更新 API Key")
|
|
async def update_key(
|
|
req: ApiKeyUpdateRequest,
|
|
key_id: str = Path(...),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> ApiKeyResponse:
|
|
"""更新 API Key 配置。"""
|
|
key = await key_service.get_api_key(db, key_id)
|
|
if not key:
|
|
raise HTTPException(status_code=404, detail="API Key 不存在")
|
|
|
|
# model_dump 默认输出 snake_case 字段名，直接传给 service
|
|
update_data = req.model_dump(exclude_none=True)
|
|
if "callable_models" in update_data and update_data["callable_models"] is not None:
|
|
update_data["callable_models"] = [
|
|
m.model_dump() if hasattr(m, "model_dump") else m
|
|
for m in update_data["callable_models"]
|
|
]
|
|
|
|
key = await key_service.update_api_key(db, key, **update_data)
|
|
await db.commit()
|
|
return _key_to_list_item(key)
|
|
|
|
|
|
@router.delete("/{key_id}", summary="删除 API Key")
|
|
async def delete_key(
|
|
key_id: str = Path(...),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> dict:
|
|
"""软删除 API Key。"""
|
|
key = await key_service.get_api_key(db, key_id)
|
|
if not key:
|
|
raise HTTPException(status_code=404, detail="API Key 不存在")
|
|
|
|
await key_service.delete_api_key(db, key)
|
|
await db.commit()
|
|
return {"status": "deleted", "id": key_id}
|
|
|
|
|
|
# === 超分配置 ===
|
|
|
|
@router.get("/{key_id}/upscale", response_model=ApiUpscaleConfigResponse, summary="获取 API Key 超分配置")
|
|
async def get_upscale_config(
|
|
key_id: str = Path(...),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> ApiUpscaleConfigResponse:
|
|
"""获取 API Key 的超分配置。"""
|
|
config = await upscale_service.get_or_create_upscale_config(db, key_id)
|
|
try:
|
|
rules = json.loads(config.rules_json) if config.rules_json else []
|
|
except (json.JSONDecodeError, TypeError):
|
|
rules = []
|
|
|
|
return ApiUpscaleConfigResponse(
|
|
data=ApiUpscaleConfigData(
|
|
enabled=config.enabled,
|
|
delete_source_after_success=config.delete_source_after_success,
|
|
rules=rules,
|
|
),
|
|
)
|
|
|
|
|
|
@router.put("/{key_id}/upscale", response_model=ApiUpscaleConfigResponse, summary="保存 API Key 超分配置")
|
|
async def save_upscale_config(
|
|
req: ApiUpscaleConfigSaveRequest,
|
|
key_id: str = Path(...),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> ApiUpscaleConfigResponse:
|
|
"""保存 API Key 的超分配置。"""
|
|
config = await upscale_service.save_upscale_config(
|
|
db=db,
|
|
api_key_id=key_id,
|
|
enabled=req.data.enabled,
|
|
delete_source_after_success=req.data.delete_source_after_success,
|
|
rules=[r.model_dump() for r in req.data.rules],
|
|
)
|
|
await db.commit()
|
|
|
|
return ApiUpscaleConfigResponse(
|
|
data=ApiUpscaleConfigData(
|
|
enabled=config.enabled,
|
|
delete_source_after_success=config.delete_source_after_success,
|
|
rules=json.loads(config.rules_json) if config.rules_json else [],
|
|
),
|
|
)
|
|
|
|
|
|
# === 使用日志 ===
|
|
|
|
@router.get("/{key_id}/usage", response_model=ApiUsageSummaryResponse, summary="获取 API Key 使用统计")
|
|
async def get_usage(
|
|
key_id: str = Path(...),
|
|
days: int = Query(30, ge=1, le=365),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> ApiUsageSummaryResponse:
|
|
"""获取 API Key 的使用统计和明细。"""
|
|
# 验证 key 存在
|
|
key = await key_service.get_api_key(db, key_id)
|
|
if not key:
|
|
raise HTTPException(status_code=404, detail="API Key 不存在")
|
|
|
|
summary = await usage_log_service.get_usage_summary(db, api_key_id=key_id, days=days)
|
|
total, logs = await usage_log_service.list_usage_logs(db, api_key_id=key_id, limit=page_size, skip=(page - 1) * page_size)
|
|
|
|
return ApiUsageSummaryResponse(
|
|
total_requests=summary["total_requests"],
|
|
total_credits_cost=summary["total_credits_cost"],
|
|
total_tokens_used=summary["total_tokens_used"],
|
|
success_count=summary["success_count"],
|
|
failed_count=summary["failed_count"],
|
|
avg_duration_ms=summary["avg_duration_ms"],
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
items=[
|
|
ApiUsageLogResponse(
|
|
id=log.id,
|
|
api_key_id=log.api_key_id,
|
|
api_generation_task_id=log.api_generation_task_id,
|
|
request_type=log.request_type,
|
|
model_name=log.model_name,
|
|
gen_type=log.gen_type,
|
|
credits_cost=log.credits_cost,
|
|
tokens_used=log.tokens_used,
|
|
request_duration_ms=log.request_duration_ms,
|
|
status=log.status,
|
|
error_message=log.error_message,
|
|
error_code=log.error_code,
|
|
created_at=log.created_at,
|
|
)
|
|
for log in logs
|
|
],
|
|
)
|
|
|
|
|
|
# === 整体消耗列表 ===
|
|
|
|
@router.get("/usage/all", response_model=dict, summary="获取整体 API 消耗列表")
|
|
async def list_all_usage(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=200),
|
|
api_key_id: str | None = None,
|
|
gen_type: str | None = None,
|
|
status_filter: str | None = Query(None, alias="status"),
|
|
start_date: datetime | None = None,
|
|
end_date: datetime | None = None,
|
|
search: str | None = None,
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> dict:
|
|
"""获取整体 API 消耗列表（跨所有 Key，支持筛选和分页）。"""
|
|
# 构建查询
|
|
query = select(ApiUsageLog, ApiKey.company_name, ApiKey.api_key_prefix).join(
|
|
ApiKey, ApiUsageLog.api_key_id == ApiKey.id
|
|
)
|
|
count_query = select(func.count(ApiUsageLog.id)).join(
|
|
ApiKey, ApiUsageLog.api_key_id == ApiKey.id
|
|
)
|
|
|
|
# 筛选条件
|
|
filters = []
|
|
if api_key_id:
|
|
filters.append(ApiUsageLog.api_key_id == api_key_id)
|
|
if gen_type:
|
|
filters.append(ApiUsageLog.gen_type == gen_type)
|
|
if status_filter:
|
|
filters.append(ApiUsageLog.status == status_filter)
|
|
if start_date:
|
|
filters.append(ApiUsageLog.created_at >= start_date)
|
|
if end_date:
|
|
filters.append(ApiUsageLog.created_at <= end_date)
|
|
if search:
|
|
search_pattern = f"%{search}%"
|
|
filters.append(
|
|
(ApiKey.company_name.ilike(search_pattern))
|
|
| (ApiKey.api_key_prefix.ilike(search_pattern))
|
|
)
|
|
|
|
for f in filters:
|
|
query = query.where(f)
|
|
count_query = count_query.where(f)
|
|
|
|
# 总数
|
|
total_result = await db.execute(count_query)
|
|
total = total_result.scalar_one()
|
|
|
|
# 分页查询
|
|
query = query.order_by(ApiUsageLog.created_at.desc()).offset(skip).limit(limit)
|
|
result = await db.execute(query)
|
|
rows = result.all()
|
|
|
|
items = []
|
|
for log, company_name, key_prefix in rows:
|
|
items.append({
|
|
"id": log.id,
|
|
"apiKeyId": log.api_key_id,
|
|
"companyName": company_name,
|
|
"apiKeyPrefix": f"{key_prefix}****" if key_prefix else None,
|
|
"taskId": log.api_generation_task_id,
|
|
"requestType": log.request_type,
|
|
"modelName": log.model_name,
|
|
"genType": log.gen_type,
|
|
"creditsCost": log.credits_cost,
|
|
"tokensUsed": log.tokens_used,
|
|
"requestDurationMs": log.request_duration_ms,
|
|
"duration": log.duration,
|
|
"resolution": log.resolution,
|
|
"status": log.status,
|
|
"errorMessage": log.error_message,
|
|
"errorCode": log.error_code,
|
|
"createdAt": log.created_at.isoformat() if log.created_at else None,
|
|
})
|
|
|
|
return {
|
|
"total": total,
|
|
"items": items,
|
|
}
|
|
|
|
|
|
@router.post("/{key_id}/quota-adjust", response_model=ApiKeyListItem, summary="调整 API Key 配额")
|
|
async def quota_adjust(
|
|
req: ApiKeyQuotaAdjustRequest,
|
|
key_id: str = Path(..., description="API Key ID"),
|
|
admin: User = Depends(get_admin_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> ApiKeyListItem:
|
|
"""调整 API Key 配额（增加总额/重置已用/设置限额/修改周期）。"""
|
|
key = await key_service.get_api_key(db, key_id)
|
|
if not key:
|
|
raise HTTPException(status_code=404, detail="API Key 不存在")
|
|
|
|
key, changes = await key_service.adjust_quota(
|
|
db,
|
|
key,
|
|
action=req.action,
|
|
quota_limit_delta=req.quota_limit_delta,
|
|
quota_limit=req.quota_limit,
|
|
quota_cycle=req.quota_cycle,
|
|
)
|
|
|
|
# 审计日志
|
|
try:
|
|
from app.services.operation_log import log_operation
|
|
await log_operation(
|
|
db=db,
|
|
user_id=str(admin.id),
|
|
username=str(admin.username),
|
|
action=f"quota_adjust:{req.action}",
|
|
method="POST",
|
|
path=f"/admin/api-keys/{key_id}/quota-adjust",
|
|
detail=json.dumps(
|
|
{**changes, "reason": req.reason},
|
|
ensure_ascii=False,
|
|
default=str,
|
|
),
|
|
)
|
|
except Exception as log_exc:
|
|
logger.warning("配额调整审计日志记录失败: %s", log_exc)
|
|
|
|
await db.commit()
|
|
return _key_to_list_item(key)
|