First fully working model with data from 2025 and 2026
This commit is contained in:
@@ -0,0 +1,138 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user