2026-09-04 09:57:39 +08:00

1322 lines
37 KiB
JavaScript
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

const { ObjectId } = require("mongodb");
const { randomUUID } = require("crypto");
const {
AUTOMATION_TYPES,
getTypeConfig,
normalizeAutomationConfig,
randomInteger,
} = require("./config");
const { decryptOrderFields } = require("../consult-order/format");
const tencentIM = require("../tencent-im");
const COLLECTION = "hlw-automation-task";
const TASK_STATUS = {
PENDING: "PENDING",
RUNNING: "RUNNING",
SUCCEEDED: "SUCCEEDED",
CANCELLED: "CANCELLED",
FAILED: "FAILED",
};
const ERROR_SOURCES = {
SYSTEM: "SYSTEM",
HIS: "HIS",
};
const TASK_CANCEL_REASONS = {
HIGH_RISK_DRUG: "HIGH_RISK_DRUG",
};
const MANUAL_OPEN_RX_MESSAGE =
"订单中含高风险药品,已停止自动开方,请医生人工开方。";
const LEASE_MS = 2 * 60 * 1000;
const POLL_INTERVAL_MS = 1000;
const RECONCILE_INTERVAL_MS = 30 * 1000;
const MAX_TASKS_PER_TICK = 10;
const MAX_RETRY_DELAY_MS = 10 * 60 * 1000;
const WORKER_ID = `${process.pid}:${randomUUID()}`;
let workerDb = null;
let pollTimer = null;
let reconcileTimer = null;
let pollIsRunning = false;
let reconcileIsRunning = false;
class AutomationError extends Error {
constructor(message, { retryable = false, source = ERROR_SOURCES.SYSTEM } = {}) {
super(message);
this.name = "AutomationError";
this.retryable = retryable;
this.source = source === ERROR_SOURCES.HIS ? ERROR_SOURCES.HIS : ERROR_SOURCES.SYSTEM;
}
}
class ManualPrescriptionRequiredError extends AutomationError {
constructor(message = MANUAL_OPEN_RX_MESSAGE) {
super(message);
this.name = "ManualPrescriptionRequiredError";
this.cancelReason = TASK_CANCEL_REASONS.HIGH_RISK_DRUG;
}
}
function getTaskKey(type, orderId, rxId) {
if (type === AUTOMATION_TYPES.PASS_RX) {
return `${type}:${rxId}`;
}
return `${type}:${orderId}`;
}
async function getConfig(db, corpId) {
const config = await db.collection("hlw-config").findOne({ corpId });
return normalizeAutomationConfig(config || {});
}
function getBusinessExpiry(order, fallbackMs = 30 * 60 * 1000) {
if (order && Number.isFinite(order.expireTime) && order.expireTime > Date.now()) {
return order.expireTime;
}
return Date.now() + fallbackMs;
}
function calculateDueAt(type, config, baseTime = Date.now(), random = Math.random) {
const { minSeconds, maxSeconds } = getTypeConfig(config, type);
const seconds = randomInteger(minSeconds, maxSeconds, random);
let dueAt = baseTime + seconds * 1000;
if (type === AUTOMATION_TYPES.OPEN_RX) {
const firstSubmitRxIntervel =
Number.isInteger(config.firstSubmitRxIntervel) && config.firstSubmitRxIntervel > 0
? config.firstSubmitRxIntervel
: 0;
dueAt = Math.max(dueAt, baseTime + firstSubmitRxIntervel * 1000);
}
if (dueAt < Date.now()) {
dueAt = Date.now() + randomInteger(0, 2, random) * 1000;
}
return dueAt;
}
async function enqueueTask(
db,
{
type,
corpId,
orderId,
rxId = "",
baseTime = Date.now(),
expiresAt,
config: providedConfig,
}
) {
if (!corpId || !orderId || !Object.values(AUTOMATION_TYPES).includes(type)) {
return { success: false, message: "自动化任务参数错误" };
}
if (type === AUTOMATION_TYPES.PASS_RX && !rxId) {
return { success: false, message: "自动审方任务缺少处方ID" };
}
const config = providedConfig || (await getConfig(db, corpId));
const stageConfig = getTypeConfig(config, type);
if (!stageConfig.enabled) {
return { success: false, skipped: true, message: "自动化配置未开启" };
}
const now = Date.now();
const taskKey = getTaskKey(type, orderId, rxId);
const dueAt = calculateDueAt(type, config, baseTime);
const task = {
taskKey,
type,
corpId,
orderId,
rxId,
status: TASK_STATUS.PENDING,
dueAt,
nextRunAt: dueAt,
expiresAt: Number.isFinite(expiresAt) ? expiresAt : now + 30 * 60 * 1000,
attempts: 0,
leaseUntil: 0,
leaseOwner: "",
lastError: "",
createdAt: now,
updatedAt: now,
};
const reactivated = await db.collection(COLLECTION).updateOne(
{ taskKey, status: TASK_STATUS.CANCELLED },
{
$set: {
...task,
createdAt: now,
},
$unset: {
completedAt: "",
result: "",
cancelReason: "",
},
}
);
if (reactivated.modifiedCount === 1) {
return {
success: true,
created: true,
reactivated: true,
taskKey,
dueAt,
};
}
let result;
try {
result = await db.collection(COLLECTION).updateOne(
{ taskKey },
{ $setOnInsert: task },
{ upsert: true }
);
} catch (error) {
if (error && error.code === 11000) {
return { success: true, created: false, taskKey, dueAt };
}
throw error;
}
return {
success: true,
created: result.upsertedCount === 1,
taskKey,
dueAt,
};
}
function getDrugCollectionName(order) {
return order && order.consultType === "onlineMedicinePurchase"
? "online-drug-info"
: "drug-info";
}
function getOrderDrugIds(order) {
const drugs = order && Array.isArray(order.drugs) ? order.drugs : [];
return drugs
.map((item) => (item && ObjectId.isValid(item._id) ? new ObjectId(item._id) : null))
.filter(Boolean);
}
async function findHighRiskDrug(db, order) {
const ids = getOrderDrugIds(order);
if (!ids.length) return null;
return db.collection(getDrugCollectionName(order)).findOne(
{ _id: { $in: ids }, riskLevel: "高风险" },
{ projection: { _id: 1, name: 1, riskLevel: 1 } }
);
}
async function cancelAutoOpenForHighRisk({
db,
orderId,
corpId,
doctorCode = "",
baseTime,
expiresAt,
}) {
const now = Date.now();
const taskKey = getTaskKey(AUTOMATION_TYPES.OPEN_RX, orderId, "");
const collection = db.collection(COLLECTION);
const cancelData = {
status: TASK_STATUS.CANCELLED,
leaseUntil: 0,
leaseOwner: "",
lastError: MANUAL_OPEN_RX_MESSAGE,
cancelReason: TASK_CANCEL_REASONS.HIGH_RISK_DRUG,
completedAt: now,
updatedAt: now,
};
const transitioned = await collection.updateOne(
{
taskKey,
$or: [
{ status: TASK_STATUS.PENDING },
{
status: TASK_STATUS.CANCELLED,
cancelReason: { $ne: TASK_CANCEL_REASONS.HIGH_RISK_DRUG },
},
],
},
{ $set: cancelData }
);
let shouldNotify = transitioned.modifiedCount === 1;
if (!shouldNotify) {
const cancelledTask = {
taskKey,
type: AUTOMATION_TYPES.OPEN_RX,
corpId,
orderId,
rxId: "",
status: TASK_STATUS.CANCELLED,
dueAt: Number.isFinite(baseTime) ? baseTime : now,
nextRunAt: Number.isFinite(baseTime) ? baseTime : now,
expiresAt: Number.isFinite(expiresAt) ? expiresAt : now + 30 * 60 * 1000,
attempts: 0,
leaseUntil: 0,
leaseOwner: "",
lastError: MANUAL_OPEN_RX_MESSAGE,
cancelReason: TASK_CANCEL_REASONS.HIGH_RISK_DRUG,
createdAt: now,
updatedAt: now,
completedAt: now,
};
try {
const inserted = await collection.updateOne(
{ taskKey },
{ $setOnInsert: cancelledTask },
{ upsert: true }
);
shouldNotify = inserted.upsertedCount === 1;
} catch (error) {
if (!error || error.code !== 11000) throw error;
}
}
// if (shouldNotify) {
// await sendManualOpenNotification(
// db,
// { type: AUTOMATION_TYPES.OPEN_RX, corpId, orderId },
// doctorCode
// );
// }
return {
success: false,
skipped: true,
manualRequired: true,
message: MANUAL_OPEN_RX_MESSAGE,
taskKey,
};
}
async function scheduleAutoAccept({ db, orderId, corpId, baseTime, expiresAt }) {
const eligible = await db.collection("consult-order").findOne(
{ orderId, corpId, orderSource: "ALIPAY_MINI" },
{ projection: { _id: 1 } }
);
if (!eligible) {
return { success: false, skipped: true, message: "非支付宝小程序订单" };
}
return enqueueTask(db, {
type: AUTOMATION_TYPES.ACCEPT,
orderId,
corpId,
baseTime,
expiresAt,
});
}
async function scheduleAutoOpenRx({ db, orderId, corpId, baseTime, expiresAt }) {
const eligible = await db.collection("consult-order").findOne(
{ orderId, corpId, orderSource: "ALIPAY_MINI" },
{ projection: { _id: 1, drugs: 1, consultType: 1, doctorCode: 1 } }
);
if (!eligible) {
return { success: false, skipped: true, message: "非支付宝小程序订单" };
}
const config = await getConfig(db, corpId);
if (!getTypeConfig(config, AUTOMATION_TYPES.OPEN_RX).enabled) {
return { success: false, skipped: true, message: "自动化配置未开启" };
}
const highRiskDrug = await findHighRiskDrug(db, eligible);
if (highRiskDrug) {
return cancelAutoOpenForHighRisk({
db,
orderId,
corpId,
doctorCode: eligible.doctorCode,
baseTime,
expiresAt,
});
}
return enqueueTask(db, {
type: AUTOMATION_TYPES.OPEN_RX,
orderId,
corpId,
baseTime,
expiresAt,
config,
});
}
async function scheduleAutoPassRx({
db,
orderId,
rxId,
corpId,
baseTime,
expiresAt,
}) {
const eligible = await db.collection("consult-order").findOne(
{ orderId, corpId, orderSource: "ALIPAY_MINI" },
{ projection: { _id: 1 } }
);
if (!eligible) {
return { success: false, skipped: true, message: "非支付宝小程序订单" };
}
return enqueueTask(db, {
type: AUTOMATION_TYPES.PASS_RX,
orderId,
rxId: rxId && rxId.toString(),
corpId,
baseTime,
expiresAt,
});
}
async function ensureIndexes(db) {
const collection = db.collection(COLLECTION);
await collection.createIndex({ taskKey: 1 }, { unique: true });
await collection.createIndex({ status: 1, nextRunAt: 1, leaseUntil: 1 });
await collection.createIndex({ orderId: 1, type: 1 });
await collection.createIndex({ completedAt: 1 });
}
async function claimNextTask(db, now = Date.now()) {
const result = await db.collection(COLLECTION).findOneAndUpdate(
{
$or: [
{
status: TASK_STATUS.PENDING,
nextRunAt: { $lte: now },
},
{
status: TASK_STATUS.RUNNING,
leaseUntil: { $lte: now },
},
],
},
{
$set: {
status: TASK_STATUS.RUNNING,
leaseUntil: now + LEASE_MS,
leaseOwner: WORKER_ID,
updatedAt: now,
},
$inc: { attempts: 1 },
},
{
sort: { nextRunAt: 1, createdAt: 1 },
returnDocument: "after",
}
);
return result && result.value ? result.value : result;
}
async function markTask(db, task, status, extra = {}) {
const now = Date.now();
const terminal = [
TASK_STATUS.SUCCEEDED,
TASK_STATUS.CANCELLED,
TASK_STATUS.FAILED,
].includes(status);
return db.collection(COLLECTION).updateOne(
{
_id: task._id,
status: TASK_STATUS.RUNNING,
leaseOwner: WORKER_ID,
},
{
$set: {
status,
leaseUntil: 0,
leaseOwner: "",
updatedAt: now,
...(terminal ? { completedAt: now } : {}),
...extra,
},
}
);
}
async function renewLease(db, task) {
await db.collection(COLLECTION).updateOne(
{
_id: task._id,
status: TASK_STATUS.RUNNING,
leaseOwner: WORKER_ID,
},
{
$set: {
leaseUntil: Date.now() + LEASE_MS,
updatedAt: Date.now(),
},
}
);
}
function getRetryDelay(attempts) {
const exponent = Math.max(0, Math.min(attempts - 1, 10));
const base = Math.min(5000 * 2 ** exponent, MAX_RETRY_DELAY_MS);
return Math.min(base + Math.floor(Math.random() * 3000), MAX_RETRY_DELAY_MS);
}
async function retryTask(db, task, error) {
const now = Date.now();
if (!Number.isFinite(task.expiresAt) || task.expiresAt <= now) {
await markTask(db, task, TASK_STATUS.FAILED, {
lastError: `业务已过期: ${error.message}`,
});
return;
}
const nextRunAt = Math.min(now + getRetryDelay(task.attempts || 1), task.expiresAt);
await markTask(db, task, TASK_STATUS.PENDING, {
nextRunAt,
lastError: error.message,
});
}
function isSingleAttemptTask(type) {
return [AUTOMATION_TYPES.OPEN_RX, AUTOMATION_TYPES.PASS_RX].includes(type);
}
function isHisUploadFailure(message) {
return typeof message === "string" && /^\s*\[his\]/i.test(message);
}
function cleanHisUploadFailureMessage(message) {
if (typeof message !== "string") return "";
return message
.replace(/^\s*\[his\]\s*/i, "")
.replace(/^处方上传(?:失败|错误)\s*[:]\s*/, "")
.trim();
}
function getFailureNotificationExt(error) {
const errorSource =
error && error.source === ERROR_SOURCES.HIS
? ERROR_SOURCES.HIS
: ERROR_SOURCES.SYSTEM;
const rawReason =
error && typeof error.message === "string" ? error.message.trim() : "";
const reason =
errorSource === ERROR_SOURCES.HIS
? `HIS上传错误${cleanHisUploadFailureMessage(rawReason) || "未知错误"}`
: `系统错误:${rawReason || "未知错误"}`;
return { errorSource, reason };
}
async function sendManualOpenNotification(db, task, knownDoctorCode = "") {
let doctorCode = knownDoctorCode;
if (!doctorCode) {
try {
const order = await db.collection("consult-order").findOne(
{ orderId: task.orderId, corpId: task.corpId },
{ projection: { doctorCode: 1 } }
);
doctorCode = order && order.doctorCode ? order.doctorCode : "";
} catch (lookupError) {
console.error(
"[自动化任务] 查询人工开方提示接收人异常:",
lookupError.message
);
}
}
if (!doctorCode) {
console.error(
`[自动化任务] 未找到订单 ${task.orderId} 的医生,无法发送人工开方提示`
);
return false;
}
try {
const result = await tencentIM(
{
type: "sendSystemNotification",
corpId: task.corpId,
formAccount: task.orderId,
toAccount: doctorCode,
SyncOtherMachine: 1,
msgBody: [
{
MsgType: "TIMCustomElem",
MsgContent: {
Data: "AUTORXMANUAL",
Desc: "notification",
Ext: MANUAL_OPEN_RX_MESSAGE,
},
},
],
},
db
);
if (!result || !result.success) {
console.error(
`[自动化任务] 发送人工开方提示失败: ${result?.message || "未知错误"}`
);
return false;
}
return true;
} catch (notificationError) {
console.error(
"[自动化任务] 发送人工开方提示异常:",
notificationError.message
);
return false;
}
}
async function sendFailureNotification(db, task, error) {
let doctorCode = "";
try {
const order = await db.collection("consult-order").findOne(
{ orderId: task.orderId, corpId: task.corpId },
{ projection: { doctorCode: 1 } }
);
doctorCode = order && order.doctorCode ? order.doctorCode : "";
} catch (lookupError) {
console.error(
"[自动化任务] 查询失败通知接收人异常:",
lookupError.message
);
}
if (
!doctorCode &&
task.type === AUTOMATION_TYPES.PASS_RX &&
ObjectId.isValid(task.rxId)
) {
try {
const rx = await db.collection("diagnostic-record").findOne(
{ _id: new ObjectId(task.rxId), orderId: task.orderId },
{ projection: { doctorCode: 1 } }
);
doctorCode = rx && rx.doctorCode ? rx.doctorCode : "";
} catch (lookupError) {
console.error(
"[自动化任务] 查询处方失败通知接收人异常:",
lookupError.message
);
}
}
if (!doctorCode) {
console.error(
`[自动化任务] 未找到订单 ${task.orderId} 的医生,无法发送失败通知`
);
return;
}
const ext = getFailureNotificationExt(error);
try {
const result = await tencentIM(
{
type: "sendSystemNotification",
corpId: task.corpId,
formAccount: task.orderId,
toAccount: doctorCode,
SyncOtherMachine: 1,
msgBody: [
{
MsgType: "TIMCustomElem",
MsgContent: {
Data: "AUTORXFAIL",
Desc: "notification",
Ext: JSON.stringify(ext),
},
},
],
},
db
);
if (!result || !result.success) {
console.error(
`[自动化任务] 发送失败通知失败: ${result?.message || "未知错误"}`
);
}
} catch (notificationError) {
console.error(
"[自动化任务] 发送失败通知异常:",
notificationError.message
);
}
}
async function assertTaskEnabled(db, task) {
const config = await getConfig(db, task.corpId);
const { enabled } = getTypeConfig(config, task.type);
if (!enabled) {
await markTask(db, task, TASK_STATUS.CANCELLED, {
lastError: "执行前检查发现自动化配置已关闭",
});
return null;
}
return config;
}
async function getAlipayOrder(db, orderId) {
const raw = await db.collection("consult-order").findOne({ orderId });
if (!raw) {
throw new AutomationError("咨询订单不存在");
}
if (raw.orderSource !== "ALIPAY_MINI") {
throw new AutomationError("仅支持支付宝小程序咨询订单");
}
return decryptOrderFields(raw);
}
async function handleAutoAccept(db, task) {
const order = await getAlipayOrder(db, task.orderId);
if (order.orderStatus === "processing") {
await scheduleAutoOpenRx({
db,
orderId: order.orderId,
corpId: order.corpId,
baseTime: order.prescriptionStartTime || Date.now(),
expiresAt: getBusinessExpiry(order),
});
return { orderId: order.orderId };
}
if (order.orderStatus !== "pending") {
throw new AutomationError(`订单状态不支持自动接诊: ${order.orderStatus}`);
}
if (Number.isFinite(order.expireTime) && order.expireTime <= Date.now()) {
throw new AutomationError("咨询订单已过期");
}
const consultOrder = require("../consult-order");
const result = await consultOrder(
{
type: "acceptConsultOrder",
orderId: order.orderId,
corpId: order.corpId,
doctorCode: order.doctorCode,
operationSource: "AUTO",
},
db
);
if (!result || !result.success) {
throw new AutomationError(result?.message || "自动接诊失败", { retryable: true });
}
return { orderId: order.orderId };
}
async function resolveDiagnosisList(db, order) {
const result = [];
const seen = new Set();
const add = (code, name) => {
const normalizedCode = typeof code === "string" ? code.trim() : "";
const normalizedName = typeof name === "string" ? name.trim() : "";
if (!normalizedCode || !normalizedName) return;
const key = `${normalizedCode}:${normalizedName}`;
if (!seen.has(key)) {
result.push({ code: normalizedCode, name: normalizedName, desc: "" });
seen.add(key);
}
};
const medInfo = order.medInfo || {};
add(medInfo.dise_codg, medInfo.dise_name);
const diseaseNames = Array.isArray(order.diseases)
? order.diseases.filter((item) => typeof item === "string" && item.trim())
: [];
if (diseaseNames.length) {
const diagnosisRecords = await db
.collection("hlw-diagnosis")
.find({ name: { $in: diseaseNames } }, { projection: { code: 1, name: 1 } })
.toArray();
for (const name of diseaseNames) {
const match = diagnosisRecords.find((item) => item.name === name);
if (!match || !match.code) {
throw new AutomationError(`诊断“${name}”无法精确匹配诊断编码`);
}
add(match.code, match.name);
}
}
// if (!result.length) {
// throw new AutomationError("订单缺少可用于开方的诊断编码");
// }
return result;
}
async function getMedicineConfig(db, corpId) {
const records = await db
.collection("hlw-config")
.find(
{ group: `${corpId}-medicine-related` },
{ projection: { key: 1, list: 1 } }
)
.toArray();
return records.reduce((map, item) => {
if (item.key && Array.isArray(item.list)) {
map[item.key] = item.list;
}
return map;
}, {});
}
function matchConfig(list, code, name) {
const items = Array.isArray(list) ? list : [];
return (
items.find((item) => code !== undefined && code !== "" && item.code == code) ||
items.find((item) => name && item.name === name)
);
}
async function resolvePrescriptionDrugs(db, order) {
const orderDrugs = Array.isArray(order.drugs) ? order.drugs : [];
if (!orderDrugs.length) {
throw new AutomationError("订单缺少药品,无法自动开方");
}
const ids = orderDrugs
.map((item) => (ObjectId.isValid(item._id) ? new ObjectId(item._id) : null))
.filter(Boolean);
if (ids.length !== orderDrugs.length) {
throw new AutomationError("订单药品ID不完整");
}
const collectionName =
order.consultType === "onlineMedicinePurchase" ? "online-drug-info" : "drug-info";
const masterDrugs = await db
.collection(collectionName)
.find({ _id: { $in: ids } })
.toArray();
if (masterDrugs.length !== orderDrugs.length) {
throw new AutomationError("订单中存在不存在的药品");
}
if (masterDrugs.some((item) => item.riskLevel === "高风险")) {
throw new ManualPrescriptionRequiredError();
}
if (masterDrugs.some((item) => item.onSale !== true)) {
throw new AutomationError("订单中存在已下架的药品");
}
const medicineConfig = await getMedicineConfig(db, order.corpId);
const prefix =
order.consultType === "onlineMedicinePurchase" ? "online-medicine" : "store-medicine";
const dosageUnitList = medicineConfig[`${prefix}-dosage-unit`] || [];
const frequencyList = medicineConfig[`${prefix}-frequence`] || [];
const usageList = medicineConfig[`${prefix}-administration`] || [];
const unitList = medicineConfig["medicine-package-unit"] || [];
return orderDrugs.map((requested) => {
const master = masterDrugs.find(
(item) => item._id.toString() === requested._id.toString()
);
const usage = matchConfig(
usageList,
requested.usageCode,
requested.usageName || master.administration_method
);
const frequency = matchConfig(
frequencyList,
requested.frequencyCode,
requested.frequencyName || master.freq
);
const dosageUnit = matchConfig(
dosageUnitList,
requested.dosage_unit_code,
requested.dosage_unit || master.dosage_unit
);
const unit = matchConfig(unitList, requested.unit || master.unit, requested.unit || master.unit);
const dosage = Number(requested.dosage);
const quantity = Number(requested.quantity);
const days = Number(master.days);
if (!usage || !frequency || !dosageUnit || !unit) {
throw new AutomationError(`药品“${master.name}”的用法用量配置无法匹配`);
}
if (!(dosage > 0) || !(quantity > 0) || !Number.isInteger(quantity)) {
throw new AutomationError(`药品“${master.name}”的剂量或数量不正确`);
}
if (!(days > 0) || !Number.isInteger(days)) {
throw new AutomationError(`药品“${master.name}”未维护有效的用药天数`);
}
if (!master.erpId || !master.insurance_code) {
throw new AutomationError(`药品“${master.name}”缺少HIS所需编码`);
}
return {
_id: master._id.toString(),
erpId: master.erpId,
dosage_form: master.dosage_form || "",
days,
dosage,
dosage_unit: dosageUnit.name,
dosage_unit_code: dosageUnit.code,
drugName: master.name,
specification: master.specification || "",
frequencyCode: frequency.code,
frequencyName: frequency.name,
insurance_code: master.insurance_code,
product_id: master.product_id,
quantity,
unit: unit.code,
usageCode: usage.code,
usageName: usage.name,
package_amount: master.package_amount,
recommended_quantity: master.recommended_quantity,
forceSelfPay: master.forceSelfPay === "Y" ? "Y" : "N",
limitUsageScope:
typeof master.usage_restriction_desc === "string" &&
master.usage_restriction_desc.trim()
? "Y"
: "N",
};
});
}
async function buildAutoPrescriptionParams(db, order) {
const doctor = await db.collection("hlw-doctor").findOne({
corpId: order.corpId,
doctorNo: order.doctorCode,
job: "doctor",
});
if (!doctor) {
throw new AutomationError("开方医生不存在");
}
if (doctor.onlineStatus !== "online") {
throw new AutomationError("开方医生暂不在线", { retryable: true });
}
const [diagnosisList, drugs] = await Promise.all([
resolveDiagnosisList(db, order),
resolvePrescriptionDrugs(db, order),
]);
const diseaseText = Array.isArray(order.diseases) ? order.diseases.join("") : "";
const complaint = [diseaseText, order.description]
.filter((item) => typeof item === "string" && item.trim())
.join(" ")
.slice(0, 500);
if (!complaint) {
throw new AutomationError("订单缺少主诉和病情描述");
}
const config = await getConfig(db, order.corpId);
const prescriptionType =
order.consultType === "onlineMedicinePurchase"
? "onlineMedicinePurchase"
: "storeMedicinePurchase";
const medicinePurchaseRxDuration =
Number.isInteger(config.medicinePurchaseRxDuration) &&
config.medicinePurchaseRxDuration > 0
? config.medicinePurchaseRxDuration
: 30;
return {
complaint,
presentIllness:
typeof order.pastHistoryStr === "string" ? order.pastHistoryStr : "",
dispose: "",
doctorCAUserId: doctor.signatureUrl || "",
patientId: order.patientId,
name: order.name,
orderId: order.orderId,
doctorCode: order.doctorCode,
doctorName: order.doctorName,
deptName: order.deptName,
unitCode: order.unitCode,
drugStoreNo: order.drugStoreNo,
orderSource: order.orderSource,
idCard: order.idCard,
blhno: order.blhno,
medOrgOrderNo: order.medorg_order_no,
address: order.address,
mobile: order.mobile,
prescriptionType,
pickUpType: order.pickUpType,
expireTime:
prescriptionType === "onlineMedicinePurchase"
? Date.now() + medicinePurchaseRxDuration * 60 * 1000
: order.expireTime,
diagnosisList,
drugs,
};
}
async function handleAutoOpenRx(db, task) {
const order = await getAlipayOrder(db, task.orderId);
const existing = await db.collection("diagnostic-record").findOne(
{
orderId: order.orderId,
status: { $in: ["INIT", "PASS"] },
},
{ projection: { _id: 1, status: 1, createTime: 1, expireTime: 1 } }
);
if (existing) {
await scheduleAutoPassRx({
db,
orderId: order.orderId,
rxId: existing._id,
corpId: order.corpId,
baseTime: existing.createTime || Date.now(),
expiresAt: existing.expireTime || getBusinessExpiry(order),
});
return { orderId: order.orderId, rxId: existing._id.toString() };
}
if (order.orderStatus !== "processing") {
throw new AutomationError(`订单状态不支持自动开方: ${order.orderStatus}`);
}
if (Number.isFinite(order.expireTime) && order.expireTime <= Date.now()) {
throw new AutomationError("咨询订单已过期");
}
const params = await buildAutoPrescriptionParams(db, order);
const diagnosticRecord = require("../diagnostic-record");
const reviewResult = await diagnosticRecord(
{
type: "getYbReviewResult",
corpId: order.corpId,
params,
},
db
);
if (!reviewResult || !reviewResult.success) {
throw new AutomationError(
reviewResult?.message || "合理用药审核未通过"
);
}
const result = await diagnosticRecord(
{
type: "addConsultDiagnosis",
corpId: order.corpId,
params,
automated: true,
operationSource: "AUTO",
},
db
);
if (!result || !result.success) {
throw new AutomationError(result?.message || "开方接口返回失败", {
retryable: true,
});
}
const rx = await db.collection("diagnostic-record").findOne(
{ orderId: order.orderId, status: { $in: ["INIT", "PASS"] } },
{ projection: { _id: 1, createTime: 1, expireTime: 1 } }
);
if (!rx) {
throw new AutomationError("自动开方成功但未查询到处方", { retryable: true });
}
await scheduleAutoPassRx({
db,
orderId: order.orderId,
rxId: rx._id,
corpId: order.corpId,
baseTime: rx.createTime || Date.now(),
expiresAt: rx.expireTime || getBusinessExpiry(order),
});
return { orderId: order.orderId, rxId: rx._id.toString() };
}
async function handleAutoPassRx(db, task) {
if (!ObjectId.isValid(task.rxId)) {
throw new AutomationError("处方ID格式错误");
}
const rx = await db.collection("diagnostic-record").findOne({
_id: new ObjectId(task.rxId),
orderId: task.orderId,
});
if (!rx) {
throw new AutomationError("处方不存在");
}
if (rx.status === "PASS") {
return { orderId: task.orderId, rxId: task.rxId };
}
if (rx.status !== "INIT") {
throw new AutomationError(`处方状态不支持自动审方: ${rx.status}`);
}
if (!rx.pharmacistNo) {
throw new AutomationError("处方未分配审方药师", { retryable: true });
}
const pharmacist = await db.collection("hlw-doctor").findOne({
corpId: task.corpId,
doctorNo: rx.pharmacistNo,
job: "pharmacist",
});
if (!pharmacist) {
throw new AutomationError("审方药师不存在");
}
if (pharmacist.onlineStatus !== "online") {
throw new AutomationError("审方药师暂不在线", { retryable: true });
}
const diagnosticRecord = require("../diagnostic-record");
const result = await diagnosticRecord(
{
type: "auditDiagnosis",
ids: [task.rxId],
status: "PASS",
pharmacistNo: rx.pharmacistNo,
corpId: task.corpId,
operationSource: "AUTO",
},
db
);
if (!result || !result.success) {
const failure =
Array.isArray(result?.failList) && result.failList[0]
? result.failList[0]
: null;
const detail = failure ? failure.message : result?.message;
throw new AutomationError(detail || "审方接口返回失败", {
retryable: true,
source: isHisUploadFailure(detail)
? ERROR_SOURCES.HIS
: ERROR_SOURCES.SYSTEM,
});
}
return { orderId: task.orderId, rxId: task.rxId };
}
async function executeTask(db, task) {
const config = await assertTaskEnabled(db, task);
if (!config) return;
let result;
if (task.type === AUTOMATION_TYPES.ACCEPT) {
result = await handleAutoAccept(db, task);
} else if (task.type === AUTOMATION_TYPES.OPEN_RX) {
result = await handleAutoOpenRx(db, task);
} else if (task.type === AUTOMATION_TYPES.PASS_RX) {
result = await handleAutoPassRx(db, task);
} else {
throw new AutomationError(`未知自动化任务类型: ${task.type}`);
}
await markTask(db, task, TASK_STATUS.SUCCEEDED, {
lastError: "",
result: result || {},
});
}
async function processClaimedTask(db, task) {
const heartbeat = setInterval(
() =>
renewLease(db, task).catch((error) => {
console.error("[自动化任务] 续租失败:", error.message);
}),
Math.floor(LEASE_MS / 3)
);
if (typeof heartbeat.unref === "function") heartbeat.unref();
try {
await executeTask(db, task);
} catch (error) {
const automationError =
error instanceof AutomationError
? error
: new AutomationError(error.message || "自动化任务执行失败", {
retryable: true,
});
if (automationError.cancelReason === TASK_CANCEL_REASONS.HIGH_RISK_DRUG) {
const marked = await markTask(db, task, TASK_STATUS.CANCELLED, {
lastError: automationError.message,
cancelReason: TASK_CANCEL_REASONS.HIGH_RISK_DRUG,
});
if (marked && (marked.modifiedCount === 1 || marked.matchedCount === 1)) {
await sendManualOpenNotification(db, task);
}
} else if (isSingleAttemptTask(task.type)) {
await markTask(db, task, TASK_STATUS.FAILED, {
lastError: automationError.message,
});
await sendFailureNotification(db, task, automationError);
} else if (automationError.retryable) {
await retryTask(db, task, automationError);
} else {
await markTask(db, task, TASK_STATUS.FAILED, {
lastError: automationError.message,
});
}
} finally {
clearInterval(heartbeat);
}
}
async function runDueTasks(db = workerDb) {
if (!db || pollIsRunning) return;
pollIsRunning = true;
try {
const tasks = [];
for (let i = 0; i < MAX_TASKS_PER_TICK; i += 1) {
const task = await claimNextTask(db);
if (!task) break;
tasks.push(task);
}
await Promise.all(tasks.map((task) => processClaimedTask(db, task)));
} catch (error) {
console.error("[自动化任务] 执行器异常:", error.message);
} finally {
pollIsRunning = false;
}
}
async function reconcileAutomationTasks(db = workerDb) {
if (!db || reconcileIsRunning) return;
reconcileIsRunning = true;
try {
const now = Date.now();
const todayStart = new Date();
todayStart.setHours(0, 0, 0, 0);
const orders = await db
.collection("consult-order")
.find(
{
orderSource: "ALIPAY_MINI",
orderStatus: { $in: ["pending", "processing"] },
$or: [
{ expireTime: { $gt: now } },
{
expireTime: { $exists: false },
createTime: { $gte: todayStart.getTime() },
},
],
},
{
projection: {
orderId: 1,
corpId: 1,
orderStatus: 1,
createTime: 1,
prescriptionStartTime: 1,
expireTime: 1,
},
}
)
.toArray();
const corpIds = [...new Set(orders.map((item) => item.corpId).filter(Boolean))];
const configs = await db
.collection("hlw-config")
.find({ corpId: { $in: corpIds } })
.toArray();
const configMap = new Map(
configs.map((item) => [item.corpId, normalizeAutomationConfig(item)])
);
for (const order of orders) {
const config = configMap.get(order.corpId) || normalizeAutomationConfig({});
if (order.orderStatus === "pending" && config.autoAcceptOrder) {
await scheduleAutoAccept({
db,
orderId: order.orderId,
corpId: order.corpId,
baseTime: order.createTime,
expiresAt: order.expireTime,
});
}
if (order.orderStatus === "processing" && config.autoOpenRx) {
const record = await db.collection("diagnostic-record").findOne(
{ orderId: order.orderId, status: { $in: ["INIT", "PASS"] } },
{ projection: { _id: 1 } }
);
if (!record) {
await scheduleAutoOpenRx({
db,
orderId: order.orderId,
corpId: order.corpId,
baseTime: order.prescriptionStartTime || order.createTime,
expiresAt: order.expireTime,
});
}
}
}
const pendingRxList = await db
.collection("diagnostic-record")
.find(
{
status: "INIT",
orderSource: "ALIPAY_MINI",
$or: [
{ expireTime: { $gt: now } },
{
expireTime: { $exists: false },
createTime: { $gte: todayStart.getTime() },
},
],
},
{
projection: {
_id: 1,
orderId: 1,
corpId: 1,
createTime: 1,
expireTime: 1,
},
}
)
.toArray();
for (const rx of pendingRxList) {
const config =
configMap.get(rx.corpId) ||
(await getConfig(db, rx.corpId));
if (config.autoPassRx) {
const order = orders.find((item) => item.orderId === rx.orderId);
await scheduleAutoPassRx({
db,
orderId: rx.orderId,
rxId: rx._id,
corpId: rx.corpId,
baseTime: rx.createTime,
expiresAt: rx.expireTime || getBusinessExpiry(order),
});
}
}
} catch (error) {
console.error("[自动化任务] 核对任务失败:", error.message);
} finally {
reconcileIsRunning = false;
}
}
async function start(db) {
if (pollTimer || reconcileTimer) return;
workerDb = db;
await ensureIndexes(db);
await reconcileAutomationTasks(db);
await runDueTasks(db);
pollTimer = setInterval(() => runDueTasks(db), POLL_INTERVAL_MS);
reconcileTimer = setInterval(
() => reconcileAutomationTasks(db),
RECONCILE_INTERVAL_MS
);
if (typeof pollTimer.unref === "function") pollTimer.unref();
if (typeof reconcileTimer.unref === "function") reconcileTimer.unref();
console.log("[自动化任务] 执行器已启动");
}
function stop() {
if (pollTimer) clearInterval(pollTimer);
if (reconcileTimer) clearInterval(reconcileTimer);
pollTimer = null;
reconcileTimer = null;
workerDb = null;
pollIsRunning = false;
reconcileIsRunning = false;
}
module.exports = {
COLLECTION,
TASK_STATUS,
AUTOMATION_TYPES,
ERROR_SOURCES,
TASK_CANCEL_REASONS,
MANUAL_OPEN_RX_MESSAGE,
AutomationError,
ManualPrescriptionRequiredError,
getFailureNotificationExt,
calculateDueAt,
enqueueTask,
scheduleAutoAccept,
scheduleAutoOpenRx,
scheduleAutoPassRx,
buildAutoPrescriptionParams,
claimNextTask,
processClaimedTask,
reconcileAutomationTasks,
runDueTasks,
start,
stop,
};