Implement executing tasks
This commit is contained in:
123
core/management/commands/backfill_contact_availability.py
Normal file
123
core/management/commands/backfill_contact_availability.py
Normal file
@@ -0,0 +1,123 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Iterable
|
||||
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
from core.models import Message, User
|
||||
from core.presence import AvailabilitySignal, record_inferred_signal
|
||||
from core.presence.inference import now_ms
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "Backfill inferred contact availability events from historical message/read-receipt activity."
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument("--days", type=int, default=30)
|
||||
parser.add_argument("--limit", type=int, default=5000)
|
||||
parser.add_argument("--service", default="")
|
||||
parser.add_argument("--user-id", default="")
|
||||
parser.add_argument("--dry-run", action="store_true", default=False)
|
||||
|
||||
def _iter_messages(self, *, days: int, limit: int, service: str, user_id: str) -> Iterable[Message]:
|
||||
cutoff_ts = now_ms() - (max(1, int(days)) * 24 * 60 * 60 * 1000)
|
||||
qs = Message.objects.filter(ts__gte=cutoff_ts).select_related(
|
||||
"user", "session", "session__identifier", "session__identifier__person"
|
||||
)
|
||||
if service:
|
||||
qs = qs.filter(source_service=str(service).strip().lower())
|
||||
if user_id:
|
||||
qs = qs.filter(user_id=str(user_id).strip())
|
||||
return qs.order_by("ts")[: max(1, int(limit))]
|
||||
|
||||
def handle(self, *args, **options):
|
||||
days = max(1, int(options.get("days") or 30))
|
||||
limit = max(1, int(options.get("limit") or 5000))
|
||||
service_filter = str(options.get("service") or "").strip().lower()
|
||||
user_filter = str(options.get("user_id") or "").strip()
|
||||
dry_run = bool(options.get("dry_run"))
|
||||
|
||||
created = 0
|
||||
scanned = 0
|
||||
|
||||
for msg in self._iter_messages(days=days, limit=limit, service=service_filter, user_id=user_filter):
|
||||
scanned += 1
|
||||
identifier = getattr(getattr(msg, "session", None), "identifier", None)
|
||||
person = getattr(identifier, "person", None)
|
||||
user = getattr(msg, "user", None)
|
||||
if not identifier or not person or not user:
|
||||
continue
|
||||
|
||||
service = str(getattr(msg, "source_service", "") or identifier.service or "").strip().lower()
|
||||
if not service:
|
||||
continue
|
||||
|
||||
base_ts = int(getattr(msg, "ts", 0) or 0)
|
||||
message_author = str(getattr(msg, "custom_author", "") or "").strip().upper()
|
||||
outgoing = message_author in {"USER", "BOT"}
|
||||
|
||||
candidates = []
|
||||
if base_ts > 0:
|
||||
candidates.append(
|
||||
{
|
||||
"source_kind": "message_out" if outgoing else "message_in",
|
||||
"availability_state": "available",
|
||||
"confidence": 0.65 if outgoing else 0.75,
|
||||
"ts": base_ts,
|
||||
"payload": {
|
||||
"origin": "backfill_contact_availability",
|
||||
"message_id": str(msg.id),
|
||||
"inferred_from": "message_activity",
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
read_ts = int(getattr(msg, "read_ts", 0) or 0)
|
||||
if read_ts > 0:
|
||||
candidates.append(
|
||||
{
|
||||
"source_kind": "read_receipt",
|
||||
"availability_state": "available",
|
||||
"confidence": 0.95,
|
||||
"ts": read_ts,
|
||||
"payload": {
|
||||
"origin": "backfill_contact_availability",
|
||||
"message_id": str(msg.id),
|
||||
"inferred_from": "read_receipt",
|
||||
"read_by": str(getattr(msg, "read_by_identifier", "") or ""),
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
for row in candidates:
|
||||
exists = user.contact_availability_events.filter(
|
||||
person=person,
|
||||
person_identifier=identifier,
|
||||
service=service,
|
||||
source_kind=row["source_kind"],
|
||||
ts=int(row["ts"]),
|
||||
).exists()
|
||||
if exists:
|
||||
continue
|
||||
created += 1
|
||||
if dry_run:
|
||||
continue
|
||||
record_inferred_signal(
|
||||
AvailabilitySignal(
|
||||
user=user,
|
||||
person=person,
|
||||
person_identifier=identifier,
|
||||
service=service,
|
||||
source_kind=row["source_kind"],
|
||||
availability_state=row["availability_state"],
|
||||
confidence=float(row["confidence"]),
|
||||
ts=int(row["ts"]),
|
||||
payload=dict(row["payload"]),
|
||||
)
|
||||
)
|
||||
|
||||
self.stdout.write(
|
||||
self.style.SUCCESS(
|
||||
f"backfill_contact_availability complete scanned={scanned} created={created} dry_run={dry_run} days={days} limit={limit}"
|
||||
)
|
||||
)
|
||||
127
core/management/commands/codex_worker.py
Normal file
127
core/management/commands/codex_worker.py
Normal file
@@ -0,0 +1,127 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
from core.models import ExternalSyncEvent, TaskProviderConfig
|
||||
from core.tasks.providers import get_provider
|
||||
from core.util import logs
|
||||
|
||||
log = logs.get_logger("codex_worker")
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = "Process queued external sync events for worker-backed providers (codex_cli)."
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument("--once", action="store_true", default=False)
|
||||
parser.add_argument("--sleep-seconds", type=float, default=2.0)
|
||||
parser.add_argument("--batch-size", type=int, default=20)
|
||||
parser.add_argument("--provider", default="codex_cli")
|
||||
|
||||
def _claim_batch(self, provider: str, batch_size: int) -> list[str]:
|
||||
ids: list[str] = []
|
||||
rows = list(
|
||||
ExternalSyncEvent.objects.filter(
|
||||
provider=provider,
|
||||
status__in=["pending", "retrying"],
|
||||
)
|
||||
.order_by("updated_at")[: max(1, batch_size)]
|
||||
.values_list("id", flat=True)
|
||||
)
|
||||
for row_id in rows:
|
||||
updated = ExternalSyncEvent.objects.filter(
|
||||
id=row_id,
|
||||
provider=provider,
|
||||
status__in=["pending", "retrying"],
|
||||
).update(status="retrying")
|
||||
if updated:
|
||||
ids.append(str(row_id))
|
||||
return ids
|
||||
|
||||
def _run_event(self, event: ExternalSyncEvent) -> None:
|
||||
provider = get_provider(event.provider)
|
||||
if not bool(getattr(provider, "run_in_worker", False)):
|
||||
return
|
||||
|
||||
cfg = (
|
||||
TaskProviderConfig.objects.filter(
|
||||
user=event.user,
|
||||
provider=event.provider,
|
||||
enabled=True,
|
||||
)
|
||||
.order_by("-updated_at")
|
||||
.first()
|
||||
)
|
||||
if cfg is None:
|
||||
event.status = "failed"
|
||||
event.error = "provider_disabled_or_missing"
|
||||
event.save(update_fields=["status", "error", "updated_at"])
|
||||
return
|
||||
|
||||
payload = dict(event.payload or {})
|
||||
action = str(payload.get("action") or "append_update").strip().lower()
|
||||
provider_payload = dict(payload.get("provider_payload") or payload)
|
||||
|
||||
if action == "create":
|
||||
result = provider.create_task(dict(cfg.settings or {}), provider_payload)
|
||||
elif action == "complete":
|
||||
result = provider.mark_complete(dict(cfg.settings or {}), provider_payload)
|
||||
elif action == "link_task":
|
||||
result = provider.link_task(dict(cfg.settings or {}), provider_payload)
|
||||
else:
|
||||
result = provider.append_update(dict(cfg.settings or {}), provider_payload)
|
||||
|
||||
event.status = "ok" if result.ok else "failed"
|
||||
event.error = str(result.error or "")
|
||||
event.payload = dict(
|
||||
payload,
|
||||
worker_processed=True,
|
||||
result=dict(result.payload or {}),
|
||||
)
|
||||
event.save(update_fields=["status", "error", "payload", "updated_at"])
|
||||
|
||||
if result.ok and result.external_key and event.task_id and not str(event.task.external_key or "").strip():
|
||||
event.task.external_key = str(result.external_key)
|
||||
event.task.save(update_fields=["external_key"])
|
||||
|
||||
def handle(self, *args, **options):
|
||||
once = bool(options.get("once"))
|
||||
sleep_seconds = max(0.2, float(options.get("sleep_seconds") or 2.0))
|
||||
batch_size = max(1, int(options.get("batch_size") or 20))
|
||||
provider_name = str(options.get("provider") or "codex_cli").strip().lower()
|
||||
|
||||
log.info(
|
||||
"codex_worker started provider=%s once=%s sleep=%s batch_size=%s",
|
||||
provider_name,
|
||||
once,
|
||||
sleep_seconds,
|
||||
batch_size,
|
||||
)
|
||||
|
||||
while True:
|
||||
claimed_ids = self._claim_batch(provider_name, batch_size)
|
||||
if not claimed_ids:
|
||||
if once:
|
||||
log.info("codex_worker exiting: no pending events")
|
||||
return
|
||||
time.sleep(sleep_seconds)
|
||||
continue
|
||||
|
||||
for row_id in claimed_ids:
|
||||
event = ExternalSyncEvent.objects.filter(id=row_id).select_related("task", "user").first()
|
||||
if event is None:
|
||||
continue
|
||||
try:
|
||||
self._run_event(event)
|
||||
except Exception as exc:
|
||||
log.exception("codex_worker failed processing id=%s", row_id)
|
||||
ExternalSyncEvent.objects.filter(id=row_id).update(
|
||||
status="failed",
|
||||
error=f"worker_exception:{exc}",
|
||||
)
|
||||
|
||||
if once:
|
||||
log.info("codex_worker processed %s event(s)", len(claimed_ids))
|
||||
return
|
||||
243
core/management/commands/recalculate_contact_availability.py
Normal file
243
core/management/commands/recalculate_contact_availability.py
Normal file
@@ -0,0 +1,243 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import defaultdict
|
||||
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
from core.models import ContactAvailabilityEvent, ContactAvailabilitySpan, Message
|
||||
from core.presence import AvailabilitySignal, record_native_signal
|
||||
from core.presence.inference import now_ms
|
||||
|
||||
|
||||
_SOURCE_ORDER = {
|
||||
"message_in": 10,
|
||||
"message_out": 20,
|
||||
"read_receipt": 30,
|
||||
"native_presence": 40,
|
||||
}
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
help = (
|
||||
"Recalculate contact availability events/spans from persisted message, "
|
||||
"read-receipt, and reaction history (deterministic rebuild)."
|
||||
)
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument("--days", type=int, default=90)
|
||||
parser.add_argument("--limit", type=int, default=20000)
|
||||
parser.add_argument("--service", default="")
|
||||
parser.add_argument("--user-id", default="")
|
||||
parser.add_argument("--dry-run", action="store_true", default=False)
|
||||
parser.add_argument("--no-reset", action="store_true", default=False)
|
||||
|
||||
def _iter_messages(self, *, days: int, limit: int, service: str, user_id: str):
|
||||
cutoff_ts = now_ms() - (max(1, int(days)) * 24 * 60 * 60 * 1000)
|
||||
qs = Message.objects.filter(ts__gte=cutoff_ts).select_related(
|
||||
"user", "session", "session__identifier", "session__identifier__person"
|
||||
)
|
||||
if service:
|
||||
qs = qs.filter(source_service=str(service).strip().lower())
|
||||
if user_id:
|
||||
qs = qs.filter(user_id=str(user_id).strip())
|
||||
return qs.order_by("ts")[: max(1, int(limit))]
|
||||
|
||||
def _build_event_rows(self, messages):
|
||||
rows = []
|
||||
for msg in messages:
|
||||
identifier = getattr(getattr(msg, "session", None), "identifier", None)
|
||||
person = getattr(identifier, "person", None)
|
||||
user = getattr(msg, "user", None)
|
||||
if not identifier or not person or not user:
|
||||
continue
|
||||
|
||||
service = str(
|
||||
getattr(msg, "source_service", "") or getattr(identifier, "service", "")
|
||||
).strip().lower()
|
||||
if not service:
|
||||
continue
|
||||
|
||||
ts = int(getattr(msg, "ts", 0) or 0)
|
||||
if ts > 0:
|
||||
author = str(getattr(msg, "custom_author", "") or "").strip().upper()
|
||||
outgoing = author in {"USER", "BOT"}
|
||||
rows.append(
|
||||
{
|
||||
"user": user,
|
||||
"person": person,
|
||||
"person_identifier": identifier,
|
||||
"service": service,
|
||||
"source_kind": "message_out" if outgoing else "message_in",
|
||||
"availability_state": "available",
|
||||
"confidence": 0.65 if outgoing else 0.75,
|
||||
"ts": ts,
|
||||
"payload": {
|
||||
"origin": "recalculate_contact_availability",
|
||||
"message_id": str(msg.id),
|
||||
"inferred_from": "message_activity",
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
read_ts = int(getattr(msg, "read_ts", 0) or 0)
|
||||
if read_ts > 0:
|
||||
rows.append(
|
||||
{
|
||||
"user": user,
|
||||
"person": person,
|
||||
"person_identifier": identifier,
|
||||
"service": service,
|
||||
"source_kind": "read_receipt",
|
||||
"availability_state": "available",
|
||||
"confidence": 0.95,
|
||||
"ts": read_ts,
|
||||
"payload": {
|
||||
"origin": "recalculate_contact_availability",
|
||||
"message_id": str(msg.id),
|
||||
"inferred_from": "read_receipt",
|
||||
"read_by": str(getattr(msg, "read_by_identifier", "") or ""),
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
reactions = list((getattr(msg, "receipt_payload", {}) or {}).get("reactions") or [])
|
||||
for reaction in reactions:
|
||||
item = dict(reaction or {})
|
||||
if bool(item.get("removed")):
|
||||
continue
|
||||
reaction_ts = int(item.get("updated_at") or 0)
|
||||
if reaction_ts <= 0:
|
||||
continue
|
||||
rows.append(
|
||||
{
|
||||
"user": user,
|
||||
"person": person,
|
||||
"person_identifier": identifier,
|
||||
"service": service,
|
||||
"source_kind": "native_presence",
|
||||
"availability_state": "available",
|
||||
"confidence": 0.9,
|
||||
"ts": reaction_ts,
|
||||
"payload": {
|
||||
"origin": "recalculate_contact_availability",
|
||||
"message_id": str(msg.id),
|
||||
"inferred_from": "reaction",
|
||||
"emoji": str(item.get("emoji") or ""),
|
||||
"actor": str(item.get("actor") or ""),
|
||||
"source_service": str(item.get("source_service") or service),
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
rows.sort(
|
||||
key=lambda row: (
|
||||
str(getattr(row["user"], "id", "")),
|
||||
str(getattr(row["person"], "id", "")),
|
||||
str(row.get("service") or ""),
|
||||
int(row.get("ts") or 0),
|
||||
_SOURCE_ORDER.get(str(row.get("source_kind") or ""), 999),
|
||||
str((row.get("payload") or {}).get("message_id") or ""),
|
||||
)
|
||||
)
|
||||
return rows
|
||||
|
||||
def handle(self, *args, **options):
|
||||
days = max(1, int(options.get("days") or 90))
|
||||
limit = max(1, int(options.get("limit") or 20000))
|
||||
service_filter = str(options.get("service") or "").strip().lower()
|
||||
user_filter = str(options.get("user_id") or "").strip()
|
||||
dry_run = bool(options.get("dry_run"))
|
||||
reset = not bool(options.get("no_reset"))
|
||||
cutoff_ts = now_ms() - (days * 24 * 60 * 60 * 1000)
|
||||
|
||||
messages = list(
|
||||
self._iter_messages(
|
||||
days=days,
|
||||
limit=limit,
|
||||
service=service_filter,
|
||||
user_id=user_filter,
|
||||
)
|
||||
)
|
||||
rows = self._build_event_rows(messages)
|
||||
|
||||
keys_to_reset = set()
|
||||
for row in rows:
|
||||
keys_to_reset.add(
|
||||
(
|
||||
str(getattr(row["user"], "id", "")),
|
||||
str(getattr(row["person"], "id", "")),
|
||||
str(row.get("service") or ""),
|
||||
)
|
||||
)
|
||||
|
||||
deleted_events = 0
|
||||
deleted_spans = 0
|
||||
if reset and keys_to_reset and not dry_run:
|
||||
for user_id, person_id, service in keys_to_reset:
|
||||
deleted_events += ContactAvailabilityEvent.objects.filter(
|
||||
user_id=user_id,
|
||||
person_id=person_id,
|
||||
service=service,
|
||||
ts__gte=cutoff_ts,
|
||||
).delete()[0]
|
||||
deleted_spans += ContactAvailabilitySpan.objects.filter(
|
||||
user_id=user_id,
|
||||
person_id=person_id,
|
||||
service=service,
|
||||
end_ts__gte=cutoff_ts,
|
||||
).delete()[0]
|
||||
|
||||
created = 0
|
||||
dedup_seen = set()
|
||||
for row in rows:
|
||||
dedup_key = (
|
||||
str(getattr(row["user"], "id", "")),
|
||||
str(getattr(row["person"], "id", "")),
|
||||
str(getattr(row["person_identifier"], "id", "")),
|
||||
str(row.get("service") or ""),
|
||||
str(row.get("source_kind") or ""),
|
||||
int(row.get("ts") or 0),
|
||||
str((row.get("payload") or {}).get("message_id") or ""),
|
||||
str((row.get("payload") or {}).get("inferred_from") or ""),
|
||||
)
|
||||
if dedup_key in dedup_seen:
|
||||
continue
|
||||
dedup_seen.add(dedup_key)
|
||||
|
||||
if not reset:
|
||||
exists = ContactAvailabilityEvent.objects.filter(
|
||||
user=row["user"],
|
||||
person=row["person"],
|
||||
person_identifier=row["person_identifier"],
|
||||
service=row["service"],
|
||||
source_kind=row["source_kind"],
|
||||
ts=row["ts"],
|
||||
).exists()
|
||||
if exists:
|
||||
continue
|
||||
|
||||
created += 1
|
||||
if dry_run:
|
||||
continue
|
||||
record_native_signal(
|
||||
AvailabilitySignal(
|
||||
user=row["user"],
|
||||
person=row["person"],
|
||||
person_identifier=row["person_identifier"],
|
||||
service=row["service"],
|
||||
source_kind=row["source_kind"],
|
||||
availability_state=row["availability_state"],
|
||||
confidence=float(row["confidence"]),
|
||||
ts=int(row["ts"]),
|
||||
payload=dict(row["payload"]),
|
||||
)
|
||||
)
|
||||
|
||||
self.stdout.write(
|
||||
self.style.SUCCESS(
|
||||
"recalculate_contact_availability complete "
|
||||
f"messages_scanned={len(messages)} candidates={len(rows)} "
|
||||
f"created={created} deleted_events={deleted_events} deleted_spans={deleted_spans} "
|
||||
f"reset={reset} dry_run={dry_run} days={days} limit={limit}"
|
||||
)
|
||||
)
|
||||
Reference in New Issue
Block a user