129 lines
3.6 KiB
Python
Executable File
129 lines
3.6 KiB
Python
Executable File
#!/bin/python3
|
|
|
|
import caldav, time, sys, signal, os
|
|
from dotenv import load_dotenv
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
from pathlib import Path
|
|
|
|
# load settings from .env
|
|
try:
|
|
load_dotenv()
|
|
CALDAV_URL = os.environ["CALDAV_URL"]
|
|
CALDAV_USER = os.environ["CALDAV_USER"]
|
|
CALDAV_PASS = os.environ["CALDAV_PASS"]
|
|
MODULES = os.environ["SUBJECTS"].split(",")
|
|
TIME_ZONE = ZoneInfo(os.environ.get("TIMEZONE", "Europe/Berlin"))
|
|
except Exception:
|
|
print("Failed to read settings from .env!\nTake a look at .env.example for the required structure!")
|
|
|
|
# apply
|
|
client = caldav.DAVClient(
|
|
url=CALDAV_URL,
|
|
username=CALDAV_USER,
|
|
password=CALDAV_PASS,
|
|
)
|
|
calendar = client.calendar(url=CALDAV_URL)
|
|
|
|
backlog_path = Path(__file__).resolve().parent/"backlog.csv"
|
|
|
|
|
|
def sync(mod, start, end):
|
|
try:
|
|
calendar.add_event(dtstart=start, dtend=end, summary=mod)
|
|
print("# Synced to calendar.")
|
|
except Exception:
|
|
print(f"Failed to synchronize, adding to backlog.")
|
|
with open(backlog_path, "a") as fd:
|
|
fd.write(f"{mod},{start.isoformat()},{end.isoformat()}\n")
|
|
|
|
|
|
# ================= check for backlog to sync =================
|
|
|
|
failed_synchs=[]
|
|
if backlog_path.exists():
|
|
with open(backlog_path, "r") as fd:
|
|
lines = fd.readlines()
|
|
|
|
if lines:
|
|
print("# Found unsynched sessions in backlog. Attempting sync now!")
|
|
|
|
line_nr = 0
|
|
for line in lines:
|
|
line_nr += 1
|
|
# this is one session. parse it:
|
|
data = line.replace("\n", "").split(",")
|
|
try:
|
|
assert(len(data)==3)
|
|
mod = data[0]
|
|
start = datetime.fromisoformat(data[1])
|
|
end = datetime.fromisoformat(data[2])
|
|
except Exception:
|
|
print(f"Failed to parse backlog in line {line_nr}")
|
|
continue
|
|
|
|
|
|
# and reattempt sync
|
|
try:
|
|
calendar.add_event(dtstart=start, dtend=end, summary=mod)
|
|
print(f"# synced event '{mod}'")
|
|
except Exception:
|
|
failed_synchs.append([mod,start,end])
|
|
print(f"failed to sync event '{mod}'")
|
|
|
|
# overwrite file with failed syncs (if there are any)
|
|
backlog_path.unlink(missing_ok=True) # delete old file
|
|
if failed_synchs:
|
|
with open(backlog_path, "w") as fd:
|
|
for mod, start, end in failed_synchs:
|
|
fd.write(f"{mod},{start.isoformat()},{end.isoformat()}\n")
|
|
|
|
|
|
|
|
# ===================== new session =========================
|
|
|
|
# parse args
|
|
if len(sys.argv) != 2:
|
|
print(f"Please provide an module: {sys.argv[0]} <module>")
|
|
sys.exit(0)
|
|
|
|
if not sys.argv[1] in MODULES:
|
|
print("Invalid module name. Please provide one of: ")
|
|
for mod in MODULES: print(f" - {mod}")
|
|
print()
|
|
sys.exit(0)
|
|
|
|
|
|
# start session
|
|
mod = sys.argv[1]
|
|
print(f"# Starting study intervall for {mod}")
|
|
start = datetime.now(TIME_ZONE)
|
|
|
|
|
|
# stop handled via sigint
|
|
studying = True
|
|
def handle_sigint(signum, frame):
|
|
global studying
|
|
studying = False
|
|
signal.signal(signal.SIGINT, handle_sigint)
|
|
|
|
# timer
|
|
while studying:
|
|
print(f"\r{str(datetime.now(TIME_ZONE)-start).split('.')[0]} <^C to end session>", end='', flush=True)
|
|
time.sleep(0.1)
|
|
|
|
# end session and ask if should be synced
|
|
end = datetime.now(TIME_ZONE)
|
|
print(f"\n# Stopped at {end:%H:%M:%S}")
|
|
ans = ""
|
|
while ans not in ["y", "n"]:
|
|
print("# Do you want to sync this session? [y|n]")
|
|
ans = input("> ").lower()
|
|
|
|
if ans == "n":
|
|
print("Stopping..")
|
|
exit(0)
|
|
|
|
# sync
|
|
sync(mod=mod, start=start, end=end)
|