Files
mediaplayer/sentence_api/repository.py
2026-08-30 17:07:55 +08:00

945 lines
36 KiB
Python

import json
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
import uuid
from .auth import hash_password, hash_token, new_token, verify_password
from .models import SentenceBoundary, SentenceBoundaryDocument
from .store import normalize_video_hash
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
class VideoRepository:
def __init__(self, database_path: Path):
self.database_path = Path(database_path)
self.database_path.parent.mkdir(parents=True, exist_ok=True)
self._initialize()
def _connect(self) -> sqlite3.Connection:
connection = sqlite3.connect(self.database_path, timeout=30)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys = ON")
connection.execute("PRAGMA journal_mode = WAL")
return connection
def _initialize(self) -> None:
with self._connect() as connection:
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS videos (
video_hash TEXT PRIMARY KEY,
title TEXT NOT NULL,
filename TEXT NOT NULL,
stored_filename TEXT NOT NULL,
content_type TEXT NOT NULL,
size_bytes INTEGER NOT NULL,
duration_ms INTEGER,
language TEXT,
status TEXT NOT NULL,
error_message TEXT,
algorithm_version TEXT,
transcription TEXT,
remote_url TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS sentences (
video_hash TEXT NOT NULL,
sentence_index INTEGER NOT NULL,
start_ms INTEGER NOT NULL,
end_ms INTEGER NOT NULL,
text TEXT,
language TEXT,
reference_speech_duration_ms INTEGER,
PRIMARY KEY (video_hash, sentence_index),
FOREIGN KEY (video_hash) REFERENCES videos(video_hash) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS attempts (
attempt_id TEXT PRIMARY KEY,
video_hash TEXT NOT NULL,
sentence_index INTEGER NOT NULL,
result_json TEXT NOT NULL,
audio_filename TEXT,
created_at TEXT NOT NULL,
FOREIGN KEY (video_hash) REFERENCES videos(video_hash) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
nickname TEXT NOT NULL,
password_hash TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS auth_tokens (
token_hash TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
created_at TEXT NOT NULL,
expires_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS enrollments (
user_id TEXT NOT NULL,
video_hash TEXT NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY (user_id, video_hash),
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY (video_hash) REFERENCES videos(video_hash) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS dub_shares (
share_id TEXT PRIMARY KEY,
video_hash TEXT NOT NULL,
title TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY (video_hash) REFERENCES videos(video_hash) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS dub_share_segments (
id INTEGER PRIMARY KEY AUTOINCREMENT,
share_id TEXT NOT NULL,
sentence_index INTEGER NOT NULL,
start_ms INTEGER NOT NULL,
end_ms INTEGER NOT NULL,
text TEXT,
audio_filename TEXT NOT NULL,
audio_size_bytes INTEGER NOT NULL,
overall_score REAL NOT NULL DEFAULT 0,
recognized_text TEXT NOT NULL DEFAULT '',
score_details TEXT NOT NULL DEFAULT '{}',
FOREIGN KEY (share_id) REFERENCES dub_shares(share_id) ON DELETE CASCADE,
UNIQUE (share_id, sentence_index)
);
"""
)
with self._connect() as connection:
columns = {
row["name"]
for row in connection.execute("PRAGMA table_info(dub_share_segments)")
}
if "overall_score" not in columns:
connection.execute(
"ALTER TABLE dub_share_segments ADD COLUMN overall_score REAL NOT NULL DEFAULT 0"
)
if "recognized_text" not in columns:
connection.execute(
"ALTER TABLE dub_share_segments ADD COLUMN recognized_text TEXT NOT NULL DEFAULT ''"
)
if "score_details" not in columns:
connection.execute(
"ALTER TABLE dub_share_segments ADD COLUMN score_details TEXT NOT NULL DEFAULT '{}'"
)
with self._connect() as connection:
attempt_columns = {
row["name"]
for row in connection.execute("PRAGMA table_info(attempts)")
}
if "user_id" not in attempt_columns:
connection.execute("ALTER TABLE attempts ADD COLUMN user_id TEXT NOT NULL DEFAULT ''")
connection.execute("CREATE INDEX IF NOT EXISTS idx_attempts_user ON attempts(user_id)")
with self._connect() as connection:
dub_columns = {
row["name"]
for row in connection.execute("PRAGMA table_info(dub_shares)")
}
if "user_id" not in dub_columns:
connection.execute("ALTER TABLE dub_shares ADD COLUMN user_id TEXT NOT NULL DEFAULT ''")
connection.execute("CREATE INDEX IF NOT EXISTS idx_dub_shares_user ON dub_shares(user_id)")
with self._connect() as connection:
video_columns = {
row["name"]
for row in connection.execute("PRAGMA table_info(videos)")
}
if "remote_url" not in video_columns:
connection.execute("ALTER TABLE videos ADD COLUMN remote_url TEXT")
def upsert_upload(
self,
*,
video_hash: str,
title: str,
filename: str,
stored_filename: str,
content_type: str,
size_bytes: int,
language: Optional[str],
) -> None:
normalized_hash = normalize_video_hash(video_hash)
now = utc_now()
with self._connect() as connection:
existing = connection.execute(
"SELECT created_at FROM videos WHERE video_hash = ?",
(normalized_hash,),
).fetchone()
created_at = existing["created_at"] if existing else now
connection.execute(
"""
INSERT INTO videos (
video_hash, title, filename, stored_filename, content_type,
size_bytes, duration_ms, language, status, error_message,
algorithm_version, transcription, remote_url, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, NULL, ?, 'uploaded', NULL, NULL, NULL, NULL, ?, ?)
ON CONFLICT(video_hash) DO UPDATE SET
title = excluded.title,
filename = excluded.filename,
stored_filename = excluded.stored_filename,
content_type = excluded.content_type,
size_bytes = excluded.size_bytes,
language = excluded.language,
status = 'uploaded',
error_message = NULL,
updated_at = excluded.updated_at
""",
(
normalized_hash,
title,
filename,
stored_filename,
content_type,
size_bytes,
language,
created_at,
now,
),
)
def mark_processing(self, video_hash: str) -> None:
self._update_status(video_hash, "processing", None)
def mark_failed(self, video_hash: str, message: str) -> None:
self._update_status(video_hash, "failed", message[:4000])
def _update_status(self, video_hash: str, status: str, error_message: Optional[str]) -> None:
with self._connect() as connection:
connection.execute(
"UPDATE videos SET status = ?, error_message = ?, updated_at = ? WHERE video_hash = ?",
(status, error_message, utc_now(), normalize_video_hash(video_hash)),
)
def save_processing_result(
self,
document: SentenceBoundaryDocument,
transcription: Optional[str],
ready: bool = True,
) -> None:
with self._connect() as connection:
connection.execute(
"DELETE FROM sentences WHERE video_hash = ?",
(document.video_hash,),
)
connection.executemany(
"""
INSERT INTO sentences (
video_hash, sentence_index, start_ms, end_ms, text,
language, reference_speech_duration_ms
) VALUES (?, ?, ?, ?, ?, ?, ?)
""",
[
(
document.video_hash,
sentence.index,
sentence.start_ms,
sentence.end_ms,
sentence.text,
sentence.language,
sentence.reference_speech_duration_ms,
)
for sentence in document.sentences
],
)
connection.execute(
"""
UPDATE videos SET duration_ms = ?, status = ?, error_message = NULL,
algorithm_version = ?, transcription = ?, updated_at = ?
WHERE video_hash = ?
""",
(
document.duration_ms,
"ready" if ready else "processing",
document.algorithm_version,
transcription,
utc_now(),
document.video_hash,
),
)
def mark_oss_uploaded(self, video_hash: str, remote_url: str) -> None:
with self._connect() as connection:
connection.execute(
"""
UPDATE videos
SET remote_url = ?, status = 'ready', error_message = NULL, updated_at = ?
WHERE video_hash = ?
""",
(remote_url, utc_now(), normalize_video_hash(video_hash)),
)
def list_videos(self) -> List[Dict[str, Any]]:
with self._connect() as connection:
rows = connection.execute(
"""
SELECT v.*, COUNT(s.sentence_index) AS sentence_count
FROM videos v
LEFT JOIN sentences s ON s.video_hash = v.video_hash
GROUP BY v.video_hash
ORDER BY v.created_at DESC
"""
).fetchall()
return [dict(row) for row in rows]
def get_video(self, video_hash: str) -> Optional[Dict[str, Any]]:
with self._connect() as connection:
row = connection.execute(
"""
SELECT v.*, COUNT(s.sentence_index) AS sentence_count
FROM videos v
LEFT JOIN sentences s ON s.video_hash = v.video_hash
WHERE v.video_hash = ?
GROUP BY v.video_hash
""",
(normalize_video_hash(video_hash),),
).fetchone()
return dict(row) if row else None
def get_document(self, video_hash: str) -> Optional[SentenceBoundaryDocument]:
video = self.get_video(video_hash)
if video is None or video["duration_ms"] is None:
return None
with self._connect() as connection:
rows = connection.execute(
"SELECT * FROM sentences WHERE video_hash = ? ORDER BY sentence_index",
(normalize_video_hash(video_hash),),
).fetchall()
return SentenceBoundaryDocument(
video_hash=video["video_hash"],
duration_ms=video["duration_ms"],
algorithm_version=video["algorithm_version"] or "unknown",
sentences=[
SentenceBoundary(
index=row["sentence_index"],
start_ms=row["start_ms"],
end_ms=row["end_ms"],
text=row["text"],
language=row["language"],
reference_speech_duration_ms=row["reference_speech_duration_ms"],
)
for row in rows
],
)
def update_sentence_text(
self,
video_hash: str,
sentence_index: int,
text: str,
language: Optional[str],
) -> Optional[SentenceBoundary]:
normalized_hash = normalize_video_hash(video_hash)
with self._connect() as connection:
cursor = connection.execute(
"""
UPDATE sentences SET text = ?, language = COALESCE(?, language)
WHERE video_hash = ? AND sentence_index = ?
""",
(text.strip(), language, normalized_hash, sentence_index),
)
if cursor.rowcount == 0:
return None
connection.execute(
"UPDATE videos SET updated_at = ? WHERE video_hash = ?",
(utc_now(), normalized_hash),
)
row = connection.execute(
"SELECT * FROM sentences WHERE video_hash = ? AND sentence_index = ?",
(normalized_hash, sentence_index),
).fetchone()
return SentenceBoundary(
index=row["sentence_index"],
start_ms=row["start_ms"],
end_ms=row["end_ms"],
text=row["text"],
language=row["language"],
reference_speech_duration_ms=row["reference_speech_duration_ms"],
)
def get_sentence(
self,
video_hash: str,
sentence_index: int,
) -> Optional[SentenceBoundary]:
normalized_hash = normalize_video_hash(video_hash)
with self._connect() as connection:
row = connection.execute(
"""
SELECT * FROM sentences
WHERE video_hash = ? AND sentence_index = ?
""",
(normalized_hash, sentence_index),
).fetchone()
if row is None:
return None
return SentenceBoundary(
index=row["sentence_index"],
start_ms=row["start_ms"],
end_ms=row["end_ms"],
text=row["text"],
language=row["language"],
reference_speech_duration_ms=row["reference_speech_duration_ms"],
)
def replace_sentence(
self,
video_hash: str,
sentence_index: int,
replacements: List[SentenceBoundary],
) -> Optional[List[SentenceBoundary]]:
"""Replace one sentence row with one or more contiguous sentence rows."""
normalized_hash = normalize_video_hash(video_hash)
if not replacements:
raise ValueError("At least one replacement sentence is required.")
with self._connect() as connection:
source = connection.execute(
"""
SELECT * FROM sentences
WHERE video_hash = ? AND sentence_index = ?
""",
(normalized_hash, sentence_index),
).fetchone()
if source is None:
return None
rows = [
SentenceBoundary(
index=index,
start_ms=replacement.start_ms,
end_ms=replacement.end_ms,
text=replacement.text.strip(),
language=replacement.language,
reference_speech_duration_ms=max(
1, replacement.reference_speech_duration_ms or 1
),
)
for index, replacement in enumerate(replacements)
]
previous_end = source["start_ms"]
for replacement in rows:
if (
replacement.start_ms < previous_end
or replacement.end_ms <= replacement.start_ms
):
raise ValueError("Replacement sentence times must be contiguous.")
if replacement.end_ms > source["end_ms"]:
raise ValueError(
"Replacement sentences cannot exceed the source sentence."
)
previous_end = replacement.end_ms
if rows[-1].end_ms != source["end_ms"]:
raise ValueError("Replacement sentences must cover the source sentence.")
# Shift following rows down first. Descending indexes avoid violating
# the (video_hash, sentence_index) primary key while making room.
added_count = len(rows) - 1
if added_count:
following = [
row["sentence_index"]
for row in connection.execute(
"""
SELECT sentence_index FROM sentences
WHERE video_hash = ? AND sentence_index > ?
ORDER BY sentence_index DESC
""",
(normalized_hash, sentence_index),
).fetchall()
]
for old_index in following:
connection.execute(
"""
UPDATE sentences SET sentence_index = ?
WHERE video_hash = ? AND sentence_index = ?
""",
(old_index + added_count, normalized_hash, old_index),
)
first = rows[0]
connection.execute(
"""
UPDATE sentences
SET start_ms = ?, end_ms = ?, text = ?, language = ?,
reference_speech_duration_ms = ?
WHERE video_hash = ? AND sentence_index = ?
""",
(
first.start_ms,
first.end_ms,
first.text,
first.language,
first.reference_speech_duration_ms,
normalized_hash,
sentence_index,
),
)
for replacement in rows[1:]:
connection.execute(
"""
INSERT INTO sentences (
video_hash, sentence_index, start_ms, end_ms, text,
language, reference_speech_duration_ms
) VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
normalized_hash,
sentence_index + replacement.index,
replacement.start_ms,
replacement.end_ms,
replacement.text,
replacement.language,
replacement.reference_speech_duration_ms,
),
)
connection.execute(
"UPDATE videos SET updated_at = ? WHERE video_hash = ?",
(utc_now(), normalized_hash),
)
result_rows = connection.execute(
"""
SELECT * FROM sentences
WHERE video_hash = ? AND sentence_index >= ? AND sentence_index < ?
ORDER BY sentence_index
""",
(normalized_hash, sentence_index, sentence_index + len(rows)),
).fetchall()
return [
SentenceBoundary(
index=row["sentence_index"],
start_ms=row["start_ms"],
end_ms=row["end_ms"],
text=row["text"],
language=row["language"],
reference_speech_duration_ms=row["reference_speech_duration_ms"],
)
for row in result_rows
]
def adjust_sentence_boundary(
self,
video_hash: str,
sentence_index: int,
delta_ms: int,
) -> Optional[List[SentenceBoundary]]:
"""Move one sentence end and link the next sentence's start to it."""
normalized_hash = normalize_video_hash(video_hash)
with self._connect() as connection:
video = connection.execute(
"SELECT duration_ms FROM videos WHERE video_hash = ?",
(normalized_hash,),
).fetchone()
if video is None or video["duration_ms"] is None:
return None
current = connection.execute(
"""
SELECT * FROM sentences
WHERE video_hash = ? AND sentence_index = ?
""",
(normalized_hash, sentence_index),
).fetchone()
if current is None:
return None
following = connection.execute(
"""
SELECT * FROM sentences
WHERE video_hash = ? AND sentence_index = ?
""",
(normalized_hash, sentence_index + 1),
).fetchone()
duration_ms = int(video["duration_ms"])
minimum_end = current["start_ms"] + 1
maximum_end = duration_ms
if following is not None:
maximum_end = min(maximum_end, following["end_ms"] - 1)
new_end_ms = current["end_ms"] + delta_ms
if new_end_ms < minimum_end or new_end_ms > maximum_end:
raise ValueError(
"The adjusted boundary must remain inside both sentences."
)
current_duration_ms = new_end_ms - current["start_ms"]
current_reference_ms = current["reference_speech_duration_ms"] or current_duration_ms
current_reference_ms = min(
current_duration_ms,
max(1, current_reference_ms + delta_ms),
)
connection.execute(
"""
UPDATE sentences
SET end_ms = ?, reference_speech_duration_ms = ?
WHERE video_hash = ? AND sentence_index = ?
""",
(
new_end_ms,
current_reference_ms,
normalized_hash,
sentence_index,
),
)
updated_indexes = [sentence_index]
if following is not None:
old_start_ms = following["start_ms"]
next_duration_ms = following["end_ms"] - new_end_ms
next_reference_ms = following["reference_speech_duration_ms"] or next_duration_ms
next_reference_ms = min(
next_duration_ms,
max(1, next_reference_ms - (new_end_ms - old_start_ms)),
)
connection.execute(
"""
UPDATE sentences
SET start_ms = ?, reference_speech_duration_ms = ?
WHERE video_hash = ? AND sentence_index = ?
""",
(
new_end_ms,
next_reference_ms,
normalized_hash,
sentence_index + 1,
),
)
updated_indexes.append(sentence_index + 1)
connection.execute(
"UPDATE videos SET updated_at = ? WHERE video_hash = ?",
(utc_now(), normalized_hash),
)
placeholders = ", ".join("?" for _ in updated_indexes)
result_rows = connection.execute(
f"""
SELECT * FROM sentences
WHERE video_hash = ? AND sentence_index IN ({placeholders})
ORDER BY sentence_index
""",
(normalized_hash, *updated_indexes),
).fetchall()
return [
SentenceBoundary(
index=row["sentence_index"],
start_ms=row["start_ms"],
end_ms=row["end_ms"],
text=row["text"],
language=row["language"],
reference_speech_duration_ms=row["reference_speech_duration_ms"],
)
for row in result_rows
]
def create_dub_share(
self,
*,
video_hash: str,
title: str,
segments: List[Dict[str, Any]],
user_id: str,
) -> Optional[Dict[str, Any]]:
normalized_hash = normalize_video_hash(video_hash)
if self.get_video(normalized_hash) is None:
return None
share_id = uuid.uuid4().hex
now = utc_now()
with self._connect() as connection:
connection.execute(
"""
INSERT INTO dub_shares (share_id, video_hash, title, user_id, created_at)
VALUES (?, ?, ?, ?, ?)
""",
(share_id, normalized_hash, title[:200], user_id, now),
)
connection.executemany(
"""
INSERT INTO dub_share_segments (
share_id, sentence_index, start_ms, end_ms, text,
audio_filename, audio_size_bytes, overall_score, recognized_text,
score_details
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
(
share_id,
segment["sentence_index"],
segment["start_ms"],
segment["end_ms"],
segment["text"],
segment["audio_filename"],
segment["audio_size_bytes"],
segment.get("overall_score", 0),
segment.get("recognized_text", ""),
segment.get("score_details", "{}"),
)
for segment in segments
],
)
return {
"share_id": share_id,
"video_hash": normalized_hash,
"title": title[:200],
"created_at": now,
}
def get_dub_share(self, share_id: str) -> Optional[Dict[str, Any]]:
with self._connect() as connection:
share = connection.execute(
"SELECT * FROM dub_shares WHERE share_id = ?", (share_id,)
).fetchone()
if share is None:
return None
rows = connection.execute(
"""
SELECT * FROM dub_share_segments
WHERE share_id = ?
ORDER BY sentence_index
""",
(share_id,),
).fetchall()
result = dict(share)
result["segments"] = [dict(row) for row in rows]
return result
def record_attempt(
self,
*,
attempt_id: str,
video_hash: str,
sentence_index: int,
result: Dict[str, Any],
audio_filename: Optional[str],
user_id: str,
) -> None:
with self._connect() as connection:
connection.execute(
"""
INSERT INTO attempts (
attempt_id, video_hash, sentence_index, result_json,
audio_filename, user_id, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
attempt_id,
normalize_video_hash(video_hash),
sentence_index,
json.dumps(result, ensure_ascii=False),
audio_filename,
user_id,
utc_now(),
),
)
def create_user(self, *, username: str, password: str, nickname: Optional[str]) -> Optional[Dict[str, Any]]:
user_id = uuid.uuid4().hex
now = utc_now()
try:
with self._connect() as connection:
row = connection.execute(
"""
INSERT INTO users (id, username, nickname, password_hash, created_at)
VALUES (?, ?, ?, ?, ?)
""",
(
user_id,
username.strip().lower(),
(nickname or username).strip()[:80],
hash_password(password),
now,
),
).fetchone()
except sqlite3.IntegrityError:
return None
return self.get_user_by_id(user_id)
def get_user_by_username(self, username: str) -> Optional[Dict[str, Any]]:
with self._connect() as connection:
row = connection.execute(
"SELECT * FROM users WHERE username = ? COLLATE NOCASE",
(username.strip(),),
).fetchone()
return dict(row) if row else None
def get_user_by_id(self, user_id: str) -> Optional[Dict[str, Any]]:
with self._connect() as connection:
row = connection.execute(
"SELECT id, username, nickname, created_at FROM users WHERE id = ?",
(user_id,),
).fetchone()
return dict(row) if row else None
def authenticate_user(self, username: str, password: str) -> Optional[Dict[str, Any]]:
with self._connect() as connection:
row = connection.execute(
"SELECT * FROM users WHERE username = ? COLLATE NOCASE",
(username.strip(),),
).fetchone()
if row is None or not verify_password(password, row["password_hash"]):
return None
return self.get_user_by_id(row["id"])
def create_session(self, user_id: str) -> str:
token = new_token()
now = datetime.now(timezone.utc)
with self._connect() as connection:
connection.execute(
"""
INSERT INTO auth_tokens (token_hash, user_id, created_at, expires_at)
VALUES (?, ?, ?, ?)
""",
(
hash_token(token),
user_id,
now.isoformat(),
(now + timedelta(days=30)).isoformat(),
),
)
return token
def resolve_session(self, token: str) -> Optional[Dict[str, Any]]:
with self._connect() as connection:
row = connection.execute(
"""
SELECT u.id, u.username, u.nickname, u.created_at
FROM auth_tokens t JOIN users u ON u.id = t.user_id
WHERE t.token_hash = ? AND t.expires_at > ?
""",
(hash_token(token), datetime.now(timezone.utc).isoformat()),
).fetchone()
return dict(row) if row else None
def delete_session(self, token: str) -> None:
with self._connect() as connection:
connection.execute(
"DELETE FROM auth_tokens WHERE token_hash = ?",
(hash_token(token),),
)
def list_courses(self, user_id: str) -> List[Dict[str, Any]]:
with self._connect() as connection:
rows = connection.execute(
"""
SELECT v.*, COUNT(s.sentence_index) AS sentence_count,
CASE WHEN e.user_id IS NULL THEN 0 ELSE 1 END AS enrolled
FROM videos v
LEFT JOIN sentences s ON s.video_hash = v.video_hash
LEFT JOIN enrollments e ON e.video_hash = v.video_hash AND e.user_id = ?
WHERE v.status = 'ready'
GROUP BY v.video_hash
ORDER BY e.created_at DESC, v.created_at DESC
""",
(user_id,),
).fetchall()
return [dict(row) for row in rows]
def enroll_course(self, user_id: str, video_hash: str) -> bool:
normalized_hash = normalize_video_hash(video_hash)
with self._connect() as connection:
video = connection.execute(
"SELECT 1 FROM videos WHERE video_hash = ? AND status = 'ready'",
(normalized_hash,),
).fetchone()
if video is None:
return False
connection.execute(
"""
INSERT OR IGNORE INTO enrollments (user_id, video_hash, created_at)
VALUES (?, ?, ?)
""",
(user_id, normalized_hash, utc_now()),
)
return True
def unenroll_course(self, user_id: str, video_hash: str) -> bool:
normalized_hash = normalize_video_hash(video_hash)
with self._connect() as connection:
cursor = connection.execute(
"DELETE FROM enrollments WHERE user_id = ? AND video_hash = ?",
(user_id, normalized_hash),
)
return cursor.rowcount > 0
def list_user_results(self, user_id: str) -> List[Dict[str, Any]]:
with self._connect() as connection:
rows = connection.execute(
"""
SELECT a.attempt_id, a.video_hash, a.sentence_index, a.result_json, a.created_at,
v.title AS course_title, s.text AS sentence_text
FROM attempts a
JOIN videos v ON v.video_hash = a.video_hash
LEFT JOIN sentences s
ON s.video_hash = a.video_hash AND s.sentence_index = a.sentence_index
WHERE a.user_id = ?
ORDER BY a.created_at DESC
LIMIT 500
""",
(user_id,),
).fetchall()
results = []
for row in rows:
payload = json.loads(row["result_json"] or "{}")
results.append(
{
"attempt_id": row["attempt_id"],
"video_hash": row["video_hash"],
"course_title": row["course_title"],
"sentence_index": row["sentence_index"],
"sentence_text": row["sentence_text"] or "",
"overall_score": float(payload.get("overall_score", 0)),
"passed": bool(payload.get("passed", False)),
"created_at": row["created_at"],
}
)
return results
def list_user_dub_shares(self, user_id: str) -> List[Dict[str, Any]]:
with self._connect() as connection:
rows = connection.execute(
"""
SELECT d.share_id, d.video_hash, d.title, d.created_at,
v.title AS course_title,
COUNT(g.id) AS segment_count,
COALESCE(AVG(NULLIF(g.overall_score, 0)), 0) AS average_score
FROM dub_shares d
JOIN videos v ON v.video_hash = d.video_hash
LEFT JOIN dub_share_segments g ON g.share_id = d.share_id
WHERE d.user_id = ?
GROUP BY d.share_id
ORDER BY d.created_at DESC
LIMIT 500
""",
(user_id,),
).fetchall()
return [dict(row) for row in rows]
def delete_video(self, video_hash: str) -> Optional[Dict[str, Any]]:
video = self.get_video(video_hash)
if video is None:
return None
with self._connect() as connection:
connection.execute(
"DELETE FROM videos WHERE video_hash = ?",
(normalize_video_hash(video_hash),),
)
return video