1056 lines
30 KiB
JavaScript
Raw Permalink Normal View History

2026-07-27 11:28:33 +08:00
const dayjs = require("dayjs");
const common = require("../common");
const weCom = require("../weCom/index.js");
const utils = require("./utils");
const logger = require("../utils/logger");
const api = require("../api");
let db = null;
exports.main = async (event, mongodb) => {
db = mongodb;
switch (event.type) {
case "createGroupmsgTask":
return await createGroupmsgTask(event);
case "getWecomGroupmesgList":
return await getWecomGroupmesgList(event);
case "addWecomMsgTemplate":
return await addWecomMsgTemplate(event);
case "stopGroupmsgTask":
return await stopGroupmsgTask(event);
case "getGroupmesgById":
return await getGroupmesgById(event);
case "getCusomterGroupmsgSendResult":
return await getCusomterGroupmsgSendResult(event);
case "addWecomMsgTemplateForTeamOrCorp":
return await addWecomMsgTemplateForTeamOrCorp(event);
case "getTeamMemberExecuteState":
return await getTeamMemberExecuteState(event);
case "remindGroupmsgSend":
return await remindGroupmsgSend(event);
case "createCurrentGroupMsgTask":
return await createCurrentGroupMsgTask(event);
case "stopExpireGroupmsgTask":
return await stopExpireGroupmsgTask(event);
case "getGroupmsgMsgidsByTaskIds":
return await getGroupmsgMsgidsByTaskIds(event);
}
};
// 创建群发任务
async function remindGroupmsgSend(context) {
return await weCom.main(context, db);
}
async function getGroupmsgMsgidsByTaskIds(context) {
try {
const { taskIds, corpId } = context;
if (!taskIds || taskIds.length === 0) {
return {
success: false,
message: "任务ID列表为空",
};
}
const list = await db
.collection("groupmsg-task")
.find({ taskId: { $in: taskIds } })
.toArray();
if (!list || list.length === 0) {
return {
success: false,
message: "未找到相关任务",
};
}
const res = await getGroupmsgResultBymsgIds(list, corpId);
return {
success: true,
data: res,
message: "获取成功",
};
} catch (error) {
return {
success: false,
message: "获取失败",
};
}
}
async function getGroupmsgResultBymsgIds(list, corpId) {
if (!list || list.length === 0) {
return [];
}
const groupmsgList = list.map(({ msgid, executor: userid, taskId }) => ({
msgid,
userid,
taskId,
}));
console.log("list", list);
const arr = groupmsgList
.filter((i) => i.msgid)
.map(async (item) => {
const { msgid, userid, taskId } = item;
const res = await weCom.main({
type: "getWecomGroupmsgSendResult",
corpId,
msgid,
userid,
});
const { send_list } = res;
const Obj =
Array.isArray(send_list) && send_list.length > 0 ? send_list[0] : {};
const { status, result } = Obj;
return {
sendSatus: status,
result,
taskId,
};
});
let sendResult = await Promise.all(arr);
console.log("sendResult", sendResult);
return sendResult;
}
async function createGroupmsgTask(context) {
let { params, accessToken, executedExternalUserIds = [] } = context;
// // 1. 为群发任务添加创建时间
params["createTime"] = dayjs().valueOf();
const _id = common.generateRandomString(24);
params["_id"] = _id;
// 2. 插入群发任务到 MongoDB
await db.collection("groupmsg-task").insertOne(params);
// 3. 如果发送来源是 "MINE",则调用 WeCom 消息接口
if (params.sendSource === "MINE") {
let { success, executedExternalUserIds: ids } = await createWecomMsg({
params,
executedExternalUserIds,
templateId: _id, // 使用任务 ID 作为模板 ID
externalUserIds: params.externalUserIds,
accessToken,
});
// 如果成功且返回了有效的用户 ID 列表,更新执行的用户 ID 列表
if (success && ids && Array.isArray(ids)) {
executedExternalUserIds = ids;
}
}
// 返回成功的结果
return {
success: true,
message: "创建群发任务成功",
executedExternalUserIds,
id: _id, // 返回任务 ID
};
}
/**
* 创建群发任务
* @param {*} params
* @param {*} executedExternalUserIds
* @param {*} templateId
* @returns
*/
async function createWecomMsg({
params,
executedExternalUserIds = [],
templateId,
accessToken,
}) {
const {
corpId,
externalUserIds = [],
attachments,
content,
executor,
} = params;
if (!externalUserIds || externalUserIds.length === 0) {
return await addWecomMsgTemplate({
corpId,
accessToken,
templateId,
executedExternalUserIds,
});
} else {
return await addWecomMsgTemplateApi({
corpId,
accessToken,
templateId,
externalUserIds,
attachments,
content,
executor,
});
}
}
async function addWecomMsgTemplateForTeamOrCorp(context) {
try {
const { params, id, accessToken } = context;
const { userIds, ...rest } = params;
const { teamIds } = rest;
let createSounrce = rest.sendSource;
let executedExternalUserIds = [];
for (const userId of userIds) {
let { customers, externalUserIds } = await getSendGroupmsgCustomers({
teamIds,
userId,
corpId: rest.corpId,
});
rest.createSounrce = createSounrce;
rest.customers = customers;
rest.executor = userId;
rest.sendSource = "MINE";
rest.parentId = id;
rest.executeStatus = "doing";
rest.startTaskDate = dayjs().startOf("day").valueOf();
rest.externalUserIds = externalUserIds;
let { executedExternalUserIds: userIds } = await createGroupmsgTask({
params: rest,
executedExternalUserIds,
accessToken,
});
executedExternalUserIds = userIds;
}
return {
success: true,
message: "创建群发任务成功",
};
} catch (error) {
return {
success: false,
message: "error",
};
}
}
// 获取团队群发, 通过userId获取到该团队下加过好友的外部联系人
async function getSendGroupmsgCustomers({ teamIds, userId, corpId }) {
// 查询到member表中符合teamId字段在teamIds 和 externalUserId 存在且不为空的所有customers, 只获取_id 和 externalUserId, 注意分页查询, 获取到所有数据
const fetchData = async (page, pageSize, db) => {
let query = {
teamId: { $in: teamIds },
externalUserId: { $exists: true, $ne: "" },
corpId,
};
let result = await db
.collection("member")
.aggregate([
{ $match: query },
{ $skip: (page - 1) * pageSize },
{ $limit: pageSize },
{ $project: { externalUserId: 1, _id: 1 } },
])
.toArray();
return result;
};
const allCustomerData = await utils.getAllData(fetchData, db);
// 筛选出所有的外部联系人ID
const externalUserIds = allCustomerData.map((item) => item.externalUserId);
if (externalUserIds.length === 0) {
return { customers: [], externalUserIds: [] };
}
const fetchData2 = async (page, pageSize, db) => {
let res = await api.getCorpApi({
type: "getExternalUserIdByUserId",
userId,
externalUserIds,
corpId,
page,
pageSize,
});
console.log("res", res.data);
return res.data;
};
const allFriendsData = await utils.getAllData(fetchData2, db);
// 筛选出所有的外部联系人ID
const externalUserIds2 = allFriendsData.map((item) => item.external_userid);
const allCustomers = allCustomerData.filter((item) =>
externalUserIds2.includes(item.externalUserId)
);
const customers = allCustomers.map((item) => item._id);
return { customers, externalUserIds: externalUserIds2 };
}
async function getGroupmesgById(content) {
const { id } = content;
try {
const task = await db.collection("groupmsg-task").findOne({ _id: id });
return {
success: true,
data: task,
message: "获取群发任务成功",
};
} catch (error) {
return {
success: false,
message: "获取群发任务失败",
};
}
}
async function getWecomGroupmesgList(context) {
const { corpId, accessToken, page, pageSize, params } = context;
const {
teamIds,
startCreateTime,
endCreateTime,
orderType = "desc",
...rest
} = params;
// 构建查询条件
let query = { corpId, ...rest };
// 如果 teamIds 存在且为数组,则使用 $in 操作符
if (Array.isArray(teamIds) && teamIds.length > 0) {
query["teamIds"] = { $in: teamIds };
}
// 处理时间范围条件
if (startCreateTime && dayjs(startCreateTime).isValid()) {
query["createTime"] = { $gte: dayjs(startCreateTime).valueOf() };
}
if (endCreateTime && dayjs(endCreateTime).isValid()) {
query["createTime"] = query["createTime"] || {}; // 如果 startCreateTime 存在,保持合并
query["createTime"]["$lte"] = dayjs(endCreateTime).valueOf();
}
// 获取企业访问令牌
try {
// 计算总数
const total = await db.collection("groupmsg-task").countDocuments(query); // 获取符合条件的总记录数
// 计算分页页数
const pages = Math.ceil(total / pageSize);
// 获取企业群发消息列表
const corpGroupmsgList = await db
.collection("groupmsg-task")
.find(query) // 使用 find() 替代 where()
.skip((page - 1) * pageSize) // 实现分页
.limit(pageSize) // 每页数量限制
.sort({ createTime: orderType === "desc" ? -1 : 1 }) // 按 createTime 倒序排序
.toArray(); // 返回数组
// 进一步处理群发消息任务
if (rest.sendSource === "MINE") {
const data = await getUserGroupmsgTask({
corpId,
accessToken,
corpGroupmsgList,
});
return {
success: true,
data,
total,
pages,
size: pageSize,
message: "获取群发任务成功",
};
} else {
const data = await getUserGroupmsgSendResult({
corpId,
accessToken,
corpGroupmsgList,
});
return {
success: true,
data,
total,
pages,
size: pageSize,
message: "获取群发任务成功",
};
}
} catch (error) {
logger.error("获取群发任务失败:", error);
return {
success: false,
message: error.message || "获取群发任务失败",
};
}
}
async function getUserGroupmsgSendResult(context) {
const { corpId, accessToken, corpGroupmsgList = [] } = context;
// 获取个人群发任务的 ID 列表
const createdTaskList = corpGroupmsgList.map((item) =>
getGroupmsgTaskByTeamId(item)
);
let msgIdList = [];
for (let i = 0; i < createdTaskList.length; i += 10) {
const promiseArr = createdTaskList.slice(i, i + 10);
let res = await Promise.all(promiseArr);
msgIdList.push(...res.flat());
}
// 获取每个任务的发送结果
const arr = msgIdList.map((item) => {
const { msgid, parentId } = item;
return weCom.main(
{
type: "getWecomGroupmsgtask",
corpId,
accessToken,
msgid,
parentId,
},
db
);
});
let sendResult = [];
// 批量执行发送结果请求,每次最多 10 个
for (let i = 0; i < arr.length; i += 10) {
const promiseArr = arr.slice(i, i + 10);
let res = await Promise.all(promiseArr);
sendResult.push(...res);
}
// 获取发送结果并处理任务
const taskCounts = await Promise.all(
corpGroupmsgList.map(async (item) => {
const taskCountResult = await db
.collection("groupmsg-task")
.countDocuments({ parentId: item._id });
return {
_id: item._id,
taskCount:
taskCountResult > 0
? taskCountResult
: item.userIds
? item.userIds.length
: 0,
};
})
);
const sendSuccessCounts = await Promise.all(
corpGroupmsgList.map(async (item) => {
const list = sendResult.filter((task) => task.parentId === item._id);
let task_list = [];
list.forEach((task) => {
task_list.push(...task.task_list);
});
return {
_id: item._id,
sendSuccessCount: task_list.filter((task) => task.status === 2).length,
};
})
);
corpGroupmsgList.forEach((item) => {
const taskCountObj = taskCounts.find((t) => t._id === item._id);
const sendSuccessCountObj = sendSuccessCounts.find(
(s) => s._id === item._id
);
item["taskCount"] = taskCountObj ? taskCountObj.taskCount : 0;
item["sendSuccessCount"] = sendSuccessCountObj
? sendSuccessCountObj.sendSuccessCount
: 0;
item["noSendCount"] = item["taskCount"] - item["sendSuccessCount"];
});
return corpGroupmsgList;
}
async function getGroupmsgTaskByTeamId(item) {
const fetchData = async (lastId = null, pageSize, db) => {
// 设置查询条件
let query = {
parentId: item._id,
corpId: item.corpId,
};
// 如果有 lastId则通过 _id 分页
if (lastId) query["_id"] = { $gt: lastId }; // 只查询比 lastId 大的记录
// 使用聚合查询代替分页的 skip 和 limit
let result = await db
.collection("groupmsg-task")
.aggregate([
{ $match: query }, // 筛选符合条件的记录
{ $sort: { _id: 1 } }, // 按照 _id 排序(升序)
{ $limit: pageSize }, // 限制返回数量
])
.toArray(); // 转换为数组形式
return result;
};
// 获取所有数据,采用分页方式查询并合并数据
const allData = await utils.getAllData(fetchData, db);
return allData;
}
async function getUserGroupmsgTask({
corpId,
accessToken,
corpGroupmsgList,
customerList = "",
}) {
const arr = corpGroupmsgList
.filter((i) => i.msgid)
.map((item) => {
const { msgid, executor } = item;
return weCom.main({
type: "getWecomGroupmsgSendResult",
corpId,
accessToken,
msgid,
userid: executor,
});
});
// 使用 Promise.all 优化并行请求
let sendResult = await Promise.all(arr);
sendResult = sendResult.flat(); // 展开结果数组,避免嵌套
// 合并所有 customerList 或 corpGroupmsgList 中的 customers
const customerIds =
customerList || corpGroupmsgList.flatMap((item) => item.customers);
const fetchData = async (page, pageSize, db) => {
let query = {
_id: { $in: customerIds }, // 查询指定的 customerIds
externalUserId: {
$in: corpGroupmsgList.flatMap((item) => item.externalUserIds),
}, // 只查询符合 externalUserId 的记录
};
// 使用聚合管道来提高查询效率,避免使用 skip 和 limit
let result = await db
.collection("member")
.aggregate([
{ $match: query },
{ $sort: { _id: 1 } }, // 按照 _id 升序排序
{ $skip: (page - 1) * pageSize }, // 跳过指定数量的记录
{ $limit: pageSize }, // 限制返回的数量
])
.toArray();
return result;
};
const allCustomerData = await utils.getAllData(fetchData, db);
// 批量处理群发任务
for (const item of corpGroupmsgList) {
if (item.msgid) {
// 获取与当前消息相关的 send_list
let { send_list } = sendResult.find(
(task) => task.msgid === item.msgid
) || { send_list: [] };
// 过滤符合条件的客户数据
let customers = allCustomerData
.filter((customer) => item.customers.includes(customer._id))
.map((customer) => {
const { age, sex, name, externalUserId, _id } = customer;
let obj = send_list.find((i) => i.external_userid === externalUserId);
return {
age,
sex,
name,
externalUserId,
_id,
...obj,
};
});
// 统计各类状态的数量
item["unexecutedCount"] = customers.filter(
(customer) => customer.status === 0 || !customer.status
).length;
let sendSuccessList = customers.filter(
(customer) => customer.status === 1
);
if (sendSuccessList.length > 0) {
item["sendTime"] = sendSuccessList[0].send_time;
}
item["sendSuccessCount"] = sendSuccessList.length;
item["sendFileCount"] = customers.filter(
(customer) => customer.status === 2 || customer.status === 3
).length;
item["sendSuccessList"] = sendSuccessList;
item["sendFileList"] = customers.filter(
(customer) => customer.status === 2 || customer.status === 3
);
item["send_list"] = send_list;
item["customersCount"] = customers.length;
} else {
// 默认情况
item["customersCount"] = item.customers.length;
item["unexecutedCount"] = item.customers.length;
item["sendSuccessCount"] = 0;
item["sendFileCount"] = 0;
item["sendTime"] = "";
item["sendSuccessList"] = [];
item["sendFileList"] = [];
}
}
return corpGroupmsgList;
}
// 添加企业微信群发消息模板
async function addWecomMsgTemplate(context) {
let {
corpId,
accessToken,
templateId,
executedExternalUserIds = [],
} = context;
try {
const template = await db
.collection("groupmsg-task")
.findOne({ _id: templateId });
// 如果没有找到模板,返回错误
if (!template) {
return {
success: false,
message: "模板未找到",
executedExternalUserIds,
};
}
let { customers = [], content = "", attachments = [], executor } = template;
// 如果没有客户数据,直接返回
if (customers.length === 0) {
return {
success: false,
message: "客户列表为空",
executedExternalUserIds,
};
}
// 使用聚合查询批量获取客户数据,并分页处理
const fetchData = async (page, pageSize, db) => {
const result = await db
.collection("member")
.aggregate([
{ $match: { _id: { $in: customers } } }, // 过滤客户
{ $skip: (page - 1) * pageSize }, // 分页
{ $limit: pageSize }, // 分页
{ $project: { externalUserId: 1 } }, // 只返回 externalUserId 字段
])
.toArray();
return result;
};
// 获取所有客户数据
const list = await utils.getAllData(fetchData, db);
console.log("list", list);
// 提取外部用户ID并去重
let externalUserIds = list
.map((item) => item.externalUserId)
.filter((item) => item);
console.log("externalUserIds", externalUserIds);
// 去重
externalUserIds = [...new Set(externalUserIds)];
// 去除已经发送过的客户
externalUserIds = externalUserIds.filter(
(item) => !executedExternalUserIds.includes(item)
);
// 更新已执行的用户ID列表
executedExternalUserIds = [...executedExternalUserIds, ...externalUserIds];
// 如果没有可发送的外部联系人,删除模板并返回
if (externalUserIds.length === 0) {
await db.collection("groupmsg-task").deleteOne({ _id: templateId });
return {
success: false,
message: "外部联系人为空",
executedExternalUserIds,
};
}
// 检查执行者是否为空
if (!executor) {
return {
success: false,
message: "发送人为空",
executedExternalUserIds,
};
}
// 处理附件,去除标题字段
attachments = attachments.map((item) => {
delete item.title;
return item;
});
// 调用发送消息模板的API
return await addWecomMsgTemplateApi({
corpId,
accessToken,
templateId,
executedExternalUserIds,
externalUserIds,
attachments,
content,
executor,
});
} catch (error) {
return {
success: false,
message: error.message || error,
};
}
}
async function addWecomMsgTemplateApi({
corpId,
accessToken,
templateId,
executedExternalUserIds = [],
externalUserIds = [],
attachments = [],
content = "",
executor = "",
}) {
let { errcode, fail_list, msgid } = await weCom.main(
{
type: "addWecomMsgTemplate",
corpid: corpId,
accessToken,
attachments,
content,
executor,
externalUserIds,
},
db
);
if (errcode === 0) {
await db.collection("groupmsg-task").updateOne(
{ _id: templateId }, // 查询条件:通过 ObjectId 查找指定的文档
{
$set: {
// 使用 $set 来设置更新的字段
msgid,
externalUserIds,
fail_list,
},
}
);
return {
success: true,
message: "创建群发任务成功",
executedExternalUserIds,
};
} else {
return {
success: false,
message: "创建群发任务失败",
executedExternalUserIds,
};
}
}
async function getTeamMemberExecuteState(context) {
const { corpId, accessToken, teamTaskId, customers } = context;
try {
// 校验必需字段
if (!teamTaskId || !corpId) {
return {
success: false,
data: [],
message: "缺少必要的参数",
};
}
// 使用 MongoDB 查询语法进行数据查询
let memberTaskList = await db
.collection("groupmsg-task")
.find({ parentId: teamTaskId })
.toArray();
// 处理每个任务并获取群发任务结果
const list = await getUserGroupmsgTask({
corpId,
accessToken,
corpGroupmsgList: memberTaskList,
customerList: customers,
});
return {
success: true,
data: list,
message: "获取成功",
};
} catch (error) {
logger.error(error); // 打印错误信息
return {
success: false,
data: [],
message: "获取失败",
};
}
}
async function getMemberCountBySendResult(context) {
const { list, customers } = context;
// 提取所有 externalUserId
let externalUserIds = list.map((item) => item.external_userid);
try {
// 使用 MongoDB 查询获取符合条件的记录数量
const count = await db.collection("member").countDocuments({
_id: { $in: customers }, // 使用 $in 操作符查找多个值
externalUserId: { $in: externalUserIds }, // 使用 $in 查找匹配的 externalUserIds
});
return count; // 返回符合条件的记录数
} catch (error) {
logger.error(error); // 输出错误信息,便于调试
return 0; // 出现错误时返回 0
}
}
async function getCusomterGroupmsgSendResult(context) {
const {
msgid,
corpId,
userid,
customers: customerIds,
accessToken,
} = context;
try {
// 获取群发消息发送结果
let { success, send_list } = await weCom.main(
{
type: "getWecomGroupmsgSendResult",
msgid,
userid,
corpId,
accessToken,
},
db
);
// 如果获取失败,直接返回失败信息
if (!success) {
return {
success: false,
message: "获取失败",
};
}
// 查询数据库获取客户信息
let res = await db
.collection("member")
.find({
_id: { $in: customerIds }, // 使用 $in 来查询多个客户
})
.toArray(); // 使用 toArray() 来获取完整的客户数据
// 处理查询结果并合并群发发送结果
let customers = res.map((item) => {
const { age, sex, name, externalUserId, _id } = item;
// 查找匹配的发送结果
let obj =
send_list && Array.isArray(send_list) && send_list.length > 0
? send_list.find((i) => i.external_userid === externalUserId)
: {};
// 返回合并后的客户信息
return {
age,
sex,
name,
externalUserId,
_id,
...obj,
};
});
return {
success: true,
message: "获取成功",
data: customers,
};
} catch (error) {
// 捕获错误并返回
logger.error("Error fetching customer group message result:", error);
return {
success: false,
message: "发生错误,无法获取数据。",
};
}
}
// 停止企业群发
async function stopWecomGroupmsgTask(content) {
let { corpId, accessToken, msgid, taskId } = content;
try {
let res = await weCom.main(
{
type: "stopWecomGroupmsgTask",
corpId,
accessToken,
msgid,
},
db
);
const { success } = res;
if (success) {
let result = await db
.collection("groupmsg-task")
.updateOne({ _id: taskId }, { $set: { executeStatus: "end" } });
if (result.modifiedCount === 1) {
return {
success: true,
message: "停止群发成功",
};
} else {
// 如果没有文档被更新可能是任务ID错误或状态已更新
return {
success: false,
message: "任务更新失败可能任务ID无效",
};
}
} else {
return res;
}
// 获取 access_token
} catch (error) {
// 捕获异常并返回
logger.error("Error stopping WeCom group message task:", error);
return {
success: false,
message: "请求失败,无法停止群发任务",
};
}
}
// 停止群发任务
async function stopGroupmsgTask(content) {
let { id, accessToken } = content;
try {
// 使用 MongoDB 的 updateOne 更新文档
await db.collection("groupmsg-task").updateOne(
{ _id: id }, // 查询条件
{ $set: { executeStatus: "end" } } // 更新操作
);
// 使用 MongoDB 的 find 查询子任务
let data = await db
.collection("groupmsg-task")
.find({
parentId: id,
executeStatus: { $ne: "end" }, // 查询 executeStatus 不等于 "end"
})
.toArray(); // 转换为数组
if (data.length > 0) {
for (const item of data) {
const { corpId, msgid, _id: taskId } = item;
// 停止子任务
await stopWecomGroupmsgTask({
corpId,
accessToken,
msgid,
taskId,
});
}
}
return {
success: true,
message: "停止群发任务成功",
};
} catch (error) {
return {
success: false,
message: error.message || "发生错误",
};
}
}
async function getexternalMemberIdByUserId({
userId,
customers,
corpId,
accessToken,
}) {
// 获取外部联系人
let res = await weCom.main(
{
type: "getMyExternalcontact",
corpId,
accessToken,
userId: [userId],
},
db
);
// 获取外部联系人列表
let externalcontacts = res.data
? res.data.map((item) => item.external_userid)
: [];
// 获取所属客户
const fetchData = async (page, pageSize, db) => {
let data = await db
.collection("member")
.find({
_id: { $in: customers }, // 过滤客户ID
externalUserId: { $in: externalcontacts }, // 过滤外部联系人ID
})
.skip((page - 1) * pageSize) // 分页
.limit(pageSize) // 限制每次查询数量
.project({ _id: 1 }) // 只获取 _id 字段
.toArray(); // 转换为数组
return data;
};
const list = await utils.getAllData(fetchData, db);
// 返回所有符合条件的 _id
return list ? list.map((item) => item._id) : [];
}
async function createCurrentGroupMsgTask(event) {
const { corpId, accessToken } = event;
let page = 0;
let pageSize = 100;
let allData = [];
while (true) {
// 使用 MongoDB 查询语法
let data = await db
.collection("groupmsg-task")
.find({
executeStatus: "notStart",
startTaskDate: { $eq: dayjs(dayjs().format("YYYY-MM-DD")).valueOf() }, // 日期匹配
})
.skip((page - 1) * pageSize) // 分页
.limit(pageSize) // 限制每次获取的数量
.toArray(); // 获取数据并转换为数组
if (data.length > 0) {
allData.push(...data);
page++;
} else {
break;
}
}
const handler = async (item) => {
// 创建群发任务
return await createGroupMsg({ ...item, corpId, accessToken });
};
// 批量处理任务
await utils.processInBatches(allData, handler, 10);
}
async function stopExpireGroupmsgTask(content) {
const { accessToken } = content;
const fetchData = async (page, pageSize, db) => {
let data = await db
.collection("groupmsg-task")
.find({
executeStatus: "doing",
endTaskDate: { $lte: dayjs(dayjs().format("YYYY-MM-DD")).valueOf() },
})
.skip((page - 1) * pageSize)
.limit(pageSize)
.toArray(); // 使用 toArray() 将查询结果转换为数组
return data;
};
// 获取所有数据
const allData = await utils.getAllData(fetchData, db);
const handler = async (item) => {
// 停止群发任务
await stopGroupmsgTask({ accessToken, ...item });
// 更新任务状态为结束
await db.collection("groupmsg-task").updateOne(
{ _id: item._id }, // 查询条件
{ $set: { executeStatus: "end" } } // 更新操作
);
return true;
};
// 批量处理数据
const res = await utils.processInBatches(allData, handler, 10);
logger.info(res);
}
async function createGroupMsg(context) {
const { _id, accessToken, ...rest } = context;
try {
// 第一步:调用 addWecomMsgTemplateForTeamOrCorp 方法处理消息模板(如果涉及外部系统交互)
await addWecomMsgTemplateForTeamOrCorp({
id: _id,
accessToken,
params: rest,
});
// 第二步:使用 MongoDB 更新操作更新任务的执行状态
const updateResult = await db.collection("groupmsg-task").updateOne(
{ _id }, // 使用 new ObjectId() 确保 _id 类型正确
{ $set: { executeStatus: "doing" } } // 使用 $set 更新 executeStatus 字段
);
// 如果没有找到对应的任务,则输出日志
if (updateResult.matchedCount === 0) {
logger.error(`未找到匹配的任务_id: ${_id}`);
} else {
logger.info(`任务 _id: ${_id} 的状态已更新为 "doing"`);
}
} catch (error) {
// 捕获并处理可能的错误
logger.error("创建群消息过程中发生错误:", error);
}
}