refactor: extract load_domain_config into single _domain.py module — remove 7 copies, full rebuild on every call

paperforge/worker/_domain.py

@@ -0,0 +1,125 @@
+"""Shared domain config module — single source of truth for domain-collections.json.
+
+This module replaces 7 copy-pasted copies of load_domain_config scattered across
+worker files. It performs a FULL REBUILD from export files on every call,
+ensuring stale domains are cleaned up and new ones are added automatically.
+
+Data classification:
+  P2 derived cache — always rebuilt from exports/*.json (ground truth)
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from paperforge.worker._utils import read_json, write_json
+
+logger = logging.getLogger(__name__)
+
+
+# ── Collection tree utilities ────────────────────────────────────────────────
+
+
+def build_collection_lookup(collections: dict) -> dict:
+    """Build parent-resolved path cache from a Zotero collection tree.
+
+    Returns a dict with:
+      path_by_key: {collection_key: "Parent/Sub/Name", ...}
+      paths_by_item_id: {item_id: [collection_path, ...], ...}
+    """
+    path_cache = {}
+    item_paths = {}
+
+    def path_for(key: str) -> str:
+        if key in path_cache:
+            return path_cache[key]
+        node = collections.get(key, {})
+        parent = node.get("parent") or ""
+        name = node.get("name", "")
+        parent_path = path_for(parent) if parent else ""
+        full_path = f"{parent_path}/{name}" if parent_path else name
+        path_cache[key] = full_path
+        return full_path
+
+    for key, node in collections.items():
+        full_path = path_for(key)
+        for item_id in node.get("items", []):
+            item_paths.setdefault(item_id, []).append(full_path)
+    return {"path_by_key": path_cache, "paths_by_item_id": item_paths}
+
+
+def export_collection_paths(export_path: Path) -> list[str]:
+    """Extract all unique collection paths from a BBT export JSON file."""
+    data = read_json(export_path)
+    if not isinstance(data, dict):
+        return []
+    collections = data.get("collections", {})
+    if not isinstance(collections, dict):
+        return []
+    lookup = build_collection_lookup(collections)
+    paths = sorted(
+        {path for path in lookup.get("path_by_key", {}).values() if str(path or "").strip()},
+        key=lambda value: (value.count("/"), value),
+    )
+    return paths
+
+
+# ── Domain config ────────────────────────────────────────────────────────────
+
+
+def load_domain_config(paths: dict[str, Path]) -> dict:
+    """Load or FULLY REBUILD the domain-collections.json config.
+
+    Every call scans exports/*.json as ground truth:
+      - Adds domains for new export files
+      - Removes domains whose export files no longer exist
+      - Updates allowed_collections from actual Zotero collection tree
+
+    Returns: {"domains": [{"domain": ..., "export_file": ..., "allowed_collections": [...]}, ...]}
+    """
+    config_path = paths["config"]
+
+    try:
+        existing = read_json(config_path) if config_path.exists() else {"domains": []}
+    except Exception:
+        existing = {"domains": []}
+
+    old_entries = existing.get("domains", [])
+    export_files = sorted(paths["exports"].glob("*.json"))
+
+    fresh_entries = []
+    for export_path in export_files:
+        domain = export_path.stem
+        try:
+            collections = export_collection_paths(export_path)
+        except Exception:
+            logger.warning("Failed to parse export file: %s", export_path.name)
+            collections = []
+        fresh_entries.append(
+            {
+                "domain": domain,
+                "export_file": export_path.name,
+                "allowed_collections": collections,
+            }
+        )
+
+    config = {"domains": fresh_entries}
+
+    # Compare before/after to decide write
+    old_sorted = sorted(old_entries, key=lambda e: e.get("domain", ""))
+    new_sorted = sorted(fresh_entries, key=lambda e: e.get("domain", ""))
+    if old_sorted != new_sorted:
+        config_path.parent.mkdir(parents=True, exist_ok=True)
+        write_json(config_path, config)
+
+    return config
+
+
+def load_domain_collections(paths: dict[str, Path]) -> dict[str, list[str]]:
+    """Convenience: return {domain: [collection_path, ...]} mapping.
+
+    Used by candidate collection resolution in sync.py.
+    """
+    config = load_domain_config(paths)
+    return {entry["domain"]: entry.get("allowed_collections", []) for entry in config.get("domains", []) if entry.get("domain")}

paperforge/worker/base_views.py

@@ -65,25 +65,6 @@ def pipeline_paths(vault: Path) -> dict[str, Path]:
     }
 
 
-def load_domain_config(paths: dict[str, Path]) -> dict:
-    """Load or create the Lite domain mapping from export JSON files."""
-    config_path = paths["config"]
-    config = read_json(config_path) if config_path.exists() else {"domains": []}
-    domains = config.setdefault("domains", [])
-    known_exports = {str(entry.get("export_file", "")) for entry in domains}
-    changed = not config_path.exists()
-    for export_path in sorted(paths["exports"].glob("*.json")):
-        if export_path.name in known_exports:
-            continue
-        domains.append({"domain": export_path.stem, "export_file": export_path.name, "allowed_collections": []})
-        known_exports.add(export_path.name)
-        changed = True
-    if changed:
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-        write_json(config_path, config)
-    return config
-
-
 def base_markdown_filter(path: Path, vault: Path) -> str:
     try:
         return str(path.relative_to(vault)).replace("\\", "/")

paperforge/worker/deep_reading.py

@@ -5,6 +5,7 @@ import re
 from pathlib import Path
 
 from paperforge.config import paperforge_paths
+from paperforge.worker._domain import load_domain_config
 from paperforge.worker._utils import (
     read_json,
     scan_library_records,
@@ -51,25 +52,6 @@ def pipeline_paths(vault: Path) -> dict[str, Path]:
     }
 
 
-def load_domain_config(paths: dict[str, Path]) -> dict:
-    """Load or create the Lite domain mapping from export JSON files."""
-    config_path = paths["config"]
-    config = read_json(config_path) if config_path.exists() else {"domains": []}
-    domains = config.setdefault("domains", [])
-    known_exports = {str(entry.get("export_file", "")) for entry in domains}
-    changed = not config_path.exists()
-    for export_path in sorted(paths["exports"].glob("*.json")):
-        if export_path.name in known_exports:
-            continue
-        domains.append({"domain": export_path.stem, "export_file": export_path.name, "allowed_collections": []})
-        known_exports.add(export_path.name)
-        changed = True
-    if changed:
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-        write_json(config_path, config)
-    return config
-
-
 # Re-exported from _utils.py for backward compatibility
 
 

paperforge/worker/ocr.py

@@ -64,25 +64,6 @@ def pipeline_paths(vault: Path) -> dict[str, Path]:
     }
 
 
-def load_domain_config(paths: dict[str, Path]) -> dict:
-    """Load or create the Lite domain mapping from export JSON files."""
-    config_path = paths["config"]
-    config = read_json(config_path) if config_path.exists() else {"domains": []}
-    domains = config.setdefault("domains", [])
-    known_exports = {str(entry.get("export_file", "")) for entry in domains}
-    changed = not config_path.exists()
-    for export_path in sorted(paths["exports"].glob("*.json")):
-        if export_path.name in known_exports:
-            continue
-        domains.append({"domain": export_path.stem, "export_file": export_path.name, "allowed_collections": []})
-        known_exports.add(export_path.name)
-        changed = True
-    if changed:
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-        write_json(config_path, config)
-    return config
-
-
 def ensure_ocr_meta(vault: Path, row: dict) -> dict:
     paths = pipeline_paths(vault)
     key = row["zotero_key"]

paperforge/worker/repair.py

@@ -5,6 +5,7 @@ import re
 from pathlib import Path
 
 from paperforge.config import load_vault_config, paperforge_paths
+from paperforge.worker._domain import load_domain_config
 from paperforge.worker._utils import (
     _resolve_formal_note_path,
     read_json,
@@ -54,25 +55,6 @@ def pipeline_paths(vault: Path) -> dict[str, Path]:
     }
 
 
-def load_domain_config(paths: dict[str, Path]) -> dict:
-    """Load or create the Lite domain mapping from export JSON files."""
-    config_path = paths["config"]
-    config = read_json(config_path) if config_path.exists() else {"domains": []}
-    domains = config.setdefault("domains", [])
-    known_exports = {str(entry.get("export_file", "")) for entry in domains}
-    changed = not config_path.exists()
-    for export_path in sorted(paths["exports"].glob("*.json")):
-        if export_path.name in known_exports:
-            continue
-        domains.append({"domain": export_path.stem, "export_file": export_path.name, "allowed_collections": []})
-        known_exports.add(export_path.name)
-        changed = True
-    if changed:
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-        write_json(config_path, config)
-    return config
-
-
 def _find_export_for_domain(paths: dict[str, Path], domain: str) -> Path | None:
     """Find the BBT export JSON file for a given domain."""
     for export_path in sorted(paths["exports"].glob("*.json")):

paperforge/worker/status.py

@@ -9,6 +9,7 @@ from json import JSONDecodeError
 from pathlib import Path
 
 from paperforge.config import load_vault_config, paperforge_paths
+from paperforge.worker._domain import load_domain_config
 from paperforge.worker._utils import (
     read_json,
     write_json,
@@ -52,25 +53,6 @@ def pipeline_paths(vault: Path) -> dict[str, Path]:
     }
 
 
-def load_domain_config(paths: dict[str, Path]) -> dict:
-    """Load or create the Lite domain mapping from export JSON files."""
-    config_path = paths["config"]
-    config = read_json(config_path) if config_path.exists() else {"domains": []}
-    domains = config.setdefault("domains", [])
-    known_exports = {str(entry.get("export_file", "")) for entry in domains}
-    changed = not config_path.exists()
-    for export_path in sorted(paths["exports"].glob("*.json")):
-        if export_path.name in known_exports:
-            continue
-        domains.append({"domain": export_path.stem, "export_file": export_path.name, "allowed_collections": []})
-        known_exports.add(export_path.name)
-        changed = True
-    if changed:
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-        write_json(config_path, config)
-    return config
-
-
 def _detect_zotero_data_dir() -> str | None:
     """Try to detect the user's Zotero data directory."""
     if os.name == "nt":

paperforge/worker/sync.py

@@ -12,6 +12,7 @@ from xml.etree import ElementTree as ET
 import requests
 
 from paperforge.config import load_vault_config, paperforge_paths
+from paperforge.worker._domain import build_collection_lookup, load_domain_config, load_domain_collections
 from paperforge.worker._utils import (
     _extract_year,
     lookup_impact_factor,
@@ -62,87 +63,6 @@ def pipeline_paths(vault: Path) -> dict[str, Path]:
     }
 
 
-def load_domain_config(paths: dict[str, Path]) -> dict:
-    """Load or create the Lite domain mapping from export JSON files."""
-    config_path = paths["config"]
-    config = read_json(config_path) if config_path.exists() else {"domains": []}
-    domains = config.setdefault("domains", [])
-    known_exports = {str(entry.get("export_file", "")) for entry in domains}
-    changed = not config_path.exists()
-    for export_path in sorted(paths["exports"].glob("*.json")):
-        if export_path.name in known_exports:
-            continue
-        domains.append({"domain": export_path.stem, "export_file": export_path.name, "allowed_collections": []})
-        known_exports.add(export_path.name)
-        changed = True
-    if changed:
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-        write_json(config_path, config)
-    return config
-
-
-def build_collection_lookup(collections: dict) -> dict:
-    path_cache = {}
-    item_paths = {}
-
-    def path_for(key: str) -> str:
-        if key in path_cache:
-            return path_cache[key]
-        node = collections.get(key, {})
-        parent = node.get("parent") or ""
-        name = node.get("name", "")
-        parent_path = path_for(parent) if parent else ""
-        full_path = f"{parent_path}/{name}" if parent_path else name
-        path_cache[key] = full_path
-        return full_path
-
-    for key, node in collections.items():
-        full_path = path_for(key)
-        for item_id in node.get("items", []):
-            item_paths.setdefault(item_id, []).append(full_path)
-    return {"path_by_key": path_cache, "paths_by_item_id": item_paths}
-
-
-def export_collection_paths(export_path: Path) -> list[str]:
-    data = read_json(export_path)
-    if not isinstance(data, dict):
-        return []
-    collections = data.get("collections", {})
-    if not isinstance(collections, dict):
-        return []
-    lookup = build_collection_lookup(collections)
-    paths = sorted(
-        {path for path in lookup.get("path_by_key", {}).values() if str(path or "").strip()},
-        key=lambda value: (value.count("/"), value),
-    )
-    return paths
-
-
-def load_domain_collection_catalog(paths: dict[str, Path]) -> dict[str, list[str]]:
-    config_path = paths["config"]
-    config = read_json(config_path) if config_path.exists() else {"domains": []}
-    domain_entries = list(config.get("domains", []))
-    entry_by_export = {
-        entry.get("export_file", ""): dict(entry) for entry in domain_entries if entry.get("export_file")
-    }
-    export_files = sorted(paths["exports"].glob("*.json"))
-    changed = False
-    for export_path in export_files:
-        entry = entry_by_export.get(export_path.name)
-        if not entry:
-            entry = {"domain": export_path.stem, "export_file": export_path.name, "allowed_collections": []}
-            entry_by_export[export_path.name] = entry
-            changed = True
-        derived = export_collection_paths(export_path)
-        if entry.get("allowed_collections", []) != derived:
-            entry["allowed_collections"] = derived
-            changed = True
-    domains = sorted(entry_by_export.values(), key=lambda entry: entry.get("domain", ""))
-    if changed or config.get("domains", []) != domains:
-        write_json(config_path, {"domains": domains})
-    return {entry.get("domain", ""): entry.get("allowed_collections", []) for entry in domains if entry.get("domain")}
-
-
 def load_export_inventory(paths: dict[str, Path]) -> dict[str, dict]:
     inventory = {"doi": {}, "pmid": {}, "title": {}}
     for export_path in sorted(paths["exports"].glob("*.json")):
@@ -971,7 +891,7 @@ def load_candidates_by_id(paths: dict[str, Path]) -> dict[str, dict]:
 
 
 def save_candidates(paths: dict[str, Path], candidate_map: dict[str, dict]) -> None:
-    collection_catalog = load_domain_collection_catalog(paths)
+    collection_catalog = load_domain_collections(paths)
     export_inventory = load_export_inventory(paths)
     rows = []
     for row in candidate_map.values():

paperforge/worker/update.py

@@ -57,25 +57,6 @@ def pipeline_paths(vault: Path) -> dict[str, Path]:
     }
 
 
-def load_domain_config(paths: dict[str, Path]) -> dict:
-    """Load or create the Lite domain mapping from export JSON files."""
-    config_path = paths["config"]
-    config = read_json(config_path) if config_path.exists() else {"domains": []}
-    domains = config.setdefault("domains", [])
-    known_exports = {str(entry.get("export_file", "")) for entry in domains}
-    changed = not config_path.exists()
-    for export_path in sorted(paths["exports"].glob("*.json")):
-        if export_path.name in known_exports:
-            continue
-        domains.append({"domain": export_path.stem, "export_file": export_path.name, "allowed_collections": []})
-        known_exports.add(export_path.name)
-        changed = True
-    if changed:
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-        write_json(config_path, config)
-    return config
-
-
 def protected_paths(vault: Path) -> set[str]:
     cfg = load_vault_config(vault)
     pf = f"{cfg['system_dir']}/PaperForge"