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

更新

Browse Source
This commit is contained in:
Your Name
2026-04-21 09:49:44 +08:00
parent c9e0419895
commit c15c348b7a
454 changed files with 5391 additions and 25651 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/app/common/service/wechat/QywxMsgArchiveService.php
+516
View File
@@ -0,0 +1,516 @@
<?php
declare(strict_types=1);
namespace app\common\service\wechat;
use app\common\model\QywxMsgArchive;
use app\common\model\QywxMsgArchiveCursor;
use app\common\model\QywxMsgArchiveMedia;
use app\common\model\QywxMsgSession;
use think\facade\Db;
use think\facade\Log;
/**
* 会话存档拉取与落库
*
* 工作流程：
* 1) 读取游标 seq
* 2) 调用 WeComFinanceSdkClient->pullChatData(seq)
* 3) 遍历解密后的消息，按 msgtype 归一化字段后 UPSERT qywx_msg_archive
* 4) 同步刷新 qywx_msg_session 列表摘要
* 5) 媒体消息登记 qywx_msg_archive_media（待下载）
* 6) 更新 cursor
*
* 媒体下载由 downloadPendingMedia() 独立跑，避免拉取慢。
*/
class QywxMsgArchiveService
{
private const MEDIA_TYPES = ['image', 'voice', 'video', 'file', 'emotion'];
/**
* 拉一次，直到本批为空或达到 maxBatches
*
* @return array{
* enabled: bool,
* batches: int,
* pulled: int,
* inserted: int,
* updated: int,
* media_pending: int,
* last_seq: int,
* errors: string[]
* }
*/
public static function pullLoop(int $maxBatches = 20, string $cursorKey = QywxMsgArchiveCursor::DEFAULT_KEY): array
{
$result = [
'enabled' => true,
'batches' => 0,
'pulled' => 0,
'inserted' => 0,
'updated' => 0,
'media_pending' => 0,
'last_seq' => 0,
'errors' => [],
];
$sdk = new WeComFinanceSdkClient();
if (!$sdk->isAvailable()) {
$result['enabled'] = false;
$result['errors'][] = $sdk->getLastError();
return $result;
}
$seq = self::getCursorSeq($cursorKey);
$result['last_seq'] = $seq;
for ($i = 0; $i < $maxBatches; $i++) {
$batch = $sdk->pullChatData($seq);
$result['batches']++;
if (!$batch['ok']) {
$result['errors'][] = $batch['error'];
self::touchCursor($cursorKey, $seq, 0, $batch['error']);
break;
}
$messages = $batch['messages'];
$count = count($messages);
if ($count === 0) {
self::touchCursor($cursorKey, $seq, 0, '');
break;
}
[$inserted, $updated, $mediaPending] = self::persistBatch($messages);
$result['pulled'] += $count;
$result['inserted'] += $inserted;
$result['updated'] += $updated;
$result['media_pending'] += $mediaPending;
$seq = (int) $batch['next_seq'];
$result['last_seq'] = $seq;
self::touchCursor($cursorKey, $seq, $count, '');
if ($count < $sdk->getLimit()) {
// 本批不足 limit，说明暂无更多，结束本轮
break;
}
}
return $result;
}
/**
* 下载 pending 的媒体文件
*
* @return array{enabled: bool, ok: int, failed: int, skipped: int, errors: string[]}
*/
public static function downloadPendingMedia(int $max = 200): array
{
$result = ['enabled' => true, 'ok' => 0, 'failed' => 0, 'skipped' => 0, 'errors' => []];
$sdk = new WeComFinanceSdkClient();
if (!$sdk->isAvailable()) {
$result['enabled'] = false;
$result['errors'][] = $sdk->getLastError();
return $result;
}
$rows = QywxMsgArchiveMedia::where('status', QywxMsgArchiveMedia::STATUS_PENDING)
->where('retry_count', '<', 5)
->order('id', 'asc')
->limit($max)
->select()
->toArray();
foreach ($rows as $row) {
$sdkFileid = (string) $row['sdkfileid'];
if ($sdkFileid === '') {
$result['skipped']++;
continue;
}
$relPath = self::buildMediaRelPath((string) $row['msgid'], $sdkFileid);
$absPath = self::absoluteMediaPath($relPath);
$ret = $sdk->downloadMedia($sdkFileid, $absPath);
$now = time();
if ($ret['ok']) {
QywxMsgArchiveMedia::where('id', $row['id'])->update([
'status' => QywxMsgArchiveMedia::STATUS_SUCCESS,
'file_path' => $relPath,
'file_size' => $ret['size'],
'error' => '',
'update_time' => $now,
]);
$result['ok']++;
} else {
QywxMsgArchiveMedia::where('id', $row['id'])->update([
'retry_count' => (int) $row['retry_count'] + 1,
'status' => ((int) $row['retry_count'] + 1) >= 5 ? QywxMsgArchiveMedia::STATUS_FAILED : QywxMsgArchiveMedia::STATUS_PENDING,
'error' => mb_substr($ret['error'], 0, 500),
'update_time' => $now,
]);
$result['failed']++;
$result['errors'][] = sprintf('msgid=%s err=%s', (string) $row['msgid'], $ret['error']);
}
}
return $result;
}
/**
* @param array<int, array<string, mixed>> $messages
*
* @return array{0:int,1:int,2:int} [inserted, updated, mediaPending]
*/
private static function persistBatch(array $messages): array
{
$inserted = 0;
$updated = 0;
$mediaPending = 0;
$now = time();
foreach ($messages as $msg) {
$norm = self::normalize($msg);
if ($norm === null) {
continue;
}
try {
$isUpdate = false;
Db::transaction(function () use ($norm, $now, &$inserted, &$updated, &$mediaPending, &$isUpdate) {
$exists = QywxMsgArchive::where('msgid', $norm['msgid'])->findOrEmpty();
if ($exists->isEmpty()) {
QywxMsgArchive::create(array_merge($norm, [
'create_time' => $now,
'update_time' => $now,
]));
$inserted++;
} else {
// recall/switch 会重发相同 msgid，更新 action / content
QywxMsgArchive::where('msgid', $norm['msgid'])->update(array_merge($norm, [
'update_time' => $now,
]));
$updated++;
$isUpdate = true;
}
if ($norm['media_id'] !== '' && in_array($norm['msgtype'], self::MEDIA_TYPES, true)) {
$existsMedia = QywxMsgArchiveMedia::where('msgid', $norm['msgid'])
->where('sdkfileid', $norm['media_id'])
->findOrEmpty();
if ($existsMedia->isEmpty()) {
QywxMsgArchiveMedia::create([
'msgid' => $norm['msgid'],
'sdkfileid' => $norm['media_id'],
'status' => QywxMsgArchiveMedia::STATUS_PENDING,
'md5sum' => $norm['md5sum'],
'create_time' => $now,
'update_time' => $now,
]);
$mediaPending++;
}
}
self::upsertSession($norm, $now);
});
} catch (\Throwable $e) {
Log::error(sprintf(
'会话存档-消息落库异常 msgid=%s err=%s',
(string) ($norm['msgid'] ?? '-'),
$e->getMessage()
));
}
}
return [$inserted, $updated, $mediaPending];
}
/**
* 把解密后的原始消息归一化为数据库字段
*
* @param array<string, mixed> $msg
*
* @return array<string, mixed>|null
*/
public static function normalize(array $msg): ?array
{
$msgid = (string) ($msg['msgid'] ?? $msg['__msgid_sdk'] ?? '');
if ($msgid === '') {
return null;
}
$action = (string) ($msg['action'] ?? 'send'); // send / recall / switch
$msgtype = (string) ($msg['msgtype'] ?? '');
$from = (string) ($msg['from'] ?? '');
$roomid = (string) ($msg['roomid'] ?? '');
$tolist = isset($msg['tolist']) && is_array($msg['tolist']) ? $msg['tolist'] : [];
$sendTime = isset($msg['msgtime']) ? (int) (((int) $msg['msgtime']) / 1000) : 0;
if ($sendTime === 0 && isset($msg['msgtime'])) {
// 兼容已经是秒的情况
$sendTime = (int) $msg['msgtime'];
}
$content = '';
$mediaId = '';
$md5sum = '';
$fileName = '';
$fileSize = 0;
$fileExt = '';
$playLength = 0;
switch ($msgtype) {
case 'text':
$content = (string) ($msg['text']['content'] ?? '');
break;
case 'markdown':
$content = (string) ($msg['markdown']['content'] ?? '');
break;
case 'image':
$mediaId = (string) ($msg['image']['sdkfileid'] ?? '');
$md5sum = (string) ($msg['image']['md5sum'] ?? '');
$fileSize = (int) ($msg['image']['filesize'] ?? 0);
$content = '[图片]';
break;
case 'voice':
$mediaId = (string) ($msg['voice']['sdkfileid'] ?? '');
$md5sum = (string) ($msg['voice']['md5sum'] ?? '');
$fileSize = (int) ($msg['voice']['voice_size'] ?? 0);
$playLength = (int) ($msg['voice']['play_length'] ?? 0);
$content = sprintf('[语音 %ds]', $playLength);
break;
case 'video':
$mediaId = (string) ($msg['video']['sdkfileid'] ?? '');
$md5sum = (string) ($msg['video']['md5sum'] ?? '');
$fileSize = (int) ($msg['video']['filesize'] ?? 0);
$playLength = (int) ($msg['video']['play_length'] ?? 0);
$content = '[视频]';
break;
case 'emotion':
$mediaId = (string) ($msg['emotion']['sdkfileid'] ?? '');
$md5sum = (string) ($msg['emotion']['md5sum'] ?? '');
$fileSize = (int) ($msg['emotion']['imagesize'] ?? 0);
$content = '[表情]';
break;
case 'file':
$mediaId = (string) ($msg['file']['sdkfileid'] ?? '');
$md5sum = (string) ($msg['file']['md5sum'] ?? '');
$fileName = (string) ($msg['file']['filename'] ?? '');
$fileExt = (string) ($msg['file']['fileext'] ?? '');
$fileSize = (int) ($msg['file']['filesize'] ?? 0);
$content = '[文件] ' . $fileName;
break;
case 'link':
$content = (string) ($msg['link']['title'] ?? '[链接]');
break;
case 'weapp':
$content = '[小程序] ' . (string) ($msg['weapp']['title'] ?? '');
break;
case 'chatrecord':
$content = '[聊天记录]';
break;
case 'location':
$content = '[位置] ' . (string) ($msg['location']['address'] ?? '');
break;
case 'card':
$content = '[名片]';
break;
case 'revoke':
$action = 'recall';
$content = '[撤回] ' . (string) ($msg['revoke']['pre_msgid'] ?? '');
break;
case 'mixed':
$content = self::summarizeMixed($msg['mixed'] ?? null);
break;
case 'meeting':
$content = '[会议] ' . (string) ($msg['meeting']['topic'] ?? '');
break;
case 'docmsg':
$content = '[文档] ' . (string) ($msg['docmsg']['title'] ?? '');
break;
default:
$content = '[' . $msgtype . ']';
}
return [
'msgid' => $msgid,
'seq' => (int) ($msg['__seq'] ?? 0),
'action' => $action,
'from_user' => $from,
'from_type' => 0, // 外层 session upsert 时再判定
'to_list' => $tolist,
'roomid' => $roomid,
'msgtype' => $msgtype,
'content' => mb_substr($content, 0, 16000),
'media_id' => $mediaId,
'md5sum' => $md5sum,
'file_name' => mb_substr($fileName, 0, 500),
'file_size' => $fileSize,
'file_ext' => $fileExt,
'play_length' => $playLength,
'raw' => json_encode($msg, JSON_UNESCAPED_UNICODE),
'send_time' => $sendTime,
];
}
/**
* @param mixed $mixed
*/
private static function summarizeMixed($mixed): string
{
if (!is_array($mixed) || !isset($mixed['item']) || !is_array($mixed['item'])) {
return '[混合消息]';
}
$parts = [];
foreach ($mixed['item'] as $el) {
if (!is_array($el)) {
continue;
}
$t = (string) ($el['type'] ?? '');
if ($t === 'text') {
$parts[] = (string) ($el['content'] ?? '');
} else {
$parts[] = '[' . $t . ']';
}
if (count($parts) >= 3) {
break;
}
}
return implode(' ', $parts);
}
private static function upsertSession(array $norm, int $now): void
{
// 外部联系人 id 的企微前缀通常是 "wm" 或 "wo"；员工 userid 无此前缀。
// 单聊时 tolist 有 1 个对端；群聊走 roomid。
$tolist = $norm['to_list'] ?? [];
if ($norm['roomid'] !== '') {
$sessionType = 2;
$staff = self::isStaffUserid($norm['from_user']) ? $norm['from_user'] : '';
if ($staff === '') {
foreach ($tolist as $to) {
if (self::isStaffUserid($to)) {
$staff = (string) $to;
break;
}
}
}
$external = '';
} else {
$sessionType = 1;
if (self::isStaffUserid($norm['from_user'])) {
$staff = $norm['from_user'];
$external = (string) ($tolist[0] ?? '');
} else {
$external = $norm['from_user'];
$staff = (string) ($tolist[0] ?? '');
}
}
if ($staff === '' && $norm['roomid'] === '') {
// 兜底：两端都辨不出，跳过 session 更新
return;
}
$where = [
'staff_userid' => $staff,
'external_userid' => $external,
'roomid' => $norm['roomid'],
];
$summary = mb_substr((string) $norm['content'], 0, 500);
$data = [
'session_type' => $sessionType,
'last_msg_id' => $norm['msgid'],
'last_msg_seq' => $norm['seq'],
'last_msg_time' => $norm['send_time'],
'last_msg_type' => $norm['msgtype'],
'last_msg_summary' => $summary,
'update_time' => $now,
];
$existing = QywxMsgSession::where($where)->findOrEmpty();
if ($existing->isEmpty()) {
QywxMsgSession::create(array_merge($where, $data, [
'unread_staff' => self::isStaffUserid($norm['from_user']) ? 0 : 1,
'create_time' => $now,
]));
} else {
// 仅当本条更新，才覆盖最近消息字段
if ((int) $existing['last_msg_time'] <= (int) $norm['send_time']) {
if (!self::isStaffUserid($norm['from_user'])) {
$data['unread_staff'] = (int) $existing['unread_staff'] + 1;
}
QywxMsgSession::where('id', $existing['id'])->update($data);
}
}
}
/**
* 粗判 userid 是员工还是外部联系人：企微外部联系人 id 以 "wm"/"wo" 开头且长度较长；
* 员工 userid 是管理员配置的，通常是字母/数字无这两个前缀。
*/
private static function isStaffUserid(string $id): bool
{
if ($id === '') {
return false;
}
$lower = strtolower($id);
if (strpos($lower, 'wm') === 0 || strpos($lower, 'wo') === 0) {
// 仍需长度判断：员工 userid 以 wm 开头的极少见；外部联系人通常 20+ 字符
return strlen($id) < 20;
}
return true;
}
private static function buildMediaRelPath(string $msgid, string $sdkFileid): string
{
$hash = substr(md5($msgid . ':' . $sdkFileid), 0, 16);
return 'qywx_archive/' . date('Ymd') . '/' . $hash;
}
private static function absoluteMediaPath(string $relPath): string
{
$base = rtrim((string) config('filesystem.disks.public.root', public_path() . 'storage'), DIRECTORY_SEPARATOR);
return $base . DIRECTORY_SEPARATOR . str_replace('/', DIRECTORY_SEPARATOR, $relPath);
}
private static function getCursorSeq(string $cursorKey): int
{
$row = QywxMsgArchiveCursor::where('cursor_key', $cursorKey)->findOrEmpty();
return $row->isEmpty() ? 0 : (int) $row['seq'];
}
private static function touchCursor(string $cursorKey, int $seq, int $pullCount, string $error): void
{
$now = time();
$row = QywxMsgArchiveCursor::where('cursor_key', $cursorKey)->findOrEmpty();
if ($row->isEmpty()) {
QywxMsgArchiveCursor::create([
'cursor_key' => $cursorKey,
'seq' => $seq,
'last_pull_time' => $now,
'last_pull_count' => $pullCount,
'last_error' => mb_substr($error, 0, 500),
'create_time' => $now,
'update_time' => $now,
]);
} else {
QywxMsgArchiveCursor::where('id', $row['id'])->update([
'seq' => $seq,
'last_pull_time' => $now,
'last_pull_count' => $pullCount,
'last_error' => mb_substr($error, 0, 500),
'update_time' => $now,
]);
}
}
}