paper_db/paper_db_common/scripts/init_db.py

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

import argparse
import sqlite3
from pathlib import Path

from common import CONFIG

CATEGORIES = CONFIG.domain.categories

SCHEMA = """
PRAGMA foreign_keys = ON;

CREATE TABLE IF NOT EXISTS categories (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL UNIQUE,
    description TEXT
);

CREATE TABLE IF NOT EXISTS papers (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    doi TEXT,
    title TEXT NOT NULL,
    title_normalized TEXT NOT NULL,
    first_author TEXT,
    year INTEGER,
    journal TEXT,
    volume TEXT,
    issue TEXT,
    pages TEXT,
    publisher TEXT,
    article_type TEXT,
    language TEXT,
    abstract TEXT,
    summary_ja TEXT,
    relevance INTEGER NOT NULL DEFAULT 0 CHECK (relevance BETWEEN 0 AND 5),
    priority TEXT NOT NULL DEFAULT 'D' CHECK (priority IN ('A','B','C','D')),
    review_status TEXT NOT NULL DEFAULT 'unreviewed',
    category_id INTEGER,
    pdf_path TEXT,
    file_size INTEGER,
    file_mtime TEXT,
    file_sha256 TEXT,
    source TEXT,
    landing_url TEXT,
    pdf_url TEXT,
    url_source TEXT,
    access_status TEXT,
    url_checked_at TEXT,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    checked_at TEXT,
    notes TEXT,
    FOREIGN KEY (category_id) REFERENCES categories(id)
);

CREATE TABLE IF NOT EXISTS authors (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    name_normalized TEXT NOT NULL,
    orcid TEXT,
    affiliation TEXT
);

CREATE TABLE IF NOT EXISTS paper_authors (
    paper_id INTEGER NOT NULL,
    author_id INTEGER NOT NULL,
    author_order INTEGER NOT NULL,
    is_corresponding INTEGER NOT NULL DEFAULT 0 CHECK (is_corresponding IN (0,1)),
    PRIMARY KEY (paper_id, author_order),
    UNIQUE (paper_id, author_id),
    FOREIGN KEY (paper_id) REFERENCES papers(id) ON DELETE CASCADE,
    FOREIGN KEY (author_id) REFERENCES authors(id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS keywords (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    name_normalized TEXT NOT NULL UNIQUE
);

CREATE TABLE IF NOT EXISTS paper_keywords (
    paper_id INTEGER NOT NULL,
    keyword_id INTEGER NOT NULL,
    PRIMARY KEY (paper_id, keyword_id),
    FOREIGN KEY (paper_id) REFERENCES papers(id) ON DELETE CASCADE,
    FOREIGN KEY (keyword_id) REFERENCES keywords(id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS processing_log (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    pdf_path TEXT,
    paper_id INTEGER,
    status TEXT NOT NULL CHECK (status IN ('registered','updated','duplicate','skipped','error')),
    message TEXT,
    processed_at TEXT NOT NULL,
    FOREIGN KEY (paper_id) REFERENCES papers(id) ON DELETE SET NULL
);

CREATE UNIQUE INDEX IF NOT EXISTS idx_papers_doi
ON papers(doi) WHERE doi IS NOT NULL AND doi <> '';

CREATE UNIQUE INDEX IF NOT EXISTS idx_papers_sha256
ON papers(file_sha256) WHERE file_sha256 IS NOT NULL AND file_sha256 <> '';

CREATE INDEX IF NOT EXISTS idx_papers_title_normalized ON papers(title_normalized);
CREATE INDEX IF NOT EXISTS idx_papers_year ON papers(year);
CREATE INDEX IF NOT EXISTS idx_papers_relevance ON papers(relevance);
CREATE INDEX IF NOT EXISTS idx_papers_priority ON papers(priority);
CREATE INDEX IF NOT EXISTS idx_papers_category_id ON papers(category_id);
CREATE INDEX IF NOT EXISTS idx_authors_name_normalized ON authors(name_normalized);
CREATE INDEX IF NOT EXISTS idx_keywords_name_normalized ON keywords(name_normalized);
"""

URL_COLUMNS = {
    "landing_url": "TEXT",
    "pdf_url": "TEXT",
    "url_source": "TEXT",
    "access_status": "TEXT",
    "url_checked_at": "TEXT",
}


def ensure_url_columns(con: sqlite3.Connection) -> None:
    columns = {
        row[1] for row in con.execute("PRAGMA table_info(papers)")
    }
    for name, column_type in URL_COLUMNS.items():
        if name not in columns:
            con.execute(
                f'ALTER TABLE papers ADD COLUMN "{name}" {column_type}'
            )


def initialize(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(path) as con:
        con.executescript(SCHEMA)
        ensure_url_columns(con)
        con.executemany(
            "INSERT OR IGNORE INTO categories(name, description) VALUES (?, NULL)",
            [(name,) for name in CATEGORIES],
        )
        result = con.execute("PRAGMA integrity_check").fetchone()[0]
        if result != "ok":
            raise RuntimeError(result)

def main() -> int:
    p = argparse.ArgumentParser()
    p.add_argument(
        "database", nargs="?", default=str(CONFIG.path.database)
    )
    args = p.parse_args()
    initialize(Path(args.database))
    print(f"Initialized: {args.database}")
    print("integrity_check: ok")
    return 0

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