264 lines
13 KiB
Python
264 lines
13 KiB
Python
"""Serial scheduling and the existing native sender adapter. No implicit sends."""
|
|
from contextlib import contextmanager
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
|
|
from proactive_contact_schedule import next_slot
|
|
|
|
|
|
def verified_receipt(receipt, task):
|
|
return bool(isinstance(receipt, dict) and receipt.get("status") == "confirmed"
|
|
and receipt.get("accountId") == task["account"]
|
|
and receipt.get("conversationId") == task["conversation"] and receipt.get("serverId")
|
|
and receipt.get("senderVerified") is True and receipt.get("textVerified") is True)
|
|
|
|
|
|
class NativeContactTransport:
|
|
def __init__(self, root):
|
|
self.root = Path(root)
|
|
self.reply_active = lambda: False
|
|
self.waiting_reason = ""
|
|
|
|
@contextmanager
|
|
def database(self, account):
|
|
from reply_database import LiveReplyDatabase
|
|
# Each reader refreshes/decrypts its own snapshot. Reusing the reply
|
|
# cache can collide with another reader's open SQLite files on Windows.
|
|
scratch = self.root / "tmp"
|
|
scratch.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix="proactive-db-", dir=scratch) as cache:
|
|
db = LiveReplyDatabase._open_database(account=account, initialize=False,
|
|
load_metadata=False, cache_dir=cache)
|
|
try:
|
|
yield db
|
|
finally:
|
|
if db is not None:
|
|
db.close()
|
|
|
|
@staticmethod
|
|
def read_baseline(db, task):
|
|
ctx = db.get_conversation_context_by_id(task["account"], task["conversation"], limit=500)
|
|
if not ctx or ctx.get("account") != task["account"] or ctx.get("conv_id") != task["conversation"]:
|
|
raise ValueError("当前客户的消息库无法按账号和会话核验")
|
|
incoming = [m for m in ctx.get("messages", []) if m.get("is_self") is False]
|
|
last = incoming[-1] if incoming else {}
|
|
return [str(last.get(k, "")) for k in ("rowid", "server_id", "dedup_key", "send_time")]
|
|
|
|
def baseline(self, task):
|
|
from wecom_native_sender import discover, NativeClient
|
|
from wecom_native_evidence import send_authorized
|
|
identity = discover()
|
|
if identity["accountId"] != task["account"]:
|
|
raise ValueError("企微账号已切换")
|
|
with NativeClient(identity["pid"]) as client:
|
|
if client.account != task["account"] or not send_authorized(client.profile, client.account):
|
|
raise ValueError("请先在自动化设置完成当前企微账号的协议发送及回执验证")
|
|
with self.database(task["account"]) as db:
|
|
return self.read_baseline(db, task)
|
|
|
|
def ready(self, task):
|
|
"""Native sends do not use the mouse/keyboard; keep reply priority only."""
|
|
self.waiting_reason = ""
|
|
if not self.reply_active():
|
|
return True
|
|
for name in ("pending_replies.json", "pending_replies.protocol.json"):
|
|
path = self.root / name
|
|
if not path.exists():
|
|
continue
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
self.waiting_reason = "正在读取自动回复状态，稍后自动继续"
|
|
return False
|
|
pending = data.get("pending", data)
|
|
for key, row in pending.items():
|
|
if not isinstance(row, dict) or key in {"seen", "manual_handoffs"}:
|
|
continue
|
|
if row.get("stage") in {"sending", "generating", "reading", "ready", "ready_to_send", "waiting_send", "queued"}:
|
|
self.waiting_reason = "自动回复正在处理客户消息，完成后自动继续，无需点击恢复"
|
|
return False
|
|
return True
|
|
|
|
@contextmanager
|
|
def lock(self):
|
|
import send_lock
|
|
with send_lock.lock("proactive-contact", 0) as acquired:
|
|
yield acquired
|
|
|
|
def send(self, task, guard):
|
|
from wecom_native_sender import NativeSender, discover
|
|
identity = discover()
|
|
if identity["accountId"] != task["account"]:
|
|
raise ValueError("企微账号已切换")
|
|
with self.database(task["account"]) as db:
|
|
sender = NativeSender(db, self.root / "protocol_sends.sqlite3")
|
|
try:
|
|
return sender.send("proactive-contact:" + task["id"], identity["pid"], task["account"],
|
|
task["conversation"], task["text"],
|
|
guard=lambda: guard() and self.read_baseline(db, task) == task["baseline"])
|
|
finally:
|
|
sender.close()
|
|
|
|
def receipt(self, task):
|
|
path = self.root / "protocol_sends.sqlite3"
|
|
if not path.exists():
|
|
return None
|
|
db = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)
|
|
try:
|
|
row = db.execute("SELECT status,result FROM sends WHERE request_id=?",
|
|
("proactive-contact:" + task["id"],)).fetchone()
|
|
if row and row[0] == "confirmed":
|
|
return json.loads(row[1])
|
|
pending = json.loads(row[1]) if row else {}
|
|
finally:
|
|
db.close()
|
|
if pending.get("nativeCallback") != "success" or not pending.get("serverId"):
|
|
return None
|
|
from wecom_native_sender import NativeSender
|
|
with self.database(task["account"]) as messages:
|
|
sender = NativeSender(messages, path)
|
|
try:
|
|
return sender.reconcile("proactive-contact:" + task["id"], task["account"],
|
|
task["conversation"], task["text"])
|
|
finally:
|
|
sender.close()
|
|
|
|
|
|
class ContactScheduler:
|
|
def __init__(self, service, *, stopped=lambda: False, interrupted=lambda task: False):
|
|
self.service = service
|
|
self.stopped, self.interrupted = stopped, interrupted
|
|
self.recovered = set()
|
|
self.next_receipt_check = {}
|
|
|
|
def wait_reason(self, task):
|
|
if self.service.transport.ready(task):
|
|
return ""
|
|
return getattr(self.service.transport, "waiting_reason", "") or "正在等待发送通道可用，可用后自动继续，无需点击恢复"
|
|
|
|
def queue(self, task, reason):
|
|
if task["status"] == "queued" and task.get("error") == reason:
|
|
return False
|
|
return self.service.store.update(task["owner"], task["id"], "queued",
|
|
expected={"scheduled", "queued"}, error=reason)
|
|
|
|
def recover(self, owner):
|
|
# The application already enforces a single desktop instance. Also hold
|
|
# the common send mutex while resolving interrupted sends.
|
|
with self.service.transport.lock() as acquired:
|
|
if not acquired:
|
|
return False
|
|
for task in self.service.store.tasks(owner, pending=True, limit=10000):
|
|
if task["status"] == "sending":
|
|
receipt = self.service.transport.receipt(task)
|
|
confirmed = verified_receipt(receipt, task)
|
|
self.service.store.finish(task, "succeeded" if confirmed else "delivery_unknown",
|
|
receipt=receipt, error="" if confirmed else "上次执行中断，发送结果待核对，禁止自动重发")
|
|
self.recovered.add(owner)
|
|
return True
|
|
|
|
def tick(self, owner):
|
|
service, store = self.service, self.service.store
|
|
if owner not in self.recovered and not self.recover(owner):
|
|
for task in store.tasks(owner, pending=True, limit=10000):
|
|
if task["status"] in {"scheduled", "queued"} and task["due"] <= service.clock():
|
|
return self.queue(task, "发送通道正在使用，释放后自动继续")
|
|
return False
|
|
if self.stopped():
|
|
return False
|
|
now = service.clock()
|
|
changed = False
|
|
if now >= self.next_receipt_check.get(owner, 0):
|
|
self.next_receipt_check[owner] = now + 30
|
|
changed = bool(service.reconcile_receipts(owner))
|
|
if not store.enabled(owner):
|
|
return changed
|
|
tasks = sorted(store.tasks(owner, pending=True, limit=10000), key=lambda t: (t["due"], t["created"]))
|
|
for task in tasks:
|
|
if task["status"] not in {"scheduled", "queued"} or task["due"] > now:
|
|
continue
|
|
due = next_slot(task["schedule"], now)
|
|
if due is None:
|
|
store.update(owner, task["id"], "skipped", error="计划日期范围已结束")
|
|
return True
|
|
if due > now:
|
|
changed = store.update(owner, task["id"], "scheduled", due=due, error="") or changed
|
|
continue
|
|
last, count = store.pacing(task["account"], now)
|
|
if count >= 100:
|
|
store.update(owner, task["id"], "paused", error="已达到本账号滚动 24 小时 100 次发送上限")
|
|
return True
|
|
if now < last + task["schedule"]["interval"]:
|
|
return self.queue(task, f"等待客户发送间隔（{task['schedule']['interval']} 秒），到时自动继续") or changed
|
|
with service.transport.lock() as acquired:
|
|
if not acquired:
|
|
return self.queue(task, "发送通道正在使用，释放后自动继续") or changed
|
|
if self.stopped() or self.interrupted(task):
|
|
return False
|
|
try:
|
|
if service.gateway.owner() != owner:
|
|
return False
|
|
waiting = self.wait_reason(task)
|
|
if waiting:
|
|
return self.queue(task, waiting) or changed
|
|
try:
|
|
service.verify(task)
|
|
except Exception as exc:
|
|
service.mark_invalid(owner, {**task, "id": task["customer"]}, str(exc))
|
|
raise
|
|
if service.transport.baseline(task) != task["baseline"]:
|
|
raise ValueError("客户已有新消息，请核对后重新创建主动联系任务")
|
|
except Exception as exc:
|
|
store.update(owner, task["id"], "paused", error=str(exc)[:300])
|
|
return True
|
|
claimed = store.claim(owner, task["id"], service.clock())
|
|
if claimed is None:
|
|
continue
|
|
reason = [""]
|
|
deferred = [False]
|
|
|
|
def guard():
|
|
deferred[0] = False
|
|
try:
|
|
current = store.get_task(owner, claimed["id"])
|
|
if (self.stopped() or self.interrupted(claimed) or not store.enabled(owner)
|
|
or service.gateway.owner() != owner or not current or current["status"] != "sending"):
|
|
raise ValueError("任务已暂停或登录账号已变化")
|
|
instant = service.clock()
|
|
slot = next_slot(claimed["schedule"], instant)
|
|
if slot is None or slot > instant:
|
|
raise ValueError("已离开允许发送的时间段")
|
|
service.verify(claimed, full=False)
|
|
waiting = self.wait_reason(claimed)
|
|
if waiting:
|
|
deferred[0] = True
|
|
raise ValueError(waiting)
|
|
return True
|
|
except Exception as exc:
|
|
reason[0] = str(exc)[:300]
|
|
return False
|
|
|
|
try:
|
|
if not guard():
|
|
store.finish(claimed, "queued" if deferred[0] else "paused", error=reason[0],
|
|
attempt_status="deferred" if deferred[0] else None)
|
|
return True
|
|
receipt = service.transport.send(claimed, guard)
|
|
if not verified_receipt(receipt, task):
|
|
store.finish(claimed, "delivery_unknown", error="发送回执无法完整核对，禁止自动重发")
|
|
else:
|
|
store.finish(claimed, "succeeded", receipt=receipt)
|
|
except Exception as exc:
|
|
kind = type(exc).__name__
|
|
# Only our own temporary guard rejection, accompanied by
|
|
# SendCancelled (guaranteed before native dispatch), may
|
|
# return to the queue. Unknown/failed dispatches never retry.
|
|
retry = kind == "SendCancelled" and deferred[0]
|
|
status = "queued" if retry else "paused" if kind in {"NativeUnavailable", "SendCancelled"} else "failed" if kind == "DeliveryFailed" else "delivery_unknown"
|
|
store.finish(claimed, status, error=reason[0] or str(exc)[:300],
|
|
attempt_status="deferred" if retry else None)
|
|
return True
|
|
return changed
|