在 Day 21 到 Day 23 中，我們建構了 Markdown 導向的沉澱管線，實現將高價值情報自動同步至 Obsidian 本地 Vault 與 GitHub 私有儲存庫。
然而，在現代團隊協作與跨裝置瀏覽中，許多人偏好使用具備豐富關聯資料庫（Relational Database）、過濾檢視（Views）、看板模式（Board View）與官方行動端支援的 Notion。
今天我們將貫徹六角架構（Hexagonal Architecture）的威力：核心領域模型與評估管線完全不需變動，只需實現 KnowledgeSyncPort 輸出端口的另一個轉接器——NotionSyncAdapter，即可將結構化情報（包含標籤、評分、二階推演與 Callout 區塊）寫入 Notion Database。
在六角架構中，輸出適配器（Driven Adapters）是可插拔的組件。核心業務邏輯僅依賴於 KnowledgeSyncPort 介面：
┌────────────────────────┐
│ IntelligenceItem │
│ (領域模型 Domain) │
└───────────┬────────────┘
│
▼
┌────────────────────────┐
│ KnowledgeSyncPort │
│ (輸出端口 Driven Port) │
└─────┬────────────┬─────┘
│ │
┌───────────────────────┘ └───────────────────────┐
▼ ▼
┌────────────────────────┐ ┌────────────────────────┐
│ ObsidianVaultAdapter │ │ NotionSyncAdapter │
│ (Markdown / Git Push) │ │ (REST API / Blocks) │
└────────────────────────┘ └────────────────────────┘
系統可以同時配置兩個轉接器，達成「本地 Markdown 備份 + 雲端 Notion 協作」的雙軌同步機制。
使用 Notion API 寫入 Database 前，需完成以下三個設定：
Intelligence-Radar-Sync。Read content、Update content 與 Insert content。ntn_... 或 secret_...）。在 Notion 中新增一個資料表（Database - Full page）。
資料庫欄位需涵蓋以下屬性：
Title (title 類型)：名稱設為 Title
Domain (select 類型)：名稱設為 Domain
Score (number 類型)：名稱設為 Score
Entry ID (rich_text 類型)：名稱設為 Entry ID
Source URL (url 類型)：名稱設為 Source URL
Date (date 類型)：名稱設為 Date
複製 Database 頁面網址，從網址中擷取 Database ID（格式為 32 碼英數組合，位於 Workspace 名稱之後、問號 ? 之前）。
...。Intelligence-Radar-Sync（未加入連線會收到 404 Object not found 錯誤）。Notion API 的資料模型分為兩層：
我們使用官方維護的 Python 客戶端 notion-client 進行封裝：
在 requirements.txt 與 pyproject.toml 中加入：
notion-client>=2.2.1
在 src/adapters/notion_adapter.py 中實現：
# src/adapters/notion_adapter.py
from __future__ import annotations
import logging
from typing import Any
from notion_client import Client
from notion_client.errors import APIResponseError
from src.core.models import IntelligenceItem
from src.ports.knowledge_sync import KnowledgeSyncPort
logger = logging.getLogger(__name__)
class NotionSyncAdapter(KnowledgeSyncPort):
def __init__(self, auth_token: str, database_id: str):
self.client = Client(auth=auth_token)
self.database_id = database_id
def _find_page_by_entry_id(self, entry_id: str) -> str | None:
"""依據 Entry ID 查詢是否已存在該筆情資（確保冪等性）"""
try:
response = self.client.databases.query(
database_id=self.database_id,
filter={
"property": "Entry ID",
"rich_text": {
"equals": str(entry_id)
}
}
)
results = response.get("results", [])
return results[0]["id"] if results else None
except APIResponseError as e:
logger.error(f"查詢 Notion 資料庫失敗: {e}")
return None
def _build_properties(self, item: IntelligenceItem) -> dict[str, Any]:
"""將領域模型映射為 Notion Database Properties"""
date_str = None
if item.timestamp:
date_str = item.timestamp.strftime("%Y-%m-%d") if hasattr(item.timestamp, "strftime") else str(item.timestamp)[:10]
properties: dict[str, Any] = {
"Title": {
"title": [{"text": {"content": item.source_title or "Untitled"}}]
},
"Entry ID": {
"rich_text": [{"text": {"content": str(item.entry_id)}}]
},
"Domain": {
"select": {"name": item.domain or "GENERAL"}
},
"Score": {
"number": float(item.verified_score or item.raw_score or 0)
}
}
if item.source_url:
properties["Source URL"] = {"url": item.source_url}
if date_str:
properties["Date"] = {"date": {"start": date_str}}
return properties
def _build_children_blocks(self, item: IntelligenceItem) -> list[dict[str, Any]]:
"""建立頁面內的 Rich Text 區塊結構（包含 Callout 與各級標題）"""
blocks = []
# 核心提煉 Callout
summary_text = item.editorial_summary or "暫無編輯精煉摘要"
blocks.append({
"object": "block",
"type": "callout",
"callout": {
"rich_text": [{"text": {"content": f"核心提煉：\n{summary_text}"}}],
"icon": {"emoji": "💡"}
}
})
# 量化對比與效能指標
blocks.append({
"object": "block",
"type": "heading_2",
"heading_2": {
"rich_text": [{"text": {"content": "📊 量化對比與效能指標"}}]
}
})
blocks.append({
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [{"text": {"content": item.metrics or "無顯著量化指標揭露。"}}]
}
})
# 技術邊界與缺陷
blocks.append({
"object": "block",
"type": "heading_2",
"heading_2": {
"rich_text": [{"text": {"content": "⚠️ 技術邊界與潛在缺陷"}}]
}
})
blocks.append({
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [{"text": {"content": item.limitations or "作者未於論文中強調顯著邊界。"}}]
}
})
# 二階效應
blocks.append({
"object": "block",
"type": "heading_2",
"heading_2": {
"rich_text": [{"text": {"content": "🔮 二階效應與跨領域衝擊"}}]
}
})
blocks.append({
"object": "block",
"type": "callout",
"callout": {
"rich_text": [{"text": {"content": item.cross_impact_notes or "暫無二階推演。" }}],
"icon": {"emoji": "🚀"}
}
})
return blocks
def sync_item(self, item: IntelligenceItem, overwrite: bool = False) -> bool:
"""同步單筆情報至 Notion，若已存在則檢查 overwrite 決定是否更新"""
existing_page_id = self._find_page_by_entry_id(str(item.entry_id))
properties = self._build_properties(item)
try:
if existing_page_id:
if not overwrite:
logger.info(f"Notion 頁面已存在，略過更新: [{item.entry_id}]")
return False
# 覆寫既有頁面屬性
self.client.pages.update(page_id=existing_page_id, properties=properties)
logger.info(f"成功更新 Notion 頁面: [{item.entry_id}]")
return True
else:
# 建立新頁面
children = self._build_children_blocks(item)
self.client.pages.create(
parent={"database_id": self.database_id},
properties=properties,
children=children
)
logger.info(f"成功建立 Notion 新頁面: [{item.entry_id}]")
return True
except APIResponseError as err:
logger.error(f"寫入 Notion API 失敗 [{item.entry_id}]: {err}")
return False
def batch_sync(self, items: list[IntelligenceItem], overwrite: bool = False) -> dict[str, int]:
stats = {"synced": 0, "skipped": 0}
for item in items:
if self.sync_item(item, overwrite=overwrite):
stats["synced"] += 1
else:
stats["skipped"] += 1
return stats
在 scripts/sync_to_notion.py 中建立可獨立運行的同步工具：
# scripts/sync_to_notion.py
from __future__ import annotations
import os
import logging
from src.adapters.sheets_adapter import SheetsAdapter
from src.adapters.notion_adapter import NotionSyncAdapter
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - [%(levelname)s] - %(message)s"
)
logger = logging.getLogger(__name__)
def main():
token = os.getenv("NOTION_API_TOKEN")
database_id = os.getenv("NOTION_DATABASE_ID")
if not token or not database_id:
logger.error("缺少 NOTION_API_TOKEN 或 NOTION_DATABASE_ID 環境變數！")
return
logger.info("連線至 Google Sheets 讀取已處理情報...")
sheets_adapter = SheetsAdapter()
candidates = sheets_adapter.fetch_processed_items(min_score=4)
logger.info(f"[*] 找到 {len(candidates)} 筆高價值情報候選項目...")
notion_adapter = NotionSyncAdapter(auth_token=token, database_id=database_id)
results = notion_adapter.batch_sync(candidates)
logger.info(f"[+] Notion 同步完成！新增/更新: {results['synced']} 篇，略過: {results['skipped']} 篇。")
if __name__ == "__main__":
main()
設定好環境變數後執行：
export NOTION_API_TOKEN="ntn_xxxxxx..."
export NOTION_DATABASE_ID="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
python -m scripts.sync_to_notion
終端機將輸出：
2026-10-08 23:40:10 - [INFO] - 連線至 Google Sheets 讀取已處理情報...
2026-10-08 23:40:11 - [INFO] - [*] 找到 2 筆高價值情報候選項目...
2026-10-08 23:40:12 - [INFO] - 成功建立 Notion 新頁面: [dfd9abe0]
2026-10-08 23:40:13 - [INFO] - 成功建立 Notion 新頁面: [ddad0487]
2026-10-08 23:40:13 - [INFO] - [+] Notion 同步完成！新增/更新: 2 篇，略過: 0 篇。
打開 Notion Database，每筆高分情報已自動轉為具備屬性欄位、評分星星與 Rich Text Callout 的精美卡片。
在整合 Notion REST API 時，有三個常見的陷阱需要防範：
text.content 區塊字串長度不可超過 2,000 字元。raw_content 或詳細摘要超出限制，API 會直接噴出 validation_error。生產環境中需將超長文字切片為多個 Rich Text 物件。Domain 選項（例如首次出現的 ROBOTICS），Notion API 會自動新增該選項，但呼叫帳號必須具備該資料庫的編輯權限。time.sleep(0.35)），避免觸發 HTTP 429 錯誤。今天我們成功擴展了情報雷達的沉澱層：
NotionSyncAdapter。PROCESSED 到 Notion 結構化關聯資料庫的映射邏輯。目前系統的前哨採集、核心研判、監控通知與知識庫沉澱（Obsidian / Notion）已全數就緒。然而在自動巡航中，四大代理人（Reviewer、Domain Expert、Synthesizer、Editor）在後台思考的推理細節仍處於「黑盒子」狀態。
在 Day 25，我們將進入可觀測性章節：「可觀測性黑盒子：設計 Agent Trace 日誌與結構化思考過程追蹤」，透過結構化追蹤揭露代理人完整的推理鏈與評分依據！
iThome鐵人賽
看影片追技術