Notion 整个空间备份

#!/usr/bin/env python3
"""手动导出唯一的 Notion 空间。Python 3.9+，仅标准库，不保存令牌。"""
import copy
import getpass
import http.cookiejar
import json
import os
import re
import shutil
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from datetime import datetime
from pathlib import Path


class ExportError(Exception):
    pass


def value(record):
    for _ in range(2):
        if isinstance(record, dict) and isinstance(record.get("value"), dict):
            record = record["value"]
    return record if isinstance(record, dict) else {}


class Exporter:
    def __init__(self, token):
        self.cookies = http.cookiejar.CookieJar()
        for domain in (".notion.so", ".notion.com"):
            self.cookies.set_cookie(http.cookiejar.Cookie(
                0, "token_v2", token, None, False, domain, True, True,
                "/", True, True, None, True, None, None, {}, False,
            ))
        # CookieJar 按域发送令牌，不向下载重定向中的其他域泄露凭证。
        self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.cookies))
        self.headers = {"Content-Type": "application/json", "User-Agent": "notion-export/1.0"}
        self.space_id = None

    def post(self, endpoint, data):
        request = urllib.request.Request(
            "https://www.notion.so/api/v3/" + endpoint,
            data=json.dumps(data).encode(), headers=self.headers,
        )
        with self.opener.open(request, timeout=60) as response:
            result = json.load(response)
        if not isinstance(result, dict):
            raise ExportError(endpoint + " 返回的数据格式不正确。")
        return result

    def prepare(self):
        spaces = {}
        for endpoint in ("getSpaces", "loadUserContent"):
            payload = self.post(endpoint, {})
            containers = [(None, payload)] + list(payload.items())
            for user_id, container in containers:
                if not isinstance(container, dict):
                    continue
                mapping = container.get("space") or container.get("recordMap", {}).get("space", {})
                if not isinstance(mapping, dict):
                    continue
                for space_id, record in mapping.items():
                    name = value(record).get("name")
                    if name:
                        owner = user_id if user_id and re.fullmatch(r"[0-9a-fA-F-]{32,36}", user_id) else None
                        if space_id not in spaces or (owner and not spaces[space_id][1]):
                            spaces[space_id] = (name, owner)
            if spaces:
                break
        if len(spaces) != 1:
            raise ExportError("找到 {} 个空间；此精简脚本仅支持账号恰好有一个可访问空间。".format(len(spaces)))
        self.space_id, (name, user_id) = next(iter(spaces.items()))
        self.headers["x-notion-space-id"] = self.space_id
        if user_id:
            self.headers["x-notion-active-user-header"] = user_id
        # 不依赖 getSpaces 是否下发令牌，主动调用已实测能返回 file_token 的接口。
        self.post("loadUserContent", {})
        if not any(c.name == "file_token" and c.value and not c.is_expired() for c in self.cookies):
            raise ExportError("Notion 未返回 file_token，无法下载。请重新获取登录令牌后重试。")
        print("已选择空间：" + name, flush=True)
        return name

    def export(self):
        started_ms = int(time.time() * 1000)
        task_id = self.post("enqueueTask", {"task": {
            "eventName": "exportSpace", "request": {
                "spaceId": self.space_id,
                "exportOptions": {"exportType": "markdown", "timeZone": "Asia/Shanghai",
                                  "locale": "en", "flattenExportFiletree": False},
                "shouldExportComments": False,
            },
        }}).get("taskId")
        if not task_id:
            raise ExportError("未收到导出任务 ID。")
        print("导出任务已提交，最多等待 1 小时……", flush=True)
        deadline = time.monotonic() + 3600
        while time.monotonic() < deadline:
            tasks = self.post("getTasks", {"taskIds": [task_id]}).get("results") or []
            task = next((t for t in tasks if isinstance(t, dict) and t.get("id") == task_id), {})
            if task.get("state") == "failure":
                raise ExportError("Notion 导出任务失败，请查看 Notion 通知。")
            if task.get("state") == "success":
                url = (task.get("status") or {}).get("exportURL")
                if url:
                    return url
                # 部分任务只在通知中提供下载链接；排除早于本次任务的通知。
                baseline = int(task.get("equeuedAt") or task.get("enqueuedAt") or started_ms)
                data = self.post("getNotificationLogV2", {
                    "spaceId": self.space_id, "size": 50,
                    "type": "unread_and_read", "variant": "no_grouping",
                })
                links = set()
                for record in data.get("recordMap", {}).get("activity", {}).values():
                    activity = value(record)
                    if activity.get("type") != "export-completed" or int(activity.get("start_time") or 0) < baseline:
                        continue
                    if activity.get("space_id") and activity["space_id"] != self.space_id:
                        continue
                    if (activity.get("taskId") or activity.get("task_id") or task_id) != task_id:
                        continue
                    for edit in activity.get("edits") or []:
                        if isinstance(edit, dict) and edit.get("link"):
                            links.add(edit["link"])
                if len(links) > 1:
                    raise ExportError("同时出现多个导出链接，请到 Notion 通知中手动下载，避免取错备份。")
                if links:
                    return links.pop()
            print("等待 Notion 生成文件及下载链接……", flush=True)
            time.sleep(10)
        raise ExportError("等待超时，可稍后在 Notion 通知中查看导出结果。")

    def download(self, url, name, directory):
        parsed = urllib.parse.urlsplit(url)
        if parsed.scheme != "https":
            raise ExportError("下载链接不是 HTTPS，已停止。")
        # 旧 API 在 notion.so 下发文件令牌，新导出链接已迁移到 file.notion.com。
        if parsed.hostname == "file.notion.com":
            for cookie in list(self.cookies):
                if cookie.name == "file_token" and cookie.domain.lstrip(".") == "notion.so":
                    cookie = copy.copy(cookie)
                    cookie.domain, cookie.domain_specified, cookie.domain_initial_dot = "file.notion.com", False, False
                    self.cookies.set_cookie(cookie)
        directory = Path(directory)
        directory.mkdir(parents=True, exist_ok=True)
        name = re.sub(r"[^\w-]", "_", name)[:60] or "Notion"
        destination = directory / (name + datetime.now().strftime("_%Y%m%d_%H%M%S_%f.zip"))
        temporary = None
        try:
            print("开始下载……", flush=True)
            with self.opener.open(url, timeout=120) as response:
                if response.status != 200:
                    raise ExportError("下载返回的 HTTP 状态异常。")
                length = response.headers.get("Content-Length")
                with tempfile.NamedTemporaryFile(dir=directory, suffix=".part", delete=False) as file:
                    temporary = Path(file.name)
                    shutil.copyfileobj(response, file, length=1024 * 1024)
            if length is not None and temporary.stat().st_size != int(length):
                raise ExportError("下载文件不完整。")
            print("检查 ZIP 完整性……", flush=True)
            with zipfile.ZipFile(temporary) as archive:
                if not archive.namelist() or archive.testzip() is not None:
                    raise ExportError("ZIP 为空或已损坏。")
            temporary.replace(destination)
            return destination
        finally:
            if temporary is not None:
                temporary.unlink(missing_ok=True)


def main():
    try:
        token = os.environ.get("NOTION_TOKEN_V2", "").strip()
        if not token:
            if not sys.stdin.isatty():
                raise ExportError("请在终端运行并输入 token_v2，或设置 NOTION_TOKEN_V2 环境变量。")
            token = getpass.getpass("粘贴 token_v2 的值（输入不显示）：").strip()
        if not token or re.search(r"[\s;]", token):
            raise ExportError("请只输入 token_v2 的值，不含 Cookie 名称或其他字段。")
        exporter = Exporter(token)
        name = exporter.prepare()
        path = exporter.download(exporter.export(), name, Path(__file__).resolve().parent / "backups")
        print("备份成功，ZIP 检查通过：\n" + str(path))
        return 0
    except KeyboardInterrupt:
        print("\n已中止；已经提交的云端导出任务可能仍会继续。")
        return 130
    except urllib.error.HTTPError as exc:
        print("请求失败，HTTP {}。401/403 请检查登录状态和权限；429 请稍后重试。".format(exc.code))
        exc.close()
    except ExportError as exc:
        print("失败：" + str(exc))
    except Exception as exc:
        # 不输出可能带令牌、签名链接或私人内容的底层异常。
        print("失败（{}），请检查网络、登录状态和保存目录后重试。".format(type(exc).__name__))
    return 1


if __name__ == "__main__":
    sys.exit(main())