87 lines
4.1 KiB
Python
87 lines
4.1 KiB
Python
"""Recover recent customer replies to proven proactive sends; never send here."""
|
|
from contextlib import closing
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import time
|
|
import json
|
|
|
|
|
|
def has_filtered_reply_recovery(root, message):
|
|
"""Explicit local repair tickets for an exact previously filtered event."""
|
|
path = Path(root) / 'protocol_inbound_recovery.json'
|
|
if not path.exists():
|
|
return False
|
|
data = json.loads(path.read_text(encoding='utf-8'))
|
|
return any(isinstance(row, dict) and row.get('reason') == 'assistant_recipient_after_proactive'
|
|
and all(row.get(k) == message.get(k) for k in ('account', 'conv_id', 'dedup_key'))
|
|
for row in data.get('requests', []))
|
|
|
|
|
|
def is_proactive_customer_turn(db, account, conversation, root):
|
|
"""A proven proactive question and its AI follow-ups override assistant role.
|
|
|
|
An assistant reminder (or an unrecognized outgoing message) ends that turn.
|
|
Receipt IDs, account and conversation determine the route, never wording.
|
|
"""
|
|
path = Path(root) / 'message_origins.sqlite3'
|
|
if not path.exists():
|
|
return False
|
|
ctx = db.get_conversation_context_by_id(account, conversation, limit=500)
|
|
if not ctx or ctx.get('account') != account or ctx.get('conv_id') != conversation:
|
|
return False
|
|
messages = ctx.get('messages') or []
|
|
with closing(sqlite3.connect(path.resolve().as_uri()+'?mode=ro', uri=True)) as conn:
|
|
for message in reversed(messages):
|
|
if message.get('is_self') is not True:
|
|
continue
|
|
if str(message.get('sender_id')) != account:
|
|
return False
|
|
server_id = str(message.get('server_id') or '')
|
|
if server_id in ('', '0'):
|
|
return False
|
|
rows = conn.execute("SELECT request,kind,created FROM origins WHERE account=? AND conversation=? "
|
|
"AND server_id=? AND status='confirmed'", (account, conversation, server_id)).fetchall()
|
|
if len(rows) != 1:
|
|
return False
|
|
request, kind, created = rows[0]
|
|
if kind == 'system' and request.startswith('proactive-contact:'):
|
|
return created >= time.time()-86400
|
|
if kind != 'ai':
|
|
return False
|
|
return False
|
|
|
|
|
|
def incoming_after_proactive(db, account, root, *, now=None, on_error=lambda exc: None):
|
|
now = time.time() if now is None else now
|
|
path = Path(root) / "message_origins.sqlite3"
|
|
if not path.exists():
|
|
return []
|
|
with closing(sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)) as conn:
|
|
targets = conn.execute("SELECT conversation,server_id,created FROM origins WHERE account=? "
|
|
"AND kind='system' AND request LIKE 'proactive-contact:%' AND status='confirmed' "
|
|
"AND server_id NOT IN ('','0') AND created>=? ORDER BY created DESC LIMIT 500",
|
|
(account, now - 86400)).fetchall()
|
|
incoming, visited = [], set()
|
|
for conversation, server_id, created in targets:
|
|
if conversation in visited:
|
|
continue
|
|
visited.add(conversation)
|
|
try:
|
|
ctx = db.get_conversation_context_by_id(account, conversation, limit=500)
|
|
if not ctx or ctx.get('account') != account or ctx.get('conv_id') != conversation:
|
|
continue
|
|
messages = ctx.get('messages') or []
|
|
anchors = [i for i, m in enumerate(messages) if str(m.get('server_id')) == server_id
|
|
and str(m.get('sender_id')) == account and m.get('is_self') is True]
|
|
if len(anchors) != 1 or anchors[0] >= len(messages) - 1:
|
|
continue
|
|
last = messages[-1]
|
|
if (last.get('account') == account and last.get('conv_id') == conversation
|
|
and last.get('is_self') is False and last.get('dedup_key')
|
|
and max(created - 2, now - 86400) <= float(last.get('send_time') or 0) <= now + 5
|
|
and float(last.get('send_time') or 0) >= float(messages[anchors[0]].get('send_time') or 0)):
|
|
incoming.append(last)
|
|
except Exception as exc:
|
|
on_error(exc)
|
|
return incoming
|