99 lines
6.0 KiB
Python
99 lines
6.0 KiB
Python
"""Local send provenance: exact receipts win; absent legacy evidence stays unknown."""
|
|
from contextlib import closing, contextmanager
|
|
import hashlib
|
|
import math
|
|
import sqlite3
|
|
import time
|
|
from pathlib import Path
|
|
from runtime_paths import application_data_dir
|
|
|
|
KINDS = frozenset({'ai', 'human', 'system', 'unknown', 'customer'})
|
|
LABELS = {'ai': 'AI 发送', 'human': '人工发送', 'system': '系统发送', 'unknown': '来源未知', 'customer': '客户发送'}
|
|
|
|
@contextmanager
|
|
def connect(root=None):
|
|
path = Path(root or application_data_dir()) / 'message_origins.sqlite3'
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with closing(sqlite3.connect(path, timeout=3)) as db, db:
|
|
db.execute('CREATE TABLE IF NOT EXISTS coverage (account TEXT PRIMARY KEY, started REAL NOT NULL)')
|
|
db.execute('CREATE TABLE IF NOT EXISTS origins (request TEXT PRIMARY KEY, account TEXT NOT NULL, conversation TEXT NOT NULL, digest TEXT NOT NULL, kind TEXT NOT NULL, created REAL NOT NULL, server_id TEXT NOT NULL DEFAULT "", status TEXT NOT NULL)')
|
|
db.execute('CREATE INDEX IF NOT EXISTS origin_receipt ON origins(account,conversation,server_id)')
|
|
db.execute('CREATE INDEX IF NOT EXISTS origin_attempt ON origins(account,conversation,digest,created)')
|
|
yield db
|
|
|
|
def digest(text):
|
|
return hashlib.sha256(str(text or '').strip().encode('utf-8')).hexdigest()
|
|
|
|
def record_attempt(request, account, conversation, text, kind='ai', *, root=None):
|
|
if not request or not account or not conversation or kind not in {'ai', 'human', 'system'}:
|
|
raise ValueError('发送来源缺少身份信息')
|
|
with connect(root) as db:
|
|
db.execute('INSERT OR IGNORE INTO coverage VALUES (?,?)', (str(account), time.time()))
|
|
db.execute("""INSERT INTO origins(request,account,conversation,digest,kind,created,status) VALUES (?,?,?,?,?,?,?)
|
|
ON CONFLICT(request) DO UPDATE SET account=excluded.account,conversation=excluded.conversation,
|
|
digest=excluded.digest,kind=excluded.kind,created=excluded.created,status='attempted',server_id=''
|
|
WHERE origins.status='cancelled'""",
|
|
(str(request), str(account), str(conversation), digest(text), kind, time.time(), 'attempted'))
|
|
|
|
def finish(request, status, server_id='', *, root=None):
|
|
with connect(root) as db:
|
|
db.execute("UPDATE origins SET status=?,server_id=? WHERE request=? AND NOT(status='confirmed' AND ?<>'confirmed')",
|
|
(status, str(server_id or ''), str(request), status))
|
|
|
|
def classify(account, conversation, sender, server_id, text, sent_at, *, root=None):
|
|
account, conversation, sender = str(account or ''), str(conversation or ''), str(sender or '')
|
|
if not account or not sender:
|
|
return 'unknown'
|
|
if sender != account:
|
|
return 'customer'
|
|
try:
|
|
sent_at = float(sent_at or 0)
|
|
if not math.isfinite(sent_at):
|
|
return 'unknown'
|
|
while sent_at > 100_000_000_000:
|
|
sent_at /= 1000
|
|
with connect(root) as db:
|
|
db.execute('INSERT OR IGNORE INTO coverage VALUES (?,?)', (account, time.time()))
|
|
if str(server_id or '') not in {'', '0', 'None'}:
|
|
exact = db.execute('SELECT kind FROM origins WHERE account=? AND conversation=? AND server_id=? AND status="confirmed"',
|
|
(account, conversation, str(server_id))).fetchone()
|
|
if exact:
|
|
return exact[0]
|
|
# A visual receipt proves sending but may not have a DB server ID yet.
|
|
# Do not guess from equal wording alone, including uncertain native sends.
|
|
candidates = db.execute("SELECT request,kind,status,server_id FROM origins WHERE account IN (?, '*') AND conversation IN (?, '*') AND status NOT IN ('cancelled','failed') AND digest=? AND created BETWEEN ? AND ?",
|
|
(account, conversation, digest(text), sent_at - 120, sent_at + 2)).fetchall()
|
|
if candidates:
|
|
return 'unknown'
|
|
started = db.execute('SELECT started FROM coverage WHERE account=?', (account,)).fetchone()[0]
|
|
return 'human' if sent_at > started and sent_at < time.time() - 15 else 'unknown'
|
|
except (OSError, ValueError, sqlite3.Error):
|
|
return 'unknown'
|
|
|
|
|
|
def reconcile_visual_receipts(connection, account, *, root=None):
|
|
"""Resolve a proven visual send once the decrypted DB catches up.
|
|
|
|
Only a unique matching outgoing DB row may provide an exact server receipt.
|
|
Equal wording with multiple candidates remains unknown.
|
|
"""
|
|
from archive_content_parser import parse_message_content
|
|
columns = {row[1] for row in connection.execute('PRAGMA table_info(message_table)')}
|
|
if not {'sender_id','conversation_id','send_time','server_id','content','content_type'}.issubset(columns):
|
|
return
|
|
with connect(root) as db:
|
|
db.execute('INSERT OR IGNORE INTO coverage VALUES (?,?)', (str(account), time.time()))
|
|
attempts = db.execute("SELECT request,conversation,digest,created FROM origins WHERE account=? AND status='visual_confirmed' AND server_id='' ORDER BY created DESC LIMIT 1000", (str(account),)).fetchall()
|
|
for request, conversation, expected, created in attempts:
|
|
query = connection.execute("""SELECT server_id,content,content_type FROM message_table
|
|
WHERE CAST(sender_id AS TEXT)=? AND conversation_id=? AND send_time BETWEEN ? AND ? LIMIT 102""",
|
|
(str(account), conversation, created - 2, created + 120))
|
|
rows = query.fetchall()
|
|
if len(rows) > 100:
|
|
continue
|
|
matches = {str(sid) for sid, content, content_type in rows
|
|
if str(sid or '') not in {'', '0', 'None'}
|
|
and digest(parse_message_content(content, content_type).get('text')) == expected}
|
|
if len(matches) == 1:
|
|
db.execute("UPDATE origins SET status='confirmed',server_id=? WHERE request=?", (matches.pop(), request))
|