1676 lines
64 KiB
PHP
Executable File
1676 lines
64 KiB
PHP
Executable File
<?php
|
||
|
||
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;
|
||
|
||
/**
|
||
* 企业微信客户逻辑层
|
||
*/
|
||
class CustomerLogic extends BaseLogic
|
||
{
|
||
private const SYNC_LOCK_KEY = 'qywx_customer_full_sync';
|
||
|
||
/** @var resource|null 跨进程互斥，避免多路同步同时 UPSERT 同一表导致 1205 */
|
||
private static $syncFileLockHandle = null;
|
||
|
||
/**
|
||
* Web 端「立即同步」：拉起 CLI 在后台跑，避免 HTTP/FPM/Nginx 60s 超时；重复点击会提示进行中。
|
||
*
|
||
* @return array<string, mixed>|false
|
||
*/
|
||
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. 获取部门成员列表（从根部门开始，递归获取所有成员）
|
||
$departmentId = config('project.qywx_sync_department_id', 1);
|
||
Log::info('开始同步企业微信客户 - 部门ID: ' . $departmentId);
|
||
|
||
$userList = $service->getDepartmentUserList($departmentId, true);
|
||
Log::info('获取到企业成员数量: ' . count($userList));
|
||
|
||
if (empty($userList)) {
|
||
Log::error('未获取到企业成员列表');
|
||
self::$error = '未获取到企业成员列表';
|
||
return false;
|
||
}
|
||
|
||
$syncCount = 0;
|
||
$updateCount = 0;
|
||
$newCount = 0;
|
||
$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');
|
||
|
||
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 {
|
||
$useBatch = (bool) config('project.qywx_sync_use_batch_detail', true);
|
||
$mergedMap = null;
|
||
/** 已走「时间筛选 + 流式 batch」分支且未抛错时，不再 list+get 全量回退 */
|
||
$createtimeFilterBatchPathDone = false;
|
||
if ($useBatch) {
|
||
try {
|
||
// 按跟进时间筛选：不构建全量合并表（10 万+ 会爆内存/慢）；流式扫 batch，仅对候选 id get+写库
|
||
if ($filterByFollowCreatetime) {
|
||
$candidates = self::collectCandidateExternalUserIdsForCreatetimeWindow(
|
||
$service,
|
||
$userList,
|
||
$ctimeFrom,
|
||
$ctimeTo,
|
||
$followCtimeMode
|
||
);
|
||
Log::info('qywx sync: 时间筛选模式 候选客户数(待 get 校验) ' . count($candidates));
|
||
foreach (array_keys($candidates) as $extId) {
|
||
$detail = $service->getExternalContactDetail($extId);
|
||
if (empty($detail)) {
|
||
continue;
|
||
}
|
||
$externalContact = $detail['external_contact'] ?? [];
|
||
$followUsers = $detail['follow_user'] ?? [];
|
||
self::upsertOneExternalContactBundle(
|
||
$externalContact,
|
||
is_array($followUsers) ? $followUsers : [],
|
||
true,
|
||
$followCtimeMode,
|
||
$ctimeFrom,
|
||
$ctimeTo,
|
||
$syncCount,
|
||
$newCount,
|
||
$updateCount,
|
||
$skippedCount
|
||
);
|
||
}
|
||
$createtimeFilterBatchPathDone = true;
|
||
} else {
|
||
$mergedMap = self::buildMergedExternalContactsMap($service, $userList);
|
||
Log::info('qywx sync: 批量接口合并后客户数 ' . count($mergedMap));
|
||
}
|
||
} catch (\Throwable $e) {
|
||
Log::warning('qywx sync: 批量拉取失败，回退 list+get — ' . $e->getMessage());
|
||
$mergedMap = null;
|
||
}
|
||
}
|
||
|
||
if ($mergedMap !== null) {
|
||
foreach ($mergedMap as $bundle) {
|
||
self::upsertOneExternalContactBundle(
|
||
$bundle['external_contact'],
|
||
$bundle['follow_users'],
|
||
$filterByFollowCreatetime,
|
||
$followCtimeMode,
|
||
$ctimeFrom,
|
||
$ctimeTo,
|
||
$syncCount,
|
||
$newCount,
|
||
$updateCount,
|
||
$skippedCount
|
||
);
|
||
}
|
||
} elseif (!$createtimeFilterBatchPathDone) {
|
||
$processedCustomers = [];
|
||
foreach ($userList as $user) {
|
||
$userId = $user['userid'] ?? '';
|
||
if (empty($userId)) {
|
||
continue;
|
||
}
|
||
|
||
Log::info('获取成员客户列表 - userId: ' . $userId . ', name: ' . ($user['name'] ?? ''));
|
||
|
||
$customerList = $service->getExternalContactList($userId);
|
||
|
||
if ($customerList === false) {
|
||
Log::error('获取客户列表失败 - 可能是API权限未开启（错误码48002）');
|
||
throw new \Exception('企业微信API权限不足，请在企业微信管理后台开启"客户联系"权限');
|
||
}
|
||
|
||
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);
|
||
|
||
Log::info('同步完成 - 总数: ' . $syncCount . ', 新增: ' . $newCount . ', 更新: ' . $updateCount);
|
||
|
||
return [
|
||
'sync_count' => $syncCount,
|
||
'new_count' => $newCount,
|
||
'update_count' => $updateCount,
|
||
'skipped_count' => $skippedCount,
|
||
];
|
||
} catch (\Throwable $e) {
|
||
Log::error('同步企业微信客户失败: ' . $e->getMessage());
|
||
self::$error = '同步失败: ' . $e->getMessage();
|
||
self::updateSyncStatus('failed', 0, $e->getMessage());
|
||
return false;
|
||
}
|
||
} catch (\Throwable $e) {
|
||
Log::error('同步企业微信客户异常: ' . $e->getMessage());
|
||
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，按单条 follow_info 做时间窗口启发式，收集可能要落库的 external_userid（去重）。
|
||
* 企微无「只拉今日」接口，仍需翻完所有分页；但不构建全量合并表，后续仅对候选调用 get 做精确过滤。
|
||
*
|
||
* @return array<string, true>
|
||
*/
|
||
private static function collectCandidateExternalUserIdsForCreatetimeWindow(
|
||
WechatWorkService $service,
|
||
array $userList,
|
||
int $ctimeFrom,
|
||
int $ctimeTo,
|
||
string $followCtimeMode
|
||
): array {
|
||
$candidates = [];
|
||
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)) {
|
||
continue;
|
||
}
|
||
$fi = $item['follow_info'] ?? [];
|
||
if (!is_array($fi)) {
|
||
$fi = [];
|
||
}
|
||
if (!self::batchFollowInfoHintMayMatchCreatetimeWindow($fi, $ctimeFrom, $ctimeTo, $followCtimeMode)) {
|
||
continue;
|
||
}
|
||
$ec = $item['external_contact'] ?? [];
|
||
$extId = trim((string) ($ec['external_userid'] ?? ''));
|
||
if ($extId !== '') {
|
||
$candidates[$extId] = true;
|
||
}
|
||
}
|
||
$cursor = (string) ($resp['next_cursor'] ?? '');
|
||
} while ($cursor !== '');
|
||
}
|
||
|
||
return $candidates;
|
||
}
|
||
|
||
/**
|
||
* 批量接口单条 follow_info 是否「可能」命中时间策略（用于缩小候选；最终以 get 全量 follow_user 为准）
|
||
*
|
||
* @param array<string, mixed> $followInfo
|
||
*/
|
||
private static function batchFollowInfoHintMayMatchCreatetimeWindow(
|
||
array $followInfo,
|
||
int $ctimeFrom,
|
||
int $ctimeTo,
|
||
string $followCtimeMode
|
||
): bool {
|
||
if ($followCtimeMode === 'any') {
|
||
return self::followCreatetimeInRange([$followInfo], $ctimeFrom, $ctimeTo);
|
||
}
|
||
$ct = self::normalizeFollowUserCreatetime($followInfo);
|
||
if ($ct > 0 && $ct < $ctimeFrom) {
|
||
return false;
|
||
}
|
||
if ($ct > 0 && $ct > $ctimeTo) {
|
||
return false;
|
||
}
|
||
|
||
return true;
|
||
}
|
||
|
||
/**
|
||
* 使用 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
|
||
*/
|
||
/**
|
||
* 客户联系事件回调：按 external_userid 拉取详情并 UPSERT（不走全量同步文件锁）。
|
||
*
|
||
* @param bool $isAddEvent 是否 add_external_contact 事件：是且本地已有该客户（含软删行）时，
|
||
* 新增一行 readd_flag=1（以前加过）的记录，原有行不动
|
||
* @param string $eventUserId 本次添加人的企微 userid（add 事件回调里的 UserID，用于取本次添加时间）
|
||
* @param int $eventTime 回调事件时间（企微 CreateTime，秒）
|
||
* @see https://developer.work.weixin.qq.com/document/path/92130
|
||
*/
|
||
public static function upsertSingleExternalContactFromApi(
|
||
string $externalUserId,
|
||
bool $isAddEvent = false,
|
||
string $eventUserId = '',
|
||
int $eventTime = 0
|
||
): void {
|
||
$externalUserId = trim($externalUserId);
|
||
$eventUserId = trim($eventUserId);
|
||
if ($externalUserId === '') {
|
||
return;
|
||
}
|
||
|
||
// 必须在写库之前判断"是否已存在"，否则无法区分新客户和重加客户
|
||
$existedBefore = false;
|
||
if ($isAddEvent) {
|
||
$existedBefore = Db::name('qywx_external_contact')
|
||
->where('external_userid', $externalUserId)
|
||
->count() > 0;
|
||
}
|
||
|
||
$service = new WechatWorkService();
|
||
$detail = [];
|
||
$lastError = null;
|
||
// 接口偶发抖动：最多重试 1 次（共 2 次请求），间隔 1.2s。仅对「非预期错误 / 空响应」重试，
|
||
// 业务预期错误（如 84061 客户未通过、60011/60111 无权限、41068 非法 ID）不重试，避免无谓压力。
|
||
$maxAttempts = 2;
|
||
for ($attempt = 1; $attempt <= $maxAttempts; $attempt++) {
|
||
$err = null;
|
||
$detail = $service->getExternalContactDetail($externalUserId, $err);
|
||
if (!empty($detail['external_contact']) && is_array($detail['external_contact'])) {
|
||
break;
|
||
}
|
||
$lastError = $err;
|
||
$errcode = is_array($err) ? ($err['errcode'] ?? null) : null;
|
||
$shouldRetry = $attempt < $maxAttempts
|
||
&& !in_array($errcode, [84061, 41068, 60011, 60111], true);
|
||
if (!$shouldRetry) {
|
||
break;
|
||
}
|
||
usleep(1_200_000);
|
||
}
|
||
|
||
if (empty($detail['external_contact']) || !is_array($detail['external_contact'])) {
|
||
$errcode = is_array($lastError) ? ($lastError['errcode'] ?? null) : null;
|
||
$errmsg = is_array($lastError) ? (string) ($lastError['errmsg'] ?? '') : '';
|
||
// 84061 = 已不是企业客户（双向解除/员工删除后无人跟进）：与企微对齐，本地软删。
|
||
// 注意：客户单方删除员工（del_follow_user）时企微仍保留客户，get 会成功返回，不会走到这里。
|
||
if ($errcode === 84061) {
|
||
self::softDeleteExternalContactRow($externalUserId);
|
||
Log::info('qywx external contact callback: 企微侧已不存在(84061)，本地软删 ext=' . $externalUserId);
|
||
|
||
return;
|
||
}
|
||
Log::warning(sprintf(
|
||
'qywx external contact callback: 拉取详情为空，跳过 UPSERT errcode=%s errmsg=%s ext=%s',
|
||
$errcode === null ? '?' : (string) $errcode,
|
||
$errmsg !== '' ? $errmsg : '-',
|
||
$externalUserId
|
||
));
|
||
|
||
return;
|
||
}
|
||
|
||
$followUsers = $detail['follow_user'] ?? [];
|
||
if (!is_array($followUsers)) {
|
||
$followUsers = [];
|
||
}
|
||
|
||
// 重复添加：不动原有行，新增一行并打「以前加过」标记（与企微"每次添加都算一条"对齐）
|
||
if ($isAddEvent && $existedBefore) {
|
||
self::insertReaddContactRow(
|
||
$detail['external_contact'],
|
||
$followUsers,
|
||
$eventUserId,
|
||
$eventTime
|
||
);
|
||
|
||
return;
|
||
}
|
||
|
||
$syncCount = 0;
|
||
$newCount = 0;
|
||
$updateCount = 0;
|
||
$skippedCount = 0;
|
||
self::upsertOneExternalContactBundle(
|
||
$detail['external_contact'],
|
||
$followUsers,
|
||
false,
|
||
'first',
|
||
0,
|
||
0,
|
||
$syncCount,
|
||
$newCount,
|
||
$updateCount,
|
||
$skippedCount
|
||
);
|
||
}
|
||
|
||
/**
|
||
* 重复添加：为已存在的客户新增一行记录（原有行保持不变）。
|
||
*
|
||
* 新行的「添加时间」取本次添加动作的时间：优先本次添加人（$eventUserId）在 follow_user
|
||
* 里的 createtime，其次回调事件时间，最后兜底当前时间——保证这行出现在"本次添加"的日期下。
|
||
*
|
||
* @param array<string, mixed> $externalContact
|
||
* @param array<int, array<string, mixed>> $followUsers
|
||
*/
|
||
private static function insertReaddContactRow(
|
||
array $externalContact,
|
||
array $followUsers,
|
||
string $eventUserId,
|
||
int $eventTime
|
||
): void {
|
||
$row = self::buildContactRow($externalContact, $followUsers);
|
||
|
||
$addTime = 0;
|
||
if ($eventUserId !== '') {
|
||
foreach ($followUsers as $fu) {
|
||
if (is_array($fu) && trim((string) ($fu['userid'] ?? '')) === $eventUserId) {
|
||
$addTime = self::normalizeFollowUserCreatetime($fu);
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
if ($addTime <= 0) {
|
||
$addTime = $eventTime > 0 ? $eventTime : time();
|
||
}
|
||
|
||
$row['external_first_add_time'] = $addTime;
|
||
$row['readd_flag'] = 1;
|
||
|
||
Db::name('qywx_external_contact')->insert($row);
|
||
|
||
$extId = (string) $row['external_userid'];
|
||
self::syncContactTagsRelation($extId, $followUsers);
|
||
self::syncContactFollowRelation($extId, $followUsers);
|
||
|
||
Log::info('qywx external contact callback: 重复添加，新增一行(readd_flag=1) ext=' . $extId);
|
||
}
|
||
|
||
/**
|
||
* 事件流水（零误差进入计数）
|
||
*
|
||
* 每次 change_external_contact 回调都调用一次，`INSERT IGNORE` 幂等：
|
||
* - 企微对非 2xx 响应最多重试 3 次，同一 (change_type,user_id,external_userid,event_time) 只会计一行；
|
||
* - 与业务 UPSERT 解耦，即便后续 DB 逻辑抛错也不影响"今日进入数"统计。
|
||
*
|
||
* 入参 $data：
|
||
* - change_type string
|
||
* - user_id string
|
||
* - external_userid string
|
||
* - state string
|
||
* - fail_reason string
|
||
* - welcome_code int 0/1（只存是否带欢迎码，不落原 code，避免泄漏）
|
||
* - event_time int 企微 CreateTime（秒）
|
||
* - raw mixed 原始消息（用于留痕；非标量会 json_encode）
|
||
*
|
||
* @param array<string, mixed> $data
|
||
*/
|
||
public static function recordExternalContactEvent(array $data): void
|
||
{
|
||
$changeType = (string) ($data['change_type'] ?? '');
|
||
if ($changeType === '') {
|
||
// 没有 ChangeType 的事件流水没有价值，直接丢弃
|
||
return;
|
||
}
|
||
|
||
$eventTime = (int) ($data['event_time'] ?? 0);
|
||
if ($eventTime <= 0) {
|
||
$eventTime = time();
|
||
}
|
||
|
||
$raw = $data['raw'] ?? null;
|
||
if ($raw !== null && !is_string($raw)) {
|
||
try {
|
||
$raw = json_encode($raw, JSON_UNESCAPED_UNICODE | JSON_PARTIAL_OUTPUT_ON_ERROR);
|
||
} catch (\Throwable) {
|
||
$raw = null;
|
||
}
|
||
}
|
||
|
||
$row = [
|
||
'change_type' => $changeType,
|
||
'user_id' => (string) ($data['user_id'] ?? ''),
|
||
'external_userid' => (string) ($data['external_userid'] ?? ''),
|
||
'state' => mb_substr((string) ($data['state'] ?? ''), 0, 128),
|
||
'fail_reason' => mb_substr((string) ($data['fail_reason'] ?? ''), 0, 64),
|
||
'welcome_code' => !empty($data['welcome_code']) ? 1 : 0,
|
||
'event_time' => $eventTime,
|
||
'create_time' => time(),
|
||
'raw' => $raw,
|
||
];
|
||
|
||
try {
|
||
// INSERT IGNORE：命中唯一键 (change_type,user_id,external_userid,event_time) 时静默跳过。
|
||
// ThinkPHP 未提供统一的 ignore helper，这里直接用原生 SQL（与 SyncImChatArchive 的做法一致）。
|
||
$cols = array_keys($row);
|
||
$table = config('database.connections.mysql.prefix') . 'qywx_external_contact_event';
|
||
$sql = 'INSERT IGNORE INTO `' . $table . '` (`' . implode('`,`', $cols) . '`) VALUES ('
|
||
. implode(',', array_fill(0, count($cols), '?')) . ')';
|
||
Db::execute($sql, array_values($row));
|
||
} catch (\Throwable $e) {
|
||
// 事件流水只用于统计，失败只记日志不阻塞主回调
|
||
Log::warning('qywx external contact event insert failed: ' . $e->getMessage());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 客户彻底流失（企微 get 返回 84061）：该客户的所有行（含历史重加行）一并软删。
|
||
*/
|
||
public static function softDeleteExternalContactRow(string $externalUserId): void
|
||
{
|
||
$externalUserId = trim($externalUserId);
|
||
if ($externalUserId === '') {
|
||
return;
|
||
}
|
||
|
||
$now = time();
|
||
Db::name('qywx_external_contact')->where('external_userid', $externalUserId)->update([
|
||
'delete_time' => $now,
|
||
'update_time' => $now,
|
||
]);
|
||
|
||
// 客户已删 → 关系表硬删（无 delete_time 列；统计场景不需要保留）
|
||
Db::name('qywx_external_contact_tag')
|
||
->where('external_userid', $externalUserId)
|
||
->delete();
|
||
Db::name('qywx_external_contact_follow')
|
||
->where('external_userid', $externalUserId)
|
||
->delete();
|
||
}
|
||
|
||
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;
|
||
}
|
||
}
|
||
|
||
$row = self::buildContactRow($externalContact, $followUsers);
|
||
|
||
self::upsertExternalContactRow($row, $syncCount, $newCount, $updateCount);
|
||
|
||
// 同步「客户↔员工↔标签」关系表（用于检索/统计；列表筛选/聚合不必再解析 follow_users JSON）
|
||
self::syncContactTagsRelation((string) $row['external_userid'], $followUsers);
|
||
|
||
// 同步「客户↔员工」现存跟进关系表（企微口径的添加时间筛选用）
|
||
self::syncContactFollowRelation((string) $row['external_userid'], $followUsers);
|
||
|
||
if (($syncCount % 25) === 0) {
|
||
usleep(8000);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 同步「客户↔员工↔标签」关系表（zyt_qywx_external_contact_tag）。
|
||
*
|
||
* 行粒度：(external_userid, follow_user_id, tag_id) 三元组——同一标签由不同员工打则各占一行。
|
||
* 用于检索/统计/聚合：按标签筛客户、统计某标签客户数、多标签 AND/OR、按分组漏斗等。
|
||
*
|
||
* 写入策略（保留 create_time，仅在改名/分组变化时刷 tag_name/group_name/type）：
|
||
* 1. 计算目标三元组集合
|
||
* 2. 删除"该 external_userid 下"不在新集合里的旧关系（含跟进人离开 / 标签移除）
|
||
* 3. 对每个新三元组 INSERT ... ON DUPLICATE KEY UPDATE 刷新冗余字段
|
||
*
|
||
* @param array<int, mixed> $followUsers /externalcontact/get 返回的 follow_user[]
|
||
* @internal 仅供 UPSERT / 回填命令复用
|
||
*/
|
||
public static function syncContactTagsRelation(string $externalUserId, array $followUsers): void
|
||
{
|
||
$externalUserId = trim($externalUserId);
|
||
if ($externalUserId === '') {
|
||
return;
|
||
}
|
||
|
||
// 1. 拍平为 (follow_user_id, tag_id) → 行 的字典；保持 follow_user_id 维度区分
|
||
$rows = [];
|
||
foreach ($followUsers as $fu) {
|
||
if (!is_array($fu)) {
|
||
continue;
|
||
}
|
||
$followUserId = trim((string) ($fu['userid'] ?? ''));
|
||
$tags = $fu['tags'] ?? [];
|
||
if (!is_array($tags)) {
|
||
continue;
|
||
}
|
||
foreach ($tags as $t) {
|
||
if (!is_array($t)) {
|
||
continue;
|
||
}
|
||
$tagId = trim((string) ($t['tag_id'] ?? ''));
|
||
if ($tagId === '') {
|
||
continue;
|
||
}
|
||
$key = $followUserId . '|' . $tagId;
|
||
$rows[$key] = [
|
||
'follow_user_id' => mb_substr($followUserId, 0, 64),
|
||
'tag_id' => mb_substr($tagId, 0, 64),
|
||
'tag_name' => mb_substr((string) ($t['tag_name'] ?? ''), 0, 128),
|
||
'group_name' => mb_substr((string) ($t['group_name'] ?? ''), 0, 128),
|
||
'type' => isset($t['type']) ? (int) $t['type'] : 1,
|
||
];
|
||
}
|
||
}
|
||
|
||
$now = time();
|
||
|
||
// 2. 删除该客户下不在目标集合里的旧关系
|
||
if ($rows === []) {
|
||
Db::name('qywx_external_contact_tag')
|
||
->where('external_userid', $externalUserId)
|
||
->delete();
|
||
|
||
return;
|
||
}
|
||
|
||
// 用 (follow_user_id || tag_id) 拼字符串过滤；记录少时 PHP 拉出来比 SQL CONCAT 更高效
|
||
$existing = Db::name('qywx_external_contact_tag')
|
||
->where('external_userid', $externalUserId)
|
||
->field(['id', 'follow_user_id', 'tag_id'])
|
||
->select()
|
||
->toArray();
|
||
$deleteIds = [];
|
||
foreach ($existing as $ex) {
|
||
$key = (string) ($ex['follow_user_id'] ?? '') . '|' . (string) ($ex['tag_id'] ?? '');
|
||
if (!isset($rows[$key])) {
|
||
$deleteIds[] = (int) $ex['id'];
|
||
}
|
||
}
|
||
if ($deleteIds !== []) {
|
||
Db::name('qywx_external_contact_tag')
|
||
->whereIn('id', $deleteIds)
|
||
->delete();
|
||
}
|
||
|
||
// 3. UPSERT 新关系（命中唯一键时只刷冗余字段 + update_time，create_time 保持原值）
|
||
foreach ($rows as $r) {
|
||
try {
|
||
Db::name('qywx_external_contact_tag')
|
||
->duplicate(['tag_name', 'group_name', 'type', 'update_time'])
|
||
->insert([
|
||
'external_userid' => $externalUserId,
|
||
'follow_user_id' => $r['follow_user_id'],
|
||
'tag_id' => $r['tag_id'],
|
||
'tag_name' => $r['tag_name'],
|
||
'group_name' => $r['group_name'],
|
||
'type' => $r['type'],
|
||
'create_time' => $now,
|
||
'update_time' => $now,
|
||
]);
|
||
} catch (\Throwable $e) {
|
||
Log::warning('qywx external contact tag upsert failed: ' . $e->getMessage(), [
|
||
'ext' => $externalUserId,
|
||
'follow_user_id' => $r['follow_user_id'],
|
||
'tag_id' => $r['tag_id'],
|
||
]);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 同步「客户↔员工」现存跟进关系表（zyt_qywx_external_contact_follow）。
|
||
*
|
||
* 与企微后台「全部客户 + 时间筛选」同口径：只保留当前仍存在的跟进关系及其 createtime；
|
||
* 关系被解除（员工删客户且未重加）后行随之删除，客户在对应日期的筛选下即不再出现。
|
||
*
|
||
* @param array<int, mixed> $followUsers /externalcontact/get 返回的 follow_user[]
|
||
* @internal 仅供 UPSERT / 回填复用
|
||
*/
|
||
public static function syncContactFollowRelation(string $externalUserId, array $followUsers): void
|
||
{
|
||
$externalUserId = trim($externalUserId);
|
||
if ($externalUserId === '') {
|
||
return;
|
||
}
|
||
|
||
$rows = [];
|
||
foreach ($followUsers as $fu) {
|
||
if (!is_array($fu)) {
|
||
continue;
|
||
}
|
||
$uid = trim((string) ($fu['userid'] ?? ''));
|
||
if ($uid === '') {
|
||
continue;
|
||
}
|
||
$rows[$uid] = self::normalizeFollowUserCreatetime($fu);
|
||
}
|
||
|
||
try {
|
||
if ($rows === []) {
|
||
Db::name('qywx_external_contact_follow')
|
||
->where('external_userid', $externalUserId)
|
||
->delete();
|
||
|
||
return;
|
||
}
|
||
|
||
Db::name('qywx_external_contact_follow')
|
||
->where('external_userid', $externalUserId)
|
||
->whereNotIn('follow_user_id', array_keys($rows))
|
||
->delete();
|
||
|
||
$now = time();
|
||
foreach ($rows as $uid => $createtime) {
|
||
Db::name('qywx_external_contact_follow')
|
||
->duplicate(['createtime' => $createtime, 'update_time' => $now])
|
||
->insert([
|
||
'external_userid' => $externalUserId,
|
||
'follow_user_id' => $uid,
|
||
'createtime' => $createtime,
|
||
'create_time' => $now,
|
||
'update_time' => $now,
|
||
]);
|
||
}
|
||
} catch (\Throwable $e) {
|
||
Log::warning('qywx contact follow relation sync failed: ' . $e->getMessage(), [
|
||
'ext' => $externalUserId,
|
||
]);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 把所有 follow_user[].tags 合并去重（按 tag_id），返回 JSON 字符串。
|
||
*
|
||
* 一个客户可能同时被多个员工跟进，每个员工各自打的 tag 在各自的 follow_user.tags 里；
|
||
* 这里按 tag_id 唯一去重，保留 tag_name / group_name / type 完整结构，便于后台直接渲染/筛选，
|
||
* 不必再解析 follow_users 大 JSON。
|
||
*
|
||
* 来源结构示例：
|
||
* "tags":[{"group_name":"自媒体1","tag_name":"自媒体1","type":1,"tag_id":"etRr2..."}]
|
||
*
|
||
* @internal 仅供 UPSERT / 回填命令复用，不属于业务对外 API
|
||
* @param array<int, mixed> $followUsers
|
||
*/
|
||
public static function extractFollowUserTags(array $followUsers): string
|
||
{
|
||
$bag = [];
|
||
foreach ($followUsers as $fu) {
|
||
if (!is_array($fu)) {
|
||
continue;
|
||
}
|
||
$tags = $fu['tags'] ?? [];
|
||
if (!is_array($tags)) {
|
||
continue;
|
||
}
|
||
foreach ($tags as $t) {
|
||
if (!is_array($t)) {
|
||
continue;
|
||
}
|
||
$tagId = trim((string) ($t['tag_id'] ?? ''));
|
||
if ($tagId === '') {
|
||
continue;
|
||
}
|
||
if (isset($bag[$tagId])) {
|
||
continue;
|
||
}
|
||
$bag[$tagId] = [
|
||
'group_name' => (string) ($t['group_name'] ?? ''),
|
||
'tag_name' => (string) ($t['tag_name'] ?? ''),
|
||
'type' => isset($t['type']) ? (int) $t['type'] : 1,
|
||
'tag_id' => $tagId,
|
||
];
|
||
}
|
||
}
|
||
|
||
return json_encode(array_values($bag), JSON_UNESCAPED_UNICODE);
|
||
}
|
||
|
||
/**
|
||
* 将企微 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);
|
||
}
|
||
|
||
/**
|
||
* 由企微 API 返回的 external_contact + follow_user 组装一行客户表数据
|
||
*
|
||
* @param array<string, mixed> $externalContact
|
||
* @param array<int, array<string, mixed>> $followUsers
|
||
* @return array<string, mixed>
|
||
*/
|
||
private static function buildContactRow(array $externalContact, array $followUsers): array
|
||
{
|
||
$now = time();
|
||
|
||
return [
|
||
'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),
|
||
'tags' => self::extractFollowUserTags($followUsers),
|
||
'follow_admin_ids' => self::resolveFollowAdminIds($followUsers),
|
||
'external_first_add_time' => self::minFollowCreatetime($followUsers),
|
||
'create_time' => $now,
|
||
'update_time' => $now,
|
||
'delete_time' => null,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 是否 MySQL 锁等待超时 / 死锁（可短重试）
|
||
*/
|
||
private static function isMysqlLockTimeoutOrDeadlock(\Throwable $e): bool
|
||
{
|
||
$m = $e->getMessage();
|
||
|
||
return str_contains($m, '1205') || str_contains($m, '1213') || str_contains($m, 'Deadlock');
|
||
}
|
||
|
||
/**
|
||
* 客户行写入 + 锁冲突重试。
|
||
*
|
||
* 表允许同一 external_userid 多行（重复添加会新增行），不再依赖唯一键 UPSERT：
|
||
* - 已有行 → 只更新"最新一行"（MAX(id)），历史重加行保持原样；
|
||
* - 无行 → 插入新行。
|
||
* 更新时不覆盖 create_time / readd_flag / external_first_add_time
|
||
* （最新行的添加时间代表"该行那次添加"的时间，由插入时决定，后续详情刷新不应改动）。
|
||
*
|
||
* @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 {
|
||
$latestId = (int) Db::name('qywx_external_contact')
|
||
->where('external_userid', (string) $row['external_userid'])
|
||
->order('id', 'desc')
|
||
->value('id');
|
||
|
||
if ($latestId > 0) {
|
||
$update = $row;
|
||
unset($update['external_userid'], $update['create_time'], $update['external_first_add_time']);
|
||
Db::name('qywx_external_contact')->where('id', $latestId)->update($update);
|
||
$updateCount++;
|
||
} else {
|
||
Db::name('qywx_external_contact')->insert($row);
|
||
$newCount++;
|
||
}
|
||
$syncCount++;
|
||
|
||
return;
|
||
} catch (\Throwable $e) {
|
||
if ($attempt >= $maxAttempts || !self::isMysqlLockTimeoutOrDeadlock($e)) {
|
||
throw $e;
|
||
}
|
||
usleep(min(800_000, 40_000 * (2 ** ($attempt - 1))));
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 标签维度统计：用于「按标签筛客户」下拉选项 + 「按标签看分布」面板
|
||
*
|
||
* 数据源：zyt_qywx_external_contact_tag（关系表）+ zyt_qywx_external_contact（剔除已软删客户）
|
||
* 输出按 group_name 分组、组内按客户数倒序，每条带 tag_id / tag_name / customer_count
|
||
*
|
||
* @return array{
|
||
* total_tags: int,
|
||
* total_relations: int,
|
||
* total_tagged_customers: int,
|
||
* groups: array<int, array{group_name: string, customer_count: int, tags: array<int, array{tag_id: string, tag_name: string, customer_count: int}>}>
|
||
* }
|
||
*/
|
||
public static function getTagStats(): array
|
||
{
|
||
// 关系表里 external_userid 可能指向已被软删的客户；这里 INNER JOIN 主表过滤未删除的
|
||
$rows = Db::name('qywx_external_contact_tag')
|
||
->alias('ect')
|
||
->join('qywx_external_contact ec', 'ec.external_userid = ect.external_userid', 'INNER')
|
||
->whereNull('ec.delete_time')
|
||
->field([
|
||
'ect.tag_id',
|
||
'ect.tag_name',
|
||
'ect.group_name',
|
||
'COUNT(DISTINCT ect.external_userid) AS customer_count',
|
||
])
|
||
->group('ect.tag_id, ect.tag_name, ect.group_name')
|
||
->order('customer_count', 'desc')
|
||
->select()
|
||
->toArray();
|
||
|
||
$groupMap = [];
|
||
$customersUnion = [];
|
||
foreach ($rows as $r) {
|
||
$g = (string) ($r['group_name'] ?? '');
|
||
if (!isset($groupMap[$g])) {
|
||
$groupMap[$g] = [
|
||
'group_name' => $g,
|
||
'customer_count' => 0,
|
||
'tags' => [],
|
||
];
|
||
}
|
||
$cnt = (int) ($r['customer_count'] ?? 0);
|
||
$groupMap[$g]['tags'][] = [
|
||
'tag_id' => (string) ($r['tag_id'] ?? ''),
|
||
'tag_name' => (string) ($r['tag_name'] ?? ''),
|
||
'customer_count' => $cnt,
|
||
];
|
||
}
|
||
|
||
// 组内已经按 customer_count desc 排好序了（外面 SQL 排序结果）；
|
||
// 组本身按"组内最大 customer_count"倒序排（最热门标签所在组优先）
|
||
foreach ($groupMap as &$g) {
|
||
$g['customer_count'] = $g['tags'][0]['customer_count'] ?? 0;
|
||
}
|
||
unset($g);
|
||
|
||
$groups = array_values($groupMap);
|
||
usort($groups, static function ($a, $b) {
|
||
return $b['customer_count'] <=> $a['customer_count'];
|
||
});
|
||
|
||
// 全表 distinct 客户数（带 tag 的客户）
|
||
$taggedCustomers = (int) Db::name('qywx_external_contact_tag')
|
||
->alias('ect')
|
||
->join('qywx_external_contact ec', 'ec.external_userid = ect.external_userid', 'INNER')
|
||
->whereNull('ec.delete_time')
|
||
->group('ect.external_userid')
|
||
->count();
|
||
|
||
$totalRelations = (int) Db::name('qywx_external_contact_tag')
|
||
->alias('ect')
|
||
->join('qywx_external_contact ec', 'ec.external_userid = ect.external_userid', 'INNER')
|
||
->whereNull('ec.delete_time')
|
||
->count();
|
||
|
||
return [
|
||
'total_tags' => count($rows),
|
||
'total_relations' => $totalRelations,
|
||
'total_tagged_customers' => $taggedCustomers,
|
||
'groups' => $groups,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* @notes 获取统计信息
|
||
*/
|
||
public static function getStats()
|
||
{
|
||
// 表内同一客户可能多行（重复添加各占一行），去重后才是真实客户数
|
||
$total = (int) Db::name('qywx_external_contact')
|
||
->whereNull('delete_time')
|
||
->fieldRaw('COUNT(DISTINCT external_userid) AS c')
|
||
->find()['c'];
|
||
|
||
// 企微管理后台「全部客户」按"客户×添加人"关系计数（同一客户被 N 名员工添加计 N 条），
|
||
// 现存跟进关系表与其同口径，便于对账。
|
||
$relationTotal = (int) Db::name('qywx_external_contact_follow')->count();
|
||
|
||
$todayStart = strtotime(date('Y-m-d 00:00:00'));
|
||
|
||
// 今日进入数 = 今日收到的 add_external_contact 事件条数。
|
||
// - 事件表 INSERT IGNORE 幂等，企微自动重试不会重复计数；
|
||
// - 与业务表 UPSERT/软删解耦，客户即使很快被删、或 /externalcontact/get 同步失败，都不影响计数；
|
||
// - 取 event_time（企微 CreateTime）而不是 create_time，跨日分钟内推送过来也归到事件真实日期。
|
||
$today = (int) Db::name('qywx_external_contact_event')
|
||
->where('change_type', 'add_external_contact')
|
||
->where('event_time', '>=', $todayStart)
|
||
->count();
|
||
|
||
/** 今日新增客户中，跟进人条数合计（同一 userid 多客户或多条均累计，不去重） */
|
||
$todayFollowStaff = 0;
|
||
$followJsonList = QywxExternalContact::whereNull('delete_time')
|
||
->whereRaw(
|
||
'(external_first_add_time >= ? OR (external_first_add_time = 0 AND create_time >= ?))',
|
||
[$todayStart, $todayStart]
|
||
)
|
||
->column('follow_users');
|
||
foreach ($followJsonList as $json) {
|
||
$raw = json_decode((string) $json, true);
|
||
if (!is_array($raw)) {
|
||
continue;
|
||
}
|
||
foreach ($raw as $fu) {
|
||
if (!is_array($fu)) {
|
||
continue;
|
||
}
|
||
$wx = trim((string) ($fu['userid'] ?? ''));
|
||
if ($wx !== '') {
|
||
$todayFollowStaff++;
|
||
}
|
||
}
|
||
}
|
||
|
||
$settings = self::getSyncSettings();
|
||
$lastSyncTime = $settings['last_sync_time'] ?? 0;
|
||
$lastSync = $lastSyncTime > 0 ? date('Y-m-d H:i:s', $lastSyncTime) : '';
|
||
|
||
return [
|
||
'total' => $total,
|
||
'relation_total' => $relationTotal,
|
||
'today' => $today,
|
||
'today_follow_staff' => $todayFollowStaff,
|
||
'lastSync' => $lastSync,
|
||
'syncStatus' => $settings['sync_status'] ?? 'idle',
|
||
'syncStatusText' => self::getSyncStatusText($settings['sync_status'] ?? 'idle'),
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 今日"进入"分布概览（直观卡片用）
|
||
*
|
||
* 数据源：qywx_external_contact_event（add_external_contact）——回调一到就幂等写入，
|
||
* 与 qywx_external_contact 的软删/UPSERT 解耦，保证"进入数"零误差。
|
||
*
|
||
* 返回：
|
||
* - total int 今日进入总数（= stats.today）
|
||
* - recent_time int 今日最近一条 add 事件时间戳（0 表示今天还没进人）
|
||
* - hourly int[24] 按 0-23 点分布（本地时区）
|
||
* - by_state array 按 state 分组 Top 5：[{state, count}]
|
||
*
|
||
* @return array<string, mixed>
|
||
*/
|
||
public static function getTodayArrivalStats(): array
|
||
{
|
||
$todayStart = strtotime(date('Y-m-d 00:00:00'));
|
||
$todayEnd = $todayStart + 86400;
|
||
|
||
$total = (int) Db::name('qywx_external_contact_event')
|
||
->where('change_type', 'add_external_contact')
|
||
->where('event_time', '>=', $todayStart)
|
||
->where('event_time', '<', $todayEnd)
|
||
->count();
|
||
|
||
$recent = (int) Db::name('qywx_external_contact_event')
|
||
->where('change_type', 'add_external_contact')
|
||
->where('event_time', '>=', $todayStart)
|
||
->where('event_time', '<', $todayEnd)
|
||
->max('event_time');
|
||
|
||
$hourly = array_fill(0, 24, 0);
|
||
if ($total > 0) {
|
||
// 用 FROM_UNIXTIME 按本地时区分桶；php.ini / mysql time_zone 有差异时，可改 DATE_FORMAT
|
||
$rows = Db::name('qywx_external_contact_event')
|
||
->fieldRaw('HOUR(FROM_UNIXTIME(event_time)) AS h, COUNT(*) AS c')
|
||
->where('change_type', 'add_external_contact')
|
||
->where('event_time', '>=', $todayStart)
|
||
->where('event_time', '<', $todayEnd)
|
||
->group('h')
|
||
->select()
|
||
->toArray();
|
||
foreach ($rows as $r) {
|
||
$h = (int) ($r['h'] ?? 0);
|
||
if ($h >= 0 && $h <= 23) {
|
||
$hourly[$h] = (int) ($r['c'] ?? 0);
|
||
}
|
||
}
|
||
}
|
||
|
||
$byState = [];
|
||
if ($total > 0) {
|
||
$stateRows = Db::name('qywx_external_contact_event')
|
||
->field(['state', 'COUNT(*) AS c'])
|
||
->where('change_type', 'add_external_contact')
|
||
->where('event_time', '>=', $todayStart)
|
||
->where('event_time', '<', $todayEnd)
|
||
->group('state')
|
||
->orderRaw('c DESC')
|
||
->limit(5)
|
||
->select()
|
||
->toArray();
|
||
foreach ($stateRows as $r) {
|
||
$byState[] = [
|
||
'state' => (string) ($r['state'] ?? ''),
|
||
'count' => (int) ($r['c'] ?? 0),
|
||
];
|
||
}
|
||
}
|
||
|
||
// 老客户回流数（去重客户维度）：今天有 add 事件，但今天之前就加过——
|
||
// 依据①事件流水里有更早的 add 事件，或②客户表首次添加时间早于今天（覆盖流水表上线前的老客户）
|
||
$returning = 0;
|
||
if ($total > 0) {
|
||
$todayExtIds = Db::name('qywx_external_contact_event')
|
||
->where('change_type', 'add_external_contact')
|
||
->where('event_time', '>=', $todayStart)
|
||
->where('event_time', '<', $todayEnd)
|
||
->group('external_userid')
|
||
->column('external_userid');
|
||
$todayExtIds = array_values(array_filter($todayExtIds));
|
||
if ($todayExtIds !== []) {
|
||
$oldSet = self::resolveOldCustomerSet($todayExtIds, $todayStart);
|
||
$returning = count($oldSet);
|
||
}
|
||
}
|
||
|
||
return [
|
||
'total' => $total,
|
||
'recent_time' => $recent,
|
||
'returning' => $returning,
|
||
'hourly' => array_values($hourly),
|
||
'by_state' => $byState,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 判定"老客户"：在 $beforeTs 之前就已加过企业的客户集合。
|
||
*
|
||
* 依据（满足其一即算老客户）：
|
||
* ① 事件流水表在 $beforeTs 之前存在该客户的 add_external_contact 记录；
|
||
* ② 客户表 external_first_add_time（企微 follow_user 最早 createtime）早于 $beforeTs——
|
||
* 覆盖事件流水表上线之前就已存在的客户。
|
||
*
|
||
* @param string[] $extIds
|
||
* @return array<string, true> external_userid 为键的集合
|
||
*/
|
||
private static function resolveOldCustomerSet(array $extIds, int $beforeTs): array
|
||
{
|
||
$extIds = array_values(array_unique(array_filter($extIds)));
|
||
if ($extIds === []) {
|
||
return [];
|
||
}
|
||
|
||
$old = [];
|
||
|
||
$earlierEventIds = Db::name('qywx_external_contact_event')
|
||
->whereIn('external_userid', $extIds)
|
||
->where('change_type', 'add_external_contact')
|
||
->where('event_time', '<', $beforeTs)
|
||
->group('external_userid')
|
||
->column('external_userid');
|
||
foreach ($earlierEventIds as $id) {
|
||
$old[(string) $id] = true;
|
||
}
|
||
|
||
$earlierFirstAddIds = Db::name('qywx_external_contact')
|
||
->whereIn('external_userid', $extIds)
|
||
->where('external_first_add_time', '>', 0)
|
||
->where('external_first_add_time', '<', $beforeTs)
|
||
->column('external_userid');
|
||
foreach ($earlierFirstAddIds as $id) {
|
||
$old[(string) $id] = true;
|
||
}
|
||
|
||
return $old;
|
||
}
|
||
|
||
/**
|
||
* 今日"进入明细"流水（分页，按时间倒序）
|
||
*
|
||
* 左联客户表补客户名/头像，左联 admin 补员工名；客户已删 / 同步未回不影响流水本身。
|
||
*
|
||
* @return array{total:int, page_no:int, page_size:int, lists:array<int,array<string,mixed>>}
|
||
*/
|
||
public static function getTodayArrivalList(int $pageNo, int $pageSize): array
|
||
{
|
||
$pageNo = max(1, $pageNo);
|
||
$pageSize = min(100, max(1, $pageSize));
|
||
$todayStart = strtotime(date('Y-m-d 00:00:00'));
|
||
$todayEnd = $todayStart + 86400;
|
||
|
||
$base = Db::name('qywx_external_contact_event')
|
||
->where('change_type', 'add_external_contact')
|
||
->where('event_time', '>=', $todayStart)
|
||
->where('event_time', '<', $todayEnd);
|
||
|
||
$total = (int) (clone $base)->count();
|
||
|
||
$rows = (clone $base)
|
||
->orderRaw('event_time DESC, id DESC')
|
||
->page($pageNo, $pageSize)
|
||
->select()
|
||
->toArray();
|
||
|
||
if ($rows === []) {
|
||
return ['total' => $total, 'page_no' => $pageNo, 'page_size' => $pageSize, 'lists' => []];
|
||
}
|
||
|
||
$extIds = array_values(array_unique(array_filter(array_map(static fn($r) => (string) ($r['external_userid'] ?? ''), $rows))));
|
||
$userIds = array_values(array_unique(array_filter(array_map(static fn($r) => (string) ($r['user_id'] ?? ''), $rows))));
|
||
|
||
$customerMap = [];
|
||
if ($extIds !== []) {
|
||
$customerRows = Db::name('qywx_external_contact')
|
||
->field(['external_userid', 'name', 'avatar'])
|
||
->whereIn('external_userid', $extIds)
|
||
->select()
|
||
->toArray();
|
||
foreach ($customerRows as $c) {
|
||
$customerMap[(string) $c['external_userid']] = $c;
|
||
}
|
||
}
|
||
|
||
$adminMap = [];
|
||
if ($userIds !== []) {
|
||
$adminMap = Admin::whereIn('work_wechat_userid', $userIds)->column('name', 'work_wechat_userid');
|
||
}
|
||
|
||
// 标记老客户：今天之前就加过企业（重加 / 被另一名员工添加的回流客户）
|
||
$oldSet = self::resolveOldCustomerSet($extIds, $todayStart);
|
||
|
||
$lists = [];
|
||
foreach ($rows as $r) {
|
||
$ext = (string) ($r['external_userid'] ?? '');
|
||
$uid = (string) ($r['user_id'] ?? '');
|
||
$lists[] = [
|
||
'id' => (int) $r['id'],
|
||
'event_time' => (int) $r['event_time'],
|
||
'user_id' => $uid,
|
||
'admin_name' => (string) ($adminMap[$uid] ?? ''),
|
||
'external_userid' => $ext,
|
||
'customer_name' => (string) ($customerMap[$ext]['name'] ?? ''),
|
||
'customer_avatar' => (string) ($customerMap[$ext]['avatar'] ?? ''),
|
||
'state' => (string) ($r['state'] ?? ''),
|
||
'welcome_code' => (int) ($r['welcome_code'] ?? 0),
|
||
'is_old_customer' => isset($oldSet[$ext]) ? 1 : 0,
|
||
];
|
||
}
|
||
|
||
return [
|
||
'total' => $total,
|
||
'page_no' => $pageNo,
|
||
'page_size' => $pageSize,
|
||
'lists' => $lists,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* @notes 获取同步设置
|
||
*/
|
||
public static function getSyncSettings()
|
||
{
|
||
$setting = QywxSyncSettings::where('setting_key', 'customer_sync')->find();
|
||
if (!$setting) {
|
||
return [
|
||
'auto_sync' => false,
|
||
'interval' => 3600,
|
||
'last_sync_time' => 0,
|
||
'sync_status' => 'idle',
|
||
];
|
||
}
|
||
|
||
$value = json_decode($setting->setting_value, true);
|
||
return $value ?: [
|
||
'auto_sync' => false,
|
||
'interval' => 3600,
|
||
'last_sync_time' => 0,
|
||
'sync_status' => 'idle',
|
||
];
|
||
}
|
||
|
||
/**
|
||
* @notes 保存同步设置
|
||
*/
|
||
public static function saveSyncSettings(array $params)
|
||
{
|
||
try {
|
||
$currentSettings = self::getSyncSettings();
|
||
$newSettings = array_merge($currentSettings, [
|
||
'auto_sync' => (bool)($params['auto_sync'] ?? false),
|
||
'interval' => (int)($params['interval'] ?? 3600),
|
||
]);
|
||
|
||
$setting = QywxSyncSettings::where('setting_key', 'customer_sync')->find();
|
||
if ($setting) {
|
||
$setting->setting_value = json_encode($newSettings, JSON_UNESCAPED_UNICODE);
|
||
$setting->update_time = time();
|
||
$setting->save();
|
||
} else {
|
||
QywxSyncSettings::create([
|
||
'setting_key' => 'customer_sync',
|
||
'setting_value' => json_encode($newSettings, JSON_UNESCAPED_UNICODE),
|
||
'create_time' => time(),
|
||
'update_time' => time(),
|
||
]);
|
||
}
|
||
|
||
return true;
|
||
} catch (\Throwable $e) {
|
||
self::$error = '保存失败: ' . $e->getMessage();
|
||
return false;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @param bool $bumpLastSyncTime 为 false 时仅改状态（用于 syncing），避免「最后同步」在跑任务期间被刷新
|
||
*/
|
||
private static function updateSyncStatus(string $status, int $syncCount = 0, string $error = '', bool $bumpLastSyncTime = true)
|
||
{
|
||
try {
|
||
$settings = self::getSyncSettings();
|
||
$settings['sync_status'] = $status;
|
||
if ($bumpLastSyncTime) {
|
||
$settings['last_sync_time'] = time();
|
||
$settings['last_sync_count'] = $syncCount;
|
||
}
|
||
if ($error) {
|
||
$settings['last_error'] = $error;
|
||
}
|
||
|
||
$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());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @notes 获取同步状态文本
|
||
*/
|
||
private static function getSyncStatusText(string $status): string
|
||
{
|
||
$map = [
|
||
'idle' => '未同步',
|
||
'syncing' => '同步中',
|
||
'success' => '同步成功',
|
||
'failed' => '同步失败',
|
||
];
|
||
return $map[$status] ?? '未知';
|
||
}
|
||
}
|