eight-hourr/backend/services/distributor.py

147 lines
3.8 KiB
Python

from __future__ import annotations
import random
from dataclasses import dataclass
from datetime import date
from config import settings
from services.gitlab_client import GitLabTask
from services.holidays_util import iter_workdays
@dataclass
class DayEntry:
date: str
description: str
take_hours: float
gitlab_task_id: int | None = None
pts_task_name: str | None = None
def _to_units(hours: float, step: float) -> int:
return int(round(hours / step))
def _from_units(units: int, step: float) -> float:
return round(units * step, 1)
def distribute_hours(num_tasks: int, max_hours: float, step: float) -> list[float]:
if num_tasks <= 0:
return []
if num_tasks == 1:
return [max_hours]
max_units = _to_units(max_hours, step)
min_units = 1
if max_units < num_tasks * min_units:
return [_from_units(min_units, step)] * num_tasks
remaining = max_units - num_tasks * min_units
weights = [random.random() for _ in range(num_tasks)]
total_weight = sum(weights) or 1
extra = [int(remaining * w / total_weight) for w in weights]
leftover = remaining - sum(extra)
idx = 0
while leftover > 0:
extra[idx % num_tasks] += 1
leftover -= 1
idx += 1
return [_from_units(min_units + extra[i], step) for i in range(num_tasks)]
def build_daily_plan(
tasks: list[GitLabTask],
workdays: list[date],
*,
max_hours: float | None = None,
hour_step: float | None = None,
tasks_per_day: int | None = None,
) -> dict[str, list[DayEntry]]:
max_hours = max_hours or settings.max_hours_per_day
hour_step = hour_step or settings.hour_step
if not workdays:
return {}
if not tasks:
return {d.isoformat(): [] for d in workdays}
per_day = tasks_per_day or min(3, max(1, len(tasks)))
plan: dict[str, list[DayEntry]] = {}
task_index = 0
for workday in workdays:
day_key = workday.isoformat()
day_tasks: list[GitLabTask] = []
for _ in range(per_day):
day_tasks.append(tasks[task_index % len(tasks)])
task_index += 1
hours = distribute_hours(len(day_tasks), max_hours, hour_step)
plan[day_key] = [
DayEntry(
date=day_key,
description=task.to_description(),
take_hours=hours[i],
gitlab_task_id=task.id,
)
for i, task in enumerate(day_tasks)
]
return plan
def build_plan_for_dates(
tasks: list[GitLabTask],
start: date,
end: date,
) -> dict[str, list[DayEntry]]:
workdays = iter_workdays(start, end)
return build_daily_plan(tasks, workdays)
def build_plan_for_day_list(
tasks: list[GitLabTask],
days: list[date],
) -> dict[str, list[DayEntry]]:
sorted_days = sorted(set(days))
return build_daily_plan(tasks, sorted_days)
def entries_from_last_state(
last_entries: list[dict],
workdays: list[date],
) -> dict[str, list[DayEntry]]:
if not last_entries or not workdays:
return {}
templates = [
DayEntry(
date="",
description=e.get("description", ""),
take_hours=float(e.get("take_hours", 2)),
gitlab_task_id=e.get("gitlab_task_id"),
)
for e in last_entries
if e.get("description")
]
if not templates:
return {}
plan: dict[str, list[DayEntry]] = {}
for workday in workdays:
day_key = workday.isoformat()
hours = distribute_hours(len(templates), settings.max_hours_per_day, settings.hour_step)
plan[day_key] = [
DayEntry(
date=day_key,
description=templates[i].description,
take_hours=hours[i],
gitlab_task_id=templates[i].gitlab_task_id,
)
for i in range(len(templates))
]
return plan