obsidian-plugin/lllin000_PaperForge
1
0
Fork 0
mirror of https://github.com/lllin000/PaperForge.git synced 2026-07-22 06:50:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor: eliminate frontmatter regex duplication — route everything through adapter

Browse source 
- adapters/obsidian_frontmatter: +read_frontmatter_bool(note_path, key) public API
- asset_index: delete private _read_frontmatter_bool/_optional copies, import from adapter
- sync_service cleanup methods: raw regex → read_frontmatter_dict()
- ld_deep: fix private worker import → adapter
- config.py paperforge_paths: +config and index keys (domain-collections, formal-library)
- sync_service.run(): pipeline_paths → self.resolve_paths()
This commit is contained in:
Research Assistant 2026-05-09 22:30:59 +08:00
parent 05629bce82
commit a30ffd3835
5 changed files with 71 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
paperforge/adapters/obsidian_frontmatter.py
View file

@ -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

8
paperforge/config.py
View file

@ -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 {})

18
paperforge/services/sync_service.py
View file

@ -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"]}

4
paperforge/skills/literature-qa/scripts/ld_deep.py
View file

@ -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."

80
paperforge/worker/asset_index.py
View file

@ -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: