312 lines
14 KiB
Python
Executable file
312 lines
14 KiB
Python
Executable file
import os
|
|
import requests
|
|
import zoneinfo
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, date, timedelta
|
|
from dotenv import load_dotenv
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import PatternFill, Font, Alignment, Border, Side
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
@dataclass
|
|
class WorkDay:
|
|
date_obj: date
|
|
phase: str
|
|
in1: str = ""
|
|
out1: str = ""
|
|
in2: str = ""
|
|
out2: str = ""
|
|
note: str = ""
|
|
is_absence: bool = False
|
|
|
|
@property
|
|
def target_time_formula(self) -> str:
|
|
# Zero target hours during semester or on pre-approved absence days.
|
|
if self.phase == "Semester":
|
|
return '=TIME(0,0,0)'
|
|
return '=TIME(0,0,0)' if self.is_absence else '=TIME(8,0,0)'
|
|
|
|
class Config:
|
|
def __init__(self):
|
|
load_dotenv()
|
|
|
|
self.email = os.environ.get("TR_EMAIL")
|
|
self.password = os.environ.get("TR_PASSWORD")
|
|
if not self.email or not self.password:
|
|
raise ValueError("Missing credentials in environment.")
|
|
|
|
# Contractual constants
|
|
self.semester_target_total = float(os.environ.get("SEMESTER_TARGET_TOTAL", "240.0"))
|
|
self.semester_weeks = float(os.environ.get("SEMESTER_WEEKS", "39.0"))
|
|
|
|
self.start_date = datetime.strptime(os.environ.get("TR_START_DATE", "2024-10-01"), "%Y-%m-%d").date()
|
|
self.local_tz = zoneinfo.ZoneInfo("Europe/Berlin")
|
|
|
|
today = datetime.now().date()
|
|
self.end_date_str = today.strftime("%Y-%m-%d")
|
|
|
|
# Academic year logic (resets Oct 1st annually)
|
|
if today.month >= 10:
|
|
self.acad_year_start = date(today.year, 10, 1)
|
|
self.current_term_end = date(today.year + 1, 7, 31)
|
|
else:
|
|
self.acad_year_start = date(today.year - 1, 10, 1)
|
|
self.current_term_end = date(today.year, 7, 31)
|
|
|
|
days_left = (self.current_term_end - today).days
|
|
self.remaining_weeks = max(0.0, round(days_left / 7.0, 1))
|
|
|
|
# Semester breaks (Vorlesungsfreie Zeit)
|
|
self.winter_break = ((2, 15), (3, 14))
|
|
self.summer_break = ((8, 1), (9, 30))
|
|
|
|
class TimeRecordingAPI:
|
|
"""Handles Auth0 PKCE session hijacking and HTTP interactions with Time Recording API."""
|
|
|
|
def __init__(self, config: Config):
|
|
self.config = config
|
|
self.token = None
|
|
self.employee_id = None
|
|
self.base_url = "https://api.timerecording.com"
|
|
|
|
def authenticate(self):
|
|
with sync_playwright() as p:
|
|
browser = p.chromium.launch(headless=True)
|
|
try:
|
|
context = browser.new_context()
|
|
page = context.new_page()
|
|
|
|
# Event listener to extract OAuth Bearer token from network headers
|
|
def capture_token(request):
|
|
if self.base_url in request.url:
|
|
auth = request.headers.get("authorization")
|
|
if auth and auth.startswith("Bearer "):
|
|
self.token = auth
|
|
|
|
page.on("request", capture_token)
|
|
page.goto("https://app.timerecording.com/de/main/dashboard")
|
|
page.fill('input[name="username"], input[type="email"]', self.config.email)
|
|
page.fill('input[name="password"], input[type="password"]', self.config.password)
|
|
page.click('button[type="submit"], button[name="action"]')
|
|
|
|
# Block until login request is executed
|
|
try:
|
|
with page.expect_request(lambda r: f"{self.base_url}/employees/me" in r.url, timeout=10000):
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
if not self.token:
|
|
raise RuntimeError("Authentication failed: Could not capture Bearer token.")
|
|
|
|
resp = context.request.get(f"{self.base_url}/employees/me", headers={"Authorization": self.token})
|
|
if not resp.ok:
|
|
raise RuntimeError(f"Failed to fetch employee ID. HTTP {resp.status}")
|
|
|
|
self.employee_id = resp.json().get("id")
|
|
finally:
|
|
browser.close()
|
|
|
|
def fetch_data(self) -> tuple[dict, list]:
|
|
headers = {"authorization": self.token, "referer": "https://app.timerecording.com/"}
|
|
|
|
r_url = f"{self.base_url}/employees/{self.employee_id}/record-buckets/{self.config.start_date}/{self.config.end_date_str}"
|
|
a_url = f"{self.base_url}/employees/{self.employee_id}/absences/{self.config.start_date}/{self.config.end_date_str}"
|
|
|
|
r_resp = requests.get(r_url, headers=headers)
|
|
a_resp = requests.get(a_url, headers=headers)
|
|
|
|
r_resp.raise_for_status()
|
|
a_resp.raise_for_status()
|
|
|
|
return r_resp.json(), a_resp.json()
|
|
|
|
class TimesheetProcessor:
|
|
"""Merges physical clock events and official absences into chronologically mapped workdays."""
|
|
|
|
def __init__(self, config: Config):
|
|
self.config = config
|
|
|
|
def get_phase(self, d: date) -> str:
|
|
w_start, w_end = date(d.year, *self.config.winter_break[0]), date(d.year, *self.config.winter_break[1])
|
|
s_start, s_end = date(d.year, *self.config.summer_break[0]), date(d.year, *self.config.summer_break[1])
|
|
|
|
if (w_start <= d <= w_end) or (s_start <= d <= s_end):
|
|
return "Fulltime"
|
|
return "Semester"
|
|
|
|
def process(self, records_data: dict, absences_data: list) -> list[WorkDay]:
|
|
days_map = {}
|
|
|
|
# 1. Parse official absences (sick leave, public holidays, vocational school)
|
|
for item in absences_data:
|
|
name = item.get("absenceType", {}).get("name", item.get("name", "Absence"))
|
|
start_d = datetime.strptime(item["start"], "%Y-%m-%d").date()
|
|
end_d = datetime.strptime(item["end"], "%Y-%m-%d").date()
|
|
|
|
curr = start_d
|
|
while curr <= end_d:
|
|
if curr >= self.config.start_date:
|
|
phase = self.get_phase(curr)
|
|
d_str = curr.strftime("%Y-%m-%d")
|
|
|
|
# Hardcode vocational school times (Berufsschule) to match local regional guidelines
|
|
if name == "Berufsschule" and phase == "Semester":
|
|
days_map[d_str] = WorkDay(curr, phase, in1="08:00:00", out1="16:00:00", note="Berufsschule")
|
|
elif phase == "Fulltime" and name != "Berufsschule":
|
|
days_map[d_str] = WorkDay(curr, phase, note=name, is_absence=True)
|
|
curr += timedelta(days=1)
|
|
|
|
# 2. Parse physical clock-in/out stamps
|
|
for d_str, records in records_data.items():
|
|
curr_date = datetime.strptime(d_str, "%Y-%m-%d").date()
|
|
if curr_date < self.config.start_date:
|
|
continue
|
|
|
|
if d_str not in days_map:
|
|
days_map[d_str] = WorkDay(curr_date, self.get_phase(curr_date))
|
|
|
|
# Skip merging logged times if day is overwritten by vocational school
|
|
if days_map[d_str].note == "Berufsschule":
|
|
continue
|
|
|
|
records.sort(key=lambda x: x['time'])
|
|
ins = [r for r in records if r['type'] == 'work_start']
|
|
outs = [r for r in records if r['type'] == 'work_stop']
|
|
|
|
def format_time(record):
|
|
dt = datetime.fromisoformat(record['time'].replace('Z', '+00:00')).astimezone(self.config.local_tz)
|
|
return dt.strftime('%H:%M:%S')
|
|
|
|
# Map up to 2 shifts per calendar day
|
|
if len(ins) > 0: days_map[d_str].in1 = format_time(ins[0])
|
|
if len(outs) > 0: days_map[d_str].out1 = format_time(outs[0])
|
|
if len(ins) > 1: days_map[d_str].in2 = format_time(ins[1])
|
|
if len(outs) > 1: days_map[d_str].out2 = format_time(outs[1])
|
|
|
|
return [days_map[k] for k in sorted(days_map.keys())]
|
|
|
|
class ExcelWriter:
|
|
"""Generates styled worksheet with dynamic formulas for balances and German break laws."""
|
|
|
|
def __init__(self, config: Config):
|
|
self.config = config
|
|
self.wb = Workbook()
|
|
|
|
def save(self, workdays: list[WorkDay], filename="work_times.xlsx"):
|
|
ws = self.wb.active
|
|
ws.title = "Work Hours"
|
|
|
|
headers = ["Date", "Phase", "In1", "Out1", "In2", "Out2", "Gross Time",
|
|
"Pause", "Net Time", "Target Time", "Semester Balance",
|
|
"Fulltime Balance", "Notes"]
|
|
ws.append(headers)
|
|
|
|
thin_border = Border(left=Side(style='thin', color='D9D9D9'), right=Side(style='thin', color='D9D9D9'),
|
|
top=Side(style='thin', color='D9D9D9'), bottom=Side(style='thin', color='D9D9D9'))
|
|
|
|
styles = {
|
|
"Semester": {"fill": PatternFill("solid", fgColor="E2EFDA"), "font": Font(name="Calibri", size=11, color="375623")},
|
|
"Fulltime": {"fill": PatternFill("solid", fgColor="FFF2CC"), "font": Font(name="Calibri", size=11, color="7F6000")}
|
|
}
|
|
|
|
for i, wd in enumerate(workdays, start=2):
|
|
gross_f = f"=(IF(D{i}>C{i},D{i}-C{i},0))+(IF(F{i}>E{i},F{i}-E{i},0))"
|
|
|
|
# German statutory breaks (ArbZG): 30m after 6h total, 45m after 9h. Deduct manual gap if exists.
|
|
pause_f = (f'=MAX(0, IF(M{i}="Berufsschule", TIME(0,0,0), IF(G{i}>=TIME(9,0,0), TIME(0,45,0), '
|
|
f'IF(G{i}>=TIME(6,0,0), TIME(0,30,0), TIME(0,0,0)))) - IF(AND(E{i}>0, D{i}>0), E{i}-D{i}, TIME(0,0,0)))')
|
|
|
|
net_f = f"=IF(G{i}>0, G{i}-H{i}, 0)"
|
|
sem_bal_f = f'=SUMIF($B$2:B{i}, "Semester", $I$2:I{i})'
|
|
|
|
# Workaround for Excel's 1900 date system which errors out on negative times
|
|
val_expr = f'ROUND(SUMIF($B$2:B{i}, "Fulltime", $I$2:I{i}) - SUMIF($B$2:B{i}, "Fulltime", $J$2:J{i}), 10)'
|
|
full_bal_f = f'=IF({val_expr}<0, "-" & TEXT(ABS({val_expr}), "[h]:mm:ss"), TEXT({val_expr}, "[h]:mm:ss"))'
|
|
|
|
row_data = [wd.date_obj, wd.phase, wd.in1 or None, wd.out1 or None, wd.in2 or None, wd.out2 or None,
|
|
gross_f, pause_f, net_f, wd.target_time_formula, sem_bal_f, full_bal_f, wd.note or None]
|
|
|
|
for col_idx, value in enumerate(row_data, start=1):
|
|
c = ws.cell(row=i, column=col_idx, value=value)
|
|
c.border = thin_border
|
|
c.alignment = Alignment(horizontal='left' if col_idx == 13 else 'center', vertical='center')
|
|
|
|
ws.cell(row=i, column=1).number_format = 'yyyy-mm-dd'
|
|
for col in range(3, 12):
|
|
ws.cell(row=i, column=col).number_format = '[h]:mm:ss'
|
|
ws.cell(row=i, column=12).number_format = '@'
|
|
|
|
p_cell = ws.cell(row=i, column=2)
|
|
p_cell.fill, p_cell.font = styles[wd.phase]["fill"], styles[wd.phase]["font"]
|
|
|
|
# Style headers
|
|
for col_idx in range(1, 14):
|
|
c = ws.cell(row=1, column=col_idx)
|
|
c.fill = PatternFill("solid", fgColor="1F497D")
|
|
c.font = Font(name="Calibri", size=11, color="FFFFFF", bold=True)
|
|
c.alignment, c.border = Alignment(horizontal='center', vertical='center'), thin_border
|
|
|
|
col_widths = {"A": 14, "B": 14, "C": 11, "D": 11, "E": 11, "F": 11, "G": 14, "H": 11, "I": 12, "J": 14, "K": 18, "L": 18, "M": 20}
|
|
for col_letter, width in col_widths.items():
|
|
ws.column_dimensions[col_letter].width = width
|
|
|
|
# Overview Sheet Calculations
|
|
ov = self.wb.create_sheet(title="Overview")
|
|
last_r = max(2, len(workdays) + 1)
|
|
|
|
ov.merge_cells("A1:B1")
|
|
ov["A1"] = "Semester Target & Pace"
|
|
ov["A1"].fill = PatternFill("solid", fgColor="1F497D")
|
|
ov["A1"].font = Font(name="Calibri", size=11, color="FFFFFF", bold=True)
|
|
ov["A1"].alignment = Alignment(horizontal='center')
|
|
|
|
wks = self.config.semester_weeks
|
|
ay_start = self.config.acad_year_start
|
|
|
|
metrics = [
|
|
("Target Hours (Total)", self.config.semester_target_total / 24.0, '[h]:mm:ss'),
|
|
("Worked Hours", f'=SUMIFS(\'Work Hours\'!I2:I{last_r}, \'Work Hours\'!B2:B{last_r}, "Semester", \'Work Hours\'!A2:A{last_r}, ">="&DATE({ay_start.year},{ay_start.month},{ay_start.day}))', '[h]:mm:ss'),
|
|
("Remaining Hours", "=B2-B3", '[h]:mm:ss'),
|
|
("Remaining Weeks", self.config.remaining_weeks, '0.0'),
|
|
("Required Pace / Wk", "=IF(B5>0, B4/B5, 0)", '[h]:mm:ss'),
|
|
("Current Deficit", f'=IF(B5>0, ((({wks}-B5)*(B2/{wks})) - B3), 0)', '[h]:mm:ss')
|
|
]
|
|
|
|
for idx, (label, val, fmt) in enumerate(metrics, start=2):
|
|
ov[f"A{idx}"] = label
|
|
ov[f"B{idx}"] = val
|
|
ov[f"A{idx}"].font = Font(name="Calibri", size=10, bold=True)
|
|
ov[f"B{idx}"].font = Font(name="Calibri", size=10)
|
|
ov[f"B{idx}"].number_format = fmt
|
|
|
|
for c in "AB":
|
|
cell = ov[f"{c}{idx}"]
|
|
cell.border = thin_border
|
|
cell.alignment = Alignment(horizontal='left' if c == "A" else 'center')
|
|
|
|
ov.column_dimensions["A"].width, ov.column_dimensions["B"].width = 24, 14
|
|
|
|
self.wb.save(filename)
|
|
print(f"Data saved to {filename}")
|
|
|
|
def main():
|
|
try:
|
|
config = Config()
|
|
|
|
api = TimeRecordingAPI(config)
|
|
api.authenticate()
|
|
records, absences = api.fetch_data()
|
|
|
|
processor = TimesheetProcessor(config)
|
|
workdays = processor.process(records, absences)
|
|
|
|
writer = ExcelWriter(config)
|
|
writer.save(workdays)
|
|
|
|
except Exception as e:
|
|
print(f"Process failed: {e}")
|
|
exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|