ZYT/zyt
2
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

更新

Browse Source
This commit is contained in:
Your Name
2026-08-25 09:36:25 +08:00
parent 01c38d8c5b
commit 47094cc617
7 changed files with 823 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/app/common/service/qywx/QywxPromotionMemberSchedulerService.php
+335
View File
@@ -0,0 +1,335 @@
<?php
declare(strict_types=1);
namespace app\common\service\qywx;
use think\facade\Db;
/** 根据实际获客回调记账，并为同一个官方链接规划下一名承接成员。 */
class QywxPromotionMemberSchedulerService
{
public static function poolIdFromState(string $state): int
{
$state = trim($state);
return preg_match('/^zyt_pool:(\d+)$/', $state, $matches) === 1
? max(0, (int) $matches[1])
: 0;
}
/** @return array{status:string,pool_id:int,member_id:int,next_member_id:int} */
public static function recordFromState(
string $state,
string $userId,
string $externalUserId,
int $eventTime = 0,
string $source = 'external_contact'
): array {
$poolId = self::poolIdFromState($state);
if ($poolId <= 0) {
return ['status' => '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<array<string,mixed>> */
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<string,mixed>|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<array<string,mixed>> $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,
]);
}
}
}
server/app/common/service/qywx/QywxPromotionRangeSyncService.php
+174
View File
@@ -0,0 +1,174 @@
<?php
declare(strict_types=1);
namespace app\common\service\qywx;
use RuntimeException;
use think\facade\Db;
/** 把回调调度出的单个目标成员同步到同一条企业微信官方获客链接。 */
class QywxPromotionRangeSyncService
{
private QywxCustomerAcquisitionApiService $api;
public function __construct(?QywxCustomerAcquisitionApiService $api = null)
{
$this->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];
}
}
server/app/common/service/qywx/QywxPromotionWeightedRandom.php
+83
View File
@@ -0,0 +1,83 @@
<?php
declare(strict_types=1);
namespace app\common\service\qywx;
use InvalidArgumentException;
/** 获客方案成员的加权随机选择；数据库锁与持久化由调度服务负责。 */
class QywxPromotionWeightedRandom
{
/**
* @param list<array<string,mixed>> $members
* @param int|null $draw 测试用固定抽签值；生产环境留空并使用 random_int。
* @return array{selected_id:int,members:list<array<string,mixed>>,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<string,mixed> $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;
}
}