src/studypulse/io.py - master - nevers/hackINTERPOL

src/studypulse/io.py
178 строк5 KB
nevers
update src/studypulse/io.py
13 янв 2026, 17:33
13 янв 2026, 17:3362f8983
"""CSV reading and validation for StudyPulse."""

from pathlib import Path
import sys
import pandas


REQUIRED_COLUMNS = [
    "student_id",
    "student_name",
    "assignment_score",
    "assignments_submitted",
    "attendance_rate",
    "days_since_last_submission",
    "days_to_deadline",
]

OPTIONAL_COLUMNS = ["comments"]


def read_csv(file_path: str | Path) -> pandas.DataFrame:
    """Read CSV file into a pandas DataFrame.

    Args:
        file_path: Path to the CSV file.

    Returns:
        pandas DataFrame containing the student data.

    Raises:
        FileNotFoundError: If the file does not exist.
        ValueError: If the file is empty.
    """
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError(f"CSV file not found: {file_path}")

    try:
        df = pandas.read_csv(path)
    except pandas.errors.EmptyDataError as err:
        raise ValueError(f"CSV file is empty: {file_path}") from err

    return df


def validate_schema(df: pandas.DataFrame) -> tuple[bool, list[str]]:
    """Validate that DataFrame has required columns.

    Args:
        df: pandas DataFrame to validate.

    Returns:
        Tuple of (is_valid, list_of_errors).
    """
    errors: list[str] = []
    if df.empty:
        errors.append("No data to validate schema.")
        return False, errors

    missing_cols = set(REQUIRED_COLUMNS) - set(df.columns)
    if missing_cols:
        errors.append(f"Missing required columns: {', '.join(sorted(missing_cols))}")

    return len(errors) == 0, errors


def validate_ranges(df: pandas.DataFrame) -> tuple[bool, list[str]]:
    """Validate that numeric columns are within expected ranges.

    Args:
        df: pandas DataFrame to validate.

    Returns:
        Tuple of (is_valid, list_of_errors).
    """
    errors: list[str] = []

    for idx, row in df.iterrows():
        row_num = idx + 1

        # Validate assignment_score (0-100)
        val = row.get("assignment_score")
        try:
            score = float(val)
            if score < 0 or score > 100:
                errors.append(
                    f"Row {row_num}: assignment_score out of range [0, 100]: {val}"
                )
        except (TypeError, ValueError):
            errors.append(f"Row {row_num}: assignment_score invalid or missing: {val}")

        # Validate assignments_submitted (non-negative integer)
        val = row.get("assignments_submitted")
        try:
            submitted = int(val)
            if submitted < 0:
                errors.append(
                    f"Row {row_num}: assignments_submitted must be >= 0: {val}"
                )
        except (TypeError, ValueError):
            errors.append(f"Row {row_num}: assignments_submitted invalid or missing: {val}")

        # Validate attendance_rate (0-1)
        val = row.get("attendance_rate")
        try:
            attendance = float(val)
            if attendance < 0 or attendance > 1:
                errors.append(
                    f"Row {row_num}: attendance_rate out of range [0, 1]: {val}"
                )
        except (TypeError, ValueError):
            errors.append(f"Row {row_num}: attendance_rate invalid or missing: {val}")

        # Validate days_since_last_submission (non-negative integer)
        val = row.get("days_since_last_submission")
        try:
            days = int(val)
            if days < 0:
                errors.append(
                    f"Row {row_num}: days_since_last_submission must be >= 0: {val}"
                )
        except (TypeError, ValueError):
            errors.append(f"Row {row_num}: days_since_last_submission invalid or missing: {val}")

    return len(errors) == 0, errors


def validate_dataframe(df: pandas.DataFrame) -> tuple[bool, list[str]]:
    """Validate DataFrame schema and data ranges.

    Args:
        df: pandas DataFrame to validate.

    Returns:
        Tuple of (is_valid, list_of_errors).
    """
    all_errors: list[str] = []

    # Check schema
    schema_valid, schema_errors = validate_schema(df)
    all_errors.extend(schema_errors)

    if not schema_valid:
        return False, all_errors

    # Check ranges
    range_valid, range_errors = validate_ranges(df)
    all_errors.extend(range_errors)

    return len(all_errors) == 0, all_errors


def load_and_validate(file_path: str | Path) -> pandas.DataFrame:
    """Load CSV and validate it, exiting with error code on failure.

    Args:
        file_path: Path to the CSV file.

    Returns:
        Validated data as pandas DataFrame.

    Raises:
        SystemExit: If validation fails (exit code 1).
    """
    try:
        df = read_csv(file_path)
    except (FileNotFoundError, ValueError) as e:
        print(f"Error reading CSV: {e}", file=sys.stderr)
        sys.exit(1)

    is_valid, errors = validate_dataframe(df)
    if not is_valid:
        print("Validation failed:", file=sys.stderr)
        for error in errors:
            print(f"  - {error}", file=sys.stderr)
        sys.exit(1)

    return df