"""Loop-break week summary for Home banners and agent context.""" from __future__ import annotations import re from datetime import date, datetime, timedelta, timezone from typing import Any from fastapi import APIRouter, Depends, Query, Request from app.deps import require_login from app.models import ApiEnvelope, ok from app.store_daily import indoors_streak router = APIRouter(prefix="/api/loop", tags=["loop"]) _LABEL_ONLY = re.compile(r"\blabel(?:ing|ling)?\b", re.IGNORECASE) def loop_summary_for( *, entry_store: Any, daily_store: Any, as_of: date, days: int = 7, ) -> dict[str, Any]: """Aggregate indoors / proof / label-only / movement / pending for a window.""" start = as_of - timedelta(days=max(1, days) - 1) rows = [r for r in daily_store._load() if start <= r.date <= as_of] entries = [] for entry in entry_store._load(): day = entry.ts.date() if hasattr(entry.ts, "date") else entry.ts if start <= day <= as_of: entries.append(entry) streak = indoors_streak(daily_store._load(), as_of=as_of) proof_days = sum(1 for r in rows if r.proof_brick_done) proof_rate = (proof_days / len(rows)) if rows else 0.0 movement_sum = sum(int(r.movement_minutes or 0) for r in rows) high_i = [e for e in entries if (e.intensity or 0) >= 7] label_fails = [ e for e in high_i if _LABEL_ONLY.search(e.remedy or "") and e.result.value in ("failed", "partial", "pending") ] label_only_fail_rate = (len(label_fails) / len(high_i)) if high_i else 0.0 pending_count = sum(1 for e in entry_store._load() if e.result.value == "pending") return { "days": days, "as_of": as_of.isoformat(), "indoors_streak": streak, "proof_rate": round(proof_rate, 4), "proof_days": proof_days, "days_with_daily": len(rows), "label_only_fail_rate": round(label_only_fail_rate, 4), "label_only_fail_n": len(label_fails), "high_intensity_n": len(high_i), "movement_minutes_sum": movement_sum, "pending_count": pending_count, } @router.get("/summary", response_model=ApiEnvelope) def get_loop_summary( request: Request, days: int = Query(default=7, ge=1, le=30), _user: str = Depends(require_login), ) -> dict[str, Any]: as_of = datetime.now(timezone.utc).date() return ok( loop_summary_for( entry_store=request.app.state.entry_store, daily_store=request.app.state.daily_store, as_of=as_of, days=days, ) )