【読書記録】ブクログではなくNotionで。楽天ブックスAPIで書影を自動取得

from __future__ import annotations

import argparse
import os
import re
import time
import unicodedata
from pathlib import Path
from typing import Any

import requests
from dotenv import load_dotenv


NOTION_API_BASE = "https://api.notion.com/v1"
NOTION_VERSION = "2026-03-11"
RAKUTEN_API_URL = (
    "https://openapi.rakuten.co.jp/services/api/BooksBook/Search/20170404"
)

TITLE_PROPERTY = "本のタイトル"
AUTHOR_PROPERTY = "著者"


def load_settings() -> dict[str, str]:
    env_path = Path(__file__).resolve().parent / ".env"
    load_dotenv(env_path)

    keys = [
        "NOTION_TOKEN",
        "NOTION_DATABASE_ID",
        "RAKUTEN_APP_ID",
        "RAKUTEN_ACCESS_KEY",
    ]
    settings = {key: os.getenv(key, "").strip() for key in keys}
    missing = [key for key, value in settings.items() if not value]

    if missing:
        raise RuntimeError(
            ".env に次の設定がありません: " + ", ".join(missing)
        )

    return settings


def notion_headers(token: str) -> dict[str, str]:
    return {
        "Authorization": f"Bearer {token}",
        "Notion-Version": NOTION_VERSION,
        "Content-Type": "application/json",
    }


def request_json(
    method: str,
    url: str,
    *,
    headers: dict[str, str],
    params: dict[str, Any] | None = None,
    json_body: dict[str, Any] | None = None,
) -> dict[str, Any]:
    response = requests.request(
        method,
        url,
        headers=headers,
        params=params,
        json=json_body,
        timeout=30,
    )

    if response.ok:
        return response.json()

    try:
        detail = response.json()
    except ValueError:
        detail = response.text

    raise RuntimeError(
        f"APIエラー: {method} {url}\n"
        f"HTTP {response.status_code}\n{detail}"
    )


def get_data_source_id(
    database_id: str,
    headers: dict[str, str],
) -> str:
    database = request_json(
        "GET",
        f"{NOTION_API_BASE}/databases/{database_id}",
        headers=headers,
    )

    data_sources = database.get("data_sources", [])
    if not data_sources:
        raise RuntimeError("Notionデータベース内にデータソースがありません。")

    # 今回のDBは1データソースなので先頭を使います。
    return data_sources[0]["id"]


def query_pages(
    data_source_id: str,
    headers: dict[str, str],
) -> list[dict[str, Any]]:
    pages: list[dict[str, Any]] = []
    cursor: str | None = None

    while True:
        body: dict[str, Any] = {"page_size": 100}
        if cursor:
            body["start_cursor"] = cursor

        result = request_json(
            "POST",
            f"{NOTION_API_BASE}/data_sources/{data_source_id}/query",
            headers=headers,
            json_body=body,
        )

        pages.extend(result.get("results", []))

        if not result.get("has_more"):
            break

        cursor = result.get("next_cursor")
        if not cursor:
            break

    return pages


def rich_text_to_plain(items: list[dict[str, Any]]) -> str:
    return "".join(item.get("plain_text", "") for item in items).strip()


def get_page_text(page: dict[str, Any], property_name: str) -> str:
    prop = page.get("properties", {}).get(property_name, {})
    prop_type = prop.get("type")

    if prop_type == "title":
        return rich_text_to_plain(prop.get("title", []))

    if prop_type == "rich_text":
        return rich_text_to_plain(prop.get("rich_text", []))

    return ""


def normalize_text(value: str) -> str:
    value = unicodedata.normalize("NFKC", value).lower()
    value = re.sub(r"\s+", "", value)
    value = re.sub(r"[「」『』（）()\[\]【】・:：,，.。!！?？\-ー_]", "", value)
    return value


def normalize_items(data: dict[str, Any]) -> list[dict[str, Any]]:
    raw_items = data.get("Items")
    if raw_items is None:
        raw_items = data.get("items", [])

    items: list[dict[str, Any]] = []
    for raw in raw_items or []:
        if not isinstance(raw, dict):
            continue

        item = raw.get("Item")
        if item is None:
            item = raw.get("item")
        if item is None:
            item = raw

        if isinstance(item, dict):
            items.append(item)

    return items


def rakuten_search(
    *,
    title: str,
    author: str,
    app_id: str,
    access_key: str,
) -> list[dict[str, Any]]:
    params: dict[str, str | int] = {
        "applicationId": app_id,
        "format": "json",
        "formatVersion": 2,
        "title": title,
        "hits": 10,
        "outOfStockFlag": 1,
    }

    if author:
        params["author"] = author

    headers = {
        "accessKey": access_key,
        "User-Agent": "NotionBookCoverSync/1.0",
    }

    data = request_json(
        "GET",
        RAKUTEN_API_URL,
        headers=headers,
        params=params,
    )
    return normalize_items(data)


def score_book(
    item: dict[str, Any],
    wanted_title: str,
    wanted_author: str,
) -> int:
    item_title = normalize_text(str(item.get("title", "")))
    item_author = normalize_text(str(item.get("author", "")))
    wanted_title_n = normalize_text(wanted_title)
    wanted_author_n = normalize_text(wanted_author)

    score = 0

    if item_title == wanted_title_n:
        score += 100
    elif wanted_title_n and wanted_title_n in item_title:
        score += 70
    elif item_title and item_title in wanted_title_n:
        score += 50

    if wanted_author_n:
        if item_author == wanted_author_n:
            score += 40
        elif wanted_author_n in item_author:
            score += 30
        elif item_author and item_author in wanted_author_n:
            score += 20

    if item.get("largeImageUrl"):
        score += 5

    return score


def find_best_book(
    *,
    title: str,
    author: str,
    app_id: str,
    access_key: str,
) -> dict[str, Any] | None:
    # まずタイトル＋著者で検索。
    items = rakuten_search(
        title=title,
        author=author,
        app_id=app_id,
        access_key=access_key,
    )

    # 見つからない場合のみタイトル単独へ緩める。
    if not items and author:
        items = rakuten_search(
            title=title,
            author="",
            app_id=app_id,
            access_key=access_key,
        )

    candidates = [
        item for item in items
        if item.get("largeImageUrl")
    ]
    if not candidates:
        return None

    return max(
        candidates,
        key=lambda item: score_book(item, title, author),
    )


def set_page_cover(
    page_id: str,
    image_url: str,
    headers: dict[str, str],
) -> None:
    request_json(
        "PATCH",
        f"{NOTION_API_BASE}/pages/{page_id}",
        headers=headers,
        json_body={
            "cover": {
                "type": "external",
                "external": {"url": image_url},
            }
        },
    )


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Notion読書DBのカバー未設定ページに楽天ブックスの書影を設定します。"
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        help="実際にNotionへ書影を設定します。指定しない場合は確認だけです。",
    )
    parser.add_argument(
        "--limit",
        type=int,
        default=1,
        help="処理する最大件数。初期値は1件です。",
    )
    parser.add_argument(
        "--title",
        help="この文字を含む本だけを対象にします。",
    )
    args = parser.parse_args()

    if args.limit < 1:
        raise RuntimeError("--limit は1以上で指定してください。")

    settings = load_settings()
    n_headers = notion_headers(settings["NOTION_TOKEN"])

    print("Notion読書DBを確認しています...")
    data_source_id = get_data_source_id(
        settings["NOTION_DATABASE_ID"],
        n_headers,
    )
    pages = query_pages(data_source_id, n_headers)

    targets: list[dict[str, Any]] = []
    for page in pages:
        title = get_page_text(page, TITLE_PROPERTY)
        author = get_page_text(page, AUTHOR_PROPERTY)

        if not title:
            continue
        if page.get("cover") is not None:
            continue
        if args.title and args.title not in title:
            continue

        targets.append(
            {
                "page_id": page["id"],
                "title": title,
                "author": author,
            }
        )

    print(f"カバー未設定の対象: {len(targets)}件")

    if not targets:
        print("処理対象はありません。")
        return 0

    mode = "書影を設定" if args.apply else "確認のみ"
    print(f"実行モード: {mode}")
    print(f"今回の上限: {args.limit}件")

    processed = 0
    updated = 0
    not_found = 0

    for target in targets[: args.limit]:
        processed += 1
        title = target["title"]
        author = target["author"]

        print("\n" + "=" * 70)
        print(f"[{processed}] {title}")
        print(f"Notion著者: {author or '(空欄)'}")

        book = find_best_book(
            title=title,
            author=author,
            app_id=settings["RAKUTEN_APP_ID"],
            access_key=settings["RAKUTEN_ACCESS_KEY"],
        )

        if not book:
            not_found += 1
            print("楽天ブックスで書影付きの本が見つかりませんでした。")
            continue

        image_url = str(book.get("largeImageUrl", ""))
        print(f"候補タイトル: {book.get('title', '')}")
        print(f"候補著者: {book.get('author', '')}")
        print(f"ISBN: {book.get('isbn', '')}")
        print(f"書影URL: {image_url}")

        if args.apply:
            set_page_cover(
                target["page_id"],
                image_url,
                n_headers,
            )
            updated += 1
            print("→ Notionのページカバーを設定しました。")
        else:
            print("→ 確認のみのため、Notionは変更していません。")

        time.sleep(1)

    print("\n" + "=" * 70)
    print("処理結果")
    print(f"確認した件数: {processed}")
    print(f"設定した件数: {updated}")
    print(f"見つからなかった件数: {not_found}")

    if not args.apply:
        print(
            "\n候補が正しければ、次のコマンドで1件だけ実際に設定します。"
            "\npy .\\notion_book_cover_sync.py --apply --limit 1"
        )

    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except requests.RequestException as exc:
        print(f"通信エラー: {exc}")
        raise SystemExit(1)
    except Exception as exc:
        print(exc)
        raise SystemExit(1)