import aiosqlite from .phone import normalize from datetime import datetime, timezone, timedelta def now_iso(): return datetime.now(timezone.utc).isoformat() class Repository: def __init__(self, path): self.path = path async def init(self): async with aiosqlite.connect(self.path) as db: await db.executescript(""" CREATE TABLE IF NOT EXISTS calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, callid INTEGER NOT NULL, legid INTEGER, phone TEXT, queue_dn TEXT NOT NULL, first_seen_at TEXT NOT NULL, last_seen_at TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active', answered_by_dn TEXT, UNIQUE(queue_dn, callid) ); CREATE TABLE IF NOT EXISTS missed_call_groups ( phone TEXT PRIMARY KEY, first_call_at TEXT NOT NULL, last_call_at TEXT NOT NULL, attempt_count INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'open', customer_id TEXT, customer_label TEXT, resolved_until TEXT, resolved_by TEXT, resolved_at TEXT, resolved_note TEXT ); CREATE TABLE IF NOT EXISTS idempotency_keys ( idem_key TEXT PRIMARY KEY, operation TEXT NOT NULL, response_json TEXT NOT NULL, created_at TEXT NOT NULL ); """) await db.commit() def normalize_phone(self, raw): """ Zentrale Telefonnummern-Normalisierung. Die eigentliche Logik liegt in app.phone.normalize(). Dadurch verwendet die Repository-/Queue-Verarbeitung dieselbe Normalisierung wie die /phone/normalize API. """ result = normalize(raw, "DE") if not result.get("valid"): return None return result.get("e164") async def observe_queue(self, participants, queue_dn, agent_dns=None): active = set() async with aiosqlite.connect(self.path) as db: for p in participants: if p.get("callid") is None: continue callid = int(p["callid"]) active.add(callid) raw = p.get("party_caller_id") is_external = p.get("party_dn_type") == "Wexternalline" e164 = self.normalize_phone(raw) if is_external else None now = now_iso() status_raw = str(p.get("status") or "").lower() participant_dn = str( p.get("partyDn") or p.get("party_dn") or "" ).strip() agent_dns_set = { str(x).strip() for x in (agent_dns or []) if str(x).strip() } if ( status_raw in ("connected", "talking", "answered") and participant_dn in agent_dns_set ): current_status = "answered" else: current_status = "active" await db.execute(""" INSERT INTO calls (callid, legid, phone, raw, queue_dn, first_seen_at, last_seen_at, status, direction, started_at, answered_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(queue_dn, callid) DO UPDATE SET legid=excluded.legid, phone=excluded.phone, raw=excluded.raw, last_seen_at=excluded.last_seen_at, status=excluded.status, answered_at=CASE WHEN calls.answered_at IS NOT NULL THEN calls.answered_at WHEN excluded.status='answered' THEN excluded.last_seen_at ELSE NULL END """, ( callid, p.get("legid"), e164, raw, queue_dn, now, now, current_status, "inbound" if is_external else "internal", now, now if current_status == "answered" else None, )) await db.commit() return active async def finalize_disappeared(self, queue_dn, active): async with aiosqlite.connect(self.path) as db: cur = await db.execute(""" SELECT id, callid, phone, raw, last_seen_at FROM calls WHERE queue_dn=? AND status IN ('active', 'answered') """, (queue_dn,)) for row_id, callid, phone, raw, last_seen_at in await cur.fetchall(): if callid in active: continue last_seen = datetime.fromisoformat(last_seen_at) if datetime.now(timezone.utc) - last_seen < timedelta(seconds=10): continue now = now_iso() cur_state = await db.execute( "SELECT answered_at, started_at FROM calls WHERE id=?", (row_id,), ) state_row = await cur_state.fetchone() answered_at = state_row[0] if state_row else None started_at = state_row[1] if state_row else None final_status = "ended" if answered_at else "missed" duration_seconds = None if started_at: started_dt = datetime.fromisoformat(started_at) ended_dt = datetime.fromisoformat(now) duration_seconds = max( 0, int((ended_dt - started_dt).total_seconds()) ) await db.execute( """ UPDATE calls SET status=?, ended_at=?, last_seen_at=?, duration_seconds=? WHERE id=? """, ( final_status, now, now, duration_seconds, row_id, ), ) if final_status == "missed" and phone: await db.execute(""" INSERT INTO missed_call_groups (phone, raw, first_call_at, last_call_at, attempt_count, status) VALUES (?, ?, ?, ?, 1, 'open') ON CONFLICT(phone) DO UPDATE SET raw=excluded.raw, last_call_at=excluded.last_call_at, attempt_count=missed_call_groups.attempt_count+1 """, (phone, raw, last_seen_at, now)) await db.commit() async def get_outbound_idempotency(self, key): async with aiosqlite.connect(self.path) as db: db.row_factory = aiosqlite.Row cur = await db.execute(""" SELECT idempotency_key, request_hash, status, response_json, created_at FROM outbound_idempotency WHERE idempotency_key=? """, (key,)) row = await cur.fetchone() return dict(row) if row else None async def claim_outbound_idempotency(self, key, request_hash, created_at): async with aiosqlite.connect(self.path) as db: try: await db.execute("BEGIN IMMEDIATE") cur = await db.execute(""" INSERT OR IGNORE INTO outbound_idempotency (idempotency_key, request_hash, status, response_json, created_at) VALUES (?, ?, 'processing', NULL, ?) """, (key, request_hash, created_at)) inserted = cur.rowcount == 1 await db.commit() return inserted except Exception: await db.rollback() raise async def finish_outbound_idempotency(self, key, response_json): async with aiosqlite.connect(self.path) as db: await db.execute(""" UPDATE outbound_idempotency SET status='completed', response_json=? WHERE idempotency_key=? """, (response_json, key)) await db.commit() async def release_outbound_idempotency(self, key): async with aiosqlite.connect(self.path) as db: await db.execute(""" DELETE FROM outbound_idempotency WHERE idempotency_key=? AND status='processing' """, (key,)) await db.commit() async def insert_outbound_call( self, callid, legid, source_dn, destination, phone, raw, started_at, ): async with aiosqlite.connect(self.path) as db: await db.execute(""" INSERT INTO calls ( callid, legid, phone, queue_dn, first_seen_at, last_seen_at, status, raw, direction, started_at, source_dn, destination ) VALUES (?, ?, ?, NULL, ?, ?, 'dialing', ?, 'outbound', ?, ?, ?) """, ( callid, legid, phone, started_at, started_at, raw, started_at, source_dn, destination, )) await db.commit() async def list_active_outbound_calls(self): async with aiosqlite.connect(self.path) as db: db.row_factory = aiosqlite.Row cur = await db.execute(""" SELECT id, callid, legid, source_dn, destination, status, started_at, answered_at, ended_at, duration_seconds FROM calls WHERE direction='outbound' AND ended_at IS NULL ORDER BY id """) return [dict(row) for row in await cur.fetchall()] async def update_outbound_connected( self, call_id, answered_at, last_seen_at, ): async with aiosqlite.connect(self.path) as db: await db.execute(""" UPDATE calls SET status='connected', answered_at=COALESCE(answered_at, ?), last_seen_at=? WHERE id=? AND direction='outbound' AND ended_at IS NULL """, ( answered_at, last_seen_at, call_id, )) await db.commit() async def update_outbound_seen( self, call_id, status, last_seen_at, ): async with aiosqlite.connect(self.path) as db: await db.execute(""" UPDATE calls SET status=?, last_seen_at=? WHERE id=? AND direction='outbound' AND ended_at IS NULL """, ( status, last_seen_at, call_id, )) await db.commit() async def finalize_outbound_call( self, call_id, ended_at, status, duration_seconds, ): async with aiosqlite.connect(self.path) as db: await db.execute(""" UPDATE calls SET status=?, ended_at=?, duration_seconds=?, last_seen_at=? WHERE id=? AND direction='outbound' AND ended_at IS NULL """, ( status, ended_at, duration_seconds, ended_at, call_id, )) await db.commit() async def get_call(self, call_id): async with aiosqlite.connect(self.path) as db: db.row_factory = aiosqlite.Row cur = await db.execute(""" SELECT id, callid, legid, phone, raw, queue_dn, status, direction, first_seen_at, last_seen_at, started_at, answered_at, ended_at, duration_seconds, agent_dn, agent_name FROM calls WHERE id=? LIMIT 1 """, (call_id,)) row = await cur.fetchone() return dict(row) if row else None async def list_calls(self, limit=100, offset=0, status=None): async with aiosqlite.connect(self.path) as db: db.row_factory = aiosqlite.Row where = "" params = [] if status: where = "WHERE status=?" params.append(status) cur = await db.execute( f""" SELECT id, callid, legid, phone, raw, queue_dn, status, direction, first_seen_at, last_seen_at, started_at, answered_at, ended_at, duration_seconds, agent_dn, agent_name FROM calls {where} ORDER BY COALESCE(started_at, first_seen_at) DESC LIMIT ? OFFSET ? """, (*params, limit, offset), ) rows = [dict(row) for row in await cur.fetchall()] cur = await db.execute( f"SELECT COUNT(*) FROM calls {where}", tuple(params), ) total = (await cur.fetchone())[0] return rows, total def _decorate(self, row): last_call = row["last_call_at"] resolved_until = row["resolved_until"] row["e164"] = row["phone"] row["status"] = ( "resolved" if resolved_until and resolved_until >= last_call else "open" ) return row async def list_groups(self, open_only=False): async with aiosqlite.connect(self.path) as db: db.row_factory = aiosqlite.Row cur = await db.execute( "SELECT * FROM missed_call_groups ORDER BY last_call_at DESC" ) rows = [dict(x) for x in await cur.fetchall()] rows = [self._decorate(x) for x in rows] if open_only: rows = [x for x in rows if x["status"] == "open"] return rows async def get_group(self, e164): async with aiosqlite.connect(self.path) as db: db.row_factory = aiosqlite.Row cur = await db.execute( "SELECT * FROM missed_call_groups WHERE phone=?", (e164,), ) row = await cur.fetchone() return self._decorate(dict(row)) if row else None async def get_idempotency(self, key, operation): async with aiosqlite.connect(self.path) as db: cur = await db.execute( """SELECT response_json FROM idempotency_keys WHERE idem_key=? AND operation=?""", (key, operation), ) row = await cur.fetchone() return row[0] if row else None async def save_idempotency(self, key, operation, response_json): async with aiosqlite.connect(self.path) as db: await db.execute(""" INSERT OR IGNORE INTO idempotency_keys (idem_key, operation, response_json, created_at) VALUES (?, ?, ?, ?) """, (key, operation, response_json, now_iso())) await db.commit() async def resolve(self, e164, until, by, note): async with aiosqlite.connect(self.path) as db: await db.execute(""" UPDATE missed_call_groups SET resolved_until=?, resolved_by=?, resolved_at=?, resolved_note=? WHERE phone=? """, (until, by, now_iso(), note, e164)) await db.commit() return await self.get_group(e164) async def reopen(self, e164, by): async with aiosqlite.connect(self.path) as db: await db.execute(""" UPDATE missed_call_groups SET resolved_until=NULL, resolved_by=?, resolved_at=?, resolved_note=NULL WHERE phone=? """, (by, now_iso(), e164)) await db.commit() return await self.get_group(e164)