存客宝应用接口初始化

This commit is contained in:
Ghost
2026-01-05 10:25:32 +08:00
parent fa37227954
commit 02797ba50a
414 changed files with 85530 additions and 75 deletions

View File

@@ -0,0 +1,57 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\AccountListJob;
class AccountListCommand extends Command
{
protected function configure()
{
$this->setName('account:list')
->setDescription('获取公司账号列表,并根据分页自动处理下一页');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理公司账号列表任务...');
try {
// 初始页码
$pageIndex = 0;
$pageSize = 100; // 每页获取100条记录
// 将第一页任务添加到队列
$this->addToQueue($pageIndex, $pageSize);
$output->writeln('公司账号列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('公司账号列表任务添加失败:' . $e->getMessage());
$output->writeln('公司账号列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
*/
protected function addToQueue($pageIndex, $pageSize)
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize
];
// 添加到队列,设置任务名为 account_list
Queue::push(AccountListJob::class, $data, 'account_list');
}
}

View File

@@ -0,0 +1,102 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use app\job\AllotChatroomJob;
use think\facade\Cache;
class AllotChatroomCommand extends Command
{
// 队列名称
protected $queueName = 'allot_chatroom';
protected function configure()
{
$this->setName('allotChatroom:run')
->setDescription('自动分配微信群聊')
->addOption('toAccountId', null, Option::VALUE_REQUIRED, '目标账号ID')
->addOption('wechatAccountKeyword', null, Option::VALUE_REQUIRED, '微信账号关键字')
->addOption('isDeleted', null, Option::VALUE_OPTIONAL, '是否已删除状态: 0=未删除(false), 1=已删除(true)', 0)
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理微信群聊自动分配任务...');
try {
// 获取命令参数
$toAccountId = $input->getOption('toAccountId');
$wechatAccountKeyword = $input->getOption('wechatAccountKeyword');
$isDeleted = $input->getOption('isDeleted');
$jobId = $input->getOption('jobId');
// 验证必填参数
if (empty($toAccountId)) {
$output->writeln('错误: 目标账号ID不能为空');
return false;
}
if (empty($wechatAccountKeyword)) {
$output->writeln('错误: 微信账号关键字不能为空');
return false;
}
$output->writeln('目标账号ID: ' . $toAccountId);
$output->writeln('微信账号关键字: ' . $wechatAccountKeyword);
$output->writeln('删除状态: ' . ($isDeleted ? '已删除' : '未删除'));
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}:{$wechatAccountKeyword}";
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中wechatAccountKeyword:{$wechatAccountKeyword},跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中wechatAccountKeyword:{$wechatAccountKeyword},跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 将任务添加到队列
$this->addToQueue($toAccountId, $wechatAccountKeyword, $isDeleted, $jobId, $queueLockKey);
$output->writeln('微信群聊自动分配任务已添加到队列');
} catch (\Exception $e) {
Log::error('微信群聊自动分配任务添加失败:' . $e->getMessage());
$output->writeln('微信群聊自动分配任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param string $toAccountId 目标账号ID
* @param string $wechatAccountKeyword 微信账号关键字
* @param bool $isDeleted 是否已删除状态
* @param string $jobId 任务ID
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($toAccountId, $wechatAccountKeyword, $isDeleted = false, $jobId = '', $queueLockKey = '')
{
$data = [
'toAccountId' => $toAccountId,
'wechatAccountKeyword' => $wechatAccountKeyword,
'isDeleted' => $isDeleted,
'jobId' => $jobId,
'queueLockKey' => $queueLockKey
];
// 添加到队列
Queue::push(AllotChatroomJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,102 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use app\job\AllotFriendJob;
use think\facade\Cache;
class AllotFriendCommand extends Command
{
// 队列名称
protected $queueName = 'allot_friends';
protected function configure()
{
$this->setName('allotFriends:run')
->setDescription('自动分配微信好友')
->addOption('toAccountId', null, Option::VALUE_REQUIRED, '目标账号ID')
->addOption('wechatAccountKeyword', null, Option::VALUE_REQUIRED, '微信账号关键字')
->addOption('isDeleted', null, Option::VALUE_OPTIONAL, '是否已删除状态: 0=未删除(false), 1=已删除(true)', 0)
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理微信好友自动分配任务...');
try {
// 获取命令参数
$toAccountId = $input->getOption('toAccountId');
$wechatAccountKeyword = $input->getOption('wechatAccountKeyword');
$isDeleted = $input->getOption('isDeleted');
$jobId = $input->getOption('jobId');
// 验证必填参数
if (empty($toAccountId)) {
$output->writeln('错误: 目标账号ID不能为空');
return false;
}
if (empty($wechatAccountKeyword)) {
$output->writeln('错误: 微信账号关键字不能为空');
return false;
}
$output->writeln('目标账号ID: ' . $toAccountId);
$output->writeln('微信账号关键字: ' . $wechatAccountKeyword);
$output->writeln('删除状态: ' . ($isDeleted ? '已删除' : '未删除'));
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}:{$wechatAccountKeyword}";
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中wechatAccountKeyword:{$wechatAccountKeyword},跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中wechatAccountKeyword:{$wechatAccountKeyword},跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 将任务添加到队列
$this->addToQueue($toAccountId, $wechatAccountKeyword, $isDeleted, $jobId, $queueLockKey);
$output->writeln('微信好友自动分配任务已添加到队列');
} catch (\Exception $e) {
Log::error('微信好友自动分配任务添加失败:' . $e->getMessage());
$output->writeln('微信好友自动分配任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param string $toAccountId 目标账号ID
* @param string $wechatAccountKeyword 微信账号关键字
* @param bool $isDeleted 是否已删除状态
* @param string $jobId 任务ID
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($toAccountId, $wechatAccountKeyword, $isDeleted = false, $jobId = '', $queueLockKey = '')
{
$data = [
'toAccountId' => $toAccountId,
'wechatAccountKeyword' => $wechatAccountKeyword,
'isDeleted' => $isDeleted,
'jobId' => $jobId,
'queueLockKey' => $queueLockKey
];
// 添加到队列
Queue::push(AllotFriendJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,50 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\AllotRuleListJob;
class AllotRuleListCommand extends Command
{
protected function configure()
{
$this->setName('allotrule:list')
->setDescription('获取分配规则列表,自动同步到数据库');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理分配规则列表任务...');
try {
// 将任务添加到队列
$this->addToQueue();
$output->writeln('分配规则列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('分配规则列表任务添加失败:' . $e->getMessage());
$output->writeln('分配规则列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
*/
protected function addToQueue()
{
$data = [
'time' => time()
];
// 添加到队列,设置任务名为 allotrule_list
Queue::push(AllotRuleListJob::class, $data, 'allotrule_list');
}
}

View File

@@ -0,0 +1,50 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\AutoCreateAllotRulesJob;
class AutoCreateAllotRulesCommand extends Command
{
protected function configure()
{
$this->setName('allotrule:autocreate')
->setDescription('自动创建微信分配规则');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理自动创建分配规则任务...');
try {
// 将任务添加到队列
$this->addToQueue();
$output->writeln('自动创建分配规则任务已添加到队列');
} catch (\Exception $e) {
Log::error('自动创建分配规则任务添加失败:' . $e->getMessage());
$output->writeln('自动创建分配规则任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
*/
protected function addToQueue()
{
$data = [
'time' => time()
];
// 添加到队列,设置任务名为 autocreate_allotrule
Queue::push(AutoCreateAllotRulesJob::class, $data, 'autocreate_allotrule');
}
}

View File

@@ -0,0 +1,554 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Db;
use think\facade\Log;
use app\common\service\WechatAccountHealthScoreService;
/**
* 统一计算微信账号健康分命令
* 一个命令完成所有评分工作:
* 1. 初始化未计算的账号(基础分只计算一次)
* 2. 更新评分记录根据wechatId和alias不一致情况
* 3. 批量更新健康分(只更新动态分)
*/
class CalculateWechatAccountScoreCommand extends Command
{
/**
* 数据库表名
*/
const TABLE_WECHAT_ACCOUNT = 's2_wechat_account';
const TABLE_WECHAT_ACCOUNT_SCORE = 's2_wechat_account_score';
protected function configure()
{
$this->setName('wechat:calculate-score')
->setDescription('统一计算微信账号健康分(包含初始化、更新评分记录、批量计算)')
->addOption('only-init', null, \think\console\input\Option::VALUE_NONE, '仅执行初始化步骤')
->addOption('only-update', null, \think\console\input\Option::VALUE_NONE, '仅执行更新评分记录步骤')
->addOption('only-batch', null, \think\console\input\Option::VALUE_NONE, '仅执行批量更新健康分步骤')
->addOption('account-id', 'a', \think\console\input\Option::VALUE_OPTIONAL, '指定账号ID仅处理该账号')
->addOption('batch-size', 'b', \think\console\input\Option::VALUE_OPTIONAL, '批处理大小', 50)
->addOption('force-recalculate', 'f', \think\console\input\Option::VALUE_NONE, '强制重新计算基础分');
}
/**
* 执行命令
*
* @param Input $input 输入对象
* @param Output $output 输出对象
* @return int 命令执行状态码0表示成功
*/
protected function execute(Input $input, Output $output)
{
// 解析命令行参数
$onlyInit = $input->getOption('only-init');
$onlyUpdate = $input->getOption('only-update');
$onlyBatch = $input->getOption('only-batch');
$accountId = $input->getOption('account-id');
$batchSize = (int)$input->getOption('batch-size');
$forceRecalculate = $input->getOption('force-recalculate');
// 参数验证
if ($batchSize <= 0) {
$batchSize = 50; // 默认批处理大小
}
// 显示执行参数
$output->writeln("==========================================");
$output->writeln("开始统一计算微信账号健康分...");
$output->writeln("==========================================");
if ($accountId) {
$output->writeln("指定账号ID: {$accountId}");
}
if ($onlyInit) {
$output->writeln("仅执行初始化步骤");
} elseif ($onlyUpdate) {
$output->writeln("仅执行更新评分记录步骤");
} elseif ($onlyBatch) {
$output->writeln("仅执行批量更新健康分步骤");
}
if ($forceRecalculate) {
$output->writeln("强制重新计算基础分");
}
$output->writeln("批处理大小: {$batchSize}");
// 记录命令开始执行的日志(仅在非交互模式下记录)
if (!$output->isVerbose()) {
Log::info('开始执行微信账号健康分计算命令', [
'accountId' => $accountId,
'onlyInit' => $onlyInit ? 'true' : 'false',
'onlyUpdate' => $onlyUpdate ? 'true' : 'false',
'onlyBatch' => $onlyBatch ? 'true' : 'false',
'batchSize' => $batchSize,
'forceRecalculate' => $forceRecalculate ? 'true' : 'false'
]);
$startTime = time();
try {
// 实例化服务
$service = new WechatAccountHealthScoreService();
} catch (\Exception $e) {
$errorMsg = "实例化WechatAccountHealthScoreService失败: " . $e->getMessage();
$output->writeln("<error>{$errorMsg}</error>");
Log::error($errorMsg);
return 1; // 返回非零状态码表示失败
}
// 初始化统计数据
$initStats = ['success' => 0, 'failed' => 0, 'errors' => []];
$updateStats = ['total' => 0];
$batchStats = ['success' => 0, 'failed' => 0, 'errors' => []];
try {
// 步骤1: 初始化未计算基础分的账号
if (!$onlyUpdate && !$onlyBatch) {
$output->writeln("\n[步骤1] 初始化未计算基础分的账号...");
$initStats = $this->initUncalculatedAccounts($service, $output, $accountId, $batchSize);
$output->writeln("初始化完成:成功 {$initStats['success']} 条,失败 {$initStats['failed']}");
}
// 步骤2: 更新评分记录根据wechatId和alias不一致情况
if (!$onlyInit && !$onlyBatch) {
$output->writeln("\n[步骤2] 更新评分记录根据wechatId和alias不一致情况...");
$updateStats = $this->updateScoreRecords($service, $output, $accountId, $batchSize);
$output->writeln("更新完成:处理了 {$updateStats['total']} 条记录");
}
// 步骤3: 批量更新健康分(只更新动态分,不重新计算基础分)
if (!$onlyInit && !$onlyUpdate) {
$output->writeln("\n[步骤3] 批量更新健康分(只更新动态分)...");
$batchStats = $this->batchUpdateHealthScore($service, $output, $accountId, $batchSize, $forceRecalculate);
$output->writeln("批量更新完成:成功 {$batchStats['success']} 条,失败 {$batchStats['failed']}");
}
// 统计信息
$endTime = time();
$duration = $endTime - $startTime;
$output->writeln("\n==========================================");
$output->writeln("任务完成!");
$output->writeln("==========================================");
$output->writeln("总耗时: {$duration}");
$output->writeln("初始化: 成功 {$initStats['success']} 条,失败 {$initStats['failed']}");
$output->writeln("更新评分记录: {$updateStats['total']}");
$output->writeln("批量更新: 成功 {$batchStats['success']} 条,失败 {$batchStats['failed']}");
// 记录命令执行完成的日志
Log::info("微信账号健康分计算命令执行完成,总耗时: {$duration} 秒," .
"初始化: 成功 {$initStats['success']} 条,失败 {$initStats['failed']} 条," .
"更新评分记录: {$updateStats['total']} 条," .
"批量更新: 成功 {$batchStats['success']} 条,失败 {$batchStats['failed']}");
if (!empty($initStats['errors'])) {
$output->writeln("\n初始化错误详情:");
Log::warning("初始化阶段出现 " . count($initStats['errors']) . " 个错误");
foreach (array_slice($initStats['errors'], 0, 10) as $error) {
$output->writeln(" 账号ID {$error['accountId']}: {$error['error']}");
Log::error("初始化错误 - 账号ID {$error['accountId']}: {$error['error']}");
}
if (count($initStats['errors']) > 10) {
$output->writeln(" ... 还有 " . (count($initStats['errors']) - 10) . " 个错误");
Log::warning("初始化错误过多只记录前10个还有 " . (count($initStats['errors']) - 10) . " 个错误未显示");
}
}
if (!empty($batchStats['errors'])) {
$output->writeln("\n批量更新错误详情:");
Log::warning("批量更新阶段出现 " . count($batchStats['errors']) . " 个错误");
foreach (array_slice($batchStats['errors'], 0, 10) as $error) {
$output->writeln(" 账号ID {$error['accountId']}: {$error['error']}");
Log::error("批量更新错误 - 账号ID {$error['accountId']}: {$error['error']}");
}
if (count($batchStats['errors']) > 10) {
$output->writeln(" ... 还有 " . (count($batchStats['errors']) - 10) . " 个错误");
Log::warning("批量更新错误过多只记录前10个还有 " . (count($batchStats['errors']) - 10) . " 个错误未显示");
}
}
} catch (\PDOException $e) {
// 数据库异常
$errorMsg = "数据库操作失败: " . $e->getMessage();
$output->writeln("\n<error>数据库错误: " . $errorMsg . "</error>");
$output->writeln($e->getTraceAsString());
// 记录数据库错误
Log::error("数据库错误: " . $errorMsg);
Log::error("错误堆栈: " . $e->getTraceAsString());
return 2; // 数据库错误状态码
} catch (\Exception $e) {
// 一般异常
$errorMsg = "命令执行失败: " . $e->getMessage();
$output->writeln("\n<error>错误: " . $errorMsg . "</error>");
$output->writeln($e->getTraceAsString());
// 记录严重错误
Log::error($errorMsg);
Log::error("错误堆栈: " . $e->getTraceAsString());
return 1; // 一般错误状态码
} catch (\Throwable $e) {
// 其他所有错误
$errorMsg = "严重错误: " . $e->getMessage();
$output->writeln("\n<error>严重错误: " . $errorMsg . "</error>");
$output->writeln($e->getTraceAsString());
// 记录严重错误
Log::critical($errorMsg);
Log::critical("错误堆栈: " . $e->getTraceAsString());
return 3; // 严重错误状态码
}
return 0; // 成功执行
}
}
/**
* 初始化未计算基础分的账号
*
* @param WechatAccountHealthScoreService $service 健康分服务实例
* @param Output $output 输出对象
* @return array 处理结果统计
* @throws \Exception 如果查询或处理过程中出现错误
*/
private function initUncalculatedAccounts($service, $output, $accountId = null, $batchSize = 50)
{
$stats = [
'total' => 0,
'success' => 0,
'failed' => 0,
'errors' => []
];
try {
// 获取所有未计算基础分的账号
// 优化查询:使用索引字段,只查询必要的字段
$query = Db::table(self::TABLE_WECHAT_ACCOUNT)
->alias('a')
->leftJoin([self::TABLE_WECHAT_ACCOUNT_SCORE => 's'], 's.accountId = a.id')
->where('a.isDeleted', 0)
->where(function($query) {
$query->whereNull('s.id')
->whereOr('s.baseScoreCalculated', 0);
});
// 如果指定了账号ID则只处理该账号
if ($accountId) {
$query->where('a.id', $accountId);
}
$accounts = $query->field('a.id, a.wechatId') // 只查询必要的字段
->select();
} catch (\Exception $e) {
Log::error("查询未计算基础分的账号失败: " . $e->getMessage());
throw new \Exception("查询未计算基础分的账号失败: " . $e->getMessage(), 0, $e);
}
$stats['total'] = count($accounts);
if ($stats['total'] == 0) {
$output->writeln("没有需要初始化的账号");
Log::info("没有需要初始化的账号");
return $stats;
}
$output->writeln("找到 {$stats['total']} 个需要初始化的账号");
Log::info("找到 {$stats['total']} 个需要初始化的账号");
// 优化批处理:使用传入的批处理大小
$batches = array_chunk($accounts, $batchSize);
$batchCount = count($batches);
Log::info("将分 {$batchCount} 批处理,每批 {$batchSize} 个账号");
foreach ($batches as $batchIndex => $batch) {
$batchStartTime = microtime(true);
$batchSuccessCount = 0;
$batchFailedCount = 0;
foreach ($batch as $account) {
try {
$service->calculateAndUpdate($account['id']);
$stats['success']++;
$batchSuccessCount++;
if ($stats['success'] % 20 == 0) { // 更频繁地显示进度
$output->write(".");
Log::debug("已成功初始化 {$stats['success']} 个账号");
}
} catch (\Exception $e) {
$stats['failed']++;
$batchFailedCount++;
$errorMsg = "初始化账号 {$account['id']} 失败: " . $e->getMessage();
Log::error($errorMsg);
$stats['errors'][] = [
'accountId' => $account['id'],
'error' => $e->getMessage()
];
}
}
$batchEndTime = microtime(true);
$batchDuration = round($batchEndTime - $batchStartTime, 2);
// 每批次完成后输出进度信息
$output->writeln(" 批次 " . ($batchIndex + 1) . "/{$batchCount} 完成,耗时 {$batchDuration} 秒,成功 {$batchSuccessCount},失败 {$batchFailedCount}");
Log::info("初始化批次 " . ($batchIndex + 1) . "/{$batchCount} 完成,耗时 {$batchDuration} 秒,成功 {$batchSuccessCount},失败 {$batchFailedCount}");
}
return $stats;
}
/**
* 更新评分记录根据wechatId和alias不一致情况
*
* @param WechatAccountHealthScoreService $service 健康分服务实例
* @param Output $output 输出对象
* @return array 处理结果统计
* @throws \Exception 如果查询或处理过程中出现错误
*/
private function updateScoreRecords($service, $output, $accountId = null, $batchSize = 50)
{
$stats = ['total' => 0];
try {
// 优化查询:合并两次查询为一次,减少数据库访问次数
$query = Db::table(self::TABLE_WECHAT_ACCOUNT)
->where('isDeleted', 0)
->where('wechatId', '<>', '')
->where('alias', '<>', '');
// 如果指定了账号ID则只处理该账号
if ($accountId) {
$query->where('id', $accountId);
}
$accounts = $query->field('id, wechatId, alias, IF(wechatId = alias, 0, 1) as isModifiedAlias')
->select();
// 分类处理查询结果
$inconsistentAccounts = [];
$consistentAccounts = [];
foreach ($accounts as $account) {
if ($account['isModifiedAlias'] == 1) {
$inconsistentAccounts[] = $account;
} else {
$consistentAccounts[] = $account;
}
}
} catch (\Exception $e) {
Log::error("查询需要更新评分记录的账号失败: " . $e->getMessage());
throw new \Exception("查询需要更新评分记录的账号失败: " . $e->getMessage(), 0, $e);
}
$allAccounts = array_merge($inconsistentAccounts, $consistentAccounts);
$stats['total'] = count($allAccounts);
if ($stats['total'] == 0) {
$output->writeln("没有需要更新的账号");
Log::info("没有需要更新的评分记录");
return $stats;
}
$output->writeln("找到 {$stats['total']} 个需要更新的账号(不一致: " . count($inconsistentAccounts) . ",一致: " . count($consistentAccounts) . "");
Log::info("找到 {$stats['total']} 个需要更新的账号(不一致: " . count($inconsistentAccounts) . ",一致: " . count($consistentAccounts) . "");
$updatedCount = 0;
// 优化批处理:使用传入的批处理大小
$batches = array_chunk($allAccounts, $batchSize);
$batchCount = count($batches);
Log::info("将分 {$batchCount} 批更新评分记录,每批 {$batchSize} 个账号");
foreach ($batches as $batchIndex => $batch) {
$batchStartTime = microtime(true);
$batchUpdatedCount = 0;
foreach ($batch as $account) {
$isModifiedAlias = isset($account['isModifiedAlias']) ?
($account['isModifiedAlias'] == 1) :
in_array($account['id'], array_column($inconsistentAccounts, 'id'));
$this->updateScoreRecord($account['id'], $isModifiedAlias, $service);
$updatedCount++;
$batchUpdatedCount++;
if ($batchUpdatedCount % 20 == 0) {
$output->write(".");
}
}
$batchEndTime = microtime(true);
$batchDuration = round($batchEndTime - $batchStartTime, 2);
// 每批次完成后输出进度信息
$output->writeln(" 批次 " . ($batchIndex + 1) . "/{$batchCount} 完成,耗时 {$batchDuration} 秒,更新 {$batchUpdatedCount} 条记录");
Log::info("更新评分记录批次 " . ($batchIndex + 1) . "/{$batchCount} 完成,耗时 {$batchDuration} 秒,更新 {$batchUpdatedCount} 条记录");
}
if ($updatedCount > 0 && $updatedCount % 100 == 0) {
$output->writeln("");
}
return $stats;
}
/**
* 批量更新健康分(只更新动态分)
*
* @param WechatAccountHealthScoreService $service 健康分服务实例
* @param Output $output 输出对象
* @return array 处理结果统计
* @throws \Exception 如果查询或处理过程中出现错误
*/
private function batchUpdateHealthScore($service, $output, $accountId = null, $batchSize = 50, $forceRecalculate = false)
{
try {
// 获取所有已计算基础分的账号
// 优化查询:只查询必要的字段,使用索引字段
$query = Db::table(self::TABLE_WECHAT_ACCOUNT_SCORE)
->where('baseScoreCalculated', 1);
// 如果指定了账号ID则只处理该账号
if ($accountId) {
$query->where('accountId', $accountId);
}
$accountIds = $query->column('accountId');
} catch (\Exception $e) {
Log::error("查询需要批量更新健康分的账号失败: " . $e->getMessage());
throw new \Exception("查询需要批量更新健康分的账号失败: " . $e->getMessage(), 0, $e);
}
$total = count($accountIds);
if ($total == 0) {
$output->writeln("没有需要更新的账号");
Log::info("没有需要批量更新健康分的账号");
return ['success' => 0, 'failed' => 0, 'errors' => []];
}
$output->writeln("找到 {$total} 个需要更新动态分的账号");
Log::info("找到 {$total} 个需要更新动态分的账号");
// 使用传入的批处理大小和强制重新计算标志
Log::info("使用批量大小 {$batchSize} 进行批量更新健康分,强制重新计算基础分: " . ($forceRecalculate ? 'true' : 'false'));
$stats = $service->batchCalculateAndUpdate($accountIds, $batchSize, $forceRecalculate);
return $stats;
}
/**
* 更新评分记录
*
* @param int $accountId 账号ID
* @param bool $isModifiedAlias 是否已修改微信号
* @param WechatAccountHealthScoreService $service 评分服务
*/
/**
* 更新评分记录
*
* @param int $accountId 账号ID
* @param bool $isModifiedAlias 是否已修改微信号
* @param WechatAccountHealthScoreService $service 评分服务
* @return bool 是否成功更新
*/
private function updateScoreRecord($accountId, $isModifiedAlias, $service)
{
Log::debug("开始更新账号 {$accountId} 的评分记录isModifiedAlias: " . ($isModifiedAlias ? 'true' : 'false'));
try {
// 获取账号数据 - 只查询必要的字段
$accountData = Db::table(self::TABLE_WECHAT_ACCOUNT)
->where('id', $accountId)
->field('id, wechatId, alias') // 只查询必要的字段
->find();
if (empty($accountData)) {
Log::warning("账号 {$accountId} 不存在,跳过更新评分记录");
return false;
}
// 确保评分记录存在 - 只查询必要的字段
$scoreRecord = Db::table(self::TABLE_WECHAT_ACCOUNT_SCORE)
->where('accountId', $accountId)
->field('accountId, baseScore, baseScoreCalculated, baseInfoScore, dynamicScore') // 只查询必要的字段
->find();
if (empty($scoreRecord)) {
// 如果记录不存在,创建并计算基础分
Log::info("账号 {$accountId} 的评分记录不存在,创建并计算基础分");
$service->calculateAndUpdate($accountId);
$scoreRecord = Db::table(self::TABLE_WECHAT_ACCOUNT_SCORE)
->where('accountId', $accountId)
->find();
}
if (empty($scoreRecord)) {
Log::warning("账号 {$accountId} 的评分记录创建失败,跳过更新");
return;
}
// 更新isModifiedAlias字段
$updateData = [
'isModifiedAlias' => $isModifiedAlias ? 1 : 0,
'updateTime' => time()
];
// 如果基础分已计算,需要更新基础信息分和基础分
if ($scoreRecord['baseScoreCalculated']) {
$oldBaseInfoScore = $scoreRecord['baseInfoScore'] ?? 0;
$newBaseInfoScore = $isModifiedAlias ? 10 : 0; // 已修改微信号得10分
if ($oldBaseInfoScore != $newBaseInfoScore) {
$oldBaseScore = $scoreRecord['baseScore'] ?? 60;
$newBaseScore = $oldBaseScore - $oldBaseInfoScore + $newBaseInfoScore;
$updateData['baseInfoScore'] = $newBaseInfoScore;
$updateData['baseScore'] = $newBaseScore;
// 重新计算健康分
$dynamicScore = $scoreRecord['dynamicScore'] ?? 0;
$healthScore = $newBaseScore + $dynamicScore;
$healthScore = max(0, min(100, $healthScore));
$updateData['healthScore'] = $healthScore;
$updateData['maxAddFriendPerDay'] = (int)floor($healthScore * 0.2);
Log::info("账号 {$accountId} 的基础信息分从 {$oldBaseInfoScore} 更新为 {$newBaseInfoScore}" .
"基础分从 {$oldBaseScore} 更新为 {$newBaseScore},健康分更新为 {$healthScore}");
}
} else {
// 基础分未计算,只更新标记和基础信息分
$updateData['baseInfoScore'] = $isModifiedAlias ? 10 : 0;
}
$result = Db::table(self::TABLE_WECHAT_ACCOUNT_SCORE)
->where('accountId', $accountId)
->update($updateData);
Log::debug("账号 {$accountId} 的评分记录更新" . ($result !== false ? "成功" : "失败"));
return $result !== false;
} catch (\Exception $e) {
Log::error("更新账号 {$accountId} 的评分记录失败: " . $e->getMessage());
return false;
}
}
}

View File

@@ -0,0 +1,57 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\CallRecordingListJob;
class CallRecordingListCommand extends Command
{
protected function configure()
{
$this->setName('call-recording:list')
->setDescription('获取通话记录列表,并根据分页自动处理下一页');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理通话记录列表任务...');
try {
// 初始页码
$pageIndex = 0;
$pageSize = 100; // 每页获取100条记录
// 将第一页任务添加到队列
$this->addToQueue($pageIndex, $pageSize);
$output->writeln('通话记录列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('通话记录列表任务添加失败:' . $e->getMessage());
$output->writeln('通话记录列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
*/
protected function addToQueue($pageIndex, $pageSize)
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize
];
// 添加到队列,设置任务名为 call_recording_list
Queue::push(CallRecordingListJob::class, $data, 'call_recording_list');
}
}

View File

@@ -0,0 +1,100 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\Db;
class CleanExpiredGroupMessages extends Command
{
protected function configure()
{
$this->setName('clean:expired_group_messages')
->setDescription('Clean expired group messages from the database')
->addOption('days', 'd', Option::VALUE_OPTIONAL, 'Number of days to keep messages (default: 90)', 90)
->addOption('dry-run', null, Option::VALUE_NONE, 'Perform a dry run without deleting any data')
->addOption('batch-size', 'b', Option::VALUE_OPTIONAL, 'Batch size for deletion (default: 1000)', 1000);
}
protected function execute(Input $input, Output $output)
{
$days = (int)$input->getOption('days');
$dryRun = $input->getOption('dry-run');
$batchSize = (int)$input->getOption('batch-size');
if ($dryRun) {
$output->writeln("<info>Running in dry-run mode. No data will be deleted.</info>");
}
$cutoffDate = date('Y-m-d H:i:s', strtotime("-{$days} days"));
$output->writeln("<info>Cleaning group messages older than {$cutoffDate} (keeping last {$days} days)</info>");
// 清理微信群组消息
$this->cleanWechatGroupMessages($cutoffDate, $dryRun, $batchSize, $output);
$output->writeln("<info>Group message cleanup completed successfully.</info>");
}
protected function cleanWechatGroupMessages($cutoffDate, $dryRun, $batchSize, Output $output)
{
$output->writeln("\nCleaning s2_wechat_group_message table...");
// 获取符合条件的消息总数
$totalCount = Db::table('s2_wechat_group_message')
->where('createTime', '<', $cutoffDate)
->count();
if ($totalCount === 0) {
$output->writeln(" <comment>No expired group messages found.</comment>");
return;
}
$output->writeln(" Found {$totalCount} group messages to clean up.");
if ($dryRun) {
$output->writeln(" <comment>Dry run mode: would delete {$totalCount} group messages.</comment>");
return;
}
// 计算需要执行的批次数
$batches = ceil($totalCount / $batchSize);
$deletedCount = 0;
$output->writeln(" Deleting in {$batches} batches of {$batchSize} records...");
// 分批删除数据
for ($i = 0; $i < $batches; $i++) {
// 获取一批要删除的ID
$ids = Db::table('s2_wechat_group_message')
->where('createTime', '<', $cutoffDate)
->limit($batchSize)
->column('id');
if (empty($ids)) {
break;
}
// 删除这批数据
$count = Db::table('s2_wechat_group_message')
->whereIn('id', $ids)
->delete();
$deletedCount += $count;
$progress = round(($deletedCount / $totalCount) * 100, 2);
$output->write(" Progress: {$progress}% ({$deletedCount}/{$totalCount})\r");
// 短暂暂停,减轻数据库负担
usleep(500000); // 暂停0.5秒
}
$output->writeln("");
$output->writeln(" <info>Successfully deleted {$deletedCount} expired group messages.</info>");
// 优化表
$output->writeln(" Optimizing table...");
Db::execute("OPTIMIZE TABLE s2_wechat_group_message");
$output->writeln(" <info>Table optimization completed.</info>");
}
}

View File

@@ -0,0 +1,100 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\Db;
class CleanExpiredMessages extends Command
{
protected function configure()
{
$this->setName('clean:expired_messages')
->setDescription('Clean expired messages from the database')
->addOption('days', 'd', Option::VALUE_OPTIONAL, 'Number of days to keep messages (default: 90)', 90)
->addOption('dry-run', null, Option::VALUE_NONE, 'Perform a dry run without deleting any data')
->addOption('batch-size', 'b', Option::VALUE_OPTIONAL, 'Batch size for deletion (default: 1000)', 1000);
}
protected function execute(Input $input, Output $output)
{
$days = (int)$input->getOption('days');
$dryRun = $input->getOption('dry-run');
$batchSize = (int)$input->getOption('batch-size');
if ($dryRun) {
$output->writeln("<info>Running in dry-run mode. No data will be deleted.</info>");
}
$cutoffDate = date('Y-m-d H:i:s', strtotime("-{$days} days"));
$output->writeln("<info>Cleaning messages older than {$cutoffDate} (keeping last {$days} days)</info>");
// 清理微信消息
$this->cleanWechatMessages($cutoffDate, $dryRun, $batchSize, $output);
$output->writeln("<info>Message cleanup completed successfully.</info>");
}
protected function cleanWechatMessages($cutoffDate, $dryRun, $batchSize, Output $output)
{
$output->writeln("\nCleaning s2_wechat_message table...");
// 获取符合条件的消息总数
$totalCount = Db::table('s2_wechat_message')
->where('createTime', '<', $cutoffDate)
->count();
if ($totalCount === 0) {
$output->writeln(" <comment>No expired messages found.</comment>");
return;
}
$output->writeln(" Found {$totalCount} messages to clean up.");
if ($dryRun) {
$output->writeln(" <comment>Dry run mode: would delete {$totalCount} messages.</comment>");
return;
}
// 计算需要执行的批次数
$batches = ceil($totalCount / $batchSize);
$deletedCount = 0;
$output->writeln(" Deleting in {$batches} batches of {$batchSize} records...");
// 分批删除数据
for ($i = 0; $i < $batches; $i++) {
// 获取一批要删除的ID
$ids = Db::table('s2_wechat_message')
->where('createTime', '<', $cutoffDate)
->limit($batchSize)
->column('id');
if (empty($ids)) {
break;
}
// 删除这批数据
$count = Db::table('s2_wechat_message')
->whereIn('id', $ids)
->delete();
$deletedCount += $count;
$progress = round(($deletedCount / $totalCount) * 100, 2);
$output->write(" Progress: {$progress}% ({$deletedCount}/{$totalCount})\r");
// 短暂暂停,减轻数据库负担
usleep(500000); // 暂停0.5秒
}
$output->writeln("");
$output->writeln(" <info>Successfully deleted {$deletedCount} expired messages.</info>");
// 优化表
$output->writeln(" Optimizing table...");
Db::execute("OPTIMIZE TABLE s2_wechat_message");
$output->writeln(" <info>Table optimization completed.</info>");
}
}

View File

@@ -0,0 +1,51 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\ContentCollectJob;
class ContentCollectCommand extends Command
{
protected function configure()
{
$this->setName('content:collect')
->setDescription('执行内容采集任务');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理内容采集任务...');
try {
// 将任务添加到队列
$this->addToQueue();
$output->writeln('内容采集任务已添加到队列');
} catch (\Exception $e) {
Log::error('内容采集任务添加失败:' . $e->getMessage());
$output->writeln('内容采集任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
*/
protected function addToQueue()
{
$data = [
'libraryId' => 0, // 0表示采集所有内容库
'timestamp' => time()
];
// 添加到队列,设置任务名为 content_collect
Queue::push(ContentCollectJob::class, $data, 'content_collect');
}
}

View File

@@ -0,0 +1,57 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\DepartmentListJob;
class DepartmentListCommand extends Command
{
protected function configure()
{
$this->setName('department:list')
->setDescription('获取部门列表,并根据分页自动处理下一页');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理部门列表任务...');
try {
// 初始页码
$pageIndex = 0;
$pageSize = 100; // 每页获取100条记录
// 将第一页任务添加到队列
$this->addToQueue($pageIndex, $pageSize);
$output->writeln('部门列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('部门列表任务添加失败:' . $e->getMessage());
$output->writeln('部门列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
*/
protected function addToQueue($pageIndex, $pageSize)
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize
];
// 添加到队列,设置任务名为 account_list
Queue::push(DepartmentListJob::class, $data, 'department_list');
}
}

View File

@@ -0,0 +1,98 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use app\job\DeviceListJob;
use think\facade\Cache;
class DeviceListCommand extends Command
{
// 队列名称
protected $queueName = 'device_list';
protected function configure()
{
$this->setName('device:list')
->setDescription('获取设备列表,并根据分页自动处理下一页')
->addOption('isDel', null, Option::VALUE_OPTIONAL, '删除状态: 0=未删除(unDeleted), 1=已删除(deleted), 2=已停用(deletedAndStop)', '')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理设备列表任务...');
try {
// 获取是否删除参数和任务ID
$isDel = $input->getOption('isDel');
$jobId = $input->getOption('jobId');
$output->writeln('删除状态参数: ' . ($isDel === '' ? '全部' : ($isDel == 0 ? '未删除' : ($isDel == 1 ? '已删除' : '已停用'))));
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}:{$isDel}";
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,删除状态:{$isDel},跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,删除状态:{$isDel},跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 为不同的删除状态和任务ID使用不同的缓存键名
$cacheKeyPrefix = "devicePage:{$jobId}";
$cacheKeySuffix = $isDel === '' ? '' : ":{$isDel}";
$cacheKey = $cacheKeyPrefix . $cacheKeySuffix;
// 从缓存获取初始页码缓存有效期1天
$pageIndex = Cache::get($cacheKey, 0);
$output->writeln("从缓存获取页码: {$pageIndex}, 缓存键: {$cacheKey}");
$pageSize = 100; // 每页获取100条记录
// 将任务添加到队列
$this->addToQueue($pageIndex, $pageSize, $isDel, $jobId, $cacheKey, $queueLockKey);
$output->writeln('设备列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('设备列表任务添加失败:' . $e->getMessage());
$output->writeln('设备列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
* @param string $isDel 删除状态
* @param string $jobId 任务ID
* @param string $cacheKey 缓存键名
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($pageIndex, $pageSize, $isDel = '', $jobId = '', $cacheKey = '', $queueLockKey = '')
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize,
'isDel' => $isDel,
'jobId' => $jobId,
'cacheKey' => $cacheKey,
'queueLockKey' => $queueLockKey
];
// 添加到队列,设置任务名为 device_list
Queue::push(DeviceListJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,60 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\FriendTaskJob;
use think\facade\Cache;
class FriendTaskCommand extends Command
{
protected function configure()
{
$this->setName('friend:task')
->setDescription('获取添加好友认为列表,并根据分页自动处理下一页');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理添加好友任务...');
try {
// 从缓存获取初始页码缓存10分钟有效
$pageIndex = Cache::get('friendTaskPage', 0);
$output->writeln('从缓存获取页码:' . $pageIndex);
$pageSize = 1000; // 每页获取1000条记录
// 将任务添加到队列
$this->addToQueue($pageIndex, $pageSize);
$output->writeln('添加好友任务已添加到队列');
} catch (\Exception $e) {
Log::error('添加好友任务添加失败:' . $e->getMessage());
$output->writeln('添加好友任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
*/
protected function addToQueue($pageIndex, $pageSize)
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize
];
// 添加到队列,设置任务名为 friend_task
Queue::push(FriendTaskJob::class, $data, 'friend_task');
}
}

View File

@@ -0,0 +1,60 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\GroupFriendsJob;
use think\facade\Cache;
class GroupFriendsCommand extends Command
{
protected function configure()
{
$this->setName('groupFriends:list')
->setDescription('获取微信群好友列表,并根据分页自动处理下一页');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理微信群好友列表任务...');
try {
// 从缓存获取初始页码,缓存有效期一天
$pageIndex = Cache::get('groupFriendsPage', 0);
$output->writeln('从缓存获取页码:' . $pageIndex);
$pageSize = 100; // 每页获取100条记录
// 将任务添加到队列
$this->addToQueue($pageIndex, $pageSize);
$output->writeln('微信群好友列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('微信群好友列表任务添加失败:' . $e->getMessage());
$output->writeln('微信群好友列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
*/
protected function addToQueue($pageIndex, $pageSize)
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize
];
// 添加到队列,设置任务名为 group_friends
Queue::push(GroupFriendsJob::class, $data, 'group_friends');
}
}

View File

@@ -0,0 +1,50 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Db;
use think\facade\Config;
class InitDatabase extends Command
{
protected function configure()
{
$this->setName('init:database')
->setDescription('初始化数据库,创建必要的表结构');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始初始化数据库...');
try {
// 读取SQL文件
$sqlFile = app()->getAppPath() . 'common/database/tk_users.sql';
if (!file_exists($sqlFile)) {
$output->error('SQL文件不存在: ' . $sqlFile);
return;
}
$sql = file_get_contents($sqlFile);
// 分割SQL语句
$sqlArr = explode(';', $sql);
// 执行SQL语句
foreach ($sqlArr as $statement) {
$statement = trim($statement);
if ($statement) {
Db::execute($statement);
$output->writeln('执行SQL: ' . mb_substr($statement, 0, 100) . '...');
}
}
$output->info('数据库初始化完成!');
} catch (\Exception $e) {
$output->error('数据库初始化失败: ' . $e->getMessage());
}
}
}

View File

@@ -0,0 +1,118 @@
<?php
namespace app\command;
use app\chukebao\model\FollowUp;
use app\chukebao\model\NoticeModel;
use app\chukebao\model\ToDo;
use library\ResponseHelper;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\Db;
use think\facade\Log;
use think\Queue;
use app\job\AllotFriendJob;
use think\facade\Cache;
class KfNoticeCommand extends Command
{
// 队列名称
protected $queueName = 'kf_notice';
protected function configure()
{
$this->setName('kfNotice:run')
->setDescription('消息通知');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理消息通知任务...');
$where = [
['isRemind', '=', 0],
['reminderTime', '<=', time()],
];
$notice = [];
Db::startTrans();
try {
$followUp = FollowUp::where($where)->alias('a')
->field('a.*,f.nickname,f.avatar,f.alias,f.wechatId')
->join(['s2_wechat_friend f'], 'a.friendId = f.id')
->where('a.isRemind',0)
->select();
if (!empty($followUp)) {
foreach ($followUp as $k => $v) {
switch ($v['type']) {
case 1:
$title = '电话回访';
break;
case 2:
$title = '发送消息';
break;
case 3:
$title = '安排会议';
break;
case 4:
$title = '发送邮件';
break;
default:
$title = '其他';
break;
}
$wechatId = !empty($v['alias']) ? $v['alias'] : $v['wechatId'];
$nickname = $v['nickname'] . '(' . $wechatId . ')';
$message = $nickname . '' . $v['description'];
$notice[] = [
'type' => 2,
'userId' => $v['userId'],
'companyId' => $v['companyId'],
'bindId' => $v['id'],
'title' => $title,
'message' => $message,
'createTime' => $v['reminderTime'],
];
}
FollowUp::where($where)->update(['isRemind' => 1]);
}
$toDo = ToDo::where($where)->alias('a')
->field('a.*,f.nickname,f.avatar,f.alias,f.wechatId')
->join(['s2_wechat_friend f'], 'a.friendId = f.id')
->where('a.isRemind',0)
->select();
if (!empty($toDo)) {
foreach ($toDo as $k => $v) {
$wechatId = !empty($v['alias']) ? $v['alias'] : $v['wechatId'];
$nickname = $v['nickname'] . '(' . $wechatId . ')';
$message = $nickname . '' . $v['description'];
$notice[] = [
'type' => 1,
'userId' => $v['userId'],
'companyId' => $v['companyId'],
'bindId' => $v['id'],
'title' => $v['title'],
'message' => $message,
'createTime' => $v['reminderTime'],
];
}
ToDo::where($where)->update(['isRemind' => 1]);
}
$noticeModel = new NoticeModel();
$noticeModel->insertAll($notice);
Db::commit();
return true;
} catch (\Exception $e) {
Db::rollback();
return false;
}
}
}

View File

@@ -0,0 +1,57 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\MessageChatroomListJob;
class MessageChatroomListCommand extends Command
{
protected function configure()
{
$this->setName('message:chatroomList')
->setDescription('获取微信群聊消息列表,并根据分页自动处理下一页');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理微信群聊消息列表...');
try {
// 初始页码
$pageIndex = 0;
$pageSize = 100; // 每页获取100条记录
// 将第一页任务添加到队列
$this->addToQueue($pageIndex, $pageSize);
$output->writeln('微信群聊消息列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('微信群聊消息列表任务添加失败:' . $e->getMessage());
$output->writeln('微信群聊消息列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
*/
protected function addToQueue($pageIndex, $pageSize)
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize
];
// 添加到队列,设置任务名为 friend_task
Queue::push(MessageChatroomListJob::class, $data, 'message_chatroom_list');
}
}

View File

@@ -0,0 +1,57 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\MessageFriendsListJob;
class MessageFriendsListCommand extends Command
{
protected function configure()
{
$this->setName('message:friendsList')
->setDescription('获取好友消息列表,并根据分页自动处理下一页');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理好友消息列表...');
try {
// 初始页码
$pageIndex = 0;
$pageSize = 100; // 每页获取100条记录
// 将第一页任务添加到队列
$this->addToQueue($pageIndex, $pageSize);
$output->writeln('好友消息列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('好友消息列表任务添加失败:' . $e->getMessage());
$output->writeln('好友消息列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
*/
protected function addToQueue($pageIndex, $pageSize)
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize
];
// 添加到队列,设置任务名为 friend_task
Queue::push(MessageFriendsListJob::class, $data, 'message_friends_list');
}
}

View File

@@ -0,0 +1,112 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Db;
class OptimizeMessageIndexes extends Command
{
protected function configure()
{
$this->setName('optimize:message_indexes')
->setDescription('Optimize database indexes for message-related tables');
}
protected function execute(Input $input, Output $output)
{
$output->writeln("Starting index optimization for message-related tables...");
// 优化 s2_wechat_message 表索引
$this->optimizeWechatMessageIndexes($output);
// 优化 s2_wechat_chatroom 表索引
$this->optimizeWechatChatroomIndexes($output);
// 优化 s2_wechat_friend 表索引
$this->optimizeWechatFriendIndexes($output);
$output->writeln("Index optimization completed successfully.");
}
protected function optimizeWechatMessageIndexes(Output $output)
{
$output->writeln("Optimizing s2_wechat_message table indexes...");
// 检查并添加 wechatChatroomId 索引
$this->addIndexIfNotExists('s2_wechat_message', 'idx_chatroom_id', 'wechatChatroomId', $output);
// 检查并添加 wechatFriendId 索引
$this->addIndexIfNotExists('s2_wechat_message', 'idx_friend_id', 'wechatFriendId', $output);
// 检查并添加 isRead 索引
$this->addIndexIfNotExists('s2_wechat_message', 'idx_is_read', 'isRead', $output);
// 检查并添加 type 索引
$this->addIndexIfNotExists('s2_wechat_message', 'idx_type', 'type', $output);
// 检查并添加 createTime 索引
$this->addIndexIfNotExists('s2_wechat_message', 'idx_create_time', 'createTime', $output);
// 检查并添加组合索引 (wechatChatroomId, isRead)
$this->addIndexIfNotExists('s2_wechat_message', 'idx_chatroom_read', 'wechatChatroomId,isRead', $output);
// 检查并添加组合索引 (wechatFriendId, isRead)
$this->addIndexIfNotExists('s2_wechat_message', 'idx_friend_read', 'wechatFriendId,isRead', $output);
}
protected function optimizeWechatChatroomIndexes(Output $output)
{
$output->writeln("Optimizing s2_wechat_chatroom table indexes...");
// 检查并添加 accountId 索引
$this->addIndexIfNotExists('s2_wechat_chatroom', 'idx_account_id', 'accountId', $output);
// 检查并添加 isDeleted 索引
$this->addIndexIfNotExists('s2_wechat_chatroom', 'idx_is_deleted', 'isDeleted', $output);
// 检查并添加组合索引 (accountId, isDeleted)
$this->addIndexIfNotExists('s2_wechat_chatroom', 'idx_account_deleted', 'accountId,isDeleted', $output);
}
protected function optimizeWechatFriendIndexes(Output $output)
{
$output->writeln("Optimizing s2_wechat_friend table indexes...");
// 检查并添加 accountId 索引
$this->addIndexIfNotExists('s2_wechat_friend', 'idx_account_id', 'accountId', $output);
// 检查并添加 isDeleted 索引
$this->addIndexIfNotExists('s2_wechat_friend', 'idx_is_deleted', 'isDeleted', $output);
// 检查并添加组合索引 (accountId, isDeleted)
$this->addIndexIfNotExists('s2_wechat_friend', 'idx_account_deleted', 'accountId,isDeleted', $output);
}
protected function addIndexIfNotExists($table, $indexName, $columns, Output $output)
{
try {
// 检查索引是否已存在
$indexExists = false;
$indexes = Db::query("SHOW INDEX FROM {$table}");
foreach ($indexes as $index) {
if ($index['Key_name'] === $indexName) {
$indexExists = true;
break;
}
}
if (!$indexExists) {
// 添加索引
Db::execute("ALTER TABLE {$table} ADD INDEX {$indexName} ({$columns})");
$output->writeln(" - Added index {$indexName} on {$table}({$columns})");
} else {
$output->writeln(" - Index {$indexName} already exists on {$table}");
}
} catch (\Exception $e) {
$output->writeln(" - Error adding index {$indexName} to {$table}: " . $e->getMessage());
}
}
}

View File

@@ -0,0 +1,51 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\OwnMomentsCollectJob;
class OwnMomentsCollectCommand extends Command
{
protected function configure()
{
$this->setName('own:moments:collect')
->setDescription('采集在线微信账号自己的朋友圈');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理自己朋友圈采集任务...');
try {
// 将任务添加到队列
$this->addToQueue();
$output->writeln('自己朋友圈采集任务已添加到队列');
} catch (\Exception $e) {
Log::error('自己朋友圈采集任务添加失败:' . $e->getMessage());
$output->writeln('自己朋友圈采集任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
*/
protected function addToQueue()
{
$data = [
'timestamp' => time()
];
// 添加到队列,设置任务名为 own_moments_collect
Queue::push(OwnMomentsCollectJob::class, $data, 'own_moments_collect');
}
}

View File

@@ -0,0 +1,121 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
class ScheduleMessageMaintenance extends Command
{
protected function configure()
{
$this->setName('schedule:message_maintenance')
->setDescription('Schedule and run message maintenance tasks')
->addOption('optimize-indexes', null, Option::VALUE_NONE, 'Run index optimization')
->addOption('clean-messages', null, Option::VALUE_NONE, 'Clean expired messages')
->addOption('days', 'd', Option::VALUE_OPTIONAL, 'Number of days to keep messages (default: 90)', 90)
->addOption('batch-size', 'b', Option::VALUE_OPTIONAL, 'Batch size for deletion (default: 1000)', 1000)
->addOption('dry-run', null, Option::VALUE_NONE, 'Perform a dry run without deleting any data');
}
protected function execute(Input $input, Output $output)
{
$optimizeIndexes = $input->getOption('optimize-indexes');
$cleanMessages = $input->getOption('clean-messages');
$days = (int)$input->getOption('days');
$batchSize = (int)$input->getOption('batch-size');
$dryRun = $input->getOption('dry-run');
// 如果没有指定任何选项,则运行所有维护任务
if (!$optimizeIndexes && !$cleanMessages) {
$optimizeIndexes = true;
$cleanMessages = true;
}
$output->writeln("<info>Starting scheduled message maintenance tasks...</info>");
$startTime = microtime(true);
// 运行索引优化
if ($optimizeIndexes) {
$this->runCommand($output, 'optimize:message_indexes');
}
// 清理过期消息
if ($cleanMessages) {
$options = [];
if ($days !== 90) {
$options[] = "--days={$days}";
}
if ($batchSize !== 1000) {
$options[] = "--batch-size={$batchSize}";
}
if ($dryRun) {
$options[] = "--dry-run";
}
$this->runCommand($output, 'clean:expired_messages', $options);
$this->runCommand($output, 'clean:expired_group_messages', $options);
}
$endTime = microtime(true);
$executionTime = round($endTime - $startTime, 2);
$output->writeln("<info>All maintenance tasks completed in {$executionTime} seconds.</info>");
}
protected function runCommand(Output $output, $command, array $options = [])
{
$output->writeln("\n<comment>Running command: {$command}</comment>");
$optionsStr = implode(' ', $options);
$fullCommand = "php think {$command} {$optionsStr}";
$output->writeln("Executing: {$fullCommand}");
$output->writeln("\n<info>Command output:</info>");
// 执行命令并实时输出结果
$descriptorSpec = [
0 => ["pipe", "r"], // stdin
1 => ["pipe", "w"], // stdout
2 => ["pipe", "w"] // stderr
];
$process = proc_open($fullCommand, $descriptorSpec, $pipes);
if (is_resource($process)) {
// 关闭标准输入
fclose($pipes[0]);
// 读取标准输出
while (!feof($pipes[1])) {
$line = fgets($pipes[1]);
if ($line !== false) {
$output->write($line);
}
}
fclose($pipes[1]);
// 读取标准错误
$errorOutput = stream_get_contents($pipes[2]);
fclose($pipes[2]);
// 获取命令执行结果
$exitCode = proc_close($process);
if ($exitCode !== 0) {
$output->writeln("\n<error>Command failed with exit code {$exitCode}</error>");
if (!empty($errorOutput)) {
$output->writeln("<error>Error output:</error>");
$output->writeln($errorOutput);
}
} else {
$output->writeln("\n<info>Command completed successfully.</info>");
}
} else {
$output->writeln("<error>Failed to execute command.</error>");
}
}
}

View File

@@ -0,0 +1,270 @@
<?php
namespace app\command;
use app\job\WorkbenchAutoLikeJob;
use think\console\Command;
use think\console\Input;
use think\console\input\Option;
use think\console\Output;
use think\Db;
use think\facade\Cache;
use think\facade\Log;
use think\Queue;
use app\api\controller\AutomaticAssign;
class SwitchFriendsCommand extends Command
{
// 队列名称
protected $queueName = 'switch_friends';
protected function configure()
{
$this->setName('switch:friends')
->setDescription('切换好友命令');
}
protected function execute(Input $input, Output $output)
{
// 清理可能损坏的缓存数据
$this->clearCorruptedCache($output);
//处理流量分过期数据
$expUserData = Db::name('workbench_traffic_config_item')
->where('expTime','<=',time())
->where('isRecycle',0)
->select();
// 根据accountId对数组进行归类
$groupedByAccount = [];
foreach ($expUserData as $friend) {
$accountId = $friend['wechatAccountId'];
if (!isset($groupedByAccount[$accountId])) {
$groupedByAccount[$accountId] = [];
}
$friendId = $friend['wechatFriendId'];
$groupedByAccount[$accountId][] = $friendId;
}
// 对每个账号的好友进行20个为一组的分组
foreach ($groupedByAccount as $accountId => $accountFriends) {
//检索主账号
$account = Db::name('users')->where('s2_accountId',$accountId)->find();
if (empty($account)) {
continue;
}
$account2 = Db::name('users')
->where('s2_accountId','>',0)
->where('companyId',$account['companyId'])
->order('s2_accountId ASC')
->find();
if (empty($account2)) {
continue;
}
$newaAccountId = $account2['s2_accountId'];
$chunks = array_chunk($accountFriends, 20);
$output->writeln('账号 ' . $newaAccountId . ' 共有 ' . count($accountFriends) . ' 个好友,分为 ' . count($chunks) . ' 组');
$automaticAssign = new AutomaticAssign();
foreach ($chunks as $chunkIndex => $chunk) {
$output->writeln('处理账号 ' . $newaAccountId . ' 第 ' . ($chunkIndex + 1) . ' 组,共 ' . count($chunk) . ' 个好友');
try {
$friendIds = implode(',', $chunk);
$res = $automaticAssign->multiAllotFriendToAccount([
'wechatFriendIds' => $friendIds,
'toAccountId' => $newaAccountId,
]);
$res = json_decode($res, true);
if ($res['code'] == 200){
//修改数据库
Db::table('s2_wechat_friend')
->where('id',$friendIds)
->update([
'accountId' => $account2['s2_accountId'],
'accountUserName' => $account2['account'],
'accountRealName' => $account2['username'],
'accountNickname' => $account2['username'],
]);
Db::name('workbench_traffic_config_item')
->whereIn('wechatFriendId',$friendIds)
->where('wechatAccountId',$accountId)
->update([
'isRecycle' => 1,
'recycleTime' => time(),
]);
$output->writeln('✓ 成功切换好友:' . $friendIds . ' 到账号:' . $newaAccountId);
} else {
$output->writeln('✗ 切换失败 - 好友:' . $friendIds . ' 到账号:' . $newaAccountId . ' 结果:' . $res['msg']);
}
} catch (\Exception $e) {
$output->writeln('✗ 切换异常 - 好友:' . implode(',', $chunk) . ' 到账号:' . $newaAccountId . ' 错误:' . $e->getMessage());
}
// 每组处理完后稍作延迟,避免请求过于频繁
if ($chunkIndex < count($chunks) - 1) {
sleep(1);
}
}
}
$cacheKey = 'allotWechatFriend';
$now = time();
$maxRetry = 5;
$retry = 0;
$switchedIds = [];
$totalProcessed = 0;
$totalSuccess = 0;
$totalFailed = 0;
$output->writeln('开始执行好友切换任务...');
do {
try {
$friends = Cache::get($cacheKey, []);
} catch (\Exception $e) {
// 如果缓存数据损坏,清空缓存并记录错误
$output->writeln('缓存数据损坏,正在清空缓存: ' . $e->getMessage());
Cache::rm($cacheKey);
$friends = [];
}
$toSwitch = [];
foreach ($friends as $friend) {
if (isset($friend['time']) && $friend['time'] < $now) {
$toSwitch[] = $friend;
}
}
if (empty($toSwitch)) {
$output->writeln('没有需要切换的好友');
return;
}
$output->writeln('找到 ' . count($toSwitch) . ' 个需要切换的好友');
$automaticAssign = new AutomaticAssign();
// 根据accountId对数组进行归类
$groupedByAccount = [];
foreach ($toSwitch as $friend) {
$accountId = $friend['accountId'];
if (!isset($groupedByAccount[$accountId])) {
$groupedByAccount[$accountId] = [];
}
$friendId = !empty($friend['friendId']) ? $friend['friendId'] : $friend['id'];
$groupedByAccount[$accountId][] = $friendId;
}
// 对每个账号的好友进行20个为一组的分组
foreach ($groupedByAccount as $accountId => $accountFriends) {
$chunks = array_chunk($accountFriends, 20);
$output->writeln('账号 ' . $accountId . ' 共有 ' . count($accountFriends) . ' 个好友,分为 ' . count($chunks) . ' 组');
$accountSuccess = 0;
$accountFailed = 0;
foreach ($chunks as $chunkIndex => $chunk) {
$output->writeln('处理账号 ' . $accountId . ' 第 ' . ($chunkIndex + 1) . ' 组,共 ' . count($chunk) . ' 个好友');
try {
$friendIds = implode(',', $chunk);
$res = $automaticAssign->multiAllotFriendToAccount([
'wechatFriendIds' => $friendIds,
'toAccountId' => $accountId,
]);
$res = json_decode($res, true);
if ($res['code'] == 200){
$output->writeln('✓ 成功切换好友:' . $friendIds . ' 到账号:' . $accountId);
$switchedIds = array_merge($switchedIds, $chunk);
$accountSuccess += count($chunk);
$totalSuccess += count($chunk);
} else {
$output->writeln('✗ 切换失败 - 好友:' . $friendIds . ' 到账号:' . $accountId . ' 结果:' . $res['msg']);
$accountFailed += count($chunk);
$totalFailed += count($chunk);
}
} catch (\Exception $e) {
$output->writeln('✗ 切换异常 - 好友:' . implode(',', $chunk) . ' 到账号:' . $accountId . ' 错误:' . $e->getMessage());
Log::error('切换好友异常: ' . $e->getMessage() . ' 好友IDs: ' . implode(',', $chunk) . ' 账号ID: ' . $accountId);
$accountFailed += count($chunk);
$totalFailed += count($chunk);
}
$totalProcessed += count($chunk);
// 每组处理完后稍作延迟,避免请求过于频繁
if ($chunkIndex < count($chunks) - 1) {
sleep(1);
}
}
$output->writeln('账号 ' . $accountId . ' 处理完成 - 成功:' . $accountSuccess . ',失败:' . $accountFailed);
}
// 过滤掉已切换的,保留未切换和新进来的
try {
$newFriends = Cache::get($cacheKey, []);
} catch (\Exception $e) {
// 如果缓存数据损坏,清空缓存并记录错误
$output->writeln('缓存数据损坏,正在清空缓存: ' . $e->getMessage());
Cache::rm($cacheKey);
$newFriends = [];
}
$updated = [];
foreach ($newFriends as $friend) {
$friendId = !empty($friend['friendId']) ? $friend['friendId'] : $friend['id'];
if (!in_array($friendId, $switchedIds)) {
$updated[] = $friend;
}
}
// 按time升序排序
usort($updated, function($a, $b) {
return ($a['time'] ?? 0) <=> ($b['time'] ?? 0);
});
try {
$success = Cache::set($cacheKey, $updated);
} catch (\Exception $e) {
// 如果缓存设置失败,记录错误并继续
$output->writeln('缓存设置失败: ' . $e->getMessage());
$success = false;
}
$retry++;
} while (!$success && $retry < $maxRetry);
$output->writeln('=== 切换任务完成 ===');
$output->writeln('总处理数量:' . $totalProcessed);
$output->writeln('成功切换:' . $totalSuccess);
$output->writeln('切换失败:' . $totalFailed);
$output->writeln('成功率:' . ($totalProcessed > 0 ? round(($totalSuccess / $totalProcessed) * 100, 2) : 0) . '%');
$output->writeln('缓存已更新并排序');
}
/**
* 清理损坏的缓存数据
* @param Output $output
*/
private function clearCorruptedCache(Output $output)
{
$cacheKey = 'allotWechatFriend';
try {
// 尝试读取缓存,如果失败则清空
$testData = Cache::get($cacheKey, []);
if (!is_array($testData)) {
$output->writeln('缓存数据格式错误,正在清空缓存');
Cache::rm($cacheKey);
}
} catch (\Exception $e) {
$output->writeln('检测到损坏的缓存数据,正在清空: ' . $e->getMessage());
Cache::rm($cacheKey);
}
}
}

View File

@@ -0,0 +1,67 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use app\job\SyncAllFriendsJob;
use think\facade\Cache;
use think\Db;
class SyncAllFriendsCommand extends Command
{
protected $queueName = 'sync_all_friends';
protected function configure()
{
$this->setName('sync:allFriends')
->setDescription('同步所有好友(自动分页队列)')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始同步所有好友...');
try {
$jobId = $input->getOption('jobId');
$queueLockKey = "queue_lock:{$this->queueName}";
Cache::rm($queueLockKey);
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,跳过执行");
return false;
}
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
$pageSize = 1000;
$accounts = Db::table('s2_wechat_account')->where('wechatAlive', 1)->select();
foreach ($accounts as $account) {
$this->addToQueue($account['wechatId'], 0, $pageSize, '', $jobId, $queueLockKey);
}
$output->writeln('同步所有好友任务已添加到队列');
} catch (\Exception $e) {
Log::error('同步所有好友任务添加失败:' . $e->getMessage());
$output->writeln('同步所有好友任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
public function addToQueue($wechatId, $pageIndex, $pageSize, $preFriendId, $jobId, $queueLockKey)
{
$data = [
'wechatId' => $wechatId,
'pageIndex' => $pageIndex,
'pageSize' => $pageSize,
'preFriendId' => $preFriendId,
'jobId' => $jobId,
'queueLockKey' => $queueLockKey
];
Queue::push(SyncAllFriendsJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,34 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Queue;
class SyncContentCommand extends Command
{
protected function configure()
{
$this->setName('content:sync')
->setDescription('同步内容库数据');
}
protected function execute(Input $input, Output $output)
{
// 将任务推送到队列
$jobHandlerClassName = 'app\job\SyncContentJob';
$jobData = [
'time' => time(),
'type' => 'sync_content'
];
$isPushed = Queue::push($jobHandlerClassName, $jobData);
if ($isPushed !== false) {
$output->writeln("同步任务已推送到队列");
} else {
$output->writeln("同步任务推送失败");
}
}
}

View File

@@ -0,0 +1,128 @@
<?php
namespace app\command;
use think\facade\Log;
use think\console\Input;
use think\console\Output;
use think\console\Command;
use think\facade\App;
use WeChatDeviceApi\Adapters\ChuKeBao\Adapter as ChuKeBaoAdapter;
// */7 * * * * cd /www/wwwroot/mckb_quwanzhi_com/Server && php think sync:wechatData >> /www/wwwroot/mckb_quwanzhi_com/Server/runtime/log/sync_wechat_data.log 2>&1
class SyncWechatDataToCkbTask extends Command
{
protected $lockFile;
public function __construct()
{
parent::__construct();
$this->lockFile = App::getRuntimePath() . 'sync_wechat_to_ckb.lock';
}
protected function configure()
{
$this->setName('sync:wechatData')
->setDescription('同步微信数据到存客宝');
}
protected function execute(Input $input, Output $output)
{
// 检查锁文件
if (file_exists($this->lockFile)) {
$lockTime = filectime($this->lockFile);
if (time() - $lockTime < 3600) {
Log::info('微信好友同步任务已在运行中,跳过本次执行');
return false;
}
unlink($this->lockFile);
}
file_put_contents($this->lockFile, time());
try {
$output->writeln("同步任务 sync_wechat_to_ckb 开始");
$ChuKeBaoAdapter = new ChuKeBaoAdapter();
$this->syncWechatAccount($ChuKeBaoAdapter);
$this->syncWechatFriend($ChuKeBaoAdapter);
$this->syncWechatDeviceLoginLog($ChuKeBaoAdapter);
$this->syncWechatDevice($ChuKeBaoAdapter);
$this->syncWechatCustomer($ChuKeBaoAdapter);
$this->syncWechatGroup($ChuKeBaoAdapter);
$this->syncWechatGroupCustomer($ChuKeBaoAdapter);
$this->syncWechatFriendToTrafficPoolBatch($ChuKeBaoAdapter);
$this->syncTrafficSourceUser($ChuKeBaoAdapter);
$this->syncTrafficSourceGroup($ChuKeBaoAdapter);
$this->syncCallRecording($ChuKeBaoAdapter);
$output->writeln("同步任务 sync_wechat_to_ckb 已结束");
return true;
} catch (\Exception $e) {
Log::error('微信好友同步任务异常:' . $e->getMessage());
return false;
} finally {
if (file_exists($this->lockFile)) {
unlink($this->lockFile);
}
}
}
protected function syncWechatFriend(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncFriendship();
}
protected function syncWechatAccount(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncWechatAccount();
}
protected function syncWechatDeviceLoginLog(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncWechatDeviceLoginLog();
}
// syncDevice
protected function syncWechatDevice(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncDevice();
}
// syncWechatCustomer
protected function syncWechatCustomer(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncWechatCustomer();
}
protected function syncWechatFriendToTrafficPoolBatch(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncWechatFriendToTrafficPoolBatch();
}
protected function syncTrafficSourceUser(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncTrafficSourceUser();
}
protected function syncTrafficSourceGroup(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncTrafficSourceGroup();
}
protected function syncWechatGroup(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncWechatGroup();
}
protected function syncWechatGroupCustomer(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncWechatGroupCustomer();
}
protected function syncCallRecording(ChuKeBaoAdapter $ChuKeBaoAdapter)
{
return $ChuKeBaoAdapter->syncCallRecording();
}
}

View File

@@ -0,0 +1,478 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Config;
use think\facade\Log;
use think\facade\Cache;
/**
* 统一任务调度器
* 支持多进程并发执行任务
*
* 使用方法:
* php think scheduler:run
*
* 在 crontab 中配置:
* * * * * * cd /path/to/project && php think scheduler:run >> /path/to/log/scheduler.log 2>&1
*/
class TaskSchedulerCommand extends Command
{
/**
* 任务配置
*/
protected $tasks = [];
/**
* 最大并发进程数
*/
protected $maxConcurrent = 10;
/**
* 当前运行的进程数
*/
protected $runningProcesses = [];
/**
* 日志目录
*/
protected $logDir = '';
protected function configure()
{
$this->setName('scheduler:run')
->setDescription('统一任务调度器,支持多进程并发执行所有定时任务');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('==========================================');
$output->writeln('任务调度器启动');
$output->writeln('时间: ' . date('Y-m-d H:i:s'));
$output->writeln('==========================================');
// 检查是否支持 pcntl 扩展
if (!function_exists('pcntl_fork')) {
$output->writeln('<error>错误:系统不支持 pcntl 扩展,无法使用多进程功能</error>');
$output->writeln('<info>提示:将使用单进程顺序执行任务</info>');
$this->maxConcurrent = 1;
}
// 加载任务配置(优先使用框架配置,其次直接引入配置文件,避免加载失败)
$this->tasks = Config::get('task_scheduler', []);
// 如果通过 Config 没有读到,再尝试直接 include 配置文件
if (empty($this->tasks)) {
// 以项目根目录为基准查找 config/task_scheduler.php
$configFile = dirname(__DIR__, 2) . DIRECTORY_SEPARATOR . 'config' . DIRECTORY_SEPARATOR . 'task_scheduler.php';
if (is_file($configFile)) {
$config = include $configFile;
if (is_array($config) && !empty($config)) {
$this->tasks = $config;
}
}
}
if (empty($this->tasks)) {
$output->writeln('<error>错误未找到任务配置task_scheduler请检查 config/task_scheduler.php 是否存在且返回数组</error>');
return false;
}
// 设置日志目录ThinkPHP5 中无 runtime_path 辅助函数,直接使用 ROOT_PATH/runtime/log
if (!defined('ROOT_PATH')) {
// CLI 下正常情况下 ROOT_PATH 已在入口脚本 define这里兜底一次
define('ROOT_PATH', dirname(__DIR__, 2));
}
$this->logDir = ROOT_PATH . DIRECTORY_SEPARATOR . 'runtime' . DIRECTORY_SEPARATOR . 'log' . DIRECTORY_SEPARATOR;
if (!is_dir($this->logDir)) {
mkdir($this->logDir, 0755, true);
}
// 获取当前时间
$currentTime = time();
$currentMinute = date('i', $currentTime);
$currentHour = date('H', $currentTime);
$currentDay = date('d', $currentTime);
$currentMonth = date('m', $currentTime);
$currentWeekday = date('w', $currentTime); // 0=Sunday, 6=Saturday
$output->writeln("当前时间: {$currentHour}:{$currentMinute}");
$output->writeln("已加载 " . count($this->tasks) . " 个任务配置");
// 筛选需要执行的任务
$tasksToRun = [];
foreach ($this->tasks as $taskId => $task) {
if (!isset($task['enabled']) || !$task['enabled']) {
continue;
}
if ($this->shouldRun($task['schedule'], $currentMinute, $currentHour, $currentDay, $currentMonth, $currentWeekday)) {
$tasksToRun[$taskId] = $task;
}
}
if (empty($tasksToRun)) {
$output->writeln('<info>当前时间没有需要执行的任务</info>');
return true;
}
$output->writeln("找到 " . count($tasksToRun) . " 个需要执行的任务");
// 执行任务
if ($this->maxConcurrent > 1 && function_exists('pcntl_fork')) {
$this->executeConcurrent($tasksToRun, $output);
} else {
$this->executeSequential($tasksToRun, $output);
}
// 清理僵尸进程
$this->cleanupZombieProcesses();
$output->writeln('==========================================');
$output->writeln('任务调度器执行完成');
$output->writeln('==========================================');
return true;
}
/**
* 判断任务是否应该执行
*
* @param string $schedule cron表达式格式分钟 小时 日 月 星期
* @param int $minute 当前分钟
* @param int $hour 当前小时
* @param int $day 当前日期
* @param int $month 当前月份
* @param int $weekday 当前星期
* @return bool
*/
protected function shouldRun($schedule, $minute, $hour, $day, $month, $weekday)
{
$parts = preg_split('/\s+/', trim($schedule));
if (count($parts) < 5) {
return false;
}
list($scheduleMinute, $scheduleHour, $scheduleDay, $scheduleMonth, $scheduleWeekday) = $parts;
// 解析分钟
if (!$this->matchCronField($scheduleMinute, $minute)) {
return false;
}
// 解析小时
if (!$this->matchCronField($scheduleHour, $hour)) {
return false;
}
// 解析日期
if (!$this->matchCronField($scheduleDay, $day)) {
return false;
}
// 解析月份
if (!$this->matchCronField($scheduleMonth, $month)) {
return false;
}
// 解析星期注意cron中0和7都表示星期日
if ($scheduleWeekday !== '*') {
$scheduleWeekday = str_replace('7', '0', $scheduleWeekday);
if (!$this->matchCronField($scheduleWeekday, $weekday)) {
return false;
}
}
return true;
}
/**
* 匹配cron字段
*
* @param string $field cron字段表达式
* @param int $value 当前值
* @return bool
*/
protected function matchCronField($field, $value)
{
// 通配符
if ($field === '*') {
return true;
}
// 列表(逗号分隔)
if (strpos($field, ',') !== false) {
$values = explode(',', $field);
foreach ($values as $v) {
if ($this->matchCronField(trim($v), $value)) {
return true;
}
}
return false;
}
// 范围(如 1-5
if (strpos($field, '-') !== false) {
list($start, $end) = explode('-', $field);
return $value >= (int)$start && $value <= (int)$end;
}
// 步长(如 */5 或 0-59/5
if (strpos($field, '/') !== false) {
$parts = explode('/', $field);
$base = $parts[0];
$step = (int)$parts[1];
if ($base === '*') {
return $value % $step === 0;
} else {
// 处理范围步长,如 0-59/5
if (strpos($base, '-') !== false) {
list($start, $end) = explode('-', $base);
if ($value >= (int)$start && $value <= (int)$end) {
return ($value - (int)$start) % $step === 0;
}
return false;
} else {
return $value % $step === 0;
}
}
}
// 精确匹配
return (int)$field === $value;
}
/**
* 并发执行任务(多进程)
*
* @param array $tasks 任务列表
* @param Output $output 输出对象
*/
protected function executeConcurrent($tasks, Output $output)
{
$output->writeln('<info>使用多进程并发执行任务(最大并发数:' . $this->maxConcurrent . '</info>');
foreach ($tasks as $taskId => $task) {
// 等待可用进程槽
while (count($this->runningProcesses) >= $this->maxConcurrent) {
$this->waitForProcesses();
usleep(100000); // 等待100ms
}
// 检查任务是否已经在运行(防止重复执行)
$lockKey = "scheduler_task_lock:{$taskId}";
$lockTime = Cache::get($lockKey);
if ($lockTime && (time() - $lockTime) < 300) { // 5分钟内不重复执行
$output->writeln("<comment>任务 {$taskId} 正在运行中,跳过</comment>");
continue;
}
// 创建子进程
$pid = pcntl_fork();
if ($pid == -1) {
// 创建进程失败
$output->writeln("<error>创建子进程失败:{$taskId}</error>");
Log::error("任务调度器:创建子进程失败", ['task' => $taskId]);
continue;
} elseif ($pid == 0) {
// 子进程:执行任务
$this->runTask($taskId, $task);
exit(0);
} else {
// 父进程记录子进程PID
$this->runningProcesses[$pid] = [
'task_id' => $taskId,
'start_time' => time(),
];
$output->writeln("<info>启动任务:{$taskId} (PID: {$pid})</info>");
// 设置任务锁
Cache::set($lockKey, time(), 600); // 10分钟过期
}
}
// 等待所有子进程完成
while (!empty($this->runningProcesses)) {
$this->waitForProcesses();
usleep(500000); // 等待500ms
}
}
/**
* 顺序执行任务(单进程)
*
* @param array $tasks 任务列表
* @param Output $output 输出对象
*/
protected function executeSequential($tasks, Output $output)
{
$output->writeln('<info>使用单进程顺序执行任务</info>');
foreach ($tasks as $taskId => $task) {
$output->writeln("<info>执行任务:{$taskId}</info>");
$this->runTask($taskId, $task);
}
}
/**
* 执行单个任务
*
* @param string $taskId 任务ID
* @param array $task 任务配置
*/
protected function runTask($taskId, $task)
{
$startTime = microtime(true);
$logFile = $this->logDir . ($task['log_file'] ?? "scheduler_{$taskId}.log");
// 确保日志目录存在
$logDir = dirname($logFile);
if (!is_dir($logDir)) {
mkdir($logDir, 0755, true);
}
// 构建命令
// 使用项目根目录下的 think 脚本(同命令行 php think
if (!defined('ROOT_PATH')) {
define('ROOT_PATH', dirname(__DIR__, 2));
}
$thinkPath = ROOT_PATH . DIRECTORY_SEPARATOR . 'think';
$command = "php {$thinkPath} {$task['command']}";
if (!empty($task['options'])) {
foreach ($task['options'] as $option) {
$command .= ' ' . escapeshellarg($option);
}
}
// 添加日志重定向
$command .= " >> " . escapeshellarg($logFile) . " 2>&1";
// 记录任务开始
$logMessage = "\n" . str_repeat('=', 60) . "\n";
$logMessage .= "任务开始执行: {$taskId}\n";
$logMessage .= "执行时间: " . date('Y-m-d H:i:s') . "\n";
$logMessage .= "命令: {$command}\n";
$logMessage .= str_repeat('=', 60) . "\n";
file_put_contents($logFile, $logMessage, FILE_APPEND);
// 执行命令
$descriptorspec = [
0 => ['file', (PHP_OS_FAMILY === 'Windows' ? 'NUL' : '/dev/null'), 'r'], // stdin
1 => ['file', $logFile, 'a'], // stdout
2 => ['file', $logFile, 'a'], // stderr
];
$process = @proc_open($command, $descriptorspec, $pipes, ROOT_PATH);
if (is_resource($process)) {
// 关闭管道
if (isset($pipes[0])) @fclose($pipes[0]);
if (isset($pipes[1])) @fclose($pipes[1]);
if (isset($pipes[2])) @fclose($pipes[2]);
// 设置超时
$timeout = $task['timeout'] ?? 3600;
$startWaitTime = time();
// 等待进程完成或超时
while (true) {
$status = proc_get_status($process);
if (!$status['running']) {
break;
}
// 检查超时
if ((time() - $startWaitTime) > $timeout) {
if (function_exists('proc_terminate')) {
proc_terminate($process, SIGTERM);
// 等待进程终止
sleep(2);
$status = proc_get_status($process);
if ($status['running']) {
// 强制终止
proc_terminate($process, SIGKILL);
}
}
Log::warning("任务执行超时", [
'task' => $taskId,
'timeout' => $timeout,
]);
break;
}
usleep(500000); // 等待500ms
}
// 关闭进程
proc_close($process);
} else {
// 如果 proc_open 失败,尝试直接执行(后台执行)
if (PHP_OS_FAMILY === 'Windows') {
pclose(popen("start /B " . $command, "r"));
} else {
exec($command . ' > /dev/null 2>&1 &');
}
}
$endTime = microtime(true);
$duration = round($endTime - $startTime, 2);
// 记录任务完成
$logMessage = "\n" . str_repeat('=', 60) . "\n";
$logMessage .= "任务执行完成: {$taskId}\n";
$logMessage .= "完成时间: " . date('Y-m-d H:i:s') . "\n";
$logMessage .= "执行时长: {$duration}\n";
$logMessage .= str_repeat('=', 60) . "\n";
file_put_contents($logFile, $logMessage, FILE_APPEND);
Log::info("任务执行完成", [
'task' => $taskId,
'duration' => $duration,
]);
}
/**
* 等待进程完成
*/
protected function waitForProcesses()
{
foreach ($this->runningProcesses as $pid => $info) {
$status = 0;
$result = pcntl_waitpid($pid, $status, WNOHANG);
if ($result == $pid || $result == -1) {
// 进程已结束
unset($this->runningProcesses[$pid]);
$duration = time() - $info['start_time'];
Log::info("子进程执行完成", [
'pid' => $pid,
'task' => $info['task_id'],
'duration' => $duration,
]);
}
}
}
/**
* 清理僵尸进程
*/
protected function cleanupZombieProcesses()
{
if (!function_exists('pcntl_waitpid')) {
return;
}
$status = 0;
while (($pid = pcntl_waitpid(-1, $status, WNOHANG)) > 0) {
// 清理僵尸进程
}
}
}

View File

@@ -0,0 +1,168 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Db;
use app\common\service\WechatAccountHealthScoreService;
/**
* 更新微信账号评分记录
* 根据wechatId和alias是否不一致来更新isModifiedAlias字段仅用于评分不修复数据
*/
class UpdateWechatAccountScoreCommand extends Command
{
protected function configure()
{
$this->setName('wechat:update-score')
->setDescription('更新微信账号评分记录根据wechatId和alias不一致情况更新isModifiedAlias字段仅用于评分');
}
protected function execute(Input $input, Output $output)
{
$output->writeln("开始更新微信账号评分记录...");
try {
// 1. 查找所有需要更新的账号
$output->writeln("步骤1: 查找需要更新的账号...");
// 查找wechatId和alias不一致的账号
$inconsistentAccounts = Db::table('s2_wechat_account')
->where('isDeleted', 0)
->where('wechatId', '<>', '')
->where('alias', '<>', '')
->whereRaw('wechatId != alias')
->field('id, wechatId, alias')
->select();
// 查找wechatId和alias一致的账号
$consistentAccounts = Db::table('s2_wechat_account')
->where('isDeleted', 0)
->where('wechatId', '<>', '')
->where('alias', '<>', '')
->whereRaw('wechatId = alias')
->field('id, wechatId, alias')
->select();
$output->writeln("发现 " . count($inconsistentAccounts) . " 条不一致记录(已修改微信号)");
$output->writeln("发现 " . count($consistentAccounts) . " 条一致记录(未修改微信号)");
// 2. 更新评分记录表中的isModifiedAlias字段
$output->writeln("步骤2: 更新评分记录表...");
$updatedCount = 0;
$healthScoreService = new WechatAccountHealthScoreService();
// 更新不一致的记录
foreach ($inconsistentAccounts as $account) {
$this->updateScoreRecord($account['id'], true, $healthScoreService);
$updatedCount++;
}
// 更新一致的记录
foreach ($consistentAccounts as $account) {
$this->updateScoreRecord($account['id'], false, $healthScoreService);
$updatedCount++;
}
$output->writeln("已更新 " . $updatedCount . " 条评分记录");
// 3. 重新计算健康分(只更新基础信息分,不重新计算基础分)
$output->writeln("步骤3: 重新计算健康分...");
$allAccountIds = array_merge(
array_column($inconsistentAccounts, 'id'),
array_column($consistentAccounts, 'id')
);
if (!empty($allAccountIds)) {
$stats = $healthScoreService->batchCalculateAndUpdate($allAccountIds, 100, false);
$output->writeln("健康分计算完成:成功 " . $stats['success'] . " 条,失败 " . $stats['failed'] . "");
if (!empty($stats['errors'])) {
$output->writeln("错误详情:");
foreach ($stats['errors'] as $error) {
$output->writeln(" 账号ID {$error['accountId']}: {$error['error']}");
}
}
}
$output->writeln("任务完成!");
} catch (\Exception $e) {
$output->writeln("错误: " . $e->getMessage());
$output->writeln($e->getTraceAsString());
}
}
/**
* 更新评分记录
*
* @param int $accountId 账号ID
* @param bool $isModifiedAlias 是否已修改微信号
* @param WechatAccountHealthScoreService $service 评分服务
*/
private function updateScoreRecord($accountId, $isModifiedAlias, $service)
{
// 获取或创建评分记录
$accountData = Db::table('s2_wechat_account')
->where('id', $accountId)
->find();
if (empty($accountData)) {
return;
}
// 确保评分记录存在
$scoreRecord = Db::table('s2_wechat_account_score')
->where('accountId', $accountId)
->find();
if (empty($scoreRecord)) {
// 如果记录不存在,创建并计算基础分
$service->calculateAndUpdate($accountId);
$scoreRecord = Db::table('s2_wechat_account_score')
->where('accountId', $accountId)
->find();
}
if (empty($scoreRecord)) {
return;
}
// 更新isModifiedAlias字段
$updateData = [
'isModifiedAlias' => $isModifiedAlias ? 1 : 0,
'updateTime' => time()
];
// 如果基础分已计算,需要更新基础信息分和基础分
if ($scoreRecord['baseScoreCalculated']) {
$oldBaseInfoScore = $scoreRecord['baseInfoScore'] ?? 0;
$newBaseInfoScore = $isModifiedAlias ? 10 : 0; // 已修改微信号得10分
if ($oldBaseInfoScore != $newBaseInfoScore) {
$oldBaseScore = $scoreRecord['baseScore'] ?? 60;
$newBaseScore = $oldBaseScore - $oldBaseInfoScore + $newBaseInfoScore;
$updateData['baseInfoScore'] = $newBaseInfoScore;
$updateData['baseScore'] = $newBaseScore;
// 重新计算健康分
$dynamicScore = $scoreRecord['dynamicScore'] ?? 0;
$healthScore = $newBaseScore + $dynamicScore;
$healthScore = max(0, min(100, $healthScore));
$updateData['healthScore'] = $healthScore;
$updateData['maxAddFriendPerDay'] = (int)floor($healthScore * 0.2);
}
} else {
// 基础分未计算,只更新标记和基础信息分
$updateData['baseInfoScore'] = $isModifiedAlias ? 10 : 0;
}
Db::table('s2_wechat_account_score')
->where('accountId', $accountId)
->update($updateData);
}
}

View File

@@ -0,0 +1,99 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use app\job\WechatChatroomJob;
use think\facade\Cache;
class WechatChatroomCommand extends Command
{
// 队列名称
protected $queueName = 'wechat_chatroom';
protected function configure()
{
$this->setName('wechatChatroom:list')
->setDescription('获取微信聊天室列表,并根据分页自动处理下一页')
->addOption('isDel', null, Option::VALUE_OPTIONAL, '删除状态: 0=未删除(false), 1=已删除(true)', '')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理微信聊天室列表任务...');
try {
// 获取是否删除参数和任务ID
$isDel = $input->getOption('isDel');
$jobId = $input->getOption('jobId');
$output->writeln('删除状态参数: ' . ($isDel === '' ? '全部' : ($isDel == 0 ? '未删除' : '已删除')));
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}:{$isDel}";
Cache::rm($queueLockKey);
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,删除状态:{$isDel},跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,删除状态:{$isDel},跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 为不同的删除状态和任务ID使用不同的缓存键名
$cacheKeyPrefix = "chatroomPage:{$jobId}";
$cacheKeySuffix = $isDel === '' ? '' : ":{$isDel}";
$cacheKey = $cacheKeyPrefix . $cacheKeySuffix;
// 从缓存获取初始页码缓存有效期1天
$pageIndex = Cache::get($cacheKey, 0);
$output->writeln("从缓存获取页码: {$pageIndex}, 缓存键: {$cacheKey}");
$pageSize = 100; // 每页获取100条记录
// 将任务添加到队列
$this->addToQueue($pageIndex, $pageSize, $isDel, $jobId, $cacheKey, $queueLockKey);
$output->writeln('微信聊天室列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('微信聊天室列表任务添加失败:' . $e->getMessage());
$output->writeln('微信聊天室列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
* @param string $isDel 删除状态
* @param string $jobId 任务ID
* @param string $cacheKey 缓存键名
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($pageIndex, $pageSize, $isDel = '', $jobId = '', $cacheKey = '', $queueLockKey = '')
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize,
'isDel' => $isDel,
'jobId' => $jobId,
'cacheKey' => $cacheKey,
'queueLockKey' => $queueLockKey
];
// 添加到队列,设置任务名为 wechat_chatroom
Queue::push(WechatChatroomJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,107 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use app\job\WechatFriendJob;
use think\facade\Cache;
class WechatFriendCommand extends Command
{
// 队列名称
protected $queueName = 'wechat_friends';
protected function configure()
{
$this->setName('wechatFriends:list')
->setDescription('获微信列表,并根据分页自动处理下一页')
->addOption('isDel', null, Option::VALUE_OPTIONAL, '删除状态: 0=未删除(false), 1=已删除(true)', '')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理微信列表任务...');
try {
// 获取是否删除参数和任务ID
$isDel = $input->getOption('isDel');
$jobId = $input->getOption('jobId');
$output->writeln('删除状态参数: ' . ($isDel === '' ? '全部' : ($isDel == 0 ? '未删除' : '已删除')));
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}:{$isDel}";
Cache::rm($queueLockKey);
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,删除状态:{$isDel},跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,删除状态:{$isDel},跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 为不同的删除状态和任务ID使用不同的缓存键名
$cacheKeyPrefix = "friendsPage:{$jobId}";
$cacheKeySuffix = $isDel === '' ? '' : ":{$isDel}";
$pageIndexCacheKey = $cacheKeyPrefix . $cacheKeySuffix;
$preFriendIdCacheKey = "preFriendId:{$jobId}" . $cacheKeySuffix;
// 从缓存获取初始页码和上次处理的好友ID
$pageIndex = Cache::get($pageIndexCacheKey, 0);
$preFriendId = Cache::get($preFriendIdCacheKey, '');
$output->writeln("从缓存获取页码: {$pageIndex}, 上次处理的好友ID: {$preFriendId}");
$output->writeln("缓存键: {$pageIndexCacheKey}, {$preFriendIdCacheKey}");
$pageSize = 100; // 每页获取100条记录
// 将任务添加到队列
$this->addToQueue($pageIndex, $pageSize, $preFriendId, $isDel, $jobId, $pageIndexCacheKey, $preFriendIdCacheKey, $queueLockKey);
$output->writeln('微信列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('微信列表任务添加失败:' . $e->getMessage());
$output->writeln('微信列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
* @param string $preFriendId 上一个好友ID
* @param string $isDel 删除状态
* @param string $jobId 任务ID
* @param string $pageIndexCacheKey 页码缓存键名
* @param string $preFriendIdCacheKey 好友ID缓存键名
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($pageIndex, $pageSize, $preFriendId = '', $isDel = '', $jobId = '', $pageIndexCacheKey = '', $preFriendIdCacheKey = '', $queueLockKey = '')
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize,
'preFriendId' => $preFriendId,
'isDel' => $isDel,
'jobId' => $jobId,
'pageIndexCacheKey' => $pageIndexCacheKey,
'preFriendIdCacheKey' => $preFriendIdCacheKey,
'queueLockKey' => $queueLockKey
];
// 添加到队列,设置任务名为 wechat_friends
Queue::push(WechatFriendJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,57 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\Queue;
use app\job\WechatListJob;
class WechatListCommand extends Command
{
protected function configure()
{
$this->setName('wechat:list')
->setDescription('获取微信客服列表,并根据分页自动处理下一页');
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理微信客服列表任务...');
try {
// 初始页码
$pageIndex = 0;
$pageSize = 500; // 每页获取100条记录
// 将第一页任务添加到队列
$this->addToQueue($pageIndex, $pageSize);
$output->writeln('微信客服列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('微信客服列表任务添加失败:' . $e->getMessage());
$output->writeln('微信客服列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
*/
protected function addToQueue($pageIndex, $pageSize)
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize
];
// 添加到队列,设置任务名为 wechat_list
Queue::push(WechatListJob::class, $data, 'wechat_list');
}
}

View File

@@ -0,0 +1,99 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use app\job\WechatMomentsJob;
use think\facade\Cache;
class WechatMomentsCommand extends Command
{
// 队列名称
protected $queueName = 'wechat_moments';
protected function configure()
{
$this->setName('wechatMoments:list')
->setDescription('获取朋友圈列表,并根据分页自动处理下一页')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理朋友圈列表任务...');
try {
// 获取任务ID
$jobId = $input->getOption('jobId');
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}";
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 为不同的任务ID使用不同的缓存键名
$pageIndexCacheKey = "momentsPage:{$jobId}";
$preMomentIdCacheKey = "preMomentId:{$jobId}";
// 从缓存获取初始页码和上次处理的朋友圈ID
$pageIndex = Cache::get($pageIndexCacheKey, 1);
$preMomentId = Cache::get($preMomentIdCacheKey, '');
$output->writeln("从缓存获取页码: {$pageIndex}, 上次处理的朋友圈ID: {$preMomentId}");
$output->writeln("缓存键: {$pageIndexCacheKey}, {$preMomentIdCacheKey}");
$pageSize = 100; // 每页获取100条记录
// 将任务添加到队列
$this->addToQueue($pageIndex, $pageSize, $preMomentId, $jobId, $pageIndexCacheKey, $preMomentIdCacheKey, $queueLockKey);
$output->writeln('朋友圈列表任务已添加到队列');
} catch (\Exception $e) {
Log::error('朋友圈列表任务添加失败:' . $e->getMessage());
$output->writeln('朋友圈列表任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param int $pageIndex 页码
* @param int $pageSize 每页大小
* @param string $preMomentId 上一个朋友圈ID
* @param string $jobId 任务ID
* @param string $pageIndexCacheKey 页码缓存键名
* @param string $preMomentIdCacheKey 朋友圈ID缓存键名
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($pageIndex, $pageSize, $preMomentId = '', $jobId = '', $pageIndexCacheKey = '', $preMomentIdCacheKey = '', $queueLockKey = '')
{
$data = [
'pageIndex' => $pageIndex,
'pageSize' => $pageSize,
'preMomentId' => $preMomentId,
'jobId' => $jobId,
'pageIndexCacheKey' => $pageIndexCacheKey,
'preMomentIdCacheKey' => $preMomentIdCacheKey,
'queueLockKey' => $queueLockKey
];
// 添加到队列,设置任务名为 wechat_moments
Queue::push(WechatMomentsJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,77 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use app\job\WorkbenchAutoLikeJob;
use think\facade\Cache;
class WorkbenchAutoLikeCommand extends Command
{
// 队列名称
protected $queueName = 'workbench_auto_like';
protected function configure()
{
$this->setName('workbench:autoLike')
->setDescription('工作台自动点赞任务队列')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理工作台自动点赞任务...');
try {
// 获取任务ID
$jobId = $input->getOption('jobId');
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}";
//Cache::rm($queueLockKey);
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 将任务添加到队列
$this->addToQueue($jobId, $queueLockKey);
$output->writeln('工作台自动点赞任务已添加到队列');
} catch (\Exception $e) {
Log::error('工作台自动点赞任务添加失败:' . $e->getMessage());
$output->writeln('工作台自动点赞任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param string $jobId 任务ID
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($jobId = '', $queueLockKey = '')
{
$data = [
'jobId' => $jobId,
'queueLockKey' => $queueLockKey
];
// 添加到队列,设置任务名为 workbench_auto_like
Queue::push(WorkbenchAutoLikeJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,76 @@
<?php
namespace app\command;
use app\job\WorkbenchGroupCreateJob;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use think\facade\Cache;
class WorkbenchGroupCreateCommand extends Command
{
// 队列名称
protected $queueName = 'workbench_groupCreate';
protected function configure()
{
$this->setName('workbench:groupCreate')
->setDescription('工作台群创建同步任务队列')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理工作台群创建同步任务...');
try {
// 获取任务ID
$jobId = $input->getOption('jobId');
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}";
Cache::rm($queueLockKey);
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 将任务添加到队列
$this->addToQueue($jobId, $queueLockKey);
$output->writeln('工作台群发同步任务已添加到队列');
} catch (\Exception $e) {
Log::error('工作台群发同步任务添加失败:' . $e->getMessage());
$output->writeln('工作台群发同步任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param string $jobId 任务ID
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($jobId = '', $queueLockKey = '')
{
$data = [
'jobId' => $jobId,
'queueLockKey' => $queueLockKey
];
// 添加到队列,设置任务名为 workbench_groupCreate
Queue::push(WorkbenchGroupCreateJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,76 @@
<?php
namespace app\command;
use app\job\WorkbenchGroupPushJob;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use think\facade\Cache;
class WorkbenchGroupPushCommand extends Command
{
// 队列名称
protected $queueName = 'workbench_groupPush';
protected function configure()
{
$this->setName('workbench:groupPush')
->setDescription('工作台群发同步任务队列')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理工作台群发同步任务...');
try {
// 获取任务ID
$jobId = $input->getOption('jobId');
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}";
Cache::rm($queueLockKey);
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 将任务添加到队列
$this->addToQueue($jobId, $queueLockKey);
$output->writeln('工作台群发同步任务已添加到队列');
} catch (\Exception $e) {
Log::error('工作台群发同步任务添加失败:' . $e->getMessage());
$output->writeln('工作台群发同步任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param string $jobId 任务ID
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($jobId = '', $queueLockKey = '')
{
$data = [
'jobId' => $jobId,
'queueLockKey' => $queueLockKey
];
// 添加到队列,设置任务名为 workbench_groupPush
Queue::push(WorkbenchGroupPushJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,125 @@
<?php
namespace app\command;
use app\job\WorkbenchImportContactJob;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\facade\Log;
use think\facade\Cache;
use think\Queue;
/**
* 工作台通讯录导入命令
* Class WorkbenchImportContactCommand
* @package app\command
*/
class WorkbenchImportContactCommand extends Command
{
/**
* 配置命令
*/
protected function configure()
{
$this->setName('workbench:import-contact')
->setDescription('执行工作台通讯录导入任务');
}
/**
* 执行命令
* @param Input $input
* @param Output $output
* @return int
*/
protected function execute(Input $input, Output $output)
{
$output->writeln('开始执行工作台通讯录导入任务...');
try {
// 检查是否有任务正在执行
$lockKey = 'workbench_import_contact_lock';
if (Cache::has($lockKey)) {
$output->writeln('通讯录导入任务正在执行中,跳过本次执行');
return 0;
}
// 设置执行锁,防止重复执行
Cache::set($lockKey, time(), 3600); // 1小时锁定时间
// 生成任务ID
$jobId = 'workbench_import_contact_' . date('YmdHis') . '_' . mt_rand(1000, 9999);
// 准备任务数据
$jobData = [
'jobId' => $jobId,
'queueLockKey' => $lockKey,
'executeTime' => time()
];
// 判断是否使用队列
if ($this->shouldUseQueue()) {
// 推送到队列
Queue::push(WorkbenchImportContactJob::class, $jobData, 'workbench_import_contact');
$output->writeln("通讯录导入任务已推送到队列任务ID: {$jobId}");
} else {
// 直接执行
$job = new WorkbenchImportContactJob();
$result = $job->execute();
// 释放锁
Cache::rm($lockKey);
if ($result !== false) {
$output->writeln('通讯录导入任务执行成功');
} else {
$output->writeln('通讯录导入任务执行失败');
return 1;
}
}
} catch (\Exception $e) {
// 释放锁
Cache::rm($lockKey ?? '');
$errorMsg = '通讯录导入任务执行异常: ' . $e->getMessage();
$output->writeln($errorMsg);
Log::error($errorMsg);
return 1;
}
return 0;
}
/**
* 判断是否应该使用队列
* @return bool
*/
protected function shouldUseQueue()
{
// 检查队列配置是否启用
$queueConfig = config('queue');
if (empty($queueConfig) || !isset($queueConfig['default'])) {
return false;
}
// 检查队列连接是否可用
try {
$connection = $queueConfig['connections'][$queueConfig['default']] ?? [];
if (empty($connection)) {
return false;
}
// 如果是数据库队列,检查表是否存在
if ($connection['type'] === 'database') {
$tableName = $connection['table'] ?? 'jobs';
$exists = \think\Db::query("SHOW TABLES LIKE '{$tableName}'");
return !empty($exists);
}
return true;
} catch (\Exception $e) {
Log::warning('队列检查失败,将使用同步执行: ' . $e->getMessage());
return false;
}
}
}

View File

@@ -0,0 +1,76 @@
<?php
namespace app\command;
use app\job\WorkbenchMomentsJob;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use think\facade\Cache;
class WorkbenchMomentsCommand extends Command
{
// 队列名称
protected $queueName = 'workbench_moments';
protected function configure()
{
$this->setName('workbench:moments')
->setDescription('工作台朋友圈同步任务队列')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID用于区分不同实例', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理工作台朋友圈同步任务...');
try {
// 获取任务ID
$jobId = $input->getOption('jobId');
$output->writeln('任务ID: ' . $jobId);
// 检查队列是否已经在运行
$queueLockKey = "queue_lock:{$this->queueName}";
Cache::rm($queueLockKey);
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,跳过执行");
return false;
}
// 设置队列运行锁有效期1小时
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
// 将任务添加到队列
$this->addToQueue($jobId, $queueLockKey);
$output->writeln('工作台朋友圈同步任务已添加到队列');
} catch (\Exception $e) {
Log::error('工作台朋友圈同步任务添加失败:' . $e->getMessage());
$output->writeln('工作台朋友圈同步任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
/**
* 添加任务到队列
* @param string $jobId 任务ID
* @param string $queueLockKey 队列锁键名
*/
public function addToQueue($jobId = '', $queueLockKey = '')
{
$data = [
'jobId' => $jobId,
'queueLockKey' => $queueLockKey
];
// 添加到队列,设置任务名为 workbench_moments
Queue::push(WorkbenchMomentsJob::class, $data, $this->queueName);
}
}

View File

@@ -0,0 +1,61 @@
<?php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use think\facade\Log;
use think\Queue;
use app\job\WorkbenchTrafficDistributeJob;
use think\facade\Cache;
class WorkbenchTrafficDistributeCommand extends Command
{
protected $queueName = 'workbench_traffic_distribute';
protected function configure()
{
$this->setName('workbench:trafficDistribute')
->setDescription('工作台流量分发任务队列')
->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID', date('YmdHis') . rand(1000, 9999));
}
protected function execute(Input $input, Output $output)
{
$output->writeln('开始处理流量分发任务...');
try {
$jobId = $input->getOption('jobId');
$output->writeln('任务ID: ' . $jobId);
$queueLockKey = "queue_lock:{$this->queueName}";
Cache::rm($queueLockKey);
if (Cache::get($queueLockKey)) {
$output->writeln("队列 {$this->queueName} 已经在运行中,跳过执行");
Log::warning("队列 {$this->queueName} 已经在运行中,跳过执行");
return false;
}
Cache::set($queueLockKey, $jobId, 3600);
$output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时");
$this->addToQueue($jobId, $queueLockKey);
$output->writeln('流量分发任务已添加到队列');
} catch (\Exception $e) {
Log::error('流量分发任务添加失败:' . $e->getMessage());
$output->writeln('流量分发任务添加失败:' . $e->getMessage());
return false;
}
return true;
}
public function addToQueue($jobId = '', $queueLockKey = '')
{
$data = [
'jobId' => $jobId,
'queueLockKey' => $queueLockKey
];
Queue::push(WorkbenchTrafficDistributeJob::class, $data, $this->queueName);
}
}