csv_loader.py

import csv
from pathlib import Path

from models import Question


def load_questions(csv_path: str) -> list[Question]:
    path = Path(csv_path)
    if not path.exists():
        raise FileNotFoundError(f"質問ファイルが見つかりません: {csv_path}")

    questions: list[Question] = []

    with open(path, encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)

        required_columns = {
            "id", "category", "question_text", "expected_keywords",
            "keyword_weight", "ai_weight", "improv_weight",
            "max_score", "scoring_criteria",
        }
        if reader.fieldnames is None:
            raise ValueError("CSVファイルにヘッダーがありません。")

        actual_columns = set(reader.fieldnames)
        missing = required_columns - actual_columns
        if missing:
            raise ValueError(f"CSVに必須カラムがありません: {missing}")

        for row_num, row in enumerate(reader, start=2):
            try:
                keyword_weight = float(row["keyword_weight"])
                ai_weight = float(row["ai_weight"])
                improv_weight = float(row["improv_weight"])
                max_score = int(row["max_score"])

                weight_sum = keyword_weight + ai_weight + improv_weight
                if abs(weight_sum - 1.0) > 0.01:
                    raise ValueError(
                        f"重みの合計が1.0ではありません: {weight_sum:.2f}"
                    )

                if max_score <= 0:
                    raise ValueError(f"max_scoreは1以上必要です: {max_score}")

                keywords = [
                    kw.strip()
                    for kw in row["expected_keywords"].split(",")
                    if kw.strip()
                ]
                if not keywords:
                    raise ValueError("expected_keywordsが空です。")

                question = Question(
                    id=int(row["id"]),
                    category=row["category"].strip(),
                    question_text=row["question_text"].strip(),
                    expected_keywords=keywords,
                    keyword_weight=keyword_weight,
                    ai_weight=ai_weight,
                    improv_weight=improv_weight,
                    max_score=max_score,
                    scoring_criteria=row["scoring_criteria"].strip(),
                    follow_up=row.get("follow_up", "").strip(),
                )
                questions.append(question)

            except (ValueError, KeyError) as e:
                raise ValueError(f"CSV {row_num}行目にエラー: {e}") from e

    if not questions:
        raise ValueError("CSVに質問が1つもありません。")

    return questions