import argparse 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) PROJECT_ROOT = Path(__file__).resolve().parents[1] DEFAULT_DATABASE = PROJECT_ROOT / "instance" / "calendar.sqlite" REQUIRED_TABLES = {"date_choices", "date_statuses"} 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): 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 inspect_database(database_path): if not database_path.is_file(): raise RuntimeError( f"Database does not exist: {database_path}\n" "Start the app once to initialize it, or pass the correct path with --database." ) with sqlite3.connect(database_path) as database: tables = { row[0] for row in database.execute("SELECT name FROM sqlite_master WHERE type = 'table'") } missing = REQUIRED_TABLES - tables if missing: raise RuntimeError( f"Database has the wrong schema; missing tables: {', '.join(sorted(missing))}" ) amount_count = database.execute("SELECT COUNT(*) FROM date_choices").fetchone()[0] status_count = database.execute("SELECT COUNT(*) FROM date_statuses").fetchone()[0] dates = database.execute( """ SELECT MIN(recorded_date), MAX(recorded_date) FROM ( SELECT choice_date AS recorded_date FROM date_choices UNION ALL SELECT status_date AS recorded_date FROM date_statuses ) """ ).fetchone() return amount_count, status_count, dates[0], dates[1] def verify_import(database_path, amounts, statuses): with sqlite3.connect(database_path) as database: stored_amounts = dict( database.execute("SELECT choice_date, amount_cents FROM date_choices") ) stored_statuses = dict( database.execute("SELECT status_date, day_status FROM date_statuses") ) bad_amounts = [key for key, value in amounts.items() if stored_amounts.get(key) != value] bad_statuses = [key for key, value in statuses.items() if stored_statuses.get(key) != value] if bad_amounts or bad_statuses: raise RuntimeError( "Post-import verification failed: " f"{len(bad_amounts)} amounts and {len(bad_statuses)} statuses do not match." ) def import_data(database_path, amounts, statuses): timestamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f") backup_path = database_path.with_name(f"{database_path.name}.backup-{timestamp}") with sqlite3.connect(database_path) as source, sqlite3.connect(backup_path) as backup: source.backup(backup) 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(), ) verify_import(database_path, amounts, statuses) return backup_path 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=DEFAULT_DATABASE, help=f"SQLite database used by the app (default: {DEFAULT_DATABASE})", ) parser.add_argument("--apply", action="store_true", help="Write the data; otherwise perform a dry run.") args = parser.parse_args() database_path = args.database.expanduser().resolve() workbook_path = args.workbook.expanduser().resolve() if not workbook_path.is_file(): parser.error(f"Workbook does not exist: {workbook_path}") try: before_amounts, before_statuses, first_date, last_date = inspect_database(database_path) except RuntimeError as error: parser.error(str(error)) print(f"Workbook: {workbook_path}") print(f"Database: {database_path}") print( f"Database before import: {before_amounts} amounts, {before_statuses} statuses" + (f", date range {first_date} through {last_date}." if first_date else ", no recorded dates.") ) amounts, statuses = read_workbook(workbook_path) if not amounts and not statuses: parser.error("Workbook produced no importable amounts or statuses; refusing to continue.") amount_conflicts, status_conflicts = existing_conflicts(database_path, 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(database_path, amounts, statuses) after_amounts, after_statuses, first_date, last_date = inspect_database(database_path) print(f"Import complete and verified: all {len(amounts)} amounts and {len(statuses)} statuses match.") print(f"Backup: {backup}") print( f"Database after import: {after_amounts} amounts, {after_statuses} statuses, " f"date range {first_date} through {last_date}." ) if __name__ == "__main__": main()