#!/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"(?<![\w/])#([A-Za-z0-9_\u4e00-\u9fff][\w\u4e00-\u9fff/\-]*)")
|
NOISE_EXACT = {
|
"ifdef", "endif", "if", "else", "elif", "ifndef", "define", "undef",
|
"include", "pragma", "all", "todo", "tag", "tags",
|
"rrggbb", "ffffff", "000000", "ff0000", "00ff00", "0000ff",
|
}
|
NOISE_PREFIX = ("setdefaults", "set", "get", "is", "en", "ff", "rr")
|
CARD_W = 220
|
CARD_H = 64
|
GAP_X = 28
|
GAP_Y = 22
|
DATE_LABEL_GAP = 4
|
DATE_LABEL_H = 14
|
DATE_LABEL_FONT_SIZE = 10
|
COLS = 5
|
START_X = 0
|
START_Y = 120
|
FRAME_PADDING_X = 28
|
FRAME_PADDING_Y = 48
|
UPDATED = 1
|
|
|
@dataclass(frozen=True)
|
class TagInfo:
|
tag: str
|
page_rel: str
|
sort_time: float
|
|
|
@dataclass(frozen=True)
|
class TagfeedPageGroup:
|
canonical: str
|
aliases: tuple[str, ...]
|
path: Path
|
|
|
def resolve_vault(explicit=None) -> 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(\"(?<![\\\\w\\\\u4e00-\\\\u9fff#])#\" + esc(targetTag) + \"(?![\\\\w\\\\u4e00-\\\\u9fff/\\\\-])\", \"i\");\n"
|
"const rows = [];\n"
|
"for (const page of dv.pages()) {\n"
|
" if (excludedPaths.some(x => 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()
|
