feat: 恢复离线期间的 QQ 消息记录到 TG（不稳定功能）

src/constants/commands.ts

@@ -1,32 +1,32 @@
-import { Api } from "telegram";
+import { Api } from 'telegram';
 
 const preSetupCommands = [
   new Api.BotCommand({
-    command: "setup",
-    description: "执行初始化配置",
+    command: 'setup',
+    description: '执行初始化配置',
   }),
 ];
 
 // 这里的 group 指群组模式，Private 指在与机器人的私聊会话中
 const groupPrivateCommands = [
   new Api.BotCommand({
-    command: "add",
-    description: "添加新的群转发",
+    command: 'add',
+    description: '添加新的群转发',
   }),
 ];
 
 const personalPrivateCommands = [
   new Api.BotCommand({
-    command: "addfriend",
-    description: "添加新的好友转发",
+    command: 'addfriend',
+    description: '添加新的好友转发',
   }),
   new Api.BotCommand({
-    command: "addgroup",
-    description: "添加新的群转发",
+    command: 'addgroup',
+    description: '添加新的群转发',
   }),
   new Api.BotCommand({
-    command: "login",
-    description: "当 QQ 处于下线状态时，使用此命令重新登录 QQ",
+    command: 'login',
+    description: '当 QQ 处于下线状态时，使用此命令重新登录 QQ',
   }),
 ];
 
@@ -34,56 +34,60 @@ const personalPrivateCommands = [
 const groupPrivateSuperAdminCommands = [
   ...groupPrivateCommands,
   new Api.BotCommand({
-    command: "newinstance",
-    description: "创建一个新的转发机器人实例",
+    command: 'newinstance',
+    description: '创建一个新的转发机器人实例',
   }),
 ];
 
 const personalPrivateSuperAdminCommands = [
   ...personalPrivateCommands,
   new Api.BotCommand({
-    command: "newinstance",
-    description: "创建一个新的转发机器人实例",
+    command: 'newinstance',
+    description: '创建一个新的转发机器人实例',
   }),
 ];
 
 // inChat 表示在关联了的转发群组中的命令
 const inChatCommands = [
   new Api.BotCommand({
-    command: "info",
-    description: "查看本群或选定消息的详情",
+    command: 'info',
+    description: '查看本群或选定消息的详情',
   }),
 ];
 
 const groupInChatCommands = [
   ...inChatCommands,
   new Api.BotCommand({
-    command: "forwardoff",
-    description: "暂停消息转发",
+    command: 'forwardoff',
+    description: '暂停消息转发',
   }),
   new Api.BotCommand({
-    command: "forwardon",
-    description: "恢复消息转发",
+    command: 'forwardon',
+    description: '恢复消息转发',
+  }),
+  new Api.BotCommand({ command: 'disable_qq_forward', description: '停止从QQ转发至TG' }),
+  new Api.BotCommand({ command: 'enable_qq_forward', description: '恢复从QQ转发至TG' }),
+  new Api.BotCommand({ command: 'disable_tg_forward', description: '停止从TG转发至QQ' }),
+  new Api.BotCommand({ command: 'enable_tg_forward', description: '恢复从TG转发至QQ' }),
+  new Api.BotCommand({
+    command: 'recover',
+    description: '恢复离线期间的 QQ 消息记录到 TG（不稳定功能，管理员专用）',
   }),
-  new Api.BotCommand({ command: "disable_qq_forward", description: "停止从QQ转发至TG" }),
-  new Api.BotCommand({ command: "enable_qq_forward", description: "恢复从QQ转发至TG" }),
-  new Api.BotCommand({ command: "disable_tg_forward", description: "停止从TG转发至QQ" }),
-  new Api.BotCommand({ command: "enable_tg_forward", description: "恢复从TG转发至QQ" }),
 ];
 
 const personalInChatCommands = [
   ...inChatCommands,
   new Api.BotCommand({
-    command: "refresh",
-    description: "刷新头像和简介",
+    command: 'refresh',
+    description: '刷新头像和简介',
   }),
   new Api.BotCommand({
-    command: "poke",
-    description: "戳一戳",
+    command: 'poke',
+    description: '戳一戳',
   }),
   new Api.BotCommand({
-    command: "nick",
-    description: "获取/设置群名片",
+    command: 'nick',
+    description: '获取/设置群名片',
   }),
 ];
 

src/controllers/InChatCommandsController.ts

@@ -1,10 +1,11 @@
-import InChatCommandsService from "../services/InChatCommandsService";
-import { getLogger, Logger } from "log4js";
-import Instance from "../models/Instance";
-import Telegram from "../client/Telegram";
-import OicqClient from "../client/OicqClient";
-import { Api } from "telegram";
-import { Group } from "oicq";
+import InChatCommandsService from '../services/InChatCommandsService';
+import { getLogger, Logger } from 'log4js';
+import Instance from '../models/Instance';
+import Telegram from '../client/Telegram';
+import OicqClient from '../client/OicqClient';
+import { Api } from 'telegram';
+import { Group } from 'oicq';
+import RecoverMessageHelper from '../helpers/RecoverMessageHelper';
 
 export default class InChatCommandsController {
   private readonly service: InChatCommandsService;
@@ -13,7 +14,8 @@ export default class InChatCommandsController {
   constructor(
     private readonly instance: Instance,
     private readonly tgBot: Telegram,
-    private readonly oicq: OicqClient
+    private readonly tgUser: Telegram,
+    private readonly oicq: OicqClient,
   ) {
     this.log = getLogger(`InChatCommandsController - ${instance.id}`);
     this.service = new InChatCommandsService(instance, tgBot, oicq);
@@ -22,55 +24,55 @@ export default class InChatCommandsController {
 
   private onTelegramMessage = async (message: Api.Message) => {
     if (!message.message) return;
-    const messageParts = message.message.split(" ");
-    if (!messageParts.length || !messageParts[0].startsWith("/")) return;
+    const messageParts = message.message.split(' ');
+    if (!messageParts.length || !messageParts[0].startsWith('/')) return;
     let command: string = messageParts.shift();
-    const params = messageParts.join(" ");
-    if (command.includes("@")) {
+    const params = messageParts.join(' ');
+    if (command.includes('@')) {
       let target: string;
-      [command, target] = command.split("@");
+      [command, target] = command.split('@');
       if (target !== this.tgBot.me.username) return false;
     }
     const pair = this.instance.forwardPairs.find(message.chat);
     if (!pair) return false;
     switch (command) {
-      case "/info":
+      case '/info':
         await this.service.info(message, pair);
         return true;
-      case "/poke":
+      case '/poke':
         await this.service.poke(message, pair);
         return true;
-      case "/forwardoff":
+      case '/forwardoff':
         pair.enable = false;
-        await message.reply({ message: "转发已禁用" });
+        await message.reply({ message: '转发已禁用' });
         return true;
-      case "/forwardon":
+      case '/forwardon':
         pair.enable = true;
-        await message.reply({ message: "转发已启用" });
+        await message.reply({ message: '转发已启用' });
         return true;
-      case "/disable_qq_forward":
+      case '/disable_qq_forward':
         pair.disableQ2TG = true;
-        await message.reply({ message: "QQ->TG已禁用" });
+        await message.reply({ message: 'QQ->TG已禁用' });
         return true;
-      case "/enable_qq_forward":
+      case '/enable_qq_forward':
         pair.disableQ2TG = false;
-        await message.reply({ message: "QQ->TG已启用" });
+        await message.reply({ message: 'QQ->TG已启用' });
         return true;
-      case "/disable_tg_forward":
+      case '/disable_tg_forward':
         pair.disableTG2Q = true;
-        await message.reply({ message: "TG->QQ已禁用" });
+        await message.reply({ message: 'TG->QQ已禁用' });
         return true;
-      case "/enable_tg_forward":
+      case '/enable_tg_forward':
         pair.disableTG2Q = false;
-        await message.reply({ message: "TG->QQ已启用" });
+        await message.reply({ message: 'TG->QQ已启用' });
         return true;
-      case "/refresh":
-        if (this.instance.workMode !== "personal" || !message.senderId?.eq(this.instance.owner)) return false;
+      case '/refresh':
+        if (this.instance.workMode !== 'personal' || !message.senderId?.eq(this.instance.owner)) return false;
         await pair.updateInfo();
-        await message.reply({ message: "<i>刷新成功</i>" });
+        await message.reply({ message: '<i>刷新成功</i>' });
         return true;
-      case "/nick":
-        if (this.instance.workMode !== "personal" || !message.senderId?.eq(this.instance.owner)) return false;
+      case '/nick':
+        if (this.instance.workMode !== 'personal' || !message.senderId?.eq(this.instance.owner)) return false;
         if (!(pair.qq instanceof Group)) return;
         if (!params) {
           await message.reply({
@@ -80,9 +82,14 @@ export default class InChatCommandsController {
         }
         const result = await pair.qq.setCard(this.instance.qqUin, params);
         await message.reply({
-          message: "设置" + (result ? "成功" : "失败"),
+          message: '设置' + (result ? '成功' : '失败'),
         });
         return true;
+      case '/recover':
+        if (!message.senderId.eq(this.instance.owner)) return true;
+        const helper = new RecoverMessageHelper(this.instance, this.tgBot, this.tgUser, this.oicq, pair, message);
+        helper.startRecover().then(() => this.log.info('恢复完成'));
+        return true;
     }
   };
 }

src/helpers/RecoverMessageHelper.ts

@@ -0,0 +1,365 @@
+import Instance from '../models/Instance';
+import Telegram from '../client/Telegram';
+import OicqClient from '../client/OicqClient';
+import { Pair } from '../models/Pair';
+import { Api } from 'telegram';
+import { GroupMessage, PrivateMessage } from 'oicq';
+import db from '../models/db';
+import { format } from 'date-and-time';
+import lottie from '../constants/lottie';
+import helper from './forwardHelper';
+import convert from './convert';
+import { fetchFile, getBigFaceUrl, getImageUrlByMd5 } from '../utils/urls';
+import { getLogger, Logger } from 'log4js';
+import path from 'path';
+import exts from '../constants/exts';
+import silk from '../encoding/silk';
+import { md5Hex } from '../utils/hashing';
+import axios from 'axios';
+import { CustomFile } from 'telegram/client/uploads';
+import fsP from 'fs/promises';
+import { file } from 'tmp-promise';
+
+export default class {
+  private readonly log: Logger;
+
+  constructor(private readonly instance: Instance,
+              private readonly tgBot: Telegram,
+              private readonly tgUser: Telegram,
+              private readonly oicq: OicqClient,
+              private readonly pair: Pair,
+              private readonly requestMessage: Api.Message) {
+    this.log = getLogger(`MessageRecoverSession - ${instance.id} ${pair.qqRoomId}`);
+  }
+
+  private statusMessage: Api.Message;
+  private historyMessages = [] as (PrivateMessage | GroupMessage)[];
+  private currentStatus = 'getMessage' as
+    'getMessage' | 'getMedia' | 'uploadMessage' | 'uploadMedia' | 'finishing' | 'done';
+  private importTxt = '';
+  // id to path
+  private filesMap = {} as { [p: string]: string };
+  private mediaUploadedCount = 0;
+
+  public async startRecover() {
+    await this.updateStatusMessage();
+    await this.getMessages();
+    this.currentStatus = 'getMedia';
+    await this.messagesToTxt();
+    console.log(this.importTxt, this.filesMap);
+    this.currentStatus = 'uploadMessage';
+    await this.updateStatusMessage();
+    await this.importMessagesAndMedia();
+    this.currentStatus = 'done';
+    await this.updateStatusMessage();
+  }
+
+  private async getMessages() {
+    let timeOrSeq = undefined as number;
+    while (true) {
+      const messages = await this.pair.qq.getChatHistory(timeOrSeq);
+      if (!messages.length) return;
+      let messagesAllExist = true;
+      timeOrSeq = messages[0] instanceof GroupMessage ? messages[0].seq : messages[0].time;
+      for (let i = messages.length - 1; i >= 0; i--) {
+        const where: {
+          instanceId: number,
+          qqSenderId: number,
+          qqRoomId: number,
+          seq: number,
+          rand?: number
+        } = {
+          instanceId: this.instance.id,
+          qqSenderId: messages[i].sender.user_id,
+          qqRoomId: this.pair.qqRoomId,
+          seq: messages[i].seq,
+        };
+        if (messages[i] instanceof PrivateMessage) {
+          where.rand = messages[i].rand;
+        }
+        const dbMessage = await db.message.findFirst({ where });
+        if (!dbMessage) {
+          messagesAllExist = false;
+          this.historyMessages.unshift(messages[i]);
+        }
+      }
+      await this.updateStatusMessage();
+      if (messagesAllExist) return;
+    }
+  }
+
+  private async messagesToTxt() {
+    let lastMediaCount = 0;
+    for (const message of this.historyMessages) {
+      let text = '';
+      const useFile = (fileKey: string, filePath: string) => {
+        if (!path.extname(fileKey)) fileKey += '.file';
+        this.filesMap[fileKey] = filePath;
+        this.importTxt += `${format(new Date(message.time * 1000), 'DD/MM/YYYY, HH:mm')} - ` +
+          `${message.nickname}: ${fileKey} (file attached)\n`;
+      };
+      for (const elem of message.message) {
+        let url: string;
+        switch (elem.type) {
+          case 'text': {
+            let tgs = lottie.getTgsIndex(elem.text);
+            if (tgs === -1) {
+              text += elem.text;
+            }
+            else {
+              useFile(`${tgs}.tgs`, `assets/tgs/tgs${tgs}.tgs`);
+            }
+            break;
+          }
+          case 'at':
+          case 'face':
+          case 'sface': {
+            text += `[${elem.text}]`;
+            break;
+          }
+          case 'bface': {
+            const fileKey = md5Hex(elem.file) + '.webp';
+            useFile(fileKey, await convert.webp(fileKey, () => fetchFile(getBigFaceUrl(elem.file))));
+            break;
+          }
+          case 'video':
+            // 先获取 URL，要传给下面
+            url = await this.pair.qq.getVideoUrl(elem.fid, elem.md5);
+          case 'image':
+          case 'flash':
+            if ('url' in elem)
+              url = elem.url;
+            try {
+              if (elem.type === 'image' && elem.asface && !(elem.file as string).toLowerCase().endsWith('.gif')) {
+                useFile(elem.file as string, await convert.webp(elem.file as string, () => fetchFile(elem.url)));
+              }
+              else {
+                useFile(elem.file as string, await convert.cachedBuffer(elem.file as string, () => fetchFile(url)));
+              }
+            }
+            catch (e) {
+              this.log.error('下载媒体失败', e);
+              // 下载失败让 Telegram 服务器下载
+              text += ` ${url} `;
+            }
+            break;
+          case 'file': {
+            const extName = path.extname(elem.name);
+            // 50M 以下文件下载转发
+            if (elem.size < 1024 * 1024 * 50 || exts.images.includes(extName.toLowerCase())) {
+              // 是图片
+              let url = await this.pair.qq.getFileUrl(elem.fid);
+              if (url.includes('?fname=')) {
+                url = url.split('?fname=')[0];
+                // Request path contains unescaped characters
+              }
+              this.log.info('正在下载媒体，长度', helper.hSize(elem.size));
+              try {
+                useFile(elem.name, await convert.cachedBuffer(elem.name, () => fetchFile(url)));
+              }
+              catch (e) {
+                this.log.error('下载媒体失败', e);
+                text += `文件: ${helper.htmlEscape(elem.name)}\n` +
+                  `大小: ${helper.hSize(elem.size)}`;
+              }
+            }
+            else {
+              text += `文件: ${helper.htmlEscape(elem.name)}\n` +
+                `大小: ${helper.hSize(elem.size)}`;
+            }
+            break;
+          }
+          case 'record': {
+            useFile(elem.md5 + '.ogg', await convert.cached(elem.md5 + '.ogg',
+              async (output) => await silk.decode(await fetchFile(elem.url), output)));
+            break;
+          }
+          case 'share': {
+            text += elem.url;
+            break;
+          }
+          case 'json': {
+            text += helper.processJson(elem.data);
+            break;
+          }
+          case 'xml': {
+            const result = helper.processXml(elem.data);
+            switch (result.type) {
+              case 'text':
+                text += helper.htmlEscape(result.text);
+                break;
+              case 'image':
+                try {
+                  useFile(result.md5, await convert.cachedBuffer(result.md5, () => fetchFile(getImageUrlByMd5(result.md5))));
+                }
+                catch (e) {
+                  this.log.error('下载媒体失败', e);
+                  text += ` ${getImageUrlByMd5(result.md5)} `;
+                }
+                break;
+              case 'forward':
+                try {
+                  const messages = await this.pair.qq.getForwardMsg(result.resId);
+                  const hash = md5Hex(result.resId);
+                  text += `转发的消息记录 ${process.env.CRV_API}/?hash=${hash}`;
+                  // 传到 Cloudflare
+                  axios.post(`${process.env.CRV_API}/add`, {
+                    auth: process.env.CRV_KEY,
+                    key: hash,
+                    data: messages,
+                  })
+                    .then(data => this.log.trace('上传消息记录到 Cloudflare', data.data))
+                    .catch(e => this.log.error('上传消息记录到 Cloudflare 失败', e));
+                }
+                catch (e) {
+                  text += '[转发多条消息（无法获取）]';
+                }
+                break;
+            }
+            break;
+          }
+          case 'rps':
+          case 'dice':
+            text += `[${elem.type === 'rps' ? '猜拳' : '骰子'}] ${elem.id}`;
+            break;
+          case 'poke':
+            text += `[戳一戳] ${elem.text}`;
+            break;
+          case 'location':
+            text += `[位置] ${elem.name}\n${elem.address}`;
+            break;
+        }
+      }
+      if (text) {
+        this.importTxt += `${format(new Date(message.time * 1000), 'DD/MM/YYYY, HH:mm')} - ` +
+          `${message.nickname}: ${text}\n`;
+      }
+      if (lastMediaCount !== Object.keys(this.filesMap).length) {
+        lastMediaCount = Object.keys(this.filesMap).length;
+        await this.updateStatusMessage();
+      }
+    }
+  }
+
+  private async importMessagesAndMedia() {
+    const tgChatForUser = await this.tgUser.getChat(this.pair.tgId);
+    const txtBuffer = Buffer.from(this.importTxt, 'utf-8');
+    const importSession = await tgChatForUser.startImportSession(
+      new CustomFile('record.txt', txtBuffer.length, '', txtBuffer),
+      Object.keys(this.filesMap).length,
+    );
+    this.currentStatus = 'uploadMedia';
+    await this.updateStatusMessage();
+    const { fileTypeFromFile } = await (Function('return import("file-type")')() as Promise<typeof import('file-type')>);
+    for (const [fileKey, filePath] of Object.entries(this.filesMap)) {
+      const type = fileKey.endsWith('.tgs') ? {
+        ext: 'tgs',
+        mime: 'application/x-tgsticker',
+      } : await fileTypeFromFile(filePath);
+      let media: Api.TypeInputMedia;
+      if (['webp', 'tgs'].includes(type.ext)) {
+        // 贴纸
+        media = new Api.InputMediaUploadedDocument({
+          file: await importSession.uploadFile(new CustomFile(
+            fileKey,
+            (await fsP.stat(filePath)).size,
+            filePath,
+          )),
+          mimeType: type.mime,
+          attributes: [],
+        });
+      }
+      else if (type.mime.startsWith('audio/')) {
+        // 语音
+        media = new Api.InputMediaUploadedDocument({
+          file: await importSession.uploadFile(new CustomFile(
+            fileKey,
+            (await fsP.stat(filePath)).size,
+            filePath,
+          )),
+          mimeType: type.mime,
+          attributes: [
+            new Api.DocumentAttributeAudio({
+              duration: 0,
+              voice: true,
+            }),
+          ],
+        });
+      }
+      else if (type.ext === 'gif') {
+        media = new Api.InputMediaUploadedDocument({
+          file: await importSession.uploadFile(new CustomFile(
+            fileKey,
+            (await fsP.stat(filePath)).size,
+            filePath,
+          )),
+          mimeType: type.mime,
+          attributes: [new Api.DocumentAttributeAnimated()],
+        });
+      }
+      else if (type.mime.startsWith('image/')) {
+        media = new Api.InputMediaUploadedPhoto({
+          file: await importSession.uploadFile(new CustomFile(
+            fileKey,
+            (await fsP.stat(filePath)).size,
+            filePath,
+          )),
+        });
+      }
+      else {
+        media = new Api.InputMediaUploadedDocument({
+          file: await importSession.uploadFile(new CustomFile(
+            fileKey,
+            (await fsP.stat(filePath)).size,
+            filePath,
+          )),
+          mimeType: type.mime,
+          attributes: [],
+        });
+      }
+      await importSession.uploadMedia(fileKey, media);
+      this.mediaUploadedCount++;
+      await this.updateStatusMessage();
+    }
+    this.currentStatus = 'finishing';
+    await this.updateStatusMessage();
+    await importSession.finish();
+  }
+
+  private lastUpdateStatusTime = 0;
+
+  private async updateStatusMessage() {
+    if (new Date().getTime() - this.lastUpdateStatusTime < 2000) return;
+    this.lastUpdateStatusTime = new Date().getTime();
+    const statusMessageText = [] as string[];
+    switch (this.currentStatus) {
+      case 'finishing':
+        statusMessageText.unshift('正在完成…');
+      case 'uploadMedia':
+        statusMessageText.unshift(`正在上传媒体… ${this.mediaUploadedCount}`);
+      case 'uploadMessage':
+        statusMessageText.unshift('正在上传消息…');
+      case 'getMedia':
+        statusMessageText.unshift(`正在下载媒体… ${Object.keys(this.filesMap).length}`);
+      case 'getMessage':
+        statusMessageText.unshift(`正在获取消息… ${this.historyMessages.length}`);
+        break;
+      case 'done':
+        statusMessageText.unshift(`成功`);
+    }
+    if (!this.statusMessage) {
+      this.statusMessage = await this.requestMessage.reply({
+        message: statusMessageText.join('\n'),
+      });
+    }
+    else {
+      try {
+        await this.statusMessage.edit({
+          text: statusMessageText.join('\n'),
+        });
+      }
+      catch (e) {
+      }
+    }
+  }
+}

src/helpers/convert.ts

@@ -20,6 +20,11 @@ const cachedConvert = async (key: string, convert: (outputPath: string) => Promi
 };
 
 const convert = {
+  cached: cachedConvert,
+  cachedBuffer: (key: string, buf: () => Promise<Buffer | Uint8Array | string>) =>
+    cachedConvert(key, async (convertedPath) => {
+      await fsP.writeFile(convertedPath, await buf());
+    }),
   // webp2png，这里 webpData 是方法因为不需要的话就不获取了
   png: (key: string, webpData: () => Promise<Buffer | Uint8Array | string>) =>
     cachedConvert(key + '.png', async (convertedPath) => {

src/models/Instance.ts

@@ -148,7 +148,7 @@ export default class Instance {
       this.requestController = new RequestController(this, this.tgBot, this.oicq);
       this.configController = new ConfigController(this, this.tgBot, this.tgUser, this.oicq);
       this.deleteMessageController = new DeleteMessageController(this, this.tgBot, this.tgUser, this.oicq);
-      this.inChatCommandsController = new InChatCommandsController(this, this.tgBot, this.oicq);
+      this.inChatCommandsController = new InChatCommandsController(this, this.tgBot, this.tgUser, this.oicq);
       this.forwardController = new ForwardController(this, this.tgBot, this.tgUser, this.oicq);
       this.fileAndFlashPhotoController = new FileAndFlashPhotoController(this, this.tgBot, this.oicq);
     })()