| """ |
| 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 |
|
|
| |
| 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}") |
|
|
| |
| 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() |
|
|
