215 lines
12 KiB
Python
215 lines
12 KiB
Python
"""Durable, account-scoped proactive tasks. No network or messaging on import."""
|
|
from contextlib import contextmanager
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
|
|
|
|
def encode(value):
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True)
|
|
|
|
|
|
class ContactStore:
|
|
def __init__(self, root):
|
|
self.path = Path(root) / "proactive_contacts.sqlite3"
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
with self.connect() as db:
|
|
db.executescript("""
|
|
CREATE TABLE IF NOT EXISTS customers (
|
|
id TEXT PRIMARY KEY, owner TEXT NOT NULL, account TEXT NOT NULL,
|
|
peer TEXT NOT NULL, data TEXT NOT NULL, UNIQUE(owner,account,peer));
|
|
CREATE TABLE IF NOT EXISTS campaigns (
|
|
id TEXT PRIMARY KEY, owner TEXT NOT NULL, created REAL NOT NULL, data TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS tasks (
|
|
id TEXT PRIMARY KEY, owner TEXT NOT NULL, campaign TEXT NOT NULL,
|
|
customer TEXT NOT NULL, account TEXT NOT NULL, status TEXT NOT NULL,
|
|
due REAL NOT NULL, created REAL NOT NULL, updated REAL NOT NULL, data TEXT NOT NULL);
|
|
CREATE INDEX IF NOT EXISTS contact_due ON tasks(owner,status,due);
|
|
CREATE TABLE IF NOT EXISTS attempts (
|
|
id TEXT PRIMARY KEY, task TEXT NOT NULL, account TEXT NOT NULL,
|
|
started REAL NOT NULL, finished REAL, status TEXT NOT NULL, detail TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS settings (owner TEXT PRIMARY KEY, enabled INTEGER NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS deleted_tasks (
|
|
task TEXT PRIMARY KEY, owner TEXT NOT NULL, deleted REAL NOT NULL, previous_status TEXT NOT NULL);
|
|
CREATE TABLE IF NOT EXISTS reply_activations (
|
|
task TEXT PRIMARY KEY, requested REAL NOT NULL);
|
|
""")
|
|
|
|
@contextmanager
|
|
def connect(self):
|
|
db = sqlite3.connect(self.path, timeout=10)
|
|
db.row_factory = sqlite3.Row
|
|
db.execute("PRAGMA synchronous=FULL")
|
|
try:
|
|
with db:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
@staticmethod
|
|
def task_row(row):
|
|
return {**json.loads(row["data"]), **{k: row[k] for k in
|
|
("id", "owner", "campaign", "customer", "account", "status", "due", "created", "updated")}}
|
|
|
|
def customers(self, owner):
|
|
with self.connect() as db:
|
|
return [dict(json.loads(r["data"]), id=r["id"]) for r in db.execute(
|
|
"SELECT * FROM customers WHERE owner=? ORDER BY rowid DESC", (owner,))]
|
|
|
|
def save_customer(self, owner, data):
|
|
with self.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
row = db.execute("SELECT id FROM customers WHERE owner=? AND account=? AND peer=?",
|
|
(owner, data["account"], data["peer"])).fetchone()
|
|
key = row[0] if row else uuid.uuid4().hex
|
|
db.execute("INSERT INTO customers VALUES(?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET data=excluded.data",
|
|
(key, owner, data["account"], data["peer"], encode(data)))
|
|
return key
|
|
|
|
def tasks(self, owner, *, pending=False, limit=500, include_deleted=False):
|
|
with self.connect() as db:
|
|
extra = " AND status IN ('scheduled','queued','sending','paused')" if pending else ""
|
|
if not include_deleted:
|
|
extra += " AND NOT EXISTS (SELECT 1 FROM deleted_tasks d WHERE d.task=tasks.id)"
|
|
return [self.task_row(r) for r in db.execute(
|
|
"SELECT * FROM tasks WHERE owner=?" + extra + " ORDER BY created DESC,rowid DESC LIMIT ?",
|
|
(owner, limit))]
|
|
|
|
def get_task(self, owner, key, *, include_deleted=False):
|
|
with self.connect() as db:
|
|
extra = "" if include_deleted else " AND NOT EXISTS (SELECT 1 FROM deleted_tasks d WHERE d.task=tasks.id)"
|
|
row = db.execute("SELECT * FROM tasks WHERE owner=? AND id=?" + extra, (owner, key)).fetchone()
|
|
return self.task_row(row) if row else None
|
|
|
|
def delete_tasks(self, owner, ids):
|
|
"""Remove from lists, stop unsent work, retain receipts/dedup/pacing evidence."""
|
|
deleted, errors = [], []
|
|
with self.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
for key in dict.fromkeys(ids):
|
|
row = db.execute("SELECT * FROM tasks WHERE owner=? AND id=?", (owner, key)).fetchone()
|
|
if not row:
|
|
errors.append("任务不存在或不属于当前账号")
|
|
continue
|
|
if row["status"] == "sending":
|
|
errors.append(f"{json.loads(row['data']).get('name', '客户')}：正在发送，请待结果返回后删除")
|
|
continue
|
|
if db.execute("SELECT 1 FROM deleted_tasks WHERE task=?", (key,)).fetchone():
|
|
continue
|
|
now = time.time()
|
|
db.execute("INSERT INTO deleted_tasks VALUES(?,?,?,?)", (key, owner, now, row["status"]))
|
|
if row["status"] in {"scheduled", "queued", "paused"}:
|
|
data = json.loads(row["data"])
|
|
data["error"] = "用户删除任务，已停止发送"
|
|
db.execute("UPDATE tasks SET status='cancelled',updated=?,data=? WHERE id=?", (now, encode(data), key))
|
|
deleted.append(key)
|
|
return deleted, errors
|
|
|
|
def create_campaign(self, owner, preview, now):
|
|
"""Confirmation is idempotent; dedup spans different campaigns for 24h."""
|
|
with self.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
if db.execute("SELECT 1 FROM campaigns WHERE owner=? AND id=?", (owner, preview["id"])).fetchone():
|
|
return 0
|
|
for task in preview["tasks"]:
|
|
existing = db.execute("SELECT data FROM tasks WHERE owner=? AND customer=? AND "
|
|
"(status IN ('scheduled','queued','sending','paused','delivery_unknown') OR "
|
|
"(status='succeeded' AND updated>?))", (owner, task["customer"], now - 86400))
|
|
if any(json.loads(r[0])["text"] == task["text"] for r in existing):
|
|
raise ValueError(f"{task['name']} 已有相同内容任务或 24 小时内已发送，请核对记录")
|
|
db.execute("INSERT INTO campaigns VALUES(?,?,?,?)", (preview["id"], owner, now,
|
|
encode({k: v for k, v in preview.items() if k != "tasks"})))
|
|
for task in preview["tasks"]:
|
|
db.execute("INSERT INTO tasks VALUES(?,?,?,?,?,?,?,?,?,?)",
|
|
(uuid.uuid4().hex, owner, preview["id"], task["customer"], task["account"],
|
|
"scheduled", task["due"], now, now, encode(task)))
|
|
return len(preview["tasks"])
|
|
|
|
def update(self, owner, key, status, *, expected=None, now=None, **changes):
|
|
with self.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
row = db.execute("SELECT * FROM tasks WHERE owner=? AND id=? AND NOT EXISTS "
|
|
"(SELECT 1 FROM deleted_tasks d WHERE d.task=tasks.id)", (owner, key)).fetchone()
|
|
if not row or (expected is not None and row["status"] not in expected):
|
|
return False
|
|
data = json.loads(row["data"])
|
|
due = changes.pop("due", row["due"])
|
|
data.update(changes)
|
|
db.execute("UPDATE tasks SET status=?,due=?,updated=?,data=? WHERE id=?",
|
|
(status, due, time.time() if now is None else now, encode(data), key))
|
|
return True
|
|
|
|
def claim(self, owner, key, now):
|
|
with self.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
row = db.execute("SELECT * FROM tasks WHERE owner=? AND id=? AND status IN ('scheduled','queued') "
|
|
"AND due<=?", (owner, key, now)).fetchone()
|
|
if not row:
|
|
return None
|
|
attempt = uuid.uuid4().hex
|
|
data = json.loads(row["data"])
|
|
data.update(attempt=attempt, attempts=int(data.get("attempts", 0)) + 1)
|
|
db.execute("UPDATE tasks SET status='sending',updated=?,data=? WHERE id=?", (now, encode(data), key))
|
|
db.execute("INSERT INTO attempts VALUES(?,?,?,?,NULL,'sending','{}')", (attempt, key, row["account"], now))
|
|
return {**data, **{k: row[k] for k in
|
|
("id", "owner", "campaign", "customer", "account", "due", "created", "updated")}, "status": "sending"}
|
|
|
|
def finish(self, task, status, *, error="", receipt=None, attempt_status=None):
|
|
now = time.time()
|
|
with self.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
row = db.execute("SELECT data FROM tasks WHERE id=? AND status='sending'", (task["id"],)).fetchone()
|
|
if not row:
|
|
return
|
|
data = json.loads(row[0])
|
|
data.update(error=error, receipt=receipt or {})
|
|
db.execute("UPDATE tasks SET status=?,updated=?,data=? WHERE id=?", (status, now, encode(data), task["id"]))
|
|
db.execute("UPDATE attempts SET finished=?,status=?,detail=? WHERE id=?",
|
|
(now, attempt_status or status, encode({"error": error, "receipt": receipt or {}}), task["attempt"]))
|
|
|
|
def pacing(self, account, now):
|
|
with self.connect() as db:
|
|
row = db.execute("SELECT MAX(started),COUNT(*) FROM attempts WHERE account=? AND started>? AND status!='deferred'",
|
|
(account, now - 86400)).fetchone()
|
|
return float(row[0] or 0), int(row[1])
|
|
|
|
def confirm_receipt(self, owner, key, receipt):
|
|
"""Reconcile a finished unknown attempt without scheduling another send."""
|
|
with self.connect() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
row = db.execute("SELECT * FROM tasks WHERE owner=? AND id=? AND status='delivery_unknown'", (owner, key)).fetchone()
|
|
if not row:
|
|
return False
|
|
data = json.loads(row["data"])
|
|
evidence = {**receipt, "previousError": data.get("error", "")}
|
|
data.update(error="", receipt=evidence)
|
|
db.execute("UPDATE tasks SET status='succeeded',updated=?,data=? WHERE id=?",
|
|
(time.time(), encode(data), key))
|
|
db.execute("UPDATE attempts SET status='succeeded',detail=? WHERE id=? AND task=? AND status='delivery_unknown'",
|
|
(encode({"receipt": evidence, "reconciled_from": "delivery_unknown"}), data.get("attempt"), key))
|
|
return True
|
|
|
|
def enabled(self, owner, value=None):
|
|
with self.connect() as db:
|
|
if value is not None:
|
|
db.execute("INSERT INTO settings VALUES(?,?) ON CONFLICT(owner) DO UPDATE SET enabled=excluded.enabled",
|
|
(owner, int(bool(value))))
|
|
row = db.execute("SELECT enabled FROM settings WHERE owner=?", (owner,)).fetchone()
|
|
return bool(row[0]) if row else True
|
|
|
|
def pending_reply_activations(self, owner):
|
|
# Include removed tasks: deleting a displayed task does not cancel replies.
|
|
with self.connect() as db:
|
|
return [self.task_row(row) for row in db.execute(
|
|
"SELECT * FROM tasks WHERE owner=? AND status='succeeded' AND updated>=? "
|
|
"AND NOT EXISTS (SELECT 1 FROM reply_activations r WHERE r.task=tasks.id) "
|
|
"ORDER BY updated LIMIT 100", (owner, time.time()-86400))]
|
|
|
|
def acknowledge_reply_activations(self, owner, ids):
|
|
with self.connect() as db:
|
|
for key in ids:
|
|
db.execute("INSERT OR IGNORE INTO reply_activations SELECT id,? FROM tasks "
|
|
"WHERE owner=? AND id=? AND status='succeeded'", (time.time(), owner, key))
|