Update page_files/categorized/Backend/category_push.py

page_files/categorized/Backend/category_push.py

@@ -124,7 +124,15 @@ def _s(v: Any) -> str:
             return ""
     except (TypeError, ValueError):
         pass
-    return str(v)
+    s = str(v)
+    # Postgres text/varchar columns cannot contain NUL (0x00). PDF text
+    # extraction occasionally emits embedded nulls (and other C0 control
+    # chars) inside quotes/comments — strip them so the INSERT doesn't fail
+    # with "A string literal cannot contain NUL (0x00) characters".
+    if "\x00" in s or any(ord(c) < 32 and c not in "\t\n\r" for c in s):
+        s = s.replace("\x00", "")
+        s = "".join(c for c in s if ord(c) >= 32 or c in "\t\n\r")
+    return s
 
 
 def _page_int(v: Any) -> Optional[int]:
@@ -180,6 +188,12 @@ def conform_to_category(
     plots_root = plots_root or os.path.join(out_dir, "rds_plots")
     dest_dir = os.path.join(plots_root, category_table, stem)
 
+    # source_sha1 scopes the DB dedup index to THIS paper. It was previously ""
+    # for every row, which collapsed the unique key across all papers. A stable
+    # hash of the stem keeps re-pushes of the same paper consistent.
+    import hashlib
+    source_sha1 = hashlib.sha1((stem or "").encode("utf-8")).hexdigest()
+
     main_rows: List[Dict[str, Any]] = []
     extras_rows: List[Dict[str, Any]] = []
     n_plots = 0
@@ -252,7 +266,7 @@ def conform_to_category(
             "unit_canonical":        _s(row.get("unit")),
             "value_si":              None,
             "source_pdf":            _s(row.get("pdf_stem")) or stem,
-            "source_sha1":           "",
+            "source_sha1":           source_sha1,
             "page":                  _page_int(row.get("source_page")),
             "source_quote":          _s(row.get("source_text")),
             "status":                "extracted",
@@ -315,6 +329,22 @@ def push_by_category(
         df_store, category_table, out_dir, stem,
         plots_root=plots_root, embed_image=embed_image)
 
+    # keep-first dedup on the exact columns of ix_..._pipeline_dedup, so two
+    # identical measurements in one paper collapse to one row instead of
+    # tripping the unique constraint mid-batch.
+    _DEDUP_KEYS = ["source_sha1", "material_key", "section",
+                   "property_name", "test_condition", "value_raw"]
+    keys = [k for k in _DEDUP_KEYS if k in main_df.columns]
+    if keys:
+        before = len(main_df)
+        main_df = main_df.drop_duplicates(subset=keys, keep="first").reset_index(drop=True)
+        dropped = before - len(main_df)
+        if dropped:
+            log.info(f"category_push: dropped {dropped} in-batch duplicate row(s) on {keys}")
+        # keep extras aligned to the surviving main rows (1:1 by position)
+        extras_df = extras_df.loc[main_df.index].reset_index(drop=True) \
+            if len(extras_df) == before else extras_df
+
     extras_table = f"{category_table}_extras"
     # idempotency: clear this paper first (tables may not exist yet -> ignore)
     with engine.begin() as conn:
@@ -324,13 +354,33 @@ def push_by_category(
             except Exception:
                 pass
 
-    main_df.to_sql(category_table, engine, if_exists="append", index=False)
+    pushed = _insert_idempotent(engine, category_table, main_df)
     extras_df.to_sql(extras_table, engine, if_exists="append", index=False)
     if _s3_ready():
         plots_dir = _s3_url(_s3_key(category_table, stem, "")).rstrip("/")
     else:
         plots_dir = os.path.join(plots_root or os.path.join(out_dir, "rds_plots"),
                                  category_table, stem)
-    return {"pushed": len(main_df), "table": category_table,
+    return {"pushed": pushed, "table": category_table,
             "extras_table": extras_table, "extras": len(extras_df),
-            "plots_copied": n_plots, "plots_dir": plots_dir}
+            "plots_copied": n_plots, "plots_dir": plots_dir}
+
+
+def _insert_idempotent(engine, table: str, df: pd.DataFrame) -> int:
+    """Append rows one INSERT ... ON CONFLICT DO NOTHING at a time, so a row that
+    still collides with the table's unique dedup index is skipped instead of
+    rolling back the whole batch. Returns the number of rows actually inserted."""
+    from sqlalchemy import text
+    if df is None or df.empty:
+        return 0
+    cols = list(df.columns)
+    collist = ", ".join(f'"{c}"' for c in cols)
+    params = ", ".join(f":{c}" for c in cols)
+    sql = text(f'INSERT INTO "{table}" ({collist}) VALUES ({params}) ON CONFLICT DO NOTHING')
+    inserted = 0
+    with engine.begin() as conn:
+        for _, r in df.iterrows():
+            payload = {c: (None if pd.isna(v) else v) for c, v in r.items()}
+            res = conn.execute(sql, payload)
+            inserted += (res.rowcount or 0)
+    return inserted