#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ tagfeed 菜单维护器 ==================== 维护 00index/Menu-Tagfeed.md 这个轻量 Markdown outline 菜单。 设计边界： - 不修改、不依赖画布索引。 - 复用 tagfeed 页面扫描、去噪、排除目录规则。 - 只补齐缺失标签，不移动、不重排用户已经整理过的 outline 节点。 - 新标签插入「待整理」节点下方的最前面。 """ import importlib.util import os import re import sys from dataclasses import dataclass from pathlib import Path from typing import Optional MENU_REL = "00index/Menu-Tagfeed.md" TAGFEED_REL = "00index/tagfeed-md" TODO_TITLE = "待整理" @dataclass(frozen=True) class TagInfo: tag: str page_rel: str sort_time: float def load_tagfeed_scan_module(): here = Path(__file__).resolve().parent mod_path = here / "sync_0inbox.py" spec = importlib.util.spec_from_file_location("tagfeed_scan", mod_path) if spec is None or spec.loader is None: raise RuntimeError(f"cannot load {mod_path}") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod def resolve_vault(explicit=None) -> Path: if explicit: return Path(explicit).expanduser().resolve() return Path(__file__).resolve().parents[2] def file_sort_time(path: Path) -> float: try: st = path.stat() except OSError: return 0.0 return float(getattr(st, "st_birthtime", st.st_mtime) or st.st_mtime) def is_excluded_rel(sync_mod, rel: str) -> bool: if rel in (".", ""): return False parts = rel.split(os.sep) top = parts[0] if top in sync_mod.EXCLUDE_TOP: return True if any(rel == p or rel.startswith(p + os.sep) for p in sync_mod.EXCLUDE_PATHS): return True return False def collect_used_tags(vault: Path, sync_mod) -> tuple[dict[str, int], dict[str, float]]: counts: dict[str, int] = {} latest: dict[str, float] = {} for root, dirs, files in os.walk(vault): dirs[:] = [d for d in dirs if d not in sync_mod.EXCLUDE_DIRS and not d.startswith(".")] rel = os.path.relpath(root, vault) if is_excluded_rel(sync_mod, rel): dirs[:] = [] continue for fn in files: if fn.startswith("."): continue ext = os.path.splitext(fn)[1].lower() if ext not in sync_mod.SCAN_EXT: continue full = Path(root) / fn tags = sync_mod.collect_tags_in_file(str(full)) if not tags: continue try: mt = full.stat().st_mtime except OSError: mt = 0.0 for tg in tags: if sync_mod.is_noise(tg): continue counts[tg] = counts.get(tg, 0) + 1 latest[tg] = max(latest.get(tg, 0.0), mt) return counts, latest def parse_target_tags(text: str) -> list[str]: m = re.search(r'const\s+targetTag\s*=\s*(?:\[([^\]]*)\]|["\']([^"\']*)["\'])', text) if not m: return [] raw = m.group(1) if m.group(1) is not None else m.group(2) out = [] for piece in raw.split(","): tag = piece.strip().strip('"').strip("'").lstrip("#").strip() if tag: out.append(tag) return out def build_page_map(vault: Path, sync_mod) -> dict[str, str]: """返回 tag -> tagfeed 页面相对路径。""" page_map: dict[str, str] = {} inbox = vault / TAGFEED_REL if inbox.is_dir(): for fp in inbox.glob("*.md"): if fp.name.startswith("."): continue page_map[fp.stem] = fp.relative_to(vault).as_posix() return page_map def wikilink(page_rel: str, tag: str) -> str: return f"[[{page_rel}|{tag}]]" def strip_frontmatter(text: str) -> str: if text.startswith("---\n"): m = re.match(r"^---\n.*?\n---\n?", text, flags=re.S) if m: return text[m.end():] return text def extract_existing_tags(lines: list[str]) -> set[str]: existing = set() link_re = re.compile(r"\[\[([^\]|]+)(?:\|([^\]]+))?\]\]") for line in lines: for m in link_re.finditer(line): path, alias = m.group(1), m.group(2) tag = alias if alias is not None else Path(path).stem tag = tag.strip().lstrip("#") if tag: existing.add(tag) return existing def rewrite_existing_links(lines: list[str], page_map: dict[str, str], sync_mod) -> list[str]: link_re = re.compile(r"\[\[([^\]|]+)(?:\|([^\]]+))?\]\]") out = [] for line in lines: def repl(m): path, alias = m.group(1), m.group(2) stem = Path(path).stem page_rel = page_map.get(stem) or page_map.get(sync_mod.tag_to_relpath(stem)) if not page_rel: return m.group(0) label = alias if alias is not None else Path(path).stem return f"[[{page_rel}|{label}]]" out.append(link_re.sub(repl, line)) return out def normalize_existing_outline(text: str) -> list[str]: body = strip_frontmatter(text) lines = [ln.rstrip() for ln in body.splitlines()] # 旧文件只有一个 "-"，视为尚未初始化。 meaningful = [ln for ln in lines if ln.strip()] if meaningful == ["-"]: return [] # Menu 文件约定为纯 outline；保留 bullet 行，丢弃空行和旧 frontmatter。 return [ln for ln in lines if ln.strip().startswith(("-", "*", "+"))] def find_todo_line(lines: list[str]) -> Optional[int]: pat = re.compile(rf"^\s*[-*+]\s+{re.escape(TODO_TITLE)}\s*$") for i, line in enumerate(lines): if pat.match(line): return i return None def merge_menu(existing_text: str, missing: list[TagInfo], page_map: dict[str, str], sync_mod) -> tuple[str, bool, int]: lines = normalize_existing_outline(existing_text) lines = rewrite_existing_links(lines, page_map, sync_mod) todo_idx = find_todo_line(lines) if todo_idx is None: lines.insert(0, f"- {TODO_TITLE}") todo_idx = 0 indent = len(lines[todo_idx]) - len(lines[todo_idx].lstrip(" ")) child_prefix = " " * (indent + 2) + "- " new_lines = [child_prefix + wikilink(t.page_rel, t.tag) for t in missing] if new_lines: lines[todo_idx + 1:todo_idx + 1] = new_lines content = "\n".join(lines).rstrip() + "\n" old_outline = "\n".join(normalize_existing_outline(existing_text)).rstrip() + "\n" changed = content != old_outline return content, changed, len(new_lines) def sync_menu(vault: Path) -> tuple[str, int, int, Path]: sync_mod = load_tagfeed_scan_module() counts, latest_seen = collect_used_tags(vault, sync_mod) page_map = build_page_map(vault, sync_mod) menu_path = vault / MENU_REL try: existing_text = menu_path.read_text(encoding="utf-8") except OSError: existing_text = "" existing_tags = extract_existing_tags(normalize_existing_outline(existing_text)) missing: list[TagInfo] = [] for tag in counts: if tag in existing_tags: continue rel_tag = sync_mod.tag_to_relpath(tag) page_rel = page_map.get(tag) or page_map.get(rel_tag) or f"{TAGFEED_REL}/{rel_tag}.md" page_time = file_sort_time(vault / page_rel) sort_time = page_time or latest_seen.get(tag, 0.0) missing.append(TagInfo(tag=tag, page_rel=page_rel, sort_time=sort_time)) missing.sort(key=lambda item: (-item.sort_time, item.tag.lower())) content, changed, inserted = merge_menu(existing_text, missing, page_map, sync_mod) if changed: menu_path.parent.mkdir(parents=True, exist_ok=True) menu_path.write_text(content, encoding="utf-8") status = "written" else: status = "unchanged" return status, len(counts), inserted, menu_path def main() -> None: vault = resolve_vault(sys.argv[1] if len(sys.argv) > 1 else None) status, total, inserted, path = sync_menu(vault) print(f"tagfeed-menu: {status} total_tags={total} inserted={inserted} path={path}") if __name__ == "__main__": main()