959 lines
27 KiB
JavaScript
959 lines
27 KiB
JavaScript
const { ObjectId } = require("mongodb");
|
||
const { randomUUID } = require("crypto");
|
||
const {
|
||
AUTOMATION_TYPES,
|
||
getTypeConfig,
|
||
normalizeAutomationConfig,
|
||
randomInteger,
|
||
} = require("./config");
|
||
const { decryptOrderFields } = require("../consult-order/format");
|
||
|
||
const COLLECTION = "hlw-automation-task";
|
||
const TASK_STATUS = {
|
||
PENDING: "PENDING",
|
||
RUNNING: "RUNNING",
|
||
SUCCEEDED: "SUCCEEDED",
|
||
CANCELLED: "CANCELLED",
|
||
FAILED: "FAILED",
|
||
};
|
||
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 } = {}) {
|
||
super(message);
|
||
this.name = "AutomationError";
|
||
this.retryable = retryable;
|
||
}
|
||
}
|
||
|
||
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 }
|
||
) {
|
||
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 = 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: "",
|
||
},
|
||
}
|
||
);
|
||
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,
|
||
};
|
||
}
|
||
|
||
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 } }
|
||
);
|
||
if (!eligible) {
|
||
return { success: false, skipped: true, message: "非支付宝小程序订单" };
|
||
}
|
||
return enqueueTask(db, {
|
||
type: AUTOMATION_TYPES.OPEN_RX,
|
||
orderId,
|
||
corpId,
|
||
baseTime,
|
||
expiresAt,
|
||
});
|
||
}
|
||
|
||
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);
|
||
await 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,
|
||
});
|
||
}
|
||
|
||
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 }, onSale: true })
|
||
.toArray();
|
||
if (masterDrugs.length !== orderDrugs.length) {
|
||
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 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 detail =
|
||
Array.isArray(result?.failList) && result.failList[0]
|
||
? result.failList[0].message
|
||
: result?.message;
|
||
throw new AutomationError(detail || "自动审方失败", { retryable: true });
|
||
}
|
||
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.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,
|
||
AutomationError,
|
||
calculateDueAt,
|
||
enqueueTask,
|
||
scheduleAutoAccept,
|
||
scheduleAutoOpenRx,
|
||
scheduleAutoPassRx,
|
||
buildAutoPrescriptionParams,
|
||
claimNextTask,
|
||
processClaimedTask,
|
||
reconcileAutomationTasks,
|
||
runDueTasks,
|
||
start,
|
||
stop,
|
||
};
|