From 61b72a467da19a8a9d4f2425d5c3c74745f65e1d Mon Sep 17 00:00:00 2001 From: Research Assistant Date: Tue, 28 Apr 2026 23:31:31 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20extract=20load=5Fdomain=5Fconfig=20?= =?UTF-8?q?into=20single=20=5Fdomain.py=20module=20=E2=80=94=20remove=207?= =?UTF-8?q?=20copies,=20full=20rebuild=20on=20every=20call?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- paperforge/worker/_domain.py | 125 ++++++++++++++++++++++++++++++ paperforge/worker/base_views.py | 19 ----- paperforge/worker/deep_reading.py | 20 +---- paperforge/worker/ocr.py | 19 ----- paperforge/worker/repair.py | 20 +---- paperforge/worker/status.py | 20 +---- paperforge/worker/sync.py | 84 +------------------- paperforge/worker/update.py | 19 ----- 8 files changed, 130 insertions(+), 196 deletions(-) create mode 100644 paperforge/worker/_domain.py diff --git a/paperforge/worker/_domain.py b/paperforge/worker/_domain.py new file mode 100644 index 00000000..03900507 --- /dev/null +++ b/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")} diff --git a/paperforge/worker/base_views.py b/paperforge/worker/base_views.py index c3490d4f..97bf76e7 100644 --- a/paperforge/worker/base_views.py +++ b/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("\\", "/") diff --git a/paperforge/worker/deep_reading.py b/paperforge/worker/deep_reading.py index c011a6c9..01e321c9 100644 --- a/paperforge/worker/deep_reading.py +++ b/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 diff --git a/paperforge/worker/ocr.py b/paperforge/worker/ocr.py index 360246a4..f64c48b3 100644 --- a/paperforge/worker/ocr.py +++ b/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"] diff --git a/paperforge/worker/repair.py b/paperforge/worker/repair.py index 6d192d06..c6e3c6a8 100644 --- a/paperforge/worker/repair.py +++ b/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")): diff --git a/paperforge/worker/status.py b/paperforge/worker/status.py index 11d04346..b0c9b213 100644 --- a/paperforge/worker/status.py +++ b/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": diff --git a/paperforge/worker/sync.py b/paperforge/worker/sync.py index 55e0022c..4bed1025 100644 --- a/paperforge/worker/sync.py +++ b/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(): diff --git a/paperforge/worker/update.py b/paperforge/worker/update.py index 24cb1bd4..0a4efbd2 100644 --- a/paperforge/worker/update.py +++ b/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"