import os import json import uuid import aiosqlite from typing import Any, Dict, List, Optional from services.config_service import get_config _db_path: Optional[str] = None def _get_db_path() -> str: global _db_path if _db_path is None: _db_path = os.path.join(os.path.dirname(__file__), "..", "data", "basketball.db") os.makedirs(os.path.dirname(_db_path), exist_ok=True) return _db_path async def get_db() -> aiosqlite.Connection: db = await aiosqlite.connect(_get_db_path()) db.row_factory = aiosqlite.Row return db async def init_db(): db = await get_db() try: await db.executescript(""" CREATE TABLE IF NOT EXISTS users ( user_id TEXT PRIMARY KEY, email TEXT UNIQUE, github_id TEXT UNIQUE, nickname TEXT, avatar_url TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_login_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS videos ( video_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, original_filename TEXT, status TEXT DEFAULT 'uploading', mode TEXT DEFAULT 'balanced', highlights_json TEXT, duration_seconds REAL, file_size_bytes INTEGER, person_query TEXT DEFAULT '', confirmed_track_id INTEGER DEFAULT NULL, original_deleted INTEGER DEFAULT 0, last_accessed_at DATETIME DEFAULT NULL, cleanup_reason TEXT DEFAULT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, expires_at DATETIME, FOREIGN KEY (user_id) REFERENCES users(user_id) ); CREATE TABLE IF NOT EXISTS email_verifications ( email TEXT NOT NULL, code TEXT NOT NULL, expires_at DATETIME NOT NULL, used INTEGER DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_videos_user_id ON videos(user_id); CREATE INDEX IF NOT EXISTS idx_videos_expires_at ON videos(expires_at); CREATE INDEX IF NOT EXISTS idx_users_email ON users(email); CREATE INDEX IF NOT EXISTS idx_users_github_id ON users(github_id); """) await db.commit() # 迁移:为已有数据库添加新字段 try: await db.execute("ALTER TABLE videos ADD COLUMN original_deleted INTEGER DEFAULT 0") except Exception: pass try: await db.execute("ALTER TABLE videos ADD COLUMN last_accessed_at DATETIME DEFAULT NULL") except Exception: pass try: await db.execute("ALTER TABLE videos ADD COLUMN cleanup_reason TEXT DEFAULT NULL") except Exception: pass await db.commit() finally: await db.close() async def execute_query(sql: str, params: tuple = ()) -> List[Dict[str, Any]]: db = await get_db() try: cursor = await db.execute(sql, params) rows = await cursor.fetchall() return [dict(row) for row in rows] finally: await db.close() async def execute_update(sql: str, params: tuple = ()) -> str: db = await get_db() try: cursor = await db.execute(sql, params) await db.commit() return cursor.lastrowid finally: await db.close() async def get_user_by_email(email: str) -> Optional[Dict[str, Any]]: rows = await execute_query("SELECT * FROM users WHERE email = ?", (email,)) return rows[0] if rows else None async def get_user_by_github_id(github_id: str) -> Optional[Dict[str, Any]]: rows = await execute_query("SELECT * FROM users WHERE github_id = ?", (github_id,)) return rows[0] if rows else None async def get_user_by_id(user_id: str) -> Optional[Dict[str, Any]]: rows = await execute_query("SELECT * FROM users WHERE user_id = ?", (user_id,)) return rows[0] if rows else None async def create_user(email: str = None, github_id: str = None, nickname: str = None, avatar_url: str = None) -> str: user_id = str(uuid.uuid4()) await execute_update( "INSERT INTO users (user_id, email, github_id, nickname, avatar_url) VALUES (?, ?, ?, ?, ?)", (user_id, email, github_id, nickname, avatar_url), ) return user_id async def update_user_login(user_id: str): await execute_update( "UPDATE users SET last_login_at = CURRENT_TIMESTAMP WHERE user_id = ?", (user_id,), ) async def save_verification_code(email: str, code: str, expires_at: str): await execute_update( "INSERT INTO email_verifications (email, code, expires_at) VALUES (?, ?, ?)", (email, code, expires_at), ) async def verify_code(email: str, code: str) -> bool: rows = await execute_query( "SELECT * FROM email_verifications WHERE email = ? AND code = ? AND used = 0 AND expires_at > datetime('now')", (email, code), ) if rows: await execute_update( "UPDATE email_verifications SET used = 1 WHERE email = ? AND code = ?", (email, code), ) return True return False async def create_video(user_id: str, original_filename: str, mode: str, file_size_bytes: int = 0, person_query: str = "") -> str: video_id = str(uuid.uuid4()) config = get_config() highlight_days = config.get("cleanup", {}).get("highlight_landscape_days", 7) await execute_update( "INSERT INTO videos (video_id, user_id, original_filename, mode, file_size_bytes, person_query, expires_at) VALUES (?, ?, ?, ?, ?, ?, datetime('now', '+' || ? || ' days'))", (video_id, user_id, original_filename, mode, file_size_bytes, person_query, str(highlight_days)), ) return video_id async def update_video_status(video_id: str, status: str, highlights_json: str = None, duration_seconds: float = None): parts = ["status = ?"] params = [status] if highlights_json is not None: parts.append("highlights_json = ?") params.append(highlights_json) if duration_seconds is not None: parts.append("duration_seconds = ?") params.append(duration_seconds) params.append(video_id) await execute_update(f"UPDATE videos SET {', '.join(parts)} WHERE video_id = ?", tuple(params)) async def get_video(video_id: str) -> Optional[Dict[str, Any]]: rows = await execute_query("SELECT * FROM videos WHERE video_id = ?", (video_id,)) return rows[0] if rows else None async def get_user_videos(user_id: str) -> List[Dict[str, Any]]: return await execute_query( "SELECT * FROM videos WHERE user_id = ? ORDER BY created_at DESC", (user_id,), ) async def delete_video(video_id: str): await execute_update("DELETE FROM videos WHERE video_id = ?", (video_id,)) async def get_confirmed_track_id(video_id: str) -> Optional[int]: rows = await execute_query("SELECT confirmed_track_id FROM videos WHERE video_id = ?", (video_id,)) if rows and rows[0].get("confirmed_track_id") is not None: return int(rows[0]["confirmed_track_id"]) return None async def set_confirmed_track_id(video_id: str, track_id: int): await execute_update( "UPDATE videos SET confirmed_track_id = ? WHERE video_id = ?", (track_id, video_id), ) async def update_video_original_deleted(video_id: str, deleted: bool = True): await execute_update( "UPDATE videos SET original_deleted = ? WHERE video_id = ?", (1 if deleted else 0, video_id), ) async def update_video_last_accessed(video_id: str): await execute_update( "UPDATE videos SET last_accessed_at = CURRENT_TIMESTAMP WHERE video_id = ?", (video_id,), ) async def update_video_cleanup_reason(video_id: str, reason: str): await execute_update( "UPDATE videos SET cleanup_reason = ? WHERE video_id = ?", (reason, video_id), ) async def acknowledge_history(video_id: str, user_id: str) -> bool: """确认已阅被清理的历史卡片(仅限 cleanup_reason 非空)""" rows = await execute_query( "SELECT cleanup_reason FROM videos WHERE video_id = ? AND user_id = ?", (video_id, user_id), ) if not rows or not rows[0].get("cleanup_reason"): return False await delete_video(video_id) return True async def acknowledge_all_history(user_id: str) -> int: """批量确认已阅所有被清理的历史卡片""" rows = await execute_query( "SELECT video_id FROM videos WHERE user_id = ? AND cleanup_reason IS NOT NULL", (user_id,), ) count = 0 for row in rows: await delete_video(row["video_id"]) count += 1 return count async def get_lru_candidates(limit: int = 100) -> List[Dict[str, Any]]: """获取 LRU 清理候选视频(按访问时间升序,未访问的排最前)""" return await execute_query( "SELECT * FROM videos WHERE status = 'completed' AND original_deleted = 1 AND (cleanup_reason IS NULL OR cleanup_reason = '') ORDER BY last_accessed_at ASC NULLS FIRST LIMIT ?", (limit,), )