| import json |
| import os |
| import sqlite3 |
| import secrets |
| from contextlib import contextmanager |
| from datetime import datetime, timezone |
| from typing import Any, Dict, List, Optional |
|
|
| from werkzeug.security import check_password_hash, generate_password_hash |
|
|
| BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)) |
| INSTANCE_DIR = os.path.join(BASE_DIR, "instance") |
| os.makedirs(INSTANCE_DIR, exist_ok=True) |
| DB_PATH = os.getenv("AISPY_DB_PATH", os.path.join(INSTANCE_DIR, "aispy.db")) |
|
|
|
|
| @contextmanager |
| def get_connection(): |
| conn = sqlite3.connect(DB_PATH) |
| conn.row_factory = sqlite3.Row |
| try: |
| yield conn |
| conn.commit() |
| finally: |
| conn.close() |
|
|
|
|
| def _utc_now() -> str: |
| return datetime.now(timezone.utc).isoformat() |
|
|
|
|
| def init_db() -> None: |
| with get_connection() as conn: |
| conn.execute( |
| """ |
| CREATE TABLE IF NOT EXISTS users ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| name TEXT NOT NULL, |
| email TEXT NOT NULL UNIQUE, |
| password_hash TEXT NOT NULL, |
| is_admin INTEGER NOT NULL DEFAULT 0, |
| created_at TEXT NOT NULL, |
| last_login_at TEXT |
| ) |
| """ |
| ) |
| conn.execute( |
| """ |
| CREATE TABLE IF NOT EXISTS analyses ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| user_id INTEGER NOT NULL, |
| request_id TEXT NOT NULL UNIQUE, |
| input_type TEXT NOT NULL, |
| file_name TEXT, |
| source_url TEXT, |
| text_claim TEXT, |
| verdict TEXT, |
| confidence REAL, |
| reasoning TEXT, |
| final_report TEXT, |
| payload_json TEXT, |
| status TEXT NOT NULL DEFAULT 'completed', |
| created_at TEXT NOT NULL, |
| FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE |
| ) |
| """ |
| ) |
| conn.execute( |
| """ |
| CREATE TABLE IF NOT EXISTS password_resets ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| user_id INTEGER NOT NULL, |
| token TEXT NOT NULL UNIQUE, |
| expires_at TEXT NOT NULL, |
| used_at TEXT, |
| created_at TEXT NOT NULL, |
| FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE |
| ) |
| """ |
| ) |
|
|
| user_columns = {row[1] for row in conn.execute("PRAGMA table_info(users)").fetchall()} |
| if "is_admin" not in user_columns: |
| conn.execute("ALTER TABLE users ADD COLUMN is_admin INTEGER NOT NULL DEFAULT 0") |
|
|
|
|
| def seed_first_admin() -> None: |
| with get_connection() as conn: |
| total_users = conn.execute("SELECT COUNT(*) AS c FROM users").fetchone()["c"] |
| total_admins = conn.execute("SELECT COUNT(*) AS c FROM users WHERE is_admin = 1").fetchone()["c"] |
| if int(total_users or 0) > 0 and int(total_admins or 0) == 0: |
| conn.execute("UPDATE users SET is_admin = 1 WHERE id = (SELECT id FROM users ORDER BY id LIMIT 1)") |
|
|
|
|
| def row_to_dict(row: Optional[sqlite3.Row]) -> Optional[Dict[str, Any]]: |
| return dict(row) if row is not None else None |
|
|
|
|
| def create_user(name: str, email: str, password: str) -> Dict[str, Any]: |
| password_hash = generate_password_hash(password) |
| admin_emails = {e.strip().lower() for e in os.getenv("AISPY_ADMIN_EMAILS", "").split(",") if e.strip()} |
| with get_connection() as conn: |
| total_users = conn.execute("SELECT COUNT(*) AS c FROM users").fetchone()["c"] |
| has_admin = conn.execute("SELECT COUNT(*) AS c FROM users WHERE is_admin = 1").fetchone()["c"] |
| is_admin = 1 if (email.strip().lower() in admin_emails or int(total_users or 0) == 0 or int(has_admin or 0) == 0) else 0 |
| cur = conn.execute( |
| """ |
| INSERT INTO users (name, email, password_hash, is_admin, created_at) |
| VALUES (?, ?, ?, ?, ?) |
| """, |
| (name.strip(), email.strip().lower(), password_hash, is_admin, _utc_now()), |
| ) |
| row = conn.execute("SELECT * FROM users WHERE id = ?", (cur.lastrowid,)).fetchone() |
| return row_to_dict(row) |
|
|
|
|
| def get_user_by_id(user_id: int) -> Optional[Dict[str, Any]]: |
| with get_connection() as conn: |
| row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone() |
| return row_to_dict(row) |
|
|
|
|
| def get_user_by_email(email: str) -> Optional[Dict[str, Any]]: |
| with get_connection() as conn: |
| row = conn.execute("SELECT * FROM users WHERE email = ?", (email.strip().lower(),)).fetchone() |
| return row_to_dict(row) |
|
|
|
|
| def verify_user(email: str, password: str) -> Optional[Dict[str, Any]]: |
| user = get_user_by_email(email) |
| if not user: |
| return None |
| if not check_password_hash(user["password_hash"], password): |
| return None |
| with get_connection() as conn: |
| conn.execute("UPDATE users SET last_login_at = ? WHERE id = ?", (_utc_now(), user["id"])) |
| return get_user_by_id(user["id"]) |
|
|
|
|
| def update_user_profile(user_id: int, *, name: Optional[str] = None, email: Optional[str] = None) -> Optional[Dict[str, Any]]: |
| fields = [] |
| values: List[Any] = [] |
| if name is not None: |
| fields.append("name = ?") |
| values.append(name.strip()) |
| if email is not None: |
| fields.append("email = ?") |
| values.append(email.strip().lower()) |
| if not fields: |
| return get_user_by_id(user_id) |
| values.append(user_id) |
| with get_connection() as conn: |
| conn.execute(f"UPDATE users SET {', '.join(fields)} WHERE id = ?", values) |
| return get_user_by_id(user_id) |
|
|
|
|
| def change_password(user_id: int, new_password: str) -> None: |
| password_hash = generate_password_hash(new_password) |
| with get_connection() as conn: |
| conn.execute("UPDATE users SET password_hash = ? WHERE id = ?", (password_hash, user_id)) |
|
|
|
|
| def create_password_reset_token(email: str, ttl_minutes: int = 30) -> Optional[Dict[str, Any]]: |
| user = get_user_by_email(email) |
| if not user: |
| return None |
| token = secrets.token_urlsafe(32) |
| expires_at = datetime.now(timezone.utc).timestamp() + ttl_minutes * 60 |
| expires_iso = datetime.fromtimestamp(expires_at, tz=timezone.utc).isoformat() |
| with get_connection() as conn: |
| conn.execute( |
| """ |
| INSERT INTO password_resets (user_id, token, expires_at, created_at) |
| VALUES (?, ?, ?, ?) |
| """, |
| (user["id"], token, expires_iso, _utc_now()), |
| ) |
| return {"token": token, "user_id": user["id"], "email": user["email"], "expires_at": expires_iso} |
|
|
|
|
| def get_password_reset_token(token: str) -> Optional[Dict[str, Any]]: |
| with get_connection() as conn: |
| row = conn.execute("SELECT * FROM password_resets WHERE token = ?", (token,)).fetchone() |
| return row_to_dict(row) |
|
|
|
|
| def use_password_reset_token(token: str) -> Optional[Dict[str, Any]]: |
| record = get_password_reset_token(token) |
| if not record or record.get("used_at"): |
| return None |
| if record.get("expires_at") and datetime.fromisoformat(record["expires_at"]) < datetime.now(timezone.utc): |
| return None |
| with get_connection() as conn: |
| conn.execute("UPDATE password_resets SET used_at = ? WHERE token = ?", (_utc_now(), token)) |
| return record |
|
|
|
|
| def set_user_admin(user_id: int, is_admin: bool) -> Optional[Dict[str, Any]]: |
| with get_connection() as conn: |
| conn.execute("UPDATE users SET is_admin = ? WHERE id = ?", (1 if is_admin else 0, user_id)) |
| return get_user_by_id(user_id) |
|
|
|
|
| def create_analysis(payload: Dict[str, Any]) -> Dict[str, Any]: |
| with get_connection() as conn: |
| cur = conn.execute( |
| """ |
| INSERT INTO analyses ( |
| user_id, request_id, input_type, file_name, source_url, text_claim, |
| verdict, confidence, reasoning, final_report, payload_json, status, created_at |
| ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| """, |
| ( |
| payload["user_id"], |
| payload["request_id"], |
| payload.get("input_type", "text"), |
| payload.get("file_name"), |
| payload.get("source_url"), |
| payload.get("text_claim"), |
| payload.get("verdict"), |
| payload.get("confidence"), |
| payload.get("reasoning"), |
| payload.get("final_report"), |
| json.dumps(payload.get("payload_json", {}), ensure_ascii=False), |
| payload.get("status", "completed"), |
| _utc_now(), |
| ), |
| ) |
| return get_analysis_by_id(cur.lastrowid) |
|
|
|
|
| def list_user_analyses(user_id: int, limit: int = 20) -> List[Dict[str, Any]]: |
| with get_connection() as conn: |
| rows = conn.execute( |
| """ |
| SELECT * FROM analyses |
| WHERE user_id = ? |
| ORDER BY id DESC |
| LIMIT ? |
| """, |
| (user_id, limit), |
| ).fetchall() |
| return [row_to_dict(row) for row in rows] |
|
|
|
|
| def get_analysis_by_id(analysis_id: int, user_id: Optional[int] = None) -> Optional[Dict[str, Any]]: |
| with get_connection() as conn: |
| if user_id is None: |
| row = conn.execute("SELECT * FROM analyses WHERE id = ?", (analysis_id,)).fetchone() |
| else: |
| row = conn.execute( |
| "SELECT * FROM analyses WHERE id = ? AND user_id = ?", |
| (analysis_id, user_id), |
| ).fetchone() |
| if not row: |
| return None |
| data = row_to_dict(row) |
| if data and data.get("payload_json"): |
| try: |
| data["payload_json"] = json.loads(data["payload_json"]) |
| except Exception: |
| pass |
| return data |
|
|
|
|
| def latest_analysis_for_user(user_id: int) -> Optional[Dict[str, Any]]: |
| items = list_user_analyses(user_id, limit=1) |
| return items[0] if items else None |
|
|
|
|
| def get_dashboard_stats(user_id: int) -> Dict[str, Any]: |
| with get_connection() as conn: |
| total = conn.execute("SELECT COUNT(*) AS c FROM analyses WHERE user_id = ?", (user_id,)).fetchone()["c"] |
| fake = conn.execute( |
| """ |
| SELECT COUNT(*) AS c |
| FROM analyses |
| WHERE user_id = ? AND LOWER(COALESCE(verdict, '')) LIKE '%fake%' |
| """, |
| (user_id,), |
| ).fetchone()["c"] |
| recent = conn.execute( |
| "SELECT created_at FROM analyses WHERE user_id = ? ORDER BY id DESC LIMIT 1", |
| (user_id,), |
| ).fetchone() |
| return { |
| "total_analyses": int(total or 0), |
| "fake_flags": int(fake or 0), |
| "last_scan_at": recent["created_at"] if recent else None, |
| } |
|
|
|
|
| def list_users(limit: int = 200) -> List[Dict[str, Any]]: |
| with get_connection() as conn: |
| rows = conn.execute( |
| "SELECT * FROM users ORDER BY id DESC LIMIT ?", |
| (limit,), |
| ).fetchall() |
| return [row_to_dict(row) for row in rows] |
|
|
|
|
| def list_all_analyses(limit: int = 200) -> List[Dict[str, Any]]: |
| with get_connection() as conn: |
| rows = conn.execute( |
| "SELECT * FROM analyses ORDER BY id DESC LIMIT ?", |
| (limit,), |
| ).fetchall() |
| return [row_to_dict(row) for row in rows] |
|
|
|
|
| def get_admin_stats() -> Dict[str, Any]: |
| with get_connection() as conn: |
| users = conn.execute("SELECT COUNT(*) AS c FROM users").fetchone()["c"] |
| analyses = conn.execute("SELECT COUNT(*) AS c FROM analyses").fetchone()["c"] |
| admins = conn.execute("SELECT COUNT(*) AS c FROM users WHERE is_admin = 1").fetchone()["c"] |
| return {"users": int(users or 0), "analyses": int(analyses or 0), "admins": int(admins or 0)} |
|
|
