calendar_reader.py

import sqlite3
import json
import os
import sys
import argparse
import datetime
from typing import List, Union

try:
    from rich import print
except ModuleNotFoundError:
    pass

parser = argparse.ArgumentParser(
    description="Reads calendar JSON files generated by https://github.com/PCS24/calendar-pdf-extractor and creates scheduled tasks in the target sqlite database, assuming it follows the schema of Static/calendar.template_db."
)


def path_checker(s: str) -> str:
    if not os.path.isfile(s):
        raise argparse.ArgumentTypeError(f'"{s}" is not an existing file path.')
    return s


parser.add_argument(
    "json-path",
    metavar="json-path",
    type=str,
    help="Path to the calendar JSON file generated by https://github.com/PCS24/calendar-pdf-extractor.",
)

parser.add_argument(
    "database-path",
    metavar="database-path",
    type=str,
    help="Path to the calendar database following the schema of Static/calendar.template_db.",
)

args = vars(parser.parse_args())
path_checker(args["json-path"])
path_checker(args["database-path"])

with open(args["json-path"], "r") as f:
    data = json.load(f)["dates"]

_SCHEDULE_T = List[List[Union[str, datetime.datetime]]]


def regular_schedule(d: datetime.datetime) -> _SCHEDULE_T:
    return [
        ["LOCKDOWN", d.replace(hour=7, minute=40)],
        ["REOPEN", d.replace(hour=14, minute=30)],
    ]


def delayed_opening_schedule(d: datetime.datetime) -> _SCHEDULE_T:
    return [
        ["LOCKDOWN", d.replace(hour=9, minute=30)],
        ["REOPEN", d.replace(hour=14, minute=30)],
    ]


def single_session_schedule(d: datetime.datetime) -> _SCHEDULE_T:
    return [
        ["LOCKDOWN", d.replace(hour=7, minute=40)],
        ["REOPEN", d.replace(hour=12, minute=0)],
    ]


now = datetime.datetime.now()
pre_tasks = []

for date_str in data["regular"]:
    pre_tasks.append(regular_schedule(datetime.datetime.fromisoformat(date_str)))

for date_str in data["early_dismissal"] + data["student_early_dismissal"]:
    pre_tasks.append(single_session_schedule(datetime.datetime.fromisoformat(date_str)))

for date_str in data["delayed"]:
    pre_tasks.append(delayed_opening_schedule(datetime.datetime.fromisoformat(date_str)))

tasks = []
for t in pre_tasks:
    for p in t:
        tasks.append(p)

with sqlite3.connect(args["database-path"]) as db:
    cur = db.cursor()
    for task in tasks:
        cur.execute(
            "INSERT INTO CALENDAR (ACTION, COMPLETED, CREATION_TIMESTAMP, SCHEDULED_TIMESTAMP) VALUES (?,?,?,?)",
            (task[0], 0, now.timestamp(), task[1].timestamp()),
        )
    db.commit()
    cur.close()

print(f"Added {len(tasks)} tasks to the calendar.")