415 lines
22 KiB
Python
415 lines
22 KiB
Python
"""Patient-bound proactive contact business rules, independent of Qt."""
|
||
from datetime import datetime
|
||
import hashlib
|
||
import time
|
||
import uuid
|
||
import threading
|
||
from collections import OrderedDict
|
||
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
|
||
|
||
from proactive_contact_schedule import CONTACT_ZONE, next_slot, normalize_schedule
|
||
from proactive_contact_store import ContactStore, encode
|
||
|
||
MAX_BATCH = 100
|
||
|
||
|
||
def binding_version(binding):
|
||
return encode({k: binding.get(k) for k in ("id", "patient_id", "updated_at", "relation_type", "status")})
|
||
|
||
|
||
def bound_people(context, account, conversation, peer):
|
||
"""Do not accept a different member's patient or a same-named conversation."""
|
||
conv = context.get("conversation") or {}
|
||
permissions = context.get("permissions") or {}
|
||
if not context.get("available") or permissions.get("can_search") is not True:
|
||
raise ValueError(context.get("reason") or permissions.get("reason") or "患者绑定暂无法核验")
|
||
if conv.get("external_id") != conversation or str(conv.get("source_account")) != account:
|
||
raise ValueError("患者绑定所属账号或会话不一致")
|
||
people = [p for p in context.get("people", []) if any(
|
||
str(i.get("external_id")) == peer for i in p.get("identities", []))]
|
||
if len(people) != 1:
|
||
raise ValueError("当前客户的联系人身份无法唯一核对")
|
||
return people[0]
|
||
|
||
|
||
class PatientGateway:
|
||
"""Use authenticated existing backend APIs; never treat a submitted ID as proof."""
|
||
def __init__(self):
|
||
self._candidate_cache = OrderedDict()
|
||
self._candidate_lock = threading.Lock()
|
||
self._candidate_epoch = 0
|
||
self._candidate_build_cache = {}
|
||
self._candidate_pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="contact-binding")
|
||
|
||
def invalidate_candidates(self):
|
||
with self._candidate_lock:
|
||
self._candidate_cache.clear()
|
||
self._candidate_epoch += 1
|
||
|
||
def close_candidates(self):
|
||
self._candidate_pool.shutdown(wait=False, cancel_futures=True)
|
||
|
||
def warm_candidates(self):
|
||
from review_assistant_contacts import active_context
|
||
from wecom_native_sender import readonly_build_validation_scope
|
||
with readonly_build_validation_scope(self._candidate_build_cache):
|
||
active_context()
|
||
|
||
@staticmethod
|
||
def _verify_patient_visibility(binding):
|
||
import backend_client
|
||
# The deployed keyword API matches names/phones, not patient IDs.
|
||
# Use the server's bound name for discovery, but only accept the exact
|
||
# patient ID. Same-name results alone are never proof of this binding.
|
||
keyword = str(binding.get("patient_name") or "").strip()[:64]
|
||
if not keyword:
|
||
raise ValueError("绑定患者缺少姓名，请在管理绑定中重新查询并核对")
|
||
patient_id = int(binding["patient_id"])
|
||
page_size = 50
|
||
for page_no in range(1, 21):
|
||
result = backend_client.search_desktop_patients(keyword, page_no=page_no, page_size=page_size)
|
||
items = result.get("items")
|
||
if not isinstance(items, list):
|
||
raise ValueError("患者查询结果格式不正确，请稍后重试")
|
||
for item in items:
|
||
try:
|
||
found_id = int(item.get("patient_id") or 0) if isinstance(item, dict) else 0
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if found_id == patient_id:
|
||
return
|
||
# total counts raw diagnoses; items are deduplicated by patient.
|
||
# A short page can still have more patients on subsequent pages.
|
||
total = max(0, int(result.get("total") or 0))
|
||
if page_no * page_size >= total:
|
||
raise ValueError("当前查询未找到绑定的患者，请在管理绑定中重新查询并核对")
|
||
if not items:
|
||
raise ValueError("患者查询结果未完整返回，暂无法核验，请稍后重试")
|
||
raise ValueError("同名患者查询结果超过核验范围，请在管理绑定中核对患者信息")
|
||
|
||
def owner(self):
|
||
import backend_client
|
||
settings = backend_client.load_settings()
|
||
account = settings.get("desktop_account") or {}
|
||
if not account.get("id") or not settings.get("desktop_token_protected"):
|
||
raise ValueError("请先登录客服助手")
|
||
return hashlib.sha256(encode([settings.get("server_url"), account["id"]]).encode()).hexdigest()
|
||
|
||
def candidates(self, query="", offset=0, *, progress=None, cancelled=lambda: False, refresh_bindings=False):
|
||
from wecom_native_sender import readonly_build_validation_scope
|
||
with readonly_build_validation_scope(self._candidate_build_cache):
|
||
return self._candidates(query, offset, progress=progress, cancelled=cancelled, refresh_bindings=refresh_bindings)
|
||
|
||
def _candidates(self, query="", offset=0, *, progress=None, cancelled=lambda: False, refresh_bindings=False):
|
||
import backend_client
|
||
from review_assistant_contacts import active_context, list_recipients, recipient_peer
|
||
query = str(query or "").strip()[:100]
|
||
identity = active_context()
|
||
if not identity.get("ready"):
|
||
raise ValueError(identity.get("reason") or "请先读取当前企微账号")
|
||
account = identity["account_id"]
|
||
page = list_recipients(account, query, 20, offset, status="sendable")
|
||
if not page.get("ready"):
|
||
raise ValueError(page.get("reason") or "客户列表不可用")
|
||
# Cache display data only, scoped to the login session and exact account.
|
||
# add/check/preview/send always use the uncached verification below.
|
||
owner = self.owner()
|
||
session = hashlib.sha256(str(backend_client.load_settings().get("desktop_token_protected") or "").encode()).hexdigest()
|
||
with self._candidate_lock:
|
||
epoch = self._candidate_epoch
|
||
scope = (owner, session, account, epoch)
|
||
items = [{"account": account, "conversation": c["id"], "name": c["name"],
|
||
"matched_name": c.get("matchedName", ""),
|
||
"bindings": [], "binding_loading": True, "reason": "正在读取患者绑定…"} for c in page["items"]]
|
||
result = {"account": account, "items": items, "offset": offset, "has_more": page["has_more"],
|
||
"total": page["total"], "query": query}
|
||
|
||
def publish():
|
||
if progress and not cancelled():
|
||
progress({**result, "items": [dict(row) for row in items]})
|
||
|
||
def lookup(conversation):
|
||
try:
|
||
ctx = backend_client.desktop_patient_context(account, conversation, timeout=4.0)
|
||
person = bound_people(ctx, account, conversation, recipient_peer(account, conversation))
|
||
bindings = []
|
||
seen = set()
|
||
for b in person.get("patient_bindings", []):
|
||
if not isinstance(b, dict) or b.get("status") != "active":
|
||
continue
|
||
try:
|
||
patient_id = int(b.get("patient_id") or 0)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
binding_id, patient_name = str(b.get("id") or "").strip(), str(b.get("patient_name") or "").strip()
|
||
if patient_id <= 0 or not binding_id or not patient_name or binding_id in seen:
|
||
continue
|
||
seen.add(binding_id)
|
||
bindings.append({"id": binding_id, "patient_id": patient_id, "patient_name": patient_name})
|
||
value = {"bindings": bindings, "binding_loading": False,
|
||
"reason": "" if bindings else "未绑定患者，请先绑定"}
|
||
with self._candidate_lock:
|
||
if epoch == self._candidate_epoch and not cancelled():
|
||
self._candidate_cache[(*scope, conversation)] = (time.monotonic(), value)
|
||
while len(self._candidate_cache) > 400:
|
||
self._candidate_cache.popitem(last=False)
|
||
return value
|
||
except Exception as exc:
|
||
return {"bindings": [], "binding_loading": False, "binding_error": True,
|
||
"reason": f"绑定暂未读取成功，可重新搜索重试：{str(exc)[:180]}"}
|
||
|
||
pending = {}
|
||
for row in items:
|
||
with self._candidate_lock:
|
||
cached = self._candidate_cache.get((*scope, row["conversation"]))
|
||
if cached and not refresh_bindings and time.monotonic() - cached[0] < 60:
|
||
row.update(cached[1])
|
||
elif refresh_bindings:
|
||
# Explicit searches must not retain a revoked binding, even if
|
||
# fresh lookup fails. Autocomplete can reuse only verified cache.
|
||
with self._candidate_lock:
|
||
self._candidate_cache.pop((*scope, row["conversation"]), None)
|
||
publish() # The local list is visible before the first remote response.
|
||
try:
|
||
for row in items:
|
||
if cancelled():
|
||
break
|
||
if row["binding_loading"]:
|
||
pending[self._candidate_pool.submit(lookup, row["conversation"])] = row
|
||
while pending and not cancelled():
|
||
done, _ = wait(pending, timeout=.05, return_when=FIRST_COMPLETED)
|
||
for future in done:
|
||
pending.pop(future).update(future.result())
|
||
if done:
|
||
publish()
|
||
finally:
|
||
for future in pending:
|
||
future.cancel()
|
||
if self.owner() != owner:
|
||
raise ValueError("登录账号已切换，请重新搜索")
|
||
return result
|
||
|
||
def check(self, account, conversation, binding_id, *, full=True):
|
||
import backend_client
|
||
from review_assistant_contacts import validate_recipient, recipient_peer
|
||
from conversation_blacklist import blocked_reason
|
||
valid = validate_recipient(account, conversation)
|
||
if valid.get("ok") is not True:
|
||
raise ValueError(valid.get("reason") or "客户不可发送")
|
||
reason = blocked_reason(account, conversation)
|
||
if reason:
|
||
raise ValueError(reason)
|
||
peer = recipient_peer(account, conversation)
|
||
ctx = backend_client.desktop_patient_context(account, conversation)
|
||
person = bound_people(ctx, account, conversation, peer)
|
||
binding = next((b for b in person.get("patient_bindings", []) if b.get("id") == binding_id
|
||
and b.get("status") == "active" and int(b.get("patient_id") or 0) > 0), None)
|
||
if not binding or not binding.get("updated_at"):
|
||
raise ValueError("患者未绑定或原绑定已失效，请重新选择")
|
||
if full:
|
||
self._verify_patient_visibility(binding)
|
||
# Recheck binding after the patient lookup, which may take seconds.
|
||
latest = backend_client.desktop_patient_context(account, conversation)
|
||
current = bound_people(latest, account, conversation, peer)
|
||
if not any(binding_version(b) == binding_version(binding) for b in current.get("patient_bindings", [])):
|
||
raise ValueError("核验期间患者绑定已变化，请刷新后重试")
|
||
return {"account": account, "conversation": conversation, "peer": peer,
|
||
"name": valid["recipient"]["name"], "staff_name": valid.get("account_name", ""),
|
||
"person_id": person["id"], "patient_id": binding["patient_id"],
|
||
"patient_name": binding.get("patient_name", ""), "binding_id": binding["id"],
|
||
"binding_version": binding_version(binding), "valid": True, "reason": "", "checked_at": time.time()}
|
||
|
||
|
||
class ContactService:
|
||
def __init__(self, root, gateway, transport, *, clock=time.time):
|
||
self.store = ContactStore(root)
|
||
self.gateway, self.transport, self.clock = gateway, transport, clock
|
||
self.previews = {}
|
||
|
||
def verify(self, customer, *, full=True):
|
||
current = self.gateway.check(customer["account"], customer["conversation"], customer["binding_id"], full=full)
|
||
if current["binding_version"] != customer["binding_version"] or current["person_id"] != customer["person_id"]:
|
||
raise ValueError("患者绑定已变化，请重新添加客户并创建任务")
|
||
return current
|
||
|
||
def mark_invalid(self, owner, customer, reason):
|
||
stored = next((c for c in self.store.customers(owner) if c["id"] == customer["id"]), None)
|
||
if stored and stored["binding_version"] == customer["binding_version"]:
|
||
self.store.save_customer(owner, {**stored, "valid": False, "reason": reason})
|
||
for task in self.store.tasks(owner, pending=True, limit=10000):
|
||
if task["customer"] == customer["id"] and task["binding_version"] == customer["binding_version"]:
|
||
self.store.update(owner, task["id"], "paused", expected={"scheduled", "queued"}, error=reason)
|
||
|
||
def add(self, owner, items):
|
||
if not isinstance(items, list) or not 1 <= len(items) <= MAX_BATCH:
|
||
raise ValueError("每批请选择 1–100 位客户")
|
||
added, rejected = [], []
|
||
for item in items:
|
||
try:
|
||
data = self.gateway.check(str(item["account"]), str(item["conversation"]), str(item["binding_id"]))
|
||
if self.gateway.owner() != owner:
|
||
raise ValueError("登录账号已切换")
|
||
old = next((c for c in self.store.customers(owner) if c["account"] == data["account"] and c["peer"] == data["peer"]), None)
|
||
if old and old["binding_version"] != data["binding_version"]:
|
||
self.mark_invalid(owner, old, "患者绑定已变化，旧任务需取消并重新创建")
|
||
if old:
|
||
data["template"] = old.get("template", "")
|
||
added.append(self.store.save_customer(owner, data))
|
||
except Exception as exc:
|
||
rejected.append({"name": item.get("name") or item.get("conversation"), "reason": str(exc)[:240]})
|
||
return {"message": f"已加入 {len(added)} 位客户，未加入 {len(rejected)} 位", "added": added, "rejected": rejected}
|
||
|
||
def selected(self, owner, ids):
|
||
if not isinstance(ids, list) or not 1 <= len(set(ids)) <= MAX_BATCH:
|
||
raise ValueError("每批请选择 1–100 位客户")
|
||
wanted = set(ids)
|
||
rows = [c for c in self.store.customers(owner) if c["id"] in wanted]
|
||
if len(rows) != len(wanted):
|
||
raise ValueError("客户不存在或不属于当前登录账号")
|
||
return rows
|
||
|
||
def set_text(self, owner, ids, text):
|
||
self.validate_text(text)
|
||
self.previews.clear()
|
||
for customer in self.selected(owner, ids):
|
||
self.store.save_customer(owner, {**customer, "template": text})
|
||
return {"message": f"已为 {len(set(ids))} 位客户保存文字。本次未创建发送任务。"}
|
||
|
||
@staticmethod
|
||
def validate_text(text):
|
||
if not isinstance(text, str) or not text.strip():
|
||
raise ValueError("请先填写发送文字")
|
||
if "\0" in text:
|
||
raise ValueError("发送文字包含不支持的字符，请重新输入")
|
||
if len(text.encode("utf-8")) > 4000:
|
||
raise ValueError("发送文字过长，请缩短后重试（最多约 1300 个汉字）")
|
||
|
||
def preview(self, owner, ids, schedule, *, text=None):
|
||
self.previews.clear()
|
||
if text is not None:
|
||
self.validate_text(text)
|
||
now = self.clock()
|
||
schedule = normalize_schedule(schedule, now)
|
||
tasks = []
|
||
cursor = now
|
||
for customer in self.selected(owner, ids):
|
||
try:
|
||
self.verify(customer)
|
||
except Exception as exc:
|
||
self.mark_invalid(owner, customer, str(exc))
|
||
raise
|
||
body = str(customer.get("template", "")) if text is None else text
|
||
if not body.strip():
|
||
raise ValueError(f"{customer['name']} 尚未保存文字，请填写本次发送文字或先为该客户保存文字")
|
||
due = next_slot(schedule, cursor)
|
||
if due is None:
|
||
raise ValueError("所选时间范围不足，请扩大时间段")
|
||
for key, value in {"客户名称": customer["name"], "客服名称": customer.get("staff_name", ""),
|
||
"日期": datetime.fromtimestamp(due, CONTACT_ZONE).strftime("%Y-%m-%d")}.items():
|
||
if "{{" + key + "}}" in body and not value:
|
||
raise ValueError(f"{customer['name']} 缺少变量：{key}")
|
||
body = body.replace("{{" + key + "}}", value)
|
||
if "{{" in body or "}}" in body:
|
||
raise ValueError("存在未知或未填写的文字变量")
|
||
self.validate_text(body)
|
||
baseline = self.transport.baseline(customer)
|
||
tasks.append({**{k: v for k, v in customer.items() if k != "id"}, "customer": customer["id"], "text": body, "schedule": schedule,
|
||
"due": due, "baseline": baseline})
|
||
cursor = due + schedule["interval"]
|
||
preview = {"id": uuid.uuid4().hex, "owner": owner, "expires": self.clock() + 600,
|
||
"tasks": tasks, "schedule": schedule}
|
||
self.previews = {preview["id"]: preview}
|
||
return {"preview": preview, "message": "预览已生成，本次尚未创建任务。请核对收件人、患者、正文和发送时间，再确认创建。"}
|
||
|
||
def confirm(self, owner, key):
|
||
preview = self.previews.get(key)
|
||
if not preview or preview["owner"] != owner or preview["expires"] < self.clock():
|
||
raise ValueError("预览已过期，请重新预览")
|
||
for task in preview["tasks"]:
|
||
self.verify(task)
|
||
if self.gateway.owner() != owner:
|
||
raise ValueError("登录账号已切换")
|
||
count = self.store.create_campaign(owner, preview, self.clock())
|
||
return {"message": f"已创建 {count} 条发送任务" if count else "此批任务已创建，请查看联系计划",
|
||
"confirmed": key, "creation": {"campaign": key, "count": len(preview["tasks"]),
|
||
"new_count": count, "kind": preview["schedule"]["kind"]}}
|
||
|
||
def control(self, owner, ids, action):
|
||
if not isinstance(ids, list) or not 1 <= len(ids) <= MAX_BATCH:
|
||
raise ValueError("请选择需要操作的任务")
|
||
results = []
|
||
errors = []
|
||
for key in dict.fromkeys(ids):
|
||
task = self.store.get_task(owner, key)
|
||
if not task:
|
||
continue
|
||
try:
|
||
if action == "cancel":
|
||
ok = self.store.update(owner, key, "cancelled", expected={"scheduled", "queued", "paused", "failed"}, error="用户取消")
|
||
elif action == "pause":
|
||
ok = self.store.update(owner, key, "paused", expected={"scheduled", "queued"}, error="用户暂停")
|
||
elif action == "resume":
|
||
if task["status"] != "paused":
|
||
raise ValueError("只有暂停且未提交的任务可恢复；失败或结果不确定请先核对记录")
|
||
self.verify(task)
|
||
if self.transport.baseline(task) != task["baseline"]:
|
||
raise ValueError("客户已有新消息，请取消旧任务并重新预览创建")
|
||
due = next_slot(task["schedule"], self.clock())
|
||
if due is None:
|
||
raise ValueError("计划时间范围已结束，请重新创建")
|
||
ok = self.store.update(owner, key, "scheduled", expected={"paused"}, due=due, error="")
|
||
else:
|
||
raise ValueError("不支持的任务操作")
|
||
message = {"pause": "已暂停，不会自动发送", "cancel": "已取消，不会发送",
|
||
"resume": "已恢复调度，尚未发送；将按计划通过协议通道自动执行"}[action] if ok else "当前状态不可操作"
|
||
results.append(f"{task['name']}：{message}")
|
||
if not ok:
|
||
errors.append(results[-1])
|
||
except Exception as exc:
|
||
results.append(f"{task['name']}：{exc}")
|
||
errors.append(results[-1])
|
||
return {"message": "；".join(results), "errors": errors}
|
||
|
||
def delete_tasks(self, owner, ids):
|
||
if (not isinstance(ids, list) or not 1 <= len(ids) <= MAX_BATCH
|
||
or any(not isinstance(key, str) or not key for key in ids)):
|
||
raise ValueError("请选择 1–100 条需要删除的任务")
|
||
if self.gateway.owner() != owner:
|
||
raise ValueError("登录账号已切换，请刷新")
|
||
deleted, errors = self.store.delete_tasks(owner, ids)
|
||
message = f"已删除 {len(deleted)} 条任务。未发送的任务已停止；已发送消息不会撤回，客户后续回复仍按自动回复设置处理。"
|
||
if errors:
|
||
message += "未删除：" + "；".join(errors)
|
||
return {"message": message, "deleted": deleted, "errors": errors}
|
||
|
||
def recheck(self, owner):
|
||
confirmed = self.reconcile_receipts(owner)
|
||
for customer in self.store.customers(owner):
|
||
try:
|
||
current = self.verify(customer)
|
||
self.store.save_customer(owner, {**customer, **current})
|
||
except Exception as exc:
|
||
self.mark_invalid(owner, customer, str(exc))
|
||
return confirmed
|
||
|
||
def reconcile_receipts(self, owner):
|
||
from proactive_contact_scheduler import verified_receipt
|
||
confirmed = 0
|
||
for task in self.store.tasks(owner, include_deleted=True):
|
||
if task["status"] != "delivery_unknown":
|
||
continue
|
||
try:
|
||
receipt = self.transport.receipt(task)
|
||
if verified_receipt(receipt, task) and self.gateway.owner() == owner:
|
||
confirmed += int(self.store.confirm_receipt(owner, task["id"], receipt))
|
||
except Exception:
|
||
# A failed read must neither resend nor turn uncertainty into a
|
||
# claimed failure. The next refresh/periodic check can retry reads.
|
||
continue
|
||
return confirmed
|
||
|
||
def snapshot(self, owner):
|
||
tasks = self.store.tasks(owner)
|
||
return {"customers": self.store.customers(owner), "tasks": tasks, "enabled": self.store.enabled(owner),
|
||
"timezone": "UTC+08:00", "transport": "已验证协议通道", "maxBatch": MAX_BATCH}
|