PART 03 — データ項目定義書
コード値定義・マスタデータ・初期投入データ・データ移行方針で定義すべきこと

"""
業務担当者提供のマスタCSVをバリデーションしてPostgreSQLに投入するスクリプト。
pip install pandas psycopg2-binary
"""
import pandas as pd
import psycopg2
from psycopg2.extras import execute_values
from dataclasses import dataclass
from typing import Optional
import re


@dataclass
class ValidationError:
    row: int
    column: str
    message: str


def validate_category_csv(df: pd.DataFrame) -> list[ValidationError]:
    """商品カテゴリCSVのバリデーション"""
    errors = []
    required_cols = ["category_code", "category_name", "parent_code", "sort_order"]

    # 必須カラム存在チェック
    for col in required_cols:
        if col not in df.columns:
            errors.append(ValidationError(0, col, f"必須カラム '{col}' が見つかりません"))
    if errors:
        return errors

    for idx, row in df.iterrows():
        row_num = idx + 2  # ヘッダー行を含めて人間が読む行番号

        # category_code: 必須・英数字アンダースコアのみ
        code = str(row.get("category_code", "")).strip()
        if not code:
            errors.append(ValidationError(row_num, "category_code", "カテゴリコードは必須です"))
        elif not re.match(r'^[A-Za-z0-9_]+$', code):
            errors.append(ValidationError(row_num, "category_code", "半角英数字とアンダースコアのみ使用可"))

        # category_name: 必須・50文字以内
        name = str(row.get("category_name", "")).strip()
        if not name:
            errors.append(ValidationError(row_num, "category_name", "カテゴリ名は必須です"))
        elif len(name) > 50:
            errors.append(ValidationError(row_num, "category_name", f"50文字以内（現在{len(name)}文字）"))

        # sort_order: 数値
        try:
            int(row.get("sort_order", 0))
        except (ValueError, TypeError):
            errors.append(ValidationError(row_num, "sort_order", "表示順は整数で入力してください"))

    # category_code の重複チェック
    dup = df["category_code"].dropna().astype(str).str.strip()
    dup_values = dup[dup.duplicated()].tolist()
    if dup_values:
        errors.append(ValidationError(0, "category_code", f"重複コードあり: {dup_values}"))

    return errors


def insert_categories(conn, df: pd.DataFrame) -> int:
    """バリデーション済みのカテゴリCSVをDBに投入"""
    records = [
        (
            str(row["category_code"]).strip(),
            str(row["category_name"]).strip(),
            str(row["parent_code"]).strip() if pd.notna(row["parent_code"]) else None,
            int(row["sort_order"]),
        )
        for _, row in df.iterrows()
    ]
    sql = """
        INSERT INTO m_category (category_code, category_name, parent_code, sort_order)
        VALUES %s
        ON CONFLICT (category_code) DO UPDATE SET
            category_name = EXCLUDED.category_name,
            sort_order = EXCLUDED.sort_order;
    """
    with conn.cursor() as cur:
        execute_values(cur, sql, records)
    conn.commit()
    return len(records)


if __name__ == "__main__":
    CSV_FILE = "category_master.csv"
    DB_CONFIG = {"host":"localhost","port":5432,"dbname":"mydb","user":"myuser","password":"mypassword"}

    df = pd.read_csv(CSV_FILE, encoding="utf-8-sig", dtype=str)
    print(f"読込件数: {len(df)}件")

    errors = validate_category_csv(df)
    if errors:
        print(f"\n❌ バリデーションエラー {len(errors)}件:")
        for e in errors:
            print(f"  行{e.row} [{e.column}] {e.message}")
    else:
        conn = psycopg2.connect(**DB_CONFIG)
        count = insert_categories(conn, df)
        conn.close()
        print(f"\n✅ {count}件を投入しました")