306 lines
9.2 KiB
Python
306 lines
9.2 KiB
Python
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import re
|
|||
|
|
from dataclasses import dataclass, field
|
|||
|
|
from typing import Any
|
|||
|
|
from urllib.parse import urlparse
|
|||
|
|
|
|||
|
|
import requests
|
|||
|
|
from requests import Response
|
|||
|
|
from sqlalchemy.orm import Session
|
|||
|
|
|
|||
|
|
from app.db import init_db
|
|||
|
|
from app.models import Box, Item, SubItem
|
|||
|
|
|
|||
|
|
NOTION_VERSION = "2026-03-11"
|
|||
|
|
NOTION_API_BASE = "https://api.notion.com/v1"
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass(slots=True)
|
|||
|
|
class ParsedSubItem:
|
|||
|
|
name: str
|
|||
|
|
note: str | None = None
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass(slots=True)
|
|||
|
|
class ParsedItem:
|
|||
|
|
name: str
|
|||
|
|
note: str | None = None
|
|||
|
|
is_container: bool = False
|
|||
|
|
subitems: list[ParsedSubItem] = field(default_factory=list)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass(slots=True)
|
|||
|
|
class ParsedBox:
|
|||
|
|
name: str
|
|||
|
|
note: str | None = None
|
|||
|
|
items: list[ParsedItem] = field(default_factory=list)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass(slots=True)
|
|||
|
|
class ImportSummary:
|
|||
|
|
boxes: list[ParsedBox]
|
|||
|
|
warnings: list[str] = field(default_factory=list)
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def box_count(self) -> int:
|
|||
|
|
return len(self.boxes)
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def item_count(self) -> int:
|
|||
|
|
return sum(len(box.items) for box in self.boxes)
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def container_item_count(self) -> int:
|
|||
|
|
return sum(1 for box in self.boxes for item in box.items if item.is_container)
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def subitem_count(self) -> int:
|
|||
|
|
return sum(len(item.subitems) for box in self.boxes for item in box.items)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class NotionClient:
|
|||
|
|
def __init__(self, token: str):
|
|||
|
|
self.session = requests.Session()
|
|||
|
|
self.session.headers.update(
|
|||
|
|
{
|
|||
|
|
"Authorization": f"Bearer {token}",
|
|||
|
|
"Notion-Version": NOTION_VERSION,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def list_block_children(self, block_id: str) -> list[dict[str, Any]]:
|
|||
|
|
results: list[dict[str, Any]] = []
|
|||
|
|
next_cursor: str | None = None
|
|||
|
|
|
|||
|
|
while True:
|
|||
|
|
params = {"page_size": 100}
|
|||
|
|
if next_cursor:
|
|||
|
|
params["start_cursor"] = next_cursor
|
|||
|
|
|
|||
|
|
response = self.session.get(
|
|||
|
|
f"{NOTION_API_BASE}/blocks/{block_id}/children",
|
|||
|
|
params=params,
|
|||
|
|
timeout=30,
|
|||
|
|
)
|
|||
|
|
self._raise_for_status(response)
|
|||
|
|
payload = response.json()
|
|||
|
|
results.extend(payload.get("results", []))
|
|||
|
|
|
|||
|
|
if not payload.get("has_more"):
|
|||
|
|
break
|
|||
|
|
next_cursor = payload.get("next_cursor")
|
|||
|
|
|
|||
|
|
return results
|
|||
|
|
|
|||
|
|
def _raise_for_status(self, response: Response) -> None:
|
|||
|
|
try:
|
|||
|
|
response.raise_for_status()
|
|||
|
|
except requests.HTTPError as exc:
|
|||
|
|
message = response.text
|
|||
|
|
raise RuntimeError(f"Notion API 请求失败: {response.status_code} {message}") from exc
|
|||
|
|
|
|||
|
|
|
|||
|
|
def extract_page_id(page_url: str) -> str:
|
|||
|
|
cleaned = page_url.strip()
|
|||
|
|
parsed = urlparse(cleaned)
|
|||
|
|
candidates = [segment for segment in parsed.path.split("/") if segment]
|
|||
|
|
if parsed.fragment:
|
|||
|
|
candidates.append(parsed.fragment)
|
|||
|
|
|
|||
|
|
matches: list[str] = []
|
|||
|
|
pattern = re.compile(
|
|||
|
|
r"([0-9a-fA-F]{32}|[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})"
|
|||
|
|
)
|
|||
|
|
for candidate in candidates:
|
|||
|
|
matches.extend(pattern.findall(candidate))
|
|||
|
|
|
|||
|
|
if not matches:
|
|||
|
|
raise ValueError("无法从 Notion 页面 URL 中提取 page id")
|
|||
|
|
|
|||
|
|
raw = matches[-1].replace("-", "").lower()
|
|||
|
|
return f"{raw[:8]}-{raw[8:12]}-{raw[12:16]}-{raw[16:20]}-{raw[20:]}"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def fetch_page_blocks(token: str, page_id: str) -> list[dict[str, Any]]:
|
|||
|
|
client = NotionClient(token)
|
|||
|
|
return _fetch_block_tree(client, page_id)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _fetch_block_tree(client: NotionClient, block_id: str) -> list[dict[str, Any]]:
|
|||
|
|
blocks = client.list_block_children(block_id)
|
|||
|
|
for block in blocks:
|
|||
|
|
if block.get("has_children"):
|
|||
|
|
block["_children"] = _fetch_block_tree(client, block["id"])
|
|||
|
|
else:
|
|||
|
|
block["_children"] = []
|
|||
|
|
return blocks
|
|||
|
|
|
|||
|
|
|
|||
|
|
def parse_notion_blocks(blocks: list[dict[str, Any]]) -> ImportSummary:
|
|||
|
|
boxes: list[ParsedBox] = []
|
|||
|
|
warnings: list[str] = []
|
|||
|
|
current_box: ParsedBox | None = None
|
|||
|
|
|
|||
|
|
for block in blocks:
|
|||
|
|
block_type = block.get("type")
|
|||
|
|
|
|||
|
|
if block_type == "heading_2":
|
|||
|
|
heading_text = extract_block_text(block)
|
|||
|
|
if not heading_text:
|
|||
|
|
warnings.append("发现空的 heading_2，已跳过")
|
|||
|
|
continue
|
|||
|
|
current_box = ParsedBox(name=heading_text)
|
|||
|
|
boxes.append(current_box)
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
if block_type == "bulleted_list_item":
|
|||
|
|
if current_box is None:
|
|||
|
|
warnings.append(
|
|||
|
|
f"发现未归属到任何 heading_2 的一级 bullet：{extract_block_text(block) or '[空文本]'}"
|
|||
|
|
)
|
|||
|
|
continue
|
|||
|
|
parsed_item = _parse_item_block(block, warnings, level=1)
|
|||
|
|
if parsed_item is not None:
|
|||
|
|
current_box.items.append(parsed_item)
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
warnings.extend(_warning_for_unsupported_block(block, level=0))
|
|||
|
|
|
|||
|
|
return ImportSummary(boxes=boxes, warnings=warnings)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _parse_item_block(
|
|||
|
|
block: dict[str, Any],
|
|||
|
|
warnings: list[str],
|
|||
|
|
*,
|
|||
|
|
level: int,
|
|||
|
|
) -> ParsedItem | None:
|
|||
|
|
item_name = extract_block_text(block)
|
|||
|
|
if not item_name:
|
|||
|
|
warnings.append(f"发现空的 bullet（层级 {level}），已跳过")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
child_blocks = block.get("_children", [])
|
|||
|
|
subitems: list[ParsedSubItem] = []
|
|||
|
|
|
|||
|
|
for child in child_blocks:
|
|||
|
|
child_type = child.get("type")
|
|||
|
|
if child_type == "bulleted_list_item":
|
|||
|
|
child_name = extract_block_text(child)
|
|||
|
|
if not child_name:
|
|||
|
|
warnings.append(f"发现空的二级 bullet（父项：{item_name}），已跳过")
|
|||
|
|
continue
|
|||
|
|
subitems.append(ParsedSubItem(name=child_name))
|
|||
|
|
|
|||
|
|
if child.get("_children"):
|
|||
|
|
warnings.append(
|
|||
|
|
f"发现超出支持层级的三级内容（父项：{item_name} -> 子项：{child_name}），已忽略更深层级"
|
|||
|
|
)
|
|||
|
|
for deep_child in child["_children"]:
|
|||
|
|
warnings.extend(_warning_for_unsupported_block(deep_child, level=3))
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
warnings.extend(_warning_for_unsupported_block(child, level=2, parent_name=item_name))
|
|||
|
|
|
|||
|
|
return ParsedItem(
|
|||
|
|
name=item_name,
|
|||
|
|
is_container=bool(subitems),
|
|||
|
|
subitems=subitems,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _warning_for_unsupported_block(
|
|||
|
|
block: dict[str, Any],
|
|||
|
|
*,
|
|||
|
|
level: int,
|
|||
|
|
parent_name: str | None = None,
|
|||
|
|
) -> list[str]:
|
|||
|
|
block_type = block.get("type", "unknown")
|
|||
|
|
text = extract_block_text(block) or "[无文本]"
|
|||
|
|
prefix = f"层级 {level} block"
|
|||
|
|
if parent_name:
|
|||
|
|
prefix += f"（父项：{parent_name}）"
|
|||
|
|
|
|||
|
|
if block_type in {"image", "file", "video", "audio", "pdf"}:
|
|||
|
|
return [f"{prefix} 类型 {block_type} 已跳过（这版不导入图片或媒体）：{text}"]
|
|||
|
|
|
|||
|
|
return [f"{prefix} 类型 {block_type} 未按导入规则处理，已跳过：{text}"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def extract_block_text(block: dict[str, Any]) -> str:
|
|||
|
|
block_type = block.get("type")
|
|||
|
|
block_data = block.get(block_type, {}) if block_type else {}
|
|||
|
|
rich_text = block_data.get("rich_text", [])
|
|||
|
|
return "".join(part.get("plain_text", "") for part in rich_text).strip()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def print_summary(summary: ImportSummary) -> None:
|
|||
|
|
print()
|
|||
|
|
print("解析结果摘要")
|
|||
|
|
print(f"- Box: {summary.box_count}")
|
|||
|
|
print(f"- Item: {summary.item_count}")
|
|||
|
|
print(f"- 其中容器型 Item: {summary.container_item_count}")
|
|||
|
|
print(f"- SubItem: {summary.subitem_count}")
|
|||
|
|
print(f"- Warnings: {len(summary.warnings)}")
|
|||
|
|
print()
|
|||
|
|
|
|||
|
|
for box in summary.boxes:
|
|||
|
|
container_names = [item.name for item in box.items if item.is_container]
|
|||
|
|
print(f"[Box] {box.name}")
|
|||
|
|
print(f" - Item 数量: {len(box.items)}")
|
|||
|
|
if container_names:
|
|||
|
|
print(f" - 容器型 Item: {', '.join(container_names)}")
|
|||
|
|
for item in box.items:
|
|||
|
|
if item.is_container:
|
|||
|
|
print(f" * {item.name} -> SubItem {len(item.subitems)} 个")
|
|||
|
|
|
|||
|
|
if summary.warnings:
|
|||
|
|
print()
|
|||
|
|
print("Warnings")
|
|||
|
|
for warning in summary.warnings:
|
|||
|
|
print(f"- {warning}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def apply_import(summary: ImportSummary, db: Session) -> dict[str, int]:
|
|||
|
|
init_db()
|
|||
|
|
|
|||
|
|
created_boxes = 0
|
|||
|
|
created_items = 0
|
|||
|
|
created_subitems = 0
|
|||
|
|
|
|||
|
|
for parsed_box in summary.boxes:
|
|||
|
|
box = Box(name=parsed_box.name, note=parsed_box.note)
|
|||
|
|
db.add(box)
|
|||
|
|
db.flush()
|
|||
|
|
created_boxes += 1
|
|||
|
|
|
|||
|
|
for parsed_item in parsed_box.items:
|
|||
|
|
item = Item(
|
|||
|
|
box=box,
|
|||
|
|
name=parsed_item.name,
|
|||
|
|
note=parsed_item.note,
|
|||
|
|
quantity=1,
|
|||
|
|
is_container=parsed_item.is_container,
|
|||
|
|
)
|
|||
|
|
db.add(item)
|
|||
|
|
db.flush()
|
|||
|
|
created_items += 1
|
|||
|
|
|
|||
|
|
for parsed_subitem in parsed_item.subitems:
|
|||
|
|
subitem = SubItem(
|
|||
|
|
parent_item=item,
|
|||
|
|
name=parsed_subitem.name,
|
|||
|
|
note=parsed_subitem.note,
|
|||
|
|
quantity=1,
|
|||
|
|
)
|
|||
|
|
db.add(subitem)
|
|||
|
|
created_subitems += 1
|
|||
|
|
|
|||
|
|
db.commit()
|
|||
|
|
return {
|
|||
|
|
"boxes": created_boxes,
|
|||
|
|
"items": created_items,
|
|||
|
|
"subitems": created_subitems,
|
|||
|
|
}
|