services/database_service.py

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,),
    )