from __future__ import annotations from datetime import date from typing import Any from config import settings from services.distributor import DayEntry from services.gitlab_client import GitLabClient from services.holidays_util import is_workday from services.planner import Planner from services.pts_client import PTSClient from services.state_store import ( get_last_entries, get_pts_session, load_state, mark_dates_filled, set_last_entries, ) from services.teams_calendar_scraper import TeamsCalendarScraper class FillService: def __init__(self) -> None: self.gitlab = GitLabClient() self.pts = PTSClient() self.planner = Planner() async def fetch_gitlab_tasks(self) -> list[dict[str, Any]]: tasks = await self.gitlab.get_assigned_tasks() return [ { "id": t.id, "iid": t.iid, "title": t.title, "reference": t.reference, "web_url": t.web_url, "source": t.source, "labels": t.labels or [], "description": t.to_description(), } for t in tasks ] async def preview_plan( self, start: date, end: date, *, use_last_if_empty: bool = True, use_llm: bool | None = None, ) -> dict[str, Any]: result = await self.planner.build_plan( start, end, use_llm=use_llm, use_last_if_empty=use_last_if_empty, ) payload = {k: v for k, v in result.items() if k != "_plan_entries"} return payload async def preview_plan_for_days( self, days: list[date], *, use_last_if_empty: bool = True, use_llm: bool | None = None, ) -> dict[str, Any]: result = await self.planner.build_plan_for_days( days, use_llm=use_llm, use_last_if_empty=use_last_if_empty, ) return {k: v for k, v in result.items() if k != "_plan_entries"} async def fill_dates( self, start: date, end: date, *, skip_existing: bool = True, dry_run: bool = False, use_llm: bool | None = None, ) -> dict[str, Any]: planned = await self.planner.build_plan(start, end, use_llm=use_llm) return await self._execute_plan( planned, skip_existing=skip_existing, dry_run=dry_run, ) async def fill_date_list( self, days: list[date], *, skip_existing: bool = True, dry_run: bool = False, use_llm: bool | None = None, ) -> dict[str, Any]: planned = await self.planner.build_plan_for_days(days, use_llm=use_llm) result = await self._execute_plan( planned, skip_existing=skip_existing, dry_run=dry_run, ) result["requested_dates"] = [d.isoformat() for d in sorted(set(days))] return result async def _execute_plan( self, planned: dict[str, Any], *, skip_existing: bool, dry_run: bool, ) -> dict[str, Any]: plan: dict[str, list[DayEntry]] = planned["_plan_entries"] source = planned["source"] if not any(plan.values()): return { "ok": False, "message": "No GitLab tasks and no previous entries to reuse.", "source": source, "analysis": planned.get("analysis", {}), "created": [], "skipped": [], } project_code = await self.pts.resolve_project_code() default_task_id = await self.pts.resolve_task_id() task_id_cache: dict[str, int] = {} created: list[dict[str, Any]] = [] skipped: list[dict[str, Any]] = [] errors: list[dict[str, Any]] = [] for day, entries in sorted(plan.items()): if not entries: continue if skip_existing: existing = await self.pts.search_reports(day, day, project_code=project_code) existing_hours = sum(float(r.get("takeHours") or 0) for r in existing) if existing_hours >= settings.max_hours_per_day - 0.01: skipped.append({"date": day, "reason": "already_filled", "hours": existing_hours}) continue for entry in entries: task_name = entry.pts_task_name or settings.pts_default_task_name if task_name not in task_id_cache: task_id_cache[task_name] = await self.pts.resolve_task_id(task_name) task_id = task_id_cache[task_name] payload = { "projectCode": project_code, "taskId": task_id, "description": entry.description, "date": day.isoformat() if hasattr(day, "isoformat") else str(day), "takeHours": entry.take_hours, "licenseIds": [], "pts_task_name": task_name, } if dry_run: created.append({"date": day, **payload, "dry_run": True}) continue try: report_id = await self.pts.create_report(payload) created.append({"date": day, "id": report_id, **payload}) except Exception as exc: # noqa: BLE001 errors.append({"date": day, "description": entry.description, "error": str(exc)}) if created and not dry_run: mark_dates_filled(list({c["date"] for c in created})) set_last_entries( [ { "description": e.description, "take_hours": e.take_hours, "gitlab_task_id": e.gitlab_task_id, } for entries in plan.values() for e in entries ][:10] ) # Include human-readable plan for the web UI (same shape as preview) plan_public: dict[str, list[dict[str, Any]]] = {} for day, entries in plan.items(): day_key = day.isoformat() if hasattr(day, "isoformat") else str(day) plan_public[day_key] = [ { "description": e.description, "take_hours": e.take_hours, "pts_task_name": e.pts_task_name, "gitlab_task_id": e.gitlab_task_id, } for e in entries ] return { "ok": len(errors) == 0, "source": source, "analysis": planned.get("analysis", {}), "project_code": project_code, "task_id": default_task_id, "created": created, "skipped": skipped, "errors": errors, "dry_run": dry_run, "workdays": planned.get("workdays", list(plan_public.keys())), "gitlab_task_count": planned.get("gitlab_task_count"), "plan": plan_public, } async def fill_today(self, *, dry_run: bool = False, use_llm: bool | None = None) -> dict[str, Any]: today = date.today() if not is_workday(today): return { "ok": True, "message": f"{today.isoformat()} is not a workday (weekend/holiday). Skipped.", "created": [], "skipped": [{"date": today.isoformat(), "reason": "holiday_or_weekend"}], } return await self.fill_dates(today, today, dry_run=dry_run, use_llm=use_llm) async def get_status(self) -> dict[str, Any]: state = load_state() session = get_pts_session() has_session = bool(session and session.get("accessToken")) status: dict[str, Any] = { "has_pts_session": has_session, "pts_session_updated_at": state.get("pts_session_updated_at"), "last_entries_count": len(get_last_entries()), "gitlab_configured": bool(settings.gitlab_token), "llm_configured": bool(settings.xai_api_key), "use_llm": settings.use_llm, "use_description_variation": settings.use_description_variation, "grok_model": settings.grok_model, "auto_fill_enabled": settings.auto_fill_enabled, "auto_fill_time": f"{settings.auto_fill_hour:02d}:{settings.auto_fill_minute:02d}", "teams_calendar_enabled": settings.teams_calendar_enabled, "pts_username": settings.pts_username or None, "teams_calendar": TeamsCalendarScraper().status(), # backward-compatible alias for old UI fields "graph_calendar": TeamsCalendarScraper().status(), } if has_session: try: pts_info = await self.pts.verify_connection() status.update(pts_info) status["pts_projects"] = pts_info["project_count"] status["pts_task_types"] = pts_info["task_type_count"] except Exception as exc: # noqa: BLE001 status["pts_error"] = str(exc) try: gitlab_tasks = await self.fetch_gitlab_tasks() status["gitlab_open_tasks"] = len(gitlab_tasks) except Exception as exc: # noqa: BLE001 status["gitlab_error"] = str(exc) return status def ensure_ready(self) -> None: if not settings.gitlab_token: raise ValueError("Missing GITLAB_TOKEN in .env") session = get_pts_session() if not session or not session.get("accessToken"): raise ValueError( "Missing PTS session. Login with PTS account/password on the web UI (or sync via Chrome extension as fallback)." )