Fatima/scripts/import_excel.py

import argparse
import shutil
import sqlite3
from datetime import date, datetime, timedelta
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from pathlib import Path

import openpyxl


GREEN = "FFC2F1C8"
YELLOW = "FFFFFF00"
ORANGE_THEME = 5
MONTH_HEADER_ROWS = (2, 11, 20)
MONTH_START_COLUMNS = (2, 10, 18, 26)


def cell_color(cell):
    color = cell.fill.fgColor
    if cell.fill.fill_type != "solid":
        return None
    if color.type == "rgb":
        return color.rgb
    if color.type == "theme" and color.theme == ORANGE_THEME:
        return "orange"
    return None


def calendar_date(year, month, week_index, weekday_index):
    first = date(year, month, 1)
    first_sunday = first - timedelta(days=(first.weekday() + 1) % 7)
    result = first_sunday + timedelta(days=week_index * 7 + weekday_index)
    return result if result.month == month else None


def amount_cents(value, coordinate):
    try:
        amount = Decimal(str(value).strip())
        if not amount.is_finite() or amount < 0:
            raise InvalidOperation
        return int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
    except (InvalidOperation, ValueError):
        raise ValueError(f"Invalid amount in {coordinate}: {value!r}") from None


def read_workbook(path):
    workbook = openpyxl.load_workbook(path, data_only=True, read_only=False)
    amounts = {}
    statuses = {}

    for sheet in workbook.worksheets:
        try:
            year = int(sheet.title)
        except ValueError:
            continue
        for quarter, header_row in enumerate(MONTH_HEADER_ROWS):
            for position, start_column in enumerate(MONTH_START_COLUMNS):
                month = quarter * 4 + position + 1
                for week_index in range(6):
                    row = header_row + 2 + week_index
                    for weekday_index in range(7):
                        cell = sheet.cell(row=row, column=start_column + weekday_index)
                        selected_date = calendar_date(year, month, week_index, weekday_index)
                        if selected_date is None:
                            continue
                        color = cell_color(cell)
                        key = selected_date.isoformat()
                        if color == GREEN:
                            # A green blank means worked, but supplies no amount to import.
                            if cell.value is not None and str(cell.value).strip():
                                amounts[key] = amount_cents(cell.value, f"{sheet.title}!{cell.coordinate}")
                        elif color == YELLOW:
                            statuses[key] = "sick"
                        elif color == "orange":
                            statuses[key] = "vacation"
    return amounts, statuses


def existing_conflicts(database_path, amounts, statuses):
    if not database_path.exists():
        return 0, 0
    with sqlite3.connect(database_path) as database:
        existing_amounts = dict(database.execute("SELECT choice_date, amount_cents FROM date_choices"))
        existing_statuses = dict(database.execute("SELECT status_date, day_status FROM date_statuses"))
    amount_conflicts = sum(key in existing_amounts and existing_amounts[key] != value for key, value in amounts.items())
    status_conflicts = sum(key in existing_statuses and existing_statuses[key] != value for key, value in statuses.items())
    return amount_conflicts, status_conflicts


def import_data(database_path, amounts, statuses):
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    backup_path = database_path.with_name(f"{database_path.name}.backup-{timestamp}")
    if database_path.exists():
        shutil.copy2(database_path, backup_path)

    with sqlite3.connect(database_path) as database:
        database.executemany(
            """
            INSERT INTO date_choices (choice_date, amount_cents) VALUES (?, ?)
            ON CONFLICT(choice_date) DO UPDATE SET
                amount_cents = excluded.amount_cents,
                updated_at = CURRENT_TIMESTAMP
            """,
            amounts.items(),
        )
        database.executemany(
            """
            INSERT INTO date_statuses (status_date, day_status) VALUES (?, ?)
            ON CONFLICT(status_date) DO UPDATE SET
                day_status = excluded.day_status,
                updated_at = CURRENT_TIMESTAMP
            """,
            statuses.items(),
        )
    return backup_path if backup_path.exists() else None


def main():
    parser = argparse.ArgumentParser(description="Import Fatima calendar data from Excel.")
    parser.add_argument("workbook", type=Path)
    parser.add_argument("--database", type=Path, default=Path("instance/calendar.sqlite"))
    parser.add_argument("--apply", action="store_true", help="Write the data; otherwise perform a dry run.")
    args = parser.parse_args()

    amounts, statuses = read_workbook(args.workbook)
    amount_conflicts, status_conflicts = existing_conflicts(args.database, amounts, statuses)
    print(f"Found {len(amounts)} amounts, {sum(v == 'vacation' for v in statuses.values())} vacation days, and {sum(v == 'sick' for v in statuses.values())} sick days.")
    print(f"Existing differing values to overwrite: {amount_conflicts} amounts, {status_conflicts} statuses.")
    if not args.apply:
        print("Dry run only; no database changes made.")
        return

    backup = import_data(args.database, amounts, statuses)
    print(f"Import complete. Backup: {backup}" if backup else "Import complete; no previous database existed.")


if __name__ == "__main__":
    main()