#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ sync_excalidraw_tagfeed.py ========================== 维护方案3：00index/00index.md。 每个标签生成一个可点击矩形，矩形 link 指向对应 tagfeed 页面。 受管边界： - 只维护名为 Frame0-all 的 frame 内部。 - frame 内保证当前标签卡片不重复、不遗漏。 - frame 外所有元素完全不判断、不删除、不补齐，供用户自由实验其它整理方案。 """ import hashlib import json import os import re import sys import urllib.parse from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Optional LZ_BASE64 = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=" OUT_REL = "00index/00index.md" INBOX_REL = "00index/tagfeed-md" PLUGIN_DATA_REL = ".obsidian/plugins/tagfeed-click-opener/data.json" ALIAS_MIRROR_REL = "00index/tagfeed-md/_Tagfeed别名表.md" STATICS_REL = "00index/tagfeed-md/_Tagfeed_statics.md" MANAGED_FRAME_NAME = "Frame0-all" SCAN_EXT = {".md", ".markdown", ".excalidraw", ".canvas"} EXCLUDE_DIRS = {".obsidian", ".trash", ".git", ".agents", ".claude", ".copilot", ".opencode", ".smart-env", ".workbuddy"} EXCLUDE_TOP = {"X2.Archived", "00index", "X1.Knomo", "X.Attachment", "P3.bobo"} EXCLUDE_PATHS = {"02DS/02copilot", "02DS/01dril-book"} TAGFEED_PAGE_EXCLUDED_PATHS = [ "X2.Archived", "00index", "X1.Knomo", "X.Attachment", "P3.bobo", "02DS/02copilot", "02DS/01dril-book", ".agents", ".claude", ".copilot", ".opencode", ".smart-env", ".workbuddy", ".trash", ] TAG_RE = re.compile(r"(? Path: if explicit: return Path(explicit).expanduser().resolve() return Path(__file__).resolve().parents[2] def stable_id(prefix: str, value: str) -> str: return hashlib.md5((prefix + ":" + value).encode("utf-8")).hexdigest()[:20] def stable_seed(value: str) -> int: return int(hashlib.md5(value.encode("utf-8")).hexdigest()[:8], 16) def index_name(i: int) -> str: # Excalidraw accepts string indexes; this keeps deterministic order for generated elements. return f"a{i:05d}" def obsidian_uri(vault: Path, page_rel: str) -> str: return ( "obsidian://open?vault=" + urllib.parse.quote(vault.name) + "&file=" + urllib.parse.quote(page_rel) ) def today_yy_mmdd() -> str: return datetime.now().strftime("%y.%m%d") def tag_to_relpath(tag: str) -> str: return tag.replace("/", "_") def tag_key(tag: str) -> str: return tag.casefold() def clean_alias_tag(value: object) -> str: tag = str(value or "").strip().lstrip("#").strip() if not tag: return "" if not re.fullmatch(r"[A-Za-z0-9_\u4e00-\u9fff][\w\u4e00-\u9fff/\-]*", tag): return "" return tag EXCALIDRAW_BLOCK_REF_RE = re.compile(r"\s+\^[A-Za-z0-9_-]+\s*$") def extract_frontmatter_prefix(raw: str) -> str: match = re.match(r"^---\s*\n.*?\n(?:---|\.\.\.)\s*\n", raw, flags=re.S) return match.group(0).rstrip() if match else "" def extract_excalidraw_text_elements(raw: str) -> str: """Return only human-readable Excalidraw Text Elements, excluding JSON/assets.""" lines = raw.splitlines() start = None for idx, line in enumerate(lines): if re.match(r"^\s*##\s+Text Elements\s*$", line, flags=re.I): start = idx + 1 break if start is None: return "" chunks: list[str] = [] current: list[str] = [] def flush() -> None: nonlocal current text = " ".join(part.strip() for part in current if part.strip()) text = re.sub(r"[ \t]{2,}", " ", text).strip() if text: chunks.append(text) current = [] for line in lines[start:]: stripped = line.strip() if stripped == "%%" or re.match(r"^\s*##\s+(?:Embedded Files|Drawing)\s*$", line, flags=re.I): break if not stripped: if current: current.append("") continue had_ref = EXCALIDRAW_BLOCK_REF_RE.search(line) is not None cleaned = EXCALIDRAW_BLOCK_REF_RE.sub("", line).strip() if cleaned: current.append(cleaned) if had_ref: flush() flush() return "\n\n".join(chunks) def readable_excalidraw_markdown(raw: str) -> str: frontmatter = extract_frontmatter_prefix(raw) text_elements = extract_excalidraw_text_elements(raw) parts = [part for part in (frontmatter, text_elements) if part] return "\n\n".join(parts) + ("\n" if parts else "") def normalize_plugin_alias_groups(raw_groups: object) -> dict[str, tuple[str, ...]]: if not isinstance(raw_groups, list): return {} canonical_by_key: dict[str, str] = {} owners: dict[str, str] = {} aliases_by_key: dict[str, list[str]] = {} for raw_group in raw_groups: if not isinstance(raw_group, dict): continue canonical = clean_alias_tag(raw_group.get("canonical")) if not canonical: continue ckey = tag_key(canonical) if ckey not in canonical_by_key: canonical_by_key[ckey] = canonical aliases_by_key[ckey] = [] owners.setdefault(ckey, ckey) for raw_group in raw_groups: if not isinstance(raw_group, dict): continue canonical = clean_alias_tag(raw_group.get("canonical")) ckey = tag_key(canonical) if not canonical or ckey not in aliases_by_key: continue aliases = raw_group.get("aliases") if not isinstance(aliases, list): continue for raw_alias in aliases: alias = clean_alias_tag(raw_alias) akey = tag_key(alias) if not alias or not akey or akey == ckey: continue owner = owners.get(akey) if owner and owner != ckey: continue owners[akey] = ckey if all(tag_key(item) != akey for item in aliases_by_key[ckey]): aliases_by_key[ckey].append(alias) return { canonical_by_key[ckey]: tuple(aliases) for ckey, aliases in sorted( aliases_by_key.items(), key=lambda item: canonical_by_key[item[0]].casefold(), ) } def load_plugin_alias_table(vault: Path) -> tuple[dict[str, tuple[str, ...]], Optional[str]]: path = vault / PLUGIN_DATA_REL if not path.exists(): return {}, None try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {}, None if not isinstance(data, dict): return {}, None updated_at = data.get("updatedAt") if not isinstance(updated_at, str): updated_at = None return normalize_plugin_alias_groups(data.get("aliasGroups")), updated_at def is_tagfeed_system_page(fp: Path) -> bool: return fp.name.startswith("_") def escape_table_cell(value: str) -> str: return value.replace("|", "\\|").replace("\n", " ") def build_alias_mirror(alias_groups: dict[str, tuple[str, ...]], updated_at: Optional[str]) -> str: updated = json.dumps(updated_at, ensure_ascii=False) if updated_at else "null" lines = [ "---", "tagfeed-system: alias-mirror", "generated-from: .obsidian/plugins/tagfeed-click-opener/data.json", f"updated: {updated}", "---", "", "# Tagfeed 别名表", "", "> 本页由 Tagfeeds / ob01 同步脚本自动生成，只用于知识库阅读和交接。请不要直接编辑本页；修改别名请去 Obsidian 插件设置页。", "", "主事实源：`.obsidian/plugins/tagfeed-click-opener/data.json`", "", ] if not alias_groups: lines.append("当前没有配置标签别名。") lines.append("") return "\n".join(lines) lines.extend(["| 主标签 | 别名 | 聚合页 |", "| --- | --- | --- |"]) for canonical, aliases in alias_groups.items(): alias_text = ", ".join(f"#{alias}" for alias in aliases) if aliases else "无" page = f"[[{INBOX_REL}/{tag_to_relpath(canonical)}]]" lines.append( f"| #{escape_table_cell(canonical)} | {escape_table_cell(alias_text)} | {page} |" ) lines.append("") return "\n".join(lines) def write_alias_mirror( vault: Path, alias_groups: dict[str, tuple[str, ...]], updated_at: Optional[str], ) -> bool: path = vault / ALIAS_MIRROR_REL path.parent.mkdir(parents=True, exist_ok=True) content = build_alias_mirror(alias_groups, updated_at) try: old = path.read_text(encoding="utf-8") if path.exists() else "" except OSError: old = "" if old == content: return False path.write_text(content, encoding="utf-8") return True def format_sync_time(ts: datetime) -> str: return ts.strftime("%Y-%m-%d %H:%M:%S") def count_missing_pages(vault: Path, tags: list[TagInfo]) -> int: missing = 0 for info in tags: page = vault / info.page_rel if not page.exists(): missing += 1 return missing def count_duplicate_cards(drawing) -> int: if not drawing: return 0 frame = find_managed_frame(drawing) if frame is None: return 0 managed = managed_tag_elements(drawing, frame) duplicate_cards = 0 for elems in managed.values(): rects = [ el for el in elems if el.get("type") == "rectangle" and element_role(el) in (None, "rect") ] duplicate_cards += max(0, len(rects) - 1) return duplicate_cards def build_statics_page( effective_tag_count: int, missing_page_count: int, duplicate_card_count: int, synced_at: datetime, tags: list[TagInfo], ) -> str: synced = format_sync_time(synced_at) lines = [ "---", "tagfeed-system: statics", "generated-from: X2.Archived/scripts/sync_excalidraw_tagfeed.py", f"updated: {json.dumps(synced, ensure_ascii=False)}", "---", "", "> [!info] Tagfeed 统计", f"> 有效标签数：{effective_tag_count}", f"> 缺页数：{missing_page_count}", f"> 重复卡片数：{duplicate_card_count}", f"> 最近同步时间：{synced}", "", "# Tagfeed Statics", "", "本页由 ob01tagfeed 同步脚本自动生成和更新，请不要手动编辑。", "", "## 全部有效标签", "", "| 序号 | 标签 | 聚合页 |", "| ---: | --- | --- |", ] for idx, info in enumerate(tags, 1): label = "`#" + escape_table_cell(info.tag) + "`" page = f"[[{info.page_rel.removesuffix('.md')}]]" lines.append(f"| {idx} | {label} | {page} |") lines.extend([ "", "统计口径：", "", "- 有效标签数：当前扫描规则下进入 tagfeed 的主标签数量。", "- 缺页数：有效主标签中尚未存在 tagfeed 聚合页的数量；已归入主标签页的别名不算缺页。", "- 重复卡片数：`Frame0-all` 内同一主标签多出来的矩形卡片数量；frame 外实验内容不统计。", "- 最近同步时间：本页被同步脚本更新时的本机时间。", "", ]) return "\n".join(lines) def write_statics_page( vault: Path, effective_tag_count: int, missing_page_count: int, duplicate_card_count: int, synced_at: datetime, tags: list[TagInfo], ) -> bool: path = vault / STATICS_REL path.parent.mkdir(parents=True, exist_ok=True) content = build_statics_page( effective_tag_count, missing_page_count, duplicate_card_count, synced_at, tags, ) try: old = path.read_text(encoding="utf-8") if path.exists() else "" except OSError: old = "" if old == content: return False path.write_text(content, encoding="utf-8") return True def is_noise(tag: str) -> bool: low = tag.lower() if low in NOISE_EXACT: return True if low.startswith(NOISE_PREFIX): return True if low.startswith("part0") or low.startswith("index_") or "threadpoolf" in low: return True if re.fullmatch(r"\d+([._]\d+)?", tag): return True if re.fullmatch(r"[0-9a-fA-F]{3}([0-9a-fA-F]{3})?", tag): return True return False def clean_fm_tag_token(value: object) -> str: tag = str(value).strip().strip("[]").strip().strip('"').strip("'").lstrip("#").strip() if not tag: return "" if not re.fullmatch(r"[A-Za-z0-9_\u4e00-\u9fff][\w\u4e00-\u9fff/\-]*", tag): return "" return tag def split_fm_value(v) -> list[str]: if v is None: return [] if isinstance(v, list): out = [] for item in v: out.extend(split_fm_value(item)) return out if isinstance(v, dict): return split_fm_value(" ".join(str(x) for x in v.values())) s = str(v).strip().strip('"').strip("'") if not s: return [] out = [] for part in re.split(r"[，,；;]", s): for tok in part.split(): tok = clean_fm_tag_token(tok) if tok: out.append(tok) return out def parse_frontmatter_tags(fm_text: str) -> list[str]: tags: list[str] = [] try: import yaml # type: ignore fm = yaml.safe_load(fm_text) or {} except Exception: fm = {} if isinstance(fm, dict) and "tags" in fm: tags.extend(split_fm_value(fm["tags"])) if tags: return tags lines = fm_text.splitlines() for i, line in enumerate(lines): match = re.match(r"^tags:\s*(.*)$", line) if not match: continue inline = match.group(1).strip() if inline: if inline.startswith("[") and inline.endswith("]"): inline = inline[1:-1] return split_fm_value(inline) out: list[str] = [] for child in lines[i + 1:]: if child.strip() == "": continue if not child.startswith((" ", "\t")): break item = re.match(r"^\s*-\s*(.+?)\s*$", child) if item: out.extend(split_fm_value(item.group(1))) return out return [] def collect_tags_in_file(path: str) -> set[str]: tags = set() try: raw = Path(path).read_text(encoding="utf-8", errors="ignore") except OSError: return tags name = Path(path).name if name.endswith(".excalidraw.md"): raw = readable_excalidraw_markdown(raw) ext = Path(path).suffix.lower() if ext in (".md", ".markdown"): m = re.match(r"^---\s*\n(.*?)\n---\s*\n", raw, re.DOTALL) body = raw if m: fm_text = m.group(1) body = raw[m.end():] for tok in parse_frontmatter_tags(fm_text): if tok: tags.add(tok) for line in body.splitlines(): for mm in TAG_RE.finditer(line): tags.add(mm.group(1)) else: for mm in TAG_RE.finditer(raw): tags.add(mm.group(1)) return tags TARGET_TAG_RE = re.compile( r"const\s+targetTag\s*=\s*(\[[^\]]*\]|\"[^\"]*\"|'[^']*')", re.M, ) def parse_target_tags(text: str) -> list[str]: m = TARGET_TAG_RE.search(text) if not m: return [] raw = m.group(1).strip() if raw.startswith("["): items = re.findall(r'"([^"]+)"|\'([^\']+)\'', raw) vals = [a or b for a, b in items] return [v.strip() for v in vals if v and v.strip()] return [raw.strip('"').strip("'").strip()] def rewrite_target_tags(text: str, tags: list[str]) -> str: if not tags: return text if len(tags) == 1: replacement = f'const targetTag = "{tags[0]}"' else: joined = ", ".join(f'"{t}"' for t in tags) replacement = f"const targetTag = [{joined}]" return TARGET_TAG_RE.sub(replacement, text, count=1) def ensure_alias_note(text: str, aliases: list[str]) -> str: trailing_newline = text.endswith("\n") lines = [line for line in text.splitlines() if not line.startswith("> 别名：")] first_nonblank = next((idx for idx, line in enumerate(lines) if line.strip()), None) if first_nonblank and lines[first_nonblank] == "---": lines = lines[first_nonblank:] if not aliases: return "\n".join(lines) + ("\n" if trailing_newline else "") alias_line = "> 别名：" + "、".join(f"#{a}" for a in aliases) out = [] inserted = False for line in lines: out.append(line) if not inserted and line.startswith("> 本页由 DataviewJS"): out.append(alias_line) inserted = True if not inserted: insert_at = 0 if len(out) >= 2 and out[0] == "---": for idx in range(1, len(out)): if out[idx] == "---": insert_at = idx + 1 break for idx, line in enumerate(out[insert_at:], start=insert_at): if line.startswith("# "): insert_at = idx + 1 break out.insert(insert_at, alias_line) return "\n".join(out) + ("\n" if trailing_newline else "") def load_tagfeed_page_groups(vault: Path) -> dict[str, TagfeedPageGroup]: base = vault / INBOX_REL groups: dict[str, TagfeedPageGroup] = {} if not base.is_dir(): return groups for fp in sorted(base.glob("*.md")): if fp.name.startswith(".") or is_tagfeed_system_page(fp): continue try: text = fp.read_text(encoding="utf-8", errors="ignore") except OSError: continue if "tagfeed-system: alias-mirror" in text: continue tags = parse_target_tags(text) if not tags: continue canonical = tags[0] aliases = tuple(dict.fromkeys(tags[1:])) groups[canonical] = TagfeedPageGroup(canonical=canonical, aliases=aliases, path=fp) return groups def normalize_tagfeed_pages( vault: Path, plugin_alias_groups: Optional[dict[str, tuple[str, ...]]] = None, ) -> tuple[dict[str, str], int, int, int]: base = vault / INBOX_REL if not base.is_dir(): return {}, 0, 0, 0 groups = load_tagfeed_page_groups(vault) plugin_alias_groups = plugin_alias_groups or {} groups_by_key = {tag_key(canonical): group for canonical, group in groups.items()} for canonical, aliases in plugin_alias_groups.items(): old_group = groups_by_key.get(tag_key(canonical)) if old_group: groups.pop(old_group.canonical, None) path = old_group.path else: path = base / f"{tag_to_relpath(canonical)}.md" groups[canonical] = TagfeedPageGroup(canonical=canonical, aliases=aliases, path=path) alias_to_canonical: dict[str, str] = {} renamed = 0 deleted = 0 rewritten = 0 for canonical, group in groups.items(): alias_to_canonical[canonical] = canonical alias_to_canonical[tag_to_relpath(canonical)] = canonical for alias in group.aliases: alias_to_canonical[alias] = canonical alias_to_canonical[tag_to_relpath(alias)] = canonical for canonical, group in groups.items(): canonical_path = base / f"{tag_to_relpath(canonical)}.md" source = group.path if source != canonical_path: if canonical_path.exists(): try: source.unlink() deleted += 1 except OSError: pass else: try: source.rename(canonical_path) renamed += 1 source = canonical_path except OSError: pass if not canonical_path.exists(): continue try: text = canonical_path.read_text(encoding="utf-8") except OSError: continue new_text = rewrite_target_tags(text, [canonical, *group.aliases]) new_text = ensure_alias_note(new_text, list(group.aliases)) if new_text != text: try: canonical_path.write_text(new_text, encoding="utf-8") rewritten += 1 except OSError: pass # remove any stale alias pages that still exist on disk for fp in sorted(base.glob("*.md")): if fp.name.startswith(".") or is_tagfeed_system_page(fp): continue stem = fp.stem canonical = alias_to_canonical.get(stem) if canonical and canonical != stem: try: fp.unlink() deleted += 1 except OSError: pass return alias_to_canonical, renamed, deleted, rewritten def js_string_array(items: list[str]) -> str: return "[" + ", ".join(json.dumps(item, ensure_ascii=False) for item in items) + "]" def rewrite_excluded_paths(text: str) -> str: replacement = f"const excludedPaths = {js_string_array(TAGFEED_PAGE_EXCLUDED_PATHS)};" if re.search(r"const\s+excludedPaths\s*=", text): return re.sub(r"const\s+excludedPaths\s*=\s*\[[^\]]*\]\s*;?", replacement, text, count=1) return re.sub( r"(const\s+targetTag\s*=\s*(?:\[[^\]]*\]|\"[^\"]*\"|'[^']*')\s*;?)", "\\1\n" + replacement, text, count=1, ) def upgrade_live_tagfeed_summary_rules(vault: Path) -> tuple[int, int]: """Keep all live DataviewJS tagfeed pages on the current extraction rules.""" base = vault / INBOX_REL if not base.is_dir(): return 0, 0 helper_marker = " for (let i = fm.bodyStart; i < lines.length; i++) {" helper = """ // A tag on an indented continuation line belongs to its nearest list item. function collectParentListBlock(i) { let parentIdx = null; let parentIndent = 0; for (let k = i - 1; k >= fm.bodyStart; k--) { const prev = lines[k]; if (!prev.trim() || HEAD_RE.test(prev) || FENCE_RE.test(prev)) break; if (prev.match(LIST_RE)) { parentIdx = k; parentIndent = indentOf(prev); break; } } if (parentIdx === null || indentOf(lines[i]) <= parentIndent) return null; const block = []; for (let j = parentIdx; j <= i; j++) { const current = lines[j]; if (!current.trim()) continue; const cm = current.match(LIST_RE); if (cm) { const currentIndent = indentOf(current); if (j !== parentIdx && currentIndent <= parentIndent) return null; const text = stripTag(cleanMatch(cm[3] || "")) || (cm[3] || "").trim(); block.push([j === parentIdx ? 0 : Math.max(1, Math.floor((currentIndent - parentIndent) / 2)), text]); } else { const text = stripTag(cleanMatch(current)); if (text) block.push([Math.max(1, Math.floor((indentOf(current) - parentIndent) / 2)), text]); } } return block.length ? { keyIdx: parentIdx, block } : null; } """ summary_marker = " // 普通段落带标签：吸收到空行为止\n" summary_fix = """ // 命中在列表项的缩进续行：摘要应回到所属列表项，而不是只截取续行尾部。 const parentBlock = collectParentListBlock(i); if (parentBlock) { addBlock(parentBlock.keyIdx, parentBlock.block); continue; } """ excalidraw_marker = "function sanitizeHumanLine(ln) {" excalidraw_helper = """const EXCALIDRAW_BLOCK_REF_RE = /\\s+\\^[A-Za-z0-9_-]+\\s*$/; function extractFrontmatterPrefix(text) { const m = String(text || "").match(/^---\\s*\\n[\\s\\S]*?\\n(?:---|\\.\\.\\.)\\s*\\n/); return m ? m[0].trimEnd() : ""; } function extractExcalidrawTextElements(text) { const lines = String(text || "").split("\\n"); const start = lines.findIndex(ln => /^\\s*##\\s+Text Elements\\s*$/i.test(ln)); if (start < 0) return ""; const chunks = []; let current = []; const flush = () => { const joined = current.map(x => x.trim()).filter(Boolean).join(" ").replace(/[ \\t]{2,}/g, " ").trim(); if (joined) chunks.push(joined); current = []; }; for (let i = start + 1; i < lines.length; i++) { const raw = lines[i]; const stripped = raw.trim(); if (stripped === "%%" || /^\\s*##\\s+(?:Embedded Files|Drawing)\\s*$/i.test(raw)) break; if (!stripped) { if (current.length) current.push(""); continue; } const hadRef = EXCALIDRAW_BLOCK_REF_RE.test(raw); const cleaned = raw.replace(EXCALIDRAW_BLOCK_REF_RE, "").trim(); if (cleaned) current.push(cleaned); if (hadRef) flush(); } flush(); return chunks.join("\\n\\n"); } function readableExcalidrawMarkdown(text) { const frontmatter = extractFrontmatterPrefix(text); const body = extractExcalidrawTextElements(text); return [frontmatter, body].filter(Boolean).join("\\n\\n"); } """ skip_excalidraw_old = ' if (f.path === currentPath || f.path.endsWith(".excalidraw.md")) continue;\n' skip_excalidraw_new = ' if (f.path === currentPath) continue;\n' read_text_line = ' try { text = await app.vault.cachedRead(f); } catch (e) { continue; }\n' read_text_with_excalidraw = ( ' try { text = await app.vault.cachedRead(f); } catch (e) { continue; }\n' ' if (f.path.endsWith(".excalidraw.md")) text = readableExcalidrawMarkdown(text);\n' ) excalidraw_transform_line = ' if (f.path.endsWith(".excalidraw.md")) text = readableExcalidrawMarkdown(text);\n' upgraded = skipped = 0 for fp in sorted(base.glob("*.md")): try: text = fp.read_text(encoding="utf-8") except OSError: continue if "tagfeed-live" not in text or "function scanText" not in text: continue updated = text if "function collectParentListBlock" not in updated: if helper_marker in updated and summary_marker in updated: updated = updated.replace(helper_marker, helper + helper_marker, 1) updated = updated.replace(summary_marker, summary_fix + summary_marker, 1) if "function readableExcalidrawMarkdown" not in updated and excalidraw_marker in updated: updated = updated.replace(excalidraw_marker, excalidraw_helper + excalidraw_marker, 1) if skip_excalidraw_old in updated: updated = updated.replace(skip_excalidraw_old, skip_excalidraw_new, 1) if ( excalidraw_transform_line not in updated and read_text_line in updated and "function readableExcalidrawMarkdown" in updated ): updated = updated.replace(read_text_line, read_text_with_excalidraw, 1) if updated != text: fp.write_text(updated, encoding="utf-8") upgraded += 1 else: skipped += 1 return upgraded, skipped def extract_tagfeed_template(vault: Path) -> str: base = vault / INBOX_REL seed = None if base.is_dir(): for fp in sorted(base.glob("*.md")): if fp.name.startswith(".") or is_tagfeed_system_page(fp): continue try: text = fp.read_text(encoding="utf-8", errors="ignore") except OSError: continue if "tagfeed-live" in text and "targetTag" in text: seed = text break frontmatter = ( "---\n" "tags:\n" " - tagfeed-live\n" "cssclasses:\n" " - tagfeed\n" "---\n\n" ) fallback_dv = ( "const targetTag = \"__TARGET_TAG__\";\n" f"const excludedPaths = {js_string_array(TAGFEED_PAGE_EXCLUDED_PATHS)};\n" "const esc = s => String(s).replace(/[.*+?^${}()|[\\]\\\\]/g, \"\\\\$&\");\n" "const tagRe = new RegExp(\"(? page.file.path === x || page.file.path.startsWith(x + \"/\"))) continue;\n" " const text = await dv.io.load(page.file.path);\n" " if (tagRe.test(text)) rows.push(page.file.link);\n" "}\n" "dv.list(rows);" ) dv = fallback_dv if seed: dm = re.search(r"```dataviewjs\n(.*?)\n```", seed, re.DOTALL) if dm: dv = dm.group(1) dv = re.sub( r"const\s+targetTag\s*=\s*(?:\[[^\]]*\]|\"[^\"]*\"|'[^']*')\s*;?", 'const targetTag = "__TARGET_TAG__";', dv, count=1, ) dv = rewrite_excluded_paths(dv) title = "# 🏷 #__TARGET_TAG__（实时聚合页）\n\n" note = "> 本页由 DataviewJS 实时计算，笔记有变动时自动刷新，无需任何后台服务。\n\n" return frontmatter + title + note + "```dataviewjs\n" + dv.strip() + "\n```\n" def create_missing_tagfeed_pages( vault: Path, tags: list[TagInfo], plugin_alias_groups: Optional[dict[str, tuple[str, ...]]] = None, ) -> tuple[int, int]: base = vault / INBOX_REL base.mkdir(parents=True, exist_ok=True) template = extract_tagfeed_template(vault) plugin_alias_groups = plugin_alias_groups or {} created = skipped = 0 seen_pages: set[str] = set() for info in tags: page_rel = Path(info.page_rel) if len(page_rel.parts) < 2 or page_rel.parts[0] != "00index" or page_rel.parts[1] != "tagfeed-md": skipped += 1 continue if page_rel.suffix.lower() != ".md": skipped += 1 continue rel_key = page_rel.as_posix() if rel_key in seen_pages: skipped += 1 continue seen_pages.add(rel_key) fp = vault / page_rel if fp.exists(): skipped += 1 continue fp.parent.mkdir(parents=True, exist_ok=True) content = template.replace("__TARGET_TAG__", info.tag) aliases = list(plugin_alias_groups.get(info.tag, ())) if aliases: content = rewrite_target_tags(content, [info.tag, *aliases]) content = ensure_alias_note(content, aliases) fp.write_text(content, encoding="utf-8") created += 1 return created, skipped def lz_base_value(ch: str) -> int: return LZ_BASE64.index(ch) def lz_decompress_from_base64(data: str) -> Optional[str]: if data is None: return "" cleaned = "".join(ch for ch in data if ch not in "\n\r") if cleaned == "": return None return lz_decompress(len(cleaned), 32, lambda idx: lz_base_value(cleaned[idx])) def lz_read_bits(data: dict, num_bits: int, reset_value: int, get_next_value) -> int: bits = 0 maxpower = 1 << num_bits power = 1 while power != maxpower: resb = data["val"] & data["position"] data["position"] >>= 1 if data["position"] == 0: data["position"] = reset_value data["val"] = get_next_value(data["index"]) data["index"] += 1 if resb > 0: bits |= power power <<= 1 return bits def lz_decompress(length: int, reset_value: int, get_next_value) -> Optional[str]: dictionary: dict[int, str] = {} enlarge_in = 4 dict_size = 4 num_bits = 3 result: list[str] = [] data = {"val": get_next_value(0), "position": reset_value, "index": 1} next_value = lz_read_bits(data, 2, reset_value, get_next_value) if next_value == 0: c = chr(lz_read_bits(data, 8, reset_value, get_next_value)) elif next_value == 1: c = chr(lz_read_bits(data, 16, reset_value, get_next_value)) elif next_value == 2: return "" else: return None dictionary[0] = "" dictionary[1] = "" dictionary[2] = "" dictionary[3] = c w = c result.append(c) while True: if data["index"] > length: return "" c_num = lz_read_bits(data, num_bits, reset_value, get_next_value) if c_num == 0: dictionary[dict_size] = chr(lz_read_bits(data, 8, reset_value, get_next_value)) dict_size += 1 c_num = dict_size - 1 enlarge_in -= 1 elif c_num == 1: dictionary[dict_size] = chr(lz_read_bits(data, 16, reset_value, get_next_value)) dict_size += 1 c_num = dict_size - 1 enlarge_in -= 1 elif c_num == 2: return "".join(result) if enlarge_in == 0: enlarge_in = 1 << num_bits num_bits += 1 if c_num in dictionary: entry = dictionary[c_num] elif c_num == dict_size: entry = w + w[0] else: return None result.append(entry) dictionary[dict_size] = w + entry[0] dict_size += 1 enlarge_in -= 1 w = entry if enlarge_in == 0: enlarge_in = 1 << num_bits num_bits += 1 def collect_tag_infos(vault: Path, alias_to_canonical: dict[str, str]) -> list[TagInfo]: counts: dict[str, int] = {} latest_seen: dict[str, float] = {} variants: dict[str, list[str]] = {} for root, dirs, files in os.walk(vault): dirs[:] = [d for d in dirs if d not in EXCLUDE_DIRS and not d.startswith(".")] rel = os.path.relpath(root, vault) top = rel.split(os.sep)[0] if top in EXCLUDE_TOP: dirs[:] = [] continue if any(rel == p or rel.startswith(p + os.sep) for p in EXCLUDE_PATHS): dirs[:] = [] continue for fn in files: if fn.startswith("."): continue ext = os.path.splitext(fn)[1].lower() if ext not in SCAN_EXT: continue full = Path(root) / fn tags = collect_tags_in_file(str(full)) if not tags: continue try: mt = full.stat().st_mtime except OSError: mt = 0.0 for tg in sorted(tags, key=lambda x: (tag_key(x), x)): if is_noise(tg): continue key = tag_key(tg) counts[key] = counts.get(key, 0) + 1 latest_seen[key] = max(latest_seen.get(key, 0.0), mt) bucket = variants.setdefault(key, []) if tg not in bucket: bucket.append(tg) page_map: dict[str, str] = {} page_map_lower: dict[str, str] = {} inbox = vault / INBOX_REL if inbox.is_dir(): for fp in inbox.glob("*.md"): if fp.name.startswith(".") or is_tagfeed_system_page(fp): continue rel = fp.relative_to(vault).as_posix() page_map[fp.stem] = rel page_map_lower.setdefault(tag_key(fp.stem), rel) alias_to_canonical_lower = {tag_key(alias): canonical for alias, canonical in alias_to_canonical.items()} tags = [] for key, tag_variants in variants.items(): canonical = None for variant in tag_variants: canonical = alias_to_canonical.get(variant) if canonical: break if canonical is None: canonical = alias_to_canonical_lower.get(key) if canonical is None: for variant in tag_variants: rel_variant = tag_to_relpath(variant) page_rel = ( page_map.get(variant) or page_map.get(rel_variant) or page_map_lower.get(tag_key(variant)) or page_map_lower.get(tag_key(rel_variant)) ) if page_rel: canonical = Path(page_rel).stem break if canonical is None: canonical = tag_variants[0] canonical_rel = tag_to_relpath(canonical) page_rel = ( page_map.get(canonical) or page_map.get(canonical_rel) or page_map_lower.get(tag_key(canonical)) or page_map_lower.get(tag_key(canonical_rel)) or f"{INBOX_REL}/{canonical_rel}.md" ) page_time = 0.0 tags.append(TagInfo(tag=canonical, page_rel=page_rel, sort_time=page_time or latest_seen.get(key, 0.0))) tags.sort(key=lambda item: (-item.sort_time, item.tag.lower())) return tags def extract_drawing_json(text: str): m = re.search(r"## Drawing\s*```json\s*(.*?)\n```", text, flags=re.S) if m: try: return json.loads(m.group(1)) except json.JSONDecodeError: return None m = re.search(r"## Drawing\s*```compressed-json\s*(.*?)\n```", text, flags=re.S) if not m: return None raw = lz_decompress_from_base64(m.group(1)) if not raw: return None try: return json.loads(raw) except json.JSONDecodeError: return None def load_existing(path: Path): if not path.exists(): return None try: return extract_drawing_json(path.read_text(encoding="utf-8", errors="ignore")) except OSError: return None def existing_tag_elements(drawing) -> dict[str, dict]: if not drawing: return {} out = {} for el in drawing.get("elements", []): data = el.get("customData") or {} tag = data.get("tagfeedTag") if tag and el.get("type") == "rectangle" and not el.get("isDeleted"): out[tag] = el return out def find_managed_frame(drawing) -> Optional[dict]: if not drawing: return None for el in drawing.get("elements", []): if el.get("type") == "frame" and el.get("name") == MANAGED_FRAME_NAME and not el.get("isDeleted"): return el return None def element_center(el: dict) -> tuple[float, float]: return ( float(el.get("x", 0)) + float(el.get("width", 0)) / 2, float(el.get("y", 0)) + float(el.get("height", 0)) / 2, ) def is_in_frame(el: dict, frame: dict) -> bool: if el.get("id") == frame.get("id"): return False if el.get("frameId") == frame.get("id"): return True x = float(frame.get("x", 0)) y = float(frame.get("y", 0)) w = float(frame.get("width", 0)) h = float(frame.get("height", 0)) cx, cy = element_center(el) return x <= cx <= x + w and y <= cy <= y + h def managed_tag_elements(drawing, frame: dict) -> dict[str, list[dict]]: out: dict[str, list[dict]] = {} for el in drawing.get("elements", []): data = el.get("customData") or {} tag = data.get("tagfeedTag") if tag and not el.get("isDeleted") and is_in_frame(el, frame): out.setdefault(tag, []).append(el) return out def system_text_elements(drawing, system_key: str) -> list[dict]: out = [] for el in drawing.get("elements", []): data = el.get("customData") or {} if data.get("tagfeedSystem") == system_key and not el.get("isDeleted"): out.append(el) return out def used_slots(existing_rects: dict[str, dict], origin_x: float = START_X, origin_y: float = START_Y) -> set[tuple[int, int]]: slots = set() for el in existing_rects.values(): try: col = round((float(el.get("x", origin_x)) - origin_x) / (CARD_W + GAP_X)) row = round((float(el.get("y", origin_y)) - origin_y) / (CARD_H + GAP_Y)) except Exception: continue if col >= 0 and row >= 0: slots.add((row, col)) return slots def infer_grid_origin(existing_rects: dict[str, dict], fallback_x: float, fallback_y: float) -> tuple[float, float]: rects = [el for el in existing_rects.values() if not el.get("isDeleted")] if not rects: return fallback_x, fallback_y rows: list[tuple[float, int]] = [] for el in rects: y = float(el.get("y", fallback_y)) for idx, (row_y, count) in enumerate(rows): if abs(row_y - y) <= 8: rows[idx] = ((row_y * count + y) / (count + 1), count + 1) break else: rows.append((y, 1)) # The actual tag list is the grid row, not incidental new cards near the frame title. dense_rows = [row_y for row_y, count in rows if count >= min(3, COLS)] origin_y = min(dense_rows) if dense_rows else min(float(el.get("y", fallback_y)) for el in rects) grid_rects = [el for el in rects if float(el.get("y", fallback_y)) >= origin_y - 8] origin_x = min(float(el.get("x", fallback_x)) for el in grid_rects) if grid_rects else fallback_x return origin_x, origin_y def append_slot(used: set[tuple[int, int]], n: int) -> tuple[int, int]: if not used: absolute = n else: absolute = max(row * COLS + col for row, col in used) + 1 + n return absolute // COLS, absolute % COLS def tag_custom_data(tag: TagInfo, role: str, created_date: Optional[str] = None) -> dict: data = {"tagfeedTag": tag.tag, "tagfeedPage": tag.page_rel, "tagfeedRole": role} if created_date: data["tagfeedCreatedDate"] = created_date return data def rectangle_for( tag: TagInfo, vault: Path, x: int, y: int, order: int, frame_id: Optional[str] = None, created_date: Optional[str] = None, ) -> dict: rect_id = stable_id("rect", tag.tag) text_id = stable_id("text", tag.tag) return { "id": rect_id, "type": "rectangle", "x": x, "y": y, "width": CARD_W, "height": CARD_H, "angle": 0, "strokeColor": "#1e293b", "backgroundColor": "#f8fafc", "fillStyle": "solid", "strokeWidth": 1, "strokeStyle": "solid", "roughness": 1, "opacity": 100, "roundness": {"type": 3}, "seed": stable_seed("rect:" + tag.tag), "version": 1, "versionNonce": stable_seed("rect-nonce:" + tag.tag), "index": index_name(order), "isDeleted": False, "groupIds": [], "frameId": frame_id, "boundElements": [{"type": "text", "id": text_id}], "updated": UPDATED, "link": obsidian_uri(vault, tag.page_rel), "locked": False, "customData": tag_custom_data(tag, "rect", created_date), } def text_for( tag: TagInfo, vault: Path, x: int, y: int, order: int, frame_id: Optional[str] = None, created_date: Optional[str] = None, ) -> dict: rect_id = stable_id("rect", tag.tag) text_id = stable_id("text", tag.tag) return { "id": text_id, "type": "text", "x": x + 14, "y": y + 18, "width": CARD_W - 28, "height": 28, "angle": 0, "strokeColor": "#0f172a", "backgroundColor": "transparent", "fillStyle": "solid", "strokeWidth": 1, "strokeStyle": "solid", "roughness": 1, "opacity": 100, "roundness": None, "seed": stable_seed("text:" + tag.tag), "version": 1, "versionNonce": stable_seed("text-nonce:" + tag.tag), "index": index_name(order), "isDeleted": False, "groupIds": [], "frameId": frame_id, "boundElements": [], "updated": UPDATED, "link": obsidian_uri(vault, tag.page_rel), "locked": False, "text": tag.tag, "fontSize": 20, "fontFamily": 5, "textAlign": "center", "verticalAlign": "middle", "containerId": rect_id, "originalText": tag.tag, "lineHeight": 1.25, "autoResize": False, "customData": tag_custom_data(tag, "label", created_date), } def date_label_for( tag: TagInfo, vault: Path, x: int, y: int, order: int, frame_id: Optional[str], created_date: str, ) -> dict: date_id = stable_id("date", tag.tag) return { "id": date_id, "type": "text", "x": x, "y": y + CARD_H + DATE_LABEL_GAP, "width": CARD_W, "height": DATE_LABEL_H, "angle": 0, "strokeColor": "#94a3b8", "backgroundColor": "transparent", "fillStyle": "solid", "strokeWidth": 1, "strokeStyle": "solid", "roughness": 1, "opacity": 100, "roundness": None, "seed": stable_seed("date:" + tag.tag), "version": 1, "versionNonce": stable_seed("date-nonce:" + tag.tag), "index": index_name(order), "isDeleted": False, "groupIds": [], "frameId": frame_id, "boundElements": [], "updated": UPDATED, "link": None, "locked": False, "text": created_date, "fontSize": DATE_LABEL_FONT_SIZE, "fontFamily": 5, "textAlign": "center", "verticalAlign": "top", "containerId": None, "originalText": created_date, "lineHeight": 1.25, "autoResize": False, "customData": tag_custom_data(tag, "date", created_date), } def title_elements(vault: Path) -> list[dict]: title_id = stable_id("title", OUT_REL) note_id = stable_id("note", OUT_REL) return [ { "id": title_id, "type": "text", "x": START_X, "y": 0, "width": 720, "height": 42, "angle": 0, "strokeColor": "#111827", "backgroundColor": "transparent", "fillStyle": "solid", "strokeWidth": 1, "strokeStyle": "solid", "roughness": 1, "opacity": 100, "roundness": None, "seed": stable_seed("title"), "version": 1, "versionNonce": stable_seed("title-nonce"), "index": "a00000", "isDeleted": False, "groupIds": [], "frameId": None, "boundElements": [], "updated": UPDATED, "link": None, "locked": False, "text": "Tagfeed页面总清单", "fontSize": 32, "fontFamily": 5, "textAlign": "left", "verticalAlign": "top", "containerId": None, "originalText": "Tagfeed页面总清单", "lineHeight": 1.25, "autoResize": True, "customData": {"tagfeedSystem": "title"}, }, { "id": note_id, "type": "text", "x": START_X, "y": 52, "width": 960, "height": 28, "angle": 0, "strokeColor": "#64748b", "backgroundColor": "transparent", "fillStyle": "solid", "strokeWidth": 1, "strokeStyle": "solid", "roughness": 1, "opacity": 100, "roundness": None, "seed": stable_seed("note"), "version": 1, "versionNonce": stable_seed("note-nonce"), "index": "a00001", "isDeleted": False, "groupIds": [], "frameId": None, "boundElements": [], "updated": UPDATED, "link": None, "locked": False, "text": "自动补齐缺失标签；已有标签卡片的位置和样式由你手动整理，脚本不重排。", "fontSize": 18, "fontFamily": 5, "textAlign": "left", "verticalAlign": "top", "containerId": None, "originalText": "自动补齐缺失标签；已有标签卡片的位置和样式由你手动整理，脚本不重排。", "lineHeight": 1.25, "autoResize": True, "customData": {"tagfeedSystem": "note"}, }, ] def element_role(el: dict) -> Optional[str]: return (el.get("customData") or {}).get("tagfeedRole") def choose_one(elements: list[dict], element_type: str, role: Optional[str] = None) -> Optional[dict]: candidates = [el for el in elements if el.get("type") == element_type and not el.get("isDeleted")] if role is not None: role_candidates = [el for el in candidates if element_role(el) == role] if not role_candidates and role in ("rect", "label"): role_candidates = [el for el in candidates if not element_role(el)] candidates = role_candidates if not candidates: return None return sorted(candidates, key=lambda el: str(el.get("index", "")))[0] def existing_created_date(*elements: Optional[dict]) -> Optional[str]: for el in elements: if not el: continue value = (el.get("customData") or {}).get("tagfeedCreatedDate") if isinstance(value, str) and re.fullmatch(r"\d{2}\.\d{4}", value): return value return None def sync_kept_element( el: dict, info: TagInfo, vault: Path, frame_id: str, role: str, created_date: Optional[str] = None, ) -> bool: before = json.dumps(el, ensure_ascii=False, sort_keys=True) if role == "date": el["link"] = None else: el["link"] = obsidian_uri(vault, info.page_rel) el["frameId"] = frame_id el["customData"] = tag_custom_data(info, role, created_date) if role == "label" and el.get("type") == "text": el["text"] = info.tag el["originalText"] = info.tag elif role == "date" and el.get("type") == "text" and created_date: el["text"] = created_date el["originalText"] = created_date after = json.dumps(el, ensure_ascii=False, sort_keys=True) return before != after def merge_elements(vault: Path, existing, tags: list[TagInfo]) -> tuple[list[dict], int, int, bool, bool]: if not existing: return [], 0, 0, False, False frame = find_managed_frame(existing) if frame is None: return existing.get("elements", []), 0, 0, False, False old_elements = existing.get("elements", []) if existing else [] managed = managed_tag_elements(existing, frame) title_items = system_text_elements(existing, "title") note_items = system_text_elements(existing, "note") current = {t.tag: t for t in tags} frame_id = frame["id"] keep_ids = set() inserted = removed = 0 updated = False kept_rects: dict[str, dict] = {} added = [] for tag, elems in managed.items(): info = current.get(tag) if info is None: removed += len(elems) continue rect = choose_one(elems, "rectangle", "rect") text = choose_one(elems, "text", "label") created_date = existing_created_date(*elems) date_label = choose_one(elems, "text", "date") if created_date else None if rect is not None: updated = sync_kept_element(rect, info, vault, frame_id, "rect", created_date) or updated keep_ids.add(rect["id"]) kept_rects[tag] = rect if text is not None: updated = sync_kept_element(text, info, vault, frame_id, "label", created_date) or updated keep_ids.add(text["id"]) if created_date and date_label is not None: updated = sync_kept_element(date_label, info, vault, frame_id, "date", created_date) or updated keep_ids.add(date_label["id"]) elif created_date and rect is not None: added.append( date_label_for( info, vault, int(float(rect.get("x", 0))), int(float(rect.get("y", 0))), 2 + len(old_elements) + len(added), frame_id, created_date, ) ) updated = True removed += sum(1 for el in elems if el.get("id") not in keep_ids) kept = [] for el in old_elements: data = el.get("customData") or {} if data.get("tagfeedSystem") in {"title", "note"}: keep_group = title_items if data.get("tagfeedSystem") == "title" else note_items if keep_group and el is not keep_group[0]: removed += 1 continue if keep_group and el is keep_group[0]: kept.append(el) continue # Only prune managed tag elements inside Frame0-all. Everything else, including # tag experiments outside the frame, remains untouched. if data.get("tagfeedTag") and is_in_frame(el, frame) and el.get("id") not in keep_ids: continue kept.append(el) if not title_items: kept = title_elements(vault) + kept inserted += 2 updated = True origin_x = float(frame.get("x", 0)) + FRAME_PADDING_X origin_y = float(frame.get("y", 0)) + FRAME_PADDING_Y origin_x, origin_y = infer_grid_origin(kept_rects, origin_x, origin_y) used = used_slots(kept_rects, origin_x, origin_y) missing = [t for t in tags if t.tag not in kept_rects] for i, tag in enumerate(missing): row, col = append_slot(used, i) x = int(origin_x + col * (CARD_W + GAP_X)) y = int(origin_y + row * (CARD_H + GAP_Y)) created_date = today_yy_mmdd() added.append(rectangle_for(tag, vault, x, y, 2 + len(kept) + len(added), frame_id=frame_id, created_date=created_date)) added.append(text_for(tag, vault, x, y, 2 + len(kept) + len(added), frame_id=frame_id, created_date=created_date)) added.append(date_label_for(tag, vault, x, y, 2 + len(kept) + len(added), frame_id, created_date)) inserted += 1 return kept + added, inserted, removed, True, updated def render_markdown(drawing: dict) -> str: text_items = [] for el in drawing["elements"]: if el.get("type") == "text" and (el.get("text") or "").strip(): text_items.append(f"{el.get('text')} ^{el.get('id')[:8]}") text_block = "\n\n".join(text_items) return ( "---\n" "excalidraw-plugin: parsed\n" "tags: [excalidraw, tagfeed-menu]\n" "---\n" "==⚠ Switch to EXCALIDRAW VIEW in the MORE OPTIONS menu of this document. ⚠==\n\n" "# Excalidraw Data\n\n" "## Text Elements\n" + text_block + "\n\n%%\n## Drawing\n" "```json\n" + json.dumps(drawing, ensure_ascii=False, indent="\t") + "\n```\n%%\n" ) def sync(vault: Path) -> tuple[str, int, int, int, int, int, Path]: synced_at = datetime.now() out = vault / OUT_REL existing = load_existing(out) live_upgraded, _ = upgrade_live_tagfeed_summary_rules(vault) plugin_alias_groups, alias_updated_at = load_plugin_alias_table(vault) alias_mirror_written = write_alias_mirror(vault, plugin_alias_groups, alias_updated_at) alias_to_canonical, renamed, deleted, rewritten = normalize_tagfeed_pages(vault, plugin_alias_groups) tags = collect_tag_infos(vault, alias_to_canonical) pages_created, pages_skipped = create_missing_tagfeed_pages(vault, tags, plugin_alias_groups) elements, inserted, removed, has_frame, updated = merge_elements(vault, existing, tags) if not has_frame: write_statics_page( vault, len(tags), count_missing_pages(vault, tags), count_duplicate_cards(existing), synced_at, tags, ) return "missingframe", len(tags), 0, 0, pages_created, pages_skipped, out drawing = dict(existing or {}) drawing.setdefault("type", "excalidraw") drawing.setdefault("version", 2) drawing.setdefault("source", "sync_excalidraw_tagfeed.py") drawing.setdefault("appState", {}) drawing.setdefault("files", {}) drawing["elements"] = elements statics_written = write_statics_page( vault, len(tags), count_missing_pages(vault, tags), count_duplicate_cards(drawing), synced_at, tags, ) content = render_markdown(drawing) old = "" try: old = out.read_text(encoding="utf-8") except OSError: pass content_changed = old != content if not inserted and not removed and not updated and not content_changed: status = "written" if pages_created or alias_mirror_written or renamed or deleted or rewritten or live_upgraded or statics_written else "unchanged" return status, len(tags), inserted, removed, pages_created, pages_skipped, out if not content_changed: status = "unchanged" else: out.parent.mkdir(parents=True, exist_ok=True) out.write_text(content, encoding="utf-8") status = "written" if renamed or deleted or rewritten or live_upgraded: status = "written" if pages_created: status = "written" if alias_mirror_written: status = "written" if statics_written: status = "written" return status, len(tags), inserted, removed, pages_created, pages_skipped, out def main() -> None: vault = resolve_vault(sys.argv[1] if len(sys.argv) > 1 else None) status, total, inserted, removed, pages_created, pages_skipped, path = sync(vault) print( "Excalidraw-Tagfeed: " f"{status} total_tags={total} inserted={inserted} removed={removed} " f"pages_created={pages_created} pages_skipped={pages_skipped} path={path}" ) if __name__ == "__main__": main()