tangchunwu Dayflow 数据同步
将 Dayflow 应用的数据库内容增量同步到 Obsidian。
同步内容
- 时间线卡片 (timeline_cards) -> Dayflow活动日志/
- 日志总结 (journal_entries) -> Dayflow每日报告/
- 观察记录 (observations) -> Dayflow提醒建议/
执行步骤
使用 Bash 工具执行以下 Python 脚本：
python3 << 'EOF'
import os
import sqlite3
from datetime import datetime
def to_datetime(ts):
if ts is None:
return None
value = int(ts)
if value > 1_000_000_000_000:
value = value / 1000
return datetime.fromtimestamp(value)
def write_if_changed(filepath, content):
old_content = None
if os.path.exists(filepath):
with open(filepath, "r", encoding="utf-8") as f:
old_content = f.read()
if old_content != content:
with open(filepath, "w", encoding="utf-8") as f:
f.write(content)
return True
return False
db_path = "/Users/tangchunwu/Library/Application Support/Dayflow/chunks.sqlite"
obsidian_base = "/Users/tangchunwu/Documents/Obsidian Vault"
activity_dir = os.path.join(obsidian_base, "Dayflow活动日志")
report_dir = os.path.join(obsidian_base, "Dayflow每日报告")
tips_dir = os.path.join(obsidian_base, "Dayflow提醒建议")
os.makedirs(activity_dir, exist_ok=True)
os.makedirs(report_dir, exist_ok=True)
os.makedirs(tips_dir, exist_ok=True)
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# 1) 同步活动日志（timeline_cards）
cursor.execute("SELECT DISTINCT day FROM timeline_cards WHERE is_deleted = 0 ORDER BY day")
days = [row[0] for row in cursor.fetchall()]
activity_updated = 0
for day in days:
cursor.execute(
"""
SELECT start, end, title, summary, detailed_summary, category, subcategory
FROM timeline_cards
WHERE day = ? AND is_deleted = 0
ORDER BY id
""",
(day,),
)
rows = cursor.fetchall()
if not rows:
continue
lines = [f"# {day} Dayflow 活动日志", ""]
for start, end, title, summary, detailed_summary, category, subcategory in rows:
title = (title or "(无标题)").strip()
start = (start or "--").strip()
end = (end or "--").strip()
summary = (summary or "").strip()
detailed_summary = (detailed_summary or "").strip()
category = (category or "").strip()
subcategory = (subcategory or "").strip()
lines.append(f"## {start} - {end}")
lines.append(f"### {title}")
lines.append("")
if category:
tag_line = f"**分类**: {category}"
if subcategory:
tag_line += f" / {subcategory}"
lines.append(tag_line)
lines.append("")
if summary:
lines.append(summary)
lines.append("")
if detailed_summary:
lines.append("**详细总结**")
lines.append("")
lines.append(detailed_summary)
lines.append("")
lines.append("---")
lines.append("")
filepath = os.path.join(activity_dir, f"{day}.md")
content = "\n".join(lines).rstrip() + "\n"
if write_if_changed(filepath, content):
activity_updated += 1
print(f"活动日志: 更新 {activity_updated} 天")
# 2) 同步每日报告
# 优先使用 journal_entries；若无数据则由 timeline_cards 自动聚合生成日报
cursor.execute(
"""
SELECT day, intentions, goals, notes, reflections, summary, status
FROM journal_entries
ORDER BY day
"""
)
reports = cursor.fetchall()
report_updated = 0
if reports:
for day, intentions, goals, notes, reflections, summary, status in reports:
filepath = os.path.join(report_dir, f"{day}.md")
blocks = []
if status:
blocks.append(("状态", str(status).strip()))
if intentions:
blocks.append(("今日意图", intentions.strip()))
if goals:
blocks.append(("目标", goals.strip()))
if summary:
blocks.append(("总结", summary.strip()))
if reflections:
blocks.append(("反思", reflections.strip()))
if notes:
blocks.append(("补充笔记", notes.strip()))
if not blocks:
continue
lines = [f"# {day} Dayflow 每日报告", ""]
for title, content in blocks:
if not content:
continue
lines.append(f"## {title}")
lines.append("")
lines.append(content)
lines.append("")
new_content = "\n".join(lines).rstrip() + "\n"
if write_if_changed(filepath, new_content):
report_updated += 1
else:
cursor.execute(
"""
SELECT day, start, end, start_ts, end_ts, title, summary, detailed_summary, category
FROM timeline_cards
WHERE is_deleted = 0
ORDER BY day, id
"""
)
rows = cursor.fetchall()
cards_by_day = {}
for day, start, end, start_ts, end_ts, title, summary, detailed_summary, category in rows:
cards_by_day.setdefault(day, []).append(
{
"start": (start or "--").strip(),
"end": (end or "--").strip(),
"start_ts": start_ts,
"end_ts": end_ts,
"title": (title or "(无标题)").strip(),
"summary": (summary or "").strip(),
"detailed": (detailed_summary or "").strip(),
"category": (category or "Unknown").strip() or "Unknown",
}
)
for day, cards in sorted(cards_by_day.items()):
if not cards:
continue
total = len(cards)
category_counts = {}
for c in cards:
category_counts[c["category"]] = category_counts.get(c["category"], 0) + 1
top_category, top_count = sorted(category_counts.items(), key=lambda x: (-x[1], x[0]))[0]
start_candidates = [to_datetime(c["start_ts"]) for c in cards if c["start_ts"] is not None]
end_candidates = [to_datetime(c["end_ts"]) for c in cards if c["end_ts"] is not None]
start_candidates = [x for x in start_candidates if x is not None]
end_candidates = [x for x in end_candidates if x is not None]
if start_candidates and end_candidates:
first_time = min(start_candidates).strftime("%H:%M")
last_time = max(end_candidates).strftime("%H:%M")
else:
first_time = cards[0]["start"]
last_time = cards[-1]["end"]
def card_score(c):
duration = 0
if c["start_ts"] is not None and c["end_ts"] is not None:
duration = max(0, int(c["end_ts"]) - int(c["start_ts"]))
text_score = len(c["summary"]) + len(c["detailed"])
return (duration, text_score)
top_cards = sorted(cards, key=card_score, reverse=True)[:5]
summary_parts = []
for c in top_cards:
text = c["summary"] or c["detailed"]
if text:
summary_parts.append(text)
if len(summary_parts) >= 3:
break
if summary_parts:
if len(one_liner) > 240:
+ "..."
else:
lines = [f"# {day} Dayflow 每日报告", ""]
lines.append("## 今日概览")
lines.append("")
lines.append(f"- 卡片数: {total}")
lines.append(f"- 活跃时段: {first_time} - {last_time}")
lines.append(f"- 主分类: {top_category} ({top_count}/{total})")
lines.append("")
lines.append("## 今日重点")
lines.append("")
for idx, c in enumerate(top_cards, start=1):
key_text = c["summary"] or c["detailed"] or "(无摘要)"
if len(key_text) > 120:
key_text = key_text[:120].rstrip() + "..."
lines.append(f"{idx}. {c['start']} - {c['end']} | {c['title']}")
lines.append(f" - {key_text}")
lines.append("")
lines.append("## 分类统计")
lines.append("")
for k, v in sorted(category_counts.items(), key=lambda x: (-x[1], x[0])):
pct = (v * 100.0) / total
lines.append(f"- {k}: {v} ({pct:.1f}%)")
lines.append("")
lines.append("## 一句总结")
lines.append("")
lines.append(one_liner)
lines.append("")
filepath = os.path.join(report_dir, f"{day}.md")
new_content = "\n".join(lines).rstrip() + "\n"
if write_if_changed(filepath, new_content):
report_updated += 1
print(f"每日报告: 更新 {report_updated} 天")
# 3) 同步提醒建议（observations）
cursor.execute(
"""
SELECT start_ts, end_ts, observation
FROM observations
ORDER BY start_ts
"""
)
obs_rows = cursor.fetchall()
obs_by_day = {}
for start_ts, end_ts, observation in obs_rows:
start_dt = to_datetime(start_ts)
end_dt = to_datetime(end_ts)
if start_dt is None:
continue
day = start_dt.strftime("%Y-%m-%d")
obs_by_day.setdefault(day, []).append((start_dt, end_dt, (observation or "").strip()))
tips_updated = 0
for day, items in sorted(obs_by_day.items()):
lines = [f"# {day} Dayflow 提醒建议", ""]
for start_dt, end_dt, text in items:
start_text = start_dt.strftime("%H:%M:%S")
end_text = end_dt.strftime("%H:%M:%S") if end_dt else "--:--:--"
lines.append(f"## {start_text} - {end_text}")
lines.append("")
lines.append(text if text else "(空)")
lines.append("")
lines.append("---")
lines.append("")
filepath = os.path.join(tips_dir, f"{day}.md")
content = "\n".join(lines).rstrip() + "\n"
if write_if_changed(filepath, content):
tips_updated += 1
print(f"提醒建议: 更新 {tips_updated} 天")
conn.close()
print(f"\n同步完成! {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
EOF
执行完成后向用户报告同步结果。