feat: add admin usage statistics and LLM configuration management

qwen3-tts-backend/db/crud.py

@@ -1,9 +1,10 @@
 import json
 from typing import Optional, List, Dict, Any
 from datetime import datetime
+from sqlalchemy import func
 from sqlalchemy.orm import Session
 
-from db.models import User, Job, VoiceCache, SystemSettings, VoiceDesign, AudiobookProject, AudiobookChapter, AudiobookCharacter, AudiobookSegment
+from db.models import User, Job, VoiceCache, SystemSettings, VoiceDesign, AudiobookProject, AudiobookChapter, AudiobookCharacter, AudiobookSegment, UsageLog
 
 def get_user_by_username(db: Session, username: str) -> Optional[User]:
     return db.query(User).filter(User.username == username).first()
@@ -385,6 +386,33 @@ def update_user_llm_config(
     return user
 
 
+def get_system_setting(db: Session, key: str):
+    setting = db.query(SystemSettings).filter(SystemSettings.key == key).first()
+    return setting.value if setting else None
+
+
+def set_system_setting(db: Session, key: str, value) -> SystemSettings:
+    setting = db.query(SystemSettings).filter(SystemSettings.key == key).first()
+    if setting:
+        setting.value = value
+        setting.updated_at = datetime.utcnow()
+    else:
+        setting = SystemSettings(key=key, value=value)
+        db.add(setting)
+    db.commit()
+    db.refresh(setting)
+    return setting
+
+
+def delete_system_setting(db: Session, key: str) -> bool:
+    setting = db.query(SystemSettings).filter(SystemSettings.key == key).first()
+    if setting:
+        db.delete(setting)
+        db.commit()
+        return True
+    return False
+
+
 def create_audiobook_project(
     db: Session,
     user_id: int,
@@ -668,3 +696,81 @@ def delete_audiobook_segments(db: Session, project_id: int) -> None:
 def delete_audiobook_characters(db: Session, project_id: int) -> None:
     db.query(AudiobookCharacter).filter(AudiobookCharacter.project_id == project_id).delete()
     db.commit()
+
+
+def create_usage_log(
+    db: Session,
+    user_id: int,
+    prompt_tokens: int,
+    completion_tokens: int,
+    model: Optional[str] = None,
+    context: Optional[str] = None,
+) -> UsageLog:
+    log = UsageLog(
+        user_id=user_id,
+        prompt_tokens=prompt_tokens,
+        completion_tokens=completion_tokens,
+        model=model,
+        context=context,
+    )
+    db.add(log)
+    db.commit()
+    return log
+
+
+def get_usage_stats(
+    db: Session,
+    date_from: Optional[datetime] = None,
+    date_to: Optional[datetime] = None,
+) -> List[Dict]:
+    llm_query = db.query(
+        UsageLog.user_id,
+        func.sum(UsageLog.prompt_tokens).label("prompt_tokens"),
+        func.sum(UsageLog.completion_tokens).label("completion_tokens"),
+    ).group_by(UsageLog.user_id)
+    if date_from:
+        llm_query = llm_query.filter(UsageLog.created_at >= date_from)
+    if date_to:
+        llm_query = llm_query.filter(UsageLog.created_at <= date_to)
+    llm_rows = llm_query.all()
+    llm_map: Dict[int, Dict] = {
+        r.user_id: {"prompt_tokens": r.prompt_tokens or 0, "completion_tokens": r.completion_tokens or 0}
+        for r in llm_rows
+    }
+
+    tts_query = db.query(
+        Job.user_id,
+        Job.backend_type,
+        func.count(Job.id).label("job_count"),
+        func.sum(func.coalesce(func.length(Job.input_data), 0)).label("char_count"),
+    ).filter(Job.status == "completed").group_by(Job.user_id, Job.backend_type)
+    if date_from:
+        tts_query = tts_query.filter(Job.created_at >= date_from)
+    if date_to:
+        tts_query = tts_query.filter(Job.created_at <= date_to)
+    tts_rows = tts_query.all()
+    tts_map: Dict[int, List] = {}
+    for r in tts_rows:
+        tts_map.setdefault(r.user_id, []).append({
+            "backend_type": r.backend_type,
+            "job_count": r.job_count,
+            "char_count": r.char_count or 0,
+        })
+
+    user_ids = set(llm_map.keys()) | set(tts_map.keys())
+    users = db.query(User).filter(User.id.in_(user_ids)).all() if user_ids else []
+    user_info = {u.id: {"username": u.username, "email": u.email} for u in users}
+
+    result = []
+    for uid in sorted(user_ids):
+        info = user_info.get(uid, {"username": f"user_{uid}", "email": ""})
+        llm = llm_map.get(uid, {"prompt_tokens": 0, "completion_tokens": 0})
+        result.append({
+            "user_id": uid,
+            "username": info["username"],
+            "email": info["email"],
+            "llm_prompt_tokens": llm["prompt_tokens"],
+            "llm_completion_tokens": llm["completion_tokens"],
+            "tts_backends": tts_map.get(uid, []),
+        })
+    return result