Update app/mapper.py

app/mapper.py

@@ -121,52 +121,98 @@ def reconcile_latest_schema(duck: duckdb.DuckDBPyConnection) -> None:
     print(f"[schema] ✅ reconciled {len(tables)} versions → canonical_latest")
 
 def canonify_df(org_id: str, hours_window: int = 24) -> pd.DataFrame:
+    """
+    Normalize, version, and persist canonical data snapshot for org_id.
+    This version pulls raw_rows as raw strings and parses JSON in Python so
+    malformed raw_rows don't crash the pipeline.
+    """
     load_dynamic_aliases()
     conn = get_conn(org_id)
     ensure_raw_table(conn)
 
-    # 1️⃣  optional timestamp filter (zero ? params)
-    cutoff_str = ""
-    if hours_window > 0:
-        cutoff = datetime.utcnow() - timedelta(hours=hours_window)
-        cutoff_str = cutoff.strftime('%Y-%m-%d %H:%M:%S')
-
-    rows = sql(conn, f"""
-        SELECT row_data
-        FROM raw_rows
-        WHERE row_data IS NOT NULL
-          AND LENGTH(row_data) > 0
-          {f"AND try_strptime(NULLIF(json_extract(row_data, '$.timestamp'), ''), '%Y-%m-%d %H:%M:%S') >= TIMESTAMP '{cutoff_str}'" if hours_window > 0 else ""}
-    """)
+    # 1) pull raw strings from DB (no JSON parsing in SQL)
+    try:
+        rows = conn.execute("SELECT row_data FROM main.raw_rows WHERE row_data IS NOT NULL AND LENGTH(row_data) > 0").fetchall()
+    except Exception as e:
+        print(f"[canonify] SQL read error: {e}")
+        rows = []
 
     if not rows:
-        print("[canonify] no rows")
+        print("[canonify] no rows to process")
+        return pd.DataFrame()
+
+    # 2) parse json strings safely in Python, skip bad ones
+    parsed = []
+    malformed_count = 0
+    for r in rows:
+        raw = r[0]
+        if not raw or not isinstance(raw, str):
+            malformed_count += 1
+            continue
+        try:
+            obj = json.loads(raw)
+        except Exception:
+            # Maybe raw is a single-object (not list) or legacy shape;
+            # attempt best-effort: ignore empty or malformed
+            malformed_count += 1
+            continue
+
+        # If this is a wrapper like {"rows": [...]} or {"data": [...]} or {"tables": {...}}
+        if isinstance(obj, dict):
+            # prefer list under rows, data, or fallback to tables flatten
+            if "rows" in obj and isinstance(obj["rows"], list):
+                parsed.extend(obj["rows"])
+            elif "data" in obj and isinstance(obj["data"], list):
+                parsed.extend(obj["data"])
+            elif "tables" in obj and isinstance(obj["tables"], dict):
+                # flatten: append all rows from all tables (optional)
+                for t_rows in obj["tables"].values():
+                    if isinstance(t_rows, list):
+                        parsed.extend(t_rows)
+            else:
+                # maybe the dict itself represents a single record
+                parsed.append(obj)
+        elif isinstance(obj, list):
+            parsed.extend(obj)
+        else:
+            # unknown shape — skip
+            malformed_count += 1
+            continue
+
+    if malformed_count:
+        print(f"[canonify] skipped {malformed_count} malformed/unsupported raw_rows")
+
+    if not parsed:
+        print("[canonify] no valid parsed rows after filtering")
         return pd.DataFrame()
 
-    # 2️⃣  normalise
-    raw = pd.DataFrame([json.loads(r[0]) for r in rows])
-    raw.columns = safe_str_transform(raw.columns)
+    # 3) build DataFrame and normalize column names
+    raw_df = pd.DataFrame(parsed)
+    if raw_df.empty:
+        print("[canonify] dataframe empty after parse")
+        return pd.DataFrame()
 
+    raw_df.columns = raw_df.columns.str.lower().str.strip()
     mapping = {}
     for canon, aliases in CANONICAL.items():
-        for col in raw.columns:
+        for col in raw_df.columns:
             if any(a in col for a in aliases):
                 mapping[col] = canon
                 break
 
-    # dynamic aliases
-    for col in raw.columns:
+    # learn dynamic aliases
+    for col in raw_df.columns:
         if col not in sum(CANONICAL.values(), []):
             for canon in CANONICAL.keys():
                 if canon in col and col not in CANONICAL[canon]:
                     CANONICAL[canon].append(col)
     save_dynamic_aliases()
 
-    renamed = raw.rename(columns=mapping)
+    renamed = raw_df.rename(columns=mapping)
     cols = [c for c in CANONICAL.keys() if c in renamed.columns]
     df = renamed[cols].copy() if cols else renamed.copy()
 
-    # 3️⃣  datatype fixes
+    # datatype conversions
     if "timestamp" in df:
         df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce")
     if "expiry_date" in df:
@@ -177,15 +223,28 @@ 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)
 
-    # 4️⃣  schema versioning & storage
+    # 4) persist canonical snapshot (use safe schema-versioning)
     os.makedirs("./db", exist_ok=True)
     duck = duckdb.connect(f"./db/{org_id}.duckdb")
 
     table_name = ensure_schema_version(duck, df)
+    # create table if not exists with the columns of 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")
+    # if table created above has no columns (rare), fallback to explicit column creation handled in ensure_schema_version
+    try:
+        duck.execute(f"INSERT INTO {table_name} SELECT * FROM df")
+    except Exception as e:
+        print(f"[canonify] insert error, retrying with explicit column checks: {e}")
+        # ensure columns exist individually
+        existing_cols = {r[0].lower() for r in duck.execute(f"PRAGMA table_info('{table_name}')").fetchall()}
+        for col in df.columns:
+            if col.lower() not in existing_cols:
+                dtype = map_pandas_to_duck(col, df[col])
+                duck.execute(f"ALTER TABLE {table_name} ADD COLUMN {col} {dtype}")
+        duck.execute(f"INSERT INTO {table_name} SELECT * FROM df")
+
     reconcile_latest_schema(duck)
     duck.close()
 
     print(f"[canonify] ✅ canonical snapshot updated for {org_id}")
-    return df
+    return df