#!/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"} 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 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 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, 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) elements, inserted, removed, has_frame, updated = merge_elements(vault, existing, tags) if not has_frame: return "missingframe", len(tags), 0, 0, 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: return "unchanged", len(tags), inserted, removed, 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" return status, len(tags), inserted, removed, out def main() -> None: vault = resolve_vault(sys.argv[1] if len(sys.argv) > 1 else None) status, total, inserted, removed, path = sync(vault) print(f"Excalidraw-Tagfeed: {status} total_tags={total} inserted={inserted} removed={removed} path={path}") if __name__ == "__main__": main()