勤務時間のCSVから休憩を除いた稼働時間を集計する

.gitignore

*.csv

calculate_work_hours.py

#!/usr/bin/env python3
"""勤務時間CSVから、休憩を除いた稼働時間を計算する。"""

import argparse
import csv
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Tuple


START_COLUMN = "勤務開始"
END_COLUMN = "勤務終了"
BREAK_COLUMN = "休憩時間（分）"
DATE_COLUMN = "勤務日"
WORK_COLUMN = "一日の稼働時間"
REQUIRED_COLUMNS = (START_COLUMN, END_COLUMN, BREAK_COLUMN, DATE_COLUMN)


def parse_time(value: str, row_number: int, column: str) -> datetime:
    try:
        return datetime.strptime(value.strip(), "%H:%M")
    except ValueError as exc:
        raise ValueError(
            f"{row_number}行目の「{column}」が時刻（HH:MM）ではありません: {value!r}"
        ) from exc


def calculate_work_minutes(row: Dict[str, str], row_number: int) -> int:
    start = parse_time(row[START_COLUMN], row_number, START_COLUMN)
    end = parse_time(row[END_COLUMN], row_number, END_COLUMN)

    # 終了が開始より前なら、日をまたいだ勤務として扱う。
    if end < start:
        end += timedelta(days=1)

    try:
        break_minutes = int(row[BREAK_COLUMN].strip())
    except ValueError as exc:
        raise ValueError(
            f"{row_number}行目の「{BREAK_COLUMN}」が整数ではありません: "
            f"{row[BREAK_COLUMN]!r}"
        ) from exc

    if break_minutes < 0:
        raise ValueError(f"{row_number}行目の休憩時間を0分以上にしてください")

    work_minutes = int((end - start).total_seconds() // 60) - break_minutes
    if work_minutes < 0:
        raise ValueError(f"{row_number}行目は休憩時間が勤務時間を超えています")
    return work_minutes


def format_minutes(minutes: int) -> str:
    hours, remaining_minutes = divmod(minutes, 60)
    return f"{hours:02d}:{remaining_minutes:02d}"


def parse_month(value: str) -> str:
    if len(value) != 6 or not value.isdigit():
        raise ValueError(f"対象月はYYYYMM形式で指定してください: {value!r}")
    try:
        month = datetime.strptime(value, "%Y%m")
    except ValueError as exc:
        raise ValueError(f"対象月はYYYYMM形式で指定してください: {value!r}") from exc
    return month.strftime("%Y-%m")


def read_and_calculate(
    input_path: Path,
    target_month: Optional[str] = None,
) -> Tuple[List[str], List[Dict[str, str]], int]:
    with input_path.open("r", encoding="utf-8-sig", newline="") as input_file:
        reader = csv.DictReader(input_file)
        if reader.fieldnames is None:
            raise ValueError("CSVにヘッダーがありません")

        missing = [column for column in REQUIRED_COLUMNS if column not in reader.fieldnames]
        if missing:
            raise ValueError(f"必要な列がありません: {', '.join(missing)}")

        fieldnames = list(reader.fieldnames)
        if WORK_COLUMN not in fieldnames:
            fieldnames.append(WORK_COLUMN)

        rows: List[Dict[str, str]] = []
        total_minutes = 0
        for row_number, row in enumerate(reader, start=2):
            if target_month is not None:
                work_date = row[DATE_COLUMN].strip()
                if not work_date.startswith(f"{target_month}-"):
                    continue
            minutes = calculate_work_minutes(row, row_number)
            row[WORK_COLUMN] = format_minutes(minutes)
            rows.append(row)
            total_minutes += minutes

        rows.sort(key=lambda row: row[DATE_COLUMN])

    return fieldnames, rows, total_minutes


def write_csv(
    output_path: Path,
    fieldnames: List[str],
    rows: List[Dict[str, str]],
    total_minutes: int,
) -> None:
    with output_path.open("w", encoding="utf-8", newline="") as output_file:
        writer = csv.DictWriter(output_file, fieldnames=fieldnames, lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)
        total_row = {column: "" for column in fieldnames}
        label_column = DATE_COLUMN if DATE_COLUMN in fieldnames else fieldnames[0]
        total_row[label_column] = "合計稼働時間"
        total_row[WORK_COLUMN] = format_minutes(total_minutes)
        writer.writerow(total_row)


def parse_args() -> Optional[argparse.Namespace]:
    parser = argparse.ArgumentParser(
        description="勤務開始から勤務終了までの時間から休憩時間を引き、稼働時間を計算します。",
        epilog=(
            "使用例:\n"
            "  python3 calculate_work_hours.py times.csv 202608\n"
            "  python3 calculate_work_hours.py times.csv 202608 -o result.csv\n"
            "  python3 calculate_work_hours.py times.csv  # 全件を出力"
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "input",
        type=Path,
        help="入力CSVのファイル名",
    )
    parser.add_argument(
        "month",
        nargs="?",
        help="対象月（YYYYMM形式、省略時は全件）",
    )
    parser.add_argument(
        "-o",
        "--output",
        type=Path,
        default=Path("result.csv"),
        help="出力CSV（デフォルト: result.csv）",
    )
    if len(sys.argv) == 1:
        parser.print_help()
        return None

    return parser.parse_args()


def main() -> int:
    args = parse_args()
    if args is None:
        return 0

    try:
        target_month = parse_month(args.month) if args.month is not None else None
        fieldnames, rows, total_minutes = read_and_calculate(args.input, target_month)
        write_csv(args.output, fieldnames, rows, total_minutes)
    except (OSError, ValueError) as exc:
        print(f"エラー: {exc}", file=sys.stderr)
        return 1

    print(f"{len(rows)}件を {args.output} に出力しました")
    print(f"合計稼働時間: {format_minutes(total_minutes)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())