|
|
| import os
|
| import sqlite3
|
| from typing import List, Optional
|
| from db import DB_PATH
|
|
|
| def _ensure_video_access_tables() -> None:
|
| os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
|
| with sqlite3.connect(DB_PATH) as conn:
|
| c = conn.cursor()
|
| c.execute(
|
| """
|
| CREATE TABLE IF NOT EXISTS video_access (
|
| lesson_id INTEGER PRIMARY KEY,
|
| visibility TEXT NOT NULL DEFAULT 'public' CHECK (visibility IN ('public','restricted','private')),
|
| updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
| )
|
| """
|
| )
|
| c.execute(
|
| """
|
| CREATE TABLE IF NOT EXISTS video_access_list (
|
| lesson_id INTEGER NOT NULL,
|
| student_email TEXT NOT NULL,
|
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
| UNIQUE(lesson_id, student_email)
|
| )
|
| """
|
| )
|
| conn.commit()
|
|
|
| _ensure_video_access_tables()
|
|
|
| def get_video_visibility(lesson_id: int) -> str:
|
| with sqlite3.connect(DB_PATH) as conn:
|
| c = conn.cursor()
|
| c.execute("SELECT visibility FROM video_access WHERE lesson_id=?", (lesson_id,))
|
| row = c.fetchone()
|
| return row[0] if row and row[0] else "public"
|
|
|
| def set_video_visibility(lesson_id: int, visibility: str) -> None:
|
| if visibility not in ("public", "restricted", "private"):
|
| raise ValueError("visibility deve ser 'public', 'restricted' ou 'private'")
|
| with sqlite3.connect(DB_PATH) as conn:
|
| c = conn.cursor()
|
| c.execute(
|
| """
|
| INSERT INTO video_access (lesson_id, visibility, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)
|
| ON CONFLICT(lesson_id) DO UPDATE SET visibility=excluded.visibility, updated_at=CURRENT_TIMESTAMP
|
| """,
|
| (lesson_id, visibility),
|
| )
|
| conn.commit()
|
|
|
| def get_video_acl_emails(lesson_id: int) -> List[str]:
|
| with sqlite3.connect(DB_PATH) as conn:
|
| c = conn.cursor()
|
| c.execute("SELECT student_email FROM video_access_list WHERE lesson_id=?", (lesson_id,))
|
| rows = c.fetchall()
|
| return [r[0] for r in rows] if rows else []
|
|
|
| def set_video_acl_emails(lesson_id: int, emails: List[str]) -> None:
|
| emails_clean = sorted({(e or "").strip().lower() for e in emails if (e or "").strip()})
|
| with sqlite3.connect(DB_PATH) as conn:
|
| c = conn.cursor()
|
| c.execute("DELETE FROM video_access_list WHERE lesson_id=?", (lesson_id,))
|
| if emails_clean:
|
| c.executemany(
|
| "INSERT INTO video_access_list (lesson_id, student_email) VALUES (?, ?)",
|
| [(lesson_id, e) for e in emails_clean]
|
| )
|
| conn.commit()
|
|
|
| def is_video_allowed_for_student(lesson_id: int, student_email: Optional[str], is_admin: bool) -> bool:
|
| if is_admin:
|
| return True
|
| vis = get_video_visibility(lesson_id)
|
| if vis == "public":
|
| return bool(student_email)
|
| if vis == "private":
|
| return False
|
| if vis == "restricted":
|
| if not student_email:
|
| return False
|
| allow = get_video_acl_emails(lesson_id)
|
| return student_email.strip().lower() in {e.strip().lower() for e in allow}
|
| return False |
