설문 CSV를 집계하기 전에 응답수·누락·중복 점검하기

python example.py

"""설문 합성 CSV의 중복 ID와 무응답을 점검한 뒤 유효 응답을 집계합니다."""

import csv
import json
import re
import sys
from collections import Counter
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path

BASE = Path(__file__).resolve().parent
INPUT = BASE / "responses.csv"
OUTPUT = BASE / "outputs"
CHOICES = ["예", "아니오", "잘 모르겠음"]


def percent(numerator, denominator):
    # 분모가 0이면 0%로 오해하지 않도록 계산 불가를 None으로 남깁니다.
    if denominator == 0:
        return None
    value = Decimal(numerator) * Decimal(100) / Decimal(denominator)
    return str(value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))


def read_rows():
    if INPUT.is_symlink() or not INPUT.is_file():
        raise ValueError("responses.csv는 일반 파일이어야 합니다.")
    rows = []
    with INPUT.open("r", encoding="utf-8-sig", newline="") as stream:
        reader = csv.DictReader(stream, strict=True)
        if reader.fieldnames != ["respondent_id", "answer"]:
            raise ValueError("열 이름과 순서는 respondent_id,answer여야 합니다.")
        for number, row in enumerate(reader, start=1):
            if None in row or any(value is None for value in row.values()):
                raise ValueError(f"데이터 {number}행: 열 수를 확인하세요.")
            respondent_id, answer = row["respondent_id"].strip(), row["answer"].strip()
            if not re.fullmatch(r"R[0-9]{3}", respondent_id):
                raise ValueError(f"데이터 {number}행: ID는 R001처럼 R과 숫자 세 자리여야 합니다.")
            if answer and answer not in CHOICES:
                raise ValueError(f"데이터 {number}행: 알 수 없는 응답 {answer!r}")
            rows.append({"data_row": number, "respondent_id": respondent_id, "answer": answer})
    return rows


def summarize(rows):
    frequencies = Counter(row["respondent_id"] for row in rows)
    duplicate_ids = sorted(key for key, count in frequencies.items() if count > 1)
    duplicates = set(duplicate_ids)
    counts = Counter()
    audit, blank_after_duplicates = [], 0
    for row in rows:
        # 중복 그룹에서는 첫 응답이나 마지막 응답을 임의로 고르지 않습니다.
        if row["respondent_id"] in duplicates:
            status = "duplicate_id"
        elif not row["answer"]:
            status = "blank_answer"
            blank_after_duplicates += 1
        else:
            status = "valid"
            counts[row["answer"]] += 1
        audit.append({**row, "status": status})

    duplicate_rows = sum(frequencies[key] for key in duplicate_ids)
    eligible = len(rows) - duplicate_rows  # 여기에는 빈 답변을 한 고유 ID도 포함됩니다.
    valid = sum(counts.values())
    summary = {
        "question": "다시 이용할 의향이 있나요?",
        "duplicate_policy": "exclude_all_rows_with_duplicate_id",
        "raw_rows": len(rows),
        "unique_ids": len(frequencies),
        "duplicate_ids": duplicate_ids,
        "duplicate_id_count": len(duplicate_ids),
        "excluded_duplicate_rows": duplicate_rows,
        "raw_blank_answer_rows": sum(not row["answer"] for row in rows),
        "eligible_unique_respondents": eligible,
        "blank_answer_rows_after_duplicates": blank_after_duplicates,
        "valid_answer_rows": valid,
        "completion_rate_percent": percent(valid, eligible),
        "choices": [
            {"answer": answer, "count": counts[answer], "denominator": valid,
             "percent": percent(counts[answer], valid)} for answer in CHOICES
        ],
    }
    return summary, audit


def main():
    if OUTPUT.exists() or OUTPUT.is_symlink():
        raise FileExistsError("outputs가 이미 있습니다. 기존 결과를 옮긴 뒤 실행하세요.")
    summary, audit = summarize(read_rows())  # 입력 검증과 집계를 마친 뒤 출력합니다.
    OUTPUT.mkdir()
    with (OUTPUT / "summary.json").open("x", encoding="utf-8") as stream:
        json.dump(summary, stream, ensure_ascii=False, indent=2)
        stream.write("\n")
    with (OUTPUT / "choices.csv").open("x", encoding="utf-8-sig", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=["answer", "count", "denominator", "percent"])
        writer.writeheader()
        writer.writerows(summary["choices"])
    with (OUTPUT / "audit_rows.csv").open("x", encoding="utf-8-sig", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=["data_row", "respondent_id", "answer", "status"])
        writer.writeheader()
        writer.writerows(audit)
    print(f"원시 {summary['raw_rows']}행 / 고유 ID {summary['unique_ids']}개")
    print(f"중복 제외 {summary['excluded_duplicate_rows']}행 / 남은 고유 응답자 {summary['eligible_unique_respondents']}명")
    print(f"빈 응답 {summary['blank_answer_rows_after_duplicates']}행 / 유효 응답 {summary['valid_answer_rows']}행")
    if not summary["valid_answer_rows"]:
        print("유효 응답이 없어 선택지 비율을 계산하지 않았습니다.")
    print(f"완료: {OUTPUT}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, csv.Error) as error:
        print(f"중지: {error}", file=sys.stderr)
        raise SystemExit(2)