更新
This commit is contained in:
@@ -16,6 +16,8 @@ namespace app\adminapi\logic\dept;
|
||||
|
||||
use app\common\enum\YesNoEnum;
|
||||
use app\common\logic\BaseLogic;
|
||||
use app\common\model\auth\Admin;
|
||||
use app\common\model\auth\AdminDept;
|
||||
use app\common\model\dept\Dept;
|
||||
|
||||
|
||||
@@ -52,6 +54,12 @@ class DeptLogic extends BaseLogic
|
||||
->select()
|
||||
->toArray();
|
||||
|
||||
$totalsWithSubtree = self::deptAdminCountTotalsWithSubtree();
|
||||
foreach ($lists as &$row) {
|
||||
$row['admin_count'] = $totalsWithSubtree[(int) $row['id']] ?? 0;
|
||||
}
|
||||
unset($row);
|
||||
|
||||
$pid = 0;
|
||||
if (!empty($lists)) {
|
||||
$pid = min(array_column($lists, 'pid'));
|
||||
@@ -59,6 +67,79 @@ class DeptLogic extends BaseLogic
|
||||
return self::getTree($lists, $pid);
|
||||
}
|
||||
|
||||
/**
|
||||
* 每个部门人数(直属 + 所有下级部门),基于全表部门树汇总;管理员来自 admin_dept,排除已删除账号
|
||||
*
|
||||
* @return array<int, int> dept_id => count
|
||||
*/
|
||||
private static function deptAdminCountTotalsWithSubtree(): array
|
||||
{
|
||||
$treeRows = Dept::field(['id', 'pid'])->select()->toArray();
|
||||
if ($treeRows === []) {
|
||||
return [];
|
||||
}
|
||||
$allIds = array_map('intval', array_column($treeRows, 'id'));
|
||||
$directMap = self::fetchDirectAdminCountMap($allIds);
|
||||
|
||||
$childrenByPid = [];
|
||||
foreach ($treeRows as $r) {
|
||||
$pid = (int) $r['pid'];
|
||||
$id = (int) $r['id'];
|
||||
if (!isset($childrenByPid[$pid])) {
|
||||
$childrenByPid[$pid] = [];
|
||||
}
|
||||
$childrenByPid[$pid][] = $id;
|
||||
}
|
||||
|
||||
$memo = [];
|
||||
$dfs = function (int $id) use (&$dfs, $childrenByPid, $directMap, &$memo): int {
|
||||
if (array_key_exists($id, $memo)) {
|
||||
return $memo[$id];
|
||||
}
|
||||
$sum = $directMap[$id] ?? 0;
|
||||
foreach ($childrenByPid[$id] ?? [] as $cid) {
|
||||
$sum += $dfs((int) $cid);
|
||||
}
|
||||
$memo[$id] = $sum;
|
||||
|
||||
return $sum;
|
||||
};
|
||||
|
||||
$out = [];
|
||||
foreach ($allIds as $id) {
|
||||
$out[$id] = $dfs($id);
|
||||
}
|
||||
|
||||
return $out;
|
||||
}
|
||||
|
||||
/**
|
||||
* 各部门直属管理员人数
|
||||
*
|
||||
* @param array<int, int> $deptIds
|
||||
* @return array<int, int>
|
||||
*/
|
||||
private static function fetchDirectAdminCountMap(array $deptIds): array
|
||||
{
|
||||
if ($deptIds === []) {
|
||||
return [];
|
||||
}
|
||||
$adminTable = (new Admin())->getTable();
|
||||
$rows = AdminDept::alias('ad')
|
||||
->join($adminTable . ' a', 'a.id = ad.admin_id')
|
||||
->whereNull('a.delete_time')
|
||||
->whereIn('ad.dept_id', $deptIds)
|
||||
->field('ad.dept_id, COUNT(*) AS admin_count')
|
||||
->group('ad.dept_id')
|
||||
->select()
|
||||
->toArray();
|
||||
$map = [];
|
||||
foreach ($rows as $r) {
|
||||
$map[(int) $r['dept_id']] = (int) $r['admin_count'];
|
||||
}
|
||||
|
||||
return $map;
|
||||
}
|
||||
|
||||
/**
|
||||
* @notes 列表树状结构
|
||||
|
||||
@@ -5,9 +5,11 @@ declare(strict_types=1);
|
||||
namespace app\adminapi\logic\qywx;
|
||||
|
||||
use app\common\logic\BaseLogic;
|
||||
use app\common\model\auth\Admin;
|
||||
use app\common\model\QywxExternalContact;
|
||||
use app\common\model\QywxSyncSettings;
|
||||
use app\common\service\wechat\WechatWorkService;
|
||||
use think\facade\Cache;
|
||||
use think\facade\Db;
|
||||
use think\facade\Log;
|
||||
|
||||
@@ -16,12 +18,140 @@ use think\facade\Log;
|
||||
*/
|
||||
class CustomerLogic extends BaseLogic
|
||||
{
|
||||
private const SYNC_LOCK_KEY = 'qywx_customer_full_sync';
|
||||
|
||||
/** @var resource|null 跨进程互斥,避免多路同步同时 UPSERT 同一表导致 1205 */
|
||||
private static $syncFileLockHandle = null;
|
||||
|
||||
/**
|
||||
* @notes 同步企业微信客户
|
||||
* Web 端「立即同步」:拉起 CLI 在后台跑,避免 HTTP/FPM/Nginx 60s 超时;重复点击会提示进行中。
|
||||
*
|
||||
* @return array<string, mixed>|false
|
||||
*/
|
||||
public static function syncCustomers()
|
||||
public static function triggerBackgroundSync()
|
||||
{
|
||||
if (Cache::get(self::SYNC_LOCK_KEY)) {
|
||||
self::$error = '同步任务正在执行中,请稍后再试或刷新页面查看状态';
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
Cache::set(self::SYNC_LOCK_KEY, time(), 7200);
|
||||
self::updateSyncStatus('syncing', 0, '', false);
|
||||
|
||||
$think = root_path() . 'think';
|
||||
if (!is_file($think)) {
|
||||
Cache::delete(self::SYNC_LOCK_KEY);
|
||||
self::$error = '未找到项目 think 入口,无法启动后台任务';
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
$phpBin = self::resolvePhpCliBinary();
|
||||
$logFile = runtime_path() . 'log/qywx_sync.log';
|
||||
$logDir = dirname($logFile);
|
||||
if (!is_dir($logDir)) {
|
||||
@mkdir($logDir, 0755, true);
|
||||
}
|
||||
|
||||
if (PHP_OS_FAMILY === 'Windows') {
|
||||
$cmd = sprintf(
|
||||
'start /B cmd /c %s %s qywx:sync-customer --force ^>^> %s 2^>^&1',
|
||||
escapeshellarg($phpBin),
|
||||
escapeshellarg($think),
|
||||
escapeshellarg($logFile)
|
||||
);
|
||||
@pclose(@popen($cmd, 'r'));
|
||||
} else {
|
||||
$cmd = sprintf(
|
||||
'%s %s qywx:sync-customer --force >> %s 2>&1 &',
|
||||
escapeshellarg($phpBin),
|
||||
escapeshellarg($think),
|
||||
escapeshellarg($logFile)
|
||||
);
|
||||
exec($cmd);
|
||||
}
|
||||
|
||||
return [
|
||||
'background' => true,
|
||||
'message' => '已在后台开始同步,请稍候刷新列表或等待状态变为「同步成功」',
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* CLI 路径:FPM 下 PHP_BINARY 可能是 php-fpm,不可用。
|
||||
*/
|
||||
private static function resolvePhpCliBinary(): string
|
||||
{
|
||||
$fromEnv = (string) env('PHP_CLI_BINARY', '');
|
||||
if ($fromEnv !== '' && is_executable($fromEnv)) {
|
||||
return $fromEnv;
|
||||
}
|
||||
if (PHP_BINARY !== '' && is_executable(PHP_BINARY) && stripos(PHP_BINARY, 'fpm') === false) {
|
||||
return PHP_BINARY;
|
||||
}
|
||||
|
||||
return 'php';
|
||||
}
|
||||
|
||||
private static function acquireSyncProcessLock(): bool
|
||||
{
|
||||
$dir = runtime_path() . 'lock';
|
||||
if (!is_dir($dir)) {
|
||||
@mkdir($dir, 0755, true);
|
||||
}
|
||||
$path = $dir . '/qywx_customer_sync.lock';
|
||||
$fp = @fopen($path, 'c+');
|
||||
if ($fp === false) {
|
||||
Log::error('qywx sync: 无法打开锁文件 ' . $path);
|
||||
|
||||
return false;
|
||||
}
|
||||
if (!flock($fp, LOCK_EX | LOCK_NB)) {
|
||||
fclose($fp);
|
||||
|
||||
return false;
|
||||
}
|
||||
self::$syncFileLockHandle = $fp;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
private static function releaseSyncProcessLock(): void
|
||||
{
|
||||
if (is_resource(self::$syncFileLockHandle)) {
|
||||
flock(self::$syncFileLockHandle, LOCK_UN);
|
||||
fclose(self::$syncFileLockHandle);
|
||||
self::$syncFileLockHandle = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @notes 同步企业微信客户(由定时任务或后台 CLI 调用,耗时可数分钟)
|
||||
*
|
||||
* @param array{follow_createtime_from?:int,follow_createtime_to?:int,follow_createtime_mode?:string} $options
|
||||
* follow_createtime_*:与 follow_createtime_mode 联用;mode 默认 first=仅「首次添加」时间在窗口内(今日新客),any=任一条跟进 createtime 在窗口内(旧逻辑,命中多)
|
||||
*/
|
||||
public static function syncCustomers(array $options = [])
|
||||
{
|
||||
try {
|
||||
@set_time_limit(0);
|
||||
} catch (\Throwable) {
|
||||
}
|
||||
|
||||
if (!self::acquireSyncProcessLock()) {
|
||||
self::$error = '已有同步任务在执行中(请勿同时运行多条 php think qywx:sync-customer 或网页「立即同步」),请等待结束后再试';
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
try {
|
||||
Db::execute('SET SESSION innodb_lock_wait_timeout = 120');
|
||||
} catch (\Throwable $e) {
|
||||
Log::warning('qywx sync: SET innodb_lock_wait_timeout ' . $e->getMessage());
|
||||
}
|
||||
|
||||
$service = new WechatWorkService();
|
||||
|
||||
// 1. 获取部门成员列表(从根部门开始,递归获取所有成员)
|
||||
@@ -40,95 +170,118 @@ class CustomerLogic extends BaseLogic
|
||||
$syncCount = 0;
|
||||
$updateCount = 0;
|
||||
$newCount = 0;
|
||||
$processedCustomers = []; // 记录已处理的客户,避免重复
|
||||
$skippedCount = 0;
|
||||
$ctimeFrom = (int) ($options['follow_createtime_from'] ?? 0);
|
||||
$ctimeTo = (int) ($options['follow_createtime_to'] ?? 0);
|
||||
$filterByFollowCreatetime = $ctimeFrom > 0 && $ctimeTo >= $ctimeFrom;
|
||||
$followCtimeMode = (string) ($options['follow_createtime_mode'] ?? 'first');
|
||||
|
||||
Db::startTrans();
|
||||
if ($filterByFollowCreatetime) {
|
||||
$tz = (string) config('app.default_timezone', 'Asia/Shanghai');
|
||||
Log::info(sprintf(
|
||||
'qywx sync: 按跟进时间筛选 mode=%s 窗口=[%s .. %s] tz=%s',
|
||||
$followCtimeMode,
|
||||
date('Y-m-d H:i:s', $ctimeFrom),
|
||||
date('Y-m-d H:i:s', $ctimeTo),
|
||||
$tz
|
||||
));
|
||||
}
|
||||
|
||||
// 不使用包层 Db 事务:同步耗时长且含大量 API 调用;模型在已有事务下会嵌套 SAVEPOINT,
|
||||
// 易导致 MySQL 1305「SAVEPOINT trans2 does not exist」。逐条写入由单次 SQL 保证原子即可。
|
||||
try {
|
||||
// 2. 遍历每个成员,获取其客户列表
|
||||
foreach ($userList as $user) {
|
||||
$userId = $user['userid'] ?? '';
|
||||
if (empty($userId)) {
|
||||
continue;
|
||||
$useBatch = (bool) config('project.qywx_sync_use_batch_detail', true);
|
||||
$mergedMap = null;
|
||||
if ($useBatch) {
|
||||
try {
|
||||
$mergedMap = self::buildMergedExternalContactsMap($service, $userList);
|
||||
Log::info('qywx sync: 批量接口合并后客户数 ' . count($mergedMap));
|
||||
} catch (\Throwable $e) {
|
||||
Log::warning('qywx sync: 批量拉取失败,回退 list+get — ' . $e->getMessage());
|
||||
$mergedMap = null;
|
||||
}
|
||||
}
|
||||
|
||||
Log::info('获取成员客户列表 - userId: ' . $userId . ', name: ' . ($user['name'] ?? ''));
|
||||
|
||||
// 获取该成员的客户列表
|
||||
$customerList = $service->getExternalContactList($userId);
|
||||
|
||||
// 如果返回false,说明API调用失败(权限问题)
|
||||
if ($customerList === false) {
|
||||
Log::error('获取客户列表失败 - 可能是API权限未开启(错误码48002)');
|
||||
throw new \Exception('企业微信API权限不足,请在企业微信管理后台开启"客户联系"权限');
|
||||
if ($mergedMap !== null) {
|
||||
foreach ($mergedMap as $bundle) {
|
||||
self::upsertOneExternalContactBundle(
|
||||
$bundle['external_contact'],
|
||||
$bundle['follow_users'],
|
||||
$filterByFollowCreatetime,
|
||||
$followCtimeMode,
|
||||
$ctimeFrom,
|
||||
$ctimeTo,
|
||||
$syncCount,
|
||||
$newCount,
|
||||
$updateCount,
|
||||
$skippedCount
|
||||
);
|
||||
}
|
||||
|
||||
Log::info('成员 ' . $userId . ' 的客户数量: ' . count($customerList));
|
||||
|
||||
if (empty($customerList)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// 3. 遍历客户列表,获取详情并保存
|
||||
foreach ($customerList as $externalUserId) {
|
||||
// 避免重复处理同一个客户
|
||||
if (isset($processedCustomers[$externalUserId])) {
|
||||
continue;
|
||||
}
|
||||
$processedCustomers[$externalUserId] = true;
|
||||
|
||||
// 获取客户详情
|
||||
$detail = $service->getExternalContactDetail($externalUserId);
|
||||
if (empty($detail)) {
|
||||
} else {
|
||||
$processedCustomers = [];
|
||||
foreach ($userList as $user) {
|
||||
$userId = $user['userid'] ?? '';
|
||||
if (empty($userId)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
$externalContact = $detail['external_contact'] ?? [];
|
||||
$followUsers = $detail['follow_user'] ?? [];
|
||||
Log::info('获取成员客户列表 - userId: ' . $userId . ', name: ' . ($user['name'] ?? ''));
|
||||
|
||||
// 保存或更新客户信息
|
||||
$data = [
|
||||
'external_userid' => $externalContact['external_userid'] ?? '',
|
||||
'name' => $externalContact['name'] ?? '',
|
||||
'avatar' => $externalContact['avatar'] ?? '',
|
||||
'type' => $externalContact['type'] ?? 1,
|
||||
'gender' => $externalContact['gender'] ?? 0,
|
||||
'unionid' => $externalContact['unionid'] ?? '',
|
||||
'position' => $externalContact['position'] ?? '',
|
||||
'corp_name' => $externalContact['corp_name'] ?? '',
|
||||
'corp_full_name' => $externalContact['corp_full_name'] ?? '',
|
||||
'external_profile' => json_encode($externalContact['external_profile'] ?? [], JSON_UNESCAPED_UNICODE),
|
||||
'follow_users' => json_encode($followUsers, JSON_UNESCAPED_UNICODE),
|
||||
'update_time' => time(),
|
||||
];
|
||||
$customerList = $service->getExternalContactList($userId);
|
||||
|
||||
$existing = QywxExternalContact::where('external_userid', $data['external_userid'])->find();
|
||||
if ($existing) {
|
||||
$existing->save($data);
|
||||
$updateCount++;
|
||||
} else {
|
||||
$data['create_time'] = time();
|
||||
QywxExternalContact::create($data);
|
||||
$newCount++;
|
||||
if ($customerList === false) {
|
||||
Log::error('获取客户列表失败 - 可能是API权限未开启(错误码48002)');
|
||||
throw new \Exception('企业微信API权限不足,请在企业微信管理后台开启"客户联系"权限');
|
||||
}
|
||||
|
||||
$syncCount++;
|
||||
Log::info('成员 ' . $userId . ' 的客户数量: ' . count($customerList));
|
||||
|
||||
if (empty($customerList)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
foreach ($customerList as $externalUserId) {
|
||||
if (isset($processedCustomers[$externalUserId])) {
|
||||
continue;
|
||||
}
|
||||
$processedCustomers[$externalUserId] = true;
|
||||
|
||||
$detail = $service->getExternalContactDetail($externalUserId);
|
||||
if (empty($detail)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
$externalContact = $detail['external_contact'] ?? [];
|
||||
$followUsers = $detail['follow_user'] ?? [];
|
||||
|
||||
self::upsertOneExternalContactBundle(
|
||||
$externalContact,
|
||||
is_array($followUsers) ? $followUsers : [],
|
||||
$filterByFollowCreatetime,
|
||||
$followCtimeMode,
|
||||
$ctimeFrom,
|
||||
$ctimeTo,
|
||||
$syncCount,
|
||||
$newCount,
|
||||
$updateCount,
|
||||
$skippedCount
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 4. 更新同步设置
|
||||
self::updateSyncStatus('success', $syncCount);
|
||||
|
||||
Db::commit();
|
||||
|
||||
Log::info('同步完成 - 总数: ' . $syncCount . ', 新增: ' . $newCount . ', 更新: ' . $updateCount);
|
||||
|
||||
return [
|
||||
'sync_count' => $syncCount,
|
||||
'new_count' => $newCount,
|
||||
'update_count' => $updateCount,
|
||||
'skipped_count' => $skippedCount,
|
||||
];
|
||||
} catch (\Throwable $e) {
|
||||
Db::rollback();
|
||||
Log::error('同步企业微信客户失败: ' . $e->getMessage());
|
||||
self::$error = '同步失败: ' . $e->getMessage();
|
||||
self::updateSyncStatus('failed', 0, $e->getMessage());
|
||||
@@ -139,6 +292,323 @@ class CustomerLogic extends BaseLogic
|
||||
self::$error = '同步异常: ' . $e->getMessage();
|
||||
self::updateSyncStatus('failed', 0, $e->getMessage());
|
||||
return false;
|
||||
} finally {
|
||||
self::releaseSyncProcessLock();
|
||||
Cache::delete(self::SYNC_LOCK_KEY);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 企微 follow_user 单条里的添加时间,统一为 Unix 秒(兼容毫秒、create_time 字段名)
|
||||
*
|
||||
* @param array<string, mixed> $fu
|
||||
*/
|
||||
public static function normalizeFollowUserCreatetime(array $fu): int
|
||||
{
|
||||
$raw = $fu['createtime'] ?? $fu['create_time'] ?? 0;
|
||||
$ct = (int) $raw;
|
||||
if ($ct > 1_000_000_000_000) {
|
||||
$ct = intdiv($ct, 1000);
|
||||
}
|
||||
|
||||
return $ct;
|
||||
}
|
||||
|
||||
/**
|
||||
* follow_user[].createtime 最小值:企微侧「首次添加为外部联系人」时间(Unix 秒),无则 0
|
||||
*
|
||||
* @param array<int, mixed> $followUsers
|
||||
*/
|
||||
public static function minFollowCreatetime(array $followUsers): int
|
||||
{
|
||||
$min = 0;
|
||||
foreach ($followUsers as $fu) {
|
||||
if (!is_array($fu)) {
|
||||
continue;
|
||||
}
|
||||
$ct = self::normalizeFollowUserCreatetime($fu);
|
||||
if ($ct > 0 && ($min === 0 || $ct < $min)) {
|
||||
$min = $ct;
|
||||
}
|
||||
}
|
||||
|
||||
return $min;
|
||||
}
|
||||
|
||||
/**
|
||||
* 「今日」起止 Unix 秒(闭区间结束为次日 0 秒前一秒),使用 app.default_timezone
|
||||
*
|
||||
* @return array{0:int,1:int}
|
||||
*/
|
||||
public static function todayCreatetimeWindowBounds(): array
|
||||
{
|
||||
$tzName = (string) config('app.default_timezone', 'Asia/Shanghai');
|
||||
$tz = new \DateTimeZone($tzName);
|
||||
$from = (new \DateTime('today', $tz))->getTimestamp();
|
||||
$to = (new \DateTime('tomorrow', $tz))->getTimestamp() - 1;
|
||||
|
||||
return [$from, $to];
|
||||
}
|
||||
|
||||
/**
|
||||
* 客户详情 follow_user 中 createtime 是否有任一条落在 [from, to](闭区间,Unix 秒)
|
||||
*
|
||||
* @param array<int, mixed> $followUsers
|
||||
*/
|
||||
private static function followCreatetimeInRange(array $followUsers, int $from, int $to): bool
|
||||
{
|
||||
foreach ($followUsers as $fu) {
|
||||
if (!is_array($fu)) {
|
||||
continue;
|
||||
}
|
||||
$ct = self::normalizeFollowUserCreatetime($fu);
|
||||
if ($ct >= $from && $ct <= $to) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* 使用 batch/get_by_user 按成员分页拉取并合并跟进人(列表接口无按时间筛选)
|
||||
*
|
||||
* @param array<int, array<string, mixed>> $userList
|
||||
* @return array<string, array{external_contact: array<string, mixed>, follow_users: array<int, array<string, mixed>>}>
|
||||
*/
|
||||
private static function buildMergedExternalContactsMap(WechatWorkService $service, array $userList): array
|
||||
{
|
||||
$merged = [];
|
||||
foreach ($userList as $user) {
|
||||
$userId = trim((string) ($user['userid'] ?? ''));
|
||||
if ($userId === '') {
|
||||
continue;
|
||||
}
|
||||
Log::info('qywx sync(batch): 成员 ' . $userId . ' ' . ($user['name'] ?? ''));
|
||||
$cursor = '';
|
||||
do {
|
||||
$resp = $service->batchGetExternalContactByUser($userId, $cursor, 100);
|
||||
$err = (int) ($resp['errcode'] ?? -1);
|
||||
if ($err !== 0) {
|
||||
throw new \RuntimeException('errcode=' . $err . ' errmsg=' . (string) ($resp['errmsg'] ?? ''));
|
||||
}
|
||||
foreach ($resp['external_contact_list'] ?? [] as $item) {
|
||||
if (is_array($item)) {
|
||||
self::mergeBatchExternalItemIntoMap($merged, $item);
|
||||
}
|
||||
}
|
||||
$cursor = (string) ($resp['next_cursor'] ?? '');
|
||||
} while ($cursor !== '');
|
||||
}
|
||||
|
||||
return $merged;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string, array{external_contact: array<string, mixed>, follow_users: array<int, array<string, mixed>>}> $merged
|
||||
* @param array<string, mixed> $item
|
||||
*/
|
||||
private static function mergeBatchExternalItemIntoMap(array &$merged, array $item): void
|
||||
{
|
||||
$ec = $item['external_contact'] ?? [];
|
||||
if (!is_array($ec)) {
|
||||
$ec = [];
|
||||
}
|
||||
$fi = $item['follow_info'] ?? [];
|
||||
if (!is_array($fi)) {
|
||||
$fi = [];
|
||||
}
|
||||
$extId = trim((string) ($ec['external_userid'] ?? ''));
|
||||
if ($extId === '') {
|
||||
return;
|
||||
}
|
||||
if (!isset($merged[$extId])) {
|
||||
$merged[$extId] = [
|
||||
'external_contact' => $ec,
|
||||
'follow_users' => [],
|
||||
];
|
||||
} else {
|
||||
$merged[$extId]['external_contact'] = self::mergeExternalContactShallow(
|
||||
$merged[$extId]['external_contact'],
|
||||
$ec
|
||||
);
|
||||
}
|
||||
$uid = trim((string) ($fi['userid'] ?? ''));
|
||||
if ($uid === '') {
|
||||
return;
|
||||
}
|
||||
foreach ($merged[$extId]['follow_users'] as $ex) {
|
||||
if (is_array($ex) && (($ex['userid'] ?? '') === $uid)) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
$merged[$extId]['follow_users'][] = $fi;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string, mixed> $base
|
||||
* @param array<string, mixed> $in
|
||||
* @return array<string, mixed>
|
||||
*/
|
||||
private static function mergeExternalContactShallow(array $base, array $in): array
|
||||
{
|
||||
foreach ($in as $k => $v) {
|
||||
if ($v === null || $v === '') {
|
||||
continue;
|
||||
}
|
||||
if (!array_key_exists($k, $base) || $base[$k] === '' || $base[$k] === null) {
|
||||
$base[$k] = $v;
|
||||
}
|
||||
}
|
||||
|
||||
return $base;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string, mixed> $externalContact
|
||||
* @param array<int, array<string, mixed>> $followUsers
|
||||
*/
|
||||
private static function upsertOneExternalContactBundle(
|
||||
array $externalContact,
|
||||
array $followUsers,
|
||||
bool $filterByFollowCreatetime,
|
||||
string $followCtimeMode,
|
||||
int $ctimeFrom,
|
||||
int $ctimeTo,
|
||||
int &$syncCount,
|
||||
int &$newCount,
|
||||
int &$updateCount,
|
||||
int &$skippedCount
|
||||
): void {
|
||||
if ($filterByFollowCreatetime) {
|
||||
if ($followCtimeMode === 'any') {
|
||||
$pass = self::followCreatetimeInRange($followUsers, $ctimeFrom, $ctimeTo);
|
||||
} else {
|
||||
$minCt = self::minFollowCreatetime($followUsers);
|
||||
$pass = $minCt >= $ctimeFrom && $minCt <= $ctimeTo;
|
||||
}
|
||||
if (!$pass) {
|
||||
$skippedCount++;
|
||||
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
$now = time();
|
||||
$row = [
|
||||
'external_userid' => $externalContact['external_userid'] ?? '',
|
||||
'name' => $externalContact['name'] ?? '',
|
||||
'avatar' => $externalContact['avatar'] ?? '',
|
||||
'type' => $externalContact['type'] ?? 1,
|
||||
'gender' => $externalContact['gender'] ?? 0,
|
||||
'unionid' => $externalContact['unionid'] ?? '',
|
||||
'position' => $externalContact['position'] ?? '',
|
||||
'corp_name' => $externalContact['corp_name'] ?? '',
|
||||
'corp_full_name' => $externalContact['corp_full_name'] ?? '',
|
||||
'external_profile' => json_encode($externalContact['external_profile'] ?? [], JSON_UNESCAPED_UNICODE),
|
||||
'follow_users' => json_encode($followUsers, JSON_UNESCAPED_UNICODE),
|
||||
'follow_admin_ids' => self::resolveFollowAdminIds($followUsers),
|
||||
'external_first_add_time' => self::minFollowCreatetime($followUsers),
|
||||
'create_time' => $now,
|
||||
'update_time' => $now,
|
||||
'delete_time' => null,
|
||||
];
|
||||
|
||||
self::upsertExternalContactRow($row, $syncCount, $newCount, $updateCount);
|
||||
if (($syncCount % 25) === 0) {
|
||||
usleep(8000);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将企微 follow_user[].userid 与后台 admin.work_wechat_userid 对齐,得到 zyt_admin.id 列表 JSON。
|
||||
*
|
||||
* @param array<int, mixed> $followUsers
|
||||
*/
|
||||
private static function resolveFollowAdminIds(array $followUsers): string
|
||||
{
|
||||
$wxUserids = [];
|
||||
foreach ($followUsers as $fu) {
|
||||
if (!is_array($fu)) {
|
||||
continue;
|
||||
}
|
||||
$uid = trim((string) ($fu['userid'] ?? ''));
|
||||
if ($uid !== '') {
|
||||
$wxUserids[] = $uid;
|
||||
}
|
||||
}
|
||||
$wxUserids = array_values(array_unique($wxUserids));
|
||||
if ($wxUserids === []) {
|
||||
return '[]';
|
||||
}
|
||||
|
||||
$idByWx = Admin::whereIn('work_wechat_userid', $wxUserids)->column('id', 'work_wechat_userid');
|
||||
$adminIds = [];
|
||||
foreach ($wxUserids as $wx) {
|
||||
if (!empty($idByWx[$wx])) {
|
||||
$adminIds[] = (int) $idByWx[$wx];
|
||||
}
|
||||
}
|
||||
|
||||
return json_encode(array_values(array_unique($adminIds)), JSON_UNESCAPED_UNICODE);
|
||||
}
|
||||
|
||||
/**
|
||||
* 是否 MySQL 锁等待超时 / 死锁(可短重试)
|
||||
*/
|
||||
private static function isMysqlLockTimeoutOrDeadlock(\Throwable $e): bool
|
||||
{
|
||||
$m = $e->getMessage();
|
||||
|
||||
return str_contains($m, '1205') || str_contains($m, '1213') || str_contains($m, 'Deadlock');
|
||||
}
|
||||
|
||||
/**
|
||||
* INSERT ... ON DUPLICATE KEY UPDATE + 锁冲突重试
|
||||
*
|
||||
* @param array<string, mixed> $row
|
||||
*/
|
||||
private static function upsertExternalContactRow(array $row, int &$syncCount, int &$newCount, int &$updateCount): void
|
||||
{
|
||||
$maxAttempts = 12;
|
||||
$attempt = 0;
|
||||
while (true) {
|
||||
$attempt++;
|
||||
try {
|
||||
$affected = Db::name('qywx_external_contact')->duplicate([
|
||||
'name',
|
||||
'avatar',
|
||||
'type',
|
||||
'gender',
|
||||
'unionid',
|
||||
'position',
|
||||
'corp_name',
|
||||
'corp_full_name',
|
||||
'external_profile',
|
||||
'follow_users',
|
||||
'follow_admin_ids',
|
||||
'external_first_add_time',
|
||||
'update_time',
|
||||
'delete_time',
|
||||
])->insert($row);
|
||||
|
||||
$syncCount++;
|
||||
// PDO MySQL:新插入 1;更新 2;值完全相同 0
|
||||
if ($affected === 1) {
|
||||
$newCount++;
|
||||
} elseif ($affected === 2) {
|
||||
$updateCount++;
|
||||
} elseif ($affected === 0) {
|
||||
$updateCount++;
|
||||
}
|
||||
|
||||
return;
|
||||
} catch (\Throwable $e) {
|
||||
if ($attempt >= $maxAttempts || !self::isMysqlLockTimeoutOrDeadlock($e)) {
|
||||
throw $e;
|
||||
}
|
||||
usleep(min(800_000, 40_000 * (2 ** ($attempt - 1))));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -150,8 +620,12 @@ class CustomerLogic extends BaseLogic
|
||||
$total = QywxExternalContact::whereNull('delete_time')->count();
|
||||
|
||||
$todayStart = strtotime(date('Y-m-d 00:00:00'));
|
||||
$today = QywxExternalContact::where('create_time', '>=', $todayStart)
|
||||
->whereNull('delete_time')
|
||||
// 今日新增:以企微首次添加时间为准;历史行未回填字段(0)时退回 create_time
|
||||
$today = QywxExternalContact::whereNull('delete_time')
|
||||
->whereRaw(
|
||||
'(external_first_add_time >= ? OR (external_first_add_time = 0 AND create_time >= ?))',
|
||||
[$todayStart, $todayStart]
|
||||
)
|
||||
->count();
|
||||
|
||||
$settings = self::getSyncSettings();
|
||||
@@ -225,24 +699,37 @@ class CustomerLogic extends BaseLogic
|
||||
}
|
||||
|
||||
/**
|
||||
* @notes 更新同步状态
|
||||
* @param bool $bumpLastSyncTime 为 false 时仅改状态(用于 syncing),避免「最后同步」在跑任务期间被刷新
|
||||
*/
|
||||
private static function updateSyncStatus(string $status, int $syncCount = 0, string $error = '')
|
||||
private static function updateSyncStatus(string $status, int $syncCount = 0, string $error = '', bool $bumpLastSyncTime = true)
|
||||
{
|
||||
try {
|
||||
$settings = self::getSyncSettings();
|
||||
$settings['sync_status'] = $status;
|
||||
$settings['last_sync_time'] = time();
|
||||
$settings['last_sync_count'] = $syncCount;
|
||||
if ($bumpLastSyncTime) {
|
||||
$settings['last_sync_time'] = time();
|
||||
$settings['last_sync_count'] = $syncCount;
|
||||
}
|
||||
if ($error) {
|
||||
$settings['last_error'] = $error;
|
||||
}
|
||||
|
||||
$setting = QywxSyncSettings::where('setting_key', 'customer_sync')->find();
|
||||
if ($setting) {
|
||||
$setting->setting_value = json_encode($settings, JSON_UNESCAPED_UNICODE);
|
||||
$setting->update_time = time();
|
||||
$setting->save();
|
||||
$payload = json_encode($settings, JSON_UNESCAPED_UNICODE);
|
||||
$now = time();
|
||||
// 直连 UPDATE,避免模型层隐式事务/事件拉长锁时间
|
||||
$n = Db::name('qywx_sync_settings')
|
||||
->where('setting_key', 'customer_sync')
|
||||
->update([
|
||||
'setting_value' => $payload,
|
||||
'update_time' => $now,
|
||||
]);
|
||||
if ($n === 0) {
|
||||
Db::name('qywx_sync_settings')->insert([
|
||||
'setting_key' => 'customer_sync',
|
||||
'setting_value' => $payload,
|
||||
'create_time' => $now,
|
||||
'update_time' => $now,
|
||||
]);
|
||||
}
|
||||
} catch (\Throwable $e) {
|
||||
Log::error('更新同步状态失败: ' . $e->getMessage());
|
||||
|
||||
Reference in New Issue
Block a user