;rgb:0000/0000/0000
This commit is contained in:
@@ -79,56 +79,80 @@ class DifyChatService
|
||||
$formatted = null;
|
||||
|
||||
foreach (self::buildAttemptPlan($normalized['kept'], $normalized['dropped']) as $attempt) {
|
||||
$requestSpecs = self::buildRequestSpecs(
|
||||
$baseUrl,
|
||||
$model,
|
||||
$inputs,
|
||||
$query,
|
||||
$user,
|
||||
false,
|
||||
$attempt['files'],
|
||||
$attempt['omitted']
|
||||
);
|
||||
$lastResponse = null;
|
||||
$lastSpec = [];
|
||||
$fileRejected = false;
|
||||
$inputRejected = false;
|
||||
|
||||
foreach ($requestSpecs as $index => $requestSpec) {
|
||||
$elapsedSeconds = (int) floor(microtime(true) - $startedAt);
|
||||
$remainingTimeout = $timeout - $elapsedSeconds;
|
||||
if ($remainingTimeout < self::MIN_TIMEOUT) {
|
||||
return self::error(
|
||||
'UPSTREAM_TIMEOUT',
|
||||
'模型响应超时，请稍后重试',
|
||||
self::elapsedMilliseconds($startedAt)
|
||||
foreach (self::buildInputAttemptPlan($inputs) as $inputIndex => $attemptInputs) {
|
||||
if ($inputIndex > 0 && !$inputRejected) {
|
||||
break;
|
||||
}
|
||||
$requestSpecs = self::buildRequestSpecs(
|
||||
$baseUrl,
|
||||
$model,
|
||||
$attemptInputs,
|
||||
$query,
|
||||
$user,
|
||||
false,
|
||||
$attempt['files'],
|
||||
$attempt['omitted']
|
||||
);
|
||||
$lastResponse = null;
|
||||
$lastSpec = [];
|
||||
|
||||
foreach ($requestSpecs as $index => $requestSpec) {
|
||||
$elapsedSeconds = (int) floor(microtime(true) - $startedAt);
|
||||
$remainingTimeout = $timeout - $elapsedSeconds;
|
||||
if ($remainingTimeout < self::MIN_TIMEOUT) {
|
||||
return self::error(
|
||||
'UPSTREAM_TIMEOUT',
|
||||
'模型响应超时，请稍后重试',
|
||||
self::elapsedMilliseconds($startedAt)
|
||||
);
|
||||
}
|
||||
|
||||
$response = self::sendRequest(
|
||||
$requestSpec['url'],
|
||||
$requestSpec['payload'],
|
||||
$apiKey,
|
||||
$remainingTimeout
|
||||
);
|
||||
$lastResponse = $response;
|
||||
$lastSpec = $requestSpec;
|
||||
$fileRejected = $fileRejected || self::isFileRejection($response, $attempt['files']);
|
||||
$inputRejected = $inputRejected || self::isInputRejection(
|
||||
$response,
|
||||
$requestSpec,
|
||||
$attemptInputs
|
||||
);
|
||||
|
||||
// /v1 在两种协议中都是合法基址。仅在明确表示路径不存在时尝试另一协议，
|
||||
// 避免因业务参数错误而重复提交同一份临床数据。
|
||||
$hasFallback = isset($requestSpecs[$index + 1]);
|
||||
if (
|
||||
$hasFallback
|
||||
&& !$inputRejected
|
||||
&& self::shouldTryNextProtocol($response, false)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
$response = self::sendRequest(
|
||||
$requestSpec['url'],
|
||||
$requestSpec['payload'],
|
||||
$apiKey,
|
||||
$remainingTimeout
|
||||
);
|
||||
$lastResponse = $response;
|
||||
$lastSpec = $requestSpec;
|
||||
|
||||
// /v1 在两种协议中都是合法基址。仅在明确表示路径不存在时尝试另一协议，
|
||||
// 避免因业务参数错误而重复提交同一份临床数据。
|
||||
$hasFallback = isset($requestSpecs[$index + 1]);
|
||||
if ($hasFallback && in_array($response['http_code'], [404, 405, 501], true)) {
|
||||
$lastResponse = $lastResponse ?? ['body' => '', 'errno' => 0, 'http_code' => 0];
|
||||
$formatted = self::formatResponse($lastResponse, $startedAt);
|
||||
self::logUpstreamFailure($lastSpec, $lastResponse, $query, $attempt['files'], $formatted);
|
||||
if (!empty($formatted['ok'])) {
|
||||
return $formatted;
|
||||
}
|
||||
// Dify 只接受应用中已声明且满足长度约束的 inputs。病例正文已经完整
|
||||
// 放在 query 中，因此 invalid_param 时可安全地用空 inputs 重试一次。
|
||||
if ($inputIndex === 0 && $inputRejected) {
|
||||
continue;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
$lastResponse = $lastResponse ?? ['body' => '', 'errno' => 0, 'http_code' => 0];
|
||||
$formatted = self::formatResponse($lastResponse, $startedAt);
|
||||
self::logUpstreamFailure($lastSpec, $lastResponse, $query, $attempt['files'], $formatted);
|
||||
if (!empty($formatted['ok'])) {
|
||||
return $formatted;
|
||||
}
|
||||
// 附件整体被拒时退回纯文本重试，附件清单已在下一轮尝试中补齐。
|
||||
if (!self::shouldRetryWithoutFiles((int) $lastResponse['http_code'], $attempt['files'])) {
|
||||
if (!$fileRejected) {
|
||||
return $formatted;
|
||||
}
|
||||
}
|
||||
@@ -191,64 +215,83 @@ class DifyChatService
|
||||
$formatted = null;
|
||||
|
||||
foreach (self::buildAttemptPlan($normalized['kept'], $normalized['dropped']) as $attempt) {
|
||||
$requestSpecs = self::buildRequestSpecs(
|
||||
$baseUrl,
|
||||
$model,
|
||||
$inputs,
|
||||
$query,
|
||||
$user,
|
||||
true,
|
||||
$attempt['files'],
|
||||
$attempt['omitted']
|
||||
);
|
||||
$lastResponse = null;
|
||||
$lastSpec = [];
|
||||
$fileRejected = false;
|
||||
$inputRejected = false;
|
||||
|
||||
foreach ($requestSpecs as $index => $requestSpec) {
|
||||
$elapsedSeconds = (int) floor(microtime(true) - $startedAt);
|
||||
$remainingTimeout = $timeout - $elapsedSeconds;
|
||||
if ($remainingTimeout < self::MIN_TIMEOUT) {
|
||||
return self::error(
|
||||
'UPSTREAM_TIMEOUT',
|
||||
'模型响应超时，请稍后重试',
|
||||
self::elapsedMilliseconds($startedAt)
|
||||
foreach (self::buildInputAttemptPlan($inputs) as $inputIndex => $attemptInputs) {
|
||||
if ($inputIndex > 0 && !$inputRejected) {
|
||||
break;
|
||||
}
|
||||
$requestSpecs = self::buildRequestSpecs(
|
||||
$baseUrl,
|
||||
$model,
|
||||
$attemptInputs,
|
||||
$query,
|
||||
$user,
|
||||
true,
|
||||
$attempt['files'],
|
||||
$attempt['omitted']
|
||||
);
|
||||
$lastResponse = null;
|
||||
$lastSpec = [];
|
||||
|
||||
foreach ($requestSpecs as $index => $requestSpec) {
|
||||
$elapsedSeconds = (int) floor(microtime(true) - $startedAt);
|
||||
$remainingTimeout = $timeout - $elapsedSeconds;
|
||||
if ($remainingTimeout < self::MIN_TIMEOUT) {
|
||||
return self::error(
|
||||
'UPSTREAM_TIMEOUT',
|
||||
'模型响应超时，请稍后重试',
|
||||
self::elapsedMilliseconds($startedAt)
|
||||
);
|
||||
}
|
||||
|
||||
$response = self::sendStreamRequest(
|
||||
$requestSpec['protocol'],
|
||||
$requestSpec['url'],
|
||||
$requestSpec['payload'],
|
||||
$apiKey,
|
||||
$remainingTimeout,
|
||||
$onDelta,
|
||||
$shouldAbort
|
||||
);
|
||||
$lastResponse = $response;
|
||||
$lastSpec = $requestSpec;
|
||||
$fileRejected = $fileRejected || self::isFileRejection($response, $attempt['files']);
|
||||
$inputRejected = $inputRejected || self::isInputRejection(
|
||||
$response,
|
||||
$requestSpec,
|
||||
$attemptInputs
|
||||
);
|
||||
|
||||
// 只在尚未向下游发送任何文本、且明确为路径不支持时尝试另一协议。
|
||||
$hasFallback = isset($requestSpecs[$index + 1]);
|
||||
if (
|
||||
$hasFallback
|
||||
&& !$inputRejected
|
||||
&& self::shouldTryNextProtocol($response, true)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
$response = self::sendStreamRequest(
|
||||
$requestSpec['protocol'],
|
||||
$requestSpec['url'],
|
||||
$requestSpec['payload'],
|
||||
$apiKey,
|
||||
$remainingTimeout,
|
||||
$onDelta,
|
||||
$shouldAbort
|
||||
);
|
||||
$lastResponse = $response;
|
||||
$lastSpec = $requestSpec;
|
||||
|
||||
// 只在尚未向下游发送任何文本、且明确为路径不支持时尝试另一协议。
|
||||
$hasFallback = isset($requestSpecs[$index + 1]);
|
||||
$lastResponse = $lastResponse ?? self::emptyStreamResponse(0);
|
||||
$formatted = self::formatStreamResponse($lastResponse, $startedAt);
|
||||
self::logUpstreamFailure($lastSpec, $lastResponse, $query, $attempt['files'], $formatted);
|
||||
if (!empty($formatted['ok'])) {
|
||||
return $formatted;
|
||||
}
|
||||
if (
|
||||
$hasFallback
|
||||
&& empty($response['emitted'])
|
||||
&& in_array($response['http_code'], [404, 405, 501], true)
|
||||
$inputIndex === 0
|
||||
&& $inputRejected
|
||||
&& empty($lastResponse['emitted'])
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
$lastResponse = $lastResponse ?? self::emptyStreamResponse(0);
|
||||
$formatted = self::formatStreamResponse($lastResponse, $startedAt);
|
||||
self::logUpstreamFailure($lastSpec, $lastResponse, $query, $attempt['files'], $formatted);
|
||||
if (!empty($formatted['ok'])) {
|
||||
return $formatted;
|
||||
}
|
||||
// 已经推给医生的文本不能重复输出，因此只在一个字都没发出去时才降级重试。
|
||||
// 附件不可达时 Dify 会在 200 流里发 event:error，同样按附件问题降级。
|
||||
$fileRejected = self::shouldRetryWithoutFiles((int) $lastResponse['http_code'], $attempt['files'])
|
||||
|| (!empty($lastResponse['upstream_error']) && $attempt['files'] !== []);
|
||||
if (!empty($lastResponse['emitted']) || !$fileRejected) {
|
||||
return $formatted;
|
||||
}
|
||||
@@ -450,6 +493,66 @@ class DifyChatService
|
||||
return $attempts;
|
||||
}
|
||||
|
||||
/**
|
||||
* Dify 应用输入变量由发布时的表单定义决定。先保留结构化输入；若上游明确
|
||||
* 拒绝输入，再使用空对象兼容旧应用。病例正文始终在 query 中，不会丢失。
|
||||
*
|
||||
* @param array<string,mixed> $inputs
|
||||
* @return array<int,array<string,mixed>>
|
||||
*/
|
||||
private static function buildInputAttemptPlan(array $inputs): array
|
||||
{
|
||||
return $inputs === [] ? [[]] : [$inputs, []];
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string,mixed> $response
|
||||
* @param array<string,mixed> $requestSpec
|
||||
* @param array<string,mixed> $inputs
|
||||
*/
|
||||
private static function isInputRejection(array $response, array $requestSpec, array $inputs): bool
|
||||
{
|
||||
if (
|
||||
$inputs === []
|
||||
|| ($requestSpec['protocol'] ?? '') !== 'dify'
|
||||
|| (int) ($response['errno'] ?? 0) !== 0
|
||||
|| !empty($response['emitted'])
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
|
||||
$httpCode = (int) ($response['http_code'] ?? 0);
|
||||
if (in_array($httpCode, [413, 422], true)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
$upstreamCode = strtolower(self::responseUpstreamCode($response));
|
||||
if ($httpCode === 400) {
|
||||
// 标准 Dify 会给出 invalid_param；部分兼容网关只保留 400，因此空码
|
||||
// 也允许一次无 inputs 重试。额度、模型或应用状态错误不能重复提交。
|
||||
return $upstreamCode === ''
|
||||
|| in_array($upstreamCode, ['invalid_param', 'payload_too_large', 'request_too_large'], true);
|
||||
}
|
||||
|
||||
return !empty($response['upstream_error'])
|
||||
&& in_array(
|
||||
$upstreamCode,
|
||||
['invalid_param', 'payload_too_large', 'request_too_large'],
|
||||
true
|
||||
);
|
||||
}
|
||||
|
||||
/** @param array<string,mixed> $response */
|
||||
private static function responseUpstreamCode(array $response): string
|
||||
{
|
||||
$upstreamCode = self::cleanUpstreamCode($response['upstream_code'] ?? '');
|
||||
if ($upstreamCode !== '' || !isset($response['body'])) {
|
||||
return $upstreamCode;
|
||||
}
|
||||
$decoded = json_decode((string) $response['body'], true);
|
||||
return is_array($decoded) ? self::cleanUpstreamCode($decoded['code'] ?? '') : '';
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<int,array<string,string>> $files
|
||||
*/
|
||||
@@ -458,6 +561,26 @@ class DifyChatService
|
||||
return $files !== [] && in_array($httpCode, self::FILE_REJECTION_CODES, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断一次上游响应是否属于“这批附件我处理不了”。
|
||||
*
|
||||
* 除了 4xx 状态码，Dify 拉不到附件时会在 200 的 SSE 流里发 event:error，
|
||||
* 这两种形态都必须触发去掉附件的降级重试。
|
||||
*
|
||||
* @param array<string,mixed> $response
|
||||
* @param array<int,array<string,string>> $files
|
||||
*/
|
||||
private static function isFileRejection(array $response, array $files): bool
|
||||
{
|
||||
if ($files === [] || (int) ($response['errno'] ?? 0) !== 0) {
|
||||
return false;
|
||||
}
|
||||
if (self::shouldRetryWithoutFiles((int) ($response['http_code'] ?? 0), $files)) {
|
||||
return true;
|
||||
}
|
||||
return !empty($response['upstream_error']);
|
||||
}
|
||||
|
||||
/**
|
||||
* 把无法随请求送达的附件写成显式清单。模型必须知道这些资料存在但读不到，
|
||||
* 才不会把“没看到”当成“没有”。
|
||||
@@ -509,6 +632,30 @@ class DifyChatService
|
||||
return $baseUrl . '/v1/' . $endpoint;
|
||||
}
|
||||
|
||||
/**
|
||||
* Decide whether an ambiguous base URL should be tried with the other wire
|
||||
* protocol. Only a missing/unsupported endpoint is a blocking-mode protocol
|
||||
* signal. A 2xx stream with no delivered delta but no valid terminal frame
|
||||
* is also safe to retry. Business validation, authentication, rate-limit and
|
||||
* server failures retain their original diagnosis instead of being hidden.
|
||||
*
|
||||
* @param array<string,mixed> $response
|
||||
*/
|
||||
private static function shouldTryNextProtocol(array $response, bool $streaming): bool
|
||||
{
|
||||
if ((int) ($response['errno'] ?? 0) !== 0) {
|
||||
return false;
|
||||
}
|
||||
$httpCode = (int) ($response['http_code'] ?? 0);
|
||||
if (in_array($httpCode, [404, 405, 501], true)) {
|
||||
return true;
|
||||
}
|
||||
if (!$streaming || $httpCode < 200 || $httpCode >= 300 || !empty($response['emitted'])) {
|
||||
return false;
|
||||
}
|
||||
return !empty($response['upstream_error']) || empty($response['finished']);
|
||||
}
|
||||
|
||||
private static function isValidBaseUrl(string $baseUrl): bool
|
||||
{
|
||||
if (preg_match('/[\x00-\x20\x7f]/', $baseUrl)) {
|
||||
@@ -579,6 +726,21 @@ class DifyChatService
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* libcurl 7.32+ exposes CURLOPT_XFERINFOFUNCTION, while CentOS/RHEL 7 commonly
|
||||
* ships libcurl 7.29 with only CURLOPT_PROGRESSFUNCTION. Resolve the option
|
||||
* by name so loading this class never evaluates an undefined PHP constant.
|
||||
*/
|
||||
private static function curlProgressOption(): ?int
|
||||
{
|
||||
foreach (['CURLOPT_XFERINFOFUNCTION', 'CURLOPT_PROGRESSFUNCTION'] as $name) {
|
||||
if (defined($name)) {
|
||||
return (int) constant($name);
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<string,mixed> $payload
|
||||
* @param callable(string):mixed $onDelta
|
||||
@@ -636,7 +798,7 @@ class DifyChatService
|
||||
self::consumeStreamBytes($protocol, $buffer, $chunk, $state, $onDelta);
|
||||
return $state['callback_error'] ? 0 : strlen($chunk);
|
||||
};
|
||||
$progress = static function () use (&$state, $shouldAbort): int {
|
||||
$progress = static function (...$unused) use (&$state, $shouldAbort): int {
|
||||
if ($shouldAbort !== null && $shouldAbort()) {
|
||||
$state['client_aborted'] = true;
|
||||
return 1;
|
||||
@@ -644,7 +806,7 @@ class DifyChatService
|
||||
return 0;
|
||||
};
|
||||
|
||||
curl_setopt_array($ch, [
|
||||
$curlOptions = [
|
||||
CURLOPT_URL => $url,
|
||||
CURLOPT_POST => true,
|
||||
CURLOPT_POSTFIELDS => $body,
|
||||
@@ -660,9 +822,13 @@ class DifyChatService
|
||||
],
|
||||
CURLOPT_HEADERFUNCTION => $header,
|
||||
CURLOPT_WRITEFUNCTION => $write,
|
||||
CURLOPT_NOPROGRESS => false,
|
||||
CURLOPT_XFERINFOFUNCTION => $progress,
|
||||
]);
|
||||
];
|
||||
$progressOption = self::curlProgressOption();
|
||||
if ($progressOption !== null) {
|
||||
$curlOptions[CURLOPT_NOPROGRESS] = false;
|
||||
$curlOptions[$progressOption] = $progress;
|
||||
}
|
||||
curl_setopt_array($ch, $curlOptions);
|
||||
|
||||
curl_exec($ch);
|
||||
$errno = curl_errno($ch);
|
||||
@@ -1044,25 +1210,23 @@ class DifyChatService
|
||||
return;
|
||||
}
|
||||
$url = (string) ($requestSpec['url'] ?? '');
|
||||
$upstreamCode = (string) ($response['upstream_code'] ?? '');
|
||||
if ($upstreamCode === '' && isset($response['body'])) {
|
||||
$decoded = json_decode((string) $response['body'], true);
|
||||
$upstreamCode = is_array($decoded)
|
||||
? self::cleanUpstreamCode($decoded['code'] ?? '')
|
||||
: '';
|
||||
}
|
||||
Log::warning('prescription ai upstream request failed', [
|
||||
$context = [
|
||||
'protocol' => (string) ($requestSpec['protocol'] ?? ''),
|
||||
'endpoint_path' => (string) (parse_url($url, PHP_URL_PATH) ?? ''),
|
||||
'http_code' => (int) ($response['http_code'] ?? 0),
|
||||
'curl_errno' => (int) ($response['errno'] ?? 0),
|
||||
// 上游自有错误码（如 invalid_param），用于区分附件超限、鉴权、模型故障。
|
||||
'upstream_code' => $upstreamCode,
|
||||
'upstream_code' => self::responseUpstreamCode($response),
|
||||
'query_bytes' => strlen($query),
|
||||
'file_count' => count($files),
|
||||
'error_code' => (string) ($formatted['error_code'] ?? 'UNKNOWN'),
|
||||
'latency_ms' => (int) ($formatted['latency_ms'] ?? 0),
|
||||
]);
|
||||
];
|
||||
// ThinkPHP 文件日志不会自动输出未参与占位符替换的 context；显式序列化
|
||||
// 这组不含凭据、主机名、患者正文的诊断字段，确保线上日志真正可用。
|
||||
Log::warning(
|
||||
'prescription ai upstream request failed ' . json_encode($context, JSON_UNESCAPED_SLASHES)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -20,6 +20,7 @@ class DirectUploadService
|
||||
/** 视频允许的扩展名（沿用 config/project.file_video） */
|
||||
public const TYPE_VIDEO = 'video';
|
||||
public const TYPE_VOICE = 'voice';
|
||||
public const TYPE_DESKTOP_PACKAGE = 'desktop_package';
|
||||
|
||||
/** 默认凭证有效期 30 分钟 */
|
||||
public const DEFAULT_DURATION = 1800;
|
||||
@@ -28,6 +29,7 @@ class DirectUploadService
|
||||
private const MAX_SIZE = [
|
||||
self::TYPE_VIDEO => 2 * 1024 * 1024 * 1024, // 2GB
|
||||
self::TYPE_VOICE => 500 * 1024 * 1024, // 500MB
|
||||
self::TYPE_DESKTOP_PACKAGE => 2 * 1024 * 1024 * 1024, // 2GB
|
||||
];
|
||||
|
||||
/**
|
||||
@@ -36,7 +38,7 @@ class DirectUploadService
|
||||
* @return array
|
||||
* @throws Exception
|
||||
*/
|
||||
public static function issueCredentials(string $type): array
|
||||
public static function issueCredentials(string $type, int $adminId = 0, string $name = ''): array
|
||||
{
|
||||
if (!isset(self::MAX_SIZE[$type])) {
|
||||
throw new Exception('不支持的上传类型: ' . $type);
|
||||
@@ -54,9 +56,27 @@ class DirectUploadService
|
||||
throw new Exception('腾讯云 COS 配置不完整');
|
||||
}
|
||||
|
||||
$keyPrefix = self::buildKeyPrefix($type);
|
||||
$keyPrefix = self::buildKeyPrefix($type, $adminId);
|
||||
$objectKey = '';
|
||||
// 兼容前后端错峰发布：旧 uploader 只传 type，不传 name。
|
||||
// 新 uploader 仍使用更严格的单对象授权；旧版则限制在当前管理员当天目录，
|
||||
// 并在 confirm 阶段校验文件名、扩展名与实际对象。
|
||||
if ($type === self::TYPE_DESKTOP_PACKAGE && trim($name) !== '') {
|
||||
$extension = strtolower((string)pathinfo($name, PATHINFO_EXTENSION));
|
||||
$objectKey = $keyPrefix
|
||||
. (int)round(microtime(true) * 1000)
|
||||
. '-'
|
||||
. bin2hex(random_bytes(8))
|
||||
. ($extension !== '' ? '.' . $extension : '');
|
||||
self::validateFileExtension($type, $objectKey, $name);
|
||||
}
|
||||
$engine = new QcloudEngine($storageConfig);
|
||||
$sts = $engine->getStsCredentials($keyPrefix, self::MAX_SIZE[$type], self::DEFAULT_DURATION);
|
||||
$sts = $engine->getStsCredentials(
|
||||
$objectKey !== '' ? $objectKey : $keyPrefix,
|
||||
self::MAX_SIZE[$type],
|
||||
self::DEFAULT_DURATION,
|
||||
$objectKey !== ''
|
||||
);
|
||||
|
||||
return [
|
||||
'provider' => 'qcloud',
|
||||
@@ -66,6 +86,7 @@ class DirectUploadService
|
||||
'host' => $sts['host'],
|
||||
'cdn_domain' => rtrim((string)($storageConfig['domain'] ?? ''), '/'),
|
||||
'key_prefix' => $keyPrefix,
|
||||
'object_key' => $objectKey,
|
||||
'max_size' => self::MAX_SIZE[$type],
|
||||
'duration' => self::DEFAULT_DURATION,
|
||||
'expired_time' => $sts['expiredTime'],
|
||||
@@ -93,8 +114,7 @@ class DirectUploadService
|
||||
}
|
||||
|
||||
$key = ltrim((string)($params['key'] ?? ''), '/');
|
||||
$allowedPrefix = self::buildKeyPrefix($type);
|
||||
if ($key === '' || strpos($key, $allowedPrefix) !== 0) {
|
||||
if (!self::isAllowedObjectKey($type, $key, (int)($params['admin_id'] ?? 0))) {
|
||||
throw new Exception('对象 Key 非法');
|
||||
}
|
||||
|
||||
@@ -112,6 +132,7 @@ class DirectUploadService
|
||||
if ($name === '') {
|
||||
$name = basename($key);
|
||||
}
|
||||
self::validateFileExtension($type, $key, $name);
|
||||
if (strlen($name) > 128) {
|
||||
$name = substr($name, 0, 123) . substr($name, -5);
|
||||
}
|
||||
@@ -137,9 +158,16 @@ class DirectUploadService
|
||||
];
|
||||
}
|
||||
|
||||
private static function buildKeyPrefix(string $type): string
|
||||
private static function buildKeyPrefix(string $type, int $adminId = 0): string
|
||||
{
|
||||
return 'uploads/' . $type . '/' . date('Ymd') . '/';
|
||||
$prefix = 'uploads/' . $type . '/';
|
||||
if ($type === self::TYPE_DESKTOP_PACKAGE) {
|
||||
if ($adminId <= 0) {
|
||||
throw new Exception('安装包上传账号无效');
|
||||
}
|
||||
$prefix .= $adminId . '/';
|
||||
}
|
||||
return $prefix . date('Ymd') . '/';
|
||||
}
|
||||
|
||||
private static function resolveFileType(string $type): int
|
||||
@@ -150,4 +178,46 @@ class DirectUploadService
|
||||
default => FileEnum::FILE_TYPE,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 桌面安装包是可执行文件，只允许发布流程所需的 EXE / ZIP。
|
||||
*/
|
||||
private static function validateFileExtension(string $type, string $key, string $name): void
|
||||
{
|
||||
if ($type !== self::TYPE_DESKTOP_PACKAGE) {
|
||||
return;
|
||||
}
|
||||
|
||||
$nameExtension = strtolower((string)pathinfo($name, PATHINFO_EXTENSION));
|
||||
$keyExtension = strtolower((string)pathinfo($key, PATHINFO_EXTENSION));
|
||||
$allowedExtensions = ['exe', 'zip'];
|
||||
if (!in_array($nameExtension, $allowedExtensions, true)
|
||||
|| $nameExtension !== $keyExtension) {
|
||||
throw new Exception('桌面安装包仅支持 EXE 或 ZIP 文件');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 安装包 Key 绑定上传管理员，并兼容跨午夜完成的上传。
|
||||
*/
|
||||
private static function isAllowedObjectKey(string $type, string $key, int $adminId): bool
|
||||
{
|
||||
if ($key === '') {
|
||||
return false;
|
||||
}
|
||||
if ($type !== self::TYPE_DESKTOP_PACKAGE) {
|
||||
return strpos($key, self::buildKeyPrefix($type)) === 0;
|
||||
}
|
||||
if ($adminId <= 0) {
|
||||
return false;
|
||||
}
|
||||
|
||||
$ownerPrefix = 'uploads/' . self::TYPE_DESKTOP_PACKAGE . '/' . $adminId . '/';
|
||||
if (strpos($key, $ownerPrefix) !== 0) {
|
||||
return false;
|
||||
}
|
||||
$date = substr($key, strlen($ownerPrefix), 8);
|
||||
return in_array($date, [date('Ymd'), date('Ymd', time() - 86400)], true)
|
||||
&& substr($key, strlen($ownerPrefix) + 8, 1) === '/';
|
||||
}
|
||||
}
|
||||
|
||||
@@ -422,9 +422,17 @@ SQL;
|
||||
* multiplies facts. Enterprise tag channels use the normalized relation
|
||||
* table, while legacy name-only channels keep a deduplicated JSON fallback.
|
||||
*
|
||||
* When the fact has an employee dimension, pass $followUserField so a tag
|
||||
* applied by employee A cannot make employee B's event match the channel.
|
||||
*
|
||||
* @param array<string, mixed>|null $channel
|
||||
*/
|
||||
public static function applyExternalUserChannelFilter(Query $query, string $field, ?array $channel): void
|
||||
public static function applyExternalUserChannelFilter(
|
||||
Query $query,
|
||||
string $field,
|
||||
?array $channel,
|
||||
?string $followUserField = null
|
||||
): void
|
||||
{
|
||||
if ($channel === null) {
|
||||
return;
|
||||
@@ -437,10 +445,14 @@ SQL;
|
||||
$tagPredicate = count($tagIds) === 1
|
||||
? 'channel_tag.tag_id = ?'
|
||||
: 'channel_tag.tag_id IN (' . implode(', ', array_fill(0, count($tagIds), '?')) . ')';
|
||||
$followUserPredicate = $followUserField === null
|
||||
? ''
|
||||
: "AND channel_tag.follow_user_id = {$followUserField} ";
|
||||
// 相关 EXISTS 走 (tag_id, external_userid) 索引，避免先物化整渠客户 ID 再 IN。
|
||||
$query->whereRaw(
|
||||
"EXISTS (SELECT 1 FROM {$tagTable} channel_tag "
|
||||
. "WHERE channel_tag.external_userid = {$field} "
|
||||
. $followUserPredicate
|
||||
. "AND {$tagPredicate} "
|
||||
. "AND EXISTS (SELECT 1 FROM {$contactTable} active_channel_contact "
|
||||
. 'WHERE active_channel_contact.external_userid = channel_tag.external_userid '
|
||||
@@ -472,6 +484,92 @@ SQL;
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Filter an add-event fact by its append-only channel snapshot. Events that
|
||||
* pre-date the snapshot migration explicitly fall back to the old projection,
|
||||
* but the fallback is constrained to the event's exact employee.
|
||||
*
|
||||
* @param array<string, mixed>|null $channel
|
||||
*/
|
||||
public static function applyExternalUserEventChannelFilter(
|
||||
Query $query,
|
||||
string $eventIdField,
|
||||
string $externalUserField,
|
||||
string $followUserField,
|
||||
?array $channel,
|
||||
bool $historicalContact = false
|
||||
): void
|
||||
{
|
||||
if ($channel === null) {
|
||||
return;
|
||||
}
|
||||
|
||||
$tagIds = self::channelTagIds($channel);
|
||||
if ($tagIds === [] || !QywxExternalContactEventTagSnapshotService::installed()) {
|
||||
if ($historicalContact) {
|
||||
self::applyHistoricalExternalUserChannelFilter(
|
||||
$query,
|
||||
$externalUserField,
|
||||
$channel,
|
||||
$followUserField
|
||||
);
|
||||
} else {
|
||||
self::applyExternalUserChannelFilter($query, $externalUserField, $channel, $followUserField);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
$snapshotTable = self::tableWithPrefix('qywx_external_contact_event_tag');
|
||||
$snapshotPredicate = count($tagIds) === 1
|
||||
? 'event_channel_tag.tag_id = ?'
|
||||
: 'event_channel_tag.tag_id IN (' . implode(', ', array_fill(0, count($tagIds), '?')) . ')';
|
||||
$snapshotMatch = "EXISTS (SELECT 1 FROM {$snapshotTable} event_channel_tag"
|
||||
. " WHERE event_channel_tag.event_id = {$eventIdField}"
|
||||
. " AND event_channel_tag.follow_user_id = {$followUserField}"
|
||||
. " AND {$snapshotPredicate})";
|
||||
$snapshotMissing = "NOT EXISTS (SELECT 1 FROM {$snapshotTable} captured_event_channel"
|
||||
. " WHERE captured_event_channel.event_id = {$eventIdField}"
|
||||
. " AND captured_event_channel.follow_user_id = {$followUserField}"
|
||||
. " AND captured_event_channel.tag_id = '')";
|
||||
|
||||
$query->where(function ($channelQuery) use (
|
||||
$snapshotMatch,
|
||||
$snapshotMissing,
|
||||
$tagIds,
|
||||
$historicalContact,
|
||||
$externalUserField,
|
||||
$followUserField,
|
||||
$channel
|
||||
): void {
|
||||
$channelQuery->whereRaw($snapshotMatch, $tagIds)
|
||||
->whereOr(function ($legacyQuery) use (
|
||||
$snapshotMissing,
|
||||
$historicalContact,
|
||||
$externalUserField,
|
||||
$followUserField,
|
||||
$channel
|
||||
): void {
|
||||
$legacyQuery->whereRaw($snapshotMissing);
|
||||
if ($historicalContact) {
|
||||
self::applyHistoricalExternalUserChannelFilter(
|
||||
$legacyQuery,
|
||||
$externalUserField,
|
||||
$channel,
|
||||
$followUserField
|
||||
);
|
||||
} else {
|
||||
self::applyExternalUserChannelFilter(
|
||||
$legacyQuery,
|
||||
$externalUserField,
|
||||
$channel,
|
||||
$followUserField
|
||||
);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Filter an external_userid fact by the channel snapshot retained in
|
||||
* qywx_external_contact.follow_users, including soft-deleted contacts.
|
||||
@@ -481,14 +579,48 @@ SQL;
|
||||
* contact row itself retains follow_users and is the best available channel
|
||||
* snapshot for this specific historical statistic.
|
||||
*
|
||||
* When $followUserField is provided, JSON_SEARCH first locates that exact
|
||||
* employee object and JSON_CONTAINS checks only its tag array.
|
||||
*
|
||||
* @param array<string, mixed>|null $channel
|
||||
*/
|
||||
public static function applyHistoricalExternalUserChannelFilter(Query $query, string $field, ?array $channel): void
|
||||
public static function applyHistoricalExternalUserChannelFilter(
|
||||
Query $query,
|
||||
string $field,
|
||||
?array $channel,
|
||||
?string $followUserField = null
|
||||
): void
|
||||
{
|
||||
if ($channel === null) {
|
||||
return;
|
||||
}
|
||||
|
||||
$contactTable = self::tableWithPrefix('qywx_external_contact');
|
||||
$tagIds = self::channelTagIds($channel);
|
||||
if ($followUserField !== null && $tagIds !== []) {
|
||||
$safeFollowUsers = "IF(JSON_VALID(historical_channel_contact.follow_users),"
|
||||
. ' historical_channel_contact.follow_users, JSON_ARRAY())';
|
||||
$userPath = "JSON_UNQUOTE(JSON_SEARCH({$safeFollowUsers}, 'one', {$followUserField},"
|
||||
. " NULL, '$[*].userid'))";
|
||||
$tagsPath = "IFNULL(REPLACE({$userPath}, '.userid', '.tags'), '$.__missing__')";
|
||||
$tagsJson = "JSON_EXTRACT({$safeFollowUsers}, {$tagsPath})";
|
||||
$tagPredicates = [];
|
||||
$bindings = [];
|
||||
foreach ($tagIds as $tagId) {
|
||||
$tagPredicates[] = "JSON_CONTAINS({$tagsJson}, JSON_OBJECT('tag_id', ?))";
|
||||
$bindings[] = $tagId;
|
||||
}
|
||||
$query->whereRaw(
|
||||
"EXISTS (SELECT 1 FROM {$contactTable} historical_channel_contact"
|
||||
. " WHERE historical_channel_contact.external_userid = {$field}"
|
||||
. " AND {$userPath} IS NOT NULL"
|
||||
. ' AND (' . implode(' OR ', $tagPredicates) . '))',
|
||||
$bindings
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
$patterns = self::buildLikePatterns($channel);
|
||||
if ($patterns === []) {
|
||||
$query->whereRaw('1 = 0');
|
||||
@@ -502,7 +634,6 @@ SQL;
|
||||
$segments[] = 'historical_channel_contact.follow_users LIKE ?';
|
||||
$bindings[] = $pattern;
|
||||
}
|
||||
$contactTable = self::tableWithPrefix('qywx_external_contact');
|
||||
$query->whereRaw(
|
||||
"EXISTS (SELECT 1 FROM {$contactTable} historical_channel_contact"
|
||||
. " WHERE historical_channel_contact.external_userid = {$field}"
|
||||
|
||||
@@ -0,0 +1,287 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use think\facade\Db;
|
||||
use think\facade\Log;
|
||||
|
||||
/**
|
||||
* 企业微信新增事件的标签快照。
|
||||
*
|
||||
* qywx_external_contact_tag 是当前状态投影，客户改标签或删除后会被覆盖/清理；
|
||||
* 本服务只在 add_external_contact 发生时写入，之后永不更新或删除。
|
||||
*/
|
||||
class QywxExternalContactEventTagSnapshotService
|
||||
{
|
||||
public const SOURCE_CONTACT_DETAIL = 1;
|
||||
public const SOURCE_PROMOTION_TASK = 2;
|
||||
|
||||
private static ?bool $installed = null;
|
||||
|
||||
/**
|
||||
* 灰度发布保护：代码先于迁移生效时，读取侧可以显式降级而不是返回 500。
|
||||
*/
|
||||
public static function installed(): bool
|
||||
{
|
||||
if (self::$installed !== null) {
|
||||
return self::$installed;
|
||||
}
|
||||
|
||||
try {
|
||||
self::$installed = Db::name('qywx_external_contact_event_tag')->getFields() !== [];
|
||||
|
||||
return self::$installed;
|
||||
} catch (\Throwable $e) {
|
||||
$message = $e->getMessage();
|
||||
if (str_contains($message, '42S02')
|
||||
|| str_contains($message, '1146')
|
||||
|| str_contains($message, 'no such table')) {
|
||||
self::$installed = false;
|
||||
|
||||
return self::$installed;
|
||||
}
|
||||
|
||||
throw $e;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 从 /externalcontact/get 的 follow_user[] 中，只截取产生事件的员工标签。
|
||||
* 找不到该员工时不写空快照，避免把一次不完整同步误判为“当时无标签”。
|
||||
*
|
||||
* @param array<int, mixed> $followUsers
|
||||
*/
|
||||
public static function captureFromFollowUsers(int $eventId, string $followUserId, array $followUsers): void
|
||||
{
|
||||
$followUserId = trim($followUserId);
|
||||
if ($eventId <= 0 || $followUserId === '') {
|
||||
return;
|
||||
}
|
||||
|
||||
foreach ($followUsers as $followUser) {
|
||||
if (!is_array($followUser)
|
||||
|| trim((string) ($followUser['userid'] ?? '')) !== $followUserId) {
|
||||
continue;
|
||||
}
|
||||
|
||||
self::capture(
|
||||
$eventId,
|
||||
$followUserId,
|
||||
is_array($followUser['tags'] ?? null) ? $followUser['tags'] : [],
|
||||
self::SOURCE_CONTACT_DETAIL
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 回调入库后，用同一推广任务的不可变 config_json 补写快照。
|
||||
*/
|
||||
public static function captureForPromotionEvent(int $eventId): void
|
||||
{
|
||||
if ($eventId <= 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
$event = Db::name('qywx_external_contact_event')->where('id', $eventId)->find();
|
||||
if (!$event || (string) ($event['change_type'] ?? '') !== 'add_external_contact') {
|
||||
return;
|
||||
}
|
||||
|
||||
$task = Db::name('qywx_promotion_automation_task')
|
||||
->where('change_type', (string) $event['change_type'])
|
||||
->where('userid', (string) $event['user_id'])
|
||||
->where('external_userid', (string) $event['external_userid'])
|
||||
->where('event_time', (int) $event['event_time'])
|
||||
->find();
|
||||
if ($task) {
|
||||
self::captureFromPromotionTask($task, $eventId);
|
||||
}
|
||||
} catch (\Throwable $e) {
|
||||
self::logFailure($e, $eventId, 'promotion_event');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 标签动作成功后追加推广配置中的确定标签，但不写完成标记。
|
||||
* 推广配置只描述自动添加的标签，不能证明客户当时没有其他标签；完整快照由后续客户详情同步完成。
|
||||
*
|
||||
* @param array<string, mixed> $task
|
||||
*/
|
||||
public static function captureFromPromotionTask(array $task, int $knownEventId = 0): void
|
||||
{
|
||||
if ((string) ($task['change_type'] ?? '') !== 'add_external_contact') {
|
||||
return;
|
||||
}
|
||||
|
||||
$actions = json_decode((string) ($task['actions_json'] ?? ''), true);
|
||||
$tagStatus = is_array($actions)
|
||||
? (string) ($actions['tags']['status'] ?? '')
|
||||
: '';
|
||||
if ($tagStatus !== 'success') {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
$eventId = $knownEventId;
|
||||
if ($eventId <= 0) {
|
||||
$eventId = (int) Db::name('qywx_external_contact_event')
|
||||
->where('change_type', 'add_external_contact')
|
||||
->where('user_id', (string) ($task['userid'] ?? ''))
|
||||
->where('external_userid', (string) ($task['external_userid'] ?? ''))
|
||||
->where('event_time', (int) ($task['event_time'] ?? 0))
|
||||
->value('id');
|
||||
}
|
||||
if ($eventId <= 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
$tags = [];
|
||||
$config = json_decode((string) ($task['config_json'] ?? ''), true);
|
||||
foreach ((array) ($config['tag_ids'] ?? []) as $tagId) {
|
||||
$tagId = trim((string) $tagId);
|
||||
if ($tagId !== '') {
|
||||
$tags[] = ['tag_id' => $tagId];
|
||||
}
|
||||
}
|
||||
if ($tags === []) {
|
||||
return;
|
||||
}
|
||||
|
||||
self::appendTags(
|
||||
$eventId,
|
||||
(string) ($task['userid'] ?? ''),
|
||||
$tags,
|
||||
self::SOURCE_PROMOTION_TASK
|
||||
);
|
||||
} catch (\Throwable $e) {
|
||||
self::logFailure($e, $knownEventId, 'promotion_task');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 追加推广任务能够证明的标签，不写 tag_id='' 完成标记。
|
||||
*
|
||||
* @param array<int, mixed> $tags
|
||||
*/
|
||||
private static function appendTags(int $eventId, string $followUserId, array $tags, int $source): void
|
||||
{
|
||||
$followUserId = mb_substr(trim($followUserId), 0, 64);
|
||||
if ($eventId <= 0 || $followUserId === '') {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
foreach (self::normalizeTags($tags) as $tag) {
|
||||
self::insertIgnore([
|
||||
'event_id' => $eventId,
|
||||
'follow_user_id' => $followUserId,
|
||||
'tag_id' => $tag['tag_id'],
|
||||
'tag_name' => $tag['tag_name'],
|
||||
'group_name' => $tag['group_name'],
|
||||
'snapshot_source' => $source,
|
||||
'create_time' => time(),
|
||||
]);
|
||||
}
|
||||
} catch (\Throwable $e) {
|
||||
self::logFailure($e, $eventId, 'append_tags');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 先写 tag_id='' 完成标记，再写真实标签；同一事务保证不会留下半份快照。
|
||||
* 完成标记已存在时直接返回，使重复回调不会把后来新增的标签补进历史事件。
|
||||
*
|
||||
* @param array<int, mixed> $tags
|
||||
*/
|
||||
private static function capture(int $eventId, string $followUserId, array $tags, int $source): void
|
||||
{
|
||||
$followUserId = mb_substr(trim($followUserId), 0, 64);
|
||||
if ($eventId <= 0 || $followUserId === '') {
|
||||
return;
|
||||
}
|
||||
|
||||
$normalized = self::normalizeTags($tags);
|
||||
|
||||
try {
|
||||
Db::transaction(static function () use ($eventId, $followUserId, $normalized, $source): void {
|
||||
$inserted = self::insertIgnore([
|
||||
'event_id' => $eventId,
|
||||
'follow_user_id' => $followUserId,
|
||||
'tag_id' => '',
|
||||
'tag_name' => '',
|
||||
'group_name' => '',
|
||||
'snapshot_source' => $source,
|
||||
'create_time' => time(),
|
||||
]);
|
||||
if ($inserted === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
foreach ($normalized as $tag) {
|
||||
self::insertIgnore([
|
||||
'event_id' => $eventId,
|
||||
'follow_user_id' => $followUserId,
|
||||
'tag_id' => $tag['tag_id'],
|
||||
'tag_name' => $tag['tag_name'],
|
||||
'group_name' => $tag['group_name'],
|
||||
'snapshot_source' => $source,
|
||||
'create_time' => time(),
|
||||
]);
|
||||
}
|
||||
});
|
||||
} catch (\Throwable $e) {
|
||||
// 快照是统计增强，迁移未执行或短时 DB 异常不能阻断企微回调主链路。
|
||||
self::logFailure($e, $eventId, 'capture');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param array<int, mixed> $tags
|
||||
* @return array<string, array{tag_id:string,tag_name:string,group_name:string}>
|
||||
*/
|
||||
private static function normalizeTags(array $tags): array
|
||||
{
|
||||
$normalized = [];
|
||||
foreach ($tags as $tag) {
|
||||
if (!is_array($tag)) {
|
||||
continue;
|
||||
}
|
||||
$tagId = mb_substr(trim((string) ($tag['tag_id'] ?? $tag['id'] ?? '')), 0, 64);
|
||||
if ($tagId === '') {
|
||||
continue;
|
||||
}
|
||||
$normalized[$tagId] = [
|
||||
'tag_id' => $tagId,
|
||||
'tag_name' => mb_substr((string) ($tag['tag_name'] ?? $tag['name'] ?? ''), 0, 128),
|
||||
'group_name' => mb_substr((string) ($tag['group_name'] ?? ''), 0, 128),
|
||||
];
|
||||
}
|
||||
|
||||
return $normalized;
|
||||
}
|
||||
|
||||
/** @param array<string, int|string> $row */
|
||||
private static function insertIgnore(array $row): int
|
||||
{
|
||||
$table = (string) config('database.connections.mysql.prefix')
|
||||
. 'qywx_external_contact_event_tag';
|
||||
$columns = array_keys($row);
|
||||
$sql = 'INSERT IGNORE INTO `' . $table . '` (`' . implode('`,`', $columns) . '`) VALUES ('
|
||||
. implode(',', array_fill(0, count($columns), '?')) . ')';
|
||||
|
||||
return Db::execute($sql, array_values($row));
|
||||
}
|
||||
|
||||
private static function logFailure(\Throwable $e, int $eventId, string $stage): void
|
||||
{
|
||||
Log::warning('qywx external contact event tag snapshot failed: ' . $e->getMessage(), [
|
||||
'event_id' => $eventId,
|
||||
'stage' => $stage,
|
||||
]);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,383 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use RuntimeException;
|
||||
|
||||
/** 推广客户自动化：短时欢迎语与可补偿关系动作分开消费。 */
|
||||
class QywxPromotionAutomationService
|
||||
{
|
||||
private QywxPromotionContactApiService $api;
|
||||
private QywxPromotionMediaService $media;
|
||||
private QywxPromotionAutomationStore $store;
|
||||
private QywxPromotionCodeCipher $cipher;
|
||||
private $clock;
|
||||
private const TERMINAL = ['sent', 'success', 'skipped', 'expired', 'uncertain', 'failed'];
|
||||
|
||||
public function __construct(
|
||||
?QywxPromotionContactApiService $api = null,
|
||||
?QywxPromotionMediaService $media = null,
|
||||
?QywxPromotionAutomationStore $store = null,
|
||||
?QywxPromotionCodeCipher $cipher = null,
|
||||
?callable $clock = null
|
||||
) {
|
||||
$this->api = $api ?? new QywxPromotionContactApiService();
|
||||
$this->media = $media ?? new QywxPromotionMediaService($this->api);
|
||||
$this->store = $store ?? new QywxPromotionAutomationStore();
|
||||
$this->cipher = $cipher ?? new QywxPromotionCodeCipher();
|
||||
$this->clock = $clock ?? static fn (): int => time();
|
||||
}
|
||||
|
||||
/**
|
||||
* 仅供验签解密后的回调调用。false代表沿用旧同步流程；已接管的入队错误必须返回HTTP500。
|
||||
* 默认仅入队；回调入口可启用即时通道，在当前请求内先发欢迎语并给正式客户打标。
|
||||
*/
|
||||
public function enqueueVerifiedEvent(array $event, bool $processImmediately = false): bool
|
||||
{
|
||||
$change = (string) ($event['ChangeType'] ?? '');
|
||||
if (!in_array($change, ['add_external_contact', 'add_half_external_contact'], true)) {
|
||||
return false;
|
||||
}
|
||||
$state = trim((string) ($event['State'] ?? ''));
|
||||
$linkId = trim((string) ($event['LinkId'] ?? $event['LinkID'] ?? ''));
|
||||
$userid = trim((string) ($event['UserID'] ?? $event['UserId'] ?? ''));
|
||||
$external = trim((string) ($event['ExternalUserID'] ?? $event['ExternalUserId'] ?? ''));
|
||||
if (($state === '' && $linkId === '') || $userid === '' || $external === '') {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
if (!$this->store->installed()) {
|
||||
return false;
|
||||
}
|
||||
$attribution = $this->store->attribution($state, $linkId, $userid);
|
||||
if ($attribution === null) {
|
||||
return false;
|
||||
}
|
||||
$now = $this->now();
|
||||
$eventTime = max(0, (int) ($event['CreateTime'] ?? 0));
|
||||
$code = (string) ($event['WelcomeCode'] ?? '');
|
||||
$config = $attribution['config'];
|
||||
$half = $change === 'add_half_external_contact';
|
||||
$welcomeStatus = 'pending';
|
||||
$reason = '';
|
||||
if (($config['welcome_mode'] ?? 'default') !== 'channel') {
|
||||
$welcomeStatus = 'skipped';
|
||||
$reason = 'mode_' . ($config['welcome_mode'] ?? 'default');
|
||||
} elseif ($code === '') {
|
||||
$welcomeStatus = 'skipped';
|
||||
$reason = 'missing_welcome_code';
|
||||
} elseif (strlen($code) > 1024) {
|
||||
$welcomeStatus = 'failed';
|
||||
$reason = 'invalid_welcome_code';
|
||||
} elseif ($eventTime <= 0 || $eventTime > $now + 5 || $eventTime + 20 <= $now) {
|
||||
$welcomeStatus = 'expired';
|
||||
$reason = 'welcome_window_elapsed_or_invalid_event_time';
|
||||
}
|
||||
$actions = [
|
||||
'welcome' => self::action($welcomeStatus, $reason),
|
||||
'tags' => self::action(!$half && !empty($config['tags_enabled']) ? 'pending' : 'skipped', $half ? 'half_contact' : 'disabled'),
|
||||
'remark' => self::action(!$half && !empty($config['remark_enabled']) ? 'pending' : 'skipped', $half ? 'half_contact' : 'disabled'),
|
||||
'description' => self::action(!$half && !empty($config['description_enabled']) ? 'pending' : 'skipped', $half ? 'half_contact' : 'disabled'),
|
||||
'dispatch' => self::action($half ? 'skipped' : 'pending', $half ? 'half_contact' : ''),
|
||||
'range' => self::action($half ? 'skipped' : 'pending', $half ? 'half_contact' : ''),
|
||||
'sync' => self::action($half ? 'skipped' : 'pending', $half ? 'half_contact' : ''),
|
||||
];
|
||||
$corp = (string) ($event['ToUserName'] ?? config('pay.wechat_work.corp_id', ''));
|
||||
$taskId = $this->store->enqueue([
|
||||
'event_key' => hash('sha256', implode('|', [$corp, $change, $userid, $external, (string) $eventTime])),
|
||||
'pool_id' => $attribution['pool_id'], 'member_admin_id' => $attribution['member_admin_id'],
|
||||
'change_type' => $change, 'userid' => $userid, 'external_userid' => $external,
|
||||
'event_time' => $eventTime, 'received_at' => $now,
|
||||
'config_json' => self::json($config), 'actions_json' => self::json($actions),
|
||||
'welcome_cipher' => $welcomeStatus === 'pending' ? $this->cipher->encrypt($code) : '',
|
||||
'welcome_code_hash' => $code !== '' ? hash('sha256', $code) : '',
|
||||
'welcome_expires_at' => $eventTime > 0 ? min($eventTime + 20, $now + 20) : 0,
|
||||
'welcome_status' => $welcomeStatus, 'welcome_next_retry' => 0,
|
||||
'status' => self::allTerminal($actions) ? 'done' : 'pending', 'next_retry' => 0,
|
||||
'lock_token' => '', 'lock_until' => 0, 'create_time' => $now, 'update_time' => $now,
|
||||
]);
|
||||
if ($processImmediately) {
|
||||
// 部署环境暂未启动常驻 worker 时仍要抢住 20 秒欢迎码窗口。
|
||||
// 标签只在正式客户事件执行；其余慢动作仍由分钟补偿处理。
|
||||
$this->consumeIds('welcome', [$taskId]);
|
||||
if (!$half) {
|
||||
$this->consumeIds('inline_metadata', [$taskId], ['tags']);
|
||||
}
|
||||
}
|
||||
return true;
|
||||
} catch (\Throwable) {
|
||||
// 不附原异常，入库SQL可能包含密文和配置；回调层返回500触发企微重试。
|
||||
throw new QywxPromotionEnqueueException('推广自动化事件未能持久化，请检查数据库迁移和私有存储');
|
||||
}
|
||||
}
|
||||
|
||||
/** 常驻秒级worker仅处理欢迎语，不被范围/客户同步或大文件上传阻塞。 */
|
||||
public function processWelcomes(int $limit = 100): array
|
||||
{
|
||||
return $this->consume('welcome', $limit);
|
||||
}
|
||||
|
||||
/** 分钟补偿：过期欢迎语只记过期，绝不尝试补发。 */
|
||||
public function retryPending(int $limit = 100): array
|
||||
{
|
||||
return $this->consume('metadata', $limit);
|
||||
}
|
||||
|
||||
public static function selectWelcome(array $config, int $eventTime): array
|
||||
{
|
||||
if (!empty($config['welcome_schedule_enabled'])) {
|
||||
foreach ((array) ($config['welcome_schedule'] ?? []) as $slot) {
|
||||
if (QywxPromotionConfig::matches($slot, $eventTime)) {
|
||||
return ['text' => (string) ($slot['text'] ?? ''), 'attachments' => (array) ($slot['attachments'] ?? [])];
|
||||
}
|
||||
}
|
||||
}
|
||||
return ['text' => (string) ($config['welcome']['text'] ?? ''), 'attachments' => (array) ($config['welcome']['attachments'] ?? [])];
|
||||
}
|
||||
|
||||
private function consume(string $lane, int $limit): array
|
||||
{
|
||||
return $this->consumeIds($lane, $this->store->due($lane, $this->now(), $limit));
|
||||
}
|
||||
|
||||
/** @param list<int> $ids @param list<string>|null $metadataActions */
|
||||
private function consumeIds(string $lane, array $ids, ?array $metadataActions = null): array
|
||||
{
|
||||
$result = ['selected' => 0, 'processed' => 0, 'failed' => 0];
|
||||
foreach ($ids as $id) {
|
||||
++$result['selected'];
|
||||
try {
|
||||
$row = $this->store->claim($id, $lane, $this->now());
|
||||
if ($row === null) {
|
||||
continue;
|
||||
}
|
||||
$actions = json_decode($row['actions_json'], true, 512, JSON_THROW_ON_ERROR);
|
||||
$config = json_decode($row['config_json'], true, 512, JSON_THROW_ON_ERROR);
|
||||
// 即时标签与欢迎语共用同一任务，但不能把仍在 20 秒窗口内待重试的欢迎语判为过期。
|
||||
if ($lane !== 'inline_metadata' && !self::terminal($actions['welcome']['status'])) {
|
||||
if ($lane === 'welcome') {
|
||||
$this->welcome($row, $actions, $config);
|
||||
} else {
|
||||
$running = $actions['welcome']['status'] === 'running';
|
||||
$this->transition($row, $actions, 'welcome', $running ? 'uncertain' : 'expired',
|
||||
$running ? 'worker_interrupted_after_send_started' : 'welcome_worker_not_available_in_window');
|
||||
}
|
||||
}
|
||||
if ($lane !== 'welcome') {
|
||||
$this->metadata($row, $actions, $config, $metadataActions);
|
||||
}
|
||||
$row['lock_until'] = 0;
|
||||
$row['update_time'] = $this->now();
|
||||
$this->store->save($row);
|
||||
++$result['processed'];
|
||||
} catch (\Throwable) {
|
||||
// 失去DB/租约时保留running状态；欢迎语恢复时视为不确定，防止重复推送。
|
||||
++$result['failed'];
|
||||
}
|
||||
}
|
||||
return $result;
|
||||
}
|
||||
|
||||
private function welcome(array &$row, array &$actions, array $config): void
|
||||
{
|
||||
if ($actions['welcome']['status'] === 'running') {
|
||||
$this->transition($row, $actions, 'welcome', 'uncertain', 'worker_interrupted_after_send_started');
|
||||
return;
|
||||
}
|
||||
if ((int) $row['welcome_expires_at'] <= $this->now() + 1) {
|
||||
$this->transition($row, $actions, 'welcome', 'expired', 'welcome_window_elapsed');
|
||||
return;
|
||||
}
|
||||
$sendStarted = false;
|
||||
try {
|
||||
$message = self::selectWelcome($config, (int) $row['event_time']);
|
||||
$text = $message['text'];
|
||||
if (str_contains($text, '{customer_name}') || str_contains($text, '{employee_name}') || str_contains($text, '{add_time}')) {
|
||||
$names = $this->names($row, $text, true);
|
||||
$text = QywxPromotionConfig::render($text, $names['customer'], $names['employee'], (int) $row['event_time'], 1200);
|
||||
}
|
||||
$truncated = strlen($text) > 4000;
|
||||
$text = mb_strcut($text, 0, 4000, 'UTF-8');
|
||||
$attachments = $this->media->materialize($message['attachments'], $config);
|
||||
$code = $this->cipher->decrypt($row['welcome_cipher']);
|
||||
if ((int) $row['welcome_expires_at'] <= $this->now() + 1) {
|
||||
$this->transition($row, $actions, 'welcome', 'expired', 'welcome_window_elapsed_during_prepare');
|
||||
return;
|
||||
}
|
||||
// running先持久化：如果HTTP成功后进程/DB断开，恢复时绝不再次使用同一code。
|
||||
$this->transition($row, $actions, 'welcome', 'running', 'send_started');
|
||||
$sendStarted = true;
|
||||
try {
|
||||
$this->api->sendWelcome($code, $text, $attachments);
|
||||
$this->transition($row, $actions, 'welcome', 'sent', $truncated ? 'sent_text_truncated_4000_bytes' : 'sent');
|
||||
} catch (QywxPromotionContactApiException $e) {
|
||||
if ($e->uncertain) {
|
||||
$this->transition($row, $actions, 'welcome', 'uncertain', 'network_result_unknown_do_not_resend', $e->getCode());
|
||||
} elseif ($e->getCode() === 41051) {
|
||||
$this->transition($row, $actions, 'welcome', 'skipped', 'welcome_code_already_consumed', 41051);
|
||||
} else {
|
||||
$this->welcomeRetry($row, $actions, 'explicit_api_rejection', $e->getCode());
|
||||
}
|
||||
} catch (\Throwable) {
|
||||
$this->transition($row, $actions, 'welcome', 'uncertain', 'send_or_persist_result_unknown_do_not_resend');
|
||||
} finally {
|
||||
unset($code);
|
||||
}
|
||||
} catch (\Throwable $e) {
|
||||
// 准备阶段没有执行发送，可以安全重试，且不会把错误原文/欢迎码写日志。
|
||||
if ($sendStarted || $actions['welcome']['status'] === 'running') {
|
||||
throw $e;
|
||||
}
|
||||
$this->welcomeRetry($row, $actions, 'prepare_failed_check_media_credentials_or_key', (int) $e->getCode());
|
||||
}
|
||||
}
|
||||
|
||||
private function welcomeRetry(array &$row, array &$actions, string $reason, int $code): void
|
||||
{
|
||||
$expired = (int) $row['welcome_expires_at'] <= $this->now() + 2;
|
||||
$this->transition($row, $actions, 'welcome', $expired ? 'expired' : 'retry', $reason, $code, $this->now() + 1);
|
||||
}
|
||||
|
||||
/** @param list<string>|null $only */
|
||||
private function metadata(array &$row, array &$actions, array $config, ?array $only = null): void
|
||||
{
|
||||
$names = null;
|
||||
$namesToProcess = ['tags', 'remark', 'description', 'dispatch', 'range', 'sync'];
|
||||
if ($only !== null) {
|
||||
$namesToProcess = array_values(array_intersect($namesToProcess, $only));
|
||||
}
|
||||
foreach ($namesToProcess as $name) {
|
||||
if (self::terminal($actions[$name]['status']) || (int) ($actions[$name]['next_retry'] ?? 0) > $this->now()) {
|
||||
continue;
|
||||
}
|
||||
$this->transition($row, $actions, $name, 'running', 'started');
|
||||
try {
|
||||
switch ($name) {
|
||||
case 'tags':
|
||||
$this->api->markTags($row['userid'], $row['external_userid'], (array) $config['tag_ids']);
|
||||
break;
|
||||
case 'remark':
|
||||
$names = $names ?? $this->names($row, (string) $config['remark_template'], false);
|
||||
$remark = QywxPromotionConfig::render($config['remark_template'], $names['customer'], $names['employee'], (int) $row['event_time'], 20);
|
||||
$this->api->remark($row['userid'], $row['external_userid'], ['remark' => $remark]);
|
||||
break;
|
||||
case 'description':
|
||||
$this->api->remark($row['userid'], $row['external_userid'], ['description' => (string) $config['description']]);
|
||||
break;
|
||||
case 'dispatch':
|
||||
$this->store->dispatch($row);
|
||||
break;
|
||||
case 'range':
|
||||
$this->store->syncRange($row);
|
||||
break;
|
||||
case 'sync':
|
||||
$this->store->syncCustomer($row);
|
||||
break;
|
||||
}
|
||||
$this->transition($row, $actions, $name, 'success', 'completed');
|
||||
} catch (\Throwable $e) {
|
||||
$attempt = (int) $actions[$name]['attempts'];
|
||||
$failed = $attempt >= 10;
|
||||
$this->transition($row, $actions, $name, $failed ? 'failed' : 'retry',
|
||||
$failed ? 'retry_limit_reached' : 'action_failed', (int) $e->getCode(),
|
||||
$this->now() + min(3600, 15 * (2 ** min(8, $attempt))));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private function names(array $row, string $template, bool $welcome): array
|
||||
{
|
||||
$names = ['customer' => '', 'employee' => ''];
|
||||
try {
|
||||
$names = $this->store->localNames($row);
|
||||
} catch (\Throwable) {
|
||||
// 本地资料失败不妨碍欢迎语使用明确的文案兜底。
|
||||
}
|
||||
$budget = fn (): bool => !$welcome || (int) $row['welcome_expires_at'] > $this->now() + 7;
|
||||
if (str_contains($template, '{customer_name}') && $names['customer'] === ''
|
||||
&& $row['change_type'] !== 'add_half_external_contact' && $budget()) {
|
||||
try {
|
||||
$detail = $this->api->getExternalContact($row['external_userid']);
|
||||
$names['customer'] = (string) ($detail['external_contact']['name'] ?? '');
|
||||
} catch (\Throwable) {
|
||||
}
|
||||
}
|
||||
if (str_contains($template, '{employee_name}') && $budget()) {
|
||||
try {
|
||||
$user = $this->api->getUser($row['userid']);
|
||||
$names['employee'] = trim((string) ($user['name'] ?? '')) ?: $names['employee'];
|
||||
} catch (\Throwable) {
|
||||
// 通讯录姓名接口权限不足时回退后台成员称呼。
|
||||
}
|
||||
}
|
||||
$names['customer'] = $names['customer'] !== '' ? $names['customer'] : '您';
|
||||
$names['employee'] = $names['employee'] !== '' ? $names['employee'] : '客户顾问';
|
||||
return $names;
|
||||
}
|
||||
|
||||
private function transition(array &$row, array &$actions, string $name, string $status, string $reason, int $code = 0, int $retryAt = 0): void
|
||||
{
|
||||
$now = $this->now();
|
||||
$action = $actions[$name];
|
||||
if ($status === 'running' || ($name === 'welcome' && $status === 'retry' && $action['status'] !== 'running')) {
|
||||
++$action['attempts'];
|
||||
}
|
||||
$action = array_replace($action, ['status' => $status, 'reason' => $reason, 'error_code' => $code,
|
||||
'next_retry' => $retryAt, 'update_time' => $now]);
|
||||
if (self::terminal($status)) {
|
||||
$action['finished_at'] = $now;
|
||||
}
|
||||
$actions[$name] = $action;
|
||||
if ($name === 'welcome') {
|
||||
$row['welcome_status'] = $status;
|
||||
$row['welcome_next_retry'] = $retryAt;
|
||||
if (self::terminal($status)) {
|
||||
$row['welcome_cipher'] = '';
|
||||
}
|
||||
}
|
||||
$row['status'] = self::allTerminal($actions) ? 'done' : 'pending';
|
||||
$retry = [];
|
||||
foreach ($actions as $key => $value) {
|
||||
if ($key !== 'welcome' && !self::terminal($value['status'])) {
|
||||
$retry[] = (int) ($value['next_retry'] ?? 0);
|
||||
}
|
||||
}
|
||||
$row['next_retry'] = $retry === [] ? 0 : min($retry);
|
||||
$row['actions_json'] = self::json($actions);
|
||||
$row['update_time'] = $now;
|
||||
$this->store->save($row, ['action' => $name, 'status' => $status, 'attempt' => $action['attempts'],
|
||||
'reason' => $reason, 'error_code' => $code, 'create_time' => $now]);
|
||||
}
|
||||
|
||||
private static function action(string $status, string $reason = ''): array
|
||||
{
|
||||
return ['status' => $status, 'reason' => $status === 'pending' ? '' : $reason, 'attempts' => 0, 'error_code' => 0, 'next_retry' => 0];
|
||||
}
|
||||
|
||||
private static function allTerminal(array $actions): bool
|
||||
{
|
||||
foreach ($actions as $action) {
|
||||
if (!self::terminal($action['status'])) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private static function terminal(string $status): bool
|
||||
{
|
||||
return in_array($status, self::TERMINAL, true);
|
||||
}
|
||||
|
||||
private static function json(array $value): string
|
||||
{
|
||||
return json_encode($value, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR);
|
||||
}
|
||||
|
||||
private function now(): int
|
||||
{
|
||||
return (int) ($this->clock)();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use app\adminapi\logic\qywx\CustomerLogic;
|
||||
use RuntimeException;
|
||||
use think\facade\Db;
|
||||
|
||||
/** DB 存储与既有同步边界；单测替换此类后不初始化业务数据库。 */
|
||||
class QywxPromotionAutomationStore
|
||||
{
|
||||
public function installed(): bool
|
||||
{
|
||||
return QywxPromotionConfig::installed();
|
||||
}
|
||||
|
||||
/** State只能定位，必须再核验真实方案、正式官方链接与实际成员关系。 */
|
||||
public function attribution(string $state, string $linkId, string $userId): ?array
|
||||
{
|
||||
if ($state !== '') {
|
||||
if (!preg_match('/^zyt_pool:([1-9][0-9]{0,9})$/', $state, $match)) {
|
||||
return null;
|
||||
}
|
||||
$poolId = (int) $match[1];
|
||||
} elseif ($linkId !== '') {
|
||||
$poolId = (int) Db::name('qywx_promotion_link')->where('remote_link_id', $linkId)
|
||||
->where('remote_status', 1)->whereNull('delete_time')->value('pool_id');
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
$pool = Db::name('qywx_promotion_pool')->where('id', $poolId)->where('status', 1)->whereNull('delete_time')->find();
|
||||
$member = Db::name('qywx_promotion_pool_member')->where('pool_id', $poolId)->where('userid', $userId)
|
||||
->whereNull('delete_time')->find();
|
||||
$links = Db::name('qywx_promotion_link')->where('pool_id', $poolId)->where('remote_status', 1)
|
||||
->where('remote_link_id', '<>', '')->whereNull('delete_time');
|
||||
if ($linkId !== '') {
|
||||
$links->where('remote_link_id', $linkId);
|
||||
}
|
||||
// 不用 enabled/当日额度验证：真实回调可能比排班切换晚到，不能漏掉已归属该方案的成员。
|
||||
if (!$pool || !$member || !$links->find()) {
|
||||
return null;
|
||||
}
|
||||
$configRow = Db::name('qywx_promotion_config')->where('pool_id', $poolId)->find();
|
||||
if (!$configRow) {
|
||||
// 尚未保存新增配置的旧方案仍保持原同步链路，不强制依赖新worker。
|
||||
return null;
|
||||
}
|
||||
return ['pool_id' => $poolId, 'member_admin_id' => (int) $member['admin_id'],
|
||||
'config' => QywxPromotionConfig::decode($configRow['config_json'])];
|
||||
}
|
||||
|
||||
public function enqueue(array $row): int
|
||||
{
|
||||
$row['welcome_code_hash'] = $row['welcome_code_hash'] ?: null;
|
||||
// 同一code可能同时出现在half/add：唯一索引把欢迎语消费权固定在第一次任务。
|
||||
for ($attempt = 0; $attempt < 2; $attempt++) {
|
||||
if ($row['welcome_code_hash'] !== null
|
||||
&& Db::name('qywx_promotion_automation_task')->where('welcome_code_hash', $row['welcome_code_hash'])->find()) {
|
||||
$actions = json_decode($row['actions_json'], true, 512, JSON_THROW_ON_ERROR);
|
||||
$actions['welcome']['status'] = 'skipped';
|
||||
$actions['welcome']['reason'] = 'same_welcome_code_already_queued';
|
||||
$row['actions_json'] = json_encode($actions, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR);
|
||||
$row['welcome_status'] = 'skipped';
|
||||
$row['welcome_cipher'] = '';
|
||||
$row['welcome_code_hash'] = null;
|
||||
$pending = array_filter($actions, static fn (array $a): bool => in_array($a['status'], ['pending', 'retry', 'running'], true));
|
||||
$row['status'] = $pending === [] ? 'done' : 'pending';
|
||||
}
|
||||
try {
|
||||
return (int) Db::name('qywx_promotion_automation_task')->insertGetId($row);
|
||||
} catch (\Throwable $e) {
|
||||
$existing = Db::name('qywx_promotion_automation_task')->where('event_key', $row['event_key'])->value('id');
|
||||
if ($existing) {
|
||||
return (int) $existing;
|
||||
}
|
||||
if ($attempt === 1 || $row['welcome_code_hash'] === null) {
|
||||
throw $e;
|
||||
}
|
||||
}
|
||||
}
|
||||
throw new RuntimeException('推广任务入队失败');
|
||||
}
|
||||
|
||||
/** 两条消费通道：常驻worker只发欢迎语，分钟任务不锁住尚有时效的欢迎语任务。 */
|
||||
public function due(string $lane, int $now, int $limit): array
|
||||
{
|
||||
$query = Db::name('qywx_promotion_automation_task')->where('status', '<>', 'done')
|
||||
->where('lock_until', '<=', $now);
|
||||
if ($lane === 'welcome') {
|
||||
$query->whereIn('welcome_status', ['pending', 'retry', 'running'])->where('welcome_next_retry', '<=', $now)
|
||||
->order('welcome_expires_at', 'asc');
|
||||
} else {
|
||||
$query->where('next_retry', '<=', $now)->where(function ($q) use ($now) {
|
||||
$q->whereNotIn('welcome_status', ['pending', 'retry', 'running'])
|
||||
->whereOr('welcome_expires_at', '<=', $now);
|
||||
})->order('id', 'asc');
|
||||
}
|
||||
return array_map('intval', $query->limit(max(1, min(500, $limit)))->column('id'));
|
||||
}
|
||||
|
||||
public function claim(int $id, string $lane, int $now): ?array
|
||||
{
|
||||
return Db::transaction(function () use ($id, $lane, $now): ?array {
|
||||
$row = Db::name('qywx_promotion_automation_task')->where('id', $id)->lock(true)->find();
|
||||
if (!$row || $row['status'] === 'done' || (int) $row['lock_until'] > $now) {
|
||||
return null;
|
||||
}
|
||||
$pendingWelcome = in_array($row['welcome_status'], ['pending', 'retry', 'running'], true);
|
||||
if (($lane === 'welcome' && (!$pendingWelcome || (int) $row['welcome_next_retry'] > $now))
|
||||
|| ($lane === 'metadata' && (($pendingWelcome && (int) $row['welcome_expires_at'] > $now) || (int) $row['next_retry'] > $now))
|
||||
|| ($lane === 'inline_metadata' && (int) $row['next_retry'] > $now)) {
|
||||
return null;
|
||||
}
|
||||
$row['lock_token'] = bin2hex(random_bytes(16));
|
||||
$row['lock_until'] = $now + (in_array($lane, ['welcome', 'inline_metadata'], true) ? 30 : 600);
|
||||
Db::name('qywx_promotion_automation_task')->where('id', $id)->update([
|
||||
'lock_token' => $row['lock_token'], 'lock_until' => $row['lock_until'], 'update_time' => $now,
|
||||
]);
|
||||
return $row;
|
||||
});
|
||||
}
|
||||
|
||||
public function save(array $row, ?array $log = null): void
|
||||
{
|
||||
Db::transaction(function () use ($row, $log): void {
|
||||
$fields = array_intersect_key($row, array_flip([
|
||||
'actions_json', 'welcome_status', 'welcome_cipher', 'welcome_next_retry', 'status',
|
||||
'next_retry', 'lock_until', 'update_time',
|
||||
]));
|
||||
// 租约令牌校验不能依赖affected rows：同秒同值更新在MySQL可能返回0。
|
||||
$current = Db::name('qywx_promotion_automation_task')->where('id', $row['id'])->lock(true)->find();
|
||||
if (!$current || !hash_equals((string) $current['lock_token'], (string) $row['lock_token'])) {
|
||||
throw new RuntimeException('推广任务处理租约已失效');
|
||||
}
|
||||
Db::name('qywx_promotion_automation_task')->where('id', $row['id'])->update($fields);
|
||||
if ($log !== null) {
|
||||
Db::name('qywx_promotion_automation_action_log')->insert($log + ['task_id' => $row['id']]);
|
||||
}
|
||||
});
|
||||
|
||||
// 标签动作成功/终止后，以任务入队时的配置冻结事件渠道；重复调用由完成标记幂等保护。
|
||||
QywxExternalContactEventTagSnapshotService::captureFromPromotionTask($row);
|
||||
}
|
||||
|
||||
public function localNames(array $task): array
|
||||
{
|
||||
return [
|
||||
'customer' => (string) (Db::name('qywx_external_contact')->where('external_userid', $task['external_userid'])->value('name') ?? ''),
|
||||
'employee' => (string) (Db::name('admin')->where('id', $task['member_admin_id'])->value('name') ?? ''),
|
||||
];
|
||||
}
|
||||
|
||||
public function dispatch(array $task): void
|
||||
{
|
||||
$result = QywxPromotionMemberSchedulerService::recordFromState('zyt_pool:' . $task['pool_id'],
|
||||
$task['userid'], $task['external_userid'], (int) $task['event_time'], 'external_contact');
|
||||
if (!in_array($result['status'] ?? '', ['counted', 'counted_blocked', 'counted_stale', 'duplicate'], true)) {
|
||||
throw new RuntimeException('推广成员记账未完成');
|
||||
}
|
||||
}
|
||||
|
||||
public function syncRange(array $task): void
|
||||
{
|
||||
// range服务自身有持久重试与版本保护；此调用负责触发。
|
||||
(new QywxPromotionRangeSyncService())->syncPool((int) $task['pool_id']);
|
||||
}
|
||||
|
||||
public function syncCustomer(array $task): void
|
||||
{
|
||||
$started = time();
|
||||
$eventId = (int) Db::name('qywx_external_contact_event')
|
||||
->where('change_type', 'add_external_contact')
|
||||
->where('user_id', (string) $task['userid'])
|
||||
->where('external_userid', (string) $task['external_userid'])
|
||||
->where('event_time', (int) $task['event_time'])
|
||||
->value('id');
|
||||
CustomerLogic::upsertSingleExternalContactFromApi(
|
||||
$task['external_userid'],
|
||||
$eventId,
|
||||
(string) $task['userid']
|
||||
);
|
||||
// 旧方法在API空结果时只log并返回void；必须核验本地实际更新，避免把未同步记为成功。
|
||||
$updated = (int) Db::name('qywx_external_contact')->where('external_userid', $task['external_userid'])->value('update_time');
|
||||
if ($updated < $started) {
|
||||
throw new RuntimeException('推广客户资料尚未同步到本地');
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use RuntimeException;
|
||||
|
||||
/** 一次性欢迎码仅加密短存；密钥不写数据库。多节点须显式共享环境密钥。 */
|
||||
class QywxPromotionCodeCipher
|
||||
{
|
||||
private ?string $key;
|
||||
|
||||
public function __construct(?string $key = null)
|
||||
{
|
||||
$this->key = $key;
|
||||
}
|
||||
|
||||
public function encrypt(string $code): string
|
||||
{
|
||||
$iv = random_bytes(12);
|
||||
$tag = '';
|
||||
$encrypted = openssl_encrypt($code, 'aes-256-gcm', $this->key(), OPENSSL_RAW_DATA, $iv, $tag);
|
||||
if ($encrypted === false) {
|
||||
throw new RuntimeException('无法加密欢迎码');
|
||||
}
|
||||
return base64_encode($iv . $tag . $encrypted);
|
||||
}
|
||||
|
||||
public function decrypt(string $cipher): string
|
||||
{
|
||||
$value = base64_decode($cipher, true);
|
||||
if ($value === false || strlen($value) <= 28) {
|
||||
throw new RuntimeException('欢迎码密文无效');
|
||||
}
|
||||
$code = openssl_decrypt(substr($value, 28), 'aes-256-gcm', $this->key(), OPENSSL_RAW_DATA, substr($value, 0, 12), substr($value, 12, 16));
|
||||
if ($code === false) {
|
||||
throw new RuntimeException('欢迎码解密失败，请核对工作进程密钥');
|
||||
}
|
||||
return $code;
|
||||
}
|
||||
|
||||
private function key(): string
|
||||
{
|
||||
if ($this->key !== null) {
|
||||
if (strlen($this->key) < 32) {
|
||||
throw new RuntimeException('欢迎码加密密钥至少32字符');
|
||||
}
|
||||
return hash('sha256', $this->key, true);
|
||||
}
|
||||
$configured = (string) config('qywx_promotion_automation.encryption_key', '');
|
||||
if ($configured !== '') {
|
||||
$this->key = $configured;
|
||||
return $this->key();
|
||||
}
|
||||
$directory = root_path('runtime') . 'qywx_promotion_private';
|
||||
if (!is_dir($directory) && !@mkdir($directory, 0700, true) && !is_dir($directory)) {
|
||||
throw new RuntimeException('无法创建欢迎码私有密钥目录');
|
||||
}
|
||||
$path = $directory . DIRECTORY_SEPARATOR . 'welcome.key';
|
||||
$stream = @fopen($path, 'c+b');
|
||||
if ($stream === false) {
|
||||
throw new RuntimeException('无法读取欢迎码私有密钥');
|
||||
}
|
||||
try {
|
||||
// 首次回调和多个worker可能同时启动；读写均持锁，避免读取尚未写完的密钥。
|
||||
if (!flock($stream, LOCK_EX)) {
|
||||
throw new RuntimeException('无法锁定欢迎码私有密钥');
|
||||
}
|
||||
@chmod($path, 0600);
|
||||
$key = trim((string) stream_get_contents($stream));
|
||||
if ($key === '') {
|
||||
$key = bin2hex(random_bytes(32));
|
||||
rewind($stream);
|
||||
if (fwrite($stream, $key) !== strlen($key) || !fflush($stream)) {
|
||||
throw new RuntimeException('无法保存欢迎码私有密钥');
|
||||
}
|
||||
}
|
||||
if (!preg_match('/^[0-9a-f]{64}$/', $key)) {
|
||||
throw new RuntimeException('欢迎码私有密钥损坏，请恢复原密钥');
|
||||
}
|
||||
$this->key = $key;
|
||||
} finally {
|
||||
flock($stream, LOCK_UN);
|
||||
fclose($stream);
|
||||
}
|
||||
return $this->key();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,291 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use DateTimeImmutable;
|
||||
use DateTimeZone;
|
||||
use RuntimeException;
|
||||
use think\facade\Db;
|
||||
|
||||
/** 获客方案配置。时间规则统一使用 Asia/Shanghai，结束时间不包含在时段内。 */
|
||||
class QywxPromotionConfig
|
||||
{
|
||||
public static function defaults(): array
|
||||
{
|
||||
return [
|
||||
'reception_mode' => 'always', 'reception_schedule' => [],
|
||||
'backup_member_admin_ids' => [], 'backup_userids' => [],
|
||||
'tags_enabled' => false, 'tag_ids' => [],
|
||||
'remark_enabled' => false, 'remark_template' => '{customer_name}',
|
||||
'description_enabled' => false, 'description' => '',
|
||||
'welcome_mode' => 'default', 'welcome' => ['text' => '', 'attachments' => []],
|
||||
'welcome_schedule_enabled' => false, 'welcome_schedule' => [],
|
||||
];
|
||||
}
|
||||
|
||||
public static function installed(): bool
|
||||
{
|
||||
try {
|
||||
foreach ([
|
||||
'qywx_promotion_config',
|
||||
'qywx_promotion_media',
|
||||
'qywx_promotion_automation_task',
|
||||
'qywx_promotion_automation_action_log',
|
||||
] as $table) {
|
||||
if (Db::name($table)->getFields() === []) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
} catch (\Throwable $error) {
|
||||
// 仅旧部署未建表时回退。数据库故障不能退回全天路由、忽略排班配置。
|
||||
if (str_contains($error->getMessage(), '42S02')
|
||||
|| str_contains($error->getMessage(), '1146')
|
||||
|| str_contains($error->getMessage(), 'no such table')) {
|
||||
return false;
|
||||
}
|
||||
throw $error;
|
||||
}
|
||||
}
|
||||
|
||||
public static function assertInstalled(): void
|
||||
{
|
||||
if (!self::installed()) {
|
||||
throw new RuntimeException('请先执行 server/sql/1.9.20260831/add_wecom_promotion_automation.sql 安装获客配置与任务表');
|
||||
}
|
||||
}
|
||||
|
||||
public static function decode(mixed $json): array
|
||||
{
|
||||
$value = is_array($json) ? $json : json_decode((string) $json, true);
|
||||
return array_replace(self::defaults(), is_array($value) ? $value : []);
|
||||
}
|
||||
|
||||
public static function forPool(int $poolId): array
|
||||
{
|
||||
if (!self::installed()) {
|
||||
return self::defaults();
|
||||
}
|
||||
return self::decode(Db::name('qywx_promotion_config')->where('pool_id', $poolId)->value('config_json'));
|
||||
}
|
||||
|
||||
public static function save(int $poolId, array $config): void
|
||||
{
|
||||
self::assertInstalled();
|
||||
if ($poolId <= 0) {
|
||||
throw new RuntimeException('分流方案不存在，无法保存自动化配置');
|
||||
}
|
||||
$row = Db::name('qywx_promotion_config')->where('pool_id', $poolId)->find();
|
||||
$json = json_encode($config, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR);
|
||||
$data = ['config_json' => $json, 'update_time' => time()];
|
||||
if ($row) {
|
||||
Db::name('qywx_promotion_config')->where('pool_id', $poolId)->update($data);
|
||||
} else {
|
||||
Db::name('qywx_promotion_config')->insert($data + ['pool_id' => $poolId, 'create_time' => time()]);
|
||||
}
|
||||
$saved = Db::name('qywx_promotion_config')->where('pool_id', $poolId)->value('config_json');
|
||||
if (!is_string($saved) || !hash_equals($json, $saved)) {
|
||||
throw new RuntimeException('获客标签与欢迎语配置未能完整落库，请重试');
|
||||
}
|
||||
}
|
||||
|
||||
/** 不接受浏览器提供的 userid；成员归属必须经过现有后台数据权限校验后再绑定。 */
|
||||
public static function normalize(array $input): array
|
||||
{
|
||||
$config = self::defaults();
|
||||
foreach (['tags_enabled', 'remark_enabled', 'description_enabled', 'welcome_schedule_enabled'] as $key) {
|
||||
$value = $input[$key] ?? false;
|
||||
if (!in_array($value, [true, false, 0, 1, '0', '1'], true)) {
|
||||
throw new RuntimeException('配置开关格式不正确');
|
||||
}
|
||||
$config[$key] = in_array($value, [true, 1, '1'], true);
|
||||
}
|
||||
$config['reception_mode'] = self::choice($input['reception_mode'] ?? 'always', ['always', 'scheduled']);
|
||||
$config['welcome_mode'] = self::choice($input['welcome_mode'] ?? 'default', ['default', 'channel', 'none']);
|
||||
$config['backup_member_admin_ids'] = self::ids($input['backup_member_admin_ids'] ?? []);
|
||||
$config['reception_schedule'] = self::schedule($input['reception_schedule'] ?? [], true);
|
||||
if ($config['reception_mode'] === 'scheduled' && $config['reception_schedule'] === []) {
|
||||
throw new RuntimeException('自动上下线模式至少需要一个接待时段');
|
||||
}
|
||||
if ($config['reception_mode'] === 'scheduled' && $config['backup_member_admin_ids'] === []) {
|
||||
throw new RuntimeException('自动上下线须配置备用员工，避免非接待时段官方链接仍路由给原成员');
|
||||
}
|
||||
if (!is_array($input['tag_ids'] ?? [])) {
|
||||
throw new RuntimeException('客户标签格式不正确');
|
||||
}
|
||||
if (count($input['tag_ids'] ?? []) > 1) {
|
||||
// 兼容旧数组字段，但不能默默截断旧方案多选；编辑时须由用户重新确认单个标签。
|
||||
throw new RuntimeException('推广方案仅支持单个客户标签，请重新选择一个标签');
|
||||
}
|
||||
$tags = [];
|
||||
foreach ($input['tag_ids'] ?? [] as $tag) {
|
||||
if (!is_string($tag) || trim($tag) === '' || strlen($tag) > 128) {
|
||||
throw new RuntimeException('企业微信标签 ID 不正确');
|
||||
}
|
||||
$tags[] = trim($tag);
|
||||
}
|
||||
$config['tag_ids'] = array_values(array_unique($tags));
|
||||
if ($config['tags_enabled'] && count($config['tag_ids']) !== 1) {
|
||||
throw new RuntimeException('启用客户标签时请选择一个企业微信标签');
|
||||
}
|
||||
$config['remark_template'] = self::text($input['remark_template'] ?? '{customer_name}', 200, '客户备注模板');
|
||||
$config['description'] = self::text($input['description'] ?? '', 150, '客户描述');
|
||||
if ($config['remark_enabled'] && $config['remark_template'] === '') {
|
||||
throw new RuntimeException('请填写客户备注模板');
|
||||
}
|
||||
if ($config['description_enabled'] && $config['description'] === '') {
|
||||
throw new RuntimeException('请填写客户描述');
|
||||
}
|
||||
$config['welcome'] = self::message($input['welcome'] ?? []);
|
||||
$config['welcome_schedule'] = self::schedule($input['welcome_schedule'] ?? [], false);
|
||||
if ($config['welcome_mode'] === 'channel') {
|
||||
self::assertMessage($config['welcome']);
|
||||
if ($config['welcome_schedule_enabled'] && $config['welcome_schedule'] === []) {
|
||||
throw new RuntimeException('请添加分时段欢迎语');
|
||||
}
|
||||
}
|
||||
return $config;
|
||||
}
|
||||
|
||||
public static function matches(array $slot, int $timestamp): bool
|
||||
{
|
||||
$date = (new DateTimeImmutable('@' . $timestamp))->setTimezone(new DateTimeZone('Asia/Shanghai'));
|
||||
$minute = $date->format('H:i');
|
||||
$start = (string) ($slot['start'] ?? '');
|
||||
$end = (string) ($slot['end'] ?? '');
|
||||
$weekdays = array_map('intval', (array) ($slot['weekdays'] ?? []));
|
||||
$day = (int) $date->format('N');
|
||||
if ($start < $end) {
|
||||
return in_array($day, $weekdays, true) && $minute >= $start && $minute < $end;
|
||||
}
|
||||
// 跨午夜时段归属于开始日期，例如周一 22:00—02:00 包含周二凌晨。
|
||||
return ($minute >= $start && in_array($day, $weekdays, true))
|
||||
|| ($minute < $end && in_array($day === 1 ? 7 : $day - 1, $weekdays, true));
|
||||
}
|
||||
|
||||
public static function render(string $template, string $customer, string $employee, int $timestamp, int $limit): string
|
||||
{
|
||||
$date = (new DateTimeImmutable('@' . $timestamp))->setTimezone(new DateTimeZone('Asia/Shanghai'));
|
||||
return mb_substr(strtr($template, [
|
||||
'{customer_name}' => $customer, '{employee_name}' => $employee,
|
||||
'{add_time}' => $date->format('Y-m-d'),
|
||||
]), 0, $limit);
|
||||
}
|
||||
|
||||
private static function schedule(mixed $value, bool $reception): array
|
||||
{
|
||||
if (!is_array($value) || count($value) > 30) {
|
||||
throw new RuntimeException('每类时间规则最多配置 30 条');
|
||||
}
|
||||
$rows = [];
|
||||
foreach ($value as $row) {
|
||||
if (!is_array($row)) {
|
||||
throw new RuntimeException('时间规则格式不正确');
|
||||
}
|
||||
$days = self::ids($row['weekdays'] ?? []);
|
||||
if ($days === [] || max($days) > 7) {
|
||||
throw new RuntimeException('请选择星期一至星期日');
|
||||
}
|
||||
$start = (string) ($row['start'] ?? '');
|
||||
$end = (string) ($row['end'] ?? '');
|
||||
if (!preg_match('/^(?:[01]\d|2[0-3]):[0-5]\d$/', $start)
|
||||
|| !preg_match('/^(?:[01]\d|2[0-3]):[0-5]\d$/', $end) || $start === $end) {
|
||||
throw new RuntimeException('时段起止时间必须不同，格式为 HH:mm；全天在线请使用全天模式');
|
||||
}
|
||||
$clean = ['weekdays' => $days, 'start' => $start, 'end' => $end];
|
||||
if ($reception) {
|
||||
$clean['member_admin_ids'] = self::ids($row['member_admin_ids'] ?? []);
|
||||
if ($clean['member_admin_ids'] === []) {
|
||||
throw new RuntimeException('每个接待时段至少选择一名接待成员');
|
||||
}
|
||||
} else {
|
||||
$clean += self::message($row);
|
||||
self::assertMessage($clean);
|
||||
}
|
||||
$rows[] = $clean;
|
||||
}
|
||||
if (!$reception) {
|
||||
// 分时欢迎语不可重叠，避免靠数组顺序决定发送内容。
|
||||
$occupied = [];
|
||||
foreach ($rows as $row) {
|
||||
[$sh, $sm] = array_map('intval', explode(':', $row['start']));
|
||||
[$eh, $em] = array_map('intval', explode(':', $row['end']));
|
||||
$from = $sh * 60 + $sm;
|
||||
$to = $eh * 60 + $em;
|
||||
$duration = ($to - $from + 1440) % 1440;
|
||||
foreach ($row['weekdays'] as $day) {
|
||||
for ($i = 0; $i < $duration; $i++) {
|
||||
$key = (($day - 1) * 1440 + $from + $i) % 10080;
|
||||
if (isset($occupied[$key])) {
|
||||
throw new RuntimeException('分时段欢迎语的时间范围不能重叠');
|
||||
}
|
||||
$occupied[$key] = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return $rows;
|
||||
}
|
||||
|
||||
public static function message(mixed $value): array
|
||||
{
|
||||
if (!is_array($value) || !is_array($value['attachments'] ?? [])) {
|
||||
throw new RuntimeException('欢迎语格式不正确');
|
||||
}
|
||||
$attachments = array_values($value['attachments'] ?? []);
|
||||
if (count($attachments) > 9) {
|
||||
throw new RuntimeException('欢迎语最多添加 9 个附件');
|
||||
}
|
||||
// 附件的详细格式与素材权限由 API/素材服务进一步验证。
|
||||
foreach ($attachments as $attachment) {
|
||||
if (!is_array($attachment) || !in_array($attachment['msgtype'] ?? '', ['image', 'link', 'miniprogram', 'video', 'file'], true)) {
|
||||
throw new RuntimeException('不支持的欢迎语附件类型');
|
||||
}
|
||||
}
|
||||
$text = self::text($value['text'] ?? '', 1200, '欢迎语');
|
||||
if (strlen($text) > 4000) {
|
||||
throw new RuntimeException('欢迎语不能超过 4000 个 UTF-8 字节（表情通常占 4 字节）');
|
||||
}
|
||||
return ['text' => $text, 'attachments' => $attachments];
|
||||
}
|
||||
|
||||
private static function assertMessage(array $message): void
|
||||
{
|
||||
if (trim($message['text']) === '' && $message['attachments'] === []) {
|
||||
throw new RuntimeException('渠道欢迎语必须包含文字或附件');
|
||||
}
|
||||
}
|
||||
|
||||
private static function choice(mixed $value, array $choices): string
|
||||
{
|
||||
if (!is_string($value) || !in_array($value, $choices, true)) {
|
||||
throw new RuntimeException('不支持的配置模式');
|
||||
}
|
||||
return $value;
|
||||
}
|
||||
|
||||
private static function ids(mixed $value): array
|
||||
{
|
||||
if (!is_array($value) || count($value) > 500) {
|
||||
throw new RuntimeException('成员或星期列表格式不正确');
|
||||
}
|
||||
$result = [];
|
||||
foreach ($value as $id) {
|
||||
if ((!is_int($id) && !(is_string($id) && ctype_digit($id))) || (int) $id <= 0) {
|
||||
throw new RuntimeException('成员或星期 ID 必须是正整数');
|
||||
}
|
||||
$result[] = (int) $id;
|
||||
}
|
||||
return array_values(array_unique($result));
|
||||
}
|
||||
|
||||
private static function text(mixed $value, int $limit, string $label): string
|
||||
{
|
||||
if (!is_string($value) || mb_strlen($value) > $limit) {
|
||||
throw new RuntimeException($label . '不能超过 ' . $limit . ' 个字符');
|
||||
}
|
||||
return trim($value);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use RuntimeException;
|
||||
|
||||
/** 不保存 Guzzle 原异常，避免请求 URL / token / welcome_code 进入日志。 */
|
||||
class QywxPromotionContactApiException extends RuntimeException
|
||||
{
|
||||
public function __construct(string $message, int $code = 0, public bool $uncertain = false)
|
||||
{
|
||||
parent::__construct($message, $code);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,284 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use GuzzleHttp\Client;
|
||||
use GuzzleHttp\Exception\GuzzleException;
|
||||
use GuzzleHttp\Psr7\Utils;
|
||||
use RuntimeException;
|
||||
use think\facade\Cache;
|
||||
|
||||
/** 客户联系可调用自建应用；不使用对外收款应用 Secret。 */
|
||||
class QywxPromotionContactApiService
|
||||
{
|
||||
private const PROMOTION_TAG_GROUP = '推广渠道';
|
||||
|
||||
private Client $client;
|
||||
private string $corpId;
|
||||
private string $secret;
|
||||
private $tokenResolver;
|
||||
|
||||
public function __construct(?Client $client = null, ?callable $tokenResolver = null)
|
||||
{
|
||||
$this->corpId = trim((string) config('qywx_customer_acquisition.corp_id', ''))
|
||||
?: trim((string) config('pay.wechat_work.corp_id', ''));
|
||||
// 获客回调的 WelcomeCode 应交由相同的可调用应用发送。专用覆盖仅用于明确配置的同应用。
|
||||
$this->secret = trim((string) config('qywx_promotion_automation.contact_secret', ''))
|
||||
?: (trim((string) config('qywx_customer_acquisition.secret', ''))
|
||||
?: trim((string) config('pay.wechat_work.customer_contact_secret', '')));
|
||||
$caPath = dirname(__DIR__, 4) . '/cacert.pem';
|
||||
$this->client = $client ?? new Client([
|
||||
'base_uri' => 'https://qyapi.weixin.qq.com/',
|
||||
'timeout' => 3, 'connect_timeout' => 2, 'http_errors' => false,
|
||||
'verify' => is_file($caPath) ? $caPath : true, 'allow_redirects' => false,
|
||||
'headers' => ['Accept' => 'application/json'],
|
||||
]);
|
||||
$this->tokenResolver = $tokenResolver;
|
||||
}
|
||||
|
||||
public function credentialFingerprint(): string
|
||||
{
|
||||
return hash('sha256', $this->corpId . '|' . $this->secret);
|
||||
}
|
||||
|
||||
public function tagOptions(): array
|
||||
{
|
||||
$result = $this->request('POST', 'externalcontact/get_corp_tag_list', []);
|
||||
$groups = [];
|
||||
foreach ((array) ($result['tag_group'] ?? []) as $group) {
|
||||
if (!is_array($group) || !empty($group['deleted'])) {
|
||||
continue;
|
||||
}
|
||||
$tags = [];
|
||||
foreach ((array) ($group['tag'] ?? []) as $tag) {
|
||||
if (is_array($tag) && empty($tag['deleted']) && !empty($tag['id'])) {
|
||||
$tags[] = ['id' => (string) $tag['id'], 'name' => (string) ($tag['name'] ?? '')];
|
||||
}
|
||||
}
|
||||
$groups[] = ['group_id' => (string) ($group['group_id'] ?? ''),
|
||||
'group_name' => (string) ($group['group_name'] ?? ''), 'tag' => $tags];
|
||||
}
|
||||
return ['tag_groups' => $groups];
|
||||
}
|
||||
|
||||
/**
|
||||
* 自定义企业客户标签：只写固定分组，先查重；创建结果不确定时只读回，不再次创建。
|
||||
* @return array{tag:array{id:string,name:string},group_id:string,group_name:string,reused:bool}
|
||||
* @see https://developer.work.weixin.qq.com/document/path/92117
|
||||
*/
|
||||
public function createTag(string $name): array
|
||||
{
|
||||
if (!mb_check_encoding($name, 'UTF-8') || preg_match('/[\p{C}\x{2028}\x{2029}]/u', $name)) {
|
||||
throw new RuntimeException('标签名称不能包含控制字符或不可见格式字符');
|
||||
}
|
||||
$name = preg_replace('/^\p{Z}+|\p{Z}+$/u', '', trim($name)) ?? '';
|
||||
if ($name === '' || mb_strlen($name, 'UTF-8') > 30) {
|
||||
throw new RuntimeException('标签名称须为 1-30 个字符');
|
||||
}
|
||||
$groups = $this->tagOptions()['tag_groups'];
|
||||
$existing = $this->findPromotionTag($groups, $name, true);
|
||||
if ($existing !== null) {
|
||||
return $existing;
|
||||
}
|
||||
$body = ['tag' => [['name' => $name]]];
|
||||
foreach ($groups as $group) {
|
||||
if (($group['group_name'] ?? '') === self::PROMOTION_TAG_GROUP && ($group['group_id'] ?? '') !== '') {
|
||||
$body['group_id'] = $group['group_id'];
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!isset($body['group_id'])) {
|
||||
// 官方保证同名分组存在时向该组添加，不额外创建同名分组；空分组不受支持。
|
||||
$body['group_name'] = self::PROMOTION_TAG_GROUP;
|
||||
}
|
||||
$failure = null;
|
||||
try {
|
||||
$response = $this->request('POST', 'externalcontact/add_corp_tag', $body, true);
|
||||
$created = $this->findPromotionTag([(array) ($response['tag_group'] ?? [])], $name, false);
|
||||
if ($created !== null) {
|
||||
return $created;
|
||||
}
|
||||
} catch (QywxPromotionContactApiException $error) {
|
||||
$failure = $error;
|
||||
}
|
||||
// 同名并发、上游缺失返回ID或网络中断，均只读回一次。永不构造本地伪标签ID。
|
||||
try {
|
||||
$confirmed = $this->findPromotionTag($this->tagOptions()['tag_groups'], $name, true);
|
||||
if ($confirmed !== null) {
|
||||
return $confirmed;
|
||||
}
|
||||
} catch (\Throwable) {
|
||||
throw new RuntimeException('标签创建结果无法确认，请刷新标签列表核对，勿重复提交', (int) ($failure?->getCode() ?? 0));
|
||||
}
|
||||
if ($failure !== null && !$failure->uncertain) {
|
||||
throw new RuntimeException('企业微信标签创建失败[' . $failure->getCode() . ']，请检查客户联系应用权限或标签额度', $failure->getCode());
|
||||
}
|
||||
throw new RuntimeException('标签创建结果无法确认，请刷新标签列表核对，勿重复提交', (int) ($failure?->getCode() ?? 0));
|
||||
}
|
||||
|
||||
private function findPromotionTag(array $groups, string $name, bool $reused): ?array
|
||||
{
|
||||
foreach ($groups as $group) {
|
||||
if (!is_array($group) || !empty($group['deleted'])
|
||||
|| ($group['group_name'] ?? '') !== self::PROMOTION_TAG_GROUP
|
||||
|| !is_string($group['group_id'] ?? null) || $group['group_id'] === '') {
|
||||
continue;
|
||||
}
|
||||
foreach ((array) ($group['tag'] ?? []) as $tag) {
|
||||
if (is_array($tag) && empty($tag['deleted']) && ($tag['name'] ?? '') === $name
|
||||
&& is_string($tag['id'] ?? null) && $tag['id'] !== '') {
|
||||
return ['tag' => ['id' => $tag['id'], 'name' => $name],
|
||||
'group_id' => $group['group_id'], 'group_name' => self::PROMOTION_TAG_GROUP, 'reused' => $reused];
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public function getExternalContact(string $externalUserId, string $cursor = ''): array
|
||||
{
|
||||
$query = ['external_userid' => $externalUserId];
|
||||
if ($cursor !== '') {
|
||||
$query['cursor'] = $cursor;
|
||||
}
|
||||
return $this->request('GET', 'externalcontact/get', $query);
|
||||
}
|
||||
|
||||
public function getUser(string $userId): array
|
||||
{
|
||||
return $this->request('GET', 'user/get', ['userid' => $userId]);
|
||||
}
|
||||
|
||||
public function markTags(string $userId, string $externalUserId, array $tagIds): void
|
||||
{
|
||||
if ($tagIds === []) {
|
||||
throw new RuntimeException('企业标签不能为空');
|
||||
}
|
||||
$this->request('POST', 'externalcontact/mark_tag', [
|
||||
'userid' => $userId, 'external_userid' => $externalUserId,
|
||||
'add_tag' => array_values(array_unique($tagIds)),
|
||||
]);
|
||||
}
|
||||
|
||||
public function remark(string $userId, string $externalUserId, array $fields): void
|
||||
{
|
||||
$body = ['userid' => $userId, 'external_userid' => $externalUserId];
|
||||
foreach (['remark' => 20, 'description' => 150] as $field => $limit) {
|
||||
if (isset($fields[$field]) && $fields[$field] !== '') {
|
||||
if (!is_string($fields[$field]) || mb_strlen($fields[$field]) > $limit) {
|
||||
throw new RuntimeException('客户备注或描述长度不正确');
|
||||
}
|
||||
$body[$field] = $fields[$field];
|
||||
}
|
||||
}
|
||||
if (count($body) === 2) {
|
||||
throw new RuntimeException('没有启用需要修改的备注字段');
|
||||
}
|
||||
$this->request('POST', 'externalcontact/remark', $body);
|
||||
}
|
||||
|
||||
public function sendWelcome(string $code, string $text, array $attachments): void
|
||||
{
|
||||
if ($code === '' || strlen($code) > 1024 || strlen($text) > 4000
|
||||
|| count($attachments) > 9 || ($text === '' && $attachments === [])) {
|
||||
throw new RuntimeException('欢迎语内容或欢迎码格式不正确');
|
||||
}
|
||||
$body = ['welcome_code' => $code];
|
||||
if ($text !== '') {
|
||||
$body['text'] = ['content' => $text];
|
||||
}
|
||||
if ($attachments !== []) {
|
||||
$body['attachments'] = array_values($attachments);
|
||||
}
|
||||
$this->request('POST', 'externalcontact/send_welcome_msg', $body, true);
|
||||
}
|
||||
|
||||
/** 仅由私有素材服务传入受控文件流，不接受 URL 或请求提供的任意路径。 */
|
||||
public function uploadMedia($stream, string $type, string $filename): array
|
||||
{
|
||||
if (!is_resource($stream) || !in_array($type, ['image', 'video', 'file'], true)) {
|
||||
throw new RuntimeException('临时素材类型或文件流不正确');
|
||||
}
|
||||
return $this->request('POST', 'media/upload', ['type' => $type], false, [
|
||||
'multipart' => [['name' => 'media', 'contents' => Utils::streamFor($stream), 'filename' => $filename]],
|
||||
'timeout' => 45,
|
||||
]);
|
||||
}
|
||||
|
||||
/** 仅明确的 token 失效响应允许重取一次；欢迎语/标签创建的网络异常不能直接重发。 */
|
||||
private function request(string $method, string $path, array $body, bool $nonIdempotent = false, array $extra = [], bool $retried = false): array
|
||||
{
|
||||
$token = $this->accessToken();
|
||||
$options = $extra + ['query' => ['access_token' => $token]];
|
||||
if ($method === 'GET' || isset($extra['multipart'])) {
|
||||
$options['query'] += $body;
|
||||
} else {
|
||||
$options['json'] = $body === [] ? (object) [] : $body;
|
||||
}
|
||||
try {
|
||||
$response = $this->client->request($method, 'cgi-bin/' . $path, $options);
|
||||
} catch (GuzzleException) {
|
||||
throw new QywxPromotionContactApiException('企业微信客户联系接口网络异常', 0, $nonIdempotent);
|
||||
}
|
||||
$decoded = json_decode((string) $response->getBody(), true);
|
||||
if ($response->getStatusCode() < 200 || $response->getStatusCode() >= 300
|
||||
|| !is_array($decoded) || !array_key_exists('errcode', $decoded)) {
|
||||
// media/upload 成功返回可没有 errcode。
|
||||
if ($path === 'media/upload' && $response->getStatusCode() === 200 && is_array($decoded) && !empty($decoded['media_id'])) {
|
||||
return $decoded;
|
||||
}
|
||||
throw new QywxPromotionContactApiException('企业微信客户联系接口响应无法确认', 0, $nonIdempotent);
|
||||
}
|
||||
$code = (int) $decoded['errcode'];
|
||||
if ($code === 0) {
|
||||
return $decoded;
|
||||
}
|
||||
if (!$retried && in_array($code, [40001, 40014, 42001], true)) {
|
||||
if ($this->tokenResolver === null) {
|
||||
Cache::delete('qywx_promotion_contact_token:' . $this->credentialFingerprint());
|
||||
}
|
||||
if (isset($extra['multipart'])) {
|
||||
$extra['multipart'][0]['contents']->rewind();
|
||||
}
|
||||
return $this->request($method, $path, $body, $nonIdempotent, $extra, true);
|
||||
}
|
||||
// 不回显上游 errmsg；部分错误会包含请求参数与一次性凭证。
|
||||
throw new QywxPromotionContactApiException('企业微信客户联系接口失败[' . $code . ']', $code);
|
||||
}
|
||||
|
||||
private function accessToken(): string
|
||||
{
|
||||
if ($this->tokenResolver !== null) {
|
||||
$token = (string) ($this->tokenResolver)();
|
||||
if ($token === '') {
|
||||
throw new RuntimeException('客户联系托管 token 为空');
|
||||
}
|
||||
return $token;
|
||||
}
|
||||
if ($this->corpId === '' || $this->secret === '') {
|
||||
throw new RuntimeException('请配置客户联系可调用自建应用的 corp_id 和 Secret');
|
||||
}
|
||||
$key = 'qywx_promotion_contact_token:' . $this->credentialFingerprint();
|
||||
$token = (string) Cache::get($key, '');
|
||||
if ($token !== '') {
|
||||
return $token;
|
||||
}
|
||||
try {
|
||||
$response = $this->client->request('GET', 'cgi-bin/gettoken', [
|
||||
'query' => ['corpid' => $this->corpId, 'corpsecret' => $this->secret],
|
||||
]);
|
||||
} catch (GuzzleException) {
|
||||
throw new QywxPromotionContactApiException('获取客户联系 token 网络异常');
|
||||
}
|
||||
$data = json_decode((string) $response->getBody(), true);
|
||||
if ($response->getStatusCode() !== 200 || !is_array($data)
|
||||
|| (int) ($data['errcode'] ?? 0) !== 0 || empty($data['access_token'])) {
|
||||
throw new QywxPromotionContactApiException('获取客户联系 token 失败', (int) ($data['errcode'] ?? 0));
|
||||
}
|
||||
$token = (string) $data['access_token'];
|
||||
Cache::set($key, $token, max(60, (int) ($data['expires_in'] ?? 7200) - 300));
|
||||
return $token;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
class QywxPromotionEnqueueException extends \RuntimeException
|
||||
{
|
||||
}
|
||||
@@ -0,0 +1,303 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use RuntimeException;
|
||||
use think\file\UploadedFile;
|
||||
|
||||
/** 私有源文件 + 可刷新三天临时素材。欢迎语关键路径仅使用缓存，不下载/上传文件。 */
|
||||
class QywxPromotionMediaService
|
||||
{
|
||||
private QywxPromotionContactApiService $api;
|
||||
private QywxPromotionMediaStore $store;
|
||||
private string $root;
|
||||
|
||||
public function __construct(?QywxPromotionContactApiService $api = null, ?QywxPromotionMediaStore $store = null, ?string $root = null)
|
||||
{
|
||||
$this->api = $api ?? new QywxPromotionContactApiService();
|
||||
$this->store = $store ?? new QywxPromotionMediaStore();
|
||||
// runtime_path()在adminapi/api/CLI间不同；使用项目级私有目录保证上传与worker共享。
|
||||
$this->root = rtrim($root ?? (root_path('runtime') . 'qywx_promotion_private' . DIRECTORY_SEPARATOR . 'media'), '/\\');
|
||||
}
|
||||
|
||||
/** @return array{asset_id:string,name:string,type:string} */
|
||||
public function upload($file, string $type, int $adminId): array
|
||||
{
|
||||
if ($adminId <= 0 || !$file instanceof UploadedFile || !$file->isValid()) {
|
||||
throw new RuntimeException('请上传有效文件');
|
||||
}
|
||||
if (!in_array($type, ['image', 'video', 'file'], true)) {
|
||||
throw new RuntimeException('素材类型仅支持 image、video、file');
|
||||
}
|
||||
$size = (int) $file->getSize();
|
||||
$limit = ($type === 'file' ? 20 : 10) * 1024 * 1024;
|
||||
if ($size <= 5 || $size > $limit) {
|
||||
throw new RuntimeException($type === 'file' ? '文件须大于5字节且不超过20MB' : '图片/视频须大于5字节且不超过10MB');
|
||||
}
|
||||
$mime = (new \finfo(FILEINFO_MIME_TYPE))->file($file->getPathname());
|
||||
$name = str_replace('\\', '/', $file->getOriginalName());
|
||||
$name = mb_substr(preg_replace('/[\x00-\x1f\x7f]/u', '', basename($name)) ?? '', 0, 180);
|
||||
$extension = strtolower(pathinfo($name, PATHINFO_EXTENSION));
|
||||
if ($type === 'image') {
|
||||
$info = @getimagesize($file->getPathname());
|
||||
if (!in_array($mime, ['image/jpeg', 'image/png'], true) || $info === false
|
||||
|| !in_array($info[2], [IMAGETYPE_JPEG, IMAGETYPE_PNG], true)) {
|
||||
throw new RuntimeException('图片仅支持真实 JPG/PNG 文件');
|
||||
}
|
||||
$extension = $mime === 'image/png' ? 'png' : 'jpg';
|
||||
} elseif ($type === 'video') {
|
||||
if ($mime !== 'video/mp4' || $extension !== 'mp4') {
|
||||
throw new RuntimeException('视频仅支持 MP4');
|
||||
}
|
||||
} else {
|
||||
// 私有存储也拒绝可执行内容/HTML/SVG；按实际 MIME 与扩展名双重检查。
|
||||
$allowed = [
|
||||
'pdf' => ['application/pdf'], 'txt' => ['text/plain'], 'csv' => ['text/plain', 'text/csv', 'application/csv'],
|
||||
'doc' => ['application/msword', 'application/x-ole-storage', 'application/CDFV2'],
|
||||
'xls' => ['application/vnd.ms-excel', 'application/x-ole-storage', 'application/CDFV2'],
|
||||
'ppt' => ['application/vnd.ms-powerpoint', 'application/x-ole-storage', 'application/CDFV2'],
|
||||
'docx' => ['application/vnd.openxmlformats-officedocument.wordprocessingml.document', 'application/zip'],
|
||||
'xlsx' => ['application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'application/zip'],
|
||||
'pptx' => ['application/vnd.openxmlformats-officedocument.presentationml.presentation', 'application/zip'],
|
||||
'zip' => ['application/zip'], 'jpg' => ['image/jpeg'], 'jpeg' => ['image/jpeg'], 'png' => ['image/png'],
|
||||
'mp4' => ['video/mp4'],
|
||||
];
|
||||
if (!isset($allowed[$extension]) || !in_array($mime, $allowed[$extension], true)) {
|
||||
throw new RuntimeException('不支持该文件格式，请上传PDF、Office、文本、ZIP、JPG/PNG或MP4');
|
||||
}
|
||||
}
|
||||
if ($name === '') {
|
||||
$name = '素材.' . $extension;
|
||||
}
|
||||
$this->ensureRoot();
|
||||
$assetId = bin2hex(random_bytes(24));
|
||||
$storageName = $assetId . '.' . $extension;
|
||||
$hash = hash_file('sha256', $file->getPathname());
|
||||
$file->move($this->root, $storageName);
|
||||
@chmod($this->root . DIRECTORY_SEPARATOR . $storageName, 0600);
|
||||
try {
|
||||
$this->store->insert([
|
||||
'asset_id' => $assetId, 'admin_id' => $adminId, 'name' => $name, 'type' => $type,
|
||||
'mime' => $mime, 'size' => $size, 'sha256' => $hash, 'storage_name' => $storageName,
|
||||
'media_id' => '', 'media_expires_at' => 0, 'credential_hash' => '',
|
||||
'last_error' => '', 'create_time' => time(), 'update_time' => time(),
|
||||
]);
|
||||
} catch (\Throwable $e) {
|
||||
@unlink($this->root . DIRECTORY_SEPARATOR . $storageName);
|
||||
throw new RuntimeException('素材入库失败，请确认已安装推广自动化数据表', 0, $e);
|
||||
}
|
||||
// 配置阶段就上传企微素材。失败保留私有文件供后续排障，不对外提供文件路径。
|
||||
$this->mediaId($assetId, $type, true);
|
||||
return ['asset_id' => $assetId, 'name' => $name, 'type' => $type];
|
||||
}
|
||||
|
||||
/** 旧方案授权由上层完成；只白名单旧配置实际已有资产，不接受请求单独声明的白名单。 */
|
||||
public function validateConfig(array $config, int $adminId, array $existingConfig = []): array
|
||||
{
|
||||
$allowed = self::assetIds($existingConfig);
|
||||
$config['welcome']['attachments'] = $this->validateAttachments((array) ($config['welcome']['attachments'] ?? []), $adminId, $allowed);
|
||||
foreach ((array) ($config['welcome_schedule'] ?? []) as $index => $slot) {
|
||||
$config['welcome_schedule'][$index]['attachments'] = $this->validateAttachments((array) ($slot['attachments'] ?? []), $adminId, $allowed);
|
||||
}
|
||||
return $config;
|
||||
}
|
||||
|
||||
public function validateAttachments(array $attachments, int $adminId, array $allowedAssetIds = []): array
|
||||
{
|
||||
if (count($attachments) > 9) {
|
||||
throw new RuntimeException('欢迎语最多9个附件');
|
||||
}
|
||||
$clean = [];
|
||||
foreach ($attachments as $attachment) {
|
||||
if (!is_array($attachment)) {
|
||||
throw new RuntimeException('附件格式不正确');
|
||||
}
|
||||
$type = (string) ($attachment['msgtype'] ?? '');
|
||||
$body = $attachment[$type] ?? null;
|
||||
if (!is_array($body)) {
|
||||
throw new RuntimeException('附件内容类型不匹配');
|
||||
}
|
||||
if (in_array($type, ['image', 'video', 'file'], true)) {
|
||||
// image.pic_url 限企微 uploadimg URL；本服务仅接受私有资产，避免伪装任意外部地址。
|
||||
$asset = $this->authorizedAsset((string) ($body['asset_id'] ?? ''), $type, $adminId, $allowedAssetIds);
|
||||
$body = ['asset_id' => $asset['asset_id']];
|
||||
} elseif ($type === 'link') {
|
||||
$body = [
|
||||
'title' => self::bytes($body['title'] ?? '', 128, '链接标题', true),
|
||||
'url' => self::url($body['url'] ?? ''),
|
||||
'desc' => self::bytes($body['desc'] ?? '', 512, '链接描述'),
|
||||
] + (!empty($body['picurl']) ? ['picurl' => self::url($body['picurl'])] : []);
|
||||
} elseif ($type === 'miniprogram') {
|
||||
$asset = $this->authorizedAsset((string) ($body['pic_asset_id'] ?? ''), 'image', $adminId, $allowedAssetIds);
|
||||
$appid = (string) ($body['appid'] ?? '');
|
||||
$page = self::bytes($body['page'] ?? '', 1024, '小程序页面', true);
|
||||
if (!preg_match('/^wx[0-9a-fA-F]{16}$/', $appid) || str_contains($page, '://')
|
||||
|| str_contains($page, '..') || preg_match('/[\x00-\x1f]/', $page)) {
|
||||
throw new RuntimeException('小程序 appid 或页面路径不正确');
|
||||
}
|
||||
$body = ['title' => self::bytes($body['title'] ?? '', 64, '小程序标题', true),
|
||||
'appid' => $appid, 'page' => $page, 'pic_asset_id' => $asset['asset_id']];
|
||||
} else {
|
||||
throw new RuntimeException('不支持的附件类型');
|
||||
}
|
||||
$clean[] = ['msgtype' => $type, $type => $body];
|
||||
}
|
||||
return $clean;
|
||||
}
|
||||
|
||||
/** 仅处理已授权并持久化的配置快照；绝不在欢迎语发送时进行网络文件上传。 */
|
||||
public function materialize(array $attachments, array $config): array
|
||||
{
|
||||
$attachments = $this->validateAttachments($attachments, 0, self::assetIds($config));
|
||||
foreach ($attachments as &$attachment) {
|
||||
$type = $attachment['msgtype'];
|
||||
if (in_array($type, ['image', 'video', 'file'], true)) {
|
||||
$attachment[$type] = ['media_id' => $this->mediaId($attachment[$type]['asset_id'], $type, false)];
|
||||
} elseif ($type === 'miniprogram') {
|
||||
$attachment[$type]['pic_media_id'] = $this->mediaId($attachment[$type]['pic_asset_id'], 'image', false);
|
||||
unset($attachment[$type]['pic_asset_id']);
|
||||
}
|
||||
}
|
||||
unset($attachment);
|
||||
return $attachments;
|
||||
}
|
||||
|
||||
public function refreshReferenced(int $limit = 100): array
|
||||
{
|
||||
$result = ['selected' => 0, 'refreshed' => 0, 'failed' => 0];
|
||||
foreach ($this->store->referencedAssetIds() as $id) {
|
||||
$asset = $this->store->find($id);
|
||||
if (!$asset || ($this->cacheValid($asset, 3600))) {
|
||||
continue;
|
||||
}
|
||||
if ($result['selected'] >= max(1, $limit)) {
|
||||
break;
|
||||
}
|
||||
++$result['selected'];
|
||||
try {
|
||||
$this->mediaId($id, $asset['type'], true, 3600);
|
||||
++$result['refreshed'];
|
||||
} catch (\Throwable) {
|
||||
++$result['failed'];
|
||||
}
|
||||
}
|
||||
return $result;
|
||||
}
|
||||
|
||||
public static function assetIds(array $config): array
|
||||
{
|
||||
$ids = [];
|
||||
$messages = array_merge([(array) ($config['welcome'] ?? [])], (array) ($config['welcome_schedule'] ?? []));
|
||||
foreach ($messages as $message) {
|
||||
foreach ((array) ($message['attachments'] ?? []) as $attachment) {
|
||||
$type = $attachment['msgtype'] ?? '';
|
||||
$key = $type === 'miniprogram' ? 'pic_asset_id' : 'asset_id';
|
||||
$id = (string) ($attachment[$type][$key] ?? '');
|
||||
if (preg_match('/^[0-9a-f]{48}$/', $id)) {
|
||||
$ids[] = $id;
|
||||
}
|
||||
}
|
||||
}
|
||||
return array_values(array_unique($ids));
|
||||
}
|
||||
|
||||
private function authorizedAsset(string $id, string $type, int $adminId, array $allowed): array
|
||||
{
|
||||
if (!preg_match('/^[0-9a-f]{48}$/', $id)) {
|
||||
throw new RuntimeException('请先上传欢迎语素材');
|
||||
}
|
||||
$asset = $this->store->find($id);
|
||||
if (!$asset || $asset['type'] !== $type || ((int) $asset['admin_id'] !== $adminId && !in_array($id, $allowed, true))) {
|
||||
throw new RuntimeException('素材不存在、类型不匹配或无权使用');
|
||||
}
|
||||
return $asset;
|
||||
}
|
||||
|
||||
private function mediaId(string $id, string $type, bool $allowUpload, int $margin = 300): string
|
||||
{
|
||||
$asset = $this->store->find($id);
|
||||
if (!$asset || $asset['type'] !== $type) {
|
||||
throw new RuntimeException('欢迎语素材不存在');
|
||||
}
|
||||
if ($this->cacheValid($asset, $margin)) {
|
||||
return (string) $asset['media_id'];
|
||||
}
|
||||
if (!$allowUpload) {
|
||||
throw new RuntimeException('欢迎语素材未预热或已过期，请检查素材刷新任务');
|
||||
}
|
||||
$stream = null;
|
||||
try {
|
||||
$path = $this->privatePath((string) $asset['storage_name']);
|
||||
if (!is_file($path) || hash_file('sha256', $path) !== $asset['sha256']) {
|
||||
throw new RuntimeException('欢迎语源文件缺失或完整性检查失败');
|
||||
}
|
||||
$stream = fopen($path, 'rb');
|
||||
$result = $this->api->uploadMedia($stream, $type, (string) $asset['name']);
|
||||
if (empty($result['media_id'])) {
|
||||
throw new RuntimeException('企微素材接口未返回 media_id');
|
||||
}
|
||||
$created = min(time(), (int) ($result['created_at'] ?? time()));
|
||||
$this->store->update($id, ['media_id' => (string) $result['media_id'],
|
||||
'media_expires_at' => $created + 3 * 86400, 'credential_hash' => $this->api->credentialFingerprint(),
|
||||
'last_error' => '', 'update_time' => time()]);
|
||||
return (string) $result['media_id'];
|
||||
} catch (\Throwable $e) {
|
||||
$this->store->update($id, ['last_error' => '素材预热失败[' . (int) $e->getCode() . ']', 'update_time' => time()]);
|
||||
throw $e;
|
||||
} finally {
|
||||
if (is_resource($stream)) {
|
||||
fclose($stream);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private function cacheValid(array $asset, int $margin): bool
|
||||
{
|
||||
return !empty($asset['media_id']) && (int) $asset['media_expires_at'] > time() + $margin
|
||||
&& hash_equals((string) $asset['credential_hash'], $this->api->credentialFingerprint());
|
||||
}
|
||||
|
||||
private function privatePath(string $name): string
|
||||
{
|
||||
if (!preg_match('/^[0-9a-f]{48}\.[a-z0-9]{1,8}$/', $name)) {
|
||||
throw new RuntimeException('素材存储标识不正确');
|
||||
}
|
||||
$root = realpath($this->root);
|
||||
$path = realpath($this->root . DIRECTORY_SEPARATOR . $name);
|
||||
if ($root === false || $path === false || !str_starts_with($path, $root . DIRECTORY_SEPARATOR)) {
|
||||
throw new RuntimeException('素材文件不在私有存储目录');
|
||||
}
|
||||
return $path;
|
||||
}
|
||||
|
||||
private function ensureRoot(): void
|
||||
{
|
||||
if (!is_dir($this->root) && !mkdir($this->root, 0700, true) && !is_dir($this->root)) {
|
||||
throw new RuntimeException('无法创建私有素材目录');
|
||||
}
|
||||
}
|
||||
|
||||
private static function bytes(mixed $value, int $limit, string $label, bool $required = false): string
|
||||
{
|
||||
if (!is_string($value) || strlen($value) > $limit || ($required && trim($value) === '')) {
|
||||
throw new RuntimeException($label . '须' . ($required ? '非空且' : '') . '不超过' . $limit . '字节');
|
||||
}
|
||||
return trim($value);
|
||||
}
|
||||
|
||||
private static function url(mixed $value): string
|
||||
{
|
||||
if (!is_string($value) || strlen($value) > 2048 || filter_var($value, FILTER_VALIDATE_URL) === false) {
|
||||
throw new RuntimeException('链接地址不正确');
|
||||
}
|
||||
$parts = parse_url($value);
|
||||
if (!in_array(strtolower((string) ($parts['scheme'] ?? '')), ['http', 'https'], true)
|
||||
|| isset($parts['user']) || isset($parts['pass'])) {
|
||||
throw new RuntimeException('链接仅支持不含账号密码的HTTP(S)地址');
|
||||
}
|
||||
// 仅向企微传递链接；服务端永远不会抓取这些URL。
|
||||
return $value;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use think\facade\Db;
|
||||
|
||||
/** 独立存储边界，测试可使用内存替身，禁止连接业务数据库。 */
|
||||
class QywxPromotionMediaStore
|
||||
{
|
||||
public function find(string $assetId): ?array
|
||||
{
|
||||
return Db::name('qywx_promotion_media')->where('asset_id', $assetId)->find() ?: null;
|
||||
}
|
||||
|
||||
public function insert(array $row): void
|
||||
{
|
||||
Db::name('qywx_promotion_media')->insert($row);
|
||||
}
|
||||
|
||||
public function update(string $assetId, array $fields): void
|
||||
{
|
||||
Db::name('qywx_promotion_media')->where('asset_id', $assetId)->update($fields);
|
||||
}
|
||||
|
||||
/** 只预热已保存方案引用的素材；未使用上传不永久续期。 */
|
||||
public function referencedAssetIds(): array
|
||||
{
|
||||
$ids = [];
|
||||
foreach (Db::name('qywx_promotion_config')->alias('cfg')
|
||||
->join('qywx_promotion_pool pool', 'pool.id = cfg.pool_id')
|
||||
->whereNull('pool.delete_time')->column('cfg.config_json') as $json) {
|
||||
$ids = array_merge($ids, QywxPromotionMediaService::assetIds(QywxPromotionConfig::decode($json)));
|
||||
}
|
||||
return array_values(array_unique($ids));
|
||||
}
|
||||
}
|
||||
@@ -11,9 +11,22 @@ class QywxPromotionMemberRange
|
||||
* @param list<array<string,mixed>> $members
|
||||
* @return array{userids:list<string>,members:list<array<string,mixed>>,eligible_count:int}
|
||||
*/
|
||||
public static function evaluate(array $members, string $today, int $now): array
|
||||
public static function evaluate(array $members, string $today, int $now, array $config = []): array
|
||||
{
|
||||
$userIds = [];
|
||||
$backups = array_fill_keys((array) ($config['backup_userids'] ?? []), true);
|
||||
$backupIds = [];
|
||||
$scheduled = ($config['reception_mode'] ?? 'always') === 'scheduled';
|
||||
$scheduledUsers = [];
|
||||
if ($scheduled) {
|
||||
foreach ((array) ($config['reception_schedule'] ?? []) as $slot) {
|
||||
if (QywxPromotionConfig::matches($slot, $now)) {
|
||||
foreach ((array) ($slot['member_userids'] ?? []) as $userId) {
|
||||
$scheduledUsers[$userId] = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
foreach ($members as &$member) {
|
||||
if ((string) ($member['today_date'] ?? '') !== $today) {
|
||||
$member['today_date'] = $today;
|
||||
@@ -25,15 +38,24 @@ class QywxPromotionMemberRange
|
||||
}
|
||||
$userId = trim((string) ($member['userid'] ?? ''));
|
||||
if ($userId !== '') {
|
||||
$userIds[$userId] = true;
|
||||
if (isset($backups[$userId])) {
|
||||
$backupIds[$userId] = true;
|
||||
} elseif (!$scheduled || isset($scheduledUsers[$userId])) {
|
||||
$userIds[$userId] = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
unset($member);
|
||||
|
||||
$usingBackup = $userIds === [] && $backupIds !== [];
|
||||
if ($usingBackup) {
|
||||
$userIds = $backupIds;
|
||||
}
|
||||
return [
|
||||
'userids' => array_keys($userIds),
|
||||
'members' => array_values($members),
|
||||
'eligible_count' => count($userIds),
|
||||
'using_backup' => $usingBackup,
|
||||
];
|
||||
}
|
||||
|
||||
|
||||
@@ -253,7 +253,7 @@ class QywxPromotionMemberSchedulerService
|
||||
string $today,
|
||||
int $now
|
||||
): array {
|
||||
$range = QywxPromotionMemberRange::evaluate($members, $today, $now);
|
||||
$range = QywxPromotionMemberRange::evaluate($members, $today, $now, QywxPromotionConfig::forPool($poolId));
|
||||
self::persistMemberCursors($range['members'], $now);
|
||||
if ($range['userids'] === []) {
|
||||
self::upsertSync($poolId, $linkId, false, $sync, $now, '所有成员均已禁用、未生效或达到今日上限');
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace app\common\service\qywx;
|
||||
|
||||
use app\common\service\DataScope\DataScopeService;
|
||||
use think\facade\Db;
|
||||
|
||||
/** 分流方案共享操作人产生的页面入口与专用数据范围。 */
|
||||
final class QywxPromotionOperatorAccess
|
||||
{
|
||||
public const PAGE_PERMISSION = 'firstvisit.wecomPromotion/overview';
|
||||
|
||||
public static function hasBasePagePermission(int $adminId, array $adminInfo = []): bool
|
||||
{
|
||||
if ((int) ($adminInfo['root'] ?? 0) === 1) {
|
||||
return true;
|
||||
}
|
||||
if ($adminId <= 0) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return Db::name('admin_role')->alias('ar')
|
||||
->join('system_role_menu rm', 'rm.role_id = ar.role_id')
|
||||
->join('system_menu m', 'm.id = rm.menu_id')
|
||||
->where('ar.admin_id', $adminId)
|
||||
->where('m.perms', self::PAGE_PERMISSION)
|
||||
->where('m.is_disable', 0)
|
||||
->count() > 0;
|
||||
}
|
||||
|
||||
public static function hasSharedPagePermission(int $adminId): bool
|
||||
{
|
||||
if ($adminId <= 0 || !self::pageMenuEnabled()) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
return Db::name('qywx_promotion_pool_operator')->alias('po')
|
||||
->join('qywx_promotion_pool p', 'p.id = po.pool_id')
|
||||
->where('po.admin_id', $adminId)
|
||||
->whereNull('po.delete_time')
|
||||
->whereNull('p.delete_time')
|
||||
->count() > 0;
|
||||
} catch (\Throwable $error) {
|
||||
if (self::isMissingTable($error)) {
|
||||
return false;
|
||||
}
|
||||
throw $error;
|
||||
}
|
||||
}
|
||||
|
||||
public static function hasPagePermission(int $adminId, array $adminInfo = []): bool
|
||||
{
|
||||
return self::hasBasePagePermission($adminId, $adminInfo)
|
||||
|| self::hasSharedPagePermission($adminId);
|
||||
}
|
||||
|
||||
/** 基础页面权限沿用角色数据范围；纯共享账号只能通过 operator pool 范围访问。 */
|
||||
public static function visibleAdminIds(int $adminId, array $adminInfo): ?array
|
||||
{
|
||||
return self::hasBasePagePermission($adminId, $adminInfo)
|
||||
? DataScopeService::getVisibleAdminIds($adminId, $adminInfo)
|
||||
: [];
|
||||
}
|
||||
|
||||
/** @return list<int> */
|
||||
public static function activePoolIds(int $adminId): array
|
||||
{
|
||||
if ($adminId <= 0) {
|
||||
return [];
|
||||
}
|
||||
try {
|
||||
$ids = Db::name('qywx_promotion_pool_operator')->alias('po')
|
||||
->join('qywx_promotion_pool p', 'p.id = po.pool_id')
|
||||
->where('po.admin_id', $adminId)
|
||||
->whereNull('po.delete_time')
|
||||
->whereNull('p.delete_time')
|
||||
->column('po.pool_id');
|
||||
} catch (\Throwable $error) {
|
||||
if (self::isMissingTable($error)) {
|
||||
return [];
|
||||
}
|
||||
throw $error;
|
||||
}
|
||||
|
||||
return array_values(array_unique(array_filter(array_map(
|
||||
static fn ($value): int => (int) $value,
|
||||
$ids
|
||||
), static fn (int $value): bool => $value > 0)));
|
||||
}
|
||||
|
||||
private static function pageMenuEnabled(): bool
|
||||
{
|
||||
return Db::name('system_menu')
|
||||
->where('perms', self::PAGE_PERMISSION)
|
||||
->where('is_disable', 0)
|
||||
->count() > 0;
|
||||
}
|
||||
|
||||
private static function isMissingTable(\Throwable $error): bool
|
||||
{
|
||||
$message = strtolower($error->getMessage());
|
||||
|
||||
return str_contains($message, '42s02')
|
||||
|| str_contains($message, '1146')
|
||||
|| str_contains($message, 'no such table');
|
||||
}
|
||||
}
|
||||
@@ -66,7 +66,7 @@ class QywxPromotionRangeSyncService
|
||||
if ($remoteLinkId === '') {
|
||||
throw new RuntimeException('官方链接 ID 为空');
|
||||
}
|
||||
$range = QywxPromotionMemberRange::evaluate($members, date('Y-m-d'), time());
|
||||
$range = QywxPromotionMemberRange::evaluate($members, date('Y-m-d'), time(), QywxPromotionConfig::forPool($poolId));
|
||||
$desiredUserIds = $range['userids'];
|
||||
if ($desiredUserIds === []) {
|
||||
$message = '所有成员均已禁用、未生效或达到今日上限；企业微信官方链接至少需要保留一名成员';
|
||||
@@ -93,7 +93,7 @@ class QywxPromotionRangeSyncService
|
||||
$this->api->updateLink([
|
||||
'link_id' => $remoteLinkId,
|
||||
'link_name' => mb_substr((string) ($pool['name'] ?? '获客分流方案'), 0, 30),
|
||||
'range' => ['user_list' => $desiredUserIds],
|
||||
'range' => ['user_list' => $desiredUserIds, 'department_list' => []],
|
||||
'skip_verify' => (int) ($link['skip_verify'] ?? 0) === 1,
|
||||
]);
|
||||
$response = $this->api->getLink($remoteLinkId);
|
||||
|
||||
@@ -116,13 +116,19 @@ class Qcloud extends Server
|
||||
|
||||
/**
|
||||
* @notes 获取 STS 临时凭证（用于浏览器直传）
|
||||
* @param string $keyPrefix 资源前缀，如 uploads/video/20260508/
|
||||
* @param string $keyScope 资源前缀或完整对象 Key
|
||||
* @param int $maxSizeBytes 单文件大小上限（字节）
|
||||
* @param int $durationSeconds 凭证有效期（秒）
|
||||
* @param bool $exactObject 是否只授权单个对象 Key
|
||||
* @return array {credentials, expiredTime, requestId}
|
||||
* @throws Exception
|
||||
*/
|
||||
public function getStsCredentials(string $keyPrefix, int $maxSizeBytes, int $durationSeconds = 1800): array
|
||||
public function getStsCredentials(
|
||||
string $keyScope,
|
||||
int $maxSizeBytes,
|
||||
int $durationSeconds = 1800,
|
||||
bool $exactObject = false
|
||||
): array
|
||||
{
|
||||
$bucket = $this->config['bucket'];
|
||||
// bucket 形如 likeadmin-1300000000，appId 即末段
|
||||
@@ -137,15 +143,19 @@ class Qcloud extends Server
|
||||
|
||||
$shortBucket = substr($bucket, 0, strrpos($bucket, '-'));
|
||||
$region = $this->config['region'];
|
||||
$prefix = ltrim($keyPrefix, '/');
|
||||
if ($prefix === '' || substr($prefix, -1) !== '/') {
|
||||
$prefix = $prefix . '/';
|
||||
$scope = ltrim($keyScope, '/');
|
||||
if ($scope === '') {
|
||||
throw new Exception('COS 授权对象不能为空');
|
||||
}
|
||||
if (!$exactObject && substr($scope, -1) !== '/') {
|
||||
$scope .= '/';
|
||||
}
|
||||
|
||||
$duration = max(900, min($durationSeconds, 7200));
|
||||
|
||||
// 自行构造 policy：对象级写动作收紧 + bucket 级 ListMultipartUploads（cos-js-sdk-v5 续传探测必需）
|
||||
$objectArn = sprintf('qcs::cos:%s:uid/%s:%s/%s*', $region, $appId, $bucket, $prefix);
|
||||
$objectResource = $exactObject ? $scope : $scope . '*';
|
||||
$objectArn = sprintf('qcs::cos:%s:uid/%s:%s/%s', $region, $appId, $bucket, $objectResource);
|
||||
$bucketArn = sprintf('qcs::cos:%s:uid/%s:%s/*', $region, $appId, $bucket);
|
||||
|
||||
$policy = [
|
||||
|
||||
Reference in New Issue
Block a user