app.py

import os
import re
import secrets
import sqlite3
import tempfile
from datetime import datetime
from pathlib import Path

import gradio as gr
import pandas as pd


DEFAULT_DB_PATH = "/data/survey_comments.db" if Path("/data").exists() else "survey_comments.db"
DB_PATH = Path(os.getenv("DB_PATH", DEFAULT_DB_PATH))
EXPORT_DIR = Path(os.getenv("EXPORT_DIR", tempfile.gettempdir()))
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "OAYS2026")

CATEGORIES = [
    "General",
    "Program Feedback",
    "Technical Issue",
    "Suggestion",
    "Complaint",
    "Other",
]

EMAIL_WITH_AT_PATTERN = re.compile(r"^[^@\s]+@[^@\s]+$")


def get_connection():
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(DB_PATH, timeout=30)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=30000")
    return conn


def init_db():
    with get_connection() as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS survey_comments (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT,
                email TEXT,
                category TEXT NOT NULL,
                subject TEXT NOT NULL DEFAULT '',
                comments TEXT NOT NULL,
                mailing_list_opt_in INTEGER NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL
            )
            """
        )

        existing_columns = pd.read_sql_query(
            "PRAGMA table_info(survey_comments)", conn
        )["name"].tolist()

        if "subject" not in existing_columns:
            conn.execute(
                """
                ALTER TABLE survey_comments
                ADD COLUMN subject TEXT NOT NULL DEFAULT ''
                """
            )

        if "mailing_list_opt_in" not in existing_columns:
            conn.execute(
                """
                ALTER TABLE survey_comments
                ADD COLUMN mailing_list_opt_in INTEGER NOT NULL DEFAULT 0
                """
            )

        conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_survey_category
            ON survey_comments(category)
            """
        )

        conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_survey_created_at
            ON survey_comments(created_at)
            """
        )

        conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_survey_email
            ON survey_comments(email)
            """
        )

        conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_survey_subject
            ON survey_comments(subject)
            """
        )

        conn.execute(
            """
            CREATE INDEX IF NOT EXISTS idx_survey_mailing_opt_in
            ON survey_comments(mailing_list_opt_in)
            """
        )

        conn.commit()


def clean_text(value):
    return (value or "").strip()


def is_valid_optional_email(email):
    if not email:
        return True
    return bool(EMAIL_WITH_AT_PATTERN.fullmatch(email))


def is_admin_password(password):
    password = password or ""
    return secrets.compare_digest(password, ADMIN_PASSWORD)


def save_comment(name, email, category, subject, comments, mailing_list_opt_in):
    name = clean_text(name)
    email = clean_text(email).lower()
    category = clean_text(category)
    subject = clean_text(subject)
    comments = clean_text(comments)
    mailing_list_opt_in = 1 if mailing_list_opt_in else 0

    if not category:
        return (
            "Category is required.",
            name,
            email,
            category,
            subject,
            comments,
            bool(mailing_list_opt_in),
        )

    if not subject:
        return (
            "Subject is required.",
            name,
            email,
            category,
            subject,
            comments,
            bool(mailing_list_opt_in),
        )

    if not comments:
        return (
            "Comments are required.",
            name,
            email,
            category,
            subject,
            comments,
            bool(mailing_list_opt_in),
        )

    if email and not is_valid_optional_email(email):
        return (
            "Email must include a valid @ address.",
            name,
            email,
            category,
            subject,
            comments,
            bool(mailing_list_opt_in),
        )

    if mailing_list_opt_in and not email:
        return (
            "Email is required if the person opts into the mailing list.",
            name,
            email,
            category,
            subject,
            comments,
            True,
        )

    created_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    with get_connection() as conn:
        conn.execute(
            """
            INSERT INTO survey_comments
                (
                    name,
                    email,
                    category,
                    subject,
                    comments,
                    mailing_list_opt_in,
                    created_at
                )
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                name,
                email,
                category,
                subject,
                comments,
                mailing_list_opt_in,
                created_at,
            ),
        )
        conn.commit()

    return "Comment saved successfully.", "", "", "General", "", "", False


def retrieve_public_comments_table():
    with get_connection() as conn:
        df = pd.read_sql_query(
            """
            SELECT
                category,
                subject,
                comments
            FROM survey_comments
            ORDER BY
                category ASC,
                created_at DESC
            """,
            conn,
        )

    if df.empty:
        return pd.DataFrame(columns=["category", "subject", "comments"])

    return df


def retrieve_public_comments_text():
    df = retrieve_public_comments_table()

    if df.empty:
        return "No comments found."

    output = []

    for category in sorted(df["category"].dropna().unique()):
        output.append("=" * 70)
        output.append(f"CATEGORY: {category}")
        output.append("=" * 70)

        subset = df[df["category"] == category]

        for _, row in subset.iterrows():
            output.append(
                f"""
Subject: {row["subject"]}

Comment:
{row["comments"]}

{"-" * 70}
"""
            )

    return "\n".join(output)


def get_full_comments_dataframe():
    with get_connection() as conn:
        df = pd.read_sql_query(
            """
            SELECT
                id,
                category,
                created_at,
                COALESCE(NULLIF(name, ''), 'Anonymous') AS name,
                COALESCE(NULLIF(email, ''), '') AS email,
                subject,
                CASE
                    WHEN mailing_list_opt_in = 1 THEN 'Yes'
                    ELSE 'No'
                END AS mailing_list_opt_in,
                comments
            FROM survey_comments
            ORDER BY
                category ASC,
                created_at DESC
            """,
            conn,
        )

    return df


def export_comments_csv(admin_password):
    if not is_admin_password(admin_password):
        return "Incorrect password. CSV export was not generated.", None

    df = get_full_comments_dataframe()
    EXPORT_DIR.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    export_path = EXPORT_DIR / f"survey_export_{timestamp}.csv"
    df.to_csv(export_path, index=False)

    return "Full comments CSV generated.", str(export_path)


def get_mailing_list(format_choice, admin_password):
    if not is_admin_password(admin_password):
        return "Incorrect password. Mailing list export was not generated.", None

    with get_connection() as conn:
        df = pd.read_sql_query(
            """
            SELECT DISTINCT
                COALESCE(NULLIF(TRIM(name), ''), '') AS name,
                TRIM(email) AS email
            FROM survey_comments
            WHERE
                mailing_list_opt_in = 1
                AND email IS NOT NULL
                AND TRIM(email) <> ''
            ORDER BY
                email ASC
            """,
            conn,
        )

    if df.empty:
        return "No opted-in mailing-list contacts found.", None

    df["name"] = df["name"].fillna("").astype(str).str.strip()
    df["email"] = df["email"].fillna("").astype(str).str.strip().str.lower()
    df = df[df["email"] != ""]
    df = df.drop_duplicates(subset=["email"]).sort_values("email")

    if format_choice == "Name and email":
        output_text = "\n".join(
            [
                f"{row['name']} <{row['email']}>" if row["name"] else row["email"]
                for _, row in df.iterrows()
            ]
        )
    elif format_choice == "Line-separated emails only":
        output_text = "\n".join(df["email"])
    elif format_choice == "Comma-separated emails only":
        output_text = ", ".join(df["email"])
    elif format_choice == "Semicolon-separated emails only":
        output_text = "; ".join(df["email"])
    else:
        output_text = "\n".join(df["email"])

    EXPORT_DIR.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    export_path = EXPORT_DIR / f"mailing_list_opt_ins_{timestamp}.csv"
    df.to_csv(export_path, index=False)

    return output_text, str(export_path)


init_db()

with gr.Blocks(title="Basic Survey Comment Collector") as demo:
    gr.Markdown("# Basic Survey Comment Collector")

    with gr.Tab("Submit Comment"):
        gr.Markdown(
            "Name and email are optional. Category, subject, and comments are required. "
            "Email is required only if the person opts into the mailing list."
        )

        name = gr.Textbox(label="Name Optional", placeholder="Optional")
        email = gr.Textbox(label="Email Optional", placeholder="Optional")

        mailing_list_opt_in = gr.Checkbox(
            label="Include my name/email on the mailing list",
            value=False,
        )

        category = gr.Dropdown(
            choices=CATEGORIES,
            value="General",
            label="Category Required",
        )

        subject = gr.Textbox(
            label="Subject Required",
            placeholder="Brief subject line",
        )

        comments = gr.Textbox(
            label="Comments Required",
            lines=5,
            placeholder="Enter comments here...",
        )

        submit_btn = gr.Button("Save Comment")
        status = gr.Textbox(label="Status", interactive=False)

        submit_btn.click(
            fn=save_comment,
            inputs=[
                name,
                email,
                category,
                subject,
                comments,
                mailing_list_opt_in,
            ],
            outputs=[
                status,
                name,
                email,
                category,
                subject,
                comments,
                mailing_list_opt_in,
            ],
        )

    with gr.Tab("View Comments"):
        gr.Markdown(
            "This public view displays only category, subject, and comment. "
            "Names, emails, dates, and mailing-list status are hidden."
        )

        retrieve_btn = gr.Button("Retrieve All Comments")

        public_comments_table = gr.Dataframe(
            label="Comments Table",
            interactive=False,
            wrap=True,
        )

        public_comments_text = gr.Textbox(
            label="Comments Grouped by Category",
            lines=20,
            interactive=False,
        )

        retrieve_btn.click(
            fn=retrieve_public_comments_table,
            inputs=[],
            outputs=public_comments_table,
        )

        retrieve_btn.click(
            fn=retrieve_public_comments_text,
            inputs=[],
            outputs=public_comments_text,
        )

    with gr.Tab("Admin Export Full CSV"):
        gr.Markdown(
            "Password-protected export. The CSV includes all stored information, "
            "including name, email, date, subject, category, mailing-list opt-in, and comments."
        )

        export_password = gr.Textbox(
            label="Export Password",
            type="password",
            placeholder="Enter password",
        )

        export_btn = gr.Button("Export Full Comments CSV")
        export_status = gr.Textbox(label="Export Status", interactive=False)
        csv_file = gr.File(label="Download Full Comments CSV")

        export_btn.click(
            fn=export_comments_csv,
            inputs=export_password,
            outputs=[export_status, csv_file],
        )

    with gr.Tab("Admin Export Mailing List"):
        gr.Markdown(
            "Password-protected export. Only people who checked the mailing-list box will appear here."
        )

        mailing_list_password = gr.Textbox(
            label="Mailing List Export Password",
            type="password",
            placeholder="Enter password",
        )

        mailing_list_format = gr.Radio(
            choices=[
                "Name and email",
                "Line-separated emails only",
                "Comma-separated emails only",
                "Semicolon-separated emails only",
            ],
            value="Name and email",
            label="Mailing List Format",
        )

        email_export_btn = gr.Button("Generate Mailing List")

        email_output = gr.Textbox(
            label="Copy/Paste Mailing List",
            lines=15,
            interactive=False,
        )

        email_file = gr.File(label="Download Mailing List CSV")

        email_export_btn.click(
            fn=get_mailing_list,
            inputs=[mailing_list_format, mailing_list_password],
            outputs=[email_output, email_file],
        )


if __name__ == "__main__":
    demo.launch()