在前面的章節中，我們完成了情報的抓取、多維度清洗、LLM 精煉評分，並成功將結構化資料儲存至 Google Sheets。
然而，Google Sheets 適合做資料流中繼站與低成本備份，若要建立一個團隊便於檢索、具備卡片視圖（Board）、排版美觀且易於閱讀「二階推演」與「技術缺陷」的個人知識庫，Notion 依然是目前最理想的展示介面。
今天我們將實作六角架構下的 NotionSyncAdapter，打通情資管線的最後一哩路，並實作冪等性（Idempotence）防護，確保重複執行時不會洗版！
在 Hexagonal Architecture 中，我們透過定義在 src/ports/knowledge_sync.py 的 Port 介面，實作出負責對接外部系統的 Adapter：
[Google Sheets Adapter]
│ (讀取 Score >= 4 的情報候選)
▼
[Sync Pipeline Script]
│ (調用 KnowledgeSyncPort)
▼
[NotionSyncAdapter]
├── 1. Query: 檢查 Entry ID 是否已存在 (確保冪等性)
│ ├── 已存在 ──> Skip / Update
│ └── 不存在 ──> Create Page
▼
[Notion Database API]
在撰寫程式前，請確保以下兩項配置完成：
secret_ 開頭的 Internal Integration Secret。在 Notion 建立一個 Database（Full-page 或 Inline 皆可），必須包含以下 Property（大小寫與型別需完全一致）：
| 欄位名稱 (Property Name) | 欄位型別 (Type) | 說明 |
|---|---|---|
| Title | title |
論文或文章標題 |
| Entry ID | rich_text (Text) |
唯一的雜湊或識別碼（查重依據） |
| Domain | select |
領域類別（如 SCIENCE, TECH 等） |
| Score | number |
綜合評分 |
| Source URL | url |
來源連結 |
| Date | date |
發布或抓取日期 |
進入該資料庫頁面，點選右上角 ... $\rightarrow$ 下拉至 Connections $\rightarrow$ 點擊 Search for connections... $\rightarrow$ 搜尋並加入你剛剛建立的 Integration。
src/ports/knowledge_sync.py)from abc import ABC, abstractmethod
from src.core.models import IntelligenceItem
class KnowledgeSyncPort(ABC):
@abstractmethod
def sync_item(self, item: IntelligenceItem, overwrite: bool = False) -> bool:
"""同步單筆情報至外部知識庫"""
pass
@abstractmethod
def batch_sync(self, items: list[IntelligenceItem], overwrite: bool = False) -> dict[str, int]:
"""批次同步情報"""
pass
src/adapters/notion_adapter.py)為了避免 SDK 版本改版造成 Invalid request URL，在查詢端點直接使用 Python 原生 requests 發送標準 REST 請求，寫入部分則沿用官方 notion-client：
from __future__ import annotations
import logging
import uuid
from typing import Any
import requests
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.auth_token = auth_token
self.client = Client(auth=auth_token)
# 清理並轉為標準 UUID 格式
raw_id = database_id.strip().replace("-", "")
self.database_id = str(uuid.UUID(raw_id))
def _find_page_by_entry_id(self, entry_id: str) -> str | None:
"""依據 Entry ID 查詢是否已存在該筆情資（確保冪等性）"""
url = f"https://api.notion.com/v1/databases/{self.database_id}/query"
headers = {
"Authorization": f"Bearer {self.auth_token}",
"Notion-Version": "2022-06-28",
"Content-Type": "application/json",
}
payload = {
"filter": {
"property": "Entry ID",
"rich_text": {
"equals": str(entry_id).strip()
}
}
}
try:
res = requests.post(url, headers=headers, json=payload, timeout=10)
if res.status_code == 200:
results = res.json().get("results", [])
return results[0]["id"] if results else None
else:
logger.error(f"查詢 Notion 資料庫失敗 [{res.status_code}]: {res.text}")
return None
except Exception as e:
logger.error(f"查詢連線失敗: {e}")
return None
def _build_properties(self, item: IntelligenceItem) -> dict[str, Any]:
"""將領域模型映射為 Notion Database 欄位屬性"""
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)from __future__ import annotations
import logging
import os
from dotenv import load_dotenv
from src.adapters.notion_adapter import NotionSyncAdapter
from src.adapters.sheets_adapter import SheetsAdapter
load_dotenv()
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()
執行同步腳本：
python -m scripts.sync_to_notion
初次執行輸出：
2026-10-21 18:30:10 - [INFO] - 連線至 Google Sheets 讀取已處理情報...
2026-10-21 18:30:13 - [INFO] - [*] 找到 8 筆高價值情報候選項目...
2026-10-21 18:30:14 - [INFO] - 成功建立 Notion 新頁面: [dfd9abe0]
...
2026-10-21 18:30:18 - [INFO] - [+] Notion 同步完成！新增/更新: 8 篇，略過: 0 篇。
重複執行（驗證冪等性）：
2026-10-21 18:38:10 - [INFO] - 連線至 Google Sheets 讀取已處理情報...
2026-10-21 18:38:13 - [INFO] - [*] 找到 8 筆高價值情報候選項目...
2026-10-21 18:38:14 - [INFO] - Notion 頁面已存在，略過更新: [dfd9abe0]
...
2026-10-21 18:38:18 - [INFO] - [+] Notion 同步完成！新增/更新: 0 篇，略過: 8 篇。
404 ObjectNotFound: Could not find database with ID...？Connected，並不代表該 Database 具備讀寫權限。... $\rightarrow$ Connections $\rightarrow$ 搜尋你的 Integration 名稱（如 Intelligence-Radar-Sync）並點擊授權。[app.notion.com/p/](https://app.notion.com/p/)...，哪一段才是真正 Database ID？[https://app.notion.com/p/](https://app.notion.com/p/){workspace_slug}/{database_id}?v={view_id}
?v= 前方那一串 32 碼英數字元。在呼叫 API 前，建議用 uuid.UUID(raw_id) 轉為標準 UUID 格式。?v= 後面是視圖（View）的 ID，不可作為 Database ID。400 Bad Request: Invalid request URL 卻依然新增成功？_find_page_by_entry_id() 查詢報錯，except 攔截後會回傳 None。這會讓系統誤以為「此資料庫還沒有這筆資料」，進而執行 pages.create，導致重複新增。notion-client SDK 在直接使用 client.request 時，容易與伺服器路由校驗衝突。requests 發送 POST [https://api.notion.com/v1/databases/](https://api.notion.com/v1/databases/){id}/query，並確保資料庫中用於過濾的 Entry ID 欄位型態為 Text（rich_text）。
iThome鐵人賽
看影片追技術