initial version
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
**/.env
|
||||
**/backlog.csv
|
||||
@@ -0,0 +1,12 @@
|
||||
# full url. for example: https://calendar.com/user/study
|
||||
CALDAV_URL=
|
||||
|
||||
# CalDAV credentials
|
||||
CALDAV_USER=
|
||||
CALDAV_PASS=
|
||||
|
||||
# timezone. for example the default: Europe/Berlin
|
||||
TIMEZONE=
|
||||
|
||||
# Comma separated list of modules
|
||||
SUBJECTS=
|
||||
Executable
+123
@@ -0,0 +1,123 @@
|
||||
#!/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
|
||||
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"))
|
||||
|
||||
# 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])
|
||||
|
||||
# overwrite file with failed syncs (if there are any)
|
||||
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)
|
||||
Reference in New Issue
Block a user