517 lines
18 KiB
PHP
517 lines
18 KiB
PHP
<?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,
|
|
]);
|
|
}
|
|
}
|
|
}
|