#!/usr/bin/env python3
|
# -*- coding: utf-8 -*-
|
"""
|
sync_0inbox.py
|
==============
|
只负责一件事：**扫描 vault 里新出现的 #tag，在
|
00index/0inbox/ 下为每个「尚未建索引页」的标签生成 tagfeed 页面**。
|
|
它与「1 号画布 00index.canvas（A）」解耦：
|
- 不再生成 B 画布（00待整理标签.canvas 已废弃，不再维护）。
|
- 不读取 A 的引用作为排除集（A 永不含 0inbox，由 clean_a_0inbox.py 守护）；
|
sync_0inbox 只判断「该 tag 是否已有 标签聚合/<path>.md 索引页（已整理）」。
|
0inbox 收件箱位于 00index/0inbox，与 标签聚合 平级、互不干扰。
|
|
收录规则（满足则**不**生成页面）
|
--------------------------------
|
1. 该 tag 已被「标签聚合 目录下」的某个索引页覆盖 → 已整理（规则2），跳过。
|
覆盖判定有二：
|
a) 标签聚合下存在同名 <tag>.md 页；
|
b) 某页的 DataviewJS 配置 `targetTag` 显式列出该标签
|
（支持一个页合并多个标签，如 targetTag=["UI","UI相关"]）。
|
这样「UI相关」被合并进 UI.md 后，不会再顽固生成 0inbox/UI相关.md。
|
2. 系统/噪音标签：纯数字、代码型（#ifdef #endif #all #setDefaults）、颜色值（#FF0000）等。
|
3. 用户指定三类自动排除：part0 开头 / index_ 开头 / 含 ThreadPoolF。
|
|
扫描范围（用户指定）
|
--------------------
|
- 顶层黑名单 EXCLUDE_TOP 整棵子树不扫：X2.Archived / 00index / X1.Knomo / X.Attachment / P3.bobo
|
- 嵌套黑名单 EXCLUDE_PATHS 整棵子树不扫：02DS/02copilot（Copilot 对话日志）、
|
02DS/01dril-book（读书笔记库）；两者均「永不从其中取内容生成 tagfeed」。
|
- 仅扫描后缀 SCAN_EXT：.md .markdown .excalidraw .canvas；其它后缀一律跳过
|
- markdown 只抽标准 `tags:` frontmatter 与正文 #tag；topic/category/keywords/type 等
|
元数据字段绝不当作标签（避免 copilot 对话、读书笔记的元数据被误建页）
|
|
幂等 & 安全
|
----------
|
只读取 vault、只写入 00index/0inbox/ 下的 .md；已存在的文件跳过不覆盖。
|
纯标准库（frontmatter 优先 PyYAML，无则内置轻量解析兜底）。
|
|
用法
|
----
|
python3 sync_0inbox.py
|
python3 sync_0inbox.py /path/to/vault
|
"""
|
|
import hashlib
|
import json
|
import os
|
import re
|
import sys
|
|
try:
|
import yaml # type: ignore
|
_HAS_YAML = True
|
except Exception:
|
yaml = None
|
_HAS_YAML = False
|
|
|
# ---------------------------------------------------------------------------
|
# frontmatter 解析
|
# ---------------------------------------------------------------------------
|
def _simple_fm_parse(text: str):
|
data = {}
|
cur_key = None
|
cur_list = None
|
for line in text.splitlines():
|
if not line.strip() or line.strip().startswith("#"):
|
continue
|
m = re.match(r"^([A-Za-z0-9_\-]+):\s*(.*)$", line)
|
if m:
|
key = m.group(1)
|
val = m.group(2).strip()
|
if val.startswith("[") and val.endswith("]"):
|
items = [x.strip().lstrip("#") for x in val[1:-1].split(",")]
|
data[key] = items
|
cur_key, cur_list = None, None
|
elif val == "":
|
cur_key, cur_list = key, []
|
data[key] = cur_list
|
else:
|
data[key] = val.lstrip("#")
|
cur_key, cur_list = None, None
|
elif re.match(r"^\s*-\s*(.*)$", line) and cur_list is not None:
|
cur_list.append(re.match(r"^\s*-\s*(.*)$", line).group(1).strip().lstrip("#"))
|
return data
|
|
|
def parse_frontmatter(fm_text: str):
|
if _HAS_YAML:
|
try:
|
return yaml.safe_load(fm_text) or {}
|
except Exception:
|
pass
|
return _simple_fm_parse(fm_text)
|
|
|
# ---------------------------------------------------------------------------
|
# 过滤规则
|
# ---------------------------------------------------------------------------
|
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")
|
|
|
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
|
# 用户指定三类自动排除：part0 开头 / index_ 开头 / 含 ThreadPoolF
|
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 hid(s: str) -> str:
|
return hashlib.md5(s.encode("utf-8")).hexdigest()[:16]
|
|
|
def resolve_vault(explicit=None) -> str:
|
if explicit:
|
return os.path.normpath(explicit)
|
here = os.path.dirname(os.path.abspath(__file__))
|
return os.path.normpath(os.path.join(here, "..", ".."))
|
|
|
TAG_RE = re.compile(r"(?<![\w/])#([A-Za-z0-9_\u4e00-\u9fff][\w\u4e00-\u9fff/\-]*)")
|
SCAN_EXT = {".md", ".markdown", ".excalidraw", ".canvas"}
|
EXCLUDE_TOP = {"X2.Archived", "00index", "X1.Knomo", "X.Attachment", "P3.bobo"}
|
# 嵌套路径排除（只排除指定子树，不影响其同级目录）。
|
# 02DS/02copilot：Obsidian Copilot 对话日志，topic: 字段与聊天正文不应被当作标签扫描。
|
# 02DS/01dril-book：读书笔记库，其正文/元数据亦不希望进入 tagfeed 体系。
|
# 两者「永不从其中取内容」（与 tagfeed 模板的 excludedPaths 保持一致）。
|
EXCLUDE_PATHS = {"02DS/02copilot", "02DS/01dril-book"}
|
INBOX_DIR = "0inbox"
|
EXCLUDE_DIRS = {".obsidian", ".trash", ".git", ".agents", ".claude",
|
".copilot", ".opencode", ".smart-env", ".workbuddy"}
|
|
# 只把标准 Obsidian 标签字段 `tags:` 当作标签；不再把 topic/category/subject/
|
# type/keywords/aliases 等元数据字段误读为标签（否则 copilot 对话的 topic:、
|
# 读书笔记的 type: 等会被当成标签疯狂建页）。行内 #标签 仍正常扫描。
|
FM_KEYWORD_KEYS = {"tags"}
|
|
|
def split_fm_value(v) -> list:
|
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()
|
if not s:
|
return []
|
s = s.strip().strip('"').strip("'")
|
parts = re.split(r"[，,；;]", s)
|
out = []
|
for p in parts:
|
for tok in p.split():
|
tok = tok.strip().strip('"').strip("'").lstrip("#")
|
if tok:
|
out.append(tok)
|
return out
|
|
|
def collect_tags_in_file(path: str) -> set:
|
ext = os.path.splitext(path)[1].lower()
|
tags = set()
|
try:
|
with open(path, encoding="utf-8", errors="ignore") as f:
|
raw = f.read()
|
except OSError:
|
return tags
|
|
if ext in (".md", ".markdown"):
|
m = re.match(r"^---\s*\n(.*?)\n---\s*\n", raw, re.DOTALL)
|
fm_text = ""
|
body = raw
|
if m:
|
fm_text = m.group(1)
|
body = raw[m.end():]
|
try:
|
fm = parse_frontmatter(fm_text) or {}
|
except Exception:
|
fm = {}
|
if isinstance(fm, dict):
|
for key in FM_KEYWORD_KEYS:
|
if key in fm:
|
for tok in split_fm_value(fm[key]):
|
tok = tok.strip().lstrip("#")
|
if tok:
|
tags.add(tok)
|
# 注意：不再遍历所有 frontmatter 值去抽取 #tag。
|
# 否则 topic/category/keywords 等元数据字段里出现的 # 文本
|
# 会被误判为标签，违背「这类字段不生成 tagfeed 页」的约定。
|
in_fence = False
|
for line in body.splitlines():
|
if line.lstrip().startswith("```"):
|
in_fence = not in_fence
|
continue
|
if in_fence:
|
continue
|
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
|
|
|
# ---------------------------------------------------------------------------
|
# tagfeed 页面模板抽取（从 vault 现有 tagfeed-live 页面）
|
# ---------------------------------------------------------------------------
|
def extract_tagfeed_template(vault: str) -> str:
|
agg_dir = os.path.join(vault, "00index", "标签聚合")
|
seed = None
|
if os.path.isdir(agg_dir):
|
for root, dirs, files in os.walk(agg_dir):
|
dirs[:] = [d for d in dirs if not d.startswith(".")]
|
for fn in files:
|
if fn.startswith(".") or not fn.endswith(".md"):
|
continue
|
fp = os.path.join(root, fn)
|
try:
|
txt = open(fp, encoding="utf-8", errors="ignore").read()
|
except OSError:
|
continue
|
if "tagfeed-live" in txt and "targetTag" in txt:
|
seed = txt
|
break
|
if seed:
|
break
|
if not seed:
|
return (
|
"---\n"
|
"tags:\n - tagfeed-live\n"
|
"cssclasses:\n - tagfeed\n"
|
"---\n\n"
|
"# 🏷 #__TARGET_TAG__（实时聚合页）\n\n"
|
"> 本页由 DataviewJS 实时计算，笔记有变动时自动刷新。\n\n"
|
"```dataviewjs\n"
|
"const targetTag = \"__TARGET_TAG__\";\n"
|
"const excludedPaths = [\"X2.Archived\", \"X0.Clippings\", \"X1.Knomo\", \"02DS/02copilot\", \"02DS/01dril-book\", \".agents\", \".claude\", \".copilot\", \".opencode\", \".smart-env\", \".workbuddy\", \".trash\"];\n"
|
"dv.paragraph(\"#\" + targetTag + \" 聚合占位\");\n"
|
"```\n"
|
)
|
fmt = re.match(r"^---\s*\n(.*?)\n---\s*\n", seed, re.DOTALL)
|
fm = fmt.group(0) if fmt else (
|
"---\n"
|
"tags:\n - tagfeed-live\n"
|
"cssclasses:\n - tagfeed\n"
|
"---\n\n"
|
)
|
dm = re.search(r"```dataviewjs\n(.*?)\n```", seed, re.DOTALL)
|
dv = dm.group(1) if dm else "dv.paragraph(\"#\" + targetTag);"
|
dv = re.sub(r'const\s+targetTag\s*=\s*(?:"[^"]*"|\[[^\]]*\])',
|
'const targetTag = "__TARGET_TAG__"', dv, count=1)
|
title = "# 🏷 #__TARGET_TAG__（实时聚合页）\n\n"
|
note = "> 本页由 DataviewJS 实时计算，笔记有变动时自动刷新，无需任何后台服务。\n\n"
|
return fm + title + note + "```dataviewjs\n" + dv + "\n```\n"
|
|
|
# ---------------------------------------------------------------------------
|
# 已整理标签集合：标签聚合目录下所有索引页所覆盖的标签
|
# ---------------------------------------------------------------------------
|
def extract_covered_tags(vault: str) -> set:
|
"""返回「标签聚合」目录下所有已整理索引页所覆盖的标签集合。
|
|
覆盖判定（二者任一即算）：
|
a) 页面文件名本身（去掉 .md）即一个已整理标签；
|
b) 页面内 DataviewJS 配置 `const targetTag = ...` 显式列出的标签，
|
支持数组形式（一个页合并多个标签，如 ["UI","UI相关"]）。
|
|
用途：这些标签视为已治理，sync_0inbox 不再为它们在 0inbox 生成页面
|
（规则2）。特别地，合并页（UI.md 覆盖 UI + UI相关）能让「被合并的」
|
标签也一并被排除，避免顽固再生。
|
|
嵌套标签分隔符 / 已规范化为 _（与 tag_to_relpath 一致）。
|
"""
|
agg_dir = os.path.join(vault, "00index", "标签聚合")
|
covered = set()
|
if not os.path.isdir(agg_dir):
|
return covered
|
tt_re = re.compile(
|
r'const\s+targetTag\s*=\s*(?:\[([^\]]*)\]|["\']([^"\']*)["\'])')
|
for root, dirs, files in os.walk(agg_dir):
|
dirs[:] = [d for d in dirs if not d.startswith(".")]
|
if os.path.basename(root) == INBOX_DIR:
|
dirs[:] = []
|
continue
|
for fn in files:
|
if fn.startswith(".") or not fn.endswith(".md"):
|
continue
|
fp = os.path.join(root, fn)
|
# a) 文件名即一个已整理标签
|
covered.add(tag_to_relpath(fn[:-3]))
|
# b) 解析页面内 targetTag 声明
|
try:
|
txt = open(fp, encoding="utf-8", errors="ignore").read()
|
except OSError:
|
continue
|
m = tt_re.search(txt)
|
if not m:
|
continue
|
raw = m.group(1) if m.group(1) is not None else m.group(2)
|
for piece in raw.split(","):
|
piece = piece.strip().strip('"').strip("'").lstrip("#").strip()
|
if piece:
|
covered.add(tag_to_relpath(piece))
|
return covered
|
|
|
def write_tagfeed_pages(vault: str, tags: list, template: str) -> tuple:
|
"""为每个待生成 tag 在 00index/0inbox/ 下生成 tagfeed 页面。
|
返回 (created, skipped)；已存在同名文件则跳过不覆盖。"""
|
inbox = os.path.join(vault, "00index", "0inbox")
|
os.makedirs(inbox, exist_ok=True)
|
created = skipped = 0
|
for tag in tags:
|
rel = tag_to_relpath(tag)
|
fp = os.path.join(inbox, rel + ".md")
|
if os.path.exists(fp):
|
skipped += 1
|
continue
|
content = template.replace("__TARGET_TAG__", tag)
|
with open(fp, "w", encoding="utf-8") as f:
|
f.write(content)
|
created += 1
|
return created, skipped
|
|
|
# ---------------------------------------------------------------------------
|
# 主流程
|
# ---------------------------------------------------------------------------
|
def main() -> None:
|
vault = resolve_vault(sys.argv[1] if len(sys.argv) > 1 else None)
|
agg_dir = os.path.join(vault, "00index", "标签聚合")
|
|
# 1) 已整理标签集合：标签聚合目录下所有索引页所覆盖的标签
|
# （文件名 + 各页 targetTag 声明，支持合并页；规则2）
|
covered = extract_covered_tags(vault)
|
|
# 2) 全 vault 扫描 #tag
|
all_tags = {}
|
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
|
# 嵌套路径排除（如 02DS/02copilot）
|
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 = os.path.join(root, fn)
|
for tg in collect_tags_in_file(full):
|
if is_noise(tg):
|
continue
|
all_tags[tg] = all_tags.get(tg, 0) + 1
|
|
# 3) 待生成 = 有使用、但「已被标签聚合某页覆盖」之外的 tag
|
pending = []
|
for tag in all_tags:
|
rel = tag_to_relpath(tag)
|
if rel in covered:
|
continue
|
pending.append(tag)
|
pending.sort(key=lambda t: (-all_tags[t], t.lower()))
|
|
# 4) 生成 0inbox tagfeed 页面
|
template = extract_tagfeed_template(vault)
|
created, skipped = write_tagfeed_pages(vault, pending, template)
|
|
print("已同步 0inbox 页面: %s" % os.path.join(vault, "00index", "0inbox"))
|
print("vault 实际 #tag(去噪)=%d 已覆盖标签=%d 待生成(0inbox)=%d"
|
% (len(all_tags), len(covered), len(pending)))
|
print("0inbox 页面: 新建=%d 已存在跳过=%d" % (created, skipped))
|
|
|
if __name__ == "__main__":
|
main()
|
