更新

2026-04-20 09:04:54 +08:00
parent 4097fe0cc6
commit 34ba9222fa
245 changed files with 537 additions and 242 deletions
@@ -192,10 +192,45 @@ class CustomerLogic extends BaseLogic
            try {
                $useBatch = (bool) config('project.qywx_sync_use_batch_detail', true);
                $mergedMap = null;
+                /** 已走「时间筛选 + 流式 batch」分支且未抛错时，不再 list+get 全量回退 */
+                $createtimeFilterBatchPathDone = false;
                if ($useBatch) {
                    try {
-                        $mergedMap = self::buildMergedExternalContactsMap($service, $userList);
-                        Log::info('qywx sync: 批量接口合并后客户数 ' . count($mergedMap));
+                        // 按跟进时间筛选：不构建全量合并表（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;
@@ -217,7 +252,7 @@ class CustomerLogic extends BaseLogic
                            $skippedCount
                        );
                    }
-                } else {
+                } elseif (!$createtimeFilterBatchPathDone) {
                    $processedCustomers = [];
                    foreach ($userList as $user) {
                        $userId = $user['userid'] ?? '';
@@ -370,6 +405,82 @@ class CustomerLogic extends BaseLogic
        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 按成员分页拉取并合并跟进人（列表接口无按时间筛选）
     *
@@ -379,8 +490,10 @@ class CustomerLogic extends BaseLogic
    private static function buildMergedExternalContactsMap(WechatWorkService $service, array $userList): array
    {
        $merged = [];
+      
        foreach ($userList as $user) {
            $userId = trim((string) ($user['userid'] ?? ''));
+           
            if ($userId === '') {
                continue;
            }
@@ -388,6 +501,7 @@ class CustomerLogic extends BaseLogic
            $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'] ?? ''));
@@ -468,6 +582,66 @@ class CustomerLogic extends BaseLogic
     * @param array<string, mixed> $externalContact
     * @param array<int, array<string, mixed>> $followUsers
     */
+    /**
+     * 客户联系事件回调：按 external_userid 拉取详情并 UPSERT（不走全量同步文件锁）。
+     *
+     * @see https://developer.work.weixin.qq.com/document/path/92130
+     */
+    public static function upsertSingleExternalContactFromApi(string $externalUserId): void
+    {
+        $externalUserId = trim($externalUserId);
+        if ($externalUserId === '') {
+            return;
+        }
+
+        $service = new WechatWorkService();
+        $detail = $service->getExternalContactDetail($externalUserId);
+        if (empty($detail['external_contact']) || !is_array($detail['external_contact'])) {
+            Log::warning('qywx external contact callback: 拉取详情为空', ['external_userid' => $externalUserId]);
+
+            return;
+        }
+
+        $followUsers = $detail['follow_user'] ?? [];
+        if (!is_array($followUsers)) {
+            $followUsers = [];
+        }
+
+        $syncCount = 0;
+        $newCount = 0;
+        $updateCount = 0;
+        $skippedCount = 0;
+        self::upsertOneExternalContactBundle(
+            $detail['external_contact'],
+            $followUsers,
+            false,
+            'first',
+            0,
+            0,
+            $syncCount,
+            $newCount,
+            $updateCount,
+            $skippedCount
+        );
+    }
+
+    /**
+     * 客户联系「删除企业客户」等事件：本地软删除一行。
+     */
+    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,
+        ]);
+    }
+
    private static function upsertOneExternalContactBundle(
        array $externalContact,
        array $followUsers,
@@ -0,0 +1,105 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\api\controller;
+
+use app\adminapi\logic\qywx\CustomerLogic;
+use EasyWeChat\Kernel\Exceptions\BadRequestException;
+use EasyWeChat\Work\Application;
+use EasyWeChat\Work\Message;
+use think\facade\Log;
+
+/**
+ * 企业微信「客户联系」事件回调（接收事件服务器）
+ *
+ * @see https://developer.work.weixin.qq.com/document/path/92130
+ */
+class QywxExternalContactCallbackController extends BaseApiController
+{
+    /** 免登录：企微服务器回调 */
+    public array $notNeedLogin = ['notify'];
+
+    public function notify()
+    {
+        $corpId = (string) config('pay.wechat_work.corp_id', '');
+        $secret = (string) config('pay.wechat_work.external_pay_secret', '');
+        $token = (string) config('pay.wechat_work.contact_callback_token', '');
+        $aesKey = (string) config('pay.wechat_work.contact_callback_aes_key', '');
+
+        if ($corpId === '' || $secret === '' || $token === '' || $aesKey === '') {
+            Log::error('qywx external contact callback: 缺少配置 corp_id / external_pay_secret / contact_callback_token / contact_callback_aes_key');
+
+            return response('config error', 503, ['Content-Type' => 'text/plain; charset=utf-8']);
+        }
+
+        try {
+            $app = new Application([
+                'corp_id' => $corpId,
+                'secret' => $secret,
+                'token' => $token,
+                'aes_key' => $aesKey,
+            ]);
+            $server = $app->getServer();
+
+            $server->addEventListener('change_external_contact', function (Message $message, \Closure $next) {
+                try {
+                    $this->handleChangeExternalContact($message);
+                } catch (\Throwable $e) {
+                    Log::error('qywx external contact callback: ' . $e->getMessage(), [
+                        'exception' => $e,
+                    ]);
+                }
+
+                return $next($message);
+            });
+
+            $psr = $server->serve();
+            $body = $psr->getBody();
+            $body->rewind();
+            $content = $body->getContents();
+
+            $headers = [];
+            $contentType = $psr->getHeaderLine('Content-Type');
+            if ($contentType !== '') {
+                $headers['Content-Type'] = $contentType;
+            }
+
+            return response($content, 200, $headers);
+        } catch (BadRequestException $e) {
+            Log::warning('qywx external contact callback: bad request ' . $e->getMessage());
+
+            return response('bad request', 400, ['Content-Type' => 'text/plain; charset=utf-8']);
+        } catch (\Throwable $e) {
+            Log::error('qywx external contact callback: serve failed ' . $e->getMessage(), ['exception' => $e]);
+
+            return response('error', 500, ['Content-Type' => 'text/plain; charset=utf-8']);
+        }
+    }
+
+    /**
+     * @param Message $message
+     */
+    private function handleChangeExternalContact($message): void
+    {
+        $changeType = (string) ($message['ChangeType'] ?? '');
+        $extId = trim((string) ($message['ExternalUserID'] ?? $message['ExternalUserId'] ?? ''));
+
+        if ($extId === '') {
+            Log::info('qywx external contact callback: 无 ExternalUserID', ['ChangeType' => $changeType]);
+
+            return;
+        }
+
+        Log::info('qywx external contact callback', ['ChangeType' => $changeType, 'ExternalUserID' => $extId]);
+
+        if ($changeType === 'del_external_contact') {
+            CustomerLogic::softDeleteExternalContactRow($extId);
+
+            return;
+        }
+
+        // 其余变更（添加/编辑/删跟进人等）：以 get 详情为准 UPSERT，避免遗漏未枚举的 ChangeType
+        CustomerLogic::upsertSingleExternalContactFromApi($extId);
+    }
+}
@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+use think\facade\Route;
+
+// 多应用下「api」应用的路由文件；URL 形如 /api/qywx/...，此处规则不含应用前缀 api。
+// @see https://doc.thinkphp.cn/v8_0/multi_app_model.html
+
+// 企业微信「客户联系」事件回调：GET 验签(echostr)、POST 收事件
+Route::rule('qywx/external-contact/notify', 'QywxExternalContactCallbackController@notify', 'GET|POST');
@@ -18,7 +18,7 @@ use think\facade\Log;
 * php think qywx:sync-customer --force --today   # 仅落库「今日首次加为外部联系人」的客户（app.default_timezone）
 * php think qywx:sync-customer --force --today --today-any-follow  # 任一条跟进在今天即落库（旧逻辑，命中多）
 *
- * 说明：企微「客户列表」API 不支持按时间筛选；--today 只减少写库条数。拉取侧默认用 batch/get_by_user 批量详情（project.qywx_sync_use_batch_detail）加速。
+ * 说明：企微无「只拉今日客户」接口，--today 仍需翻完各成员 batch 分页；开启后会流式筛候选 id，只对候选 get+写库（省内存与 UPSERT）。增量实时请配客户联系回调。
 *
 * 配置crontab（每小时执行一次）：
 * 0 * * * * cd /path/to/project && php think qywx:sync-customer >> /dev/null 2>&1