diff --git a/server/app/command/QywxSyncPromotionRanges.php b/server/app/command/QywxSyncPromotionRanges.php new file mode 100644 index 000000000..f2a2739f5 --- /dev/null +++ b/server/app/command/QywxSyncPromotionRanges.php @@ -0,0 +1,33 @@ +setName('qywx:sync-promotion-ranges') + ->setDescription('同步企业微信官方获客链接的下一名承接成员'); + } + + protected function execute(Input $input, Output $output): int + { + $result = (new QywxPromotionRangeSyncService())->syncPending(100); + $output->writeln(sprintf( + 'QYWX_PROMOTION_RANGE_SYNC selected=%d synced=%d failed=%d', + $result['selected'], + $result['synced'], + $result['failed'] + )); + + return 0; + } +} diff --git a/server/app/common/service/qywx/QywxPromotionMemberSchedulerService.php b/server/app/common/service/qywx/QywxPromotionMemberSchedulerService.php new file mode 100644 index 000000000..0cb116730 --- /dev/null +++ b/server/app/common/service/qywx/QywxPromotionMemberSchedulerService.php @@ -0,0 +1,335 @@ + 'ignored_state', 'pool_id' => 0, 'member_id' => 0, 'next_member_id' => 0]; + } + + return self::record($poolId, $userId, $externalUserId, $eventTime, $source); + } + + /** @return array{status:string,pool_id:int,member_id:int,next_member_id:int} */ + public static function recordFromRemoteLink( + string $remoteLinkId, + string $userId, + string $externalUserId, + int $eventTime = 0, + string $source = 'customer_acquisition' + ): array { + $poolId = (int) (Db::name('qywx_promotion_link') + ->where('remote_link_id', trim($remoteLinkId)) + ->whereNull('delete_time') + ->value('pool_id') ?? 0); + if ($poolId <= 0) { + return ['status' => 'ignored_link', 'pool_id' => 0, 'member_id' => 0, 'next_member_id' => 0]; + } + + return self::record($poolId, $userId, $externalUserId, $eventTime, $source); + } + + /** + * 管理端变更成员开关、权重或上限后,确保当前远端成员仍可用;必要时立即排队切换。 + * + * @return array{pool_id:int,next_member_id:int,queued:bool,blocked:bool} + */ + public static function reconcilePool(int $poolId): array + { + return Db::transaction(function () use ($poolId): array { + $now = time(); + $today = date('Y-m-d', $now); + $linkId = self::promotionLinkId($poolId); + $sync = self::lockedSyncRow($poolId); + $members = self::lockedMembers($poolId); + if ($linkId <= 0 || $members === []) { + return ['pool_id' => $poolId, 'next_member_id' => 0, 'queued' => false, 'blocked' => true]; + } + + $desiredId = (int) ($sync['desired_member_id'] ?? 0); + $desiredEligible = false; + foreach ($members as &$member) { + if ((string) ($member['today_date'] ?? '') !== $today) { + $member['today_date'] = $today; + $member['today_count'] = 0; + } + if ((int) $member['id'] === $desiredId) { + $desiredEligible = QywxPromotionWeightedRandom::eligible($member, $now); + } + } + unset($member); + self::persistMemberCursors($members, $now); + if ($desiredEligible) { + if ((int) ($sync['status'] ?? 0) === 4) { + Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->update([ + 'status' => 0, + 'next_retry' => 0, + 'last_error' => '', + 'update_time' => $now, + ]); + } + return ['pool_id' => $poolId, 'next_member_id' => $desiredId, 'queued' => false, 'blocked' => false]; + } + + return self::selectAndQueueLocked($poolId, $linkId, $sync, $members, $today, $now); + }); + } + + public static function initialisePool(int $poolId, int $promotionLinkId, string $selectedUserId): int + { + $memberId = (int) (Db::name('qywx_promotion_pool_member') + ->where('pool_id', $poolId) + ->where('userid', $selectedUserId) + ->whereNull('delete_time') + ->value('id') ?? 0); + if ($memberId <= 0 || $promotionLinkId <= 0) { + return 0; + } + $now = time(); + $existing = Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->find(); + $data = [ + 'promotion_link_id' => $promotionLinkId, + 'desired_member_id' => $memberId, + 'applied_member_id' => $memberId, + 'status' => 0, + 'attempts' => 0, + 'next_retry' => 0, + 'lock_token' => '', + 'lock_until' => 0, + 'last_error' => '', + 'update_time' => $now, + ]; + if ($existing) { + $version = max(1, (int) ($existing['desired_version'] ?? 0) + 1); + $data['desired_version'] = $version; + $data['applied_version'] = $version; + Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->update($data); + } else { + $data += ['pool_id' => $poolId, 'desired_version' => 1, 'applied_version' => 1, 'create_time' => $now]; + Db::name('qywx_promotion_range_sync')->insert($data); + } + + return $memberId; + } + + /** @return array{status:string,pool_id:int,member_id:int,next_member_id:int} */ + private static function record( + int $poolId, + string $userId, + string $externalUserId, + int $eventTime, + string $source + ): array { + $userId = trim($userId); + $externalUserId = trim($externalUserId); + if ($poolId <= 0 || $userId === '' || $externalUserId === '') { + return ['status' => 'ignored_identity', 'pool_id' => $poolId, 'member_id' => 0, 'next_member_id' => 0]; + } + + return Db::transaction(function () use ($poolId, $userId, $externalUserId, $eventTime, $source): array { + $pool = Db::name('qywx_promotion_pool')->where('id', $poolId)->whereNull('delete_time')->lock(true)->find(); + if (!$pool) { + return ['status' => 'ignored_pool', 'pool_id' => $poolId, 'member_id' => 0, 'next_member_id' => 0]; + } + $members = self::lockedMembers($poolId); + $actualMember = null; + foreach ($members as $member) { + if ((string) ($member['userid'] ?? '') === $userId) { + $actualMember = $member; + break; + } + } + if ($actualMember === null) { + return ['status' => 'ignored_member', 'pool_id' => $poolId, 'member_id' => 0, 'next_member_id' => 0]; + } + $actualMemberId = (int) $actualMember['id']; + $eventKey = hash('sha256', $poolId . '|' . $userId . '|' . $externalUserId); + try { + Db::name('qywx_promotion_dispatch_event')->insert([ + 'event_key' => $eventKey, + 'pool_id' => $poolId, + 'member_id' => $actualMemberId, + 'userid' => $userId, + 'external_userid' => $externalUserId, + 'source' => mb_substr($source, 0, 32), + 'event_time' => max(0, $eventTime), + 'create_time' => time(), + ]); + } catch (\Throwable $e) { + if (!Db::name('qywx_promotion_dispatch_event')->where('event_key', $eventKey)->find()) { + throw $e; + } + + return ['status' => 'duplicate', 'pool_id' => $poolId, 'member_id' => $actualMemberId, 'next_member_id' => 0]; + } + + $now = time(); + $today = date('Y-m-d', $now); + foreach ($members as &$member) { + if ((string) ($member['today_date'] ?? '') !== $today) { + $member['today_date'] = $today; + $member['today_count'] = 0; + } + if ((int) $member['id'] === $actualMemberId) { + $member['today_count'] = (int) ($member['today_count'] ?? 0) + 1; + $member['total_count'] = (int) ($member['total_count'] ?? 0) + 1; + $member['last_assigned_time'] = max($now, max(0, $eventTime)); + } + } + unset($member); + self::persistMemberCursors($members, $now); + + $linkId = self::promotionLinkId($poolId); + $sync = self::lockedSyncRow($poolId); + $desiredId = (int) ($sync['desired_member_id'] ?? 0); + if ($linkId <= 0 || ($desiredId > 0 && $desiredId !== $actualMemberId)) { + return ['status' => 'counted_stale', 'pool_id' => $poolId, 'member_id' => $actualMemberId, 'next_member_id' => $desiredId]; + } + + $planned = self::selectAndQueueLocked($poolId, $linkId, $sync, $members, $today, $now); + + return [ + 'status' => $planned['blocked'] ? 'counted_blocked' : 'counted', + 'pool_id' => $poolId, + 'member_id' => $actualMemberId, + 'next_member_id' => $planned['next_member_id'], + ]; + }); + } + + /** @return array{pool_id:int,next_member_id:int,queued:bool,blocked:bool} */ + private static function selectAndQueueLocked( + int $poolId, + int $linkId, + ?array $sync, + array $members, + string $today, + int $now + ): array { + $selection = QywxPromotionWeightedRandom::select($members, $today, $now); + self::persistMemberCursors($selection['members'], $now); + $selectedId = (int) $selection['selected_id']; + if ($selectedId <= 0) { + self::upsertSync($poolId, $linkId, (int) ($sync['desired_member_id'] ?? 0), false, $sync, $now, '所有成员均已禁用、未生效或达到今日上限'); + + return ['pool_id' => $poolId, 'next_member_id' => 0, 'queued' => false, 'blocked' => true]; + } + $changed = $selectedId !== (int) ($sync['desired_member_id'] ?? 0); + self::upsertSync($poolId, $linkId, $selectedId, $changed, $sync, $now); + + return ['pool_id' => $poolId, 'next_member_id' => $selectedId, 'queued' => $changed, 'blocked' => false]; + } + + private static function upsertSync( + int $poolId, + int $linkId, + int $desiredMemberId, + bool $pending, + ?array $existing, + int $now, + string $error = '' + ): void { + $version = max(1, (int) ($existing['desired_version'] ?? 0) + ($pending ? 1 : 0)); + $data = [ + 'promotion_link_id' => $linkId, + 'desired_member_id' => $desiredMemberId, + 'desired_version' => $version, + 'status' => $error !== '' ? 4 : ($pending ? 1 : (int) ($existing['status'] ?? 0)), + 'next_retry' => $error !== '' ? strtotime('tomorrow', $now) : ($pending ? $now : 0), + 'last_error' => mb_substr($error, 0, 500), + 'update_time' => $now, + ]; + if ($pending) { + $data['attempts'] = 0; + } + if ($existing) { + Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->update($data); + } else { + $data += [ + 'pool_id' => $poolId, + 'applied_member_id' => 0, + 'applied_version' => 0, + 'attempts' => 0, + 'lock_token' => '', + 'lock_until' => 0, + 'create_time' => $now, + ]; + Db::name('qywx_promotion_range_sync')->insert($data); + } + } + + /** @return list> */ + private static function lockedMembers(int $poolId): array + { + return Db::name('qywx_promotion_pool_member') + ->where('pool_id', $poolId) + ->whereNull('delete_time') + ->order('id', 'asc') + ->lock(true) + ->select()->toArray(); + } + + /** @return array|null */ + private static function lockedSyncRow(int $poolId): ?array + { + $row = Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->lock(true)->find(); + + return $row ?: null; + } + + private static function promotionLinkId(int $poolId): int + { + $syncLinkId = (int) (Db::name('qywx_promotion_range_sync') + ->where('pool_id', $poolId)->value('promotion_link_id') ?? 0); + if ($syncLinkId > 0) { + return $syncLinkId; + } + + return (int) (Db::name('qywx_promotion_link') + ->where('pool_id', $poolId) + ->where('remote_link_id', '<>', '') + ->where('remote_status', '<>', 2) + ->whereNull('delete_time') + ->order('id', 'desc') + ->value('id') ?? 0); + } + + /** @param list> $members */ + private static function persistMemberCursors(array $members, int $now): void + { + foreach ($members as $member) { + Db::name('qywx_promotion_pool_member')->where('id', (int) $member['id'])->update([ + 'current_weight' => (int) ($member['current_weight'] ?? 0), + 'today_count' => max(0, (int) ($member['today_count'] ?? 0)), + 'today_date' => (string) ($member['today_date'] ?? '') ?: null, + 'total_count' => max(0, (int) ($member['total_count'] ?? 0)), + 'last_assigned_time' => max(0, (int) ($member['last_assigned_time'] ?? 0)), + 'update_time' => $now, + ]); + } + } +} diff --git a/server/app/common/service/qywx/QywxPromotionRangeSyncService.php b/server/app/common/service/qywx/QywxPromotionRangeSyncService.php new file mode 100644 index 000000000..232120420 --- /dev/null +++ b/server/app/common/service/qywx/QywxPromotionRangeSyncService.php @@ -0,0 +1,174 @@ +api = $api ?? new QywxCustomerAcquisitionApiService(); + } + + /** @return array{status:string,pool_id:int,member_id:int} */ + public function syncPool(int $poolId): array + { + $claim = Db::transaction(function () use ($poolId): ?array { + $row = Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->lock(true)->find(); + if (!$row) { + return null; + } + $status = (int) ($row['status'] ?? 0); + $now = time(); + if ($status === 0 || $status === 4 || ($status === 2 && (int) ($row['lock_until'] ?? 0) > $now)) { + return null; + } + if ($status === 3 && (int) ($row['next_retry'] ?? 0) > $now) { + return null; + } + $token = bin2hex(random_bytes(16)); + Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->update([ + 'status' => 2, + 'lock_token' => $token, + 'lock_until' => $now + 90, + 'attempts' => (int) ($row['attempts'] ?? 0) + 1, + 'last_error' => '', + 'update_time' => $now, + ]); + $row['lock_token'] = $token; + + return $row; + }); + if ($claim === null) { + return ['status' => 'noop', 'pool_id' => $poolId, 'member_id' => 0]; + } + + $token = (string) $claim['lock_token']; + $desiredVersion = (int) ($claim['desired_version'] ?? 0); + $memberId = (int) ($claim['desired_member_id'] ?? 0); + try { + $pool = Db::name('qywx_promotion_pool')->where('id', $poolId)->whereNull('delete_time')->find(); + $link = Db::name('qywx_promotion_link') + ->where('id', (int) ($claim['promotion_link_id'] ?? 0)) + ->whereNull('delete_time')->find(); + $member = Db::name('qywx_promotion_pool_member') + ->where('id', $memberId)->where('pool_id', $poolId)->whereNull('delete_time')->find(); + if (!$pool || !$link || !$member || (int) ($member['enabled'] ?? 0) !== 1) { + throw new RuntimeException('分流方案、官方链接或目标成员已失效'); + } + $remoteLinkId = trim((string) ($link['remote_link_id'] ?? '')); + $userId = trim((string) ($member['userid'] ?? '')); + if ($remoteLinkId === '' || $userId === '') { + throw new RuntimeException('官方链接 ID 或目标成员 userid 为空'); + } + + $this->api->updateLink([ + 'link_id' => $remoteLinkId, + 'link_name' => mb_substr((string) ($pool['name'] ?? '获客分流方案'), 0, 30), + 'range' => ['user_list' => [$userId]], + 'skip_verify' => (int) ($link['skip_verify'] ?? 0) === 1, + ]); + $response = $this->api->getLink($remoteLinkId); + $remote = QywxCustomerAcquisitionLinkService::normaliseRemoteResponse($response, $remoteLinkId); + $actualUserIds = $remote['range_userids']; + if ($actualUserIds !== [$userId]) { + throw new RuntimeException('企业微信返回的成员范围与待同步成员不一致'); + } + $url = $remote['url']; + $snapshot = json_encode($remote['snapshot'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES); + Db::name('qywx_promotion_link')->where('id', (int) $link['id'])->update([ + 'wecom_url' => $url, + 'remote_status' => 1, + 'range_user_json' => json_encode($actualUserIds, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES), + 'range_department_json' => json_encode($remote['range_department_ids'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES), + 'remote_snapshot' => $snapshot === false ? null : $snapshot, + 'last_sync_time' => time(), + 'sync_error' => '', + 'update_time' => time(), + ]); + + $freshVersion = (int) (Db::name('qywx_promotion_range_sync') + ->where('pool_id', $poolId)->value('desired_version') ?? 0); + Db::name('qywx_promotion_range_sync') + ->where('pool_id', $poolId) + ->where('lock_token', $token) + ->update([ + 'status' => $freshVersion === $desiredVersion ? 0 : 1, + 'applied_member_id' => $memberId, + 'applied_version' => $desiredVersion, + 'next_retry' => $freshVersion === $desiredVersion ? 0 : time(), + 'attempts' => 0, + 'lock_token' => '', + 'lock_until' => 0, + 'last_error' => '', + 'update_time' => time(), + ]); + + return ['status' => 'synced', 'pool_id' => $poolId, 'member_id' => $memberId]; + } catch (\Throwable $e) { + $attempts = max(1, (int) ($claim['attempts'] ?? 0) + 1); + Db::name('qywx_promotion_range_sync') + ->where('pool_id', $poolId) + ->where('lock_token', $token) + ->update([ + 'status' => 3, + 'next_retry' => time() + min(300, 15 * $attempts), + 'lock_token' => '', + 'lock_until' => 0, + 'last_error' => mb_substr($e->getMessage(), 0, 500), + 'update_time' => time(), + ]); + Db::name('qywx_promotion_link') + ->where('id', (int) ($claim['promotion_link_id'] ?? 0)) + ->update(['sync_error' => mb_substr($e->getMessage(), 0, 500), 'update_time' => time()]); + + throw $e; + } + } + + /** @return array{selected:int,synced:int,failed:int} */ + public function syncPending(int $limit = 100): array + { + $now = time(); + $blockedPoolIds = Db::name('qywx_promotion_range_sync') + ->where('status', 4) + ->where('next_retry', '>', 0) + ->where('next_retry', '<=', $now) + ->limit(min(500, max(1, $limit))) + ->column('pool_id'); + foreach ($blockedPoolIds as $blockedPoolId) { + QywxPromotionMemberSchedulerService::reconcilePool((int) $blockedPoolId); + } + $poolIds = Db::name('qywx_promotion_range_sync') + ->where(function ($query) use ($now): void { + $query->whereIn('status', [1, 3])->where('next_retry', '<=', $now) + ->whereOr(function ($running) use ($now): void { + $running->where('status', 2)->where('lock_until', '<=', $now); + }); + }) + ->order('next_retry', 'asc') + ->limit(min(500, max(1, $limit))) + ->column('pool_id'); + $synced = 0; + $failed = 0; + foreach ($poolIds as $poolId) { + try { + $result = $this->syncPool((int) $poolId); + if ($result['status'] === 'synced') { + $synced++; + } + } catch (\Throwable) { + $failed++; + } + } + + return ['selected' => count($poolIds), 'synced' => $synced, 'failed' => $failed]; + } +} diff --git a/server/app/common/service/qywx/QywxPromotionWeightedRandom.php b/server/app/common/service/qywx/QywxPromotionWeightedRandom.php new file mode 100644 index 000000000..de74cc025 --- /dev/null +++ b/server/app/common/service/qywx/QywxPromotionWeightedRandom.php @@ -0,0 +1,83 @@ +> $members + * @param int|null $draw 测试用固定抽签值;生产环境留空并使用 random_int。 + * @return array{selected_id:int,members:list>,eligible_count:int,total_weight:int} + */ + public static function select(array $members, string $today, int $now, ?int $draw = null): array + { + $eligibleIndexes = []; + $totalWeight = 0; + foreach ($members as $index => &$member) { + if ((string) ($member['today_date'] ?? '') !== $today) { + $member['today_date'] = $today; + $member['today_count'] = 0; + } + // current_weight 是旧版平滑轮询游标;随机模式不再使用,统一归零。 + $member['current_weight'] = 0; + if (!self::eligible($member, $now)) { + continue; + } + $weight = max(1, (int) ($member['weight'] ?? 1)); + $eligibleIndexes[] = ['index' => $index, 'weight' => $weight]; + $totalWeight += $weight; + } + unset($member); + + if ($eligibleIndexes === []) { + return [ + 'selected_id' => 0, + 'members' => array_values($members), + 'eligible_count' => 0, + 'total_weight' => 0, + ]; + } + + $ticket = $draw ?? random_int(1, $totalWeight); + if ($ticket < 1 || $ticket > $totalWeight) { + throw new InvalidArgumentException('加权随机抽签值超出有效范围'); + } + $cursor = 0; + $selectedIndex = (int) $eligibleIndexes[0]['index']; + foreach ($eligibleIndexes as $candidate) { + $cursor += (int) $candidate['weight']; + if ($ticket <= $cursor) { + $selectedIndex = (int) $candidate['index']; + break; + } + } + + return [ + 'selected_id' => (int) ($members[$selectedIndex]['id'] ?? 0), + 'members' => array_values($members), + 'eligible_count' => count($eligibleIndexes), + 'total_weight' => $totalWeight, + ]; + } + + /** @param array $member */ + public static function eligible(array $member, int $now): bool + { + if ((int) ($member['enabled'] ?? 0) !== 1 || (int) ($member['weight'] ?? 0) <= 0) { + return false; + } + $start = max(0, (int) ($member['active_start'] ?? 0)); + $end = max(0, (int) ($member['active_end'] ?? 0)); + if (($start > 0 && $start > $now) || ($end > 0 && $end < $now)) { + return false; + } + $limit = max(0, (int) ($member['daily_limit'] ?? 0)); + + return $limit === 0 || (int) ($member['today_count'] ?? 0) < $limit; + } +} diff --git a/server/sql/1.9.20260824/upgrade_qywx_promotion_member_dispatch.sql b/server/sql/1.9.20260824/upgrade_qywx_promotion_member_dispatch.sql new file mode 100644 index 000000000..84a1ed4ee --- /dev/null +++ b/server/sql/1.9.20260824/upgrade_qywx_promotion_member_dispatch.sql @@ -0,0 +1,81 @@ +-- 企业微信获客助手:单官方直链的成员规则与回调驱动调度 + +START TRANSACTION; + +CREATE TABLE IF NOT EXISTS `zyt_qywx_promotion_pool_member` ( + `id` bigint unsigned NOT NULL AUTO_INCREMENT, + `pool_id` int unsigned NOT NULL DEFAULT 0, + `admin_id` int unsigned NOT NULL DEFAULT 0, + `userid` varchar(128) NOT NULL DEFAULT '' COMMENT '企业微信 userid 快照', + `enabled` tinyint unsigned NOT NULL DEFAULT 1, + `weight` smallint unsigned NOT NULL DEFAULT 1, + `current_weight` bigint NOT NULL DEFAULT 0 COMMENT '旧版平滑轮询游标,随机调度模式保留兼容', + `daily_limit` int unsigned NOT NULL DEFAULT 0 COMMENT '0为不限;按回调确认的实际获客计数', + `today_count` int unsigned NOT NULL DEFAULT 0, + `today_date` date NULL DEFAULT NULL, + `total_count` bigint unsigned NOT NULL DEFAULT 0, + `active_start` int unsigned NOT NULL DEFAULT 0, + `active_end` int unsigned NOT NULL DEFAULT 0, + `last_assigned_time` int unsigned NOT NULL DEFAULT 0, + `remark` varchar(255) NOT NULL DEFAULT '', + `create_time` int unsigned NOT NULL DEFAULT 0, + `update_time` int unsigned NOT NULL DEFAULT 0, + `delete_time` int unsigned NULL DEFAULT NULL, + PRIMARY KEY (`id`), + UNIQUE KEY `uk_pool_userid` (`pool_id`,`userid`), + KEY `idx_pool_enabled` (`pool_id`,`enabled`,`delete_time`), + KEY `idx_admin` (`admin_id`), + KEY `idx_today` (`today_date`,`today_count`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='获客分流方案成员规则'; + +CREATE TABLE IF NOT EXISTS `zyt_qywx_promotion_dispatch_event` ( + `id` bigint unsigned NOT NULL AUTO_INCREMENT, + `event_key` char(64) NOT NULL COMMENT '方案+成员+客户的幂等键', + `pool_id` int unsigned NOT NULL DEFAULT 0, + `member_id` bigint unsigned NOT NULL DEFAULT 0, + `userid` varchar(128) NOT NULL DEFAULT '', + `external_userid` varchar(128) NOT NULL DEFAULT '', + `source` varchar(32) NOT NULL DEFAULT '', + `event_time` int unsigned NOT NULL DEFAULT 0, + `create_time` int unsigned NOT NULL DEFAULT 0, + PRIMARY KEY (`id`), + UNIQUE KEY `uk_event_key` (`event_key`), + KEY `idx_pool_time` (`pool_id`,`event_time`), + KEY `idx_member_time` (`member_id`,`event_time`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='获客成员实际承接幂等流水'; + +CREATE TABLE IF NOT EXISTS `zyt_qywx_promotion_range_sync` ( + `pool_id` int unsigned NOT NULL, + `promotion_link_id` int unsigned NOT NULL DEFAULT 0, + `desired_member_id` bigint unsigned NOT NULL DEFAULT 0, + `desired_version` bigint unsigned NOT NULL DEFAULT 0, + `applied_member_id` bigint unsigned NOT NULL DEFAULT 0, + `applied_version` bigint unsigned NOT NULL DEFAULT 0, + `status` tinyint unsigned NOT NULL DEFAULT 0 COMMENT '0已同步 1待同步 2同步中 3失败待重试 4无可用成员', + `attempts` int unsigned NOT NULL DEFAULT 0, + `next_retry` int unsigned NOT NULL DEFAULT 0, + `lock_token` char(32) NOT NULL DEFAULT '', + `lock_until` int unsigned NOT NULL DEFAULT 0, + `last_error` varchar(500) NOT NULL DEFAULT '', + `create_time` int unsigned NOT NULL DEFAULT 0, + `update_time` int unsigned NOT NULL DEFAULT 0, + PRIMARY KEY (`pool_id`), + KEY `idx_pending` (`status`,`next_retry`,`lock_until`), + KEY `idx_link` (`promotion_link_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='获客官方链接当前成员范围同步队列'; + +INSERT INTO `zyt_dev_crontab` ( + `name`, `type`, `system`, `remark`, `command`, `params`, + `status`, `expression`, `create_time`, `update_time` +) +SELECT + '同步企微获客分流成员', 1, 0, + '按回调实际承接结果、成员权重与每日上限更新同一条企业微信官方获客链接', + 'qywx:sync-promotion-ranges', '', 1, '* * * * *', UNIX_TIMESTAMP(), UNIX_TIMESTAMP() +FROM DUAL +WHERE NOT EXISTS ( + SELECT 1 FROM `zyt_dev_crontab` + WHERE `command` = 'qywx:sync-promotion-ranges' AND `delete_time` IS NULL +); + +COMMIT; diff --git a/server/tests/QywxCustomerAcquisitionLinkServiceTest.php b/server/tests/QywxCustomerAcquisitionLinkServiceTest.php new file mode 100644 index 000000000..6d1bc93b8 --- /dev/null +++ b/server/tests/QywxCustomerAcquisitionLinkServiceTest.php @@ -0,0 +1,71 @@ + 0, + 'errmsg' => 'ok', + 'link' => [ + 'link_name' => '官网获客', + 'url' => $base, + 'create_time' => 1787610000, + 'skip_verify' => true, + ], + 'range' => [ + 'user_list' => ['HuoYiSheng', 'HuoYiSheng', 'Li'], + 'department_list' => ['2'], + ], + 'priority_option' => ['priority_type' => 1], +], 'cawcdea778939b9097'); +$assert( + $officialDetail['range_userids'] === ['HuoYiSheng', 'Li'], + 'get 接口根级 range.user_list 未被正确解析' +); +$assert( + $officialDetail['range_department_ids'] === ['2'], + 'get 接口根级 range.department_list 未被正确解析' +); +$assert( + ($officialDetail['priority_option']['priority_type'] ?? 0) === 1, + 'get 接口根级 priority_option 未被正确解析' +); +$assert( + isset($officialDetail['snapshot']['range']), + '远端快照应保留完整响应,而不是只保留 link 节点' +); + +echo "QYWX_CUSTOMER_ACQUISITION_LINK_SERVICE_OK\n"; diff --git a/server/tests/QywxPromotionWeightedRandomTest.php b/server/tests/QywxPromotionWeightedRandomTest.php new file mode 100644 index 000000000..3e78ea2e5 --- /dev/null +++ b/server/tests/QywxPromotionWeightedRandomTest.php @@ -0,0 +1,46 @@ + 1, 'enabled' => 1, 'weight' => 2, 'current_weight' => 9, 'daily_limit' => 0, 'today_count' => 0, 'today_date' => $today], + ['id' => 2, 'enabled' => 1, 'weight' => 1, 'current_weight' => -3, 'daily_limit' => 0, 'today_count' => 0, 'today_date' => $today], + ['id' => 3, 'enabled' => 1, 'weight' => 1, 'current_weight' => 4, 'daily_limit' => 0, 'today_count' => 0, 'today_date' => $today], +]; + +$assert(QywxPromotionWeightedRandom::select($members, $today, $now, 1)['selected_id'] === 1, '权重 2 的第一个区间映射错误'); +$assert(QywxPromotionWeightedRandom::select($members, $today, $now, 2)['selected_id'] === 1, '权重 2 的第二个区间映射错误'); +$assert(QywxPromotionWeightedRandom::select($members, $today, $now, 3)['selected_id'] === 2, '第二名成员的随机区间映射错误'); +$fourth = QywxPromotionWeightedRandom::select($members, $today, $now, 4); +$assert($fourth['selected_id'] === 3, '第三名成员的随机区间映射错误'); +$assert($fourth['total_weight'] === 4 && $fourth['eligible_count'] === 3, '随机池权重合计错误'); +$assert(array_column($fourth['members'], 'current_weight') === [0, 0, 0], '旧版平滑游标没有归零'); + +$limited = QywxPromotionWeightedRandom::select([ + ['id' => 1, 'enabled' => 0, 'weight' => 10, 'current_weight' => 0, 'daily_limit' => 0, 'today_count' => 0, 'today_date' => $today], + ['id' => 2, 'enabled' => 1, 'weight' => 5, 'current_weight' => 0, 'daily_limit' => 3, 'today_count' => 3, 'today_date' => $today], + ['id' => 3, 'enabled' => 1, 'weight' => 1, 'current_weight' => 0, 'daily_limit' => 0, 'today_count' => 8, 'today_date' => $today], +], $today, $now, 1); +$assert($limited['selected_id'] === 3 && $limited['eligible_count'] === 1, '禁用成员或达到上限的成员仍进入随机池'); + +$reset = QywxPromotionWeightedRandom::select([ + ['id' => 4, 'enabled' => 1, 'weight' => 1, 'current_weight' => 0, 'daily_limit' => 1, 'today_count' => 1, 'today_date' => '2026-08-24'], +], $today, $now, 1); +$assert($reset['selected_id'] === 4 && $reset['members'][0]['today_count'] === 0, '跨日数量没有自动重置'); +$assert(QywxPromotionMemberSchedulerService::poolIdFromState('zyt_pool:123') === 123, 'customer_channel 方案 ID 解析失败'); +$assert(QywxPromotionMemberSchedulerService::poolIdFromState('qywx_ca:123') === 0, '不应接管其他系统的 customer_channel'); + +echo "QYWX_PROMOTION_WEIGHTED_RANDOM_OK\n";