Spaces:
Sleeping
Sleeping
File size: 5,893 Bytes
558d330 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | """
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 []
|