821675ab4a
Specs stay at repo root (cross-VM). Move deploy and code into logical projects with README per domain, updated manifest.yaml, and symlinks at legacy paths for VM122 backward compatibility.
115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
"""Cálculo de durações do funil onboarding (Spec 014)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timezone
|
||
|
||
|
||
def _parse_iso(iso: str | None) -> datetime | None:
|
||
if not iso:
|
||
return None
|
||
try:
|
||
dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def format_duration(seconds: float | int | None) -> str:
|
||
if seconds is None:
|
||
return "—"
|
||
sec = max(0, int(round(float(seconds))))
|
||
if sec < 60:
|
||
return f"{sec}s"
|
||
mins, rem = divmod(sec, 60)
|
||
if mins < 60:
|
||
return f"{mins}m {rem}s"
|
||
hrs, mins = divmod(mins, 60)
|
||
if hrs < 48:
|
||
return f"{hrs}h {mins}m"
|
||
days, hrs = divmod(hrs, 24)
|
||
return f"{days}d {hrs}h"
|
||
|
||
|
||
def enrich_timeline_events(events: list[dict]) -> list[dict]:
|
||
if not events:
|
||
return []
|
||
start_dt = _parse_iso(events[0].get("created_at") or events[0].get("at"))
|
||
prev_dt = None
|
||
enriched: list[dict] = []
|
||
for idx, ev in enumerate(events):
|
||
at = ev.get("created_at") or ev.get("at")
|
||
cur_dt = _parse_iso(at)
|
||
from_prev = None
|
||
from_start = None
|
||
if cur_dt and prev_dt:
|
||
from_prev = (cur_dt - prev_dt).total_seconds()
|
||
if cur_dt and start_dt:
|
||
from_start = (cur_dt - start_dt).total_seconds()
|
||
row = dict(ev)
|
||
row["duration_from_prev_sec"] = from_prev if idx > 0 else 0
|
||
row["duration_from_start_sec"] = from_start
|
||
row["duration_from_prev_label"] = format_duration(from_prev) if idx > 0 else "—"
|
||
row["duration_from_start_label"] = format_duration(from_start)
|
||
enriched.append(row)
|
||
if cur_dt:
|
||
prev_dt = cur_dt
|
||
return enriched
|
||
|
||
|
||
def build_timing_report(events: list[dict], *, now_iso: str | None = None) -> dict:
|
||
enriched = enrich_timeline_events(events)
|
||
if not enriched:
|
||
return {
|
||
"timing_enabled": True,
|
||
"events": [],
|
||
"total_duration_sec": None,
|
||
"total_duration_label": "—",
|
||
"started_at": None,
|
||
"completed_at": None,
|
||
"idle_since_sec": None,
|
||
"idle_since_label": "—",
|
||
}
|
||
last = enriched[-1]
|
||
start_dt = _parse_iso(enriched[0].get("created_at") or enriched[0].get("at"))
|
||
last_dt = _parse_iso(last.get("created_at") or last.get("at"))
|
||
completed_types = {"onboarding.completed", "onboarding.failed"}
|
||
last_type = last.get("event_type") or last.get("event")
|
||
is_done = last_type in completed_types
|
||
now_dt = _parse_iso(now_iso) or datetime.now(timezone.utc)
|
||
# Sessão activa: tempo total = agora − início (relógio a correr).
|
||
# Concluída: tempo total = último evento − início.
|
||
if is_done and last_dt and start_dt:
|
||
total_sec = (last_dt - start_dt).total_seconds()
|
||
elif start_dt:
|
||
total_sec = (now_dt - start_dt).total_seconds()
|
||
else:
|
||
total_sec = last.get("duration_from_start_sec")
|
||
idle_sec = None
|
||
if not is_done and last_dt:
|
||
idle_sec = (now_dt - last_dt).total_seconds()
|
||
return {
|
||
"timing_enabled": True,
|
||
"events": enriched,
|
||
"total_duration_sec": total_sec,
|
||
"total_duration_label": format_duration(total_sec),
|
||
"started_at": enriched[0].get("created_at") or enriched[0].get("at"),
|
||
"last_event_at": last.get("created_at") or last.get("at"),
|
||
"completed_at": last.get("created_at") or last.get("at") if is_done else None,
|
||
"idle_since_sec": idle_sec,
|
||
"idle_since_label": format_duration(idle_sec) if idle_sec is not None else "—",
|
||
"is_completed": is_done,
|
||
}
|
||
|
||
|
||
def apply_module_timing(events: list[dict]) -> tuple[list[dict], dict | None]:
|
||
from app.modules import store as module_store
|
||
|
||
if not module_store.is_module_enabled("funnel-timing") or not events:
|
||
return events, None
|
||
report = build_timing_report(events)
|
||
enriched = report.pop("events", events)
|
||
meta = {k: v for k, v in report.items() if k != "timing_enabled"}
|
||
return enriched, meta
|