55 lines
1.4 KiB
Python
55 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from core.models import ContactAvailabilitySettings, Person, PersonIdentifier, User
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AvailabilitySignal:
|
|
user: User
|
|
person: Person
|
|
person_identifier: PersonIdentifier | None
|
|
service: str
|
|
source_kind: str
|
|
availability_state: str
|
|
confidence: float = 0.8
|
|
ts: int = 0
|
|
payload: dict | None = None
|
|
|
|
|
|
def get_settings(user: User) -> ContactAvailabilitySettings:
|
|
settings_row, _ = ContactAvailabilitySettings.objects.get_or_create(user=user)
|
|
return settings_row
|
|
|
|
|
|
def record_native_signal(signal: AvailabilitySignal) -> AvailabilitySignal | None:
|
|
"""
|
|
Compatibility adapter for existing router call sites.
|
|
|
|
Availability state is now derived from behavioral events in Manticore, so this
|
|
function no longer persists a separate ORM availability row.
|
|
"""
|
|
settings_row = get_settings(signal.user)
|
|
if not settings_row.enabled:
|
|
return None
|
|
return signal
|
|
|
|
|
|
def record_inferred_signal(signal: AvailabilitySignal) -> AvailabilitySignal | None:
|
|
settings_row = get_settings(signal.user)
|
|
if not settings_row.enabled or not settings_row.inference_enabled:
|
|
return None
|
|
return signal
|
|
|
|
|
|
def ensure_fading_state(
|
|
*,
|
|
user: User,
|
|
person: Person,
|
|
person_identifier: PersonIdentifier | None,
|
|
service: str,
|
|
at_ts: int | None = None,
|
|
) -> None:
|
|
return None
|