Senad Uka
fd45ff84dc
- Pipeline-down fallback: dialplan plays gogo-fallback after a failed AudioSocket (§15) + script to synthesize the Bosnian prompt via TTS - Voice heartbeat file → /health/voice endpoint + live status in admin panel - Retention job also purges stale call registrations; chat sessions metered; 100%-usage super-admin alert email (GOGO_ADMIN_ALERT_EMAIL) - .env.example, docs/DEPLOY.md (compose stack, GPU node, backups pg_dump, update procedure, security notes) - §16 Definition of Done as an automated test: softphone-style call → fake partner availability → request pushed → webhook confirm → console SMS - M0 PoC scripts: STT accuracy harness (CER/WER + spoken-digit check; dry-run verified with espeak-ng samples + faster-whisper small on CPU — phone number extracted exactly), TTS bake-off (Azure vs ElevenLabs, latency+cost), and end-to-end latency smoke test speaking real AudioSocket to the live pipeline Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
121 lines
4.3 KiB
Python
121 lines
4.3 KiB
Python
"""Background jobs: proposal expiry/reminders, partner polling fallback, retention.
|
|
|
|
APScheduler (in-process) — no extra infra needed for MVP scale (§4).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|
from sqlalchemy import select
|
|
|
|
from gogo.db import get_sessionmaker
|
|
from gogo.domain import BookingStatus, Slot
|
|
from gogo.models import BookingRequest, Tenant
|
|
from gogo.proposals import engine as proposal_engine
|
|
from gogo.scheduling.base import get_provider
|
|
|
|
log = logging.getLogger("gogo.jobs")
|
|
|
|
|
|
async def run_expiry_job() -> None:
|
|
async with get_sessionmaker()() as session:
|
|
expired = await proposal_engine.process_expirations(session)
|
|
await session.commit()
|
|
if expired:
|
|
log.info("expired %d proposals", expired)
|
|
|
|
|
|
async def run_partner_polling_job() -> None:
|
|
"""Polling fallback (§B.3) for partner tenants that cannot call webhooks."""
|
|
async with get_sessionmaker()() as session:
|
|
pending = (
|
|
(
|
|
await session.execute(
|
|
select(BookingRequest).where(
|
|
BookingRequest.status == BookingStatus.pending.value,
|
|
BookingRequest.partner_request_id.is_not(None),
|
|
)
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for req in pending:
|
|
tenant = (
|
|
await session.execute(select(Tenant).where(Tenant.id == req.tenant_id))
|
|
).scalar_one()
|
|
if tenant.scheduling_provider != "partner_api":
|
|
continue
|
|
provider = await get_provider(session, tenant)
|
|
if not provider.config.get("polling_fallback"):
|
|
continue
|
|
try:
|
|
data = await provider.poll_status(req.partner_request_id)
|
|
except Exception: # noqa: BLE001
|
|
log.exception("polling failed for request %s", req.id)
|
|
continue
|
|
status = data.get("status")
|
|
if status == "confirmed" and data.get("confirmed_slot"):
|
|
slot = Slot.model_validate(data["confirmed_slot"])
|
|
await proposal_engine.confirm_request(session, tenant, req, slot, recheck=False)
|
|
elif status == "resolved":
|
|
await proposal_engine.resolve_request(session, tenant, req)
|
|
elif status == "rejected":
|
|
await proposal_engine.reject_request(session, tenant, req)
|
|
await session.commit()
|
|
|
|
|
|
async def run_retention_job() -> None:
|
|
"""Delete call audio older than retention window; keep transcripts (§5.4)."""
|
|
import os
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from gogo.config import get_settings
|
|
from gogo.models import Call
|
|
|
|
cutoff = datetime.now(UTC) - timedelta(days=get_settings().audio_retention_days)
|
|
async with get_sessionmaker()() as session:
|
|
rows = (
|
|
(
|
|
await session.execute(
|
|
select(Call).where(
|
|
Call.recording_path.is_not(None), Call.started_at < cutoff
|
|
)
|
|
)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for call in rows:
|
|
path = call.recording_path
|
|
if path and os.path.exists(path):
|
|
try:
|
|
os.remove(path)
|
|
except OSError:
|
|
log.exception("failed to delete recording %s", path)
|
|
continue
|
|
call.recording_path = None
|
|
|
|
# drop stale dialplan handoff rows (they are only needed for minutes)
|
|
from sqlalchemy import delete
|
|
|
|
from gogo.models import CallRegistration
|
|
|
|
reg_cutoff = datetime.now(UTC) - timedelta(days=7)
|
|
await session.execute(
|
|
delete(CallRegistration).where(CallRegistration.created_at < reg_cutoff)
|
|
)
|
|
await session.commit()
|
|
if rows:
|
|
log.info("retention: cleared %d recordings", len(rows))
|
|
|
|
|
|
def build_scheduler() -> AsyncIOScheduler:
|
|
scheduler = AsyncIOScheduler()
|
|
scheduler.add_job(run_expiry_job, "interval", minutes=5, id="proposal_expiry")
|
|
scheduler.add_job(run_partner_polling_job, "interval", minutes=5, id="partner_polling")
|
|
scheduler.add_job(run_retention_job, "cron", hour=4, minute=0, id="retention")
|
|
return scheduler
|