Spaces:
Running
Running
File size: 4,967 Bytes
954286a b1c84b5 954286a b1c84b5 954286a b1c84b5 954286a b1c84b5 954286a b1c84b5 954286a b1c84b5 954286a | 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 | """
PhilVerify — Firebase / Firestore Client
Initializes firebase-admin SDK and provides typed helpers for persistence.
Setup:
1. Go to Firebase Console → Project Settings → Service Accounts
2. Click "Generate new private key" → save as `serviceAccountKey.json`
in the PhilVerify project root (already in .gitignore)
3. Set FIREBASE_PROJECT_ID in .env
Collections:
verifications/ — one doc per verification run
trends/summary — aggregated entity/topic counters
"""
import logging
import os
from functools import lru_cache
from pathlib import Path
logger = logging.getLogger(__name__)
_SERVICEACCOUNT_PATH = Path(__file__).parent / "serviceAccountKey.json"
_db = None # Firestore client singleton
def get_firestore():
"""Return the Firestore client, or None if Firebase is not configured."""
global _db
if _db is not None:
return _db
try:
import firebase_admin
from firebase_admin import credentials, firestore
if firebase_admin._DEFAULT_APP_NAME in firebase_admin._apps:
_db = firestore.client()
return _db
if _SERVICEACCOUNT_PATH.exists():
# Service account key file available (local dev + CI)
cred = credentials.Certificate(str(_SERVICEACCOUNT_PATH))
firebase_admin.initialize_app(cred)
logger.info("Firebase initialized via service account key")
elif os.getenv("GOOGLE_APPLICATION_CREDENTIALS") or os.getenv("K_SERVICE"):
# Cloud Run (K_SERVICE is always set) or explicit ADC path
cred = credentials.ApplicationDefault()
firebase_admin.initialize_app(cred)
logger.info("Firebase initialized via Application Default Credentials")
else:
logger.warning(
"Firebase not configured — no serviceAccountKey.json and no "
"GOOGLE_APPLICATION_CREDENTIALS env var. History will use in-memory store."
)
return None
_db = firestore.client()
return _db
except ImportError:
logger.warning("firebase-admin not installed — Firestore disabled")
return None
except Exception as e:
logger.error("Firebase init error: %s — falling back to in-memory store", e)
return None
async def save_verification(data: dict) -> bool:
"""
Persist a verification result to Firestore.
Returns True on success, False if Firebase is unavailable.
"""
db = get_firestore()
if db is None:
return False
try:
db.collection("verifications").document(data["id"]).set(data)
logger.debug("Verification %s saved to Firestore", data["id"])
return True
except Exception as e:
logger.error("Firestore write error: %s", e)
return False
async def get_verifications(
limit: int = 20,
offset: int = 0,
verdict_filter: str | None = None,
) -> list[dict]:
"""Fetch verification history from Firestore ordered by timestamp desc."""
db = get_firestore()
if db is None:
return []
try:
from google.cloud.firestore_v1.base_query import FieldFilter
query = (
db.collection("verifications")
.order_by("timestamp", direction="DESCENDING")
)
if verdict_filter:
query = query.where(filter=FieldFilter("verdict", "==", verdict_filter))
docs = query.limit(limit + offset).stream()
results = [doc.to_dict() for doc in docs]
return results[offset : offset + limit]
except Exception as e:
logger.error("Firestore read error: %s", e)
return []
def get_all_verifications_sync() -> list[dict]:
"""Synchronously fetch ALL verification records from Firestore (used by trends aggregation)."""
db = get_firestore()
if db is None:
return []
try:
docs = (
db.collection("verifications")
.order_by("timestamp", direction="DESCENDING")
.limit(10_000) # hard cap — more than enough for trends analysis
.stream()
)
return [doc.to_dict() for doc in docs]
except Exception as e:
logger.error("Firestore get_all_verifications_sync error: %s", e)
return []
async def get_verification_count(verdict_filter: str | None = None) -> int:
"""Return total count of verifications (with optional verdict filter)."""
db = get_firestore()
if db is None:
return 0
try:
from google.cloud.firestore_v1.base_query import FieldFilter
query = db.collection("verifications")
if verdict_filter:
query = query.where(filter=FieldFilter("verdict", "==", verdict_filter))
# Use aggregation query (Firestore native count)
result = query.count().get()
return result[0][0].value
except Exception as e:
logger.error("Firestore count error: %s", e)
return 0
|