Update app/mapper.py

app/mapper.py

@@ -3,7 +3,7 @@ import os
 import json
 import duckdb
 import pandas as pd
-from datetime import datetime
+from datetime import datetime, timedelta
 from app.db import get_conn, ensure_raw_table
 from app.utils.detect_industry import _ALIAS
 
@@ -90,20 +90,22 @@ def reconcile_latest_schema(duck: duckdb.DuckDBPyConnection) -> None:
     duck.execute("CREATE OR REPLACE TABLE main.canonical_latest AS " + union_query)
     print(f"[schema] ✅ reconciled {len(tables)} versions → canonical_latest")
 
-# ----------------------  Canonify core logic  ---------------------- #
 def canonify_df(org_id: str, hours_window: int = 24) -> pd.DataFrame:
     load_dynamic_aliases()
     conn = get_conn(org_id)
     ensure_raw_table(conn)
 
-    # 1️⃣  bullet-proof timestamp filter
-    rows = sql(conn, """
+    # 1️⃣  zero-parameter timestamp filter
+    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 try_strptime(NULLIF(json_extract(row_data, '$.timestamp'), ''),
                            '%Y-%m-%d %H:%M:%S')
-              >= now() - INTERVAL ? HOUR
-    """, hours_window)
+          >= TIMESTAMP '{cutoff_str}'
+    """)
 
     if not rows:
         print("[canonify] no rows")
@@ -119,6 +121,7 @@ def canonify_df(org_id: str, hours_window: int = 24) -> pd.DataFrame:
             if any(a in col for a in aliases):
                 mapping[col] = canon
                 break
+
     # dynamic aliases
     for col in raw.columns:
         if col not in sum(CANONICAL.values(), []):
@@ -151,6 +154,6 @@ def canonify_df(org_id: str, hours_window: int = 24) -> pd.DataFrame:
     duck.execute(f"INSERT INTO {table_name} SELECT * FROM df")
     reconcile_latest_schema(duck)
     duck.close()
-    #print  df 
+
     print(f"[canonify] ✅ canonical snapshot updated for {org_id}")
     return df