124 lines
4.6 KiB
Python
124 lines
4.6 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",
|
|
))
|
|
existing_ids = set(db.scalars(select(TimeEntry.external_id).where(
|
|
TimeEntry.source == "zed",
|
|
TimeEntry.device_id == device_id,
|
|
TimeEntry.project_slug == project_slug,
|
|
)))
|
|
|
|
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())
|
|
# Blocks whose stable ID already exists (accepted, rejected, or still
|
|
# suggested) must not be re-added: the unique constraint would reject
|
|
# the insert, and user decisions about a block must stick.
|
|
if entry_id in existing_ids:
|
|
continue
|
|
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 overlaps_entries(entry: TimeEntry, accepted_entries: list[TimeEntry]) -> bool:
|
|
return any(
|
|
other.id != entry.id and other.start_at < entry.end_at and other.end_at > entry.start_at
|
|
for other in accepted_entries
|
|
)
|
|
|
|
|
|
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}
|
|
|
|
|
|
def group_entries_by_day(entries: list[TimeEntry]) -> dict:
|
|
days: dict = defaultdict(list)
|
|
for entry in entries:
|
|
days[entry.start_at.astimezone(TZ).date()].append(entry)
|
|
for day_entries in days.values():
|
|
day_entries.sort(key=lambda item: item.start_at)
|
|
return dict(days)
|
|
|
|
|
|
def group_commits_by_day(commits: list[RawEvent]) -> dict:
|
|
days: dict = defaultdict(list)
|
|
for commit in commits:
|
|
if commit.occurred_at is not None:
|
|
days[commit.occurred_at.astimezone(TZ).date()].append(commit)
|
|
for day_commits in days.values():
|
|
day_commits.sort(key=lambda item: item.occurred_at)
|
|
return dict(days)
|