added seperate config and calendar modules and refactored
This commit is contained in:
+10
-14
@@ -3,28 +3,24 @@ import logging
|
||||
import os, sys, threading, time
|
||||
|
||||
from src.parser import (
|
||||
get_this_semesters_sessions,
|
||||
calculate_totals,
|
||||
get_last_n_days,
|
||||
get_stats,
|
||||
)
|
||||
from src.cache import Cache
|
||||
from src.web_calendar import Calendar
|
||||
from src import config
|
||||
|
||||
app = Flask(__name__)
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
sessions_cache = Cache()
|
||||
calendar = Calendar()
|
||||
|
||||
REFETCH_MINUTES = os.environ.get("REFETCH_INTERVAL_MINUTES", "5")
|
||||
if not REFETCH_MINUTES.isdigit():
|
||||
app.logger.error("Invalid refetch interval")
|
||||
sys.exit(1)
|
||||
REFETCH_SECONDS = int(REFETCH_MINUTES) * 60
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
# initial data fetch (required to work)
|
||||
try:
|
||||
sessions = get_this_semesters_sessions()
|
||||
sessions = calendar.fetch_sessions_since(config.CURRENT_SEMESTER_START)
|
||||
sessions_cache.set(sessions)
|
||||
app.logger.info(f"Fetched {len(sessions)} sessions from calDAV server")
|
||||
except Exception:
|
||||
app.logger.exception("Failed to fetch from calDAV server.")
|
||||
sys.exit(1)
|
||||
@@ -57,12 +53,12 @@ def api_sessions():
|
||||
# background daemon
|
||||
def fetch_job():
|
||||
while True:
|
||||
time.sleep(REFETCH_SECONDS)
|
||||
time.sleep(config.REFETCH_SECONDS)
|
||||
try:
|
||||
sessions_cache.set(get_this_semesters_sessions())
|
||||
app.logger.info(f"Fetched {len(sessions)} sessions from calDAV server")
|
||||
except Exception:
|
||||
app.logger.warning("Failed to fetch from calDAV server, keeping old data")
|
||||
sessions = calendar.fetch_sessions_since(config.CURRENT_SEMESTER_START)
|
||||
sessions_cache.set(sessions)
|
||||
except Exception as e:
|
||||
app.logger.warning(f"Failed to fetch from calDAV server, keeping old data. Error: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
if os.environ.get("WERKZEUG_RUN_MAIN") == "true" or not app.debug:
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
from zoneinfo import ZoneInfo
|
||||
from dotenv import load_dotenv
|
||||
from datetime import datetime, timedelta, date, time as dtime
|
||||
import sys,logging,os
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
load_dotenv()
|
||||
|
||||
CALDAV_URL = os.environ["CALDAV_URL"]
|
||||
CALDAV_USER = os.environ["CALDAV_USER"]
|
||||
CALDAV_PASS = os.environ["CALDAV_PASS"]
|
||||
TIME_ZONE = ZoneInfo(os.environ.get("TIMEZONE", "Europe/Berlin"))
|
||||
SEMESTER_STARTS = os.environ["SEMESTER_STARTS"]
|
||||
if SEMESTER_STARTS: SEMESTER_STARTS = SEMESTER_STARTS.split(",")
|
||||
except Exception:
|
||||
logger.exception("Failed to read config from .env!\nTake a look at .env.example for the required structure!")
|
||||
sys.exit(1)
|
||||
|
||||
REFETCH_MINUTES = os.environ.get("REFETCH_INTERVAL_MINUTES", "5")
|
||||
if not REFETCH_MINUTES.isdigit():
|
||||
logger.error("Invalid refetch interval")
|
||||
sys.exit(1)
|
||||
REFETCH_SECONDS = int(REFETCH_MINUTES) * 60
|
||||
|
||||
# get the date of the start of the semester were currently in
|
||||
# or return None if no semesters were specified
|
||||
def get_current_semester_start() -> date:
|
||||
global SEMESTER_STARTS
|
||||
if not SEMESTER_STARTS: # array of MM-DD strings
|
||||
return None
|
||||
|
||||
# first check if one the dates was previously this year
|
||||
today = datetime.now().date()
|
||||
found_start = None
|
||||
try:
|
||||
dates = [datetime.strptime(str(today.year)+"-"+date_str, "%Y-%m-%d").date() for date_str in SEMESTER_STARTS]
|
||||
except Exception:
|
||||
logger.warning("Failed to parse semester start date. Falling back to None!")
|
||||
return None
|
||||
|
||||
for date in dates:
|
||||
if date > today:
|
||||
continue
|
||||
if found_start and date < found_start:
|
||||
continue
|
||||
found_start = date
|
||||
|
||||
if found_start:
|
||||
return found_start
|
||||
|
||||
# otherwise use the last date of last year
|
||||
try:
|
||||
last_year_dates = [date.replace(year=date.year - 1) for date in dates]
|
||||
except Exception:
|
||||
logger.warning("Failed to parse semester start date. Falling back to None!")
|
||||
return None
|
||||
|
||||
return max(last_year_dates)
|
||||
|
||||
|
||||
CURRENT_SEMESTER_START = get_current_semester_start()
|
||||
+4
-100
@@ -1,105 +1,9 @@
|
||||
import logging
|
||||
from datetime import datetime, timedelta, date
|
||||
import os, random, logging
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from datetime import datetime, timedelta, date, time as dtime
|
||||
|
||||
from zoneinfo import ZoneInfo
|
||||
import caldav, os, random
|
||||
from src import web_calendar, config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 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"]
|
||||
TIME_ZONE = ZoneInfo(os.environ.get("TIMEZONE", "Europe/Berlin"))
|
||||
SEMESTER_STARTS = os.environ["SEMESTER_STARTS"]
|
||||
if SEMESTER_STARTS: SEMESTER_STARTS = SEMESTER_STARTS.split(",")
|
||||
except Exception:
|
||||
logger.error("Failed to read settings from .env!\nTake a look at .env.example for the required structure!")
|
||||
|
||||
client = caldav.DAVClient(
|
||||
url=CALDAV_URL,
|
||||
username=CALDAV_USER,
|
||||
password=CALDAV_PASS,
|
||||
)
|
||||
calendar = client.calendar(url=CALDAV_URL)
|
||||
|
||||
# we need to convert all-day events from date to daytime (add a time and tz)
|
||||
# dont know when id even need an all day event but better be safe so why not
|
||||
def to_datetime(value):
|
||||
if isinstance(value, date) and not isinstance(value, datetime):
|
||||
value = datetime.combine(value, dtime.min, tzinfo=TIME_ZONE) # just set time to 00:00
|
||||
return value
|
||||
|
||||
# grab ALL sessions and return as array of dicts
|
||||
def get_all_sessions() -> list:
|
||||
events = calendar.events()
|
||||
events.sort(key=lambda e: to_datetime(e.vobject_instance.vevent.dtstart.value))
|
||||
|
||||
event_list = []
|
||||
for event in events:
|
||||
event.load() # fetch full data
|
||||
e = event.vobject_instance.vevent
|
||||
event_list.append({
|
||||
"subject":e.summary.value,
|
||||
"ts":e.dtstart.value.strftime("%d.%m.%Y %H:%M"),
|
||||
"date":e.dtstart.value.strftime("%Y-%m-%d"),
|
||||
"minutes":int((e.dtend.value - e.dtstart.value).total_seconds() / 60)
|
||||
})
|
||||
return event_list
|
||||
|
||||
# grab only the sessions that fall into the current semester
|
||||
# (or all if none was given)
|
||||
def get_this_semesters_sessions() -> list:
|
||||
sessions = get_all_sessions()
|
||||
semester_start = get_current_semester_start()
|
||||
logger.info(f"keeping sessions after semester start [{str(semester_start)}]")
|
||||
|
||||
if not semester_start:
|
||||
# no semester starts given, just pass all sessions
|
||||
return sessions
|
||||
|
||||
# only keep sessions with a date greater then semester start
|
||||
# man i love list comprehension
|
||||
return [session for session in sessions if datetime.strptime(session["date"], "%Y-%m-%d").date() >= semester_start]
|
||||
|
||||
|
||||
# get the date of the start of the semester were currently in
|
||||
# or return None if no semesters were specified
|
||||
def get_current_semester_start() -> date:
|
||||
if not SEMESTER_STARTS: # array of MM-DD strings
|
||||
return None
|
||||
|
||||
# first check if one the dates was previously this year
|
||||
today = datetime.now().date()
|
||||
found_start = None
|
||||
try:
|
||||
dates = [datetime.strptime(str(today.year)+"-"+date_str, "%Y-%m-%d").date() for date_str in SEMESTER_STARTS]
|
||||
except Exception:
|
||||
logger.warning("Failed to parse semester start date. Falling back to None!")
|
||||
return None
|
||||
|
||||
for date in dates:
|
||||
if date > today:
|
||||
continue
|
||||
if found_start and date < found_start:
|
||||
continue
|
||||
found_start = date
|
||||
|
||||
if found_start:
|
||||
return found_start
|
||||
|
||||
# otherwise use the last date of last year
|
||||
try:
|
||||
last_year_dates = [date.replace(year=date.year - 1) for date in dates]
|
||||
except Exception:
|
||||
logger.warning("Failed to parse semester start date. Falling back to None!")
|
||||
return None
|
||||
|
||||
return max(last_year_dates)
|
||||
|
||||
# return dict with all subjects and their respective total hours studied
|
||||
def calculate_totals(sessions) -> dict:
|
||||
@@ -148,7 +52,7 @@ def get_total_month(sessions) -> int:
|
||||
for session in sessions:
|
||||
year_month = str(session["date"])[:7]
|
||||
|
||||
if year_month == datetime.now(TIME_ZONE).strftime("%Y-%m"):
|
||||
if year_month == datetime.now(config.TIME_ZONE).strftime("%Y-%m"):
|
||||
total_min += session["minutes"]
|
||||
return total_min
|
||||
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
from datetime import date, datetime, time as dtime
|
||||
import caldav, logging
|
||||
|
||||
from src import config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# we need to convert all-day events from date to daytime (add a time and tz)
|
||||
# dont know when id even need an all day event but better be safe so why not
|
||||
def date_to_datetime(value):
|
||||
if isinstance(value, date) and not isinstance(value, datetime):
|
||||
value = datetime.combine(value, dtime.min, tzinfo=config.TIME_ZONE) # just set time to 00:00
|
||||
return value
|
||||
|
||||
|
||||
# calendar class to interface with calDAV calender
|
||||
|
||||
class Calendar:
|
||||
def __init__(self):
|
||||
self._client = caldav.DAVClient(
|
||||
url=config.CALDAV_URL,
|
||||
username=config.CALDAV_USER,
|
||||
password=config.CALDAV_PASS,
|
||||
)
|
||||
self._calendar = self._client.calendar(url=config.CALDAV_URL)
|
||||
|
||||
def fetch_events(self):
|
||||
return self._calendar.events()
|
||||
|
||||
# grab ALL sessions and return as array of dicts
|
||||
def fetch_all_sessions(self) -> list:
|
||||
events = self.fetch_events()
|
||||
events.sort(key=lambda e: date_to_datetime(e.vobject_instance.vevent.dtstart.value))
|
||||
|
||||
event_list = []
|
||||
for event in events:
|
||||
event.load() # fetch full data
|
||||
e = event.vobject_instance.vevent
|
||||
event_list.append({
|
||||
"subject":e.summary.value,
|
||||
"ts":e.dtstart.value.strftime("%d.%m.%Y %H:%M"),
|
||||
"date":e.dtstart.value.strftime("%Y-%m-%d"),
|
||||
"minutes":int((e.dtend.value - e.dtstart.value).total_seconds() / 60)
|
||||
})
|
||||
|
||||
logger.info(f"Fetched {len(event_list)} sessions from calDAV server")
|
||||
return event_list
|
||||
|
||||
# grab only the sessions that fall into the current semester
|
||||
# (or all if none was given)
|
||||
def fetch_sessions_since(self, date: date) -> list:
|
||||
sessions = self.fetch_all_sessions()
|
||||
logger.info(f"keeping sessions after [{date}]")
|
||||
|
||||
if not date:
|
||||
# no semester starts given, just pass all sessions
|
||||
return sessions
|
||||
|
||||
# only keep sessions with a date greater then semester start
|
||||
# man i love list comprehension
|
||||
return [session for session in sessions if datetime.strptime(session["date"], "%Y-%m-%d").date() >= date]
|
||||
|
||||
Reference in New Issue
Block a user