#!/usr/bin/env python3
|
# -*- coding: utf-8 -*-
|
"""
|
tagfeed.py — Logseq 风格的 Obsidian 标签页生成器（无插件依赖）
|
|
给定一个标签，扫描整个 vault，把所有带该标签的"块"（bullet / memo / 段落 / 小节）
|
按日期倒序聚合到一个页面里，样式模仿 Logseq 的 query 结果：
|
|
## 📅 2026-08-05 · 周二
|
- 🔗 [[26.0805]] · GTD
|
- 块内容（子 bullet 原样保留）
|
|
日期判定优先级：
|
1. 块上方最近的日期标题（Knomo 的 `## [[2026-08-04]]` 形式）
|
2. 文件名里的日期（`26.0805.md` → 2026-08-05，`2026-08-04-xxx.md` 等）
|
3. frontmatter 里的 date/created 字段
|
4. 文件修改时间
|
|
用法：
|
python3 tagfeed.py <标签> # 生成聚合页（标签可带可不带 #）
|
python3 tagfeed.py <标签> --open # 生成后在 Obsidian 中打开
|
python3 tagfeed.py <标签> --out 某目录 # 指定输出目录（相对 vault，默认 00index）
|
python3 tagfeed.py <标签> --title 我的标题 # 自定义页面标题
|
python3 tagfeed.py --tags # 列出全库标签及出现次数
|
python3 tagfeed.py <标签> --vault 路径 # 指定其他 vault
|
|
输出文件： <vault>/<out>/tag-<标签>.md
|
重复运行即刷新（覆盖旧文件）。
|
"""
|
|
import argparse
|
import collections
|
import datetime
|
import difflib
|
import hashlib
|
import json
|
import os
|
import re
|
import subprocess
|
import sys
|
import time
|
import urllib.parse
|
|
DEFAULT_VAULT = os.environ.get(
|
"TAGFEED_VAULT", os.path.expanduser("~/Downloads/Syn/Ob/Ob.temp")
|
)
|
SKIP_DIRS = {".obsidian", ".git", ".trash", "X.Attachment", ".hg", ".svn"}
|
TAGFEED_SKIP_REL_PREFIXES = ("00index", "X1.Knomo", "02DS/01dril-book")
|
MARKER = "<!-- tagfeed" # 生成文件头部的标记，扫描时跳过
|
WEEKDAYS = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
|
MAX_SECTION_LINES = 300 # 小节捕获的行数上限，防止异常文件
|
PREVIEW_MAX_LINES = 3 # 聚合页内每个块最多预览的行数
|
PREVIEW_MAX_CHARS = 220 # 聚合页内每个块最多预览的字符数
|
WATCH_DEFAULT_VAULTS = [ # watch 模式默认监听的 vault
|
os.path.expanduser("~/Downloads/Syn/Ob/Ob.temp"),
|
os.path.expanduser("~/Downloads/Syn/Ob/Ob.bobo"),
|
]
|
DEFAULT_DIARY_DIR = os.path.expanduser("~/Downloads/Syn/Ob/Ob.temp/X0.Diary")
|
DIARY_MARK_BEGIN = "<!-- daily-activity:begin (tagfeed 自动生成，勿手改此块内) -->"
|
DIARY_MARK_END = "<!-- daily-activity:end -->"
|
DIARY_BACKFILL_DAYS = 7 # 回填最近 N 天的活动
|
ACTIVITY_LIST_MAX = 30 # 每种类型最多列出的文件数，超出折叠
|
DIARY_SKIP_DIRS = {"X.Attachment", "attachments", "Assets", "assets"}
|
DIARY_SKIP_REL_PREFIXES = ("00index", "X1.Knomo", "X2.Archived", "02DS/01dril-book")
|
DIARY_COPILOT_ROOT_REL = "02DS/02copilot"
|
DIARY_COPILOT_ALLOWED_REL = "02DS/02copilot/copilot-conversations"
|
TRACKED_SUFFIXES = (".md", ".canvas", ".excalidraw", ".csv", ".tsv",
|
".xlsx", ".xls", ".pdf", ".docx", ".doc", ".epub")
|
DIARY_ACTIVITY_TYPES = [ # 参与"每日足迹"追踪的文件类型（按顺序判定）
|
("白板", lambda f: f.endswith(".excalidraw.md") or f.endswith(".excalidraw") or f.endswith(".canvas")),
|
("表格", lambda f: f.lower().endswith((".csv", ".tsv", ".xlsx", ".xls"))),
|
("文档", lambda f: f.lower().endswith((".pdf", ".docx", ".doc", ".epub"))),
|
("笔记", lambda f: f.endswith(".md")),
|
]
|
CANVAS_DIR = "00index" # 目录脑图索引的文件夹
|
CANVAS_NAME = "00index.md" # 目录索引文件名
|
CANVAS_INDEX_ENABLED = False # 禁止自动生成/复活画布索引
|
CANVAS_FILE_W, CANVAS_FILE_H = 240, 54 # 文件节点尺寸
|
CANVAS_FOLDER_W, CANVAS_FOLDER_H = 150, 60 # 文件夹/根节点尺寸
|
CANVAS_H_GAP, CANVAS_V_GAP = 90, 24 # 列间距 / 节点纵向间距
|
CANVAS_COLORS = ["1", "2", "3", "4", "5", "6"] # 历史兼容：当前不会启用
|
CANVAS_DEFAULT_COLOR = "1" # 历史兼容：当前不会启用
|
CANVAS_SKIP_DIRS = {"tagfeed-md"} # 目录索引时排除 tagfeed 页面目录
|
ACTIVITY_ICONS = {"白板": "🖼", "表格": "📊", "文档": "📄", "笔记": "📝"}
|
# ---- 任务动态（Task Board 任务/子任务变更追踪）----
|
TASK_LINE_RE = re.compile(r"^(\s*)-\s*\[( |x|X|/|-)\]\s+(.*)$")
|
TASK_ID_RE = re.compile(r"🆔\s*(\d+)")
|
TASK_META_RE = re.compile(r"📅\s*[\d\-/]+|🆔\s*\d+|⛔\s*\d+|🔁\s*\S+")
|
WIKILINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|([^\]]+))?\]\]")
|
MDLINK_RE = re.compile(r"\[([^\]]*)\]\([^)]*\)")
|
ANY_TAG_RE = re.compile(r"(?<![\w&/])#([\w\u4e00-\u9fff][\w\u4e00-\u9fff/\-]*)")
|
STATUS_LABEL = {" ": "待办", "/": "进行中", "x": "完成", "-": "取消"}
|
TASK_EVENT_MAX_TEXT = 40
|
TASK_EVENT_LIST_MAX = 30 # 日记任务动态小节每天最多显示条数
|
DIARY_STATE_PATH = os.path.expanduser("~/.tagfeed-diary-state.json")
|
HASH_CAP_BYTES = 8 * 1024 * 1024 # 大文件只哈希首尾各 8MB + 文件大小
|
|
|
# ---------------------------------------------------------------- 工具函数
|
|
def tag_regex(tag: str) -> re.Pattern:
|
"""精确匹配一个标签：前面不能是词字符/#，后面不能是词字符 / - （避免 #bobo 匹配 #bobo-xyz）"""
|
return re.compile(r"(?<![\w#])#" + re.escape(tag) + r"(?![\w/\-])")
|
|
|
ALL_TAGS_RE = re.compile(r"(?<![\w#])#([\w\u4e00-\u9fff][\w\u4e00-\u9fff/\-]*)")
|
|
LIST_ITEM_RE = re.compile(r"^(\s*)([-*+]|\d+[.)])(?:\s+(.*))?$")
|
HEADING_RE = re.compile(r"^(#{1,6})\s+(.*)$")
|
DATE_HEADING_RE = re.compile(r"\[\[(\d{4})-(\d{2})-(\d{2})\]\]")
|
TIME_RE = re.compile(r"(\d{1,2}):(\d{2})(?::(\d{2}))?")
|
FENCE_RE = re.compile(r"^\s*(```|~~~)")
|
INLINE_CODE_RE = re.compile(r"`[^`\n]*`")
|
INLINE_COMMENT_RE = re.compile(r"%%.*?%%", re.S)
|
INLINE_HTML_RE = re.compile(r"<!--.*?-->", re.S)
|
|
|
def indent_of(line: str) -> int:
|
"""缩进宽度（tab 按 2 空格计）"""
|
expanded = line.expandtabs(2)
|
return len(expanded) - len(expanded.lstrip(" "))
|
|
|
def clean_for_match(line: str) -> str:
|
"""去掉行内代码 / %%注释%% / HTML 注释，避免误匹配"""
|
line = INLINE_CODE_RE.sub("", line)
|
line = INLINE_COMMENT_RE.sub("", line)
|
line = INLINE_HTML_RE.sub("", line)
|
return line
|
|
|
def valid_date(y: int, m: int, d: int):
|
try:
|
return datetime.date(y, m, d)
|
except ValueError:
|
return None
|
|
|
def date_from_filename(basename: str):
|
"""支持 YYYY-MM-DD 与日记命名 YY.MMDD（如 26.0805 → 2026-08-05）"""
|
m = re.search(r"(\d{4})-(\d{2})-(\d{2})", basename)
|
if m:
|
dt = valid_date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
if dt:
|
return dt
|
m = re.match(r"^(\d{2})\.(\d{2})(\d{2})(?:\D|$)", basename)
|
if m:
|
return valid_date(2000 + int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
return None
|
|
|
def parse_frontmatter(text: str):
|
"""返回 (tags:set, date:datetime.date|None, body_start_line:int)"""
|
tags, fdate = set(), None
|
lines = text.split("\n")
|
if not lines or lines[0].strip() != "---":
|
return tags, fdate, 0
|
for i in range(1, min(len(lines), 200)):
|
if lines[i].strip() in ("---", "..."):
|
block = "\n".join(lines[1:i])
|
tm = re.search(r"^tags:\s*\[(.*?)\]", block, re.M)
|
if tm:
|
for t in tm.group(1).split(","):
|
t = t.strip().strip("\"'").lstrip("#").strip()
|
if t:
|
tags.add(t)
|
else:
|
tm2 = re.search(r"^tags:\s*\n((?:[ \t]+-[ \t]+.*\n?)+)", block, re.M)
|
if tm2:
|
for ln in tm2.group(1).splitlines():
|
mm = re.match(r"^\s*-\s*[\"']?([\w\u4e00-\u9fff/\-]+)[\"']?\s*$", ln)
|
if mm:
|
tags.add(mm.group(1))
|
for key in ("date", "created", "published"):
|
dm = re.search(rf"^{key}:\s*[\"']?(\d{{4}})-(\d{{2}})-(\d{{2}})", block, re.M)
|
if dm:
|
fdate = valid_date(*map(int, dm.groups()))
|
break
|
return tags, fdate, i + 1
|
return set(), None, 0
|
|
|
def strip_tag_from_text(text: str, rx: re.Pattern) -> str:
|
text = rx.sub("", text)
|
text = re.sub(r"[ \t]{2,}", " ", text)
|
return text.strip()
|
|
|
# ---------------------------------------------------------------- 扫描
|
|
class Entry:
|
"""一个命中的块"""
|
__slots__ = ("date", "time", "order", "relpath", "note", "breadcrumb",
|
"lines", "page_level", "notes", "anchor")
|
|
def __init__(self, date, time, order, relpath, note, breadcrumb, lines,
|
page_level=False, anchor=None):
|
self.date = date
|
self.time = time # (h,m,s) 或 None
|
self.order = order # 文件内原始顺序
|
self.relpath = relpath
|
self.note = note
|
self.breadcrumb = breadcrumb
|
self.lines = lines # [(indent_level, text)] indent_level: 0=顶层
|
self.page_level = page_level
|
self.notes = [note] # 来源笔记列表（去重合并后可能有多个）
|
self.anchor = anchor # 所在小节的原始标题（用于 [[笔记#标题]] 跳转）
|
|
|
def rel_is_or_under(rel: str, prefix: str) -> bool:
|
return rel == prefix or rel.startswith(prefix + "/")
|
|
|
def should_skip_tagfeed_path(path: str, vault: str) -> bool:
|
rel = os.path.relpath(path, vault)
|
if rel == ".":
|
return False
|
rel = rel.replace(os.sep, "/")
|
return any(rel_is_or_under(rel, prefix) for prefix in TAGFEED_SKIP_REL_PREFIXES)
|
|
|
def iter_md_files(vault: str):
|
for root, dirs, files in os.walk(vault):
|
dirs[:] = [d for d in dirs
|
if d not in SKIP_DIRS
|
and not should_skip_tagfeed_path(os.path.join(root, d), vault)]
|
for f in files:
|
if not f.endswith(".md") or f.endswith(".excalidraw.md"):
|
continue
|
path = os.path.join(root, f)
|
if should_skip_tagfeed_path(path, vault):
|
continue
|
yield path
|
|
|
def scan_file(path: str, relpath: str, tag: str, rx: re.Pattern, collect_all: bool = False):
|
"""
|
扫描单个文件，返回 (entries, all_tags_counter)。
|
collect_all=True 时只统计全库标签（--tags 模式），不做块提取。
|
"""
|
try:
|
with open(path, encoding="utf-8", errors="replace") as fh:
|
text = fh.read()
|
except OSError:
|
return [], collections.Counter()
|
|
if MARKER in text[:1000]:
|
return [], collections.Counter() # 跳过本工具生成的聚合页
|
|
fm_tags, fm_date, body_start = parse_frontmatter(text)
|
if fm_tags & {"tagfeed", "tagfeed-live"}:
|
return [], collections.Counter() # 跳过 tagfeed 系统页（含 dataviewjs 实时页，
|
# 它们的标题/说明里含 #标签 字样，不能被索引）
|
file_date = date_from_filename(os.path.splitext(os.path.basename(path))[0])
|
try:
|
mtime_date = datetime.date.fromtimestamp(os.path.getmtime(path))
|
except OSError:
|
mtime_date = None
|
|
lines = text.split("\n")
|
all_tags = collections.Counter()
|
if collect_all:
|
in_fence = False
|
for i, ln in enumerate(lines):
|
if i < body_start:
|
continue
|
if FENCE_RE.match(ln):
|
in_fence = not in_fence
|
continue
|
if in_fence:
|
continue
|
for t in ALL_TAGS_RE.findall(clean_for_match(ln)):
|
all_tags[t] += 1
|
for t in fm_tags:
|
all_tags[t] += 1
|
return [], all_tags
|
|
note = os.path.splitext(os.path.basename(path))[0]
|
entries = []
|
seen_blocks = set()
|
order = 0
|
|
in_fence = False
|
in_pct = False # %% 块注释
|
in_html = False # HTML 块注释
|
current_heading = None # 最近的小节标题（面包屑用）
|
current_heading_raw = None # 原始标题文本（[[笔记#标题]] 跳转锚点用）
|
current_date_heading = None # Knomo 风格 ## [[YYYY-MM-DD]]
|
|
def entry_date():
|
return current_date_heading or file_date or fm_date or mtime_date
|
|
def anchor():
|
return None if current_date_heading else current_heading_raw
|
|
def breadcrumb():
|
if current_heading and not DATE_HEADING_RE.search(current_heading):
|
t = ALL_TAGS_RE.sub("", current_heading)
|
t = re.sub(r"\s+", " ", t).strip()
|
if t:
|
return t
|
return None
|
|
def add_lines_block(start, content_lines, first_line_idx):
|
"""content_lines: [(indent, text)] 已规整"""
|
nonlocal order
|
key = (path, first_line_idx)
|
if key in seen_blocks or not content_lines:
|
return
|
seen_blocks.add(key)
|
d = entry_date()
|
if d is None:
|
return
|
t = None
|
m = TIME_RE.search(content_lines[0][1][:30])
|
if m:
|
t = (int(m.group(1)), int(m.group(2)), int(m.group(3) or 0))
|
order += 1
|
entries.append(Entry(d, t, order, relpath, note, breadcrumb(),
|
content_lines, anchor=anchor()))
|
|
def collect_children(i, parent_indent):
|
"""收集 list item i 的子 bullet（缩进更深者），返回规整后的 [(level, text)]"""
|
kids = []
|
j = i + 1
|
raw = []
|
while j < len(lines):
|
ln = lines[j]
|
if ln.strip() == "":
|
# 允许空行，只要后面还有更深的 item
|
k = j + 1
|
while k < len(lines) and lines[k].strip() == "":
|
k += 1
|
if k < len(lines):
|
mk = LIST_ITEM_RE.match(lines[k])
|
if mk and indent_of(lines[k]) > parent_indent:
|
j = k
|
continue
|
break
|
mi = LIST_ITEM_RE.match(ln)
|
if mi and indent_of(ln) > parent_indent:
|
raw.append((indent_of(ln), (mi.group(3) or "").rstrip()))
|
j += 1
|
else:
|
break
|
for ind, txt in raw:
|
kids.append(((ind - parent_indent) // 2, txt))
|
return kids
|
|
for i in range(body_start, len(lines)):
|
ln = lines[i]
|
|
if FENCE_RE.match(ln):
|
in_fence = not in_fence
|
continue
|
if in_fence:
|
continue
|
if in_pct:
|
if "%%" in ln:
|
in_pct = False
|
continue
|
if in_html:
|
if "-->" in ln:
|
in_html = False
|
continue
|
|
stripped = ln.strip()
|
if stripped.startswith("%%") and "%%" not in stripped[2:]:
|
in_pct = True
|
continue
|
if stripped.startswith("<!--") and "-->" not in stripped:
|
in_html = True
|
continue
|
|
hm = HEADING_RE.match(ln)
|
if hm:
|
htext = hm.group(2)
|
dh = DATE_HEADING_RE.search(htext)
|
if dh:
|
current_date_heading = valid_date(*map(int, dh.groups()))
|
current_heading = None
|
current_heading_raw = None
|
else:
|
current_heading = htext
|
current_heading_raw = htext.strip()
|
# 标题行里带标签 → 捕获整个小节
|
if rx.search(clean_for_match(htext)):
|
sec = [(0, "**" + strip_tag_from_text(htext, rx) + "**")]
|
j = i + 1
|
level = len(hm.group(1))
|
count = 0
|
while j < len(lines) and count < MAX_SECTION_LINES:
|
sl = lines[j]
|
shm = HEADING_RE.match(sl)
|
if shm and len(shm.group(1)) <= level:
|
break
|
if FENCE_RE.match(sl):
|
break
|
if sl.strip():
|
smi = LIST_ITEM_RE.match(sl)
|
if smi:
|
sec.append((1 + indent_of(sl) // 2, (smi.group(3) or "").rstrip()))
|
else:
|
sec.append((1, sl.strip()))
|
count += 1
|
j += 1
|
add_lines_block(i, sec, i)
|
continue
|
|
clean = clean_for_match(ln)
|
if rx.search(clean) is None:
|
continue
|
|
# ---- 命中：判断块类型
|
mi = LIST_ITEM_RE.match(ln)
|
if mi:
|
ind = indent_of(ln)
|
item_text = mi.group(3) or ""
|
cleaned = strip_tag_from_text(item_text, rx)
|
kids = collect_children(i, ind)
|
if cleaned:
|
block = [(0, cleaned)] + kids
|
else:
|
# 纯标签 bullet：保留标签可见，子级按自然深度挂在下面
|
block = [(0, item_text.strip())] + kids
|
add_lines_block(i, block, i)
|
continue
|
|
# 段落 / 独立标签行
|
if ALL_TAGS_RE.sub("", clean).strip() in ("", "%%"):
|
# 1) 向下：吸收紧随其后的连续 list items（section 标签语义，如 #taglauncher 下的任务）
|
sec = []
|
j = i + 1
|
while j < len(lines):
|
mj = LIST_ITEM_RE.match(lines[j])
|
if mj:
|
sec.append((indent_of(lines[j]), (mj.group(3) or "").rstrip()))
|
j += 1
|
else:
|
break
|
key_idx = i
|
if not sec:
|
# 2) 向上：标签属于紧邻上方的块（Logseq 块尾标签语义，如歌词末尾的 #lyric）
|
# 向上收集到空行/标题/围栏为止；若途中遇到 bullet，
|
# 则从该 bullet 起捕获整块（bullet + 其续行/子级，直到标签行）
|
up = []
|
k = i - 1
|
bullet_idx = None
|
while k >= body_start:
|
bk = lines[k]
|
if bk.strip() == "" or HEADING_RE.match(bk) or FENCE_RE.match(bk):
|
break
|
up.append(k)
|
if LIST_ITEM_RE.match(bk):
|
bullet_idx = k
|
break
|
k -= 1
|
if bullet_idx is not None:
|
inds = []
|
for bl in lines[bullet_idx:i]:
|
mb = LIST_ITEM_RE.match(bl)
|
if mb:
|
inds.append(indent_of(bl))
|
sec.append((indent_of(bl), (mb.group(3) or "").rstrip()))
|
else:
|
base = inds[-1] + 2 if inds else 0 # 续行挂在上级 bullet 之下
|
sec.append((base, bl.strip()))
|
key_idx = bullet_idx
|
elif up:
|
sec = [(0, lines[k2].strip()) for k2 in reversed(up)]
|
if sec:
|
mi0 = min(ind for ind, _ in sec)
|
norm = [((ind - mi0) // 2, t) for ind, t in sec]
|
add_lines_block(i, norm, key_idx)
|
continue
|
|
# 普通段落带标签：吸收到空行为止
|
para = [strip_tag_from_text(clean, rx)]
|
j = i + 1
|
while j < len(lines):
|
nl = lines[j]
|
if nl.strip() == "" or HEADING_RE.match(nl) or LIST_ITEM_RE.match(nl) or FENCE_RE.match(nl):
|
break
|
para.append(nl.strip())
|
j += 1
|
add_lines_block(i, [(0, " ".join(p for p in para if p))], i)
|
|
# frontmatter 标签 → 页面级条目（附正文开头预览）
|
if tag in fm_tags:
|
d = fm_date or file_date or mtime_date
|
if d:
|
order += 1
|
prev = []
|
for ln2 in lines[body_start:]:
|
s = ln2.strip()
|
if not s:
|
if prev:
|
break
|
continue
|
if s.startswith("#") or s == "---":
|
continue
|
prev.append(ALL_TAGS_RE.sub("", s))
|
if len(prev) >= 2 or sum(len(x) for x in prev) >= 150:
|
break
|
preview = " ".join(p for p in prev if p)[:160]
|
entries.append(Entry(d, None, order, relpath, note, None,
|
[(0, preview)] if preview else [], page_level=True))
|
|
return entries, all_tags
|
|
|
def dedupe(entries):
|
"""
|
同一日期下、同一段内容在多个笔记里重复出现时（典型：日记同步到 Knomo 月度归档），
|
合并为一条，来源链接并排。归并键 = (日期, 内容指纹)。
|
不同日期的相同内容不合并（那是每天的重复记录，Logseq 也会逐日显示）。
|
"""
|
from collections import OrderedDict
|
merged = OrderedDict()
|
for e in entries:
|
fp = (e.date, tuple((lvl, t) for lvl, t in e.lines))
|
if fp in merged:
|
m = merged[fp]
|
if e.note not in m.notes:
|
m.notes.append(e.note)
|
if e.time and (m.time is None or e.time < m.time):
|
m.time = e.time
|
# 若先扫到的来源没有面包屑，而后面的有，则采用后者（与扫描顺序无关，结果确定）
|
if not m.breadcrumb and e.breadcrumb:
|
m.breadcrumb = e.breadcrumb
|
m.anchor = e.anchor
|
m.relpath = e.relpath
|
m.note = e.note
|
else:
|
clone = Entry(e.date, e.time, e.order, e.relpath, e.note,
|
e.breadcrumb, e.lines, e.page_level, e.anchor)
|
clone.notes = [e.note]
|
merged[fp] = clone
|
return list(merged.values())
|
|
|
# ---------------------------------------------------------------- 渲染
|
|
def render(tag: str, entries, vault_name: str) -> str:
|
by_date = collections.defaultdict(list)
|
for e in entries:
|
by_date[e.date].append(e)
|
|
notes = sorted({n for e in entries for n in e.notes})
|
out = []
|
out.append("---")
|
out.append("tags:")
|
out.append(" - tagfeed")
|
out.append("cssclasses:")
|
out.append(" - tagfeed")
|
out.append("---")
|
out.append(f"{MARKER}: {tag} -->")
|
out.append("")
|
out.append(f"# 🏷 #{tag}")
|
out.append("")
|
out.append(f"> **{len(entries)}** 个内容块 ｜ **{len(notes)}** 篇笔记 ｜ 按日期倒序")
|
out.append("")
|
|
for d in sorted(by_date, reverse=True):
|
group = by_date[d]
|
wd = WEEKDAYS[d.weekday()]
|
out.append(f"## 📅 {d.isoformat()} · {wd}")
|
out.append("")
|
|
# 按 (笔记, 面包屑) 分组；组内：有时间的倒序（新在上），无时间的保持原顺序
|
sub = collections.OrderedDict()
|
for e in group:
|
key = (e.relpath, e.breadcrumb or "")
|
sub.setdefault(key, []).append(e)
|
|
def sub_sort_key(item):
|
key, es = item
|
times = [e.time for e in es if e.time]
|
has_time = bool(times)
|
best = max(times) if times else None
|
return (0 if has_time else 1,
|
tuple(-x for x in best) if best else (0,),
|
key[0])
|
|
for (relpath, crumb), es in sorted(sub.items(), key=sub_sort_key):
|
all_notes = []
|
for e in es:
|
for n in e.notes:
|
if n not in all_notes:
|
all_notes.append(n)
|
links = " · ".join(f"[[{n}]]" for n in sorted(all_notes))
|
head = f"- 🔗 {links}" + (f" · {crumb}" if crumb else "")
|
out.append(head)
|
for e in sorted(es, key=lambda e: (e.time is None,
|
tuple(-x for x in e.time) if e.time else (0,),
|
e.order)):
|
block_lines = e.lines
|
truncated = False
|
if not e.page_level:
|
shown, chars = [], 0
|
for lvl, txt in block_lines:
|
if len(shown) >= PREVIEW_MAX_LINES or chars > PREVIEW_MAX_CHARS:
|
truncated = True
|
break
|
shown.append((lvl, txt))
|
chars += len(txt)
|
block_lines = shown
|
for lvl, txt in block_lines:
|
prefix = "\t" * (lvl + 1)
|
txt = txt if txt.strip() else ""
|
out.append(f"{prefix}- {txt}".rstrip())
|
if truncated or e.page_level:
|
target = e.note + (f"#{e.anchor}" if e.anchor else "")
|
label = "页面级标签 · 查看全文" if e.page_level else "查看全文"
|
out.append(f"\t- [[{target}|↗ {label}]]")
|
out.append("")
|
out.append("")
|
|
return "\n".join(out).rstrip() + "\n"
|
|
|
# ---------------------------------------------------------------- 主流程
|
|
def cmd_tags(vault: str):
|
counter = collections.Counter()
|
for path in iter_md_files(vault):
|
_, tags = scan_file(path, os.path.relpath(path, vault), "", None, collect_all=True)
|
counter.update(tags)
|
if not counter:
|
print("（未发现任何标签）")
|
return
|
print(f"{'次数':>5} 标签")
|
for t, c in counter.most_common(60):
|
print(f"{c:>5} #{t}")
|
|
|
def generate_tag(vault: str, tag: str, out_rel_dir: str = "00index",
|
title: str = None, allow_empty: bool = False, scan_fn=None):
|
"""生成/更新一个标签的聚合页。
|
返回 (status, out_path, (blocks, notes, days))，
|
status ∈ {"written", "unchanged", "empty"}。
|
内容与现有页面相同时不写盘（防止触发文件监听循环）。
|
scan_fn: 可选的 (path, rel, tag, rx) -> entries 缓存包装。"""
|
rx = tag_regex(tag)
|
entries = []
|
for path in iter_md_files(vault):
|
rel = os.path.relpath(path, vault)
|
if scan_fn is not None:
|
es = scan_fn(path, rel, tag, rx)
|
else:
|
es, _ = scan_file(path, rel, tag, rx)
|
entries.extend(es)
|
|
out_dir = os.path.join(vault, out_rel_dir.strip("/"))
|
safe = re.sub(r"[/\\:*?\"<>|]", "-", tag)
|
out_rel = os.path.join(out_rel_dir.strip("/"), f"tag-{safe}.md")
|
out_path = os.path.join(vault, out_rel)
|
|
if not entries and not allow_empty:
|
return "empty", out_path, (0, 0, 0)
|
|
entries = dedupe(entries)
|
content = render(tag, entries, os.path.basename(vault))
|
if title:
|
content = content.replace(f"# 🏷 #{tag}", f"# {title}", 1)
|
|
# 哈希门控：内容没变就不写，避免无谓的 mtime 变动和自我触发
|
try:
|
with open(out_path, encoding="utf-8") as fh:
|
if fh.read() == content:
|
dates = len({e.date for e in entries})
|
notes = len({n for e in entries for n in e.notes}) if entries else 0
|
return "unchanged", out_path, (len(entries), notes, dates)
|
except OSError:
|
pass
|
|
os.makedirs(out_dir, exist_ok=True)
|
with open(out_path, "w", encoding="utf-8") as fh:
|
fh.write(content)
|
|
dates = len({e.date for e in entries})
|
notes = len({n for e in entries for n in e.notes}) if entries else 0
|
return "written", out_path, (len(entries), notes, dates)
|
|
|
def discover_subscribed(vault: str):
|
"""扫描 vault 中已存在的 tagfeed 聚合页，返回 {tag: 页面路径}。
|
存在聚合页 = 订阅了该标签的自动更新；删掉页面 = 取消订阅。"""
|
subs = {}
|
for root, dirs, files in os.walk(vault):
|
dirs[:] = [d for d in dirs if d not in SKIP_DIRS]
|
for f in files:
|
if not f.endswith(".md") or not f.startswith("tag-"):
|
continue
|
p = os.path.join(root, f)
|
try:
|
with open(p, encoding="utf-8", errors="replace") as fh:
|
head = fh.read(1000)
|
except OSError:
|
continue
|
if MARKER not in head:
|
continue
|
m = re.search(re.escape(MARKER) + r":\s*(.+?)\s*-->", head)
|
if m:
|
subs[m.group(1)] = p
|
return subs
|
|
|
# ---------------------------------------------------------------- 每日足迹
|
|
def diary_rel_name(d: datetime.date) -> str:
|
"""日记文件名：YYYY-MM-DD.md（如 2026-08-06 → 2026-08-06.md）。
|
2026-08-25 起日记改用 YYYY-MM-DD 格式，与 Obsidian daily-notes 设置一致。"""
|
return d.isoformat() + ".md"
|
|
|
def classify_activity(fname: str):
|
for label, test in DIARY_ACTIVITY_TYPES:
|
if test(fname):
|
return label
|
return None
|
|
|
def should_skip_diary_path(path: str, vault: str) -> bool:
|
rel = os.path.relpath(path, vault)
|
if rel == ".":
|
return False
|
rel = rel.replace(os.sep, "/")
|
parts = rel.split("/")
|
if any(part in SKIP_DIRS or part in DIARY_SKIP_DIRS or part.startswith(".") for part in parts):
|
return True
|
if any(rel_is_or_under(rel, prefix) for prefix in DIARY_SKIP_REL_PREFIXES):
|
return True
|
if rel == DIARY_COPILOT_ROOT_REL:
|
return False
|
if rel_is_or_under(rel, DIARY_COPILOT_ROOT_REL):
|
return not rel_is_or_under(rel, DIARY_COPILOT_ALLOWED_REL)
|
return False
|
|
|
def iter_tracked_files(vault: str):
|
"""所有参与"每日足迹"追踪的文件（含 canvas/表格/文档等非 md 类型）"""
|
for root, dirs, files in os.walk(vault):
|
dirs[:] = [d for d in dirs
|
if d not in SKIP_DIRS and d not in DIARY_SKIP_DIRS
|
and not d.startswith(".")
|
and not should_skip_diary_path(os.path.join(root, d), vault)]
|
for f in files:
|
if f == ".DS_Store" or not f.lower().endswith(TRACKED_SUFFIXES):
|
continue
|
if any(c in f for c in "[]|#\n\r"):
|
continue # 文件名含会破坏 wikilink 的字符，跳过
|
path = os.path.join(root, f)
|
if should_skip_diary_path(path, vault):
|
continue
|
yield path
|
|
|
def in_diary_dir(path: str, vault: str) -> bool:
|
rel = os.path.relpath(os.path.dirname(path), vault)
|
return rel == "X0.Diary" or rel.startswith("X0.Diary" + os.sep)
|
|
|
def is_generated_page(path: str) -> bool:
|
"""tagfeed 生成的聚合页不算用户活动。
|
|
两种系统页都要排除：
|
1) 文件名以 tag- 开头、且头部含 <!-- tagfeed 标记（旧的批量聚合页）
|
2) frontmatter 的 tags: 含 tagfeed 或 tagfeed-live（实时聚合页，
|
如 00index/tagfeed-md/<tag>.md，文件名不以 tag- 开头，
|
头部也无 <!-- tagfeed 标记，单靠前一种判断会漏掉）
|
避免这些"系统自生成页"被当成用户笔记塞进每日活动足迹。"""
|
if not path.endswith(".md"):
|
return False
|
b = os.path.basename(path)
|
if b.startswith("tag-") and b.endswith(".md"):
|
try:
|
with open(path, encoding="utf-8", errors="replace") as fh:
|
return MARKER in fh.read(600)
|
except OSError:
|
return False
|
# 文件名不以 tag- 开头：靠 frontmatter 的 tags 字段判断
|
try:
|
with open(path, encoding="utf-8", errors="replace") as fh:
|
head = fh.read(4000)
|
except OSError:
|
return False
|
if "---" not in head:
|
return False
|
fm = head.split("---", 2)
|
if len(fm) < 3:
|
return False
|
block = fm[1]
|
if re.search(r"^tags:\s*\[(.*?)\]", block, re.M | re.S):
|
if re.search(r"\b(?:tagfeed|tagfeed-live)\b", block):
|
return True
|
if re.search(r"^tags:\s*$", block, re.M):
|
# 列表式 tags:
|
m = re.search(r"^tags:\s*\n((?:[ \t]+-[ \t]+.*\n?)+)", block, re.M)
|
if m:
|
if re.search(r"tagfeed|tagfeed-live", m.group(1)):
|
return True
|
return False
|
|
|
def link_for(path: str, vault: str, name_counts) -> str:
|
"""生成 Obsidian wikilink：名字全库唯一用短链接，否则带目录路径。
|
md 去掉扩展名；canvas/excalidraw 等保留（Obsidian 约定）。"""
|
name = os.path.basename(path)
|
stem = name[:-3] if name.endswith(".md") else name
|
if name_counts[stem] <= 1:
|
return f"[[{stem}]]"
|
rel = os.path.relpath(path, vault)
|
target = rel[:-3] if rel.endswith(".md") else rel
|
return f"[[{target}|{stem}]]"
|
|
|
def file_hash(path: str) -> str:
|
"""内容哈希。小文件全量；大文件只哈希首尾各 HASH_CAP_BYTES + 文件大小。"""
|
try:
|
size = os.path.getsize(path)
|
h = hashlib.sha256()
|
h.update(str(size).encode())
|
with open(path, "rb") as fh:
|
if size <= HASH_CAP_BYTES * 2:
|
for chunk in iter(lambda: fh.read(1 << 20), b""):
|
h.update(chunk)
|
else:
|
h.update(fh.read(HASH_CAP_BYTES))
|
fh.seek(-HASH_CAP_BYTES, os.SEEK_END)
|
h.update(fh.read(HASH_CAP_BYTES))
|
return h.hexdigest()
|
except OSError:
|
return ""
|
|
|
def load_diary_state() -> dict:
|
try:
|
with open(DIARY_STATE_PATH, encoding="utf-8") as fh:
|
return json.load(fh).get("files", {})
|
except (OSError, ValueError):
|
return {}
|
|
|
def save_diary_state(files: dict):
|
"""写状态文件，顺带清理：不存在的文件、超过 14 天的编辑记录。"""
|
cutoff = (datetime.date.today() - datetime.timedelta(days=14)).isoformat()
|
for p in [p for p in files if not os.path.exists(p)]:
|
del files[p]
|
for rec in files.values():
|
rec["edited"] = {d: t for d, t in rec.get("edited", {}).items() if d >= cutoff}
|
tmp = DIARY_STATE_PATH + ".tmp"
|
try:
|
with open(tmp, "w", encoding="utf-8") as fh:
|
json.dump({"files": files}, fh, ensure_ascii=False)
|
os.replace(tmp, DIARY_STATE_PATH)
|
except OSError:
|
pass
|
|
|
def parse_task_snapshot(text: str) -> dict:
|
"""解析笔记里所有复选框任务行 → 快照 {key: [status, 原始文本, 标签串, 父任务key]}。
|
Task Board 管理的任务带 🆔 N，用 "id:N" 作稳定键；无 ID 的用 "txt:文本"。
|
父子关系按缩进判断（子任务缩进更深）。"""
|
tasks = {}
|
stack = [] # [(indent, key)] 父任务链
|
for line in text.split("\n"):
|
m = TASK_LINE_RE.match(line)
|
if not m:
|
continue
|
indent = indent_of(m.group(1))
|
status = m.group(2).lower()
|
raw = m.group(3).strip()
|
mid = TASK_ID_RE.search(raw)
|
key = ("id:" + mid.group(1)) if mid else ("txt:" + raw)
|
if key in tasks:
|
continue # 同键重复极少见，以首个为准
|
while stack and stack[-1][0] >= indent:
|
stack.pop()
|
parent = stack[-1][1] if stack else ""
|
tags_str = " ".join("#" + t for t in ANY_TAG_RE.findall(raw))
|
tasks[key] = [status, raw, tags_str, parent]
|
stack.append((indent, key))
|
return tasks
|
|
|
def diff_task_snapshots(old, new):
|
"""比对新旧任务快照 → 事件列表 (kind, key)。
|
kind: added/status/edited/removed；key 指向新快照（removed 指向旧快照）。
|
无 ID 任务的文本改动用模糊匹配兜底（相似度≥0.6 视为同一任务）。"""
|
pairs = [] # (old_key, new_key)
|
matched_new = set()
|
for k in new:
|
if k in old:
|
pairs.append((k, k))
|
matched_new.add(k)
|
rest_old = set(k for k, _ in pairs)
|
rest_new = [k for k in new if k not in matched_new]
|
used = set()
|
for ok in old:
|
if ok in rest_old or not ok.startswith("txt:"):
|
continue
|
best, br = None, 0.0
|
for nk in rest_new:
|
if nk in used or not nk.startswith("txt:"):
|
continue
|
r = difflib.SequenceMatcher(None, ok[4:], nk[4:]).ratio()
|
if r > br:
|
br, best = r, nk
|
if best is not None and br >= 0.6:
|
pairs.append((ok, best))
|
used.add(best)
|
matched_old = set(p[0] for p in pairs)
|
events = []
|
for ok, nk in pairs:
|
o, n = old[ok], new[nk]
|
if o[0] != n[0]:
|
events.append(("status", nk))
|
elif o[1] != n[1]:
|
events.append(("edited", nk))
|
for k in new:
|
if k not in matched_new and k not in used:
|
events.append(("added", k))
|
for k in old:
|
if k not in matched_old:
|
events.append(("removed", k))
|
return events
|
|
|
def clean_task_text(raw: str) -> str:
|
"""任务行 → 干净的展示文本：去元数据 emoji、链接转显示文字、去标签、超长截断"""
|
t = TASK_META_RE.sub(" ", raw)
|
t = WIKILINK_RE.sub(lambda m: m.group(2) or m.group(1), t)
|
t = MDLINK_RE.sub(lambda m: m.group(1), t)
|
t = ANY_TAG_RE.sub(" ", t)
|
t = re.sub(r"\s+", " ", t).strip()
|
if len(t) > TASK_EVENT_MAX_TEXT:
|
t = t[:TASK_EVENT_MAX_TEXT] + "…"
|
return t
|
|
|
def detect_task_events(path, rec, tstr, d_iso):
|
"""对当前文件做任务快照 diff：有变化就把事件追加进 rec["tasks"][d_iso]，
|
并更新快照。首次见到该文件只建基线不产生事件。返回新增事件数。"""
|
try:
|
with open(path, encoding="utf-8", errors="replace") as fh:
|
text = fh.read()
|
except OSError:
|
return 0
|
new_snap = parse_task_snapshot(text)
|
old_snap = rec.get("task_snapshot")
|
rec["task_snapshot"] = new_snap
|
if not old_snap:
|
return 0
|
events = diff_task_snapshots(old_snap, new_snap)
|
if not events:
|
return 0
|
bucket = rec.setdefault("tasks", {}).setdefault(d_iso, [])
|
n = 0
|
for kind, key in events:
|
src = new_snap if key in new_snap else old_snap
|
info = src.get(key)
|
if info is None:
|
continue
|
status, raw, tags_str, parent_key = info
|
pinfo = src.get(parent_key)
|
parent_disp = clean_task_text(pinfo[1]) if pinfo else ""
|
if not tags_str and pinfo:
|
tags_str = pinfo[2] # 子任务没标签 → 继承父任务标签
|
if kind == "status":
|
verb = "→ " + STATUS_LABEL.get(status, status)
|
elif kind == "added":
|
verb = "➕ 新建"
|
elif kind == "edited":
|
verb = "✏️ 编辑"
|
else:
|
verb = "🗑️ 删除"
|
bucket.append([tstr, verb, clean_task_text(raw), tags_str, parent_disp])
|
n += 1
|
if len(bucket) > 100:
|
del bucket[:len(bucket) - 100]
|
return n
|
|
|
def build_activity_block(vault, items_by_type, name_counts, task_events=None) -> str:
|
lines = [DIARY_MARK_BEGIN, "## 🕐 活动足迹（自动记录）", ""]
|
if task_events:
|
lines += ["### ✅ 任务动态", ""]
|
shown = sorted(task_events, key=lambda e: e[0], reverse=True)
|
for tstr, verb, text, tags, parent in shown[:TASK_EVENT_LIST_MAX]:
|
line = f"- {tstr} {verb} · {text}"
|
if tags:
|
line += f" `{tags}`"
|
lines.append(line)
|
if parent:
|
lines.append(f"\t- ↳ 父任务：{parent}")
|
if len(shown) > TASK_EVENT_LIST_MAX:
|
lines.append(f"- …其余 {len(shown) - TASK_EVENT_LIST_MAX} 条省略")
|
lines.append("")
|
lines += ["> ✏️ = 编辑过 · 👀 = 只打开/碰过，内容没变", ""]
|
for label, _ in DIARY_ACTIVITY_TYPES:
|
items = items_by_type.get(label)
|
if not items:
|
continue
|
icon = ACTIVITY_ICONS.get(label, "·")
|
lines.append(f"- {icon} **{label}** · {len(items)} 项")
|
edited = sorted([it for it in items if it[2] == "edited"],
|
key=lambda it: it[0], reverse=True)
|
touched = sorted([it for it in items if it[2] != "edited"],
|
key=lambda it: it[0], reverse=True)
|
shown = (edited + touched)[:ACTIVITY_LIST_MAX]
|
for tstr, path, kind in shown:
|
mark = "✏️" if kind == "edited" else "👀"
|
lines.append(f"\t- {tstr} {mark} {link_for(path, vault, name_counts)}")
|
if len(items) > ACTIVITY_LIST_MAX:
|
lines.append(f"\t- …其余 {len(items) - ACTIVITY_LIST_MAX} 项省略")
|
lines += ["", DIARY_MARK_END]
|
return "\n".join(lines)
|
|
|
def update_diaries(vault: str, days_back: int, log=print, known_written=None):
|
"""扫描 vault，把最近 days_back 天内变动过的文件写进对应日期的日记。
|
用内容哈希区分「✏️ 编辑过」和「👀 只碰过」。
|
只替换日记中标记块内的内容，手写内容不受影响。返回更新的日记数。"""
|
if known_written is None:
|
known_written = {}
|
diary_dir = os.path.join(vault, "X0.Diary")
|
today = datetime.date.today()
|
cutoff = today - datetime.timedelta(days=days_back)
|
state = load_diary_state()
|
state_dirty = False
|
|
by_day = collections.defaultdict(lambda: collections.defaultdict(list))
|
name_counts = collections.Counter()
|
for p in iter_tracked_files(vault):
|
if in_diary_dir(p, vault) or is_generated_page(p):
|
continue
|
prev = known_written.get(p)
|
try:
|
st = os.stat(p)
|
except OSError:
|
continue
|
if prev is not None and abs(st.st_mtime - prev) < 0.01:
|
continue # 刚由本工具写出的文件
|
|
rec = state.get(p)
|
d = datetime.date.fromtimestamp(st.st_mtime)
|
tstr = datetime.datetime.fromtimestamp(st.st_mtime).strftime("%H:%M")
|
d_iso = d.isoformat()
|
|
if rec is not None and abs(rec.get("mtime", 0) - st.st_mtime) < 0.01:
|
# mtime 没变：沿用上次分类，零开销
|
current_kind = "edited" if d_iso in rec.get("edited", {}) else "touched"
|
else:
|
# mtime 变了：对比哈希判断是编辑还是只碰过
|
new_hash = file_hash(p)
|
if rec is not None and rec.get("hash") and rec["hash"] == new_hash:
|
current_kind = "touched"
|
else:
|
current_kind = "edited" # 内容真的变了（或首次见到，保守记为编辑）
|
if rec is None:
|
rec = {"edited": {}}
|
if current_kind == "edited":
|
rec["edited"][d_iso] = tstr
|
rec["mtime"] = st.st_mtime
|
rec["hash"] = new_hash
|
state[p] = rec
|
state_dirty = True
|
# 任务动态：md 文件内容变了 → diff 任务快照（首次只见建基线）
|
if current_kind == "edited" and p.endswith(".md"):
|
try:
|
if detect_task_events(p, rec, tstr, d_iso) > 0:
|
state_dirty = True
|
except Exception:
|
pass # 任务解析失败不影响足迹主流程
|
|
label = classify_activity(os.path.basename(p))
|
if not label:
|
continue
|
stem = os.path.basename(p)
|
name_counts[stem[:-3] if stem.endswith(".md") else stem] += 1
|
|
# 该文件的展示条目：历史编辑记录（窗口内的）+ 当前 mtime 日期的事件
|
entries = {}
|
for ds, et in rec.get("edited", {}).items():
|
try:
|
dd = datetime.date.fromisoformat(ds)
|
except ValueError:
|
continue
|
if cutoff <= dd <= today:
|
entries[ds] = (et, "edited")
|
if cutoff <= d <= today and d_iso not in entries:
|
entries[d_iso] = (tstr, current_kind)
|
for ds, (et, kind) in entries.items():
|
by_day[datetime.date.fromisoformat(ds)][label].append((et, p, kind))
|
|
# 任务动态：汇总所有文件在窗口内各天的事件（顺带清理过期桶和已删文件）
|
task_by_day = collections.defaultdict(list)
|
prune_before = (cutoff - datetime.timedelta(days=7)).isoformat()
|
for p, rec in list(state.items()):
|
if should_skip_diary_path(p, vault) or in_diary_dir(p, vault) or is_generated_page(p):
|
continue
|
if not os.path.exists(p):
|
del state[p] # 文件已删：其任务事件也不再展示
|
state_dirty = True
|
continue
|
tasks = rec.get("tasks")
|
if not tasks:
|
continue
|
for ds in list(tasks.keys()):
|
if ds < prune_before:
|
del tasks[ds]
|
state_dirty = True
|
continue
|
try:
|
dd = datetime.date.fromisoformat(ds)
|
except ValueError:
|
continue
|
if cutoff <= dd <= today:
|
task_by_day[dd].extend(tasks[ds])
|
|
if state_dirty:
|
save_diary_state(state)
|
|
changed = 0
|
days = sorted(set(list(by_day.keys()) + list(task_by_day.keys()) + [
|
today - datetime.timedelta(days=i) for i in range(days_back + 1)
|
]), reverse=True)
|
for d in days:
|
if d < cutoff or d > today:
|
continue
|
dpath = os.path.join(diary_dir, diary_rel_name(d))
|
try:
|
with open(dpath, encoding="utf-8") as fh:
|
existing = fh.read()
|
except OSError:
|
existing = None
|
has_block = existing is not None and DIARY_MARK_BEGIN in existing
|
|
types = by_day.get(d)
|
tev = task_by_day.get(d)
|
if not types and not tev:
|
if not has_block:
|
continue # 没内容也没旧块，跳过（不凭空创建空日记）
|
new = re.sub(re.escape(DIARY_MARK_BEGIN) + r".*?" + re.escape(DIARY_MARK_END),
|
"", existing, flags=re.S)
|
new = re.sub(r"\n{3,}", "\n\n", new).rstrip("\n") + "\n"
|
else:
|
block = build_activity_block(vault, types or {}, name_counts,
|
task_events=tev)
|
if existing is None:
|
new = f"# {diary_rel_name(d)[:-3]}\n\n{block}\n"
|
elif has_block:
|
new = re.sub(re.escape(DIARY_MARK_BEGIN) + r".*?" + re.escape(DIARY_MARK_END),
|
lambda _m: block, existing, flags=re.S)
|
else:
|
new = existing.rstrip("\n") + "\n\n" + block + "\n"
|
|
if existing is not None and new == existing:
|
continue
|
os.makedirs(diary_dir, exist_ok=True)
|
with open(dpath, "w", encoding="utf-8") as fh:
|
fh.write(new)
|
known_written[dpath] = os.stat(dpath).st_mtime
|
changed += 1
|
if log:
|
log(f" 日记 {diary_rel_name(d)}: 活动足迹已更新")
|
return changed
|
|
|
def canvas_signature(vault, rel_dir=CANVAS_DIR):
|
"""00index 目录结构指纹：所有 md 文件与文件夹的相对路径有序元组。
|
文件改名不改变 mtime（全局指纹看不见），但会改变路径集合，
|
因此画布更新必须靠这个独立指纹触发。"""
|
base = os.path.join(vault, rel_dir)
|
if not os.path.isdir(base):
|
return None
|
items = []
|
for root, dirs, files in os.walk(base):
|
dirs[:] = [d for d in dirs
|
if d not in SKIP_DIRS and not d.startswith(".")
|
and d not in CANVAS_SKIP_DIRS]
|
for d in dirs:
|
items.append("d:" + os.path.relpath(os.path.join(root, d), base))
|
for f in files:
|
if f.startswith(".") or f == ".DS_Store":
|
continue
|
if f.endswith(".md"):
|
items.append("f:" + os.path.relpath(os.path.join(root, f), base))
|
return tuple(sorted(items))
|
|
|
def generate_canvas_index(vault, rel_dir=CANVAS_DIR, canvas_name=CANVAS_NAME):
|
"""为 rel_dir 目录生成向右展开的脑图画布。
|
|
当前 foldernote 由方案3的 00index/00index.md 承担；watch 模式默认不会调用此函数。
|
|
结构：根节点 → 文件夹节点 → 文件节点（file 类型，可点击打开）。
|
触发规则：文件增/删/改名才改变画布；文件内容修改不影响画布内容，
|
内容哈希门控保证无变化时不写盘。返回 (status, canvas_path)。
|
|
颜色策略（重要）：
|
- 生成器**不再自动给任何节点/连线分配颜色**（旧逻辑用 CANVAS_COLORS
|
把每个顶层分支染成红/绿/蓝……，会覆盖用户手动设置的颜色，已停用）。
|
- 改为「保留旧画布里已有的、非默认的颜色」：节点 id 由路径确定性生成，
|
重画时同一文件 id 不变，因此可从旧画布继承用户手动设定的 color；
|
**新出现的节点不带颜色**，由用户自己上色。
|
- 旧版遗留的默认色 "1"（浅蓝，曾由 CANVAS_COLORS 自动盖在每个节点上）
|
视为「无色」，重画时不再继承——历史画布上被自动涂成 "1" 的节点会
|
自然变回无色；而用户**手动**选过的其它颜色（如 "2"~"6" 或十六进制）
|
一律保留，生成器绝不改动。
|
- 画布生成时排除 CANVAS_SKIP_DIRS。
|
"""
|
base = os.path.join(vault, rel_dir)
|
if not os.path.isdir(base):
|
return "missing", None
|
|
# ---- 1. 建目录树（同层：文件夹在前、文件在后，各按名称排序）----
|
def build(abs_dir):
|
children = []
|
try:
|
entries = sorted(os.listdir(abs_dir))
|
except OSError:
|
return children
|
for e in entries:
|
if e.startswith(".") or e == ".DS_Store":
|
continue
|
full = os.path.join(abs_dir, e)
|
if os.path.isdir(full):
|
if e in SKIP_DIRS or e in CANVAS_SKIP_DIRS:
|
continue
|
children.append({"dir": e, "children": build(full)})
|
elif e.endswith(".md"):
|
children.append({"file": e[:-3]})
|
return children
|
|
tree = {"dir": os.path.basename(rel_dir.rstrip("/")), "children": build(base)}
|
|
# ---- 2. 布局：子树高度自底向上，列宽取每层最大节点宽 ----
|
def node_size(n):
|
return (CANVAS_FILE_W, CANVAS_FILE_H) if "file" in n \
|
else (CANVAS_FOLDER_W, CANVAS_FOLDER_H)
|
|
def subtree_h(n):
|
if not n.get("children"):
|
return node_size(n)[1]
|
total = sum(subtree_h(c) for c in n["children"]) \
|
+ CANVAS_V_GAP * (len(n["children"]) - 1)
|
return max(total, node_size(n)[1])
|
|
col_w = {}
|
|
def measure(n, depth):
|
w = node_size(n)[0]
|
if w > col_w.get(depth, 0):
|
col_w[depth] = w
|
for c in n.get("children", []):
|
measure(c, depth + 1)
|
|
measure(tree, 0)
|
xs = [0] * (max(col_w) + 1)
|
for d in range(1, len(xs)):
|
xs[d] = xs[d - 1] + col_w[d - 1] + CANVAS_H_GAP
|
|
def node_id(s):
|
return hashlib.md5(s.encode("utf-8")).hexdigest()[:16]
|
|
nodes, edges = [], []
|
|
# ---- 3. 读取旧画布里已有的颜色，按 id 匹配后保留（仅保留、不再自动上色）----
|
out_path = os.path.join(base, canvas_name)
|
old_node_color, old_edge_color = {}, {}
|
try:
|
with open(out_path, encoding="utf-8") as fh:
|
old = json.load(fh)
|
for nd in old.get("nodes", []):
|
c = nd.get("color")
|
# 跳过旧版生成器遗留的默认色 "1"（视作无色，不继承），其余颜色（用户手动设的）一律保留
|
if c is not None and c != CANVAS_DEFAULT_COLOR:
|
old_node_color[nd["id"]] = c
|
for ed in old.get("edges", []):
|
c = ed.get("color")
|
if c is not None and c != CANVAS_DEFAULT_COLOR:
|
old_edge_color[ed["id"]] = c
|
except (OSError, ValueError):
|
pass
|
|
# ---- 4. 放置节点：父节点在其子树跨度内垂直居中 ----
|
def place(n, depth, top, segs):
|
# segs = 该节点祖先在 rel_dir 之下的路径段
|
h = subtree_h(n)
|
w, own_h = node_size(n)
|
y = int(round(top + (h - own_h) / 2.0))
|
if "file" in n:
|
own = segs + [n["file"]]
|
nid = node_id("f:" + "/".join(own))
|
rel = "/".join([rel_dir.rstrip("/")] + own)
|
# 用文字节点+wikilink 代替 file 节点：file 节点会渲染笔记预览
|
# （满屏"笔记属性"），文字节点直接显示笔记名，点击同样可打开
|
if "[[" in n["file"] or "]]" in n["file"]:
|
text = n["file"] # 极端文件名兜底
|
else:
|
text = "[[" + rel + "|" + n["file"] + "]]"
|
node = {"id": nid, "type": "text", "text": text,
|
"x": xs[depth], "y": y, "width": w, "height": own_h}
|
if nid in old_node_color: # 仅保留旧画布里手动设的颜色
|
node["color"] = old_node_color[nid]
|
nodes.append(node)
|
return nid
|
# 根节点 = rel_dir 本身，路径段为空；其余目录节点含自身名字
|
own = [] if depth == 0 else segs + [n["dir"]]
|
nid = node_id("root") if depth == 0 else node_id("d:" + "/".join(own))
|
label = ("🗂 " if depth == 0 else "📁 ") + n["dir"]
|
node = {"id": nid, "type": "text", "text": label,
|
"x": xs[depth], "y": y, "width": w, "height": own_h}
|
if nid in old_node_color:
|
node["color"] = old_node_color[nid]
|
nodes.append(node)
|
kids = n.get("children", [])
|
total_h = sum(subtree_h(c) for c in kids) + CANVAS_V_GAP * (len(kids) - 1) \
|
if kids else 0
|
child_top = top + (h - total_h) / 2.0
|
for c in kids:
|
cid = place(c, depth + 1, child_top, own)
|
eid = node_id("e:" + nid + cid)
|
edge = {"id": eid, "fromNode": nid, "fromSide": "right",
|
"toNode": cid, "toSide": "left"}
|
if eid in old_edge_color:
|
edge["color"] = old_edge_color[eid]
|
edges.append(edge)
|
child_top += subtree_h(c) + CANVAS_V_GAP
|
return nid
|
|
place(tree, 0, 0, [])
|
|
# ---- 5. 内容哈希门控：无变化不写盘 ----
|
content = json.dumps({"nodes": nodes, "edges": edges},
|
ensure_ascii=False, indent="\t") + "\n"
|
try:
|
with open(out_path, encoding="utf-8") as fh:
|
if fh.read() == content:
|
return "unchanged", out_path
|
except OSError:
|
pass
|
with open(out_path, "w", encoding="utf-8") as fh:
|
fh.write(content)
|
return "written", out_path
|
|
|
def cmd_watch(vaults, interval: float):
|
def log(msg):
|
print(f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)
|
|
log("tagfeed watch 启动，监听: " + ", ".join(vaults))
|
known_written = {} # 输出页路径 -> 我方写入时的 mtime（防止自我触发）
|
last_max = {} # vault -> 已处理的内容指纹
|
baseline = set()
|
scan_cache = {} # (path, tag) -> (mtime_ns, entries)，增量缓存
|
canvas_sigs = {} # vault -> 00index 结构指纹（改名不改 mtime，须独立轮询）
|
|
def cached_scan(path, rel, tag, rx):
|
"""按文件 mtime 缓存解析结果：没改过的文件不重新读盘/跑正则"""
|
try:
|
mt = os.stat(path).st_mtime_ns
|
except OSError:
|
return []
|
key = (path, tag)
|
hit = scan_cache.get(key)
|
if hit is not None and hit[0] == mt:
|
return hit[1]
|
es, _ = scan_file(path, rel, tag, rx)
|
scan_cache[key] = (mt, es)
|
return es
|
|
def content_fingerprint(vault):
|
"""全库内容文件的指纹 (文件数, mtime 总和)。
|
新增/修改/删除文件都会改变指纹。tagfeed 生成页不计入。
|
覆盖所有被追踪的类型（md/白板/表格/文档），日记更新也靠它触发。"""
|
count, total = 0, 0
|
for p in iter_tracked_files(vault):
|
try:
|
st = os.stat(p)
|
except OSError:
|
continue
|
prev = known_written.get(p)
|
if prev is not None and abs(st.st_mtime - prev) < 0.01:
|
continue # tagfeed 刚写出的页面，不算内容变更
|
if os.path.basename(p).startswith("tag-"):
|
try:
|
with open(p, encoding="utf-8", errors="replace") as fh:
|
if MARKER in fh.read(600):
|
known_written[p] = st.st_mtime
|
continue # 别处生成的 tagfeed 页，同样跳过
|
except OSError:
|
pass
|
count += 1
|
total += int(st.st_mtime_ns // 1000000) # 毫秒粒度，避免浮点误差
|
return (count, total)
|
|
def refresh(vault):
|
subs = discover_subscribed(vault)
|
n = 0
|
for tag, out_path in subs.items():
|
out_rel_dir = os.path.relpath(os.path.dirname(out_path), vault)
|
try:
|
status, path, _ = generate_tag(vault, tag, out_rel_dir,
|
allow_empty=True, scan_fn=cached_scan)
|
except Exception as ex:
|
log(f" 更新 #{tag} 失败: {ex}")
|
continue
|
if status == "written":
|
known_written[path] = os.stat(path).st_mtime
|
n += 1
|
# 每日足迹：把变动的文件写进对应日期的日记（只对有日记目录的 vault）
|
if os.path.isdir(os.path.join(vault, "X0.Diary")):
|
try:
|
dn = update_diaries(vault, DIARY_BACKFILL_DAYS, log=log,
|
known_written=known_written)
|
except Exception as ex:
|
log(f" 日记足迹更新失败: {ex}")
|
dn = 0
|
else:
|
dn = 0
|
# 方案3：Excalidraw 版 tagfeed 总入口，当前 00index 的唯一 foldernote。
|
# 只维护 Frame0-all 内的标签卡片，尽量保留用户手工整理后的位置和样式。
|
try:
|
excalidraw_menu_py = os.path.join(vault, "X2.Archived", "scripts", "sync_excalidraw_tagfeed.py")
|
if os.path.isfile(excalidraw_menu_py):
|
r = subprocess.run([sys.executable, excalidraw_menu_py, vault],
|
capture_output=True, text=True, timeout=180)
|
if r.returncode != 0:
|
log(f" Excalidraw-Tagfeed 同步失败: {(r.stderr or '').strip()[-200:]}")
|
else:
|
out = (r.stdout or "").strip()
|
m = re.search(r"Excalidraw-Tagfeed:\s+(\w+)\s+total_tags=(\d+)\s+inserted=(\d+)\s+removed=(\d+)\s+path=(.+)", out)
|
if m:
|
excalidraw_menu_path = m.group(5).strip()
|
if os.path.exists(excalidraw_menu_path):
|
known_written[excalidraw_menu_path] = os.stat(excalidraw_menu_path).st_mtime
|
if m.group(1) == "written":
|
log(f" Excalidraw-Tagfeed 已更新: 补入 {m.group(3)} 个标签，移除 {m.group(4)} 个重复/过期标签")
|
elif out:
|
log(f" Excalidraw-Tagfeed 输出: {out[-200:]}")
|
except Exception as ex:
|
log(f" Excalidraw-Tagfeed 同步异常: {ex}")
|
return n, dn
|
|
while True:
|
for vault in vaults:
|
if not os.path.isdir(vault):
|
continue
|
name = os.path.basename(vault)
|
if CANVAS_INDEX_ENABLED:
|
try:
|
sig = canvas_signature(vault)
|
except Exception:
|
sig = None
|
if sig is not None and sig != canvas_sigs.get(vault):
|
canvas_sigs[vault] = sig
|
try:
|
cstat, cpath = generate_canvas_index(vault)
|
if cstat == "written":
|
known_written[cpath] = os.stat(cpath).st_mtime
|
log(f"{name}: 目录脑图画布已更新 "
|
+ os.path.relpath(cpath, vault))
|
except Exception as ex:
|
log(f"{name}: 目录脑图画布更新失败: {ex}")
|
try:
|
fp = content_fingerprint(vault)
|
except Exception as ex:
|
log(f"扫描出错 {vault}: {ex}")
|
continue
|
if vault not in baseline:
|
baseline.add(vault)
|
n, dn = refresh(vault)
|
last_max[vault] = fp
|
log(f"{name}: 首次追赶完成，更新 {n} 个聚合页、{dn} 个日记")
|
continue
|
if fp != last_max.get(vault):
|
time.sleep(1.5) # 防抖：等保存稳定
|
fp2 = content_fingerprint(vault)
|
n, dn = refresh(vault)
|
last_max[vault] = fp2
|
log(f"{name}: 检测到变更，更新 {n} 个聚合页、{dn} 个日记")
|
time.sleep(interval)
|
|
|
def main():
|
ap = argparse.ArgumentParser(description="Logseq 风格 Obsidian tagfeed 页生成器")
|
ap.add_argument("tag", nargs="?", help="标签名（带不带 # 都行）")
|
ap.add_argument("--vault", action="append",
|
help=f"vault 路径，可多次指定（默认 {DEFAULT_VAULT}；watch 模式默认两个 vault）")
|
ap.add_argument("--out", default="00index", help="输出目录，相对 vault（默认 00index）")
|
ap.add_argument("--title", default=None, help="自定义页面标题（默认 # 🏷 <标签>）")
|
ap.add_argument("--open", action="store_true", help="生成后在 Obsidian 中打开")
|
ap.add_argument("--tags", action="store_true", help="列出全库标签统计后退出")
|
ap.add_argument("--watch", action="store_true",
|
help="常驻监听：vault 有变更时自动刷新所有已生成的聚合页和日记足迹")
|
ap.add_argument("--diary", action="store_true",
|
help=f"一次性回填最近 {DIARY_BACKFILL_DAYS} 天的日记活动足迹后退出")
|
ap.add_argument("--days", type=int, default=DIARY_BACKFILL_DAYS,
|
help=f"--diary 回填的天数（默认 {DIARY_BACKFILL_DAYS}）")
|
ap.add_argument("--interval", type=float, default=3.0,
|
help="watch 模式轮询间隔秒数（默认 3）")
|
args = ap.parse_args()
|
|
if args.watch:
|
scheme3_watch = os.path.join(os.path.dirname(__file__), "excalidraw_tagfeed_watch.py")
|
os.execv(sys.executable, [sys.executable, scheme3_watch])
|
return
|
|
vault = os.path.abspath(os.path.expanduser(args.vault[0] if args.vault else DEFAULT_VAULT))
|
if not os.path.isdir(vault):
|
sys.exit(f"错误：vault 不存在：{vault}")
|
|
if args.tags:
|
cmd_tags(vault)
|
return
|
if args.diary:
|
n = update_diaries(vault, args.days)
|
print(f"✅ 日记活动足迹：更新了 {n} 篇日记（回填最近 {args.days} 天）")
|
return
|
if not args.tag:
|
ap.error("请提供标签名，例如：python3 tagfeed.py bobo")
|
|
tag = args.tag.lstrip("#").strip()
|
if not tag:
|
sys.exit("错误：标签为空")
|
|
status, out_path, (blocks, notes, days) = generate_tag(
|
vault, tag, args.out, args.title)
|
if status == "empty":
|
print(f"未找到 #{tag} 的任何内容。（用 --tags 查看全库标签列表）")
|
return
|
|
print(f"✅ #{tag}: {blocks} 个块 / {notes} 篇笔记 / {days} 天")
|
print(f" → {os.path.relpath(out_path, vault)}" +
|
("（内容无变化，未写盘）" if status == "unchanged" else ""))
|
|
if args.open:
|
vault_name = os.path.basename(vault)
|
out_rel = os.path.relpath(out_path, vault)
|
uri = ("obsidian://open?vault=" + urllib.parse.quote(vault_name)
|
+ "&file=" + urllib.parse.quote(os.path.splitext(out_rel)[0]))
|
subprocess.Popen(["open", uri])
|
|
|
if __name__ == "__main__":
|
main()
|
