Upload folder using huggingface_hub

brain/db_handler.py

@@ -0,0 +1,179 @@
+import os
+import json
+import time
+import sqlite3
+import logging
+import threading
+from pathlib import Path
+from typing import List, Dict, Any, Optional
+from supabase import create_client, Client
+
+logger = logging.getLogger(__name__)
+
+# Local fallback queue. If Supabase is unreachable, save_report writes here
+# so the report is never lost. flush_pending() can be called periodically to
+# replay. The HF container has ephemeral disk, so this only protects against
+# transient outages, not full container loss — that's acceptable.
+_DEFAULT_FALLBACK_PATH = os.getenv(
+    "OPS_FALLBACK_DB", str(Path(__file__).parent.parent / ".ops_fallback.db")
+)
+
+
+class StoreDB:
+    def __init__(self, url: str, key: str, fallback_path: str = _DEFAULT_FALLBACK_PATH):
+        if not url or not key:
+            raise ValueError("StoreDB: Supabase url and key are required")
+        self.supabase: Client = create_client(url, key)
+        self._fb_path = fallback_path
+        self._fb_lock = threading.Lock()
+        self._init_fallback()
+
+    # ─── Fallback queue (SQLite WAL) ────────────────────────────────────────────
+    def _init_fallback(self) -> None:
+        try:
+            with sqlite3.connect(self._fb_path) as conn:
+                # WAL = crash-safe + concurrent readers
+                conn.execute("PRAGMA journal_mode=WAL;")
+                conn.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS pending_reports (
+                        id          INTEGER PRIMARY KEY AUTOINCREMENT,
+                        payload     TEXT    NOT NULL,
+                        attempts    INTEGER NOT NULL DEFAULT 0,
+                        last_error  TEXT,
+                        created_at  REAL    NOT NULL
+                    );
+                    """
+                )
+                conn.commit()
+        except Exception as e:
+            logger.warning("StoreDB: could not init fallback queue at %s: %s", self._fb_path, e)
+
+    def _queue_pending(self, payload: Dict[str, Any], last_error: str) -> None:
+        try:
+            with self._fb_lock, sqlite3.connect(self._fb_path) as conn:
+                conn.execute(
+                    "INSERT INTO pending_reports (payload, attempts, last_error, created_at) "
+                    "VALUES (?, ?, ?, ?)",
+                    (json.dumps(payload, default=str), 99, last_error[:500], time.time()),
+                )
+                conn.commit()
+        except Exception as e:
+            logger.error("StoreDB: failed to queue pending report: %s", e)
+
+    def flush_pending(self, max_items: int = 50) -> int:
+        """
+        Try to replay queued reports into Supabase. Returns number successfully flushed.
+        Safe to call periodically (e.g. hourly cron / health endpoint).
+        """
+        flushed = 0
+        try:
+            with self._fb_lock, sqlite3.connect(self._fb_path) as conn:
+                conn.row_factory = sqlite3.Row
+                rows = conn.execute(
+                    "SELECT id, payload FROM pending_reports ORDER BY id ASC LIMIT ?",
+                    (max_items,),
+                ).fetchall()
+        except Exception as e:
+            logger.warning("StoreDB.flush_pending: could not read queue: %s", e)
+            return 0
+
+        for row in rows:
+            try:
+                payload = json.loads(row["payload"])
+                self.supabase.table("store_reports").insert(payload).execute()
+                with self._fb_lock, sqlite3.connect(self._fb_path) as conn:
+                    conn.execute("DELETE FROM pending_reports WHERE id = ?", (row["id"],))
+                    conn.commit()
+                flushed += 1
+            except Exception as e:
+                logger.warning("StoreDB.flush_pending: replay failed for id=%s: %s", row["id"], e)
+                with self._fb_lock, sqlite3.connect(self._fb_path) as conn:
+                    conn.execute(
+                        "UPDATE pending_reports SET attempts = attempts + 1, last_error = ? WHERE id = ?",
+                        (str(e)[:500], row["id"]),
+                    )
+                    conn.commit()
+        return flushed
+
+    def pending_count(self) -> int:
+        try:
+            with self._fb_lock, sqlite3.connect(self._fb_path) as conn:
+                cur = conn.execute("SELECT COUNT(*) FROM pending_reports")
+                return int(cur.fetchone()[0])
+        except Exception:
+            return -1
+
+    # ─── Public API ─────────────────────────────────────────────────────────────
+    def save_report(self, report_data: Dict[str, Any]) -> Any:
+        """
+        Saves a parsed report to the 'store_reports' table.
+        On Supabase failure, queues the report locally so it's not lost.
+        """
+        data = {
+            "store_id": report_data.get("store_id"),
+            "sales": (report_data.get("metrics") or {}).get("sales"),
+            "inventory_status": (report_data.get("metrics") or {}).get("inventory_status"),
+            "staffing": (report_data.get("metrics") or {}).get("staffing"),
+            "issues": report_data.get("issues") or [],
+            "analysis": report_data.get("analysis"),
+            "actions": report_data.get("actions_needed") or [],
+            "report_date": time.strftime('%Y-%m-%d') # Force ISO date for Dashboard sync
+        }
+        try:
+            return self.supabase.table("store_reports").insert(data).execute()
+        except Exception as e:
+            logger.error("StoreDB.save_report: Supabase insert failed, queuing locally: %s", e)
+            self._queue_pending(data, str(e))
+            # Re-raise so the caller (Telegram handler) can decide UX.
+            raise
+
+    def get_latest_reports(self, limit: int = 20) -> Any:
+        return (
+            self.supabase.table("store_reports")
+            .select("*")
+            .order("created_at", desc=True)
+            .limit(limit)
+            .execute()
+        )
+
+    def get_all_store_summaries(self) -> Any:
+        return self.supabase.table("store_reports").select("*").execute()
+
+    def get_recent_operator_actions(self, limit: int = 50) -> Any:
+        """For the dashboard's operator log panel."""
+        return (
+            self.supabase.table("operator_logs")
+            .select("*")
+            .order("timestamp", desc=True)
+            .limit(limit)
+            .execute()
+        )
+
+    def log_operator_action(self, store_id: str, action: str, notes: str) -> Any:
+        """
+        Logs a manual operator correction/intervention to a separate audit table.
+        """
+        data = {
+            "store_id": store_id,
+            "action_type": action,
+            "notes": notes,
+            "timestamp": "now()",
+        }
+        return self.supabase.table("operator_logs").insert(data).execute()
+
+    def save_bulk_reports(self, reports: List[Dict[str, Any]]) -> Any:
+        """
+        Saves a list of reports in a single batch operation.
+        Essential for processing large Excel/CSV imports without hitting API limits.
+        """
+        try:
+            return self.supabase.table("store_reports").insert(reports).execute()
+        except Exception as e:
+            logger.error("StoreDB.save_bulk_reports: Batch insert failed: %s", e)
+            for report in reports:
+                try:
+                    self.save_report(report)
+                except:
+                    pass
+            raise e

brain/ops_brain.py

@@ -0,0 +1,157 @@
+import os
+import json
+import time
+import logging
+from typing import List, Dict, Any, Optional
+from groq import Groq
+
+logger = logging.getLogger(__name__)
+
+# Default model. Groq deprecated `llama-3.3-70b-versatile` — llama-3.1-70b
+# is the supported current SKU. Override via env for fast A/B.
+DEFAULT_MODEL = os.getenv("OPS_BRAIN_MODEL", "llama-3.1-70b-versatile")
+
+GROQ_TIMEOUT_S = 15.0
+GROQ_MAX_RETRIES = 2
+GROQ_RETRY_BACKOFF_S = 1.5
+
+
+def _safe_json_loads(raw: str) -> Optional[Dict[str, Any]]:
+    """
+    Parse JSON from a model response, tolerating stray prose / markdown fences.
+    Returns None if unparseable.
+    """
+    if not raw:
+        return None
+    text = raw.strip()
+
+    # Strip ```json ... ``` fences if present.
+    if text.startswith("```"):
+        first_nl = text.find("\n")
+        if first_nl != -1:
+            text = text[first_nl + 1 :]
+        if text.endswith("```"):
+            text = text[:-3]
+
+    # First attempt: strict.
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        pass
+
+    # Second attempt: find the first '{' and last '}' and try that slice.
+    start = text.find("{")
+    end = text.rfind("}")
+    if start != -1 and end != -1 and end > start:
+        try:
+            return json.loads(text[start : end + 1])
+        except json.JSONDecodeError:
+            pass
+
+    logger.error("ops_brain: could not parse JSON from model output: %r", raw[:300])
+    return None
+
+
+def _groq_chat(client: Groq, prompt: str, model: str) -> Optional[Dict[str, Any]]:
+    """Call Groq with timeout, retry, and JSON repair."""
+    last_err: Optional[Exception] = None
+    for attempt in range(1, GROQ_MAX_RETRIES + 1):
+        try:
+            completion = client.chat.completions.create(
+                model=model,
+                messages=[{"role": "user", "content": prompt}],
+                response_format={"type": "json_object"},
+                timeout=GROQ_TIMEOUT_S,
+            )
+            raw = completion.choices[0].message.content
+            parsed = _safe_json_loads(raw)
+            if parsed is not None:
+                return parsed
+            last_err = ValueError("json parse failed")
+        except Exception as e:  # groq.GroqError, httpx.TimeoutException, etc.
+            last_err = e
+            logger.warning("ops_brain: groq attempt %d/%d failed: %s", attempt, GROQ_MAX_RETRIES, e)
+
+        if attempt < GROQ_MAX_RETRIES:
+            time.sleep(GROQ_RETRY_BACKOFF_S * attempt)
+
+    logger.error("ops_brain: giving up after %d attempts. last_err=%s", GROQ_MAX_RETRIES, last_err)
+    return None
+
+
+class OpsManagerAI:
+    def __init__(self, api_key: str, model: str = DEFAULT_MODEL):
+        if not api_key or not api_key.strip():
+            raise ValueError("OpsManagerAI: api_key is empty")
+        self.client = Groq(api_key=api_key)
+        self.model = model
+
+    def process_telegram_message(self, text: str) -> Dict[str, Any]:
+        """
+        Parses store reports into structured JSON using Groq.
+        Returns a safe-default dict (with store_id=None) if the model fails,
+        so the bot never crashes the handler thread.
+        """
+        prompt = f"""
+        You are a Professional AI Operations Manager. Your task is to parse store reports into a strict JSON format.
+
+        The user provides reports in a specific template like:
+        'Daily Update [ Store Name ] Date: [Date] 💵 Sales:; (Value) 🛍️Transactions: [Value] 📊Average Transaction (AT): [Value] ⬆️⬇️AT Yesterday[ Value]'
+
+        Input Text: {text}
+
+        Required JSON Output:
+        {{
+            "store_id": "Extract the name inside the brackets [ ]",
+            "metrics": {{
+                "sales": float or null,
+                "inventory_status": "Good|Warning|Critical",
+                "staffing": "OK|Understaffed|Overstaffed"
+            }},
+            "issues": ["List any mentioned anomalies or low stock, otherwise empty"],
+            "analysis": "A brief analytical summary of the store's health today",
+            "actions_needed": ["Concrete, actionable steps based on the sales/AT trends"]
+        }}
+        Return ONLY the JSON object. No preamble.
+        """
+
+        parsed = _groq_chat(self.client, prompt, self.model)
+        if parsed is not None and parsed.get("store_id"):
+            return parsed
+
+        # Safe fallback so caller can decide to reply with an error instead of crashing.
+        return {
+            "store_id": None,
+            "metrics": {"sales": None, "inventory_status": None, "staffing": None},
+            "issues": [],
+            "analysis": "AI could not extract a valid store report from the message.",
+            "actions_needed": [],
+        }
+
+    def generate_hot_list_analysis(self, all_stores_data: List[Dict]) -> Dict[str, Any]:
+        """
+        Analyze the full fleet to find critical areas.
+        """
+        prompt = f"""
+        Analyze the following store data and provide a strategic operation summary.
+        Data: {json.dumps(all_stores_data)}
+
+        Return JSON:
+        {{
+            "top_performers": ["Store name"],
+            "critical_stores": ["Store name"],
+            "global_dev_area": "General area needing improvement",
+            "strategic_priority": "Top immediate action for the owner"
+        }}
+        Return ONLY JSON.
+        """
+
+        parsed = _groq_chat(self.client, prompt, self.model)
+        if parsed is not None:
+            return parsed
+        return {
+            "top_performers": [],
+            "critical_stores": [],
+            "global_dev_area": "AI analysis unavailable.",
+            "strategic_priority": "Manual review required.",
+        }