88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
import hashlib
|
|
import os
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .models import RawEvent, TimeEntry
|
|
|
|
TZ = ZoneInfo(os.environ.get("APP_TIMEZONE", "Europe/Vienna"))
|
|
|
|
|
|
def parse_timestamp(value: str) -> datetime:
|
|
value = value.replace("Z", "+00:00")
|
|
parsed = datetime.fromisoformat(value)
|
|
if parsed.tzinfo is None:
|
|
return parsed.replace(tzinfo=TZ)
|
|
return parsed
|
|
|
|
|
|
def stable_id(*parts: str) -> str:
|
|
return hashlib.sha256("\x1f".join(parts).encode()).hexdigest()
|
|
|
|
|
|
def derive_zed_suggestions(db: Session, device_id: str, project_slug: str, idle_seconds: int = 300) -> int:
|
|
events = db.scalars(
|
|
select(RawEvent).where(
|
|
RawEvent.source == "zed_heartbeat",
|
|
RawEvent.device_id == device_id,
|
|
RawEvent.project_slug == project_slug,
|
|
).order_by(RawEvent.occurred_at)
|
|
).all()
|
|
active = [event for event in events if event.payload.get("zed_active") is True and event.occurred_at is not None]
|
|
|
|
# Accepted suggestions become manual entries and are intentionally retained.
|
|
db.execute(delete(TimeEntry).where(
|
|
TimeEntry.source == "zed",
|
|
TimeEntry.device_id == device_id,
|
|
TimeEntry.project_slug == project_slug,
|
|
TimeEntry.kind == "zed_inferred",
|
|
TimeEntry.status == "suggested",
|
|
))
|
|
|
|
blocks: list[tuple[datetime, datetime, int]] = []
|
|
start = previous = None
|
|
for event in active:
|
|
current = event.occurred_at
|
|
if current is None:
|
|
continue
|
|
if previous is None or (current - previous).total_seconds() > idle_seconds:
|
|
if start is not None and previous is not None and previous > start:
|
|
blocks.append((start, previous, int((previous - start).total_seconds())))
|
|
start = current
|
|
previous = current
|
|
if start is not None and previous is not None and previous > start:
|
|
blocks.append((start, previous, int((previous - start).total_seconds())))
|
|
|
|
for start_at, end_at, duration in blocks:
|
|
entry_id = stable_id(device_id, project_slug, start_at.isoformat(), end_at.isoformat())
|
|
db.add(TimeEntry(
|
|
source="zed", device_id=device_id, external_id=entry_id, kind="zed_inferred",
|
|
status="suggested", start_at=start_at, end_at=end_at, duration_seconds=duration,
|
|
task="Zed activity", project_slug=project_slug,
|
|
))
|
|
return len(blocks)
|
|
|
|
|
|
def overlaps_accepted(db: Session, entry: TimeEntry) -> bool:
|
|
matching = db.scalars(select(TimeEntry).where(
|
|
TimeEntry.status == "accepted", TimeEntry.id != entry.id,
|
|
TimeEntry.start_at < entry.end_at, TimeEntry.end_at > entry.start_at,
|
|
)).all()
|
|
return bool(matching)
|
|
|
|
|
|
def daily_summary(entries: list[TimeEntry]) -> dict:
|
|
totals: dict[str, int] = defaultdict(int)
|
|
suggestions = 0
|
|
for entry in entries:
|
|
day = entry.start_at.astimezone(TZ).date().isoformat()
|
|
if entry.status == "accepted":
|
|
totals[day] += entry.duration_seconds
|
|
elif entry.status == "suggested":
|
|
suggestions += entry.duration_seconds
|
|
return {"by_day": dict(totals), "suggested_seconds": suggestions}
|