担当者・保存場所・未解決事項を含むプロジェクト引き継ぎチェックリストを作る

item_id,handover_item,owner,location,status,open_issue
H001,Source code,Min,repo/main,READY,
H002,Test dataset,Jae,shared-drive/project/data,READY,
H003,Calibration procedure,Min,shared-drive/project/docs/calibration.pdf,OPEN,Final approval is still pending
H004,Supplier contact list,,shared-drive/project/admin/suppliers.csv,READY,
H005,Open bug list,Jae,,OPEN,Two export bugs remain under review

import csv
from pathlib import Path

SOURCE = Path("handover_items.csv")
OUTPUT_DIR = Path("outputs") / "handover_check_result"
REPORT = OUTPUT_DIR / "handover_review.csv"
REQUIRED_COLUMNS = [
    "item_id", "handover_item", "owner", "location", "status", "open_issue"
]
VALID_STATUS = {"READY", "OPEN"}


def add_issue(issues, row_number, item_id, issue):
    issues.append({
        "row_number": row_number,
        "item_id": item_id,
        "issue": issue,
    })


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}")

    issues = []
    seen_ids = set()
    item_count = 0
    open_count = 0

    with SOURCE.open("r", encoding="utf-8-sig", newline="") as stream:
        reader = csv.DictReader(stream)
        if reader.fieldnames != REQUIRED_COLUMNS:
            raise ValueError(
                f"Header mismatch. Expected {REQUIRED_COLUMNS}, got {reader.fieldnames}."
            )

        for row_number, row in enumerate(reader, start=2):
            item_count += 1
            item_id = row["item_id"].strip()
            handover_item = row["handover_item"].strip()
            owner = row["owner"].strip()
            location = row["location"].strip()
            status = row["status"].strip()
            open_issue = row["open_issue"].strip()

            if not item_id:
                add_issue(issues, row_number, item_id, "MISSING_ITEM_ID")
            elif item_id in seen_ids:
                add_issue(issues, row_number, item_id, "DUPLICATE_ITEM_ID")
            seen_ids.add(item_id)

            if not handover_item:
                add_issue(issues, row_number, item_id, "MISSING_ITEM_NAME")
            if not owner:
                add_issue(issues, row_number, item_id, "MISSING_OWNER")
            if not location:
                add_issue(issues, row_number, item_id, "MISSING_LOCATION")
            if status not in VALID_STATUS:
                add_issue(issues, row_number, item_id, "INVALID_STATUS")
            elif status == "OPEN":
                open_count += 1
                if not open_issue:
                    add_issue(issues, row_number, item_id, "MISSING_OPEN_ISSUE")
            elif open_issue:
                add_issue(issues, row_number, item_id, "READY_HAS_OPEN_ISSUE")

    OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_DIR.mkdir()
    with REPORT.open("x", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(
            stream,
            fieldnames=["row_number", "item_id", "issue"],
        )
        writer.writeheader()
        writer.writerows(issues)

    affected_items = len({row["item_id"] for row in issues})
    print(f"Checklist items: {item_count}.")
    print(f"Open items: {open_count}.")
    print(f"Validation issues: {len(issues)}.")
    print(f"Affected items: {affected_items}.")
    print(f"Report: {REPORT.as_posix()}")


if __name__ == "__main__":
    main()

row_number,item_id,issue
5,H004,MISSING_OWNER
6,H005,MISSING_LOCATION

Checklist items: 5.
Open items: 2.
Validation issues: 2.
Affected items: 2.
Report: outputs/handover_check_result/handover_review.csv