306 lines
13 KiB
Python
306 lines
13 KiB
Python
"""Durable account-scoped outbound deny list shared by desktop engines.
|
|
|
|
The server-side blacklist (management backend) is the source of truth; this
|
|
local database is its cache. Engines keep reading only this file, so a blocked
|
|
conversation stays blocked while the computer is offline.
|
|
|
|
Every row carries a ``sync_state``:
|
|
|
|
pending added on this computer and not yet confirmed by the server
|
|
(also every entry that existed before the server list did)
|
|
synced present in the last server snapshot; deleted locally once a later
|
|
snapshot no longer contains it (removed from the phone console)
|
|
rejected refused by the server (malformed identifiers); still blocks here,
|
|
no longer re-uploaded
|
|
|
|
The cache is bound to the logged-in account (``sync_meta.tenant``): each ZYT
|
|
account has its own server list.
|
|
"""
|
|
from contextlib import closing
|
|
from datetime import datetime
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from runtime_paths import application_data_dir
|
|
|
|
BLOCKED_REASON = '该会话已加入忽略，系统禁止发送消息'
|
|
_OPTIONAL_COLUMNS = ('name', 'account_name', 'created_at', 'reason')
|
|
|
|
|
|
def _path(root):
|
|
return (Path(root or application_data_dir()) / 'conversation_blacklist.sqlite3').resolve()
|
|
|
|
|
|
def entries(root=None):
|
|
path = _path(root)
|
|
if not path.exists():
|
|
return set()
|
|
with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=2)) as db:
|
|
return {(str(a), str(c)) for a, c in db.execute('SELECT account,conversation FROM blocked')}
|
|
|
|
|
|
def list_entries(root=None):
|
|
"""Read all accounts, including entries absent from the current message list."""
|
|
path = _path(root)
|
|
if not path.exists():
|
|
return []
|
|
with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=2)) as db:
|
|
db.row_factory = sqlite3.Row
|
|
columns = {row[1] for row in db.execute('PRAGMA table_info(blocked)')}
|
|
fields = ['account', 'conversation'] + [
|
|
name if name in columns else "'' AS " + name
|
|
for name in _OPTIONAL_COLUMNS
|
|
] + ['sync_state' if 'sync_state' in columns else "'pending' AS sync_state"]
|
|
return [dict(row) for row in db.execute(
|
|
'SELECT ' + ','.join(fields) + ' FROM blocked ORDER BY account,conversation')]
|
|
|
|
|
|
def _initialize(db):
|
|
db.execute('CREATE TABLE IF NOT EXISTS blocked (account TEXT NOT NULL, conversation TEXT NOT NULL, PRIMARY KEY(account,conversation))')
|
|
columns = {row[1] for row in db.execute('PRAGMA table_info(blocked)')}
|
|
for name in _OPTIONAL_COLUMNS:
|
|
if name not in columns:
|
|
db.execute("ALTER TABLE blocked ADD COLUMN " + name + " TEXT NOT NULL DEFAULT ''")
|
|
if 'sync_state' not in columns:
|
|
# Entries written by older versions have never reached the server.
|
|
db.execute("ALTER TABLE blocked ADD COLUMN sync_state TEXT NOT NULL DEFAULT 'pending'")
|
|
db.execute('CREATE TABLE IF NOT EXISTS sync_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)')
|
|
|
|
|
|
def _connect(root):
|
|
path = _path(root)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
return closing(sqlite3.connect(path, timeout=5))
|
|
|
|
|
|
def _clean(value):
|
|
return str(value or '').strip()
|
|
|
|
|
|
def set_blocked(account, conversation, blocked, *, root=None, name='', account_name='', reason=''):
|
|
account, conversation = _clean(account), _clean(conversation)
|
|
if not account or not conversation or type(blocked) is not bool:
|
|
raise ValueError('请选择有效账号和会话')
|
|
with _connect(root) as db, db:
|
|
_initialize(db)
|
|
if blocked:
|
|
db.execute("""INSERT INTO blocked (account,conversation,name,account_name,created_at,reason,sync_state)
|
|
VALUES (?,?,?,?,?,?,'pending') ON CONFLICT(account,conversation) DO UPDATE SET
|
|
name=CASE WHEN excluded.name<>'' THEN excluded.name ELSE blocked.name END,
|
|
account_name=CASE WHEN excluded.account_name<>'' THEN excluded.account_name ELSE blocked.account_name END,
|
|
reason=CASE WHEN excluded.reason<>'' THEN excluded.reason ELSE blocked.reason END""",
|
|
(account, conversation, str(name or ''), str(account_name or ''),
|
|
datetime.now().strftime('%Y-%m-%d %H:%M:%S'), str(reason or '')[:200]))
|
|
else:
|
|
db.execute('DELETE FROM blocked WHERE account=? AND conversation=?', (account, conversation))
|
|
|
|
|
|
def remove_entries(items, *, root=None, only_pending=False):
|
|
"""Validate the whole selection, then remove exact account/conversation pairs atomically.
|
|
|
|
``only_pending`` removes a row only while it has not reached the server: a sync
|
|
running in parallel may have just uploaded it, and then only the server may
|
|
remove it.
|
|
"""
|
|
pairs = _validated_pairs(items)
|
|
path = _path(root)
|
|
if not path.exists():
|
|
return 0
|
|
with closing(sqlite3.connect(path, timeout=5)) as db, db:
|
|
if only_pending:
|
|
_initialize(db)
|
|
before = db.total_changes
|
|
sql = 'DELETE FROM blocked WHERE account=? AND conversation=?'
|
|
if only_pending:
|
|
sql += " AND sync_state<>'synced'"
|
|
db.executemany(sql, sorted(pairs))
|
|
return db.total_changes - before
|
|
|
|
|
|
def _validated_pairs(items):
|
|
if not isinstance(items, list) or not items:
|
|
raise ValueError('请先勾选需要移除的会话')
|
|
pairs = set()
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
raise ValueError('忽略选择无效，请刷新后重试')
|
|
account, conversation = item.get('account'), item.get('conversation')
|
|
if not isinstance(account, str) or not account.strip() or not isinstance(conversation, str) or not conversation.strip():
|
|
raise ValueError('忽略选择无效，请刷新后重试')
|
|
pairs.add((account.strip(), conversation.strip()))
|
|
return pairs
|
|
|
|
|
|
def local_states(root=None):
|
|
"""{(account, conversation): 'pending' | 'synced'} for every local entry."""
|
|
path = _path(root)
|
|
if not path.exists():
|
|
return {}
|
|
with _connect(root) as db, db:
|
|
_initialize(db)
|
|
return {(str(a), str(c)): str(s or 'pending')
|
|
for a, c, s in db.execute('SELECT account,conversation,sync_state FROM blocked')}
|
|
|
|
|
|
def pending_entries(root=None):
|
|
"""Entries that still have to be uploaded to the server."""
|
|
path = _path(root)
|
|
if not path.exists():
|
|
return []
|
|
with _connect(root) as db, db:
|
|
_initialize(db)
|
|
db.row_factory = sqlite3.Row
|
|
return [
|
|
{'account': row['account'], 'conversation': row['conversation'], 'name': row['name'],
|
|
'account_name': row['account_name'], 'reason': row['reason']}
|
|
for row in db.execute(
|
|
"SELECT * FROM blocked WHERE sync_state='pending' ORDER BY account,conversation LIMIT 500")
|
|
]
|
|
|
|
|
|
def mark_rejected(rows, *, root=None):
|
|
"""The server refused these (malformed identifiers): keep blocking, stop re-uploading."""
|
|
pairs = [(_clean(row.get('account')), _clean(row.get('conversation')))
|
|
for row in rows or [] if isinstance(row, dict)]
|
|
pairs = [pair for pair in pairs if pair[0] and pair[1]]
|
|
if not pairs:
|
|
return 0
|
|
with _connect(root) as db, db:
|
|
_initialize(db)
|
|
before = db.total_changes
|
|
db.executemany("UPDATE blocked SET sync_state='rejected' WHERE account=? AND conversation=? "
|
|
"AND sync_state='pending'", pairs)
|
|
return db.total_changes - before
|
|
|
|
|
|
def pending_count(root=None):
|
|
path = _path(root)
|
|
if not path.exists():
|
|
return 0
|
|
try:
|
|
with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=2)) as db:
|
|
columns = {row[1] for row in db.execute('PRAGMA table_info(blocked)')}
|
|
if 'sync_state' not in columns:
|
|
return int(db.execute('SELECT COUNT(*) FROM blocked').fetchone()[0])
|
|
return int(db.execute("SELECT COUNT(*) FROM blocked WHERE sync_state='pending'").fetchone()[0])
|
|
except sqlite3.Error:
|
|
return 0
|
|
|
|
|
|
def _meta(root):
|
|
path = _path(root)
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=2)) as db:
|
|
return {str(k): str(v) for k, v in db.execute('SELECT key,value FROM sync_meta')}
|
|
except sqlite3.Error:
|
|
return {}
|
|
|
|
|
|
def synced_version(root=None, tenant=None):
|
|
"""Server version of the cached list; None when it belongs to another account."""
|
|
meta = _meta(root)
|
|
if tenant is not None and meta.get('tenant', '') != str(tenant):
|
|
return None
|
|
try:
|
|
return int(meta['server_version'])
|
|
except (KeyError, TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def adopt_tenant(tenant, *, root=None):
|
|
"""Bind the cache to the logged-in account (its server list is per ZYT account).
|
|
|
|
When a different account logs in on this computer, rows synced for the previous
|
|
account become ``pending`` instead of being dropped: they keep blocking and are
|
|
uploaded to the new account, where a supervisor can review them. Returns True
|
|
when the binding changed.
|
|
"""
|
|
tenant = _clean(tenant)
|
|
if not tenant:
|
|
return False
|
|
current = _meta(root).get('tenant', '')
|
|
if current == tenant:
|
|
return False
|
|
with _connect(root) as db, db:
|
|
_initialize(db)
|
|
stored = db.execute("SELECT value FROM sync_meta WHERE key='tenant'").fetchone()
|
|
if stored and str(stored[0]) == tenant:
|
|
return False
|
|
if stored and str(stored[0]):
|
|
db.execute("UPDATE blocked SET sync_state='pending' WHERE sync_state='synced'")
|
|
db.execute("DELETE FROM sync_meta WHERE key='server_version'")
|
|
db.execute("INSERT INTO sync_meta(key,value) VALUES('tenant',?) "
|
|
"ON CONFLICT(key) DO UPDATE SET value=excluded.value", (tenant,))
|
|
return True
|
|
|
|
|
|
def _local_time(value):
|
|
text = _clean(value)
|
|
try:
|
|
return datetime.fromisoformat(text.replace('Z', '+00:00')).astimezone().strftime('%Y-%m-%d %H:%M:%S')
|
|
except ValueError:
|
|
return text or datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
|
|
def apply_server_snapshot(items, version, *, root=None, tenant=None):
|
|
"""Replace the synced part of the cache with the server's full list.
|
|
|
|
Rows still ``pending`` are kept even when the snapshot lacks them: they were
|
|
added here after the request left (or the upload is still to come), and
|
|
blocking is the safe side. Returns counts for logging.
|
|
"""
|
|
server = {}
|
|
for item in items or []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
account, conversation = _clean(item.get('account')), _clean(item.get('conversation'))
|
|
if account and conversation:
|
|
server[(account, conversation)] = item
|
|
added = removed = 0
|
|
with _connect(root) as db, db:
|
|
_initialize(db)
|
|
if tenant is not None:
|
|
stored = db.execute("SELECT value FROM sync_meta WHERE key='tenant'").fetchone()
|
|
if stored and str(stored[0]) and str(stored[0]) != str(tenant):
|
|
# The login changed while this snapshot was in flight; adopt_tenant()
|
|
# re-binds first, so a stale snapshot must not touch the new binding.
|
|
return {'added': 0, 'removed': 0, 'total': 0, 'stale': True}
|
|
local = {(str(a), str(c)): str(s or 'pending')
|
|
for a, c, s in db.execute('SELECT account,conversation,sync_state FROM blocked')}
|
|
for key, state in local.items():
|
|
if key not in server and state == 'synced':
|
|
db.execute('DELETE FROM blocked WHERE account=? AND conversation=?', key)
|
|
removed += 1
|
|
for key, item in server.items():
|
|
if key not in local:
|
|
added += 1
|
|
db.execute("""INSERT INTO blocked (account,conversation,name,account_name,created_at,reason,sync_state)
|
|
VALUES (?,?,?,?,?,?,'synced') ON CONFLICT(account,conversation) DO UPDATE SET
|
|
name=CASE WHEN excluded.name<>'' THEN excluded.name ELSE blocked.name END,
|
|
account_name=CASE WHEN excluded.account_name<>'' THEN excluded.account_name ELSE blocked.account_name END,
|
|
reason=excluded.reason,sync_state='synced'""",
|
|
(key[0], key[1], _clean(item.get('name')), _clean(item.get('account_name')),
|
|
_local_time(item.get('created_at')), _clean(item.get('reason'))[:200]))
|
|
if version is not None:
|
|
db.execute("INSERT INTO sync_meta(key,value) VALUES('server_version',?) "
|
|
"ON CONFLICT(key) DO UPDATE SET value=excluded.value", (str(int(version)),))
|
|
if tenant is not None:
|
|
db.execute("INSERT INTO sync_meta(key,value) VALUES('tenant',?) "
|
|
"ON CONFLICT(key) DO UPDATE SET value=excluded.value", (str(tenant),))
|
|
return {'added': added, 'removed': removed, 'total': len(server)}
|
|
|
|
|
|
def blocked_reason(account, conversation, *, root=None):
|
|
try:
|
|
return BLOCKED_REASON if (str(account or ''), str(conversation or '')) in entries(root) else ''
|
|
except (OSError, sqlite3.Error):
|
|
return '忽略暂不可读取，已暂停系统发送，请重试'
|
|
|
|
|
|
def state_reason(state):
|
|
state = state if isinstance(state, dict) else {}
|
|
event = state.get('database_event') or {}
|
|
return blocked_reason(state.get('account') or event.get('account'),
|
|
state.get('conv_id') or event.get('conv_id'))
|