maddin
149 lines
4.2 KiB
Python
149 lines
4.2 KiB
Python
from datetime import date, timedelta
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import User, WeeklyTargetRule
|
|
|
|
DEFAULT_REFERENCE_WEEK_START = date(1970, 1, 5) # Montag
|
|
|
|
|
|
def monday_of(day: date) -> date:
|
|
return day - timedelta(days=day.weekday())
|
|
|
|
|
|
def week_starts_between(start_week_start: date, end_week_start: date) -> list[date]:
|
|
weeks: list[date] = []
|
|
current = start_week_start
|
|
while current <= end_week_start:
|
|
weeks.append(current)
|
|
current += timedelta(days=7)
|
|
return weeks
|
|
|
|
|
|
def list_rules_for_user(db: Session, user_id: str) -> list[WeeklyTargetRule]:
|
|
stmt = (
|
|
select(WeeklyTargetRule)
|
|
.where(WeeklyTargetRule.user_id == user_id)
|
|
.order_by(WeeklyTargetRule.effective_from.asc())
|
|
)
|
|
return db.execute(stmt).scalars().all()
|
|
|
|
|
|
def target_for_week(
|
|
rules: list[WeeklyTargetRule],
|
|
week_start: date,
|
|
fallback_minutes: int,
|
|
) -> int:
|
|
target = fallback_minutes
|
|
for rule in rules:
|
|
if rule.effective_from <= week_start:
|
|
target = rule.weekly_target_minutes
|
|
else:
|
|
break
|
|
return target
|
|
|
|
|
|
def target_map_for_weeks(
|
|
rules: list[WeeklyTargetRule],
|
|
week_starts: list[date],
|
|
fallback_minutes: int,
|
|
) -> dict[date, int]:
|
|
result: dict[date, int] = {}
|
|
for week_start in week_starts:
|
|
result[week_start] = target_for_week(rules, week_start, fallback_minutes)
|
|
return result
|
|
|
|
|
|
def upsert_rule(db: Session, user_id: str, effective_from: date, weekly_target_minutes: int) -> None:
|
|
stmt = select(WeeklyTargetRule).where(
|
|
WeeklyTargetRule.user_id == user_id,
|
|
WeeklyTargetRule.effective_from == effective_from,
|
|
)
|
|
rule = db.execute(stmt).scalar_one_or_none()
|
|
|
|
if rule:
|
|
rule.weekly_target_minutes = weekly_target_minutes
|
|
return
|
|
|
|
db.add(
|
|
WeeklyTargetRule(
|
|
user_id=user_id,
|
|
effective_from=effective_from,
|
|
weekly_target_minutes=weekly_target_minutes,
|
|
)
|
|
)
|
|
|
|
|
|
def ensure_user_has_default_target_rule(db: Session, user: User) -> None:
|
|
stmt = select(WeeklyTargetRule.id).where(WeeklyTargetRule.user_id == user.id).limit(1)
|
|
existing = db.execute(stmt).scalar_one_or_none()
|
|
if existing:
|
|
return
|
|
|
|
db.add(
|
|
WeeklyTargetRule(
|
|
user_id=user.id,
|
|
effective_from=DEFAULT_REFERENCE_WEEK_START,
|
|
weekly_target_minutes=user.weekly_target_minutes,
|
|
)
|
|
)
|
|
|
|
|
|
def ensure_all_users_have_default_target_rules(db: Session) -> None:
|
|
users = db.execute(select(User)).scalars().all()
|
|
changed = False
|
|
for user in users:
|
|
before_count = db.execute(
|
|
select(WeeklyTargetRule.id).where(WeeklyTargetRule.user_id == user.id).limit(1)
|
|
).scalar_one_or_none()
|
|
if before_count:
|
|
continue
|
|
ensure_user_has_default_target_rule(db, user)
|
|
changed = True
|
|
|
|
if changed:
|
|
db.commit()
|
|
|
|
|
|
def apply_weekly_target_change(
|
|
db: Session,
|
|
*,
|
|
user: User,
|
|
selected_week_start: date,
|
|
new_target_minutes: int,
|
|
scope: str,
|
|
) -> None:
|
|
rules = list_rules_for_user(db, user.id)
|
|
fallback = user.weekly_target_minutes
|
|
|
|
if scope == "all_weeks":
|
|
db.execute(delete(WeeklyTargetRule).where(WeeklyTargetRule.user_id == user.id))
|
|
db.add(
|
|
WeeklyTargetRule(
|
|
user_id=user.id,
|
|
effective_from=DEFAULT_REFERENCE_WEEK_START,
|
|
weekly_target_minutes=new_target_minutes,
|
|
)
|
|
)
|
|
return
|
|
|
|
if scope == "from_current_week":
|
|
db.execute(
|
|
delete(WeeklyTargetRule).where(
|
|
WeeklyTargetRule.user_id == user.id,
|
|
WeeklyTargetRule.effective_from >= selected_week_start,
|
|
)
|
|
)
|
|
upsert_rule(db, user.id, selected_week_start, new_target_minutes)
|
|
return
|
|
|
|
if scope == "current_week":
|
|
next_week_start = selected_week_start + timedelta(days=7)
|
|
target_next_week_before = target_for_week(rules, next_week_start, fallback)
|
|
upsert_rule(db, user.id, selected_week_start, new_target_minutes)
|
|
upsert_rule(db, user.id, next_week_start, target_next_week_before)
|
|
return
|
|
|
|
raise ValueError("Ungueltiger Scope")
|