88 lines
3.5 KiB
Python
88 lines
3.5 KiB
Python
# scripts/step1_notion_sync.py
|
||
import os
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
from notion_client import Client
|
||
import logging
|
||
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
|
||
def extract_property_value(property_data):
|
||
"""從 Notion 頁面屬性中提取純文字值。"""
|
||
prop_type = property_data.get('type')
|
||
if prop_type == 'title':
|
||
return property_data['title'][0]['plain_text'] if property_data.get('title') else ""
|
||
elif prop_type == 'rich_text':
|
||
return "\n".join([text['plain_text'] for text in property_data.get('rich_text', [])])
|
||
elif prop_type == 'select' and property_data.get('select'):
|
||
return property_data['select']['name']
|
||
elif prop_type == 'date' and property_data.get('date'):
|
||
return property_data['date']['start']
|
||
return None
|
||
|
||
|
||
|
||
def create_project_from_page(api_key: str, page_id: str, project_name: str, projects_dir: Path):
|
||
"""根據單一 Notion 頁面建立一個新的本地專案資料夾。"""
|
||
sanitized_name = re.sub(r'[\\/*?:"<>|]', "", project_name).replace(" ", "_").lower()
|
||
project_path = projects_dir / sanitized_name
|
||
output_json_path = project_path / "data.json"
|
||
|
||
if project_path.exists():
|
||
return False, f"⚠️ 專案 '{sanitized_name}' 已存在。", None
|
||
|
||
project_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
try:
|
||
client = Client(auth=api_key)
|
||
page_data = client.pages.retrieve(page_id=page_id)
|
||
properties = page_data['properties']
|
||
page_entry = {"id": page_data['id']}
|
||
for prop_name, prop_data in properties.items():
|
||
page_entry[prop_name] = extract_property_value(prop_data)
|
||
|
||
with open(output_json_path, 'w', encoding='utf-8') as f:
|
||
json.dump(page_entry, f, ensure_ascii=False, indent=2)
|
||
return True, f"✅ 專案 '{sanitized_name}' 建立成功！", sanitized_name
|
||
except Exception as e:
|
||
return False, f"❌ 處理頁面 (ID: {page_id}) 時出錯: {e}", None
|
||
|
||
# --- 新增的更新函式 ---
|
||
def update_project_from_notion(api_key: str, project_path: Path):
|
||
"""
|
||
從 Notion 更新單一專案的 data.json 檔案。
|
||
它會讀取本地 data.json 以取得 page_id，然後抓取最新資料。
|
||
"""
|
||
data_file = project_path / "data.json"
|
||
if not data_file.exists():
|
||
return False, f"❌ 錯誤：在專案 '{project_path.name}' 中找不到 data.json。"
|
||
|
||
try:
|
||
with open(data_file, 'r', encoding='utf-8') as f:
|
||
local_data = json.load(f)
|
||
|
||
page_id = local_data.get('id')
|
||
if not page_id:
|
||
return False, "❌ 錯誤：data.json 檔案中缺少 Notion page_id。"
|
||
|
||
# 使用 page_id 從 Notion API 抓取最新資料 [4][6]
|
||
logging.info(f"正在從 Notion 更新頁面 ID: {page_id}")
|
||
client = Client(auth=api_key)
|
||
page_data = client.pages.retrieve(page_id=page_id)
|
||
|
||
properties = page_data['properties']
|
||
updated_entry = {"id": page_data['id']}
|
||
for prop_name, prop_data in properties.items():
|
||
updated_entry[prop_name] = extract_property_value(prop_data)
|
||
|
||
# 覆寫本地的 data.json 檔案
|
||
with open(data_file, 'w', encoding='utf-8') as f:
|
||
json.dump(updated_entry, f, ensure_ascii=False, indent=2)
|
||
|
||
return True, f"✅ 專案 '{project_path.name}' 已成功從 Notion 同步更新！"
|
||
|
||
except Exception as e:
|
||
logging.error(f"更新專案 '{project_path.name}' 時發生錯誤: {e}")
|
||
return False, f"❌ 更新失敗: {e}"
|