Update app/mapper.py

app/mapper.py

@@ -1,9 +1,12 @@
-import os, json, duckdb, pandas as pd
+# mapper.py  –  production-hardened
+import os
+import json
+import duckdb
+import pandas as pd
 from datetime import datetime
 from app.db import get_conn, ensure_raw_table
 from app.utils.detect_industry import _ALIAS
 
-
 # ----------------------  Canonical schema base  ---------------------- #
 CANONICAL = {
     "timestamp":  ["timestamp", "date", "sale_date", "created_at"],
@@ -15,45 +18,44 @@ CANONICAL = {
     "promo_flag": ["promo", "promotion", "is_promo", "discount_code"],
     "expiry_date":["expiry_date", "best_before", "use_by", "expiration"],
 }
-
 ALIAS_FILE = "./db/alias_memory.json"
 
-def safe_str_transform(series: pd.Series) -> pd.Series:
-    """Apply .str.lower() & .str.strip() only if dtype is object/string."""
-    if pd.api.types.is_string_dtype(series):
-        return series.str.lower().str.strip()
-    return series
-# ----------------------  Alias memory helpers  ---------------------- #
+# ----------  helpers  ---------- #
+def safe_str_transform(s: pd.Series) -> pd.Series:
+    if pd.api.types.is_string_dtype(s):
+        return s.str.lower().str.strip()
+    return s
+
+def sql(conn, stmt: str, *args):
+    """Centralised parameter binding → no more int-vs-tuple mistakes."""
+    return conn.execute(stmt, args).fetchall()
+
+def add_column_if_not_exists(conn, table: str, col: str, dtype: str) -> None:
+    cols = {c[0] for c in conn.execute(f"DESCRIBE {table}").fetchall()}
+    if col.lower() not in cols:
+        conn.execute(f"ALTER TABLE {table} ADD COLUMN {col} {dtype}")
+
+# ----------  alias memory  ---------- #
 def load_dynamic_aliases() -> None:
-    """Load learned aliases and merge into CANONICAL."""
     if os.path.exists(ALIAS_FILE):
         try:
             with open(ALIAS_FILE) as f:
                 dynamic_aliases = json.load(f)
             for k, v in dynamic_aliases.items():
                 if k in CANONICAL:
-                    for alias in v:
-                        if alias not in CANONICAL[k]:
-                            CANONICAL[k].append(alias)
+                    CANONICAL[k].extend([a for a in v if a not in CANONICAL[k]])
                 else:
                     CANONICAL[k] = v
         except Exception as e:
             print(f"[mapper] ⚠️ failed to load alias memory: {e}")
 
-
 def save_dynamic_aliases() -> None:
-    """Persist learned aliases for next runs."""
     os.makedirs(os.path.dirname(ALIAS_FILE), exist_ok=True)
     with open(ALIAS_FILE, "w") as f:
         json.dump(CANONICAL, f, indent=2)
 
-
-# ----------------------  Schema versioning helpers  ---------------------- #
-def ensure_schema_version(duck, df: pd.DataFrame) -> str:
-    """
-    Ensure schema versioning and track evolution.
-    Returns the active canonical table name (e.g., main.canonical_v2).
-    """
+# ----------  schema versioning  ---------- #
+def ensure_schema_version(duck: duckdb.DuckDBPyConnection, df: pd.DataFrame) -> str:
     duck.execute("CREATE SCHEMA IF NOT EXISTS main")
     duck.execute("""
         CREATE TABLE IF NOT EXISTS main.schema_versions (
@@ -62,7 +64,6 @@ def ensure_schema_version(duck, df: pd.DataFrame) -> str:
             created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
         )
     """)
-
     latest = duck.execute("SELECT * FROM main.schema_versions ORDER BY version DESC LIMIT 1").fetchone()
     new_signature = sorted(df.columns.tolist())
 
@@ -70,81 +71,55 @@ def ensure_schema_version(duck, df: pd.DataFrame) -> str:
         latest_cols = sorted(json.loads(latest[1]))
         if latest_cols == new_signature:
             return f"main.canonical_v{latest[0]}"
-        else:
-            new_version = latest[0] + 1
-            duck.execute("INSERT INTO main.schema_versions (version, columns) VALUES (?, ?)",
-                         (new_version, json.dumps(new_signature)))
-            print(f"[schema] → new version detected: canonical_v{new_version}")
-            return f"main.canonical_v{new_version}"
+        new_version = latest[0] + 1
     else:
-        duck.execute("INSERT INTO main.schema_versions (version, columns) VALUES (?, ?)",
-                     (1, json.dumps(new_signature)))
-        print("[schema] → initialized canonical_v1")
-        return "main.canonical_v1"
-
+        new_version = 1
+    duck.execute("INSERT INTO main.schema_versions (version, columns) VALUES (?, ?)",
+                 (new_version, json.dumps(new_signature)))
+    print(f"[schema] → canonical_v{new_version}")
+    return f"main.canonical_v{new_version}"
 
-def reconcile_latest_schema(duck):
-    """
-    Merge all canonical_v* tables into main.canonical_latest
-    preserving new columns and filling missing values with NULL.
-    """
+def reconcile_latest_schema(duck: duckdb.DuckDBPyConnection) -> None:
     tables = [r[0] for r in duck.execute("""
         SELECT table_name FROM information_schema.tables
         WHERE table_name LIKE 'canonical_v%'
     """).fetchall()]
     if not tables:
         return
-
     union_query = " UNION ALL ".join([f"SELECT * FROM {t}" for t in tables])
     duck.execute("CREATE OR REPLACE TABLE main.canonical_latest AS " + union_query)
-    print(f"[schema] ✅ reconciled {len(tables)} schema versions → canonical_latest")
-
+    print(f"[schema] ✅ reconciled {len(tables)} versions → canonical_latest")
 
 # ----------------------  Canonify core logic  ---------------------- #
 def canonify_df(org_id: str, hours_window: int = 24) -> pd.DataFrame:
-    """
-    Normalize, version, and persist canonical data snapshot for org_id.
-    """
     load_dynamic_aliases()
     conn = get_conn(org_id)
     ensure_raw_table(conn)
 
-    # --------------------------
-    # ⏱  Safe timestamp filtering
-    # --------------------------
-    try:
-        rows = conn.execute(
-            """
-            SELECT row_data
-            FROM raw_rows
-            WHERE strptime(json_extract(row_data, '$.timestamp'), '%Y-%m-%d %H:%M:%S')
-                  >= now() - INTERVAL ? HOUR
-            """,
-            (hours_window,)
-        ).fetchall()
-    except Exception as e:
-        print(f"[canonify] ⚠️ fallback to all rows due to timestamp parse error: {e}")
-        rows = conn.execute("SELECT row_data FROM raw_rows").fetchall()
+    # 1️⃣  bullet-proof timestamp filter
+    rows = sql(conn, """
+        SELECT row_data
+        FROM raw_rows
+        WHERE try_strptime(NULLIF(json_extract(row_data, '$.timestamp'), ''),
+                           '%Y-%m-%d %H:%M:%S')
+              >= now() - INTERVAL ? HOUR
+    """, hours_window)
 
     if not rows:
-        print("[canonify] no rows to process")
+        print("[canonify] no rows")
         return pd.DataFrame()
 
-    # --------------------------
-    # 🧩 DataFrame normalization
-    # --------------------------
+    # 2️⃣  normalise
     raw = pd.DataFrame([json.loads(r[0]) for r in rows])
     raw.columns = safe_str_transform(raw.columns)
 
-    # Flexible alias mapping 
     mapping = {}
     for canon, aliases in CANONICAL.items():
         for col in raw.columns:
             if any(a in col for a in aliases):
                 mapping[col] = canon
                 break
-
-    # 🧠 Learn new aliases dynamically
+    # dynamic aliases
     for col in raw.columns:
         if col not in sum(CANONICAL.values(), []):
             for canon in CANONICAL.keys():
@@ -152,12 +127,11 @@ def canonify_df(org_id: str, hours_window: int = 24) -> pd.DataFrame:
                     CANONICAL[canon].append(col)
     save_dynamic_aliases()
 
-    # Apply canonical renaming
     renamed = raw.rename(columns=mapping)
     cols = [c for c in CANONICAL.keys() if c in renamed.columns]
     df = renamed[cols].copy() if cols else renamed.copy()
 
-    # 🔢 Normalize datatypes
+    # 3️⃣  datatype fixes
     if "timestamp" in df:
         df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce")
     if "expiry_date" in df:
@@ -168,19 +142,15 @@ def canonify_df(org_id: str, hours_window: int = 24) -> pd.DataFrame:
         if col in df:
             df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0)
 
-    # --------------------------
-    # 🪣 Schema versioning + storage
-    # --------------------------
+    # 4️⃣  schema versioning & storage
     os.makedirs("./db", exist_ok=True)
     duck = duckdb.connect(f"./db/{org_id}.duckdb")
 
     table_name = ensure_schema_version(duck, df)
     duck.execute(f"CREATE TABLE IF NOT EXISTS {table_name} AS SELECT * FROM df LIMIT 0")
     duck.execute(f"INSERT INTO {table_name} SELECT * FROM df")
-
-    # 🧩 Always refresh canonical_latest for unified analytics
     reconcile_latest_schema(duck)
     duck.close()
 
     print(f"[canonify] ✅ canonical snapshot updated for {org_id}")
-    return df
+    return df