CSV で参考文献リストを管理し、DOI の重複や欠落を確認する

ref_id,title,year,doi
R001,Synthetic bracket study,2024,10.0000/demo.alpha
R002,Synthetic vibration study,2025,10.0000/demo.beta
R003,Synthetic bracket study copy,2024,https://doi.org/10.0000/demo.alpha
R004,Synthetic fatigue note,2023,
R005,Synthetic surrogate study,2026,doi:10.0000/DEMO.GAMMA
R006,Synthetic optimization study,2026,10.0000/demo.delta

import csv
from pathlib import Path

SOURCE = Path("references.csv")
OUTPUT_DIR = Path("outputs") / "reference_list_result"
NORMALIZED = OUTPUT_DIR / "normalized_references.csv"
ISSUES = OUTPUT_DIR / "issues.csv"
REQUIRED = {"ref_id", "title", "year", "doi"}


def normalize_doi(value: str) -> str:
    text = value.strip()
    lowered = text.lower()
    prefixes = (
        "https://doi.org/",
        "http://doi.org/",
        "doi:",
    )
    for prefix in prefixes:
        if lowered.startswith(prefix):
            text = text[len(prefix):].strip()
            break
    return text.lower()


def main() -> None:
    if not SOURCE.is_file():
        raise FileNotFoundError(f"Source CSV not found: {SOURCE}")
    if OUTPUT_DIR.exists():
        raise FileExistsError(f"Output folder already exists: {OUTPUT_DIR}")

    rows = []
    with SOURCE.open("r", encoding="utf-8-sig", newline="") as stream:
        reader = csv.DictReader(stream)
        if reader.fieldnames is None or not REQUIRED.issubset(reader.fieldnames):
            raise ValueError("CSV is missing a required column.")

        seen_ids = set()
        for row in reader:
            ref_id = row["ref_id"].strip()
            if not ref_id:
                raise ValueError("ref_id must not be blank.")
            if ref_id in seen_ids:
                raise ValueError(f"Duplicate ref_id: {ref_id}")
            seen_ids.add(ref_id)

            rows.append({
                "ref_id": ref_id,
                "title": row["title"],
                "year": row["year"],
                "doi": row["doi"],
                "normalized_doi": normalize_doi(row["doi"]),
            })

    by_doi = {}
    for row in rows:
        doi = row["normalized_doi"]
        if doi:
            by_doi.setdefault(doi, []).append(row["ref_id"])

    duplicate_dois = {
        doi: ids for doi, ids in by_doi.items() if len(ids) > 1
    }

    issue_rows = []
    for row in rows:
        doi = row["normalized_doi"]
        if not doi:
            issue_rows.append({
                "ref_id": row["ref_id"],
                "issue": "MISSING_DOI",
                "normalized_doi": "",
                "related_ref_ids": "",
            })
        elif doi in duplicate_dois:
            issue_rows.append({
                "ref_id": row["ref_id"],
                "issue": "DUPLICATE_DOI",
                "normalized_doi": doi,
                "related_ref_ids": ";".join(duplicate_dois[doi]),
            })

    OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_DIR.mkdir()

    normalized_fields = ["ref_id", "title", "year", "doi", "normalized_doi"]
    with NORMALIZED.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=normalized_fields)
        writer.writeheader()
        writer.writerows(rows)

    issue_fields = ["ref_id", "issue", "normalized_doi", "related_ref_ids"]
    with ISSUES.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=issue_fields)
        writer.writeheader()
        writer.writerows(issue_rows)

    missing_count = sum(row["issue"] == "MISSING_DOI" for row in issue_rows)
    print(f"References checked: {len(rows)}.")
    print(f"Missing DOI rows: {missing_count}.")
    print(f"Duplicate DOI groups: {len(duplicate_dois)}.")
    print(f"Issue rows: {len(issue_rows)}.")
    print(f"Output folder: {OUTPUT_DIR.as_posix()}")


if __name__ == "__main__":
    main()

ref_id,issue,normalized_doi,related_ref_ids
R001,DUPLICATE_DOI,10.0000/demo.alpha,R001;R003
R003,DUPLICATE_DOI,10.0000/demo.alpha,R001;R003
R004,MISSING_DOI,,

References checked: 6.
Missing DOI rows: 1.
Duplicate DOI groups: 1.
Issue rows: 3.
Output folder: outputs/reference_list_result