|
|
| import os
|
| import sqlite3
|
| import pandas as pd
|
| from typing import Optional
|
| from db import DB_PATH
|
|
|
| def _ensure_comments_table() -> 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 comments (
|
| id INTEGER PRIMARY KEY AUTOINCREMENT,
|
| course_id INTEGER NOT NULL,
|
| student_email TEXT NOT NULL,
|
| student_name TEXT,
|
| content TEXT NOT NULL,
|
| parent_id INTEGER,
|
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
| )
|
| """
|
| )
|
| conn.commit()
|
|
|
| def add_comment(course_id: int, student_email: str, student_name: str, content: str, parent_id: Optional[int] = None) -> None:
|
| _ensure_comments_table()
|
| content = (content or "").strip()
|
| if not content:
|
| raise ValueError("Comentário vazio.")
|
| with sqlite3.connect(DB_PATH) as conn:
|
| c = conn.cursor()
|
| c.execute(
|
| "INSERT INTO comments (course_id, student_email, student_name, content, parent_id) VALUES (?, ?, ?, ?, ?)",
|
| (course_id, student_email.strip().lower(), (student_name or "").strip(), content, parent_id)
|
| )
|
| conn.commit()
|
|
|
| def delete_comment(comment_id: int) -> None:
|
| _ensure_comments_table()
|
| with sqlite3.connect(DB_PATH) as conn:
|
| c = conn.cursor()
|
| c.execute("DELETE FROM comments WHERE parent_id=?", (comment_id,))
|
| c.execute("DELETE FROM comments WHERE id=?", (comment_id,))
|
| conn.commit()
|
|
|
| def get_comments(course_id: int) -> pd.DataFrame:
|
| _ensure_comments_table()
|
| with sqlite3.connect(DB_PATH) as conn:
|
| df = pd.read_sql_query(
|
| "SELECT id, course_id, student_email, student_name, content, parent_id, created_at FROM comments WHERE course_id=? ORDER BY created_at ASC",
|
| conn,
|
| params=(course_id,)
|
| )
|
| return df |
