#!/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()
|
