#!/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
|
|
注意：
|
--watch 已退役。tagfeed 当前只通过 Obsidian 命令“Tagfeed：手动更新全局Tagfeed”触发全局同步。
|
每日活动足迹已剥离到 daily_activity.py，不再由 tagfeed 负责。
|
|
输出文件： <vault>/<out>/tag-<标签>.md
|
重复运行即刷新（覆盖旧文件）。
|
"""
|
|
import argparse
|
import collections
|
import datetime
|
import hashlib
|
import json
|
import os
|
import re
|
import subprocess
|
import sys
|
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"),
|
]
|
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 页面目录
|
|
|
# ---------------------------------------------------------------- 工具函数
|
|
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
|
|
|
EXCALIDRAW_BLOCK_REF_RE = re.compile(r"\s+\^[A-Za-z0-9_-]+\s*$")
|
|
|
def extract_frontmatter_prefix(raw: str) -> str:
|
match = re.match(r"^---\s*\n.*?\n(?:---|\.\.\.)\s*\n", raw, flags=re.S)
|
return match.group(0).rstrip() if match else ""
|
|
|
def extract_excalidraw_text_elements(raw: str) -> str:
|
"""只抽取 Excalidraw 的人类可读 Text Elements，跳过 Drawing JSON 和附件区。"""
|
lines = raw.splitlines()
|
start = None
|
for idx, line in enumerate(lines):
|
if re.match(r"^\s*##\s+Text Elements\s*$", line, flags=re.I):
|
start = idx + 1
|
break
|
if start is None:
|
return ""
|
|
chunks = []
|
current = []
|
|
def flush():
|
nonlocal current
|
text = " ".join(part.strip() for part in current if part.strip())
|
text = re.sub(r"[ \t]{2,}", " ", text).strip()
|
if text:
|
chunks.append(text)
|
current = []
|
|
for line in lines[start:]:
|
stripped = line.strip()
|
if stripped == "%%" or re.match(r"^\s*##\s+(?:Embedded Files|Drawing)\s*$", line, flags=re.I):
|
break
|
if not stripped:
|
if current:
|
current.append("")
|
continue
|
had_ref = EXCALIDRAW_BLOCK_REF_RE.search(line) is not None
|
cleaned = EXCALIDRAW_BLOCK_REF_RE.sub("", line).strip()
|
if cleaned:
|
current.append(cleaned)
|
if had_ref:
|
flush()
|
flush()
|
return "\n\n".join(chunks)
|
|
|
def readable_excalidraw_markdown(raw: str) -> str:
|
frontmatter = extract_frontmatter_prefix(raw)
|
text_elements = extract_excalidraw_text_elements(raw)
|
parts = [part for part in (frontmatter, text_elements) if part]
|
return "\n\n".join(parts) + ("\n" if parts else "")
|
|
|
def 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"):
|
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 os.path.basename(path).endswith(".excalidraw.md"):
|
text = readable_excalidraw_markdown(text)
|
|
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
|
|
def collect_parent_list_block(i):
|
"""Return the parent list block when a tag is on one of its indented text lines."""
|
parent_idx = None
|
parent_indent = 0
|
for k in range(i - 1, body_start - 1, -1):
|
prev = lines[k]
|
if not prev.strip() or HEADING_RE.match(prev) or FENCE_RE.match(prev):
|
break
|
if LIST_ITEM_RE.match(prev):
|
parent_idx = k
|
parent_indent = indent_of(prev)
|
break
|
if parent_idx is None or indent_of(lines[i]) <= parent_indent:
|
return None
|
|
block = []
|
for j in range(parent_idx, i + 1):
|
current = lines[j]
|
if not current.strip():
|
continue
|
cm = LIST_ITEM_RE.match(current)
|
if cm:
|
current_indent = indent_of(current)
|
if j != parent_idx and current_indent <= parent_indent:
|
return None
|
text = strip_tag_from_text(clean_for_match(cm.group(3) or ""), rx) or (cm.group(3) or "").strip()
|
level = 0 if j == parent_idx else max(1, (current_indent - parent_indent) // 2)
|
else:
|
text = strip_tag_from_text(clean_for_match(current), rx)
|
if not text:
|
continue
|
level = max(1, (indent_of(current) - parent_indent) // 2)
|
block.append((level, text))
|
return (parent_idx, block) if block else None
|
|
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
|
|
# 标签在缩进续行中时，归属到最近的父列表项。
|
parent_block = collect_parent_list_block(i)
|
if parent_block:
|
parent_idx, block = parent_block
|
add_lines_block(parent_idx, block, parent_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 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 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}）")
|
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="已退役：tagfeed 不再使用系统 watcher，请用 Obsidian 手动命令同步")
|
ap.add_argument("--diary", action="store_true",
|
help="已剥离：每日活动足迹请使用 daily_activity.py")
|
ap.add_argument("--days", type=int, default=7,
|
help="兼容旧参数；tagfeed 不再处理每日活动足迹")
|
ap.add_argument("--interval", type=float, default=3.0,
|
help="兼容旧参数；tagfeed watcher 已退役")
|
args = ap.parse_args()
|
|
if args.watch:
|
print("Tagfeed watcher 已退役；请在 Obsidian Command+P 执行：Tagfeed：手动更新全局Tagfeed")
|
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:
|
print("每日活动足迹已从 tagfeed 剥离；请运行：python3 /Users/ar/bin/daily_activity.py --days N")
|
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()
|
