161 lines
4.6 KiB
Python
161 lines
4.6 KiB
Python
from flask import Flask, Response
|
|
from icalendar import Calendar, Event
|
|
from datetime import datetime
|
|
import pytz
|
|
import uuid
|
|
|
|
|
|
class Webserver:
|
|
def __init__(self, ip, port, db_instance):
|
|
self.ip = ip
|
|
self.port = port
|
|
self.db = db_instance
|
|
|
|
self.app = Flask(__name__)
|
|
self.tz = pytz.timezone("Europe/Amsterdam")
|
|
|
|
self._register_routes()
|
|
|
|
def _register_routes(self):
|
|
self.app.add_url_rule("/", "index", lambda: "Calendar server running ✅")
|
|
self.app.add_url_rule("/calendar.ics", "calendar", self._calendar_endpoint)
|
|
|
|
def _fetch_shifts(self):
|
|
con = self.db._get_connection()
|
|
cur = con.cursor()
|
|
|
|
cur.execute("""
|
|
SELECT shift_start, shift_end, department, description
|
|
FROM shifts
|
|
ORDER BY shift_start ASC
|
|
""")
|
|
|
|
rows = cur.fetchall()
|
|
con.close()
|
|
|
|
shifts = []
|
|
for row in rows:
|
|
start_str, end_str, department, description = row
|
|
|
|
# Convert string → datetime
|
|
start = self._parse_datetime(start_str)
|
|
end = self._parse_datetime(end_str)
|
|
|
|
shifts.append({
|
|
"start": start,
|
|
"end": end,
|
|
"description": f"{description} ",
|
|
"department": f"{department} "
|
|
})
|
|
|
|
return shifts
|
|
|
|
def _parse_datetime(self, dt_str):
|
|
dt = datetime.fromisoformat(dt_str)
|
|
if dt.tzinfo is None:
|
|
dt = self.tz.localize(dt)
|
|
return dt
|
|
|
|
def _calendar_endpoint(self):
|
|
cal = Calendar()
|
|
cal.add('prodid', '-//Shift Calendar//example//')
|
|
cal.add('version', '2.0')
|
|
|
|
shifts = self._fetch_shifts()
|
|
|
|
for shift in shifts:
|
|
event = Event()
|
|
event.add('uid', str(uuid.uuid4()))
|
|
event.add('dtstart', shift["start"])
|
|
event.add('dtend', shift["end"])
|
|
event.add('summary', shift["department"])
|
|
event.add('description', shift["description"])
|
|
event.add('dtstamp', datetime.now(tz=self.tz))
|
|
|
|
cal.add_component(event)
|
|
|
|
return Response(cal.to_ical(), mimetype="text/calendar")
|
|
|
|
def run(self):
|
|
print(f"🚀 Starting server on {self.ip}:{self.port}")
|
|
from flask import Flask, Response
|
|
from icalendar import Calendar, Event
|
|
from datetime import datetime
|
|
import pytz
|
|
import uuid
|
|
|
|
|
|
class Webserver:
|
|
def __init__(self, ip, port, db_instance):
|
|
self.ip = ip
|
|
self.port = port
|
|
self.db = db_instance
|
|
|
|
self.app = Flask(__name__)
|
|
self.tz = pytz.timezone("Europe/Amsterdam")
|
|
|
|
self._register_routes()
|
|
|
|
def _register_routes(self):
|
|
self.app.add_url_rule("/", "index", lambda: "Calendar server running ✅")
|
|
self.app.add_url_rule("/calendar.ics", "calendar", self._calendar_endpoint)
|
|
|
|
def _fetch_shifts(self):
|
|
con = self.db._get_connection()
|
|
cur = con.cursor()
|
|
|
|
cur.execute("""
|
|
SELECT shift_start, shift_end, department, description
|
|
FROM shifts
|
|
ORDER BY shift_start ASC
|
|
""")
|
|
|
|
rows = cur.fetchall()
|
|
con.close()
|
|
|
|
shifts = []
|
|
for row in rows:
|
|
start_str, end_str, department, description = row
|
|
|
|
# Convert string → datetime
|
|
start = self._parse_datetime(start_str)
|
|
end = self._parse_datetime(end_str)
|
|
|
|
shifts.append({
|
|
"start": start,
|
|
"end": end,
|
|
"description": f"{description} ",
|
|
"department": f"{department} "
|
|
})
|
|
|
|
return shifts
|
|
|
|
def _parse_datetime(self, dt_str):
|
|
dt = datetime.fromisoformat(dt_str)
|
|
if dt.tzinfo is None:
|
|
dt = self.tz.localize(dt)
|
|
return dt
|
|
|
|
def _calendar_endpoint(self):
|
|
cal = Calendar()
|
|
cal.add('prodid', '-//Shift Calendar//example//')
|
|
cal.add('version', '2.0')
|
|
|
|
shifts = self._fetch_shifts()
|
|
|
|
for shift in shifts:
|
|
event = Event()
|
|
event.add('uid', str(uuid.uuid4()))
|
|
event.add('dtstart', shift["start"])
|
|
event.add('dtend', shift["end"])
|
|
event.add('summary', shift["department"])
|
|
event.add('description', shift["description"])
|
|
event.add('dtstamp', datetime.now(tz=self.tz))
|
|
|
|
cal.add_component(event)
|
|
|
|
return Response(cal.to_ical(), mimetype="text/calendar")
|
|
|
|
def run(self):
|
|
print(f"🚀 Starting server on {self.ip}:{self.port}")
|
|
self.app.run(host=self.ip, port=self.port, threaded=True) |