import uuid from datetime import datetime, timezone from pydantic import BaseModel, Field, field_validator def _calc_heartbeat_status(last_heartbeat_at: datetime | None) -> str: """Berechnet den Liveness-Status aus dem letzten Heartbeat-Zeitstempel.""" if last_heartbeat_at is None: return "offline" now = datetime.now(timezone.utc) # last_heartbeat_at könnte timezone-naive sein (ältere DB-Einträge) if last_heartbeat_at.tzinfo is None: last_heartbeat_at = last_heartbeat_at.replace(tzinfo=timezone.utc) delta = (now - last_heartbeat_at).total_seconds() if delta <= 90: return "online" if delta <= 300: # 5 Minuten return "stale" return "offline" # ── Admin-Schemas ───────────────────────────────────────────────────────────── class KioskDeviceCreate(BaseModel): name: str = Field(..., min_length=1, max_length=255) location: str | None = Field(None, max_length=255) public_key: str # Ed25519 PEM oder OpenSSH ip_whitelist: str | None = None # CIDR-Liste kommasepariert, z.B. "10.0.0.0/24,192.168.1.0/24" @field_validator("name") @classmethod def name_not_blank(cls, v: str) -> str: if not v.strip(): raise ValueError("Name darf nicht nur aus Leerzeichen bestehen.") return v.strip() @field_validator("public_key") @classmethod def public_key_not_blank(cls, v: str) -> str: if not v.strip(): raise ValueError("Public Key darf nicht leer sein.") return v.strip() class KioskDeviceUpdate(BaseModel): name: str | None = Field(None, min_length=1, max_length=255) location: str | None = Field(None, max_length=255) ip_whitelist: str | None = None @field_validator("name") @classmethod def name_not_blank(cls, v: str | None) -> str | None: if v is not None: if not v.strip(): raise ValueError("Name darf nicht nur aus Leerzeichen bestehen.") return v.strip() return v class KioskDeviceOut(BaseModel): model_config = {"from_attributes": True} id: uuid.UUID company_id: uuid.UUID name: str location: str | None status: str # pending / approved / revoked public_key: str | None key_algorithm: str last_heartbeat_at: datetime | None client_version: str | None offline_queue_size: int ip_whitelist: str | None created_at: datetime heartbeat_status: str = "offline" # wird in model_validator gesetzt @classmethod def model_validate(cls, obj, *args, **kwargs): # type: ignore[override] instance = super().model_validate(obj, *args, **kwargs) instance.heartbeat_status = _calc_heartbeat_status(instance.last_heartbeat_at) return instance class KioskApproveResponse(BaseModel): id: uuid.UUID status: str message: str # ── Heartbeat-Schemas ───────────────────────────────────────────────────────── class HeartbeatRequest(BaseModel): uptime_seconds: int | None = None current_user_id: uuid.UUID | None = None browser_version: str | None = None queued_offline_entries: int = 0 client_version: str | None = None class HeartbeatResponse(BaseModel): server_time: str # ISO-8601-Zeitstempel des Servers heartbeat_interval_sec: int device_status: str # pending / approved / revoked