#!/usr/bin/env python3
|
# -*- coding: utf-8 -*-
|
"""ob02 daily activity generator.
|
|
This script owns the daily-activity block in X0.Diary/YYYY-MM-DD.md.
|
It is intentionally independent from the tagfeed workflow: tagfeed should not
|
watch the vault or write daily notes.
|
"""
|
|
import argparse
|
import collections
|
import datetime
|
import difflib
|
import hashlib
|
import json
|
import os
|
import re
|
|
|
DEFAULT_VAULT = os.environ.get(
|
"DAILY_ACTIVITY_VAULT", os.path.expanduser("~/Downloads/Syn/Ob/Ob.temp")
|
)
|
SKIP_DIRS = {".obsidian", ".git", ".trash", "X.Attachment", ".hg", ".svn"}
|
TAGFEED_MARKER = "<!-- tagfeed"
|
DIARY_MARK_BEGIN = "<!-- daily-activity:begin (ob02 自动生成，勿手改此块内) -->"
|
DIARY_MARK_END = "<!-- daily-activity:end -->"
|
DIARY_BLOCK_RE = re.compile(
|
r"<!--\s*daily-activity:begin[^>]*-->.*?<!--\s*daily-activity:end\s*-->",
|
re.S,
|
)
|
DIARY_BACKFILL_DAYS = 7
|
ACTIVITY_LIST_MAX = 30
|
DIARY_SKIP_DIRS = {"X.Attachment", "attachments", "Assets", "assets"}
|
DIARY_SKIP_REL_PREFIXES = ("00index", "X0.Diary", "X1.Knomo", "X2.Archived", "02DS/01dril-book")
|
DIARY_COPILOT_ROOT_REL = "02DS/02copilot"
|
DIARY_COPILOT_ALLOWED_REL = "02DS/02copilot/copilot-conversations"
|
TRACKED_SUFFIXES = (".md", ".markdown", ".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")),
|
]
|
ACTIVITY_ICONS = {"白板": "🖼", "表格": "📊", "文档": "📄", "笔记": "📝"}
|
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(
|
os.environ.get("DAILY_ACTIVITY_STATE_PATH", "~/.tagfeed-diary-state.json")
|
)
|
HASH_CAP_BYTES = 8 * 1024 * 1024
|
|
|
def rel_is_or_under(rel: str, prefix: str) -> bool:
|
return rel == prefix or rel.startswith(prefix + "/")
|
|
|
def indent_of(line: str) -> int:
|
expanded = line.expandtabs(2)
|
return len(expanded) - len(expanded.lstrip(" "))
|
|
|
def diary_rel_name(d: datetime.date) -> str:
|
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):
|
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
|
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:
|
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 TAGFEED_MARKER in fh.read(600)
|
except OSError:
|
return False
|
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):
|
m = re.search(r"^tags:\s*\n((?:[ \t]+-[ \t]+.*\n?)+)", block, re.M)
|
if m and re.search(r"tagfeed|tagfeed-live", m.group(1)):
|
return True
|
return False
|
|
|
def link_for(path: str, vault: str, name_counts) -> str:
|
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:
|
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):
|
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:
|
tasks = {}
|
stack = []
|
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):
|
pairs = []
|
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:
|
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):
|
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):
|
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:
|
current_kind = "edited" if d_iso in rec.get("edited", {}) else "touched"
|
else:
|
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
|
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
|
|
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))
|
read_mtime_ns = None
|
try:
|
read_mtime_ns = os.stat(dpath).st_mtime_ns
|
with open(dpath, encoding="utf-8") as fh:
|
existing = fh.read()
|
except OSError:
|
existing = None
|
has_block = existing is not None and DIARY_BLOCK_RE.search(existing) is not None
|
|
types = by_day.get(d)
|
tev = task_by_day.get(d)
|
if not types and not tev:
|
if not has_block:
|
continue
|
new = DIARY_BLOCK_RE.sub("", existing)
|
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 = DIARY_BLOCK_RE.sub(lambda _m: block, existing)
|
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)
|
if read_mtime_ns is not None:
|
try:
|
current_mtime_ns = os.stat(dpath).st_mtime_ns
|
except OSError:
|
current_mtime_ns = None
|
if current_mtime_ns is not None and current_mtime_ns != read_mtime_ns:
|
if log:
|
log(f" 日记 {diary_rel_name(d)}: 跳过活动足迹更新（文件正在被其他程序改写）")
|
continue
|
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 main():
|
ap = argparse.ArgumentParser(description="ob02 每日活动足迹生成器")
|
ap.add_argument("--vault", default=DEFAULT_VAULT, help=f"vault 路径（默认 {DEFAULT_VAULT}）")
|
ap.add_argument("--days", type=int, default=DIARY_BACKFILL_DAYS,
|
help=f"回填最近 N 天（默认 {DIARY_BACKFILL_DAYS}）")
|
ap.add_argument("--today", action="store_true", help="只更新今天")
|
args = ap.parse_args()
|
|
vault = os.path.abspath(os.path.expanduser(args.vault))
|
if not os.path.isdir(vault):
|
raise SystemExit(f"错误：vault 不存在：{vault}")
|
days = 0 if args.today else max(0, args.days)
|
n = update_diaries(vault, days)
|
scope = "只检查今天" if args.today else f"回填最近 {days} 天"
|
print(f"✅ 每日活动足迹：更新了 {n} 篇日记（{scope}）")
|
|
|
if __name__ == "__main__":
|
main()
|
