fuj-management/scripts/attendance.py

"""Shared attendance/fee logic for FUJ Tuesday practices."""

import csv
import io
import urllib.request
from datetime import datetime

from config import ATTENDANCE_SHEET_ID as SHEET_ID, JUNIOR_SHEET_GID

EXPORT_URL = f"https://docs.google.com/spreadsheets/d/{SHEET_ID}/export?format=csv&gid=0"
JUNIOR_EXPORT_URL = f"https://docs.google.com/spreadsheets/d/{SHEET_ID}/export?format=csv&gid={JUNIOR_SHEET_GID}"

ADULT_FEE_DEFAULT = 750  # CZK, for 2+ practices in a month
ADULT_FEE_SINGLE = 200  # CZK, for exactly 1 practice in a month
ADULT_FEE_MONTHLY_RATE = {
    "2026-03": 350
}

JUNIOR_FEE_DEFAULT = 500  # CZK for 2+ practices
JUNIOR_MONTHLY_RATE = {
    "2025-09": 250
}
ADULT_MERGED_MONTHS = {
    #"2025-12": "2026-01",   # keys are merged into values
    #"2025-09": "2025-10"
}

JUNIOR_MERGED_MONTHS = {
    "2025-12": "2026-01",   # keys are merged into values
    "2025-09": "2025-10"
}

COL_NAME = 0
COL_TIER = 1
FIRST_DATE_COL = 3


def fetch_csv(url: str = EXPORT_URL) -> list[list[str]]:
    """Fetch the attendance Google Sheet as parsed CSV rows."""
    req = urllib.request.Request(url)
    with urllib.request.urlopen(req) as resp:
        text = resp.read().decode("utf-8")
    reader = csv.reader(io.StringIO(text))
    return list(reader)


def parse_dates(header_row: list[str]) -> list[tuple[int, datetime]]:
    """Return (column_index, date) pairs for all date columns."""
    dates = []
    
    for i in range(FIRST_DATE_COL, len(header_row)):
        raw = header_row[i].strip()
        if not raw:
            continue
            
        try:
            # Try DD.MM.YYYY
            dt = datetime.strptime(raw, "%d.%m.%Y")
            dates.append((i, dt))
        except ValueError:
            try:
                # Fallback to MM/DD/YYYY
                dt = datetime.strptime(raw, "%m/%d/%Y")
                dates.append((i, dt))
            except ValueError:
                pass
            
    return dates


def group_by_month(dates: list[tuple[int, datetime]], merged_months: dict[str, str]) -> dict[str, list[int]]:
    """Group column indices by YYYY-MM, handling merged months."""
    months: dict[str, list[int]] = {}
    for col, dt in dates:
        key = dt.strftime("%Y-%m")
        # Apply merged month mapping if configured
        target_key = merged_months.get(key, key)
        months.setdefault(target_key, []).append(col)
    return months


def calculate_fee(attendance_count: int, month_key: str) -> int:
    """Apply fee rules: 0 → 0, 1 → 200, 2+ → configured rate (default 750)."""
    if attendance_count == 0:
        return 0
    if attendance_count == 1:
        return ADULT_FEE_SINGLE
    return ADULT_FEE_MONTHLY_RATE.get(month_key, ADULT_FEE_DEFAULT)


def calculate_junior_fee(attendance_count: int, month_key: str) -> str | int:
    """Apply junior fee rules: 0 → 0, 1 → '?', 2+ → Configured Rate (default 500)."""
    if attendance_count == 0:
        return 0
    if attendance_count == 1:
        return "?"
    return JUNIOR_MONTHLY_RATE.get(month_key, JUNIOR_FEE_DEFAULT)


def get_members(rows: list[list[str]]) -> list[tuple[str, str, list[str]]]:
    """Parse member rows. Returns list of (name, tier, row).
    
    Stopped at row where first column contains '# last line'.
    Skips rows starting with '#'.
    """
    members = []
    for row in rows[1:]:
        if not row or len(row) <= COL_NAME:
            continue
            
        first_col = row[COL_NAME].strip()
        
        # Terminator for rows to process
        if "# last line" in first_col.lower():
            break
            
        # Ignore comments
        if first_col.startswith("#"):
            continue
            
        if not first_col or first_col.lower() in ("jméno", "name", "jmeno"):
            continue
            
        tier = row[COL_TIER].strip().upper() if len(row) > COL_TIER else ""
        members.append((first_col, tier, row))
    return members


def get_junior_members_from_sheet(rows: list[list[str]]) -> list[tuple[str, str, list[str]]]:
    """Parse junior member rows from the junior sheet.
    
    Stopped at row where first column contains '# Treneri'.
    Returns list of (name, tier, row) for members where tier is not 'X'.
    """
    members = []
    for row in rows[1:]:
        if not row or len(row) <= COL_NAME:
            continue
            
        first_col = row[COL_NAME].strip()
        
        # Terminator for rows to process in junior sheet
        if "# treneri" in first_col.lower() or "# trenéři" in first_col.lower():
            break
            
        # Ignore comments
        if first_col.startswith("#"):
            continue
            
        if not first_col or first_col.lower() in ("jméno", "name", "jmeno"):
            continue
            
        tier = row[COL_TIER].strip().upper() if len(row) > COL_TIER else ""
        if tier == "X":
            continue
            
        members.append((first_col, tier, row))
    return members


def get_members_with_fees() -> tuple[list[tuple[str, str, dict[str, int]]], list[str]]:
    """Fetch attendance data and compute fees.

    Returns:
        (members, sorted_months) where members is a list of
        (name, tier, {month_key: (fee, count)}) for ALL members (all tiers).
        sorted_months is the list of YYYY-MM keys in order.
    """
    rows = fetch_csv()
    if len(rows) < 2:
        return [], []

    header_row = rows[0]
    dates = parse_dates(header_row)
    if not dates:
        return [], []

    months = group_by_month(dates, ADULT_MERGED_MONTHS)
    sorted_months = sorted(months.keys())
    members_raw = get_members(rows)

    members = []
    for name, tier, row in members_raw:
        month_fees = {}
        for month_key in sorted_months:
            cols = months[month_key]
            count = sum(
                1
                for c in cols
                if c < len(row) and row[c].strip().upper() == "TRUE"
            )
            fee = calculate_fee(count, month_key) if tier == "A" else 0
            month_fees[month_key] = (fee, count)
        members.append((name, tier, month_fees))

    return members, sorted_months


def get_junior_members_with_fees() -> tuple[list[tuple[str, str, dict[str, tuple[str | int, int, int, int]]]], list[str]]:
    """Fetch attendance data from both sheets and compute junior fees.
    
    Merges members by exact name match.
    
    Returns:
        (members, sorted_months) where members is a list of
        (name, tier, {month_key: (fee, total_count, adult_count, junior_count)}).
    """
    main_rows = fetch_csv(EXPORT_URL)
    junior_rows = fetch_csv(JUNIOR_EXPORT_URL)
    
    if len(main_rows) < 2 or len(junior_rows) < 2:
        return [], []

    main_dates = parse_dates(main_rows[0])
    junior_dates = parse_dates(junior_rows[0])
    
    main_months = group_by_month(main_dates, JUNIOR_MERGED_MONTHS)
    junior_months = group_by_month(junior_dates, JUNIOR_MERGED_MONTHS)
    
    # Collect all unique sorted months
    all_months = set(main_months.keys()).union(set(junior_months.keys()))
    sorted_months = sorted(list(all_months))

    from typing import Any
    merged_members: dict[str, Any] = {}
    
    # Process Junior Tier from Main Sheet (Adult Practices)
    main_members_raw = get_members(main_rows)
    for name, tier, row in main_members_raw:
        if tier != "J":
            continue
            
        if name not in merged_members:
            merged_members[name] = {"tier": tier, "months": {}}
            
        for month_key in sorted_months:
            if month_key not in merged_members[name]["months"]:
                merged_members[name]["months"][month_key] = {"adult": 0, "junior": 0}
                
            cols = main_months.get(month_key, [])
            adult_count = sum(
                1
                for c in cols
                if c < len(row) and row[c].strip().upper() == "TRUE"
            )
            merged_members[name]["months"][month_key]["adult"] += adult_count
            
    # Process Junior Sheet (Junior Practices)
    junior_members_raw = get_junior_members_from_sheet(junior_rows)
    for name, tier, row in junior_members_raw:
        if name not in merged_members:
            merged_members[name] = {"tier": tier, "months": {}}
            
        for month_key in sorted_months:
            if month_key not in merged_members[name]["months"]:
                merged_members[name]["months"][month_key] = {"adult": 0, "junior": 0}
                
            cols = junior_months.get(month_key, [])
            junior_count = sum(
                1
                for c in cols
                if c < len(row) and row[c].strip().upper() == "TRUE"
            )
            merged_members[name]["months"][month_key]["junior"] += junior_count

    # Compile the final result format
    members = []
    for name, data in merged_members.items():
        month_fees = {}
        for month_key in sorted_months:
            adult_count = data["months"].get(month_key, {}).get("adult", 0)
            junior_count = data["months"].get(month_key, {}).get("junior", 0)
            total_count = adult_count + junior_count
            
            fee = calculate_junior_fee(total_count, month_key)
            month_fees[month_key] = (fee, total_count, adult_count, junior_count)
            
        members.append((name, data["tier"], month_fees))

    return members, sorted_months