#!/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"
|
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 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 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 = tok.strip().lstrip("#")
|
if tok:
|
out.append(tok)
|
return out
|
|
|
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
|
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():]
|
fm = {}
|
try:
|
import yaml # type: ignore
|
fm = yaml.safe_load(fm_text) or {}
|
except Exception:
|
pass
|
if isinstance(fm, dict) and "tags" in fm:
|
for tok in split_fm_value(fm["tags"]):
|
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:
|
alias_line = ""
|
if aliases:
|
alias_line = "> 别名：" + "、".join(f"#{a}" for a in aliases)
|
lines = text.splitlines()
|
if not aliases:
|
return "\n".join(line for line in lines if not line.startswith("> 别名：")) + ("\n" if text.endswith("\n") else "")
|
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:
|
out.insert(0, alias_line)
|
return "\n".join(out) + ("\n" if text.endswith("\n") 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("."):
|
continue
|
try:
|
text = fp.read_text(encoding="utf-8", errors="ignore")
|
except OSError:
|
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) -> 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)
|
alias_to_canonical: dict[str, str] = {}
|
renamed = 0
|
deleted = 0
|
rewritten = 0
|
|
for canonical, group in groups.items():
|
alias_to_canonical[canonical] = canonical
|
for alias in group.aliases:
|
alias_to_canonical[alias] = canonical
|
|
for canonical, group in groups.items():
|
canonical_path = base / f"{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("."):
|
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 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("."):
|
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/\\\\-])\");\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]) -> tuple[int, int]:
|
base = vault / INBOX_REL
|
base.mkdir(parents=True, exist_ok=True)
|
template = extract_tagfeed_template(vault)
|
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)
|
fp.write_text(template.replace("__TARGET_TAG__", info.tag), 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 = {}
|
latest_seen = {}
|
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 tags:
|
if is_noise(tg):
|
continue
|
counts[tg] = counts.get(tg, 0) + 1
|
latest_seen[tg] = max(latest_seen.get(tg, 0.0), mt)
|
|
page_map: dict[str, str] = {}
|
inbox = vault / INBOX_REL
|
if inbox.is_dir():
|
for fp in inbox.glob("*.md"):
|
if fp.name.startswith("."):
|
continue
|
page_map[fp.stem] = fp.relative_to(vault).as_posix()
|
tags = []
|
for tag in counts:
|
rel_tag = tag_to_relpath(tag)
|
canonical = alias_to_canonical.get(tag, tag)
|
canonical_rel = tag_to_relpath(canonical)
|
page_rel = (
|
page_map.get(canonical)
|
or page_map.get(canonical_rel)
|
or page_map.get(tag)
|
or page_map.get(rel_tag)
|
or f"{INBOX_REL}/{canonical_rel}.md"
|
)
|
page_time = 0.0
|
tags.append(TagInfo(tag=tag, page_rel=page_rel, sort_time=page_time or latest_seen.get(tag, 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## Drawing\n"
|
"```json\n"
|
+ json.dumps(drawing, ensure_ascii=False, indent="\t")
|
+ "\n```\n"
|
)
|
|
|
def sync(vault: Path) -> tuple[str, int, int, int, int, int, Path]:
|
out = vault / OUT_REL
|
existing = load_existing(out)
|
alias_to_canonical, renamed, deleted, rewritten = normalize_tagfeed_pages(vault)
|
tags = collect_tag_infos(vault, alias_to_canonical)
|
pages_created, pages_skipped = create_missing_tagfeed_pages(vault, tags)
|
elements, inserted, removed, has_frame, updated = merge_elements(vault, existing, tags)
|
if not has_frame:
|
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
|
if not inserted and not removed and not updated:
|
status = "written" if pages_created else "unchanged"
|
return status, len(tags), inserted, removed, pages_created, pages_skipped, out
|
content = render_markdown(drawing)
|
old = ""
|
try:
|
old = out.read_text(encoding="utf-8")
|
except OSError:
|
pass
|
if old == content:
|
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:
|
status = "written"
|
if pages_created:
|
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()
|
