837c5f66fb
- anchor_writer.py: 记忆写入与情感打标 - anchor_search.py: 多维语义检索引擎 - anchor_decay.py: 遗忘曲线 + 归档 + 随机返场 - anchor_vault_sync.py: Vault 索引生成 - README.md: 项目说明文档
147 lines
5.0 KiB
Python
147 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Anchor Vault Sync & Search
|
||
轻量级 Obsidian Vault 索引与检索脚本
|
||
路径: ~/.hermes/anchor_vault_sync.py
|
||
"""
|
||
import os
|
||
import sys
|
||
import json
|
||
import re
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
VAULT_PATH = "/Users/fyah/Documents/如梦初醒/memory/test"
|
||
INDEX_PATH = os.path.expanduser("~/.hermes/anchor_vault_index.json")
|
||
SKIP_DIRS = {".obsidian", ".trash", ".smart-env", ".space", ".makemd", ".git"}
|
||
PREVIEW_LEN = 300
|
||
|
||
def parse_yaml_frontmatter(content):
|
||
"""极简 YAML 解析，只取 tags 和 title"""
|
||
tags = []
|
||
title = ""
|
||
if content.startswith("---"):
|
||
parts = content.split("---", 2)
|
||
if len(parts) >= 2:
|
||
yaml_block = parts[1]
|
||
for line in yaml_block.splitlines():
|
||
if line.startswith("tags:"):
|
||
# 处理 tags: [a, b] 或 tags:\n - a\n - b
|
||
rest = line[5:].strip()
|
||
if rest.startswith("["):
|
||
tags = [t.strip().strip("'\"") for t in rest[1:-1].split(",")]
|
||
else:
|
||
tags = []
|
||
elif line.startswith(" - "):
|
||
if "tags" in locals() and tags is not None: # 简单判断是否在 tags 块下
|
||
tags.append(line.strip().strip("- ").strip("'\""))
|
||
elif line.startswith("title:"):
|
||
title = line[6:].strip().strip("'\"")
|
||
return tags, title
|
||
|
||
def scan_vault():
|
||
"""扫描 Vault 并生成索引"""
|
||
print(f"[*] 开始扫描: {VAULT_PATH}")
|
||
index = []
|
||
count = 0
|
||
|
||
for root, dirs, files in os.walk(VAULT_PATH):
|
||
# 过滤隐藏目录
|
||
dirs[:] = [d for d in dirs if d not in SKIP_DIRS and not d.startswith(".")]
|
||
|
||
for file in files:
|
||
if not file.endswith(".md"):
|
||
continue
|
||
|
||
filepath = os.path.join(root, file)
|
||
rel_path = os.path.relpath(filepath, VAULT_PATH)
|
||
|
||
try:
|
||
with open(filepath, "r", encoding="utf-8") as f:
|
||
content = f.read()
|
||
|
||
tags, title = parse_yaml_frontmatter(content)
|
||
mtime = os.path.getmtime(filepath)
|
||
mtime_str = datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M")
|
||
|
||
# 提取预览（去掉 YAML 头和空行）
|
||
body = content.split("---", 2)[-1].strip()
|
||
preview = body[:PREVIEW_LEN].replace("\n", " ").strip()
|
||
if len(body) > PREVIEW_LEN:
|
||
preview += "..."
|
||
|
||
index.append({
|
||
"path": rel_path,
|
||
"title": title or Path(file).stem,
|
||
"tags": tags,
|
||
"mtime": mtime_str,
|
||
"preview": preview
|
||
})
|
||
count += 1
|
||
except Exception as e:
|
||
print(f"[!] 读取失败 {rel_path}: {e}")
|
||
|
||
# 保存索引
|
||
with open(INDEX_PATH, "w", encoding="utf-8") as f:
|
||
json.dump(index, f, ensure_ascii=False, indent=2)
|
||
|
||
print(f"[+] 同步完成。共索引 {count} 个文件。")
|
||
print(f"[+] 索引已保存至: {INDEX_PATH}")
|
||
return index
|
||
|
||
def search_vault(query):
|
||
"""基于关键词搜索索引"""
|
||
if not os.path.exists(INDEX_PATH):
|
||
print("[-] 索引不存在，请先运行 sync。")
|
||
return []
|
||
|
||
with open(INDEX_PATH, "r", encoding="utf-8") as f:
|
||
index = json.load(f)
|
||
|
||
results = []
|
||
q_lower = query.lower()
|
||
|
||
for item in index:
|
||
# 匹配标题、标签、预览
|
||
score = 0
|
||
if q_lower in item["title"].lower(): score += 3
|
||
if any(q_lower in t.lower() for t in item["tags"]): score += 2
|
||
if q_lower in item["preview"].lower(): score += 1
|
||
|
||
if score > 0:
|
||
item["score"] = score
|
||
results.append(item)
|
||
|
||
# 按分数排序
|
||
results.sort(key=lambda x: x["score"], reverse=True)
|
||
|
||
if not results:
|
||
print(f"[-] 未找到与 '{query}' 相关的内容。")
|
||
return []
|
||
|
||
print(f"[+] 找到 {len(results)} 条相关记忆（Top 5）：")
|
||
for r in results[:5]:
|
||
print(f"\n📄 {r['title']} ({r['path']})")
|
||
print(f"🏷️ 标签: {', '.join(r['tags']) if r['tags'] else '无'}")
|
||
print(f"🕒 更新: {r['mtime']}")
|
||
print(f"👁️ 预览: {r['preview'][:100]}...")
|
||
|
||
return results
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) < 2:
|
||
print("用法: python3 anchor_vault_sync.py [sync|search <关键词>]")
|
||
sys.exit(1)
|
||
|
||
cmd = sys.argv[1]
|
||
if cmd == "sync":
|
||
scan_vault()
|
||
elif cmd == "search":
|
||
if len(sys.argv) < 3:
|
||
print("[-] 搜索需要关键词。用法: search <关键词>")
|
||
sys.exit(1)
|
||
query = " ".join(sys.argv[2:])
|
||
search_vault(query)
|
||
else:
|
||
print(f"[-] 未知命令: {cmd}")
|