Spaces:
Sleeping
Sleeping
| """ | |
| web_app/database.py — Database audit logging system. | |
| Supports dual-driver data persistence: | |
| - Direct Cloud Supabase PostgreSQL storage (if environment variables exist). | |
| - Automatic Local SQLite database fallback. | |
| """ | |
| import os | |
| import sqlite3 | |
| import datetime | |
| from pathlib import Path | |
| from typing import List, Dict | |
| DB_PATH = Path(__file__).parent / "financial_audit.db" | |
| def init_db(): | |
| """Initializes the local SQLite database if not present.""" | |
| conn = sqlite3.connect(str(DB_PATH)) | |
| cursor = conn.cursor() | |
| cursor.execute(""" | |
| CREATE TABLE IF NOT EXISTS financial_analysis_audit ( | |
| id INTEGER PRIMARY KEY AUTOINCREMENT, | |
| timestamp TEXT NOT NULL, | |
| user_email TEXT NOT NULL, | |
| company_ticker TEXT NOT NULL, | |
| company_name TEXT NOT NULL, | |
| selected_formats TEXT NOT NULL, | |
| ai_model_used TEXT NOT NULL, | |
| status TEXT NOT NULL | |
| ) | |
| """) | |
| conn.commit() | |
| conn.close() | |
| def save_audit_log(user_email: str, company_ticker: str, company_name: str, selected_formats: List[str], ai_models: List[str], status: str): | |
| """ | |
| Saves an audit log entry. | |
| Tries Supabase first, falls back to local SQLite on failure or if not configured. | |
| """ | |
| timestamp = datetime.datetime.utcnow().isoformat() | |
| selected_formats_str = ", ".join(selected_formats) | |
| ai_models_str = ", ".join(ai_models) if ai_models else "none" | |
| supabase_url = os.environ.get("SUPABASE_URL") | |
| supabase_key = os.environ.get("SUPABASE_KEY") | |
| if supabase_url and supabase_key: | |
| try: | |
| from supabase import create_client, Client | |
| supabase: Client = create_client(supabase_url, supabase_key) | |
| data = { | |
| "user_email": user_email, | |
| "company_ticker": company_ticker, | |
| "company_name": company_name, | |
| "selected_formats": selected_formats, # stored as text array in Postgres | |
| "ai_model_used": ai_models_str, | |
| "status": status | |
| } | |
| supabase.table("financial_analysis_audit").insert(data).execute() | |
| print("Audit log successfully written to Cloud Supabase DB.") | |
| return | |
| except Exception as e: | |
| print(f"Supabase logging failed: {e}. Falling back to SQLite.") | |
| # Local SQLite Fallback | |
| try: | |
| init_db() | |
| conn = sqlite3.connect(str(DB_PATH)) | |
| cursor = conn.cursor() | |
| cursor.execute(""" | |
| INSERT INTO financial_analysis_audit (timestamp, user_email, company_ticker, company_name, selected_formats, ai_model_used, status) | |
| VALUES (?, ?, ?, ?, ?, ?, ?) | |
| """, (timestamp, user_email, company_ticker, company_name, selected_formats_str, ai_models_str, status)) | |
| conn.commit() | |
| conn.close() | |
| print("Audit log written to local SQLite database.") | |
| except Exception as sqle: | |
| print(f"Local SQLite database audit write error: {sqle}") | |
| def get_audit_logs() -> List[Dict]: | |
| """ | |
| Fetches the history of audit logs. | |
| Tries Supabase first, falls back to local SQLite. | |
| """ | |
| supabase_url = os.environ.get("SUPABASE_URL") | |
| supabase_key = os.environ.get("SUPABASE_KEY") | |
| if supabase_url and supabase_key: | |
| try: | |
| from supabase import create_client, Client | |
| supabase: Client = create_client(supabase_url, supabase_key) | |
| response = supabase.table("financial_analysis_audit").select("*").order("timestamp", desc=True).limit(50).execute() | |
| logs = [] | |
| for row in response.data: | |
| # Convert list array from supabase to comma-separated string for display alignment | |
| formats = row.get("selected_formats", []) | |
| formats_str = ", ".join(formats) if isinstance(formats, list) else str(formats) | |
| # Format timestamp | |
| ts = row.get("timestamp", "") | |
| if ts: | |
| try: | |
| ts = datetime.datetime.fromisoformat(ts.replace("Z", "+00:00")).strftime("%Y-%m-%d %H:%M:%S") | |
| except: | |
| pass | |
| logs.append({ | |
| "timestamp": ts, | |
| "user_email": row.get("user_email", "N/A"), | |
| "company_ticker": row.get("company_ticker", "N/A"), | |
| "company_name": row.get("company_name", "N/A"), | |
| "selected_formats": formats_str, | |
| "ai_model_used": row.get("ai_model_used", "N/A"), | |
| "status": row.get("status", "N/A") | |
| }) | |
| return logs | |
| except Exception as e: | |
| print(f"Supabase fetch logs failed: {e}. Falling back to SQLite.") | |
| # SQLite Fallback | |
| try: | |
| init_db() | |
| conn = sqlite3.connect(str(DB_PATH)) | |
| conn.row_factory = sqlite3.Row | |
| cursor = conn.cursor() | |
| cursor.execute("SELECT * FROM financial_analysis_audit ORDER BY timestamp DESC LIMIT 50") | |
| rows = cursor.fetchall() | |
| logs = [] | |
| for r in rows: | |
| ts = r["timestamp"] | |
| if ts and "T" in ts: | |
| try: | |
| ts = datetime.datetime.fromisoformat(ts).strftime("%Y-%m-%d %H:%M:%S") | |
| except: | |
| pass | |
| logs.append({ | |
| "timestamp": ts, | |
| "user_email": r["user_email"], | |
| "company_ticker": r["company_ticker"], | |
| "company_name": r["company_name"], | |
| "selected_formats": r["selected_formats"], | |
| "ai_model_used": r["ai_model_used"], | |
| "status": r["status"] | |
| }) | |
| conn.close() | |
| return logs | |
| except Exception as sqle: | |
| print(f"Local SQLite logs retrieve error: {sqle}") | |
| return [] | |