217 lines
8.8 KiB
Python
217 lines
8.8 KiB
Python
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()
|