diff --git a/paperforge/adapters/obsidian_frontmatter.py b/paperforge/adapters/obsidian_frontmatter.py index 20331c40..5f84bc0d 100644 --- a/paperforge/adapters/obsidian_frontmatter.py +++ b/paperforge/adapters/obsidian_frontmatter.py @@ -40,6 +40,17 @@ def _read_frontmatter_bool_from_text(text: str, key: str, default: bool = False) return match.group(1).lower() == "true" +def read_frontmatter_bool(note_path: Path, key: str, default: bool = False) -> bool: + """Read a boolean field from a formal note's YAML frontmatter (Path-based).""" + if not note_path or not note_path.exists(): + return default + try: + text = note_path.read_text(encoding="utf-8") + return _read_frontmatter_bool_from_text(text, key, default) + except Exception: + return default + + def _read_frontmatter_optional_bool_from_text(text: str, key: str) -> Optional[bool]: match = re.search(rf"^{re.escape(key)}:\s*(?:[\"'])?(true|false)(?:[\"'])?\s*$", text, re.MULTILINE | re.IGNORECASE) if not match: @@ -47,6 +58,17 @@ def _read_frontmatter_optional_bool_from_text(text: str, key: str) -> Optional[b return match.group(1).lower() == "true" +def read_frontmatter_optional_bool(note_path: Path, key: str) -> Optional[bool]: + """Read an optional boolean field from a formal note's YAML frontmatter (Path-based).""" + if not note_path or not note_path.exists(): + return None + try: + text = note_path.read_text(encoding="utf-8") + return _read_frontmatter_optional_bool_from_text(text, key) + except Exception: + return None + + def _legacy_control_flags(paths: dict[str, Path], zotero_key: str) -> dict[str, Optional[bool]]: records_root = paths.get("library_records") if not records_root or not records_root.exists(): @@ -220,30 +242,28 @@ def has_deep_reading_content(text: str) -> bool: if not body: return False - clarity_ok = bool(re.search( - r'-\s*\*\*Clarity\*\*（清晰度）：(.+)', body - )) and not re.search( - r'-\s*\*\*Clarity\*\*（清晰度）：\s*$', body, re.MULTILINE + clarity_ok = bool(re.search(r"-\s*\*\*Clarity\*\*（清晰度）：(.+)", body)) and not re.search( + r"-\s*\*\*Clarity\*\*（清晰度）：\s*$", body, re.MULTILINE ) - figure_sec = _extract_section(body, r'\*\*Figure 导读\*\*') + figure_sec = _extract_section(body, r"\*\*Figure 导读\*\*") figure_ok = False if figure_sec: for line in figure_sec.splitlines(): s = line.strip() - if s.startswith('- ') and '：' in s: - _, after = s.split('：', 1) - if after.strip() and after.strip() != '（待补充）': + if s.startswith("- ") and "：" in s: + _, after = s.split("：", 1) + if after.strip() and after.strip() != "（待补充）": figure_ok = True break - issue_sec = _extract_section(body, r'\*\*遗留问题\*\*') + issue_sec = _extract_section(body, r"\*\*遗留问题\*\*") if not issue_sec: - issue_sec = _extract_section(body, r'####\s*遗留问题') + issue_sec = _extract_section(body, r"####\s*遗留问题") issue_ok = False if issue_sec: dirty = [l.strip() for l in issue_sec.splitlines() if l.strip()] - substantive = [l for l in dirty if l not in ('-', '（待补充）', '', '**遗留问题**')] + substantive = [l for l in dirty if l not in ("-", "（待补充）", "", "**遗留问题**")] issue_ok = bool(substantive) return clarity_ok and figure_ok and issue_ok @@ -251,8 +271,9 @@ def has_deep_reading_content(text: str) -> bool: def _extract_section(body: str, section_header: str) -> str | None: m = re.search( - section_header + r'\n*(.*?)(?=\n(?:#{1,3})\s|\Z)', - body, re.DOTALL, + section_header + r"\n*(.*?)(?=\n(?:#{1,3})\s|\Z)", + body, + re.DOTALL, ) if m: return m.group(1).strip() @@ -298,7 +319,7 @@ def read_frontmatter_dict(text: str) -> dict: import re import yaml - fm_match = re.match(r'^---\s*\n(.*?)\n---', text, re.DOTALL) + fm_match = re.match(r"^---\s*\n(.*?)\n---", text, re.DOTALL) if not fm_match: return {} @@ -312,7 +333,7 @@ def read_frontmatter_dict(text: str) -> dict: result = {} for line in fm_match.group(1).splitlines(): line = line.strip() - if ':' in line: - key, _, val = line.partition(':') + if ":" in line: + key, _, val = line.partition(":") result[key.strip()] = val.strip() return result diff --git a/paperforge/config.py b/paperforge/config.py index 37016ffc..8f634566 100644 --- a/paperforge/config.py +++ b/paperforge/config.py @@ -333,6 +333,9 @@ def paperforge_paths( "worker_script": worker_script, "skill_dir": skill_path, "ld_deep_script": ld_deep_script, + # ── v2.2: canonical locations below paperforge/ ── + "config": paperforge / "config" / "domain-collections.json", + "index": paperforge / "indexes" / "formal-library.json", } @@ -404,10 +407,7 @@ def migrate_paperforge_json(vault: Path) -> bool: # --- Build the cleaned output --- # 1. Non-path top-level keys to preserve (everything except path keys and vault_config) - non_path_keys = { - k: v for k, v in data.items() - if k not in CONFIG_PATH_KEYS and k != "vault_config" - } + non_path_keys = {k: v for k, v in data.items() if k not in CONFIG_PATH_KEYS and k != "vault_config"} # 2. Build vault_config: start with existing vault_config, fill gaps from top-level vault_config = dict(data.get("vault_config", {}) or {}) diff --git a/paperforge/services/sync_service.py b/paperforge/services/sync_service.py index 093b308c..5b04b289 100644 --- a/paperforge/services/sync_service.py +++ b/paperforge/services/sync_service.py @@ -98,13 +98,12 @@ class SyncService: continue try: content = record_file.read_text(encoding="utf-8") - key_match = re.search(r"^zotero_key:\s*(.+)$", content, re.MULTILINE) - if not key_match: + fm = read_frontmatter_dict(content) + key = str(fm.get("zotero_key", "")).strip() + if not key: continue - key = key_match.group(1).strip() - title_match = re.search(r"^title:\s*[\"']?(.+)[\"']?\s*$", content, re.MULTILINE) - title = title_match.group(1) if title_match else "" - has_pdf = "has_pdf: true" in content + title = str(fm.get("title", "")) + has_pdf = fm.get("has_pdf", False) is True normalized = re.sub(r"[^a-z0-9]", "", title.lower())[:20] records_info[key] = { "file": record_file, @@ -182,8 +181,8 @@ class SyncService: for flat_note in list(domain_dir.glob("*.md")): try: text = flat_note.read_text(encoding="utf-8") - m = re.search(r"^zotero_key:\s*\"?(\S+?)\"?\s*$", text, re.MULTILINE) - key = m.group(1) if m else "" + fm = read_frontmatter_dict(text) + key = str(fm.get("zotero_key", "")).strip() except Exception: continue if key and key in ws_keys: @@ -245,11 +244,10 @@ class SyncService: ) # ── Phase 2: Index ── - from paperforge.worker._utils import pipeline_paths from paperforge.worker._domain import load_domain_config from paperforge.worker.base_views import ensure_base_views - paths = pipeline_paths(self.vault) + paths = self.resolve_paths() config = load_domain_config(paths) ensure_base_views(self.vault, paths, config) domain_lookup = {entry["export_file"]: entry["domain"] for entry in config["domains"]} diff --git a/paperforge/skills/literature-qa/scripts/ld_deep.py b/paperforge/skills/literature-qa/scripts/ld_deep.py index 04f2725c..3e613f39 100644 --- a/paperforge/skills/literature-qa/scripts/ld_deep.py +++ b/paperforge/skills/literature-qa/scripts/ld_deep.py @@ -1374,9 +1374,9 @@ def prepare_deep_reading(vault: Path, zotero_key: str, force: bool = False) -> d if key_match: domain_match = re.search(r"^domain:\s*(.+)$", text, re.MULTILINE) domain = domain_match.group(1).strip() if domain_match else candidate.parent.name - from paperforge.worker.asset_index import _read_frontmatter_bool + from paperforge.adapters.obsidian_frontmatter import read_frontmatter_bool - if not _read_frontmatter_bool(candidate, "analyze"): + if not read_frontmatter_bool(candidate, "analyze"): result["message"] = ( f"[ERROR] analyze=False for {zotero_key} in {candidate}. " "Set analyze: true first." diff --git a/paperforge/worker/asset_index.py b/paperforge/worker/asset_index.py index 72c5fb21..a36406ff 100644 --- a/paperforge/worker/asset_index.py +++ b/paperforge/worker/asset_index.py @@ -35,6 +35,11 @@ from pathlib import Path import filelock from paperforge import __version__ as _paperforge_version +from paperforge.adapters.obsidian_frontmatter import ( + _legacy_control_flags, + read_frontmatter_bool, + read_frontmatter_optional_bool, +) from paperforge.config import paperforge_paths logger = logging.getLogger(__name__) @@ -207,49 +212,8 @@ def migrate_legacy_index(vault: Path) -> bool: # --------------------------------------------------------------------------- -# Single-entry builder (shared by full rebuild and incremental refresh) -# --------------------------------------------------------------------------- - - -def _read_frontmatter_bool(note_path: Path, key: str, default: bool = False) -> bool: - """Read a boolean field from a formal note's YAML frontmatter.""" - if not note_path or not note_path.exists(): - return default - try: - text = note_path.read_text(encoding="utf-8") - m = re.search(rf"^{key}:\s*(?:[\"'])?(true|false)(?:[\"'])?\s*$", text, re.MULTILINE | re.IGNORECASE) - if m: - return m.group(1).lower() == "true" - except Exception: - pass - return default - - -def _read_frontmatter_optional_bool(note_path: Path, key: str) -> bool | None: - if not note_path or not note_path.exists(): - return None - try: - text = note_path.read_text(encoding="utf-8") - m = re.search(rf"^{re.escape(key)}:\s*(?:[\"'])?(true|false)(?:[\"'])?\s*$", text, re.MULTILINE | re.IGNORECASE) - if m: - return m.group(1).lower() == "true" - except Exception: - pass - return None - - -def _read_legacy_control_bool(records_root: Path, zotero_key: str, key: str) -> bool | None: - if not records_root or not records_root.exists() or not zotero_key: - return None - for record_path in records_root.rglob(f"{zotero_key}.md"): - try: - text = record_path.read_text(encoding="utf-8") - except Exception: - continue - match = re.search(rf"^{re.escape(key)}:\s*(?:[\"'])?(true|false)(?:[\"'])?\s*$", text, re.MULTILINE | re.IGNORECASE) - if match: - return match.group(1).lower() == "true" - return None +# ── Single-entry builder (shared by full rebuild and incremental refresh) ── +# read_frontmatter_bool / read_frontmatter_optional_bool imported from adapters def _build_entry(item: dict, vault: Path, paths: dict, domain: str, zotero_dir: Path) -> dict: @@ -287,9 +251,7 @@ def _build_entry(item: dict, vault: Path, paths: dict, domain: str, zotero_dir: key = item["key"] collection_meta = collection_fields(item.get("collections", [])) - pdf_attachments = [ - a for a in item.get("attachments", []) if a.get("contentType") == "application/pdf" - ] + pdf_attachments = [a for a in item.get("attachments", []) if a.get("contentType") == "application/pdf"] meta_path = paths["ocr"] / key / "meta.json" meta = read_json(meta_path) if meta_path.exists() else {} if meta: @@ -326,15 +288,15 @@ def _build_entry(item: dict, vault: Path, paths: dict, domain: str, zotero_dir: first_author = authors[0] if authors else "" extra = item.get("extra", "") impact_factor = lookup_impact_factor(item.get("journal", ""), extra, vault) - legacy_records_root = paths.get("library_records") - legacy_do_ocr = _read_legacy_control_bool(legacy_records_root, key, "do_ocr") - legacy_analyze = _read_legacy_control_bool(legacy_records_root, key, "analyze") - note_do_ocr = _read_frontmatter_optional_bool(main_note_path, "do_ocr") + legacy_flags = _legacy_control_flags(paths, key) + legacy_do_ocr = legacy_flags.get("do_ocr") + legacy_analyze = legacy_flags.get("analyze") + note_do_ocr = read_frontmatter_optional_bool(main_note_path, "do_ocr") if note_do_ocr is None: - note_do_ocr = _read_frontmatter_optional_bool(note_path, "do_ocr") - note_analyze = _read_frontmatter_optional_bool(main_note_path, "analyze") + note_do_ocr = read_frontmatter_optional_bool(note_path, "do_ocr") + note_analyze = read_frontmatter_optional_bool(main_note_path, "analyze") if note_analyze is None: - note_analyze = _read_frontmatter_optional_bool(note_path, "analyze") + note_analyze = read_frontmatter_optional_bool(note_path, "analyze") do_ocr_value = note_do_ocr if note_do_ocr is not None else legacy_do_ocr if do_ocr_value is None: @@ -364,9 +326,7 @@ def _build_entry(item: dict, vault: Path, paths: dict, domain: str, zotero_dir: "do_ocr": do_ocr_value, "analyze": analyze_value, "pdf_path": ( - obsidian_wikilink_for_pdf(pdf_attachments[0]["path"], vault, zotero_dir) - if pdf_attachments - else "" + obsidian_wikilink_for_pdf(pdf_attachments[0]["path"], vault, zotero_dir) if pdf_attachments else "" ), "ocr_status": meta.get("ocr_status", "pending"), "ocr_job_id": meta.get("ocr_job_id", ""), @@ -379,7 +339,9 @@ def _build_entry(item: dict, vault: Path, paths: dict, domain: str, zotero_dir: if note_path.exists() and has_deep_reading_content(note_path.read_text(encoding="utf-8")) else "pending" ), - "note_path": str((main_note_path if main_note_path.exists() else note_path).relative_to(vault)).replace("\\", "/"), + "note_path": str((main_note_path if main_note_path.exists() else note_path).relative_to(vault)).replace( + "\\", "/" + ), "deep_reading_md_path": ( str(main_note_path.relative_to(vault)).replace("\\", "/") if main_note_path.exists() and has_deep_reading_content(main_note_path.read_text(encoding="utf-8")) @@ -621,9 +583,7 @@ def summarize_index(vault: Path) -> dict | None: "ai_context_ready": 0, } health_keys = ["pdf_health", "ocr_health", "note_health", "asset_health"] - health_agg: dict[str, dict[str, int]] = { - k: {"healthy": 0, "unhealthy": 0} for k in health_keys - } + health_agg: dict[str, dict[str, int]] = {k: {"healthy": 0, "unhealthy": 0} for k in health_keys} maturity_dist: dict[str, int] = {str(i): 0 for i in range(1, 7)} for entry in items: