Switch ingest IDs to code+year and handle duplicate ID rename

smoke_signal_tab.py

@@ -770,6 +770,141 @@ def next_book_id(df: pd.DataFrame) -> str:
     return "SS-BOOK-9999"
 
 
+_TITLE_STOPWORDS = {"the", "a", "an"}
+
+
+def _derive_book_code(title: str, code_hint: str = "") -> str:
+    hint = re.sub(r"[^a-z]", "", str(code_hint or "").lower())
+    if len(hint) >= 3:
+        return hint[:3]
+
+    words = re.findall(r"[a-z]+", str(title or "").lower())
+    if words and words[0] in _TITLE_STOPWORDS and len(words) > 1:
+        words = words[1:]
+    letters = "".join(words)
+    if not letters:
+        return "bok"
+    consonants = "".join(ch for ch in letters if ch not in "aeiou")
+    base = consonants[:3] if len(consonants) >= 3 else letters[:3]
+    return (base + "xxx")[:3]
+
+
+def _extract_year(*parts: str) -> str:
+    for part in parts:
+        match = re.search(r"\b(1[6-9]\d{2}|20\d{2})\b", str(part or ""))
+        if match:
+            return match.group(1)
+    return ""
+
+
+def _suggest_book_id(
+    title: str,
+    notes: str = "",
+    code_hint: str = "",
+    year_hint: str = "",
+    existing_ids: Optional[set[str]] = None,
+    current_id: str = "",
+) -> str:
+    code = _derive_book_code(title, code_hint=code_hint)
+    year = _extract_year(year_hint, notes, title) or "0000"
+    base = f"{code}{year}"
+    if existing_ids is None:
+        return base
+    if base not in existing_ids or base == current_id:
+        return base
+    for suffix in "abcdefghijklmnopqrstuvwxyz":
+        candidate = f"{base}{suffix}"
+        if candidate not in existing_ids or candidate == current_id:
+            return candidate
+    return base
+
+
+def _rename_book_id_references(old_id: str, new_id: str) -> None:
+    if not old_id or not new_id or old_id == new_id:
+        return
+
+    # Profile JSON
+    old_profile = PROFILES_DIR / f"{old_id}_page_profile.json"
+    new_profile = PROFILES_DIR / f"{new_id}_page_profile.json"
+    if old_profile.exists():
+        try:
+            data = json.load(open(old_profile, encoding="utf-8"))
+            data["book_id"] = new_id
+            with open(new_profile, "w", encoding="utf-8") as f:
+                json.dump(data, f, indent=2)
+            old_profile.unlink(missing_ok=True)
+        except Exception:
+            pass
+
+    # Render directory
+    old_render_dir = RENDERS_DIR / old_id
+    new_render_dir = RENDERS_DIR / new_id
+    if old_render_dir.exists() and not new_render_dir.exists():
+        old_render_dir.rename(new_render_dir)
+
+    # OCR raw directory + file
+    old_ocr_dir = OCR_RAW_DIR / old_id
+    new_ocr_dir = OCR_RAW_DIR / new_id
+    if old_ocr_dir.exists() and not new_ocr_dir.exists():
+        old_ocr_dir.rename(new_ocr_dir)
+    if new_ocr_dir.exists():
+        old_raw = new_ocr_dir / f"{old_id}_ocr_raw.json"
+        new_raw = new_ocr_dir / f"{new_id}_ocr_raw.json"
+        if old_raw.exists() and not new_raw.exists():
+            old_raw.rename(new_raw)
+        if new_raw.exists():
+            try:
+                raw_data = json.load(open(new_raw, encoding="utf-8"))
+                raw_data["book_id"] = new_id
+                with open(new_raw, "w", encoding="utf-8") as f:
+                    json.dump(raw_data, f, indent=2)
+            except Exception:
+                pass
+
+    # Queue and decision CSVs
+    for csv_path in [QUEUE_CSV, DECISIONS_CSV]:
+        if not csv_path.exists():
+            continue
+        try:
+            cdf = pd.read_csv(csv_path)
+            if "book_id" in cdf.columns:
+                cdf.loc[cdf["book_id"] == old_id, "book_id"] = new_id
+            if "region_id" in cdf.columns:
+                region_series = cdf["region_id"].astype(str)
+                mask = region_series.str.startswith(f"{old_id}_")
+                cdf.loc[mask, "region_id"] = region_series[mask].str.replace(
+                    f"{old_id}_", f"{new_id}_", n=1, regex=False
+                )
+            cdf.to_csv(csv_path, index=False)
+        except Exception:
+            pass
+
+    # Gold JSONL
+    if GOLD_FILE.exists():
+        tmp_path = GOLD_FILE.with_suffix(".tmp")
+        try:
+            with open(GOLD_FILE, "r", encoding="utf-8") as src, open(tmp_path, "w", encoding="utf-8") as dst:
+                for line in src:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    try:
+                        obj = json.loads(line)
+                    except Exception:
+                        dst.write(line + "\n")
+                        continue
+                    if obj.get("book_id") == old_id:
+                        obj["book_id"] = new_id
+                    region_id = str(obj.get("region_id", ""))
+                    if region_id.startswith(f"{old_id}_"):
+                        obj["region_id"] = region_id.replace(f"{old_id}_", f"{new_id}_", 1)
+                    dst.write(json.dumps(obj, ensure_ascii=False) + "\n")
+            tmp_path.replace(GOLD_FILE)
+        except Exception:
+            if tmp_path.exists():
+                tmp_path.unlink(missing_ok=True)
+
+
 def load_queue_df() -> pd.DataFrame:
     if not QUEUE_CSV.exists():
         return pd.DataFrame()
@@ -899,7 +1034,7 @@ def save_book_scope(book_id: str, include_spec: str, exclude_spec: str, safe_tit
 
 
 # ── Step 1: INGEST ─────────────────────────────────────────────────────────────
-def ingest_pdfs(files, rights_class: str, notes: str) -> tuple:
+def ingest_pdfs(files, rights_class: str, notes: str, book_code_hint: str = "", publication_year: str = "") -> tuple:
     """Register uploaded PDFs into the source manifest."""
     if not files:
         return _ingest_status_html("idle"), pd.DataFrame(), "No files uploaded."
@@ -918,11 +1053,25 @@ def ingest_pdfs(files, rights_class: str, notes: str) -> tuple:
 
         file_hash = sha256_file(path)
 
+        desired_book_id = _suggest_book_id(
+            Path(path.name).stem,
+            notes=notes,
+            code_hint=book_code_hint,
+            year_hint=publication_year,
+            existing_ids=set(df["book_id"].tolist()) if not df.empty else set(),
+        )
+
         # Check duplicate — update rights/notes if changed
         if not df.empty and file_hash in df["sha256"].values:
             existing_book_id = df.loc[df["sha256"] == file_hash, "book_id"].values[0]
             existing_status = df.loc[df["sha256"] == file_hash, "status"].values[0]
             existing_rights = df.loc[df["sha256"] == file_hash, "rights_class"].values[0]
+            if desired_book_id and desired_book_id != existing_book_id and desired_book_id not in set(df["book_id"].tolist()):
+                _rename_book_id_references(existing_book_id, desired_book_id)
+                df.loc[df["sha256"] == file_hash, "book_id"] = desired_book_id
+                existing_book_id = desired_book_id
+                log.append(log_line(f"↻ Renamed duplicate book ID → {desired_book_id}"))
+
             if rights_class != "unknown" and existing_rights != rights_class:
                 df.loc[df["sha256"] == file_hash, "rights_class"] = rights_class
                 df.loc[df["sha256"] == file_hash, "notes"] = notes
@@ -930,7 +1079,7 @@ def ingest_pdfs(files, rights_class: str, notes: str) -> tuple:
                 if existing_status == "pending":
                     auto_profile_ids.append(existing_book_id)
             else:
-                log.append(log_line(f"↩ Duplicate: {path.name} (rights={existing_rights})"))
+                log.append(log_line(f"↩ Duplicate: {path.name} ({existing_book_id}, rights={existing_rights})"))
             dup_count += 1
             continue
 
@@ -949,7 +1098,9 @@ def ingest_pdfs(files, rights_class: str, notes: str) -> tuple:
         except Exception:
             pass
 
-        book_id = next_book_id(df)
+        book_id = desired_book_id
+        if (not book_id) or (book_id in set(df["book_id"].tolist())):
+            book_id = next_book_id(df)
         new_row = pd.DataFrame([{
             "book_id":          book_id,
             "filename":         path.name,
@@ -2390,6 +2541,16 @@ def smoke_signal_tab():
                             choices=["public-domain","licensed-owned","controlled-internal","unknown"],
                             value="unknown",
                         )
+                        book_code_input = gr.Textbox(
+                            label="Book Code (3 letters)",
+                            placeholder="e.g. grf",
+                            lines=1,
+                        )
+                        pub_year_input = gr.Textbox(
+                            label="Publishing Year",
+                            placeholder="e.g. 1999",
+                            lines=1,
+                        )
                         ingest_notes = gr.Textbox(label="Notes", placeholder="Source, edition, etc.", lines=2)
                         ingest_btn = gr.Button("Register + Auto-Profile →", elem_classes=["ss-btn-run"])
 
@@ -2411,7 +2572,7 @@ def smoke_signal_tab():
 
                 gr.Markdown("**Update rights class for existing book:**")
                 with gr.Row():
-                    update_book_id = gr.Textbox(label="Book ID", placeholder="SS-BOOK-0001", scale=1)
+                    update_book_id = gr.Textbox(label="Book ID", placeholder="grf1999", scale=1)
                     update_rights_dd = gr.Dropdown(
                         label="New Rights Class",
                         choices=["public-domain","licensed-owned","controlled-internal","unknown"],
@@ -2427,7 +2588,7 @@ def smoke_signal_tab():
 
                 gr.Markdown("**Save Book Pages + Safe Title (persisted):**")
                 with gr.Row():
-                    scope_book_id = gr.Textbox(label="Book ID", placeholder="SS-BOOK-0001", scale=1)
+                    scope_book_id = gr.Textbox(label="Book ID", placeholder="grf1999", scale=1)
                     scope_include = gr.Textbox(label="Story Pages Include", placeholder="all or 7-27 or 7,8,9,11-27", scale=1)
                     scope_exclude = gr.Textbox(label="Story Pages Exclude", placeholder="e.g. 1,2,3,17,25", scale=1)
                     scope_title = gr.Textbox(label="Safe Book Title", placeholder="e.g. the-gruffalo", scale=1)
@@ -2440,7 +2601,7 @@ def smoke_signal_tab():
 
                 ingest_btn.click(
                     ingest_pdfs,
-                    inputs=[pdf_upload, rights_dd, ingest_notes],
+                    inputs=[pdf_upload, rights_dd, ingest_notes, book_code_input, pub_year_input],
                     outputs=[ingest_status, manifest_table, ingest_log],
                 )
                 gr.HTML('<div style="height:16px"></div>')