paper_db/paper_db_common/scripts/validate_db.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable

from common import CONFIG, resolve_pdf_path


@dataclass(frozen=True)
class MissingPdf:
    """見つからないPDFの情報。"""

    paper_id: int
    title: str
    pdf_path: str
    resolved_path: str | None
    reason: str


@dataclass(frozen=True)
class IntegrityResult:
    """SQLite整合性検証の結果。"""

    integrity: str
    foreign_key_count: int
    errors: tuple[str, ...]


def parse_args() -> argparse.Namespace:
    """コマンドライン引数を解析する。"""
    parser = argparse.ArgumentParser(
        description="papers.dbの整合性を検証する"
    )
    parser.add_argument(
        "database",
        nargs="?",
        default=str(CONFIG.path.database),
        help="検証するSQLiteデータベース",
    )
    parser.add_argument(
        "--pdf-root",
        default=str(CONFIG.path.pdf_root),
        help="PDF_ROOTのパス",
    )
    return parser.parse_args()


def open_database(database: Path) -> sqlite3.Connection:
    """SQLiteデータベースへ接続する。"""
    connection = sqlite3.connect(database)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys=ON")
    return connection


def check_integrity(
    connection: sqlite3.Connection,
) -> IntegrityResult:
    """SQLite本体と外部キーの整合性を検証する。"""
    errors: list[str] = []

    integrity = connection.execute(
        "PRAGMA integrity_check"
    ).fetchone()[0]

    if integrity != "ok":
        errors.append(f"integrity_check: {integrity}")

    foreign_keys = connection.execute(
        "PRAGMA foreign_key_check"
    ).fetchall()

    foreign_key_count = len(foreign_keys)

    if foreign_key_count:
        errors.append(
            f"foreign keys: {foreign_key_count}"
        )

    return IntegrityResult(
        integrity=integrity,
        foreign_key_count=foreign_key_count,
        errors=tuple(errors),
    )


def validation_queries() -> tuple[tuple[str, str], ...]:
    """DB内容の検証用SQLを返す。"""
    return (
        (
            "duplicate DOI",
            """
            SELECT doi
            FROM papers
            WHERE doi IS NOT NULL
              AND doi <> ''
            GROUP BY doi
            HAVING COUNT(*) > 1
            """,
        ),
        (
            "duplicate SHA256",
            """
            SELECT file_sha256
            FROM papers
            WHERE file_sha256 IS NOT NULL
              AND file_sha256 <> ''
            GROUP BY file_sha256
            HAVING COUNT(*) > 1
            """,
        ),
        (
            "empty title",
            """
            SELECT id
            FROM papers
            WHERE title IS NULL
               OR TRIM(title) = ''
            """,
        ),
        (
            "invalid relevance",
            """
            SELECT id
            FROM papers
            WHERE relevance NOT BETWEEN 0 AND 5
            """,
        ),
        (
            "invalid priority",
            """
            SELECT id
            FROM papers
            WHERE priority NOT IN ('A', 'B', 'C', 'D')
            """,
        ),
    )


def check_database_values(
    connection: sqlite3.Connection,
) -> list[str]:
    """重複や値域を検証する。"""
    errors: list[str] = []

    for label, sql in validation_queries():
        rows = connection.execute(sql).fetchall()

        if rows:
            errors.append(f"{label}: {len(rows)}")

    return errors


def iter_papers_with_pdf(
    connection: sqlite3.Connection,
) -> Iterable[sqlite3.Row]:
    """PDFパスが登録された論文を返す。"""
    sql = """
        SELECT id, title, pdf_path
        FROM papers
        WHERE pdf_path IS NOT NULL
          AND TRIM(pdf_path) <> ''
        ORDER BY id
    """
    return connection.execute(sql)


def inspect_pdf(
    pdf_root: Path,
    row: sqlite3.Row,
) -> MissingPdf | None:
    """1件のPDFが存在するか確認する。"""
    try:
        path = resolve_pdf_path(
            pdf_root,
            row["pdf_path"],
        )

        if path.is_file():
            return None

        return MissingPdf(
            paper_id=row["id"],
            title=row["title"],
            pdf_path=row["pdf_path"],
            resolved_path=str(path),
            reason="file does not exist",
        )

    except Exception as exc:
        return MissingPdf(
            paper_id=row["id"],
            title=row["title"],
            pdf_path=row["pdf_path"],
            resolved_path=None,
            reason=f"{type(exc).__name__}: {exc}",
        )


def find_missing_pdfs(
    connection: sqlite3.Connection,
    pdf_root: Path,
) -> list[MissingPdf]:
    """存在しないPDFを検索する。"""
    missing: list[MissingPdf] = []

    for row in iter_papers_with_pdf(connection):
        item = inspect_pdf(pdf_root, row)

        if item is not None:
            missing.append(item)

    return missing


def print_missing_pdfs(
    missing: Iterable[MissingPdf],
) -> None:
    """見つからないPDFの詳細を表示する。"""
    items = list(missing)

    if not items:
        return

    print()
    print("=== Missing PDFs ===")

    for item in items:
        print(f"paper_id: {item.paper_id}")
        print(f"title: {item.title}")
        print(f"pdf_path: {item.pdf_path}")
        print(
            "resolved_path:",
            item.resolved_path
            or "(resolve failed)",
        )
        print(f"reason: {item.reason}")
        print()


def print_summary(
    integrity: IntegrityResult,
    errors: list[str],
) -> None:
    """検証結果を表示する。"""
    print(f"integrity_check: {integrity.integrity}")
    print(
        "foreign_key_violations:",
        integrity.foreign_key_count,
    )
    print(f"errors: {len(errors)}")

    for error in errors:
        print(f"- {error}")


def validate(
    database: Path,
    pdf_root: Path,
) -> list[str]:
    """データベース全体を検証する。"""
    with open_database(database) as connection:
        integrity = check_integrity(connection)
        value_errors = check_database_values(
            connection
        )
        missing = find_missing_pdfs(
            connection,
            pdf_root,
        )

    errors = list(integrity.errors)
    errors.extend(value_errors)

    if missing:
        errors.append(
            f"missing PDF: {len(missing)}"
        )

    print_missing_pdfs(missing)
    print_summary(integrity, errors)

    return errors


def main() -> int:
    """メイン処理。"""
    args = parse_args()

    database = Path(
        args.database
    ).expanduser().resolve()

    pdf_root = Path(
        args.pdf_root
    ).expanduser().resolve()

    errors = validate(
        database,
        pdf_root,
    )

    return 1 if errors else 0


if __name__ == "__main__":
    raise SystemExit(main())