#!/usr/bin/env python3 """ 雙儲存研究筆記同步執行工具 將研究筆記同步到指定專案，或從專案同步回 Research_zoo 使用方式： # 從 Research_zoo 同步到專案（首次遷移） python sync_note.py migrate "research/2026/01/xxx.md" --to real-estate-advisor # 從主本同步到副本 python sync_note.py sync "research/2026/01/xxx.md" --direction primary-to-secondary # 從副本同步回主本 python sync_note.py sync "research/2026/01/xxx.md" --direction secondary-to-primary # 批次遷移 python sync_note.py migrate-batch --config migrate_config.yaml """ import sys sys.stdout.reconfigure(encoding='utf-8') import re import yaml import shutil from pathlib import Path from datetime import datetime from typing import Optional, Dict # 設定 RESEARCH_ZOO_ROOT = Path(r"C:\Users\User\Documents\GitHub\Research_zoo") PROJECTS = { "real-estate-advisor": Path(r"C:\Users\User\Documents\GitHub\real-estate-advisor") } def extract_yaml_frontmatter(content: str) -> tuple[Optional[Dict], str, str]: """ 從 Markdown 內容擷取 YAML frontmatter Returns: (yaml_dict, yaml_text, body_text) """ match = re.match(r'^(---\n)(.*?)(\n---\n)(.*)', content, re.DOTALL) if not match: return None, "", content try: yaml_dict = yaml.safe_load(match.group(2)) return yaml_dict, match.group(2), match.group(4) except yaml.YAMLError: return None, match.group(2), match.group(4) def increment_version(version: str, level: str = "minor") -> str: """ 遞增版本號 Args: version: 目前版本 (如 "1.0.0") level: 遞增層級 ("major", "minor", "patch") """ parts = version.split('.') if len(parts) != 3: parts = ['1', '0', '0'] major, minor, patch = int(parts[0]), int(parts[1]), int(parts[2]) if level == "major": major += 1 minor = 0 patch = 0 elif level == "minor": minor += 1 patch = 0 else: # patch patch += 1 return f"{major}.{minor}.{patch}" def add_dual_storage_metadata(content: str, role: str, linked_path: str, linked_repo: str) -> str: """ 在 YAML frontmatter 中新增 dual_storage 欄位 """ match = re.match(r'^(---\n)(.*?)(\n---)', content, re.DOTALL) if not match: return content yaml_text = match.group(2) rest = content[match.end():] # 檢查是否已有 dual_storage if 'dual_storage:' in yaml_text: print(" 警告: 檔案已有 dual_storage 設定，跳過") return content today = datetime.now().strftime('%Y-%m-%d') dual_storage_block = f""" # ===== 雙儲存追蹤（Dual Storage） ===== dual_storage: enabled: true role: "{role}" linked_path: "{linked_path}" linked_repo: "{linked_repo}" primary_version: "1.0.0" local_version: "1.0.0" last_sync: {today} sync_direction: "initial" """ # 找到 notes_for_future_me 或 --- 結尾之前插入 if 'notes_for_future_me:' in yaml_text: # 在 notes_for_future_me 之前插入 yaml_text = yaml_text.replace( 'notes_for_future_me:', dual_storage_block.strip() + '\n\nnotes_for_future_me:' ) else: # 在結尾插入 yaml_text = yaml_text.rstrip() + '\n' + dual_storage_block.strip() return f"---\n{yaml_text}\n---{rest}" def migrate_note(source_path: Path, target_project: str) -> bool: """ 將筆記遷移到目標專案 1. 在目標專案建立副本 2. 更新原檔的 dual_storage metadata (role: primary) 3. 更新副本的 dual_storage metadata (role: secondary) """ if target_project not in PROJECTS: print(f"錯誤: 未知的專案 '{target_project}'") print(f"可用專案: {list(PROJECTS.keys())}") return False project_root = PROJECTS[target_project] project_research = project_root / "research" # 計算相對路徑 try: relative_path = source_path.relative_to(RESEARCH_ZOO_ROOT / "research") except ValueError: print(f"錯誤: {source_path} 不在 Research_zoo/research 資料夾內") return False target_path = project_research / relative_path print(f"\n遷移筆記:") print(f" 來源: {source_path}") print(f" 目標: {target_path}") # 讀取原檔內容 try: content = source_path.read_text(encoding='utf-8') except Exception as e: print(f"錯誤: 讀取來源檔案失敗 - {e}") return False # 確保目標資料夾存在 target_path.parent.mkdir(parents=True, exist_ok=True) # 準備兩邊的 metadata primary_linked_path = str(target_path).replace('\\', '/') secondary_linked_path = str(source_path).replace('\\', '/') # 更新原檔 (primary) primary_content = add_dual_storage_metadata( content, role="primary", linked_path=primary_linked_path, linked_repo=target_project ) # 建立副本 (secondary) secondary_content = add_dual_storage_metadata( content, role="secondary", linked_path=secondary_linked_path, linked_repo="Research_zoo" ) # 寫入檔案 try: source_path.write_text(primary_content, encoding='utf-8') print(f" [OK] 更新原檔 (primary)") target_path.write_text(secondary_content, encoding='utf-8') print(f" [OK] 建立副本 (secondary)") return True except Exception as e: print(f"錯誤: 寫入檔案失敗 - {e}") return False def sync_note(source_path: Path, direction: str) -> bool: """ 同步筆記內容 direction: "primary-to-secondary": 從主本同步到副本 "secondary-to-primary": 從副本同步回主本 """ try: content = source_path.read_text(encoding='utf-8') except Exception as e: print(f"錯誤: 讀取檔案失敗 - {e}") return False metadata, yaml_text, body = extract_yaml_frontmatter(content) if not metadata: print("錯誤: 無法解析 YAML frontmatter") return False dual_storage = metadata.get('dual_storage', {}) if not dual_storage.get('enabled'): print("錯誤: 此檔案未啟用 dual_storage") return False linked_path = Path(dual_storage.get('linked_path', '')) if not linked_path.exists(): print(f"錯誤: 連結檔案不存在 - {linked_path}") return False current_role = dual_storage.get('role') today = datetime.now().strftime('%Y-%m-%d') if direction == "primary-to-secondary": if current_role != "primary": print("警告: 當前檔案不是 primary，但嘗試執行 primary→secondary 同步") # 遞增版本 new_version = increment_version(dual_storage.get('primary_version', '1.0.0'), 'minor') # 更新 primary dual_storage['primary_version'] = new_version dual_storage['local_version'] = new_version dual_storage['last_sync'] = today dual_storage['sync_direction'] = "primary→secondary" # 更新 secondary linked_content = linked_path.read_text(encoding='utf-8') linked_metadata, linked_yaml, linked_body = extract_yaml_frontmatter(linked_content) if linked_metadata and 'dual_storage' in linked_metadata: linked_metadata['dual_storage']['local_version'] = new_version linked_metadata['dual_storage']['last_sync'] = today linked_metadata['dual_storage']['sync_direction'] = "primary→secondary" # 複製 body 內容 # (這裡簡化處理，實際可能需要更複雜的合併邏輯) print(f" 版本更新: {dual_storage.get('primary_version')} → {new_version}") elif direction == "secondary-to-primary": # 類似邏輯，方向相反 pass print(f" 同步完成: {source_path.name}") return True def main(): import argparse parser = argparse.ArgumentParser(description="雙儲存研究筆記同步工具") subparsers = parser.add_subparsers(dest='command', help='可用命令') # migrate 命令 migrate_parser = subparsers.add_parser('migrate', help='將筆記遷移到專案') migrate_parser.add_argument('source', help='來源檔案路徑 (相對於 Research_zoo)') migrate_parser.add_argument('--to', required=True, help='目標專案名稱') # sync 命令 sync_parser = subparsers.add_parser('sync', help='同步筆記') sync_parser.add_argument('source', help='來源檔案路徑') sync_parser.add_argument('--direction', required=True, choices=['primary-to-secondary', 'secondary-to-primary'], help='同步方向') # list 命令 list_parser = subparsers.add_parser('list', help='列出可遷移的筆記') list_parser.add_argument('--project', required=True, help='目標專案名稱') list_parser.add_argument('--domain', default='不動產', help='篩選領域') args = parser.parse_args() if args.command == 'migrate': source = RESEARCH_ZOO_ROOT / args.source if not source.exists(): # 嘗試直接使用路徑 source = Path(args.source) if not source.exists(): print(f"錯誤: 找不到檔案 {args.source}") sys.exit(1) success = migrate_note(source, args.to) sys.exit(0 if success else 1) elif args.command == 'sync': source = Path(args.source) if not source.exists(): source = RESEARCH_ZOO_ROOT / args.source if not source.exists(): print(f"錯誤: 找不到檔案 {args.source}") sys.exit(1) success = sync_note(source, args.direction) sys.exit(0 if success else 1) elif args.command == 'list': print(f"列出 {args.project} 相關的可遷移筆記 (domain: {args.domain})...") # TODO: 實作列出邏輯 pass else: parser.print_help() if __name__ == "__main__": main()