修复前端样式生成器不显示文本消息

close #120

Dockerfile

@@ -2,13 +2,13 @@
 # 构建前端
 #
 
-FROM node:16.14.0-bullseye AS builder
+FROM node:18.17.0-bullseye AS builder
 ARG BASE_PATH='/root/blivechat'
 WORKDIR "${BASE_PATH}/frontend"
 
 # 前端依赖
 COPY frontend/package.json ./
-RUN npm i --registry=https://registry.npmmirror.com
+RUN npm i
 
 # 编译前端
 COPY frontend ./
@@ -24,8 +24,9 @@ ARG EXT_DATA_PATH='/mnt/data'
 WORKDIR "${BASE_PATH}"
 
 # 后端依赖
+COPY blivedm/requirements.txt blivedm/
 COPY requirements.txt ./
-RUN pip3 install --no-cache-dir -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt
+RUN pip3 install --no-cache-dir -i https://mirrors.aliyun.com/pypi/simple -r requirements.txt
 
 # 数据目录
 COPY . ./

README.md

@@ -55,7 +55,7 @@
     npm i
     npm run build
     ```
-2. 运行服务器（需要Python3.6以上版本）：
+2. 运行服务器（需要Python3.8以上版本）：
     ```sh
     pip3 install -r requirements.txt
     python3 main.py
@@ -108,6 +108,8 @@ server {
 	ssl_certificate /PATH/TO/CERT.crt;
 	ssl_certificate_key /PATH/TO/CERT_KEY.key;
 
+	set $blivechat_path /PATH/TO/BLIVECHAT;
+
 	client_body_buffer_size 256k;
 	client_max_body_size 1.1m;
 
@@ -120,12 +122,20 @@ server {
 
 	# 静态文件
 	location / {
-		root /PATH/TO/BLIVECHAT/frontend/dist;
-		# 如果文件不存在，交给前端路由
-		try_files $uri $uri/ /index.html;
+		root $blivechat_path/frontend/dist;
+		try_files $uri $uri/ @index;
+	}
+	# 不存在的文件请求转发到index.html，交给前端路由
+	location @index {
+		rewrite ^ /index.html last;
+	}
+	location = /index.html {
+		root $blivechat_path/frontend/dist;
+		# index.html不缓存，防止更新后前端还是旧版
+		add_header Cache-Control no-cache;
 	}
 	location /emoticons {
-		alias /PATH/TO/BLIVECHAT/data/emoticons;
+		alias $blivechat_path/data/emoticons;
 	}
 	# 动态API
 	location /api {

api/base.py

@@ -10,6 +10,8 @@ class ApiHandler(tornado.web.RequestHandler):  # noqa
         self.json_args = None
 
     def prepare(self):
+        self.set_header('Cache-Control', 'no-cache')
+
         if not self.request.headers.get('Content-Type', '').startswith('application/json'):
             return
         try:

api/chat.py

@@ -63,6 +63,7 @@ def make_text_message_data(
     translation: str = '',
     content_type: int = ContentType.TEXT,
     content_type_params: list = None,
+    text_emoticons: Iterable[Tuple[str, str]] = None
 ):
     # 为了节省带宽用list而不是dict
     return [
@@ -96,6 +97,8 @@ def make_text_message_data(
         content_type,
         # 14: contentTypeParams
         content_type_params if content_type_params is not None else [],
+        # 15: textEmoticons
+        text_emoticons if text_emoticons is not None else [],
     ]
 
 
@@ -129,26 +132,26 @@ class ChatHandler(tornado.websocket.WebSocketHandler):  # noqa
 
     def open(self):
         logger.info('client=%s connected', self.request.remote_ip)
-        self._heartbeat_timer_handle = asyncio.get_event_loop().call_later(
+        self._heartbeat_timer_handle = asyncio.get_running_loop().call_later(
             self.HEARTBEAT_INTERVAL, self._on_send_heartbeat
         )
         self._refresh_receive_timeout_timer()
 
     def _on_send_heartbeat(self):
         self.send_cmd_data(Command.HEARTBEAT, {})
-        self._heartbeat_timer_handle = asyncio.get_event_loop().call_later(
+        self._heartbeat_timer_handle = asyncio.get_running_loop().call_later(
             self.HEARTBEAT_INTERVAL, self._on_send_heartbeat
         )
 
     def _refresh_receive_timeout_timer(self):
         if self._receive_timeout_timer_handle is not None:
             self._receive_timeout_timer_handle.cancel()
-        self._receive_timeout_timer_handle = asyncio.get_event_loop().call_later(
+        self._receive_timeout_timer_handle = asyncio.get_running_loop().call_later(
             self.RECEIVE_TIMEOUT, self._on_receive_timeout
         )
 
     def _on_receive_timeout(self):
-        logger.warning('client=%s timed out', self.request.remote_ip)
+        logger.info('client=%s timed out', self.request.remote_ip)
         self._receive_timeout_timer_handle = None
         self.close()
 
@@ -165,15 +168,13 @@ class ChatHandler(tornado.websocket.WebSocketHandler):  # noqa
 
     def on_message(self, message):
         try:
-            # 超时没有加入房间也断开
-            if self.has_joined_room:
-                self._refresh_receive_timeout_timer()
-
             body = json.loads(message)
             cmd = body['cmd']
 
             if cmd == Command.HEARTBEAT:
-                pass
+                # 超时没有加入房间也断开
+                if self.has_joined_room:
+                    self._refresh_receive_timeout_timer()
 
             elif cmd == Command.JOIN_ROOM:
                 if self.has_joined_room:
@@ -189,7 +190,7 @@ class ChatHandler(tornado.websocket.WebSocketHandler):  # noqa
                     pass
 
                 services.chat.client_room_manager.add_client(self.room_id, self)
-                asyncio.ensure_future(self._on_joined_room())
+                asyncio.create_task(self._on_joined_room())
 
             else:
                 logger.warning('client=%s unknown cmd=%d, body=%s', self.request.remote_ip, cmd, body)
@@ -265,10 +266,15 @@ class ChatHandler(tornado.websocket.WebSocketHandler):  # noqa
             'translation': ''
         }
         self.send_cmd_data(Command.ADD_TEXT, text_data)
+        text_data[4] = 'te[dog]st'
+        text_data[11] = uuid.uuid4().hex
+        text_data[15] = [('[dog]', 'http://i0.hdslb.com/bfs/live/4428c84e694fbf4e0ef6c06e958d9352c3582740.png')]
+        self.send_cmd_data(Command.ADD_TEXT, text_data)
         text_data[2] = '主播'
         text_data[3] = 3
         text_data[4] = "I can eat glass, it doesn't hurt me."
         text_data[11] = uuid.uuid4().hex
+        text_data[15] = []
         self.send_cmd_data(Command.ADD_TEXT, text_data)
         self.send_cmd_data(Command.ADD_MEMBER, member_data)
         self.send_cmd_data(Command.ADD_SUPER_CHAT, sc_data)
@@ -276,7 +282,7 @@ class ChatHandler(tornado.websocket.WebSocketHandler):  # noqa
         sc_data['price'] = 100
         sc_data['content'] = '敏捷的棕色狐狸跳过了懒狗'
         self.send_cmd_data(Command.ADD_SUPER_CHAT, sc_data)
-        # self.send_message(Command.DEL_SUPER_CHAT, {'ids': [sc_data['id']]})
+        # self.send_cmd_data(Command.DEL_SUPER_CHAT, {'ids': [sc_data['id']]})
         self.send_cmd_data(Command.ADD_GIFT, gift_data)
         gift_data['id'] = uuid.uuid4().hex
         gift_data['totalCoin'] = 1245000
@@ -306,7 +312,15 @@ class RoomInfoHandler(api.base.ApiHandler):  # noqa
     async def _get_room_info(room_id):
         try:
             async with utils.request.http_session.get(
-                blivedm_client.ROOM_INIT_URL, params={'room_id': room_id}
+                blivedm_client.ROOM_INIT_URL,
+                headers={
+                    **utils.request.BILIBILI_COMMON_HEADERS,
+                    'Origin': 'https://live.bilibili.com',
+                    'Referer': f'https://live.bilibili.com/{room_id}'
+                },
+                params={
+                    'room_id': room_id
+                }
             ) as res:
                 if res.status != 200:
                     logger.warning('room=%d _get_room_info failed: %d %s', room_id,

api/main.py

@@ -19,13 +19,22 @@ EMOTICON_BASE_URL = '/emoticons'
 class MainHandler(tornado.web.StaticFileHandler):  # noqa
     """为了使用Vue Router的history模式，把不存在的文件请求转发到index.html"""
     async def get(self, path, include_body=True):
+        if path == '':
+            await self._get_index(include_body)
+            return
+
         try:
             await super().get(path, include_body)
         except tornado.web.HTTPError as e:
             if e.status_code != 404:
                 raise
             # 不存在的文件请求转发到index.html，交给前端路由
-            await super().get('index.html', include_body)
+            await self._get_index(include_body)
+
+    async def _get_index(self, include_body=True):
+        # index.html不缓存，防止更新后前端还是旧版
+        self.set_header('Cache-Control', 'no-cache')
+        await super().get('index.html', include_body)
 
 
 class ServerInfoHandler(api.base.ApiHandler):  # noqa
@@ -56,7 +65,7 @@ class UploadEmoticonHandler(api.base.ApiHandler):  # noqa
         if not file.content_type.lower().startswith('image/'):
             raise tornado.web.HTTPError(415)
 
-        url = await asyncio.get_event_loop().run_in_executor(
+        url = await asyncio.get_running_loop().run_in_executor(
             None, self._save_file, file.body, self.request.remote_ip
         )
         self.write({

config.py

@@ -49,17 +49,20 @@ def get_config():
 
 class AppConfig:
     def __init__(self):
+        self.host = '127.0.0.1'
+        self.port = 12450
         self.database_url = 'sqlite:///data/database.db'
         self.tornado_xheaders = False
         self.loader_url = ''
+        self.open_browser_at_startup = True
         self.enable_upload_file = True
 
-        self.fetch_avatar_interval = 3.5
-        self.fetch_avatar_max_queue_size = 2
-        self.avatar_cache_size = 50000
+        self.fetch_avatar_max_queue_size = 1
+        self.avatar_cache_size = 10000
 
         self.enable_translate = True
         self.allow_translate_rooms = set()
+        self.translate_max_queue_size = 10
         self.translation_cache_size = 50000
         self.translator_configs = []
 
@@ -77,18 +80,22 @@ class AppConfig:
 
     def _load_app_config(self, config: configparser.ConfigParser):
         app_section = config['app']
+        self.host = app_section.get('host', self.host)
+        self.port = app_section.getint('port', fallback=self.port)
         self.database_url = app_section.get('database_url', self.database_url)
         self.tornado_xheaders = app_section.getboolean('tornado_xheaders', fallback=self.tornado_xheaders)
         self.loader_url = app_section.get('loader_url', self.loader_url)
+        self.open_browser_at_startup = app_section.getboolean('open_browser_at_startup',
+                                                              fallback=self.open_browser_at_startup)
         self.enable_upload_file = app_section.getboolean('enable_upload_file', fallback=self.enable_upload_file)
 
-        self.fetch_avatar_interval = app_section.getfloat('fetch_avatar_interval', fallback=self.fetch_avatar_interval)
         self.fetch_avatar_max_queue_size = app_section.getint('fetch_avatar_max_queue_size',
                                                               fallback=self.fetch_avatar_max_queue_size)
         self.avatar_cache_size = app_section.getint('avatar_cache_size', fallback=self.avatar_cache_size)
 
         self.enable_translate = app_section.getboolean('enable_translate', fallback=self.enable_translate)
         self.allow_translate_rooms = _str_to_list(app_section.get('allow_translate_rooms', ''), int, set)
+        self.translate_max_queue_size = app_section.getint('translate_max_queue_size', self.translate_max_queue_size)
         self.translation_cache_size = app_section.getint('translation_cache_size', self.translation_cache_size)
 
     def _load_translator_configs(self, config: configparser.ConfigParser):
@@ -103,7 +110,6 @@ class AppConfig:
                 translator_config = {
                     'type': type_,
                     'query_interval': section.getfloat('query_interval'),
-                    'max_queue_size': section.getint('max_queue_size')
                 }
                 if type_ == 'TencentTranslateFree':
                     translator_config['source_language'] = section['source_language']

data/config.example.ini

@@ -2,8 +2,13 @@
 # If you want to modify the configuration, copy this file and rename it to "config.ini" and edit
 
 [app]
-# 数据库配置，见https://docs.sqlalchemy.org/en/13/core/engines.html#database-urls
-# See https://docs.sqlalchemy.org/en/13/core/engines.html#database-urls
+# 服务器监听的地址
+# The address the server listens on
+host = 127.0.0.1
+port = 12450
+
+# 数据库配置，见 https://docs.sqlalchemy.org/en/20/core/engines.html#database-urls
+# See https://docs.sqlalchemy.org/en/20/core/engines.html#database-urls
 database_url = sqlite:///data/database.db
 
 # 如果使用了nginx之类的反向代理服务器，设置为true
@@ -15,22 +20,22 @@ tornado_xheaders = false
 # Use a loader so that you can run OBS before blivechat. If empty, no loader is used
 loader_url = https://xfgryujk.sinacloud.net/blivechat/loader.html
 
+# 启动时打开浏览器
+# Open browser at startup
+open_browser_at_startup = true
+
 # 允许上传自定义表情文件
 # Enable uploading custom emote file
 enable_upload_file = true
 
 
-# 获取头像间隔时间（秒）。如果小于3秒有很大概率被服务器拉黑
-# Interval between fetching avatars (seconds). At least 3 seconds is recommended
-fetch_avatar_interval = 3.5
-
-# 获取头像最大队列长度，注意最长等待时间等于 最大队列长度 * 请求间隔时间
+# 获取头像最大队列长度
 # Maximum queue length for fetching avatar
-fetch_avatar_max_queue_size = 2
+fetch_avatar_max_queue_size = 4
 
-# 头像缓存数量
-# Number of avatar caches
-avatar_cache_size = 50000
+# 内存中头像缓存数量
+# Number of avatar caches in memory
+avatar_cache_size = 10000
 
 
 # 允许自动翻译到日语
@@ -42,6 +47,10 @@ enable_translate = true
 # Example: allow_translate_rooms = 4895312,22347054,21693691
 allow_translate_rooms =
 
+# 翻译最大队列长度
+# Maximum queue length for translating
+translate_max_queue_size = 10
+
 # 翻译缓存数量
 # Number of translation caches
 translation_cache_size = 50000
@@ -55,17 +64,15 @@ translation_cache_size = 50000
 # 翻译器配置，索引到下面的配置节。可以以逗号分隔配置多个翻译器，翻译时会自动负载均衡
 # 配置多个翻译器可以增加额度、增加QPS、容灾
 # 不同配置可以使用同一个类型，但要使用不同的账号，否则还是会遇到额度、调用频率限制
-translator_configs = tencent_translate_free,bilibili_translate_free
+translator_configs = tencent_translate_free
 
 
 [tencent_translate_free]
 # 类型：腾讯翻译白嫖版。使用了网页版的接口，**将来可能失效**
 type = TencentTranslateFree
 
-# 请求间隔时间（秒），等于 1 / QPS。目前没有遇到此接口有调用频率限制，10QPS应该够用了
-query_interval = 0.1
-# 最大队列长度，注意最长等待时间等于 最大队列长度 * 请求间隔时间
-max_queue_size = 100
+# 请求间隔时间（秒），等于 1 / QPS
+query_interval = 1
 
 # 自动：auto；中文：zh；日语：jp；英语：en；韩语：kr
 # 完整语言列表见文档：https://cloud.tencent.com/document/product/551/15619
@@ -75,16 +82,6 @@ source_language = zh
 target_language = jp
 
 
-[bilibili_translate_free]
-# 类型：B站翻译白嫖版。使用了B站直播网页的接口，**将来可能失效**。目前B站翻译后端是百度翻译
-type = BilibiliTranslateFree
-
-# 请求间隔时间（秒），等于 1 / QPS。目前此接口频率限制是3秒一次
-query_interval = 3.1
-# 最大队列长度，注意最长等待时间等于 最大队列长度 * 请求间隔时间
-max_queue_size = 3
-
-
 [tencent_translate]
 # 文档：https://cloud.tencent.com/product/tmt
 # 定价：https://cloud.tencent.com/document/product/551/35017
@@ -99,8 +96,6 @@ type = TencentTranslate
 
 # 请求间隔时间（秒），等于 1 / QPS。理论上最高QPS为5，实际测试是3
 query_interval = 0.333
-# 最大队列长度，注意最长等待时间等于 最大队列长度 * 请求间隔时间
-max_queue_size = 30
 
 # 自动：auto；中文：zh；日语：ja；英语：en；韩语：ko
 # 完整语言列表见文档：https://cloud.tencent.com/document/product/551/15619
@@ -122,17 +117,15 @@ region = ap-shanghai
 [baidu_translate]
 # 文档：https://fanyi-api.baidu.com/
 # 定价：https://fanyi-api.baidu.com/product/112
-# * 标准版完全免费，不限使用字符量（QPS=1）
-# * 高级版每月前200万字符免费，超出后仅收取超出部分费用（QPS=10），49元/百万字符
-# * 尊享版每月前200万字符免费，超出后仅收取超出部分费用（QPS=100），49元/百万字符
+# * 标准版每月前5万字符免费，超出仅收取超出部分费用（QPS=1），按49元/百万字符计费
+# * 高级版每月前100万字符免费，超出仅收取超出部分费用（QPS=10），按49元/百万字符计费
+# * 尊享版每月前200万字符免费，超出后仅收取超出部分费用（QPS=100），按49元/百万字符计费
 
 # 类型：百度翻译
 type = BaiduTranslate
 
 # 请求间隔时间（秒），等于 1 / QPS
 query_interval = 1.5
-# 最大队列长度，注意最长等待时间等于 最大队列长度 * 请求间隔时间
-max_queue_size = 9
 
 # 自动：auto；中文：zh；日语：jp；英语：en；韩语：kor
 # 完整语言列表见文档：https://fanyi-api.baidu.com/doc/21

frontend/.eslintrc.js

@@ -5,7 +5,7 @@ module.exports = {
     "node": true
   },
   "parserOptions": {
-    "parser": "babel-eslint"
+    "parser": "@babel/eslint-parser"
   },
   "extends": [
     "plugin:vue/essential",
@@ -75,5 +75,7 @@ module.exports = {
     "no-shadow": "warn", // 变量名和外部作用域重复
 
     "no-console": "off", // 线上尽量不要用console输出，看不到的
+
+    "vue/multi-word-component-names": "off", // Vue组件名允许用1个单词
   }
 }

frontend/jsconfig.json

@@ -7,6 +7,9 @@
       "@/*": ["src/*"]
     }
   },
+  "vueCompilerOptions": {
+    "target": 2.7,
+  },
   "include": [
     "./src/**/*.js",
     "./src/**/*.vue"

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "blivechat",
-  "version": "1.6.0",
+  "version": "1.7.0",
   "private": true,
   "scripts": {
     "serve": "vue-cli-service serve",
@@ -8,32 +8,28 @@
     "lint": "vue-cli-service lint"
   },
   "dependencies": {
-    "axios": "^0.21.1",
-    "core-js": "^3.6.5",
+    "axios": "^1.4.0",
+    "core-js": "^3.8.3",
     "downloadjs": "^1.4.7",
-    "element-ui": "^2.9.1",
-    "lodash": "^4.17.19",
-    "vue": "^2.6.10",
-    "vue-i18n": "^8.11.2",
-    "vue-router": "^3.0.6"
+    "element-ui": "^2.15.13",
+    "lodash": "^4.17.21",
+    "vue": "^2.7.14",
+    "vue-i18n": "^8.28.2",
+    "vue-router": "^3.6.5"
   },
   "devDependencies": {
-    "@vue/cli-plugin-babel": "^4.5.12",
-    "@vue/cli-plugin-eslint": "^4.5.12",
-    "@vue/cli-service": "~4.5.12",
-    "babel-eslint": "^10.1.0",
+    "@babel/core": "^7.12.16",
+    "@babel/eslint-parser": "^7.12.16",
+    "@vue/cli-plugin-babel": "~5.0.0",
+    "@vue/cli-plugin-eslint": "~5.0.0",
+    "@vue/cli-service": "~5.0.0",
     "babel-plugin-component": "^1.1.1",
-    "eslint": "^6.7.2",
-    "eslint-plugin-vue": "^6.2.2",
-    "vue-template-compiler": "^2.5.21"
-  },
-  "postcss": {
-    "plugins": {
-      "autoprefixer": {}
-    }
+    "eslint": "^7.32.0",
+    "eslint-plugin-vue": "^9.16.1"
   },
   "browserslist": [
     "> 1%",
-    "last 2 versions"
+    "last 2 versions",
+    "not dead"
   ]
 }

frontend/src/api/chat/ChatClientDirect/index.js

@@ -107,7 +107,7 @@ export default class ChatClientDirect {
 
   sendAuth() {
     let authParams = {
-      uid: 0,
+      uid: this.roomOwnerUid,
       roomid: this.roomId,
       protover: 3,
       platform: 'web',
@@ -183,7 +183,6 @@ export default class ChatClientDirect {
   }
 
   onWsMessage(event) {
-    this.refreshReceiveTimeoutTimer()
     if (!(event.data instanceof ArrayBuffer)) {
       console.warn('未知的websocket消息类型，data=', event.data)
       return
@@ -226,6 +225,7 @@ export default class ChatClientDirect {
     }
     case OP_HEARTBEAT_REPLY: {
       // 服务器心跳包，包含人气值，这里没用
+      this.refreshReceiveTimeoutTimer()
       break
     }
     default: {
@@ -322,6 +322,8 @@ export default class ChatClientDirect {
       authorType = 0
     }
 
+    let textEmoticons = this.parseTextEmoticons(info)
+
     let data = {
       avatarUrl: await avatar.getAvatarUrl(uid),
       timestamp: info[0][4] / 1000,
@@ -336,11 +338,26 @@ export default class ChatClientDirect {
       medalLevel: roomId === this.roomId ? medalLevel : 0,
       id: getUuid4Hex(),
       translation: '',
-      emoticon: info[0][13].url || null
+      emoticon: info[0][13].url || null,
+      textEmoticons: textEmoticons,
     }
     this.onAddText(data)
   }
 
+  parseTextEmoticons(info) {
+    try {
+      let modeInfo = info[0][15]
+      let extra = JSON.parse(modeInfo.extra)
+      if (!extra.emots) {
+        return []
+      }
+      let res = Object.values(extra.emots).map(emoticon => [emoticon.descript, emoticon.url])
+      return res
+    } catch {
+      return []
+    }
+  }
+
   sendGiftCallback(command) {
     if (!this.onAddGift) {
       return

frontend/src/api/chat/ChatClientRelay.js

@@ -10,8 +10,7 @@ const COMMAND_UPDATE_TRANSLATION = 7
 // const CONTENT_TYPE_TEXT = 0
 const CONTENT_TYPE_EMOTICON = 1
 
-const HEARTBEAT_INTERVAL = 10 * 1000
-const RECEIVE_TIMEOUT = HEARTBEAT_INTERVAL + (5 * 1000)
+const RECEIVE_TIMEOUT = 15 * 1000
 
 export default class ChatClientRelay {
   constructor(roomId, autoTranslate) {
@@ -28,7 +27,6 @@ export default class ChatClientRelay {
     this.websocket = null
     this.retryCount = 0
     this.isDestroying = false
-    this.heartbeatTimerId = null
     this.receiveTimeoutTimerId = null
   }
 
@@ -66,16 +64,9 @@ export default class ChatClientRelay {
         }
       }
     }))
-    this.heartbeatTimerId = window.setInterval(this.sendHeartbeat.bind(this), HEARTBEAT_INTERVAL)
     this.refreshReceiveTimeoutTimer()
   }
 
-  sendHeartbeat() {
-    this.websocket.send(JSON.stringify({
-      cmd: COMMAND_HEARTBEAT
-    }))
-  }
-
   refreshReceiveTimeoutTimer() {
     if (this.receiveTimeoutTimerId) {
       window.clearTimeout(this.receiveTimeoutTimerId)
@@ -95,10 +86,6 @@ export default class ChatClientRelay {
 
   onWsClose() {
     this.websocket = null
-    if (this.heartbeatTimerId) {
-      window.clearInterval(this.heartbeatTimerId)
-      this.heartbeatTimerId = null
-    }
     if (this.receiveTimeoutTimerId) {
       window.clearTimeout(this.receiveTimeoutTimerId)
       this.receiveTimeoutTimerId = null
@@ -112,11 +99,15 @@ export default class ChatClientRelay {
   }
 
   onWsMessage(event) {
-    this.refreshReceiveTimeoutTimer()
-
     let { cmd, data } = JSON.parse(event.data)
     switch (cmd) {
     case COMMAND_HEARTBEAT: {
+      this.refreshReceiveTimeoutTimer()
+
+      // 不能由定时器触发发心跳包，因为浏览器会把不活动页面的定时器调到1分钟以上
+      this.websocket.send(JSON.stringify({
+        cmd: COMMAND_HEARTBEAT
+      }))
       break
     }
     case COMMAND_ADD_TEXT: {
@@ -145,7 +136,8 @@ export default class ChatClientRelay {
         medalLevel: data[10],
         id: data[11],
         translation: data[12],
-        emoticon: emoticon
+        emoticon: emoticon,
+        textEmoticons: data[15],
       }
       this.onAddText(data)
       break

frontend/src/api/chat/ChatClientTest.js

@@ -3,20 +3,57 @@ import * as constants from '@/components/ChatRenderer/constants'
 import * as avatar from './avatar'
 
 const NAMES = [
-  'xfgryujk', 'Simon', 'Il Harper', 'Kinori', 'shugen', 'yuyuyzl', '3Shain', '光羊', '黑炎', 'Misty', '孤梦星影',
-  'ジョナサン・ジョースター', 'ジョセフ・ジョースター', 'ディオ・ブランドー', '空條承太郎', '博丽灵梦', '雾雨魔理沙',
-  'Rick Astley'
+  '光羊',
+  '黑炎',
+  '孤梦星影',
+  '博丽灵梦',
+  '雾雨魔理沙',
+  '空條承太郎',
+  'ディオ・ブランドー',
+  'ジョセフ・ジョースター',
+  'ジョナサン・ジョースター',
+  'Simon',
+  'Misty',
+  'Kinori',
+  'shugen',
+  '3Shain',
+  'yuyuyzl',
+  'xfgryujk',
+  'Il Harper',
+  'Rick Astley',
 ]
 
 const CONTENTS = [
-  '草', 'kksk', '8888888888', '888888888888888888888888888888', '老板大气，老板身体健康',
-  'The quick brown fox jumps over the lazy dog', "I can eat glass, it doesn't hurt me",
-  '我不做人了，JOJO', '無駄無駄無駄無駄無駄無駄無駄無駄', '欧啦欧啦欧啦欧啦欧啦欧啦欧啦欧啦', '逃げるんだよォ！',
-  '嚯，朝我走过来了吗，没有选择逃跑而是主动接近我么', '不要停下来啊', '已经没有什么好怕的了',
-  'I am the bone of my sword. Steel is my body, and fire is my blood.', '言いたいことがあるんだよ！',
-  '我忘不掉夏小姐了。如果不是知道了夏小姐，说不定我已经对这个世界没有留恋了', '迷えば、敗れる',
-  'Farewell, ashen one. May the flame guide thee', '竜神の剣を喰らえ！', '竜が我が敌を喰らう！',
-  '有一说一，这件事大家懂的都懂，不懂的，说了你也不明白，不如不说', '让我看看', '我柜子动了，我不玩了'
+  '草',
+  '让我看看',
+  '不要停下来啊',
+  '我不做人了，JOJO',
+  '已经没有什么好怕的了',
+  '我柜子动了，我不玩了',
+  '老板大气，老板身体健康',
+  '我醉提酒游寒山，爽滑慢舔',
+  '無駄無駄無駄無駄無駄無駄無駄無駄',
+  '欧啦欧啦欧啦欧啦欧啦欧啦欧啦欧啦',
+  '所有没好全部康复呀，我的癌也全部康复呀',
+  '嚯，朝我走过来了吗，没有选择逃跑而是主动接近我么',
+  '有一说一，这件事大家懂的都懂，不懂的，说了你也不明白，不如不说',
+  '如来来了吗？如来嘛~他真来了吗？如~来~到底来没来？如来~如来他真来了吗？如来~你看看，来没来？如~来~',
+  '迷えば、敗れる',
+  '逃げるんだよォ！',
+  '竜神の剣を喰らえ！',
+  '竜が我が敌を喰らう！',
+  '言いたいことがあるんだよ！',
+  '知らず知らず隠してた 本当の声を響かせてよほら',
+  'kksk',
+  '8888888888',
+  'Never gonna give you up',
+  'Never gonna let you down',
+  '888888888888888888888888888888',
+  'I am the storm that is approaching',
+  "I can eat glass, it doesn't hurt me",
+  'The quick brown fox jumps over the lazy dog',
+  'Farewell, ashen one. May the flame guide thee',
+  'I am the bone of my sword. Steel is my body, and fire is my blood.',
 ]
 
 const EMOTICONS = [
@@ -78,7 +115,8 @@ const MESSAGE_GENERATORS = [
           medalLevel: randInt(0, 40),
           id: getUuid4Hex(),
           translation: '',
-          emoticon: null
+          emoticon: null,
+          textEmoticons: [],
         }
       }
     }
@@ -102,7 +140,8 @@ const MESSAGE_GENERATORS = [
           medalLevel: randInt(0, 40),
           id: getUuid4Hex(),
           translation: '',
-          emoticon: randomChoose(EMOTICONS)
+          emoticon: randomChoose(EMOTICONS),
+          textEmoticons: [],
         }
       }
     }
@@ -194,9 +233,6 @@ function randInt(min, max) {
 
 export default class ChatClientTest {
   constructor() {
-    this.minSleepTime = 800
-    this.maxSleepTime = 1200
-
     this.onAddText = null
     this.onAddGift = null
     this.onAddMember = null
@@ -219,7 +255,14 @@ export default class ChatClientTest {
   }
 
   refreshTimer() {
-    this.timerId = window.setTimeout(this.onTimeout.bind(this), randInt(this.minSleepTime, this.maxSleepTime))
+    // 模仿B站的消息间隔模式
+    let sleepTime
+    if (randInt(0, 4) == 0) {
+      sleepTime = randInt(1000, 2000)
+    } else {
+      sleepTime = randInt(0, 400)
+    }
+    this.timerId = window.setTimeout(this.onTimeout.bind(this), sleepTime)
   }
 
   onTimeout() {

frontend/src/components/ChatRenderer/index.vue

@@ -264,15 +264,19 @@ export default {
       } else {
         let curTime = new Date()
         let interval = curTime - this.lastEnqueueTime
-        // 让发消息速度变化不要太频繁
-        if (interval > 1000) {
+        // 真实的进队列时间间隔模式大概是这样：2500, 300, 300, 300, 2500, 300, ...
+        // B站消息有缓冲，会一次发多条消息。这里把波峰视为发送了一次真实的WS消息，所以要过滤掉间隔太小的
+        if (interval > 1000 || this.enqueueIntervals.length < 5) {
           this.enqueueIntervals.push(interval)
           if (this.enqueueIntervals.length > 5) {
             this.enqueueIntervals.splice(0, this.enqueueIntervals.length - 5)
           }
+          // 这边估计得尽量大，只要不太早把消息缓冲发完就是平滑的。有MESSAGE_MAX_INTERVAL保底，不会让消息延迟太大
+          // 其实可以用单调队列求最大值，偷懒不写了
           this.estimatedEnqueueInterval = Math.max(...this.enqueueIntervals)
-          this.lastEnqueueTime = curTime
         }
+        // 上次入队时间还是要设置，否则会太早把消息缓冲发完，然后较长时间没有新消息
+        this.lastEnqueueTime = curTime
       }
 
       // 把messages分成messageGroup，每个组里最多有1个需要平滑的消息

frontend/src/utils/trie.js

@@ -41,7 +41,7 @@ export class Trie {
     return this.get(key) !== null
   }
 
-  greedyMatch(str) {
+  lazyMatch(str) {
     let node = this._root
     for (let char of str) {
       let nextNode = node.children[char]

frontend/src/views/Room.vue

@@ -33,7 +33,8 @@ export default {
     return {
       config: chatConfig.deepCloneDefaultConfig(),
       chatClient: null,
-      pronunciationConverter: null
+      pronunciationConverter: null,
+      textEmoticons: {}, // 官方的文本表情，运行时从弹幕消息收集
     }
   },
   computed: {
@@ -64,6 +65,9 @@ export default {
           res.set(emoticon.keyword, emoticon)
         }
       }
+      for (let emoticon of Object.values(this.textEmoticons)) {
+        res.set(emoticon.keyword, emoticon)
+      }
       return res
     }
   },
@@ -163,6 +167,15 @@ export default {
       if (!this.config.showDanmaku || !this.filterTextMessage(data) || this.mergeSimilarText(data.content)) {
         return
       }
+
+      // 更新官方文本表情
+      for (let [keyword, url] of data.textEmoticons) {
+        if (!(keyword in this.textEmoticons)) {
+          let emoticon = { keyword, url }
+          this.$set(this.textEmoticons, keyword, emoticon)
+        }
+      }
+
       let message = {
         id: data.id,
         type: constants.MESSAGE_TYPE_TEXT,
@@ -272,7 +285,7 @@ export default {
       let blockKeywordsTrie = this.blockKeywordsTrie
       for (let i = 0; i < content.length; i++) {
         let remainContent = content.substring(i)
-        if (blockKeywordsTrie.greedyMatch(remainContent) !== null) {
+        if (blockKeywordsTrie.lazyMatch(remainContent) !== null) {
           return false
         }
       }
@@ -312,8 +325,8 @@ export default {
         return richContent
       }
 
-      // 没有自定义表情，只能是文本
-      if (this.config.emoticons.length === 0) {
+      // 没有文本表情，只能是纯文本
+      if (this.config.emoticons.length === 0 && Object.keys(this.textEmoticons).length === 0) {
         richContent.push({
           type: constants.CONTENT_TYPE_TEXT,
           text: data.content
@@ -321,13 +334,13 @@ export default {
         return richContent
       }
 
-      // 可能含有自定义表情，需要解析
+      // 可能含有文本表情，需要解析
       let emoticonsTrie = this.emoticonsTrie
       let startPos = 0
       let pos = 0
       while (pos < data.content.length) {
         let remainContent = data.content.substring(pos)
-        let matchEmoticon = emoticonsTrie.greedyMatch(remainContent)
+        let matchEmoticon = emoticonsTrie.lazyMatch(remainContent)
         if (matchEmoticon === null) {
           pos++
           continue

frontend/vue.config.js

@@ -1,4 +1,5 @@
-const API_BASE_URL = 'http://localhost:12450'
+// 不能用localhost，https://forum.dfinity.org/t/development-workflow-quickly-test-code-modifications/1793/21
+const API_BASE_URL = 'http://127.0.0.1:12450'
 
 module.exports = {
   devServer: {

main.py

@@ -16,6 +16,7 @@ import services.avatar
 import services.chat
 import services.translate
 import update
+import utils.request
 
 logger = logging.getLogger(__name__)
 
@@ -28,7 +29,7 @@ routes = [
     (r'/api/avatar_url', api.chat.AvatarHandler),
 
     (rf'{api.main.EMOTICON_BASE_URL}/(.*)', tornado.web.StaticFileHandler, {'path': api.main.EMOTICON_UPLOAD_PATH}),
-    (r'/(.*)', api.main.MainHandler, {'path': config.WEB_ROOT, 'default_filename': 'index.html'})
+    (r'/(.*)', api.main.MainHandler, {'path': config.WEB_ROOT})
 ]
 
 
@@ -37,10 +38,14 @@ def main():
 
     init_logging(args.debug)
     config.init()
+
+    utils.request.init()
     models.database.init(args.debug)
+
     services.avatar.init()
     services.translate.init()
     services.chat.init()
+
     update.check_update()
 
     run_server(args.host, args.port, args.debug)
@@ -48,8 +53,8 @@ def main():
 
 def parse_args():
     parser = argparse.ArgumentParser(description='用于OBS的仿YouTube风格的bilibili直播评论栏')
-    parser.add_argument('--host', help='服务器host，默认为127.0.0.1', default='127.0.0.1')
-    parser.add_argument('--port', help='服务器端口，默认为12450', type=int, default=12450)
+    parser.add_argument('--host', help='服务器host，默认和配置中的一样', default=None)
+    parser.add_argument('--port', help='服务器端口，默认和配置中的一样', type=int, default=None)
     parser.add_argument('--debug', help='调试模式', action='store_true')
     return parser.parse_args()
 
@@ -62,7 +67,6 @@ def init_logging(debug):
     )
     logging.basicConfig(
         format='{asctime} {levelname} [{name}]: {message}',
-        datefmt='%Y-%m-%d %H:%M:%S',
         style='{',
         level=logging.INFO if not debug else logging.DEBUG,
         handlers=[stream_handler, file_handler]
@@ -73,13 +77,18 @@ def init_logging(debug):
 
 
 def run_server(host, port, debug):
+    cfg = config.get_config()
+    if host is None:
+        host = cfg.host
+    if port is None:
+        port = cfg.port
+
     app = tornado.web.Application(
         routes,
         websocket_ping_interval=10,
         debug=debug,
         autoreload=False
     )
-    cfg = config.get_config()
     try:
         app.listen(
             port,
@@ -92,10 +101,9 @@ def run_server(host, port, debug):
         logger.warning('Address is used %s:%d', host, port)
         return
     finally:
-        url = 'http://localhost/' if port == 80 else f'http://localhost:{port}/'
-        # 防止更新版本后浏览器加载缓存
-        url += '?_v=' + update.VERSION
-        webbrowser.open(url)
+        if cfg.open_browser_at_startup:
+            url = 'http://localhost/' if port == 80 else f'http://localhost:{port}/'
+            webbrowser.open(url)
     logger.info('Server started: %s:%d', host, port)
     tornado.ioloop.IOLoop.current().start()
 

models/bilibili.py

@@ -1,11 +1,14 @@
 # -*- coding: utf-8 -*-
+import datetime
+
 import sqlalchemy
+from sqlalchemy.orm import Mapped, mapped_column
 
 import models.database
 
 
 class BilibiliUser(models.database.OrmBase):
     __tablename__ = 'bilibili_users'
-    uid = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
-    avatar_url = sqlalchemy.Column(sqlalchemy.String(100))
-    update_time = sqlalchemy.Column(sqlalchemy.DateTime)
+    uid: Mapped[int] = mapped_column(sqlalchemy.BigInteger, primary_key=True)  # 创建表后最好手动改成unsigned
+    avatar_url: Mapped[str] = mapped_column(sqlalchemy.String(100))
+    update_time: Mapped[datetime.datetime]

models/database.py

@@ -1,25 +1,32 @@
 # -*- coding: utf-8 -*-
 from typing import *
 
-import sqlalchemy.ext.declarative
 import sqlalchemy.orm
 
 import config
 
-OrmBase = sqlalchemy.ext.declarative.declarative_base()
-_engine = None
-_DbSession: Optional[Type[sqlalchemy.orm.Session]] = None
+_engine: Optional[sqlalchemy.Engine] = None
+
+
+class OrmBase(sqlalchemy.orm.DeclarativeBase):
+    pass
 
 
 def init(_debug):
     cfg = config.get_config()
-    global _engine, _DbSession
-    # engine = sqlalchemy.create_engine(cfg.database_url, echo=debug)
-    _engine = sqlalchemy.create_engine(cfg.database_url)
-    _DbSession = sqlalchemy.orm.sessionmaker(bind=_engine)
+    global _engine
+    _engine = sqlalchemy.create_engine(
+        cfg.database_url,
+        pool_size=5,  # 保持的连接数
+        max_overflow=5,  # 临时的额外连接数
+        pool_timeout=3,  # 连接数达到最大时获取新连接的超时时间
+        # pool_pre_ping=True,  # 获取连接时先检测是否可用
+        pool_recycle=60 * 60,  # 回收超过1小时的连接，防止数据库服务器主动断开不活跃的连接
+        # echo=debug,  # 输出SQL语句
+    )
 
     OrmBase.metadata.create_all(_engine)
 
 
-def get_session():
-    return _DbSession()
+def get_session() -> sqlalchemy.orm.Session:
+    return sqlalchemy.orm.Session(_engine)

requirements.txt

@@ -1,5 +1,5 @@
-aiohttp==3.7.4
-Brotli==1.0.9
+-r blivedm/requirements.txt
+cachetools==5.3.1
 pycryptodome==3.10.1
-sqlalchemy==1.4.31
-tornado==6.1.0
+sqlalchemy==2.0.19
+tornado==6.3.2

services/avatar.py

@@ -1,12 +1,15 @@
 # -*- coding: utf-8 -*-
 import asyncio
+import dataclasses
 import datetime
+import hashlib
 import logging
 import re
+import urllib.parse
 from typing import *
 
 import aiohttp
-import sqlalchemy
+import cachetools
 import sqlalchemy.exc
 
 import config
@@ -19,151 +22,92 @@ logger = logging.getLogger(__name__)
 
 DEFAULT_AVATAR_URL = '//static.hdslb.com/images/member/noface.gif'
 
-_main_event_loop = asyncio.get_event_loop()
+_avatar_fetchers: List['AvatarFetcher'] = []
 # user_id -> avatar_url
-_avatar_url_cache: Dict[int, str] = {}
+_avatar_url_cache: Optional[cachetools.TTLCache] = None
 # 正在获取头像的Future，user_id -> Future
 _uid_fetch_future_map: Dict[int, asyncio.Future] = {}
-# 正在获取头像的user_id队列
-_uid_queue_to_fetch: Optional[asyncio.Queue] = None
-# 上次被B站ban时间
-_last_fetch_banned_time: Optional[datetime.datetime] = None
+# 正在获取头像的任务队列
+_task_queue: Optional['asyncio.Queue[FetchTask]'] = None
+
+
+@dataclasses.dataclass
+class FetchTask:
+    user_id: int
+    future: 'asyncio.Future[Optional[str]]'
 
 
 def init():
     cfg = config.get_config()
-    global _uid_queue_to_fetch
-    _uid_queue_to_fetch = asyncio.Queue(cfg.fetch_avatar_max_queue_size)
-    asyncio.ensure_future(_get_avatar_url_from_web_consumer())
+    global _avatar_url_cache, _task_queue
+    _avatar_url_cache = cachetools.TTLCache(cfg.avatar_cache_size, 10 * 60)
+    _task_queue = asyncio.Queue(cfg.fetch_avatar_max_queue_size)
+    asyncio.get_event_loop().create_task(_do_init())
 
 
-async def get_avatar_url(user_id):
+async def _do_init():
+    fetchers = [
+        UserSpaceAvatarFetcher(5.5),
+        MedalAnchorAvatarFetcher(3),
+        UserCardAvatarFetcher(3),
+        GameUserCenterAvatarFetcher(3),
+    ]
+    await asyncio.gather(*(fetcher.init() for fetcher in fetchers))
+    global _avatar_fetchers
+    _avatar_fetchers = fetchers
+
+
+async def get_avatar_url(user_id) -> str:
     avatar_url = await get_avatar_url_or_none(user_id)
     if avatar_url is None:
         avatar_url = DEFAULT_AVATAR_URL
     return avatar_url
 
 
-async def get_avatar_url_or_none(user_id):
-    avatar_url = get_avatar_url_from_memory(user_id)
+async def get_avatar_url_or_none(user_id) -> Optional[str]:
+    if user_id == 0:
+        return None
+
+    # 查内存
+    avatar_url = _get_avatar_url_from_memory(user_id)
     if avatar_url is not None:
         return avatar_url
-    avatar_url = await get_avatar_url_from_database(user_id)
-    if avatar_url is not None:
+
+    # 查数据库
+    user = await _get_avatar_url_from_database(user_id)
+    if user is not None:
+        avatar_url = user.avatar_url
+        _update_avatar_cache_in_memory(user_id, avatar_url)
+        # 如果距离数据库上次更新太久，则在后台从接口获取，并更新所有缓存
+        if (datetime.datetime.now() - user.update_time).days >= 1:
+            asyncio.create_task(_refresh_avatar_cache_from_web(user_id))
         return avatar_url
-    return await get_avatar_url_from_web(user_id)
+
+    # 从接口获取
+    avatar_url = await _get_avatar_url_from_web(user_id)
+    if avatar_url is not None:
+        update_avatar_cache(user_id, avatar_url)
+        return avatar_url
+
+    return None
 
 
-def get_avatar_url_from_memory(user_id):
-    return _avatar_url_cache.get(user_id, None)
-
-
-def get_avatar_url_from_database(user_id) -> Awaitable[Optional[str]]:
-    return asyncio.get_event_loop().run_in_executor(
-        None, _do_get_avatar_url_from_database, user_id
-    )
-
-
-def _do_get_avatar_url_from_database(user_id):
-    try:
-        with models.database.get_session() as session:
-            user = session.query(bl_models.BilibiliUser).filter(
-                bl_models.BilibiliUser.uid == user_id
-            ).one_or_none()
-            if user is None:
-                return None
-            avatar_url = user.avatar_url
-
-            # 如果离上次更新太久就更新所有缓存
-            if (datetime.datetime.now() - user.update_time).days >= 3:
-                def refresh_cache():
-                    _avatar_url_cache.pop(user_id, None)
-                    get_avatar_url_from_web(user_id)
-
-                _main_event_loop.call_soon(refresh_cache)
-            else:
-                # 否则只更新内存缓存
-                _update_avatar_cache_in_memory(user_id, avatar_url)
-    except sqlalchemy.exc.OperationalError:
-        # SQLite会锁整个文件，忽略就行
-        return None
-    except sqlalchemy.exc.SQLAlchemyError:
-        logger.exception('_do_get_avatar_url_from_database failed:')
-        return None
-    return avatar_url
-
-
-def get_avatar_url_from_web(user_id) -> Awaitable[Optional[str]]:
-    # 如果已有正在获取的future则返回，防止重复获取同一个uid
-    future = _uid_fetch_future_map.get(user_id, None)
-    if future is not None:
-        return future
-    # 否则创建一个获取任务
-    _uid_fetch_future_map[user_id] = future = _main_event_loop.create_future()
-    future.add_done_callback(lambda _future: _uid_fetch_future_map.pop(user_id, None))
-    try:
-        _uid_queue_to_fetch.put_nowait(user_id)
-    except asyncio.QueueFull:
-        future.set_result(None)
-    return future
-
-
-async def _get_avatar_url_from_web_consumer():
-    while True:
-        try:
-            user_id = await _uid_queue_to_fetch.get()
-            future = _uid_fetch_future_map.get(user_id, None)
-            if future is None:
-                continue
-
-            # 防止在被ban的时候获取
-            global _last_fetch_banned_time
-            if _last_fetch_banned_time is not None:
-                cur_time = datetime.datetime.now()
-                if (cur_time - _last_fetch_banned_time).total_seconds() < 3 * 60 + 3:
-                    # 3分钟以内被ban，解封大约要15分钟
-                    future.set_result(None)
-                    continue
-                else:
-                    _last_fetch_banned_time = None
-
-            asyncio.ensure_future(_get_avatar_url_from_web_coroutine(user_id, future))
-
-            # 限制频率，防止被B站ban
-            cfg = config.get_config()
-            await asyncio.sleep(cfg.fetch_avatar_interval)
-        except Exception:  # noqa
-            logger.exception('_get_avatar_url_from_web_consumer error:')
-
-
-async def _get_avatar_url_from_web_coroutine(user_id, future):
-    try:
-        avatar_url = await _do_get_avatar_url_from_web(user_id)
-    except BaseException as e:
-        future.set_exception(e)
-    else:
-        future.set_result(avatar_url)
-
-
-async def _do_get_avatar_url_from_web(user_id):
-    try:
-        async with utils.request.http_session.get(
-            'https://api.bilibili.com/x/space/acc/info', params={'mid': user_id}
-        ) as r:
-            if r.status != 200:
-                logger.warning('Failed to fetch avatar: status=%d %s uid=%d', r.status, r.reason, user_id)
-                if r.status == 412:
-                    # 被B站ban了
-                    global _last_fetch_banned_time
-                    _last_fetch_banned_time = datetime.datetime.now()
-                return None
-            data = await r.json()
-    except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
-        return None
-
-    avatar_url = process_avatar_url(data['data']['face'])
+async def _refresh_avatar_cache_from_web(user_id):
+    avatar_url = await _get_avatar_url_from_web(user_id)
+    if avatar_url is None:
+        return
     update_avatar_cache(user_id, avatar_url)
-    return avatar_url
+
+
+def update_avatar_cache(user_id, avatar_url):
+    _update_avatar_cache_in_memory(user_id, avatar_url)
+    _update_avatar_cache_in_database(user_id, avatar_url)
+
+
+def update_avatar_cache_if_expired(user_id, avatar_url):
+    # 内存缓存过期了才更新，减少写入数据库的频率
+    if _get_avatar_url_from_memory(user_id) is None:
+        update_avatar_cache(user_id, avatar_url)
 
 
 def process_avatar_url(avatar_url):
@@ -177,25 +121,51 @@ def process_avatar_url(avatar_url):
     return avatar_url
 
 
-def update_avatar_cache(user_id, avatar_url):
-    _update_avatar_cache_in_memory(user_id, avatar_url)
-    asyncio.get_event_loop().run_in_executor(
-        None, _update_avatar_cache_in_database, user_id, avatar_url
-    )
+def _get_avatar_url_from_memory(user_id) -> Optional[str]:
+    return _avatar_url_cache.get(user_id, None)
 
 
 def _update_avatar_cache_in_memory(user_id, avatar_url):
     _avatar_url_cache[user_id] = avatar_url
-    cfg = config.get_config()
-    while len(_avatar_url_cache) > cfg.avatar_cache_size:
-        _avatar_url_cache.pop(next(iter(_avatar_url_cache)), None)
 
 
-def _update_avatar_cache_in_database(user_id, avatar_url):
+def _get_avatar_url_from_database(user_id) -> Awaitable[Optional[bl_models.BilibiliUser]]:
+    loop = asyncio.get_running_loop()
+    return loop.run_in_executor(None, _do_get_avatar_url_from_database, user_id)
+
+
+def _do_get_avatar_url_from_database(user_id) -> Optional[bl_models.BilibiliUser]:
     try:
         with models.database.get_session() as session:
-            user = session.query(bl_models.BilibiliUser).filter(
-                bl_models.BilibiliUser.uid == user_id
+            user: bl_models.BilibiliUser = session.scalars(
+                sqlalchemy.select(bl_models.BilibiliUser).filter(
+                    bl_models.BilibiliUser.uid == user_id
+                )
+            ).one_or_none()
+            if user is None:
+                return None
+            return user
+    except sqlalchemy.exc.OperationalError:
+        # SQLite会锁整个文件，忽略就行
+        return None
+    except sqlalchemy.exc.SQLAlchemyError:
+        logger.exception('_do_get_avatar_url_from_database failed:')
+        return None
+
+
+def _update_avatar_cache_in_database(user_id, avatar_url) -> Awaitable[None]:
+    return asyncio.get_running_loop().run_in_executor(
+        None, _do_update_avatar_cache_in_database, user_id, avatar_url
+    )
+
+
+def _do_update_avatar_cache_in_database(user_id, avatar_url):
+    try:
+        with models.database.get_session() as session:
+            user = session.scalars(
+                sqlalchemy.select(bl_models.BilibiliUser).filter(
+                    bl_models.BilibiliUser.uid == user_id
+                )
             ).one_or_none()
             if user is None:
                 user = bl_models.BilibiliUser(
@@ -206,7 +176,377 @@ def _update_avatar_cache_in_database(user_id, avatar_url):
             user.update_time = datetime.datetime.now()
             session.commit()
     except (sqlalchemy.exc.OperationalError, sqlalchemy.exc.IntegrityError):
-        # SQLite会锁整个文件，忽略就行，另外还有多线程导致ID重复的问题
+        # SQLite会锁整个文件，忽略就行。另外还有多线程导致ID重复的问题，这里对一致性要求不高就没加for update
         pass
     except sqlalchemy.exc.SQLAlchemyError:
-        logger.exception('_update_avatar_cache_in_database failed:')
+        logger.exception('_do_update_avatar_cache_in_database failed:')
+
+
+def _get_avatar_url_from_web(user_id) -> Awaitable[Optional[str]]:
+    # 如果已有正在获取的future则返回，防止重复获取同一个uid
+    future = _uid_fetch_future_map.get(user_id, None)
+    if future is not None:
+        return future
+    # 否则创建一个获取任务
+    future = asyncio.get_running_loop().create_future()
+
+    task = FetchTask(
+        user_id=user_id,
+        future=future
+    )
+    if not _push_task(task):
+        future.set_result(None)
+        return future
+
+    _uid_fetch_future_map[user_id] = future
+    future.add_done_callback(lambda _future: _uid_fetch_future_map.pop(user_id, None))
+    return future
+
+
+def _push_task(task: FetchTask):
+    if not _has_available_avatar_fetcher():
+        return False
+
+    try:
+        _task_queue.put_nowait(task)
+        return True
+    except asyncio.QueueFull:
+        return False
+
+
+def _pop_task() -> Awaitable[FetchTask]:
+    return _task_queue.get()
+
+
+def _cancel_all_tasks_if_no_available_avatar_fetcher():
+    if _has_available_avatar_fetcher():
+        return
+
+    logger.warning('No available avatar fetcher')
+    while not _task_queue.empty():
+        task = _task_queue.get_nowait()
+        task.future.set_result(None)
+
+
+def _has_available_avatar_fetcher():
+    return any(fetcher.is_available for fetcher in _avatar_fetchers)
+
+
+class AvatarFetcher:
+    def __init__(self, query_interval):
+        self._query_interval = query_interval
+        self._be_available_event = asyncio.Event()
+        self._be_available_event.set()
+
+        self._cool_down_timer_handle = None
+
+    async def init(self):
+        asyncio.create_task(self._fetch_consumer())
+        return True
+
+    @property
+    def is_available(self):
+        return self._cool_down_timer_handle is None
+
+    def _on_availability_change(self):
+        if self.is_available:
+            self._be_available_event.set()
+        else:
+            self._be_available_event.clear()
+            _cancel_all_tasks_if_no_available_avatar_fetcher()
+
+    async def _fetch_consumer(self):
+        cls_name = type(self).__name__
+        while True:
+            try:
+                if not self.is_available:
+                    logger.info('%s waiting to become available', cls_name)
+                    await self._be_available_event.wait()
+                    logger.info('%s became available', cls_name)
+
+                task = await _pop_task()
+                # 为了简化代码，约定只会在_fetch_wrapper里变成不可用，所以获取task之后这里还是可用的
+                assert self.is_available
+
+                start_time = datetime.datetime.now()
+                await self._fetch_wrapper(task)
+                cost_time = (datetime.datetime.now() - start_time).total_seconds()
+
+                # 限制频率，防止被B站ban
+                await asyncio.sleep(self._query_interval - cost_time)
+            except Exception:  # noqa
+                logger.exception('%s error:', cls_name)
+
+    async def _fetch_wrapper(self, task: FetchTask) -> Optional[str]:
+        try:
+            avatar_url = await self._do_fetch(task.user_id)
+        except BaseException as e:
+            task.future.set_exception(e)
+            return None
+
+        task.future.set_result(avatar_url)
+        return avatar_url
+
+    async def _do_fetch(self, user_id) -> Optional[str]:
+        raise NotImplementedError
+
+    def _cool_down(self, sleep_time):
+        if self._cool_down_timer_handle is not None:
+            return
+
+        self._cool_down_timer_handle = asyncio.get_running_loop().call_later(
+            sleep_time, self._on_cool_down_timeout
+        )
+        self._on_availability_change()
+
+    def _on_cool_down_timeout(self):
+        self._cool_down_timer_handle = None
+        self._on_availability_change()
+
+
+class UserSpaceAvatarFetcher(AvatarFetcher):
+    # wbi密码表
+    WBI_KEY_INDEX_TABLE = [
+        46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35,
+        27, 43, 5, 49, 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13
+    ]
+
+    def __init__(self, query_interval):
+        super().__init__(query_interval)
+
+        # wbi鉴权口令
+        self._wbi_key = ''
+
+    async def _do_fetch(self, user_id) -> Optional[str]:
+        if self._wbi_key == '':
+            await self._refresh_wbi_key()
+            if self._wbi_key == '':
+                return None
+
+        try:
+            async with utils.request.http_session.get(
+                'https://api.bilibili.com/x/space/wbi/acc/info',
+                headers={
+                    **utils.request.BILIBILI_COMMON_HEADERS,
+                    'Origin': 'https://space.bilibili.com',
+                    'Referer': f'https://space.bilibili.com/{user_id}/'
+                },
+                params=self._add_wbi_sign({'mid': user_id}),
+            ) as r:
+                if r.status != 200:
+                    logger.warning(
+                        'UserSpaceAvatarFetcher failed to fetch avatar: status=%d %s uid=%d',
+                        r.status, r.reason, user_id
+                    )
+                    if r.status == 412:
+                        # 被B站ban了
+                        self._cool_down(3 * 60)
+                        await self._refresh_wbi_key()
+                    return None
+                data = await r.json()
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            return None
+
+        code = data['code']
+        if code != 0:
+            logger.info(
+                'UserSpaceAvatarFetcher failed to fetch avatar: code=%d %s uid=%d',
+                code, data['message'], user_id
+            )
+            if code == -401:
+                # 被B站ban了
+                self._cool_down(3 * 60)
+                await self._refresh_wbi_key()
+            elif code == -403:
+                # 签名错误
+                self._wbi_key = ''
+                await self._refresh_wbi_key()
+            return None
+
+        return process_avatar_url(data['data']['face'])
+
+    async def _refresh_wbi_key(self):
+        wbi_key = await self._get_wbi_key()
+        if wbi_key != '':
+            self._wbi_key = wbi_key
+
+    async def _get_wbi_key(self):
+        try:
+            async with utils.request.http_session.get(
+                'https://api.bilibili.com/nav',
+                headers=utils.request.BILIBILI_COMMON_HEADERS,
+            ) as r:
+                if r.status != 200:
+                    logger.warning('UserSpaceAvatarFetcher failed to get wbi key: status=%d %s', r.status, r.reason)
+                    return ''
+                data = await r.json()
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            logger.exception('UserSpaceAvatarFetcher failed to get wbi key:')
+            return ''
+
+        try:
+            wbi_img = data['data']['wbi_img']
+            img_key = wbi_img['img_url'].rpartition('/')[2].partition('.')[0]
+            sub_key = wbi_img['sub_url'].rpartition('/')[2].partition('.')[0]
+        except KeyError:
+            logger.warning('UserSpaceAvatarFetcher failed to get wbi key: data=%s', data)
+            return ''
+
+        shuffled_key = img_key + sub_key
+        wbi_key = []
+        for index in self.WBI_KEY_INDEX_TABLE:
+            if index < len(shuffled_key):
+                wbi_key.append(shuffled_key[index])
+        return ''.join(wbi_key)
+
+    def _add_wbi_sign(self, params: dict):
+        if self._wbi_key == '':
+            return params
+
+        wts = str(int(datetime.datetime.now().timestamp()))
+        params_to_sign = {**params, 'wts': wts}
+
+        # 按key字典序排序
+        params_to_sign = {
+            key: params_to_sign[key]
+            for key in sorted(params_to_sign.keys())
+        }
+        # 过滤一些字符
+        for key, value in params_to_sign.items():
+            value = ''.join(
+                ch
+                for ch in str(value)
+                if ch not in "!'()*"
+            )
+            params_to_sign[key] = value
+
+        str_to_sign = urllib.parse.urlencode(params_to_sign) + self._wbi_key
+        w_rid = hashlib.md5(str_to_sign.encode('utf-8')).hexdigest()
+        return {
+            **params,
+            'wts': wts,
+            'w_rid': w_rid
+        }
+
+
+class MedalAnchorAvatarFetcher(AvatarFetcher):
+    async def _do_fetch(self, user_id) -> Optional[str]:
+        try:
+            async with utils.request.http_session.get(
+                'https://api.live.bilibili.com/xlive/web-room/v1/index/getDanmuMedalAnchorInfo',
+                headers={
+                    **utils.request.BILIBILI_COMMON_HEADERS,
+                    'Origin': 'https://live.bilibili.com',
+                    'Referer': 'https://live.bilibili.com/'
+                },
+                params={
+                    'ruid': user_id,
+                    'token': '',
+                    'platform': 'web',
+                    'jsonp': 'jsonp'
+                }
+            ) as r:
+                if r.status != 200:
+                    logger.warning(
+                        'MedalAnchorAvatarFetcher failed to fetch avatar: status=%d %s uid=%d',
+                        r.status, r.reason, user_id
+                    )
+                    if r.status == 412:
+                        # 被B站ban了
+                        self._cool_down(3 * 60)
+                    return None
+                data = await r.json()
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            return None
+
+        code = data['code']
+        if code != 0:
+            # 这里虽然失败但不会被ban一段时间
+            logger.info(
+                'MedalAnchorAvatarFetcher failed to fetch avatar: code=%d %s uid=%d',
+                code, data['message'], user_id
+            )
+            return None
+
+        return process_avatar_url(data['data']['rface'])
+
+
+class UserCardAvatarFetcher(AvatarFetcher):
+    async def _do_fetch(self, user_id) -> Optional[str]:
+        try:
+            async with utils.request.http_session.get(
+                'https://api.bilibili.com/x/web-interface/card',
+                headers={
+                    **utils.request.BILIBILI_COMMON_HEADERS,
+                    'Origin': 'https://t.bilibili.com',
+                    'Referer': 'https://t.bilibili.com/'
+                },
+                params={
+                    'mid': user_id,
+                    'photo': 'true'
+                }
+            ) as r:
+                if r.status != 200:
+                    logger.warning(
+                        'UserCardAvatarFetcher failed to fetch avatar: status=%d %s uid=%d',
+                        r.status, r.reason, user_id
+                    )
+                    if r.status == 412:
+                        # 被B站ban了
+                        self._cool_down(3 * 60)
+                    return None
+                data = await r.json()
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            return None
+
+        code = data['code']
+        if code != 0:
+            # 这里虽然失败但不会被ban一段时间
+            logger.info(
+                'UserCardAvatarFetcher failed to fetch avatar: code=%d %s uid=%d',
+                code, data['message'], user_id
+            )
+            return None
+
+        return process_avatar_url(data['data']['card']['face'])
+
+
+class GameUserCenterAvatarFetcher(AvatarFetcher):
+    async def _do_fetch(self, user_id) -> Optional[str]:
+        try:
+            async with utils.request.http_session.get(
+                'https://line3-h5-mobile-api.biligame.com/game/center/h5/user/space/info',
+                headers={
+                    **utils.request.BILIBILI_COMMON_HEADERS,
+                    'Origin': 'https://app.biligame.com',
+                    'Referer': 'https://app.biligame.com/'
+                },
+                params={
+                    'uid': user_id,
+                    'sdk_type': '1'
+                }
+            ) as r:
+                if r.status != 200:
+                    # 这个接口经常502
+                    logger.info(
+                        'GameUserCenterAvatarFetcher failed to fetch avatar: status=%d %s uid=%d',
+                        r.status, r.reason, user_id
+                    )
+                    if r.status == 412:
+                        # 被B站ban了
+                        self._cool_down(3 * 60)
+                    return None
+                data = await r.json()
+        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+            return None
+
+        code = data['code']
+        if code != 0:
+            # 这里虽然失败但不会被ban一段时间
+            logger.info(
+                'GameUserCenterAvatarFetcher failed to fetch avatar: code=%d %s uid=%d',
+                code, data['message'], user_id
+            )
+            return None
+
+        return process_avatar_url(data['data']['face'])

services/chat.py

@@ -1,11 +1,15 @@
 # -*- coding: utf-8 -*-
 import asyncio
+import base64
+import binascii
+import json
 import logging
 import uuid
 from typing import *
 
 import api.chat
 import blivedm.blivedm as blivedm
+import blivedm.blivedm.models.pb as blivedm_pb
 import config
 import services.avatar
 import services.translate
@@ -39,7 +43,7 @@ class LiveClientManager:
         logger.info('room=%d creating live client', room_id)
         self._live_clients[room_id] = live_client = LiveClient(room_id)
         live_client.add_handler(_live_msg_handler)
-        asyncio.ensure_future(self._init_live_client(live_client))
+        asyncio.create_task(self._init_live_client(live_client))
         logger.info('room=%d live client created, %d live clients', room_id, len(self._live_clients))
 
     async def _init_live_client(self, live_client: 'LiveClient'):
@@ -56,7 +60,7 @@ class LiveClientManager:
             return
         logger.info('room=%d removing live client', room_id)
         live_client.remove_handler(_live_msg_handler)
-        asyncio.ensure_future(live_client.stop_and_close())
+        asyncio.create_task(live_client.stop_and_close())
         logger.info('room=%d live client removed, %d live clients', room_id, len(self._live_clients))
 
         client_room_manager.del_room(room_id)
@@ -130,7 +134,7 @@ class ClientRoomManager:
 
     def delay_del_room(self, room_id, timeout):
         self._clear_delay_del_timer(room_id)
-        self._delay_del_timer_handles[room_id] = asyncio.get_event_loop().call_later(
+        self._delay_del_timer_handles[room_id] = asyncio.get_running_loop().call_later(
             timeout, self._on_delay_del_room, room_id
         )
 
@@ -203,6 +207,19 @@ class LiveMsgHandler(blivedm.BaseHandler):
     # 重新定义XXX_callback是为了减少对字段名的依赖，防止B站改字段名
     def __danmu_msg_callback(self, client: LiveClient, command: dict):
         info = command['info']
+        dm_v2 = command.get('dm_v2', '')
+
+        proto: Optional[blivedm_pb.SimpleDm] = None
+        if dm_v2 != '':
+            try:
+                proto = blivedm_pb.SimpleDm.loads(base64.b64decode(dm_v2))
+            except (binascii.Error, KeyError, TypeError, ValueError):
+                pass
+        if proto is not None:
+            face = proto.user.face
+        else:
+            face = ''
+
         if len(info[3]) != 0:
             medal_level = info[3][0]
             medal_room_id = info[3][3]
@@ -215,11 +232,13 @@ class LiveMsgHandler(blivedm.BaseHandler):
             msg_type=info[0][9],
             dm_type=info[0][12],
             emoticon_options=info[0][13],
+            mode_info=info[0][15],
 
             msg=info[1],
 
             uid=info[2][0],
             uname=info[2][1],
+            face=face,
             admin=info[2][2],
             urank=info[2][5],
             mobile_verify=info[2][6],
@@ -263,7 +282,7 @@ class LiveMsgHandler(blivedm.BaseHandler):
             price=data['price'],
             message=data['message'],
             start_time=data['start_time'],
-            id_=data['id'],
+            id=data['id'],
             uid=data['uid'],
             uname=data['user_info']['uname'],
             face=data['user_info']['face'],
@@ -279,11 +298,15 @@ class LiveMsgHandler(blivedm.BaseHandler):
     }
 
     async def _on_danmaku(self, client: LiveClient, message: blivedm.DanmakuMessage):
-        asyncio.ensure_future(self.__on_danmaku(client, message))
+        asyncio.create_task(self.__on_danmaku(client, message))
 
     async def __on_danmaku(self, client: LiveClient, message: blivedm.DanmakuMessage):
-        # 先异步调用再获取房间，因为返回时房间可能已经不存在了
-        avatar_url = await services.avatar.get_avatar_url(message.uid)
+        avatar_url = message.face
+        if avatar_url != '':
+            services.avatar.update_avatar_cache_if_expired(message.uid, avatar_url)
+        else:
+            # 先异步调用再获取房间，因为返回时房间可能已经不存在了
+            avatar_url = await services.avatar.get_avatar_url(message.uid)
 
         room = client_room_manager.get_room(client.tmp_room_id)
         if room is None:
@@ -307,7 +330,9 @@ class LiveMsgHandler(blivedm.BaseHandler):
             content_type = api.chat.ContentType.TEXT
             content_type_params = None
 
-        need_translate = self._need_translate(message.msg, room)
+        text_emoticons = self._parse_text_emoticons(message)
+
+        need_translate = content_type != api.chat.ContentType.EMOTICON and self._need_translate(message.msg, room)
         if need_translate:
             translation = services.translate.get_translation_from_cache(message.msg)
             if translation is None:
@@ -335,15 +360,32 @@ class LiveMsgHandler(blivedm.BaseHandler):
             translation=translation,
             content_type=content_type,
             content_type_params=content_type_params,
+            text_emoticons=text_emoticons,
         ))
 
         if need_translate:
             await self._translate_and_response(message.msg, room.room_id, msg_id)
 
+    @staticmethod
+    def _parse_text_emoticons(message: blivedm.DanmakuMessage):
+        try:
+            extra = json.loads(message.mode_info['extra'])
+            # {"[dog]":{"emoticon_id":208,"emoji":"[dog]","descript":"[dog]","url":"http://i0.hdslb.com/bfs/live/4428c8
+            # 4e694fbf4e0ef6c06e958d9352c3582740.png","width":20,"height":20,"emoticon_unique":"emoji_208","count":1}}
+            emoticons = extra['emots']
+            if emoticons is None:
+                return []
+            res = [
+                (emoticon['descript'], emoticon['url'])
+                for emoticon in emoticons.values()
+            ]
+            return res
+        except (json.JSONDecodeError, TypeError, KeyError):
+            return []
+
     async def _on_gift(self, client: LiveClient, message: blivedm.GiftMessage):
         avatar_url = services.avatar.process_avatar_url(message.face)
-        # 服务器白给的头像URL，直接缓存
-        services.avatar.update_avatar_cache(message.uid, avatar_url)
+        services.avatar.update_avatar_cache_if_expired(message.uid, avatar_url)
 
         # 丢人
         if message.coin_type != 'gold':
@@ -364,7 +406,7 @@ class LiveMsgHandler(blivedm.BaseHandler):
         })
 
     async def _on_buy_guard(self, client: LiveClient, message: blivedm.GuardBuyMessage):
-        asyncio.ensure_future(self.__on_buy_guard(client, message))
+        asyncio.create_task(self.__on_buy_guard(client, message))
 
     @staticmethod
     async def __on_buy_guard(client: LiveClient, message: blivedm.GuardBuyMessage):
@@ -385,8 +427,7 @@ class LiveMsgHandler(blivedm.BaseHandler):
 
     async def _on_super_chat(self, client: LiveClient, message: blivedm.SuperChatMessage):
         avatar_url = services.avatar.process_avatar_url(message.face)
-        # 服务器白给的头像URL，直接缓存
-        services.avatar.update_avatar_cache(message.uid, avatar_url)
+        services.avatar.update_avatar_cache_if_expired(message.uid, avatar_url)
 
         room = client_room_manager.get_room(client.tmp_room_id)
         if room is None:
@@ -415,14 +456,16 @@ class LiveMsgHandler(blivedm.BaseHandler):
         })
 
         if need_translate:
-            asyncio.ensure_future(self._translate_and_response(message.message, room.room_id, msg_id))
+            asyncio.create_task(self._translate_and_response(
+                message.message, room.room_id, msg_id, services.translate.Priority.HIGH
+            ))
 
     async def _on_super_chat_delete(self, client: LiveClient, message: blivedm.SuperChatDeleteMessage):
         room = client_room_manager.get_room(client.tmp_room_id)
         if room is None:
             return
 
-        room.send_cmd_data(api.chat.Command.ADD_SUPER_CHAT, {
+        room.send_cmd_data(api.chat.Command.DEL_SUPER_CHAT, {
             'ids': list(map(str, message.ids))
         })
 
@@ -437,8 +480,8 @@ class LiveMsgHandler(blivedm.BaseHandler):
         )
 
     @staticmethod
-    async def _translate_and_response(text, room_id, msg_id):
-        translation = await services.translate.translate(text)
+    async def _translate_and_response(text, room_id, msg_id, priority=services.translate.Priority.NORMAL):
+        translation = await services.translate.translate(text, priority)
         if translation is None:
             return
 

services/translate.py

@@ -1,7 +1,9 @@
 # -*- coding: utf-8 -*-
 import asyncio
 import base64
+import dataclasses
 import datetime
+import enum
 import functools
 import hashlib
 import hmac
@@ -14,6 +16,7 @@ from typing import *
 import Crypto.Cipher.AES as cry_aes  # noqa
 import Crypto.Util.Padding as cry_pad  # noqa
 import aiohttp
+import cachetools
 
 import config
 import utils.request
@@ -27,13 +30,33 @@ NO_TRANSLATE_TEXTS = {
 
 _translate_providers: List['TranslateProvider'] = []
 # text -> res
-_translate_cache: Dict[str, str] = {}
+_translate_cache: Optional[cachetools.LRUCache] = None
 # 正在翻译的Future，text -> Future
 _text_future_map: Dict[str, asyncio.Future] = {}
+# 正在翻译的任务队列，索引是优先级
+_task_queues: List['asyncio.Queue[TranslateTask]'] = []
+
+
+class Priority(enum.IntEnum):
+    HIGH = 0
+    NORMAL = 1
+
+
+@dataclasses.dataclass
+class TranslateTask:
+    priority: Priority
+    text: str
+    future: 'asyncio.Future[Optional[str]]'
+    remain_retry_count: int
 
 
 def init():
-    asyncio.ensure_future(_do_init())
+    cfg = config.get_config()
+    global _translate_cache, _task_queues
+    _translate_cache = cachetools.LRUCache(cfg.translation_cache_size)
+    # 总队列长度会超过translate_max_queue_size，不用这么严格
+    _task_queues = [asyncio.Queue(cfg.translate_max_queue_size) for _ in range(len(Priority))]
+    asyncio.get_event_loop().create_task(_do_init())
 
 
 async def _do_init():
@@ -54,21 +77,17 @@ def create_translate_provider(cfg):
     type_ = cfg['type']
     if type_ == 'TencentTranslateFree':
         return TencentTranslateFree(
-            cfg['query_interval'], cfg['max_queue_size'], cfg['source_language'],
-            cfg['target_language']
+            cfg['query_interval'], cfg['source_language'], cfg['target_language']
         )
-    elif type_ == 'BilibiliTranslateFree':
-        return BilibiliTranslateFree(cfg['query_interval'], cfg['max_queue_size'])
     elif type_ == 'TencentTranslate':
         return TencentTranslate(
-            cfg['query_interval'], cfg['max_queue_size'], cfg['source_language'],
-            cfg['target_language'], cfg['secret_id'], cfg['secret_key'],
-            cfg['region']
+            cfg['query_interval'], cfg['source_language'], cfg['target_language'],
+            cfg['secret_id'], cfg['secret_key'], cfg['region']
         )
     elif type_ == 'BaiduTranslate':
         return BaiduTranslate(
-            cfg['query_interval'], cfg['max_queue_size'], cfg['source_language'],
-            cfg['target_language'], cfg['app_id'], cfg['secret']
+            cfg['query_interval'], cfg['source_language'], cfg['target_language'],
+            cfg['app_id'], cfg['secret']
         )
     return None
 
@@ -95,14 +114,14 @@ def get_translation_from_cache(text):
     return _translate_cache.get(key, None)
 
 
-def translate(text) -> Awaitable[Optional[str]]:
+def translate(text, priority=Priority.NORMAL) -> Awaitable[Optional[str]]:
     key = text.strip().lower()
     # 如果已有正在翻译的future则返回，防止重复翻译
     future = _text_future_map.get(key, None)
     if future is not None:
         return future
     # 否则创建一个翻译任务
-    future = asyncio.get_event_loop().create_future()
+    future = asyncio.get_running_loop().create_future()
 
     # 查缓存
     res = _translate_cache.get(key, None)
@@ -110,25 +129,18 @@ def translate(text) -> Awaitable[Optional[str]]:
         future.set_result(res)
         return future
 
-    # 负载均衡，找等待时间最少的provider
-    min_wait_time = None
-    min_wait_time_provider = None
-    for provider in _translate_providers:
-        if not provider.is_available:
-            continue
-        wait_time = provider.wait_time
-        if min_wait_time is None or wait_time < min_wait_time:
-            min_wait_time = wait_time
-            min_wait_time_provider = provider
-
-    # 没有可用的
-    if min_wait_time_provider is None:
+    task = TranslateTask(
+        priority=priority,
+        text=text,
+        future=future,
+        remain_retry_count=3 if priority == Priority.HIGH else 1
+    )
+    if not _push_task(task):
         future.set_result(None)
         return future
 
     _text_future_map[key] = future
     future.add_done_callback(functools.partial(_on_translate_done, key))
-    min_wait_time_provider.translate(text, future)
     return future
 
 
@@ -142,76 +154,149 @@ def _on_translate_done(key, future):
     if res is None:
         return
     _translate_cache[key] = res
-    cfg = config.get_config()
-    while len(_translate_cache) > cfg.translation_cache_size:
-        _translate_cache.pop(next(iter(_translate_cache)), None)
+
+
+def _push_task(task: TranslateTask):
+    if not _has_available_translate_provider():
+        return False
+
+    queue = _task_queues[task.priority]
+    if not queue.full():
+        queue.put_nowait(task)
+        return True
+
+    if task.priority != Priority.HIGH:
+        return False
+
+    # 高优先级的尝试降级，挤掉低优先级的任务
+    queue = _task_queues[Priority.NORMAL]
+    if queue.full():
+        lower_task = queue.get_nowait()
+        lower_task.future.set_result(None)
+    queue.put_nowait(task)
+    return True
+
+
+async def _pop_task() -> TranslateTask:
+    # 按优先级遍历，看是否已经有任务
+    for queue in _task_queues:
+        if not queue.empty():
+            return queue.get_nowait()
+
+    done_future_set, pending_future_set = await asyncio.wait(
+        [asyncio.create_task(queue.get()) for queue in _task_queues],
+        return_when=asyncio.FIRST_COMPLETED
+    )
+    for future in pending_future_set:
+        future.cancel()
+
+    # 如果有多个队列都取到任务了，只返回优先级最高的那一个，剩下的放回队列
+    assert len(done_future_set) != 0
+    tasks = [await future for future in done_future_set]
+    if len(tasks) > 1:
+        tasks.sort(key=lambda task_: task_.priority)
+
+    res = None
+    for task in tasks:
+        if res is None:
+            res = task
+            continue
+
+        if not _push_task(task):
+            task.future.set_result(None)
+    return res
+
+
+def _cancel_all_tasks_if_no_available_translate_provider():
+    if _has_available_translate_provider():
+        return
+
+    logger.warning('No available translate provider')
+    for queue in _task_queues:
+        while not queue.empty():
+            task = queue.get_nowait()
+            task.future.set_result(None)
+
+
+def _has_available_translate_provider():
+    return any(provider.is_available for provider in _translate_providers)
 
 
 class TranslateProvider:
-    async def init(self):
-        return True
-
-    @property
-    def is_available(self):
-        return True
-
-    @property
-    def wait_time(self):
-        return 0
-
-    def translate(self, text, future):
-        raise NotImplementedError
-
-
-class FlowControlTranslateProvider(TranslateProvider):
-    def __init__(self, query_interval, max_queue_size):
+    def __init__(self, query_interval):
         self._query_interval = query_interval
-        # (text, future)
-        self._text_queue = asyncio.Queue(max_queue_size)
+        self._be_available_event = asyncio.Event()
+        self._be_available_event.set()
 
     async def init(self):
-        asyncio.ensure_future(self._translate_consumer())
+        asyncio.create_task(self._translate_consumer())
         return True
 
     @property
     def is_available(self):
-        return not self._text_queue.full()
+        return True
 
-    @property
-    def wait_time(self):
-        return self._text_queue.qsize() * self._query_interval
-
-    def translate(self, text, future):
-        try:
-            self._text_queue.put_nowait((text, future))
-        except asyncio.QueueFull:
-            future.set_result(None)
+    def _on_availability_change(self):
+        if self.is_available:
+            self._be_available_event.set()
+        else:
+            self._be_available_event.clear()
+            _cancel_all_tasks_if_no_available_translate_provider()
 
     async def _translate_consumer(self):
+        cls_name = type(self).__name__
         while True:
             try:
-                text, future = await self._text_queue.get()
-                asyncio.ensure_future(self._translate_coroutine(text, future))
+                if not self.is_available:
+                    logger.info('%s waiting to become available', cls_name)
+                    await self._be_available_event.wait()
+                    logger.info('%s became available', cls_name)
+
+                task = await _pop_task()
+                # 为了简化代码，约定只会在_translate_wrapper里变成不可用，所以获取task之后这里还是可用的
+                assert self.is_available
+
+                start_time = datetime.datetime.now()
+                await self._translate_wrapper(task)
+                cost_time = (datetime.datetime.now() - start_time).total_seconds()
+
                 # 频率限制
-                await asyncio.sleep(self._query_interval)
+                await asyncio.sleep(self._query_interval - cost_time)
             except Exception:  # noqa
-                logger.exception('FlowControlTranslateProvider error:')
+                logger.exception('%s error:', cls_name)
 
-    async def _translate_coroutine(self, text, future):
+    async def _translate_wrapper(self, task: TranslateTask) -> Optional[str]:
         try:
-            res = await self._do_translate(text)
+            exc = None
+            task.remain_retry_count -= 1
+            res = await self._do_translate(task.text)
         except BaseException as e:
-            future.set_exception(e)
-        else:
-            future.set_result(res)
+            exc = e
+            res = None
+        if res is not None:
+            task.future.set_result(res)
+            return res
 
-    async def _do_translate(self, text):
+        if task.remain_retry_count > 0:
+            # 还可以重试则放回队列
+            if not _push_task(task):
+                task.future.set_result(None)
+        else:
+            # 否则设置异常或None结果
+            if exc is not None:
+                task.future.set_exception(exc)
+            else:
+                task.future.set_result(None)
+        return None
+
+    async def _do_translate(self, text) -> Optional[str]:
         raise NotImplementedError
 
 
-class TencentTranslateFree(FlowControlTranslateProvider):
-    def __init__(self, query_interval, max_queue_size, source_language, target_language):
-        super().__init__(query_interval, max_queue_size)
+class TencentTranslateFree(TranslateProvider):
+    def __init__(self, query_interval, source_language, target_language):
+        super().__init__(query_interval)
+        self._be_available_event.clear()  # _do_init之后才可用
         self._source_language = source_language
         self._target_language = target_language
 
@@ -226,9 +311,7 @@ class TencentTranslateFree(FlowControlTranslateProvider):
     async def init(self):
         if not await super().init():
             return False
-        if not await self._do_init():
-            return False
-        self._reinit_future = asyncio.ensure_future(self._reinit_coroutine())
+        self._reinit_future = asyncio.create_task(self._reinit_coroutine())
         return True
 
     async def _do_init(self):
@@ -246,7 +329,7 @@ class TencentTranslateFree(FlowControlTranslateProvider):
                     self._server_time_delta = int((datetime.datetime.now().timestamp() - server_time) * 1000)
                 except (KeyError, ValueError):
                     self._server_time_delta = 0
-        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+        except (aiohttp.ClientError, asyncio.TimeoutError):
             logger.exception('TencentTranslateFree init error:')
             return False
 
@@ -281,7 +364,7 @@ class TencentTranslateFree(FlowControlTranslateProvider):
                                    reauthuri, r.status, r.reason)
                     return False
                 data = await r.json()
-        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+        except (aiohttp.ClientError, asyncio.TimeoutError):
             logger.exception('TencentTranslateFree init error:')
             return False
 
@@ -298,14 +381,24 @@ class TencentTranslateFree(FlowControlTranslateProvider):
         self._uc_iv = uc_iv
         self._qtv = qtv
         self._qtk = qtk
+
+        self._on_availability_change()
         return True
 
     async def _reinit_coroutine(self):
         try:
             while True:
-                await asyncio.sleep(30)
                 logger.debug('TencentTranslateFree reinit')
-                asyncio.ensure_future(self._do_init())
+                start_time = datetime.datetime.now()
+                try:
+                    await self._do_init()
+                except asyncio.CancelledError:
+                    raise
+                except BaseException:  # noqa
+                    pass
+                cost_time = (datetime.datetime.now() - start_time).total_seconds()
+
+                await asyncio.sleep(30 - cost_time)
         except asyncio.CancelledError:
             pass
 
@@ -313,20 +406,15 @@ class TencentTranslateFree(FlowControlTranslateProvider):
     def is_available(self):
         return '' not in (self._uc_key, self._uc_iv, self._qtv, self._qtk) and super().is_available
 
-    async def _translate_coroutine(self, text, future):
-        try:
-            res = await self._do_translate(text)
-        except BaseException as e:
-            future.set_exception(e)
-            self._on_fail()
-            return
-        future.set_result(res)
-        if res is None:
-            self._on_fail()
-        else:
+    async def _translate_wrapper(self, task: TranslateTask) -> Optional[str]:
+        res = await super()._translate_wrapper(task)
+        if res is not None:
             self._fail_count = 0
+        else:
+            self._on_fail()
+        return res
 
-    async def _do_translate(self, text):
+    async def _do_translate(self, text) -> Optional[str]:
         try:
             async with utils.request.http_session.post(
                 'https://fanyi.qq.com/api/translate',
@@ -347,7 +435,7 @@ class TencentTranslateFree(FlowControlTranslateProvider):
                     return None
                 self._update_uc_key(r)
                 data = await r.json()
-        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
+        except (aiohttp.ClientError, asyncio.TimeoutError):
             return None
         if data['errCode'] != 0:
             logger.warning('TencentTranslateFree failed: %d %s', data['errCode'], data['errMsg'])
@@ -355,7 +443,7 @@ class TencentTranslateFree(FlowControlTranslateProvider):
         res = ''.join(record['targetText'] for record in data['translate']['records'])
         if res == '' and text.strip() != '':
             # qtv、qtk过期
-            logger.warning('TencentTranslateFree result is empty %s', data)
+            logger.info('TencentTranslateFree result is empty %s', data)
             return None
         return res
 
@@ -404,8 +492,8 @@ class TencentTranslateFree(FlowControlTranslateProvider):
 
     def _on_fail(self):
         self._fail_count += 1
-        # 为了可靠性，连续失败10次时冷却直到下次重新init
-        if self._fail_count >= 10:
+        # 为了可靠性，连续失败5次时冷却直到下次重新init
+        if self._fail_count >= 5:
             self._cool_down()
 
     def _cool_down(self):
@@ -415,39 +503,13 @@ class TencentTranslateFree(FlowControlTranslateProvider):
         self._qtv = self._qtk = ''
         self._fail_count = 0
 
-
-class BilibiliTranslateFree(FlowControlTranslateProvider):
-    def __init__(self, query_interval, max_queue_size):
-        super().__init__(query_interval, max_queue_size)
-
-    async def _do_translate(self, text):
-        try:
-            async with utils.request.http_session.get(
-                'https://api.live.bilibili.com/av/v1/SuperChat/messageTranslate',
-                params={
-                    'room_id': '21396545',
-                    'ruid': '407106379',
-                    'parent_area_id': '9',
-                    'area_id': '371',
-                    'msg': text
-                }
-            ) as r:
-                if r.status != 200:
-                    logger.warning('BilibiliTranslateFree request failed: status=%d %s', r.status, r.reason)
-                    return None
-                data = await r.json()
-        except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
-            return None
-        if data['code'] != 0:
-            logger.warning('BilibiliTranslateFree failed: %d %s', data['code'], data['msg'])
-            return None
-        return data['data']['message_trans']
+        self._on_availability_change()
 
 
-class TencentTranslate(FlowControlTranslateProvider):
-    def __init__(self, query_interval, max_queue_size, source_language, target_language,
+class TencentTranslate(TranslateProvider):
+    def __init__(self, query_interval, source_language, target_language,
                  secret_id, secret_key, region):
-        super().__init__(query_interval, max_queue_size)
+        super().__init__(query_interval)
         self._source_language = source_language
         self._target_language = target_language
         self._secret_id = secret_id
@@ -460,7 +522,7 @@ class TencentTranslate(FlowControlTranslateProvider):
     def is_available(self):
         return self._cool_down_timer_handle is None and super().is_available
 
-    async def _do_translate(self, text):
+    async def _do_translate(self, text) -> Optional[str]:
         try:
             async with self._request_tencent_cloud(
                 'TextTranslate',
@@ -545,18 +607,20 @@ class TencentTranslate(FlowControlTranslateProvider):
             # 需要手动处理，等5分钟
             sleep_time = 5 * 60
         if sleep_time != 0:
-            self._cool_down_timer_handle = asyncio.get_event_loop().call_later(
+            self._cool_down_timer_handle = asyncio.get_running_loop().call_later(
                 sleep_time, self._on_cool_down_timeout
             )
+            self._on_availability_change()
 
     def _on_cool_down_timeout(self):
         self._cool_down_timer_handle = None
+        self._on_availability_change()
 
 
-class BaiduTranslate(FlowControlTranslateProvider):
-    def __init__(self, query_interval, max_queue_size, source_language, target_language,
+class BaiduTranslate(TranslateProvider):
+    def __init__(self, query_interval, source_language, target_language,
                  app_id, secret):
-        super().__init__(query_interval, max_queue_size)
+        super().__init__(query_interval)
         self._source_language = source_language
         self._target_language = target_language
         self._app_id = app_id
@@ -568,7 +632,7 @@ class BaiduTranslate(FlowControlTranslateProvider):
     def is_available(self):
         return self._cool_down_timer_handle is None and super().is_available
 
-    async def _do_translate(self, text):
+    async def _do_translate(self, text) -> Optional[str]:
         try:
             async with utils.request.http_session.post(
                 'https://fanyi-api.baidu.com/api/trans/vip/translate',
@@ -607,9 +671,11 @@ class BaiduTranslate(FlowControlTranslateProvider):
             # 账户余额不足，需要手动处理，等5分钟
             sleep_time = 5 * 60
         if sleep_time != 0:
-            self._cool_down_timer_handle = asyncio.get_event_loop().call_later(
+            self._cool_down_timer_handle = asyncio.get_running_loop().call_later(
                 sleep_time, self._on_cool_down_timeout
             )
+            self._on_availability_change()
 
     def _on_cool_down_timeout(self):
         self._cool_down_timer_handle = None
+        self._on_availability_change()

update.py

@@ -5,11 +5,11 @@ import aiohttp
 
 import utils.request
 
-VERSION = 'v1.6.0'
+VERSION = 'v1.7.0'
 
 
 def check_update():
-    asyncio.ensure_future(_do_check_update())
+    asyncio.get_event_loop().create_task(_do_check_update())
 
 
 async def _do_check_update():

utils/request.py

@@ -1,4 +1,24 @@
 # -*- coding: utf-8 -*-
+import asyncio
+from typing import *
+
 import aiohttp
 
-http_session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10))
+# 不带这堆头部有时候也能成功请求，但是带上后成功的概率更高
+BILIBILI_COMMON_HEADERS = {
+    'Origin': 'https://www.bilibili.com',
+    'Referer': 'https://www.bilibili.com/',
+    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)'
+                  ' Chrome/114.0.0.0 Safari/537.36'
+}
+
+http_session: Optional[aiohttp.ClientSession] = None
+
+
+def init():
+    # ClientSession要在异步函数中创建
+    async def do_init():
+        global http_session
+        http_session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10))
+
+    asyncio.get_event_loop().run_until_complete(do_init())