285 lines
12 KiB
Python
285 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""ZYT 患者订单只读适配器。
|
||
|
||
订单查询必须经过 ZYT 的管理端接口，桌面端不直接打开 ``C:\\zyt`` 的源码、
|
||
数据库或配置文件。本模块只负责 HTTP、响应解析和脱敏；患者、租户和会话归属
|
||
由 ``archive_api`` 的桌面代理再次校验。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import socket
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
|
||
class ZytOrderError(RuntimeError):
|
||
"""ZYT 订单查询失败。"""
|
||
|
||
|
||
class ZytOrderSessionExpired(ZytOrderError):
|
||
"""ZYT 管理端会话已过期。"""
|
||
|
||
|
||
class ZytOrderPermissionDenied(ZytOrderError):
|
||
"""当前 ZYT 账号没有订单查询权限。"""
|
||
|
||
|
||
class ZytOrderNotFound(ZytOrderError):
|
||
"""订单不存在或不在当前账号数据范围内。"""
|
||
|
||
|
||
class ZytOrderTrackingUnavailable(ZytOrderError):
|
||
"""订单还没有快递单号，或物流服务暂时查不到真实轨迹。"""
|
||
|
||
def __init__(self, message: str, *, code: str = "tracking_unavailable"):
|
||
super().__init__(message)
|
||
self.code = code
|
||
|
||
|
||
def _expired_message(message: str) -> bool:
|
||
text = str(message or "").lower().replace(" ", "")
|
||
return any(token in text for token in (
|
||
"登录超时", "登录已失效", "登录失效", "登录已过期", "登录过期",
|
||
"请重新登录", "请先登录", "token过期", "token失效", "unauthorized",
|
||
))
|
||
|
||
|
||
def mask_tracking_number(value: Any) -> str:
|
||
"""只保留首两位和末四位，避免把完整运单号交给模型。"""
|
||
|
||
number = re.sub(r"\s+", "", str(value or ""))
|
||
if not number:
|
||
return ""
|
||
if len(number) <= 6:
|
||
return "*" * len(number)
|
||
return f"{number[:2]}****{number[-4:]}"
|
||
|
||
|
||
def _date_text(value: Any) -> str:
|
||
if value in (None, ""):
|
||
return ""
|
||
if isinstance(value, (int, float)):
|
||
try:
|
||
return datetime.fromtimestamp(float(value), tz=timezone.utc).astimezone().strftime(
|
||
"%Y-%m-%d %H:%M:%S"
|
||
)
|
||
except (OverflowError, OSError, ValueError):
|
||
return ""
|
||
return str(value).strip()[:40]
|
||
|
||
|
||
def _first(row: dict[str, Any], *names: str) -> Any:
|
||
for name in names:
|
||
value = row.get(name)
|
||
if value not in (None, ""):
|
||
return value
|
||
return ""
|
||
|
||
|
||
def _audit_text(value: Any) -> str:
|
||
if isinstance(value, str) and value.strip() and not value.strip().isdigit():
|
||
return value.strip()
|
||
try:
|
||
return {0: "待审核", 1: "已通过", 2: "已驳回"}.get(int(value), "未知")
|
||
except (TypeError, ValueError):
|
||
return str(value or "")
|
||
|
||
|
||
def _fulfillment_text(value: Any) -> str:
|
||
if isinstance(value, str) and value.strip() and not value.strip().isdigit():
|
||
return value.strip()
|
||
try:
|
||
return {
|
||
1: "待双审通过", 2: "待发货", 3: "已完成", 4: "已取消",
|
||
5: "已发货", 6: "已签收", 7: "进行中", 8: "暂不制药",
|
||
9: "拒收", 10: "退款", 11: "保留药方", 12: "制药缓发",
|
||
}.get(int(value), "未知")
|
||
except (TypeError, ValueError):
|
||
return str(value or "")
|
||
|
||
|
||
class ZytOrderClient:
|
||
"""调用 ZYT 已存在的管理端订单、详情和物流只读接口。"""
|
||
|
||
ORDER_LIST_PATH = "/adminapi/firstvisit.myPatient/orders"
|
||
ORDER_DETAIL_PATH = "/adminapi/firstvisit.myPatient/orderDetail"
|
||
TRACKING_PATH = "/adminapi/tcm.prescriptionOrder/logisticsTrace"
|
||
|
||
def __init__(self, api_url: str, token: str, *, timeout: float = 12.0):
|
||
self.api_url = str(api_url or "").strip().rstrip("/")
|
||
self.token = str(token or "").strip()
|
||
self.timeout = max(2.0, min(float(timeout), 30.0))
|
||
parsed = urllib.parse.urlparse(self.api_url)
|
||
if parsed.scheme not in {"http", "https"} or not parsed.netloc or parsed.username:
|
||
raise ZytOrderError("ZYT 订单接口地址配置不正确")
|
||
if not self.token:
|
||
raise ZytOrderSessionExpired("ZYT 订单查询会话缺失，请重新验证账号")
|
||
|
||
def _get(self, path: str, params: dict[str, Any]) -> Any:
|
||
query = urllib.parse.urlencode(
|
||
{key: value for key, value in params.items() if value not in (None, "")}
|
||
)
|
||
url = f"{self.api_url}{path}" + (f"?{query}" if query else "")
|
||
request = urllib.request.Request(
|
||
url,
|
||
headers={
|
||
"Accept": "application/json",
|
||
"User-Agent": "WeCom-RPA-Order-Bridge/1.0",
|
||
# ZYT 管理端中间件使用 token 头，不使用 Bearer。
|
||
"token": self.token,
|
||
},
|
||
method="GET",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=self.timeout) as response:
|
||
payload = json.loads(response.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as exc:
|
||
if exc.code == 401:
|
||
raise ZytOrderSessionExpired("ZYT 订单查询登录已过期，请重新验证账号") from exc
|
||
if exc.code == 403:
|
||
raise ZytOrderPermissionDenied("当前 ZYT 账号没有患者订单查询权限") from exc
|
||
raise ZytOrderError(f"ZYT 订单接口请求失败（HTTP {exc.code}）") from exc
|
||
except (urllib.error.URLError, TimeoutError, socket.timeout) as exc:
|
||
raise ZytOrderError(
|
||
f"无法连接 ZYT 订单接口：{getattr(exc, 'reason', exc)}"
|
||
) from exc
|
||
except (ValueError, TypeError, UnicodeDecodeError) as exc:
|
||
raise ZytOrderError("ZYT 订单接口返回了无效数据") from exc
|
||
|
||
if not isinstance(payload, dict):
|
||
raise ZytOrderError("ZYT 订单接口返回了无效数据")
|
||
if str(payload.get("code")) != "1":
|
||
message = str(payload.get("msg") or payload.get("message") or "").strip() if isinstance(payload, dict) else ""
|
||
if "权限" in message or "无权" in message or "禁止" in message:
|
||
raise ZytOrderPermissionDenied(message)
|
||
if str(payload.get("code")) in {"-1", "401", "403"} or _expired_message(message):
|
||
raise ZytOrderSessionExpired(message or "ZYT 订单查询会话已失效")
|
||
if "订单不存在" in message or "无权操作" in message:
|
||
raise ZytOrderNotFound(message)
|
||
if "未填写快递单号" in message or "暂无快递单号" in message:
|
||
raise ZytOrderTrackingUnavailable(message, code="tracking_number_missing")
|
||
if path == self.TRACKING_PATH:
|
||
raise ZytOrderTrackingUnavailable(message or "物流服务暂时查不到轨迹")
|
||
raise ZytOrderError(message or "ZYT 订单查询失败")
|
||
return payload.get("data")
|
||
|
||
@staticmethod
|
||
def _list_payload(data: Any) -> tuple[list[dict[str, Any]], int]:
|
||
if isinstance(data, list):
|
||
return [dict(row) for row in data if isinstance(row, dict)], len(data)
|
||
if not isinstance(data, dict):
|
||
return [], 0
|
||
raw = data.get("lists")
|
||
if not isinstance(raw, list):
|
||
raw = data.get("items")
|
||
if not isinstance(raw, list):
|
||
raw = data.get("data")
|
||
rows = [dict(row) for row in raw if isinstance(row, dict)] if isinstance(raw, list) else []
|
||
try:
|
||
total = max(0, int(data.get("count", data.get("total", len(rows)))))
|
||
except (TypeError, ValueError):
|
||
total = len(rows)
|
||
return rows, total
|
||
|
||
def list_orders(
|
||
self,
|
||
*,
|
||
diagnosis_id: int = 0,
|
||
keyword: str = "",
|
||
page_no: int = 1,
|
||
page_size: int = 20,
|
||
start_date: str = "",
|
||
end_date: str = "",
|
||
) -> dict[str, Any]:
|
||
page_no = max(1, int(page_no))
|
||
page_size = max(1, min(int(page_size), 50))
|
||
keyword = str(keyword or "").strip()[:80]
|
||
# MyPatientOrderLists 的数字 keyword 可按 diagnosis_id 精确命中；绑定
|
||
# 没有诊单号时才使用调用方给出的关键词。
|
||
if not keyword and int(diagnosis_id or 0) > 0:
|
||
keyword = str(int(diagnosis_id))
|
||
data = self._get(
|
||
self.ORDER_LIST_PATH,
|
||
{
|
||
"keyword": keyword,
|
||
"page_no": page_no,
|
||
"page_size": page_size,
|
||
"start_date": str(start_date or "").strip(),
|
||
"end_date": str(end_date or "").strip(),
|
||
},
|
||
)
|
||
rows, total = self._list_payload(data)
|
||
return {"items": rows, "total": total, "page_no": page_no, "page_size": page_size}
|
||
|
||
def get_order_detail(self, order_id: int) -> dict[str, Any]:
|
||
order_id = int(order_id)
|
||
if order_id <= 0:
|
||
raise ValueError("订单编号不正确")
|
||
data = self._get(self.ORDER_DETAIL_PATH, {"id": order_id})
|
||
if not isinstance(data, dict):
|
||
raise ZytOrderNotFound("订单不存在或无权查看")
|
||
return dict(data)
|
||
|
||
def get_tracking(self, order_id: int, *, phone_tail: str = "") -> dict[str, Any]:
|
||
order_id = int(order_id)
|
||
if order_id <= 0:
|
||
raise ValueError("订单编号不正确")
|
||
data = self._get(
|
||
self.TRACKING_PATH,
|
||
{"id": order_id, "phone_tail": re.sub(r"\D", "", str(phone_tail or ""))[-4:]},
|
||
)
|
||
return dict(data) if isinstance(data, dict) else {}
|
||
|
||
|
||
def sanitize_order(row: dict[str, Any], *, queried_at: str = "") -> dict[str, Any]:
|
||
"""仅保留客服回答所需事实，并移除姓名、地址、金额和处方明细。"""
|
||
|
||
row = dict(row or {})
|
||
tracking_number = str(_first(row, "tracking_number", "trackingNumber") or "").strip()
|
||
return {
|
||
"order_id": int(_first(row, "id", "order_id") or 0),
|
||
"order_no": str(_first(row, "order_no", "orderNo") or "")[:80],
|
||
"diagnosis_id": int(_first(row, "diagnosis_id", "diagnosisId") or 0),
|
||
"order_time": _date_text(_first(row, "create_time", "order_time", "createTime")),
|
||
"prescription_audit": _audit_text(_first(row, "prescription_audit_text", "prescription_audit_status")),
|
||
"payment_audit": _audit_text(_first(row, "payment_slip_audit_text", "payment_audit_text", "payment_slip_audit_status")),
|
||
"fulfillment_status": _fulfillment_text(_first(row, "fulfillment_text", "fulfillment_status")),
|
||
"express_company": str(_first(row, "express_company_name", "express_company") or "")[:40],
|
||
"tracking_number_masked": mask_tracking_number(tracking_number),
|
||
"has_tracking_number": bool(tracking_number),
|
||
"queried_at": queried_at,
|
||
}
|
||
|
||
|
||
def sanitize_tracking(payload: dict[str, Any], *, queried_at: str = "") -> dict[str, Any]:
|
||
"""把物流接口的多种字段形状归一为一条可审计的最新轨迹。"""
|
||
|
||
data = dict(payload or {})
|
||
traces = data.get("traces") or data.get("trace") or data.get("list") or []
|
||
if not isinstance(traces, list):
|
||
traces = []
|
||
normalized = []
|
||
for trace in traces[:50]:
|
||
if not isinstance(trace, dict):
|
||
continue
|
||
normalized.append({
|
||
"time": _date_text(_first(trace, "trace_time", "accept_time", "time", "created_at")),
|
||
"description": str(_first(trace, "trace_context", "remark", "context", "description") or "")[:500],
|
||
})
|
||
normalized.sort(key=lambda item: item.get("time") or "", reverse=True)
|
||
latest = normalized[0] if normalized else {}
|
||
return {
|
||
"status": str(_first(data, "state_text", "current_state_text", "status", "state") or ""),
|
||
"time": str(latest.get("time") or _first(data, "latest_trace_time", "last_query_time") or ""),
|
||
"description": str(latest.get("description") or _first(data, "latest_trace_context", "hint") or "")[:500],
|
||
"traces": normalized[:20],
|
||
"queried_at": queried_at,
|
||
}
|
||
|