""" CropAI Database & Cloud Persistence Module Supports Google Cloud Firestore for scalable Cloud Run deployment with local thread-safe persistence fallback. """ import json import sqlite3 import logging import threading import uuid from typing import Dict, Any, List, Optional from datetime import datetime, timezone from pathlib import Path from .config import settings logger = logging.getLogger("cropai.database") DB_FILE = Path(__file__).parent / "cropai_production.db" class StorageManager: """Manages persistence for diagnoses and expert consultation tickets.""" def __init__(self): self._lock = threading.Lock() self._firestore_client = None self._init_firestore() self._init_local_db() def _init_firestore(self): """Initializes Google Cloud Firestore if GCP environment is configured.""" if settings.gcp_project_id: try: from google.cloud import firestore self._firestore_client = firestore.Client(project=settings.gcp_project_id) logger.info(f"Initialized Google Cloud Firestore for project: {settings.gcp_project_id}") except Exception as e: logger.warning(f"Google Cloud Firestore not available: {e}. Using local storage.") self._firestore_client = None def _init_local_db(self): """Initializes local SQLite schema for resilient offline/container persistence.""" with self._lock: conn = sqlite3.connect(DB_FILE) try: cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS expert_tickets ( ticket_id TEXT PRIMARY KEY, farmer_name TEXT NOT NULL, phone_number TEXT NOT NULL, location TEXT, crop TEXT NOT NULL, disease_suspected TEXT, confidence_score REAL, notes TEXT, status TEXT DEFAULT 'PENDING_REVIEW', assigned_expert TEXT, created_at TEXT NOT NULL ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS diagnoses_log ( image_id TEXT PRIMARY KEY, crop TEXT NOT NULL, disease TEXT NOT NULL, severity TEXT NOT NULL, confidence REAL NOT NULL, is_healthy INTEGER NOT NULL, created_at TEXT NOT NULL, payload_json TEXT NOT NULL ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS kisan_polls ( ballot_id TEXT PRIMARY KEY, problem_id TEXT NOT NULL, district TEXT NOT NULL, state TEXT NOT NULL, crop_affected TEXT, farmer_comment TEXT, device_id TEXT, created_at TEXT NOT NULL ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_kisan_polls_problem ON kisan_polls(problem_id)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_kisan_polls_district ON kisan_polls(district)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_kisan_polls_device ON kisan_polls(device_id)") conn.commit() finally: conn.close() def save_ticket(self, ticket_data: Dict[str, Any]) -> str: """Saves an expert review ticket to Firestore or local SQLite.""" created_at_str = ticket_data.get("created_at") or datetime.now(timezone.utc).isoformat() ticket_data["created_at"] = created_at_str # 1. Try Firestore if self._firestore_client: try: doc_ref = self._firestore_client.collection(settings.firestore_collection_tickets).document(ticket_data["ticket_id"]) doc_ref.set(ticket_data) logger.info(f"Saved ticket {ticket_data['ticket_id']} to Firestore.") except Exception as e: logger.error(f"Firestore save error: {e}") # 2. Persist to SQLite with self._lock: conn = sqlite3.connect(DB_FILE) try: cursor = conn.cursor() cursor.execute(""" INSERT OR REPLACE INTO expert_tickets ( ticket_id, farmer_name, phone_number, location, crop, disease_suspected, confidence_score, notes, status, assigned_expert, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( ticket_data["ticket_id"], ticket_data["farmer_name"], ticket_data["phone_number"], ticket_data.get("location"), ticket_data["crop"], ticket_data.get("disease_suspected"), ticket_data.get("confidence_score"), ticket_data.get("notes"), ticket_data.get("status", "PENDING_REVIEW"), ticket_data.get("assigned_expert", "Senior Agronomist"), created_at_str )) conn.commit() finally: conn.close() return ticket_data["ticket_id"] def log_diagnosis(self, diagnosis_data: Dict[str, Any]): """Logs an AI diagnosis event for audit, analytics, and disease surveillance.""" timestamp_str = diagnosis_data.get("timestamp") or datetime.now(timezone.utc).isoformat() if self._firestore_client: try: doc_ref = self._firestore_client.collection(settings.firestore_collection_diagnoses).document(diagnosis_data["image_id"]) doc_ref.set(diagnosis_data) except Exception as e: logger.debug(f"Firestore diagnosis log error: {e}") with self._lock: conn = sqlite3.connect(DB_FILE) try: cursor = conn.cursor() cursor.execute(""" INSERT OR REPLACE INTO diagnoses_log ( image_id, crop, disease, severity, confidence, is_healthy, created_at, payload_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( diagnosis_data["image_id"], diagnosis_data["crop"], diagnosis_data["disease"], diagnosis_data["severity"], diagnosis_data["confidence"], 1 if diagnosis_data.get("is_healthy", False) else 0, timestamp_str, json.dumps(diagnosis_data) )) conn.commit() finally: conn.close() def get_recent_tickets(self, limit: int = 20) -> List[Dict[str, Any]]: """Retrieves recent expert tickets.""" with self._lock: conn = sqlite3.connect(DB_FILE) try: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute("SELECT * FROM expert_tickets ORDER BY created_at DESC LIMIT ?", (limit,)) rows = cursor.fetchall() return [dict(row) for row in rows] finally: conn.close() def is_device_rate_limited(self, device_id: str, problem_id: str, hours: int = 24) -> bool: """Checks if a device has already cast a vote for the same issue in the last N hours.""" if not device_id: return False with self._lock: conn = sqlite3.connect(DB_FILE) try: cursor = conn.cursor() cursor.execute(""" SELECT created_at FROM kisan_polls WHERE device_id = ? AND problem_id = ? ORDER BY created_at DESC LIMIT 1 """, (device_id, problem_id)) row = cursor.fetchone() if not row: return False try: last_vote = datetime.fromisoformat(row[0].replace("Z", "+00:00")) now = datetime.now(timezone.utc) if (now - last_vote).total_seconds() < hours * 3600: return True except Exception: pass return False finally: conn.close() def save_poll_ballot(self, ballot_data: Dict[str, Any]) -> str: """Persists a verified farmer poll ballot into SQLite.""" created_at = ballot_data.get("timestamp") or datetime.now(timezone.utc).isoformat() ballot_id = ballot_data.get("ballot_id") or f"VOTE-{datetime.now().strftime('%Y%m%d')}-{uuid.uuid4().hex[:6].upper()}" with self._lock: conn = sqlite3.connect(DB_FILE) try: cursor = conn.cursor() cursor.execute(""" INSERT OR REPLACE INTO kisan_polls ( ballot_id, problem_id, district, state, crop_affected, farmer_comment, device_id, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( ballot_id, ballot_data["problem_id"], ballot_data.get("district", "Surguja"), ballot_data.get("state", "Chhattisgarh"), ballot_data.get("crop_affected", "Paddy (Rice)"), ballot_data.get("farmer_comment", ""), ballot_data.get("device_id", ""), created_at )) conn.commit() finally: conn.close() return ballot_id def get_poll_counts(self, district: Optional[str] = None) -> Dict[str, int]: """Returns live vote counts aggregated per problem ID.""" counts: Dict[str, int] = {} with self._lock: conn = sqlite3.connect(DB_FILE) try: cursor = conn.cursor() if district: cursor.execute(""" SELECT problem_id, COUNT(*) FROM kisan_polls WHERE district = ? COLLATE NOCASE GROUP BY problem_id """, (district,)) else: cursor.execute("SELECT problem_id, COUNT(*) FROM kisan_polls GROUP BY problem_id") rows = cursor.fetchall() for row in rows: counts[row[0]] = row[1] return counts finally: conn.close() def get_recent_ballots(self, limit: int = 50, district: Optional[str] = None) -> List[Dict[str, Any]]: """Retrieves most recent cast ballots.""" with self._lock: conn = sqlite3.connect(DB_FILE) try: conn.row_factory = sqlite3.Row cursor = conn.cursor() if district: cursor.execute(""" SELECT * FROM kisan_polls WHERE district = ? COLLATE NOCASE ORDER BY created_at DESC LIMIT ? """, (district, limit)) else: cursor.execute("SELECT * FROM kisan_polls ORDER BY created_at DESC LIMIT ?", (limit,)) rows = cursor.fetchall() return [dict(row) for row in rows] finally: conn.close() def seed_poll_baseline_if_empty(self, baseline_options: List[Dict[str, Any]]): """Seeds baseline grassroot ballots if the table is freshly initialized.""" with self._lock: conn = sqlite3.connect(DB_FILE) try: cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM kisan_polls") total = cursor.fetchone()[0] if total > 0: return logger.info("Initializing Kisan Jan-Mat SQLite database with baseline community sample votes...") districts = ["Surguja", "Bilaspur", "Raipur", "Bastar", "Janjgir-Champa", "Rajnandgaon"] sample_comments = { "spurious_pesticides": "दुकानदार से ली गई दवा से इल्ली नहीं मरी, पूरा 2 एकड़ खेत खराब हुआ।", "stray_cattle_wildlife": "रात में नीलगाय का झुंड आकर पूरी चना की फसल चट कर गया।", "water_electricity_scarcity": "टेल-एंड तक नहर का पानी नहीं पहुंचा, बोरवेल में कम वोल्टेज से मोटर नहीं चली।", "unseasonal_weather": "धान कटाई के समय ओले और तेज आंधी से बालियां झड़ गईं।", "mandi_cold_storage": "मंडी में टमाटर ₹3 किलो बिका, किराया भी नहीं निकला।", "uncontrolled_pest_epidemic": "अचानक भूरा माहू लग गया, 3 दिन में खेत जल गया।" } now_iso = datetime.now(timezone.utc).isoformat() for opt in baseline_options: p_id = opt["id"] count = opt.get("default_votes", 50) for i in range(count): b_id = f"INIT-{p_id[:4].upper()}-{i+1:04d}" dist = districts[i % len(districts)] comment = sample_comments.get(p_id, "") if i % 15 == 0 else "" cursor.execute(""" INSERT OR IGNORE INTO kisan_polls ( ballot_id, problem_id, district, state, crop_affected, farmer_comment, device_id, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( b_id, p_id, dist, "Chhattisgarh", "Paddy (Rice)", comment, f"DEV-SEED-{i%30}", now_iso )) conn.commit() logger.info("Kisan Jan-Mat baseline successfully seeded.") finally: conn.close() def get_diagnosis_outbreak_clusters(self, days: int = 14) -> List[Dict[str, Any]]: """Aggregates real user diagnoses from diagnoses_log to detect actual field outbreak clusters.""" with self._lock: conn = sqlite3.connect(DB_FILE) try: conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute(""" SELECT crop, disease, severity, COUNT(*) as report_count FROM diagnoses_log WHERE is_healthy = 0 GROUP BY crop, disease ORDER BY report_count DESC LIMIT 10 """) rows = cursor.fetchall() return [dict(row) for row in rows] finally: conn.close() db_manager = StorageManager()