🔖 0.4.0.post1

* 忽略 .idea 文件夹

* 新增查分段指令

* 修复未知段位可能导致程序无限死循环的异常

* 我 keys 呢

* 🎨 整理 import

* 🎨 规范引号

* 🐛 记得await

* ⚡️ 一些优化

* 🐛 多个逗号

* 🚨 修正typing hint

* 🐛 返回值是 tuple 哦

* 🐛 少个逗号

* 🐛 你得删前缀啊

* 🐛 怎么能用 is 呢

* 🐛 记得await

* 试图匹配查询格式

* 💬 小改返回消息样式

* 🎨 修改变量名

* 🐛 修复查询大写问题

* 🐛 使用 get_db 获取数据库对象

---------

Co-authored-by: scdhh <wallfjjd@gmail.com>

.github/dependabot.yml

@@ -0,0 +1,12 @@
+# To get started with Dependabot version updates, you'll need to specify which
+# package ecosystems to update and where the package manifests are located.
+# Please see the documentation for all configuration options:
+# https://docs.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
+
+version: 2
+updates:
+  - package-ecosystem: "pip" # See documentation for possible values
+    directory: "/" # Location of package manifests
+    target-branch: "dev"
+    schedule:
+      interval: "daily"

.github/workflows/Release.yml

@@ -0,0 +1,25 @@
+name: Release CI
+
+on:
+  push:
+    tags:
+      - "*"
+
+jobs:
+  release:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-python@v4
+        with:
+          python-version: '3.10'
+      - name: Install Poetry
+        shell: bash
+        run: |
+          pip install poetry
+      - name: Build
+        shell: bash
+        run: |
+          poetry install
+          poetry env use python
+          poetry publish --build -u ${{ secrets.USERNAME }} -p ${{ secrets.PASSWORD }} -n

.github/workflows/codeql-analysis.yml

@@ -0,0 +1,72 @@
+# For most projects, this workflow file will not need changing; you simply need
+# to commit it to your repository.
+#
+# You may wish to alter this file to override the set of languages analyzed,
+# or to provide custom queries or build logic.
+#
+# ******** NOTE ********
+# We have attempted to detect the languages in your repository. Please check
+# the `language` matrix defined below to confirm you have the correct set of
+# supported CodeQL languages.
+#
+name: "CodeQL"
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    # The branches below must be a subset of the branches above
+    branches: [ main ]
+  schedule:
+    - cron: '17 6 * * 5'
+
+jobs:
+  analyze:
+    name: Analyze
+    runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      security-events: write
+
+    strategy:
+      fail-fast: false
+      matrix:
+        language: [ 'python' ]
+        # CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python', 'ruby' ]
+        # Learn more about CodeQL language support at https://aka.ms/codeql-docs/language-support
+
+    steps:
+    - name: Checkout repository
+      uses: actions/checkout@v3
+
+    # Initializes the CodeQL tools for scanning.
+    - name: Initialize CodeQL
+      uses: github/codeql-action/init@v2
+      with:
+        languages: ${{ matrix.language }}
+        # If you wish to specify custom queries, you can do so here or in a config file.
+        # By default, queries listed here will override any specified in a config file.
+        # Prefix the list here with "+" to use these queries and those in the config file.
+        
+        # Details on CodeQL's query packs refer to : https://docs.github.com/en/code-security/code-scanning/automatically-scanning-your-code-for-vulnerabilities-and-errors/configuring-code-scanning#using-queries-in-ql-packs
+        # queries: security-extended,security-and-quality
+
+        
+    # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
+    # If this step fails, then you should remove it and run the build manually (see below)
+    - name: Autobuild
+      uses: github/codeql-action/autobuild@v2
+
+    # ℹ️ Command-line programs to run using the OS shell.
+    # 📚 See https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_idstepsrun
+
+    #   If the Autobuild fails above, remove it and uncomment the following three lines. 
+    #   modify them (or add more) to build your code if your project, please refer to the EXAMPLE below for guidance.
+
+    # - run: |
+    #   echo "Run, Build Application using script"
+    #   ./location_of_script_within_repo/buildscript.sh
+
+    - name: Perform CodeQL Analysis
+      uses: github/codeql-action/analyze@v2

.gitignore

@@ -1 +1,6 @@
+.idea
 dist
+test*
+Untitled*
+*copy*
+.vscode

.pylintrc

@@ -0,0 +1,31 @@
+[MAIN]
+# A comma-separated list of package or module names from where C extensions may
+# be loaded. Extensions are loading into the active Python interpreter and may
+# run arbitrary code.
+extension-pkg-allow-list=lxml, pydantic
+
+# A comma-separated list of package or module names from where C extensions may
+# be loaded. Extensions are loading into the active Python interpreter and may
+# run arbitrary code. (This is an alternative name to extension-pkg-allow-list
+# for backward compatibility.)
+extension-pkg-whitelist=lxml, pydantic
+
+# List of module names for which member attributes should not be checked
+# (useful for modules/projects where namespaces are manipulated during runtime
+# and thus existing member attributes cannot be deduced by static analysis). It
+# supports qualified module names, as well as Unix pattern matching.
+ignored-modules=ujson
+
+disable=
+    C0114, # missing-module-docstring
+
+[MESSAGES CONTROL]
+# Enable the message, report, category or checker with the given id(s). You can
+# either give multiple identifier separated by comma (,) or put this option
+# multiple time (only on the command line, not in the configuration file where
+# it should appear only once). See also the "--disable" option for examples.
+enable=c-extension-no-member
+
+[FORMAT]
+# Maximum number of characters on a single line.
+max-line-length=120

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2022 scdhh
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -5,18 +5,31 @@ TETRIS Stats
 目前支持
 * [TETR.IO](https://tetr.io/)
 * [茶服](https://teatube.cn/tos/)
-
-计划支持
 * [TOP](http://tetrisonline.pl/)
 
 安装
 ----
 
-* 使用 pip
+* 使用 nb-cli
+```
+nb plugin install nonebot-plugin-tetris-stats
+```
 
+* 使用 pip
 ```
 pip install nonebot-plugin-tetris-stats
 ```
+* 对于 Windows
+```
+# CMD or PowerShell
+playwright install firefox
+```
+* 对于 Linux
+```
+# 似乎 playwright官方 只支持 Ubuntu, 如果你是其他系统请自行尝试解决依赖问题
+playwright install firefox
+playwright install-deps firefox
+```
 
 使用
 ----
@@ -33,6 +46,7 @@ pip install nonebot-plugin-tetris-stats
 
 * [NoneBot2](https://v2.nonebot.dev/)
 * [OneBot](https://onebot.dev/)
+* [go-cqhttp](https://github.com/Mrs4s/go-cqhttp/)
 
 开源
 ----

nonebot_plugin_tetris_stats/IODataProcessing.py

@@ -1,137 +0,0 @@
-from nonebot.log import logger
-
-from asyncio import gather
-import aiohttp
-
-# 封装请求函数
-async def request(Url: str) -> dict[str, bool|dict[str, any]]:
-    data = {}
-    try:
-        async with aiohttp.ClientSession() as session:
-            async with session.get(Url) as resp:
-                data['Status'] = True
-                data['Data'] = await resp.json()
-                data['Success'] = data['Data']['success']
-    except aiohttp.client_exceptions.ClientConnectorError as e:
-        logger.error(f'[TETRIS STATS] IODataProcessing.request: 请求错误\n{e}')
-        data['Status'] = False
-    finally:
-        return data
-
-# 获取用户数据
-async def getUserData(userName: str = None, userID: str = None) -> dict[str, dict[str, any]]:
-    if userName is not None and userID is None:
-        userDataUrl = f'https://ch.tetr.io/api/users/{userName}'
-        userData = await request(Url=userDataUrl)
-    elif userName is None and userID is not None:
-        userDataUrl = f'https://ch.tetr.io/api/users/{userID}'
-        userData = await request(Url=userDataUrl)
-    else:
-        raise ValueError('[TETRIS STATS] IODataProcessing.getUserData: 参数错误')
-    return userData
-
-# 获取Solo数据
-async def getSoloData(userName: str = None, userID: str = None) -> dict[str, dict[str, any]]:
-    if userName is not None and userID is None:
-        userSoloUrl = f'https://ch.tetr.io/api/users/{userName}/records'
-        soloData = await request(Url = userSoloUrl)
-    elif userName is None and userID is not None:
-        userSoloUrl = f'https://ch.tetr.io/api/users/{userID}/records'
-        soloData = await request(Url = userSoloUrl)
-    else:
-        raise ValueError('[TETRIS STATS] IODataProcessing.getSoloData: 参数错误')
-    return soloData
-
-# 获取用户ID
-async def getUserID(userData: dict = None, userName: str = None) -> str:
-    if userName is not None and userData is None:
-        userData = await getUserData(userName=userName)
-    elif userData is None and userName is None:
-        raise ValueError('[TETRIS STATS] IODataProcessing.getUserID: 参数错误')
-    return userData['Data']['data']['user']['_id']
-
-# 获取排位统计数据
-async def getLeagueStats(userData: dict) -> dict[str, bool|int|str|float]:
-    league = userData['Data']['data']['user']['league']
-    leagueStats = {}
-    if league['gamesplayed'] == 0:
-        leagueStats['Played'] = False
-    else:
-        leagueStats['Played'] = True
-        leagueStats['PPS'] = league['pps']
-        leagueStats['APM'] = league['apm']
-        leagueStats['VS'] = 0 if league['vs'] is None else league['vs']
-        leagueStats['Rank'] = False if league['rank'] == 'z' else league['rank'].upper()
-        if league['rating'] != -1:
-            leagueStats['Ranked'] = True
-            leagueStats['Rating'] = round(league['rating'], 2)
-            leagueStats['Glicko'] = round(league['glicko'], 2)
-            leagueStats['RD'] = round(league['rd'], 2)
-        else:
-            leagueStats['Ranked'] = False
-        leagueStats['Standing'] = league['standing']
-        leagueStats['LPM'] = round((league['pps'] * 24), 2)
-        leagueStats['APL'] = round((leagueStats['APM'] / leagueStats['LPM']), 2)
-        leagueStats['ADPM'] = round((leagueStats['VS'] * 0.6), 2)
-        leagueStats['ADPL'] = round((leagueStats['ADPM'] / leagueStats['LPM']), 2)
-    return leagueStats
-
-# 获取40L统计数据
-async def getSprintStats(soloData: dict) -> dict[str, bool|int|float]:
-    sprintStats = {}
-    if soloData['Data']['data']['records']['40l']['record'] is None:
-        sprintStats['Played'] = False
-    else:
-        sprintStats['Played'] = True
-        sprintStats['Rank'] = False if soloData['Data']['data']['records']['40l']['rank'] is None else soloData['Data']['data']['records']['40l']['rank']
-        sprintStats['Time'] = round(soloData['Data']['data']['records']['40l']['record']['endcontext']['finalTime'] / 1000, 2)
-    return sprintStats
-
-# 获取Blitz统计数据
-async def getBlitzStats(soloData: dict) -> dict[str, bool|int]:
-    blitzStats = {}
-    if soloData['Data']['data']['records']['blitz']['record'] is None:
-        blitzStats['Played'] = False
-    else:
-        blitzStats['Played'] = True
-        blitzStats['Rank'] = False if soloData['Data']['data']['records']['blitz']['rank'] is None else soloData['Data']['data']['records']['blitz']['rank']
-        blitzStats['Score'] = soloData['Data']['data']['records']['blitz']['record']['endcontext']['score']
-    return blitzStats
-
-# 生成消息
-async def generateMessage(userName: str = None, userID: str = None) -> str:
-    userData, soloData = await gather(getUserData(userName = userName, userID = userID), getSoloData(userName = userName, userID = userID))
-    if userData['Status'] is False:
-        return '用户信息请求失败'
-    if userData['Success'] is False:
-        return f'用户信息请求错误:\n{userData["Data"]["error"]}'
-    userName = userData['Data']['data']['user']['username'].upper()
-    leagueStats = await getLeagueStats(userData)
-    message = ''
-    if leagueStats['Played'] is False:
-        message += f'用户 {userName} 没有排位统计数据'
-    else:
-        if leagueStats['Rank'] is False and leagueStats['Ranked'] is False:
-            message += f'用户 {userName} 暂未完成定级赛'
-        elif leagueStats['Rank'] is False and leagueStats['Ranked'] is True:
-            message += f'用户 {userName} 暂无段位, {leagueStats["Rating"]} TR'
-        else:
-            message += f'{leagueStats["Rank"]} 段用户 {userName} {leagueStats["Rating"]} TR (#{leagueStats["Standing"]})'
-        message += f', 段位分 {leagueStats["Glicko"]}±{leagueStats["RD"]}, 最近十场的数据:' if leagueStats['Ranked'] is True else ', 最近十场的数据:'
-        message += f'\nL\'PM: {leagueStats["LPM"]} ( {leagueStats["PPS"]} pps )'
-        message += f'\nAPM: {leagueStats["APM"]} ( x{leagueStats["APL"]} )'
-        if leagueStats["VS"] != 0:
-            message += f'\nADPM: {leagueStats["ADPM"]} ( x{leagueStats["ADPL"]} ) ( {leagueStats["VS"]}vs )'
-    if soloData['Status'] is False:
-        return f'{message}\nSolo统计数据请求失败'
-    sprintStats = await getSprintStats(soloData)
-    blitzStats = await getBlitzStats(soloData)
-    if sprintStats['Played'] is True:
-        message += f'\n40L: {sprintStats["Time"]}s'
-        if sprintStats['Rank'] is not False:
-            message += f' ( #{sprintStats["Rank"]} )'
-    if blitzStats['Played'] is True:
-        message += f'\nBlitz: {blitzStats["Score"]}'
-        if blitzStats['Rank'] is not False:
-            message += f' ( #{blitzStats["Rank"]} )'
-    return message

nonebot_plugin_tetris_stats/MessageAnalyzer.py

@@ -1,69 +0,0 @@
-from re import match
-
-# userBind
-async def handleBindMessage(message: str, gameType: str) -> dict[str, bool | str]:
-    _CMD_ALIASES = {'IO': ['io绑定', 'iobind'],
-                    'TOP': ['top绑定', 'topbind']}
-    # 剔除命令前缀
-    for i in _CMD_ALIASES[gameType]:
-        if message.startswith(i):
-            message = message.replace(i, '')
-            message = message.strip()
-            break
-    if message == '' or message.isspace():
-        return {'Success': False, 'Type': None, 'Message': '用户名为空'}
-    else:
-        return await checkName(message, gameType)
-
-# statsQuery
-async def handleStatsQueryMessage(message: str, gameType: str) -> dict[str, bool | str]:
-    _CMD_ALIASES = {'IO': ['io查', 'iostats'],
-                    'TOS': ['tos查', 'tostats', '茶服查', '茶服stats'],
-                    'TOP': ['top查', 'topstats']}
-    _ME = ['我', '自己', '私', '俺', 'me']
-    message = (message.strip()).lower()
-    # 剔除命令前缀
-    for i in _CMD_ALIASES[gameType]:
-        if message.startswith(i):
-            message = message.replace(i, '')
-            message = message.strip()
-            break
-    if message == '' or message.isspace():
-        return {'Success': False, 'Type': None, 'Message': '用户名为空'}
-    else:
-        if message.startswith('[cq:at,qq='):
-            try:
-                QQNumber = int((str(message)).split(
-                    '[cq:at,qq=')[1].split(']')[0])
-            except ValueError:
-                return {'Success': False, 'Type': None, 'Message': 'QQ号码不合法'}
-            else:
-                return {'Success': True, 'Type': 'AT', 'Message': None, 'QQNumber': QQNumber}
-        elif message in _ME:
-            return {'Success': True, 'Type': 'ME', 'Message': None}
-        else:
-            return await checkName(message, gameType)
-
-async def checkName(name: str, gameType: str) -> dict[str, bool | str]:
-    if gameType == 'IO':
-        if match(r'^[a-f0-9]{24}$', name):
-            return {'Success': True, 'Type': 'ID', 'Message': None, 'User': name}
-        elif match(r'^[a-zA-Z0-9_-]{3,16}$', name):
-            return {'Success': True, 'Type': 'Name', 'Message': None, 'User': name}
-        else:
-            return {'Success': False, 'Type': None, 'Message': '用户名不合法'}
-    elif gameType == 'TOP':
-        if match(r'^[a-zA-Z0-9_]{1,16}$', name):
-            return {'Success': True, 'Type': 'Name', 'Message': None, 'User': name}
-        else:
-            return {'Success': False, 'Type': None, 'Message': '用户名不合法'}
-    elif gameType == 'TOS':
-        if (match(r'^(?!\.)(?!com[0-9]$)(?!con$)(?!lpt[0-9]$)(?!nul$)(?!prn$)[^\-][^\+][^\|\*\?\\\s\!:<>/$"]*[^\.\|\*\?\\\s\!:<>/$"]+$', name)
-            and name.isdigit() is False
-            and 2 <= len(name) <= 18):
-            # 虽然我也不想这么长 但是似乎确实得这么长
-            return {'Success': True, 'Type': 'Name', 'Message': None, 'User': name}
-        elif name.isdigit() is True:
-            return {'Success': True, 'Type': 'QQ', 'Message': None, 'QQNumber': name}
-        else:
-            return {'Success': False, 'Type': None, 'Message': '用户名不合法'}

nonebot_plugin_tetris_stats/SQL.py

@@ -1,52 +0,0 @@
-from nonebot.log import logger
-
-import sqlite3
-import os
-
-_DB_FILE = 'data/nonebot-plugin-tetris-stats/data.db'
-
-# 初始化数据库
-async def initDB():
-    if not os.path.exists(os.path.dirname(_DB_FILE)):
-        os.makedirs(os.path.dirname(_DB_FILE))
-    db = sqlite3.connect(_DB_FILE)
-    cursor = db.cursor()
-    cursor.execute('''CREATE TABLE IF NOT EXISTS IOBIND
-                    (QQ INTEGER NOT NULL,
-                    USER TEXT NOT NULL)''')
-    cursor.execute('''CREATE TABLE IF NOT EXISTS TOPBIND
-                    (QQ INTEGER NOT NULL,
-                    USER TEXT NOT NULL)''')
-    db.commit()
-    db.close()
-    logger.info('数据库初始化完成')
-
-# 查询绑定信息
-async def queryBindInfo(QQNumber: int, gameType: str) -> dict:
-    db = sqlite3.connect(_DB_FILE)
-    cursor = db.cursor()
-    cursor.execute(f'SELECT USER FROM {gameType}BIND WHERE QQ = {QQNumber}')
-    user = cursor.fetchone()
-    db.commit()
-    db.close()
-    if user is None:
-        return {'Hit': False, 'User': None}
-    else:
-        return {'Hit': True, 'User': user[0]}
-
-# 写入绑定信息
-async def writeBindInfo(QQNumber: int, user: str, gameType: str) -> str:
-    info = await queryBindInfo(QQNumber, gameType)
-    db = sqlite3.connect(_DB_FILE)
-    cursor = db.cursor()
-    if info['Hit'] is True:
-        cursor.execute(
-            f'UPDATE {gameType}BIND SET USER = ? WHERE QQ = ?', (user, QQNumber))
-        message = '更新成功'
-    elif info['Hit'] is False:
-        cursor.execute(
-            f'INSERT INTO {gameType}BIND (QQ, USER) VALUES (?, ?)', (QQNumber, user))
-        message = '绑定成功'
-    db.commit()
-    db.close()
-    return message

nonebot_plugin_tetris_stats/TOSDataProcessing.py

@@ -1,143 +0,0 @@
-from nonebot.log import logger
-
-from asyncio import gather
-import aiohttp
-
-# 封装请求函数
-async def request(Url: str) -> dict[str, bool|dict[str, any]]:
-    data = {}
-    try:
-        async with aiohttp.ClientSession() as session:
-            async with session.get(Url) as resp:
-                data['Status'] = True
-                data['Data'] = await resp.json()
-                data['Success'] = data['Data']['success']
-    except aiohttp.client_exceptions.ClientConnectorError as e:
-        logger.error(f'[TETRIS STATS] TOSDataProcessing.request: 请求错误\n{e}')
-        data['Status'] = False
-    finally:
-        return data
-
-# 获取用户信息
-async def getUserInfo(userName: str = None, teaID: int = None) -> dict[str, bool|dict[str, any]]:
-    if userName is not None and teaID is None:
-        userDataUrl = f'https://teatube.cn:8888/getUsernameInfo?username={userName}'
-        userInfo = await request(Url=userDataUrl)
-    elif teaID is not None and userName is None:
-        userDataUrl = f'https://teatube.cn:8888/getTeaIdInfo?teaId={teaID}'
-        userInfo = await request(Url=userDataUrl)
-    else:
-        raise ValueError('[TETRIS STATS] TOSDataProcessing.getUserInfo: 参数错误')
-    return userInfo
-
-# 获取用户数据
-async def getUserData(userName: str = None, teaID: int = None, otherParameter: str = '') -> dict[str, bool|dict[str, any]]:
-    if userName is not None and teaID is None:
-        userDataUrl = f'https://teatube.cn:8888/getProfile?id={userName}{otherParameter}'
-        userData = await request(Url=userDataUrl)
-    elif teaID is not None and userName is None:
-        userDataUrl = f'https://teatube.cn:8888/getProfile?id={teaID}{otherParameter}'
-        userData = await request(Url=userDataUrl)
-    else:
-        raise ValueError('[TETRIS STATS] TOSDataProcessing.getUserData: 参数错误')
-    return userData
-
-# 获取Rank数据
-async def getRankStats(userInfo: dict) -> dict[str, bool|float]:
-    rankStats = {}
-    if int(userInfo['Data']['data']['rankedGames']) == 0:
-        rankStats['Played'] = False
-    else:
-        rankStats['Played'] = True
-        rankStats['Rating'] = round(float(userInfo['Data']['data']['ratingNow']), 2)
-        rankStats['RD'] = round(float(userInfo['Data']['data']['rdNow']), 2)
-        rankStats['Vol'] = round(float(userInfo['Data']['data']['volNow']), 3)
-    return rankStats
-
-# 获取游戏数据
-async def getGameData(userData: dict) -> dict[str, bool|int|float]:
-    gameData = {}
-    if userData['Data']['data'] == []:
-        gameData['Played'] = False
-    else:
-        gameData['Played'] = True
-        weightedTotalLpm = weightedTotalApm = weightedTotalAdpm = weightedTotalTime = num = 0
-        for i in userData['Data']['data']:
-            # 排除单人局和时间为0的游戏
-            if i['num_players'] == 1 or i['time'] == 0:
-                continue
-            # 茶：不计算没挖掘的局，即使apm和lpm也如此
-            if i['dig'] is None:
-                break
-            # 加权计算
-            time = i['time'] / 1000
-            lpm = 24 * (i['pieces'] / time)
-            apm = (i['attack'] / time) * 60
-            adpm = ((i['attack'] + i['dig']) / time) * 60
-            weightedTotalLpm += lpm * time
-            weightedTotalApm += apm * time
-            weightedTotalAdpm += adpm * time
-            weightedTotalTime += time
-            num += 1
-            if num == 50:
-                break
-        gameData['NUM'] = num
-        gameData['LPM'] = round((weightedTotalLpm / weightedTotalTime), 2)
-        gameData['APM'] = round((weightedTotalApm / weightedTotalTime), 2)
-        gameData['ADPM'] = round((weightedTotalAdpm / weightedTotalTime), 2)
-        gameData['PPS'] = round((gameData['LPM'] / 24), 2)
-        gameData['APL'] = round((gameData['APM'] / gameData['LPM']), 2)
-        gameData['ADPL'] = round((gameData['ADPM'] / gameData['LPM']), 2)
-        gameData['VS'] = round((gameData['ADPM'] / 60 * 100), 2)
-        # TODO: 如果有效局数不满50, 没有无dig信息的局, 且userData['Data']['data']内有50个局, 则继续往前获取信息
-    return gameData
-
-# 获取PB数据
-async def getPBData(userInfo: dict) -> dict[str, bool|float|str]:
-    PBData = {}
-    if int(userInfo['Data']['data']['PBSprint']) == 2147483647:
-        PBData['Sprint'] = False
-    else:
-        PBData['Sprint'] = round(float(userInfo['Data']['data']['PBSprint']) / 1000, 2)
-    if int(userInfo['Data']['data']['PBMarathon']) == 0:
-        PBData['Marathon'] = False
-    else:
-        PBData['Marathon'] = userInfo['Data']['data']['PBMarathon']
-    if int(userInfo['Data']['data']['PBChallenge']) == 0:
-        PBData['Challenge'] = False
-    else:
-        PBData['Challenge'] = userInfo['Data']['data']['PBChallenge']
-    return PBData
-
-# 生成消息
-async def generateMessage(userName: str = None, teaID: int = None) -> str:
-    userInfo, userData = await gather(getUserInfo(userName=userName, teaID=teaID), getUserData(userName=userName, teaID=teaID))
-    if userInfo['Status'] is False:
-        return f'用户信息请求失败'
-    if userInfo['Success'] is False:
-        return f'用户信息请求错误:\n{userInfo["Data"]["error"]}'
-    rankStats = await getRankStats(userInfo)
-    PBData = await getPBData(userInfo)
-    message = ''
-    if rankStats['Played'] is False:
-        message += f'用户 {userInfo["Data"]["data"]["name"]}（{userInfo["Data"]["data"]["teaId"]}）暂无段位统计数据'
-    elif rankStats['Played'] is True:
-        message += f'用户 {userInfo["Data"]["data"]["name"]} ({userInfo["Data"]["data"]["teaId"]}) , 段位分 {rankStats["Rating"]}±{rankStats["RD"]} ({rankStats["Vol"]}) '
-    if userData['Status'] is False:
-        message = f'{message.rstrip()}\n游戏数据请求失败'
-    elif userData['Status'] is True:
-        gameData = await getGameData(userData)
-        if gameData['Played'] is False:
-            message += ', 暂无游戏数据'
-        elif gameData['Played'] is True:
-            message += f', 最近 {gameData["NUM"]} 局数据'
-            message += f'\nL\'PM: {gameData["LPM"]} ( {gameData["PPS"]} pps )'
-            message += f'\nAPM：{gameData["APM"]} ( x{gameData["APL"]} )'
-            message += f'\nADPM：{gameData["ADPM"]} ( x{gameData["ADPL"]} ) ( {gameData["VS"]}vs )'
-    if PBData['Sprint'] is not False:
-        message += f'\n40L: {PBData["Sprint"]}s'
-    if PBData['Marathon']is not False:
-        message += f'\nMarathon: {PBData["Marathon"]}'
-    if PBData['Challenge'] is not False:
-        message += f'\nChallenge: {PBData["Challenge"]}'
-    return message

nonebot_plugin_tetris_stats/__init__.py

@@ -1,86 +1 @@
-from re import I
-
-from nonebot import get_driver, on_regex
-from nonebot.adapters.onebot.v11 import GROUP, MessageEvent
-from nonebot.matcher import Matcher
-
-from .MessageAnalyzer import *
-from .SQL import *
-
-from .IODataProcessing import generateMessage as IOgenerateMessage
-from .IODataProcessing import getUserID as IOgetUserID
-from .TOSDataProcessing import generateMessage as TOSgenerateMessage
-
-driver = get_driver()
-
-@driver.on_startup
-async def startUP():
-    await initDB()
-
-ioBind = on_regex(pattern=r'^io绑定|^iobind', flags=I, permission=GROUP)
-ioStats = on_regex(pattern=r'^io查|^iostats', flags=I, permission=GROUP)
-
-tosStats = on_regex(pattern=r'^tos查|^tostats|^茶服查|^茶服stats',
-                    flags=I, permission=GROUP)
-
-topBind = on_regex(pattern=r'^top绑定|^topbind', flags=I, permission=GROUP)
-topStats = on_regex(pattern=r'^top查|^topstats', flags=I, permission=GROUP)
-
-@ioBind.handle()
-async def bindIOUser(event: MessageEvent, matcher: Matcher):
-    decodedMessage = await handleBindMessage(message=str(event.get_message()), gameType='IO')
-    if decodedMessage['Success'] is True:
-        if decodedMessage['Type'] == 'ID':
-            user = decodedMessage['User']
-        elif decodedMessage['Type'] == 'Name':
-            user = await IOgetUserID(userName=decodedMessage['User'])
-        message = await writeBindInfo(QQNumber=event.sender.user_id, user=user, gameType='IO')
-    elif decodedMessage['Success'] is False:
-        message = decodedMessage['Message']
-    await matcher.send(message=message)
-
-@ioStats.handle()
-async def handleIOStatsQuery(event: MessageEvent, matcher: Matcher):
-    decodedMessage = await handleStatsQueryMessage(message=str(event.get_message()), gameType='IO')
-    if decodedMessage['Success'] is True:
-        if decodedMessage['Type'] == 'AT':
-            bindInfo = await queryBindInfo(QQNumber=decodedMessage['QQNumber'], gameType='IO')
-            if bindInfo['Hit'] is True:
-                message = (f'* 由于无法验证绑定信息，不能保证查询到的用户为本人\n{await IOgenerateMessage(userID=bindInfo["User"])}')
-            elif bindInfo['Hit'] is False:
-                message = '未查询到绑定信息'
-        elif decodedMessage['Type'] == 'ME':
-            bindInfo = await queryBindInfo(QQNumber=event.sender.user_id, gameType='IO')
-            if bindInfo['Hit'] is True:
-                message = (f'* 由于无法验证绑定信息，不能保证查询到的用户为本人\n{await IOgenerateMessage(userID=bindInfo["User"])}')
-            elif bindInfo['Hit'] is False:
-                message = '您还没有绑定账号'
-        elif decodedMessage['Type'] == 'ID':
-            message = await IOgenerateMessage(userID=decodedMessage['User'])
-        elif decodedMessage['Type'] == 'Name':
-            message = await IOgenerateMessage(userName=decodedMessage['User'])
-    elif decodedMessage['Success'] is False:
-        message = decodedMessage['Message']
-    await matcher.finish(message=message)
-
-@tosStats.handle()
-async def handleTOSStatsQuery(event: MessageEvent, matcher: Matcher):
-    decodedMessage = await handleStatsQueryMessage(message=str(event.get_message()), gameType='TOS')
-    if decodedMessage['Success'] is True:
-        if decodedMessage['Type'] == 'AT' or decodedMessage['Type'] == 'QQ':
-            message = await TOSgenerateMessage(teaID=decodedMessage['QQNumber'])
-        elif decodedMessage['Type'] == 'ME':
-            message = await TOSgenerateMessage(teaID=event.sender.user_id)
-        elif decodedMessage['Type'] == 'Name':
-            message = await TOSgenerateMessage(userName=decodedMessage['User'])
-    elif decodedMessage['Success'] is False:
-        message = decodedMessage['Message']
-    await matcher.finish(message=message)
-
-@topBind.handle()
-async def bindTOPUser(event: MessageEvent, matcher: Matcher):
-    await matcher.send(message='TODO')
-
-@topStats.handle()
-async def handleTOPStatsQuery(event: MessageEvent, matcher: Matcher):
-    await matcher.send(message='TODO')
+from . import game_data_processor

nonebot_plugin_tetris_stats/game_data_processor/__init__.py

@@ -0,0 +1 @@
+from . import io_data_processor, top_data_processor, tos_data_processor

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/__init__.py

@@ -0,0 +1,40 @@
+from re import I
+
+from nonebot import on_regex
+from nonebot.adapters.onebot.v11 import GROUP, MessageEvent
+from nonebot.matcher import Matcher
+
+from .processor import Processor
+
+IOBind = on_regex(pattern=r'^io绑定|^iobind', flags=I, permission=GROUP)
+IOStats = on_regex(pattern=r'^io查|^iostats', flags=I, permission=GROUP)
+IORank = on_regex(pattern=r'^io段位|^iorank', flags=I, permission=GROUP)
+
+@IOBind.handle()
+async def _(event: MessageEvent, matcher: Matcher):
+    await matcher.finish(
+        await Processor.handle_bind(
+            message=event.raw_message,
+            qq_number=event.sender.user_id
+        )
+    )
+
+
+@IOStats.handle()
+async def _(event: MessageEvent, matcher: Matcher):
+    if event.is_tome():
+        await matcher.finish('不能查询bot的信息')
+    await matcher.finish(
+        await Processor.handle_query(
+            message=event.raw_message,
+            qq_number=event.sender.user_id
+        )
+    )
+
+@IORank.handle()
+async def _(event: MessageEvent, matcher: Matcher):
+    await matcher.finish(
+        await Processor.handle_rank(
+            message=event.raw_message
+        )
+    )

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/processor.py

@@ -0,0 +1,289 @@
+import math
+from asyncio import gather
+from typing import Any
+
+from nonebot.log import logger
+
+from ...utils.database import DataBase
+from ...utils.message_analyzer import (
+    handle_bind_message,
+    handle_rank_message,
+    handle_stats_query_message,
+)
+from .request import Request
+
+
+class Processor:
+    @classmethod
+    async def handle_bind(cls, message: str, qq_number: int | None) -> str:
+        '''处理绑定消息'''
+        decoded_message = await handle_bind_message(message=message, game_type='IO')
+        if decoded_message[0] is None:
+            return decoded_message[1][0]
+        if decoded_message[0] == 'ID':
+            user_id_stats = await cls.check_user_id(decoded_message[1][1])
+            if user_id_stats[0] is False:
+                return user_id_stats[1]
+            user_id = decoded_message[1][1]
+        elif decoded_message[0] == 'Name':
+            user_data = await cls.get_user_data(user_name=decoded_message[1][1])
+            if user_data[0] is False:
+                return '用户信息请求失败'
+            if user_data[1] is False:
+                return f'用户信息请求错误:\n{user_data[2]["error"]}'
+            user_id = await cls.get_user_id(user_data[2])
+            if qq_number is None:  # 理论上是不会有None出现的, ide快乐行属于是（
+                logger.error('获取QQ号失败')
+                return '获取QQ号失败'
+            return (
+                await DataBase.write_bind_info(
+                    qq_number=qq_number,
+                    user=user_id,
+                    game_type='IO'
+                )
+            )
+        logger.error('预期外行为, 请上报GitHub')
+        return '出现预期外行为，请查看后台信息'
+
+    @classmethod
+    async def handle_rank(cls, message: str):
+        query_rank = await handle_rank_message(message)
+        rank_info = await DataBase.query_rank_info_today(rank=query_rank.lower())
+
+        if rank_info is None:
+            ranks_percentiles = {
+                'x': 1,
+                'u': 5,
+                'ss': 11,
+                's+': 17,
+                's': 23,
+                's-': 30,
+                'a+': 38,
+                'a': 46,
+                'a-': 54,
+                'b+': 62,
+                'b': 70,
+                'b-': 78,
+                'c+': 84,
+                'c': 90,
+                'c-': 95,
+                'd+': 97.5,
+                'd': 100,
+            }
+
+            if query_rank.lower() not in (i for i in ranks_percentiles.keys()):
+                return '未知段位'
+
+            result = await Request.request('https://ch.tetr.io/api/users/lists/league/all')
+            users: list = result[2]['data']['users']
+
+            def avg(rank_users: list, column: str, playercount: int | None = None) -> float:
+                return sum(i['league'][column] for i in rank_users) / (playercount or len(rank_users))
+
+            for rank, percentile in ranks_percentiles.items():
+                offset = math.floor((percentile / 100) * len(users)) - 1
+                tr = users[offset]['league']['rating']
+
+                rank_users = list(filter(lambda x: x['league']['rank'] == rank, users))
+                playercount = len(rank_users)
+
+                avg_apm = avg(rank_users, 'apm', playercount)
+                avg_pps = avg(rank_users, 'pps', playercount)
+                avg_vs = avg(rank_users, 'vs', playercount)
+
+                await DataBase.write_rank_info_today(
+                    rank=rank,
+                    trline=tr,
+                    playercount=playercount,
+                    avgapm=avg_apm,
+                    avgpps=avg_pps,
+                    avgvs=avg_vs
+                    )
+
+            return await Processor.handle_rank(message=message)
+        else:
+            avg_apm = round(rank_info[3], 2)
+            avg_pps = round(rank_info[4], 2)
+            avg_vs = round(rank_info[5], 2)
+            avg_lpm = round((avg_pps * 24), 2)
+            avg_apl = round((avg_apm / avg_lpm), 2)
+            avg_adpm = round((avg_vs * 0.6), 2)
+            avg_adpl = round((avg_adpm / avg_lpm), 2)
+
+            message = f'{query_rank.upper()} 段, 分数线 {round(rank_info[1], 2)} TR, {rank_info[2]} 名玩家'
+            message += f'\n对比昨日趋势: {rank_info[0]}'
+            message += '\n平均数据: '
+            message += f"\nL'PM: {avg_lpm} ( {avg_pps} pps )"
+            message += f'\nAPM: {avg_apm} ( x{avg_apl} )'
+            message += f'\nADPM: {avg_adpm} ( x{avg_adpl} ) ( {avg_vs}vs )'
+            message += '\n'
+            message += f'\n数据更新时间: {rank_info[6]}'
+
+        return message
+
+    @classmethod
+    async def handle_query(cls, message: str, qq_number: int | None):
+        '''处理查询消息'''
+        decoded_message = await handle_stats_query_message(message=message, game_type='IO')
+        if decoded_message[0] is None:
+            return decoded_message[1][0]
+        if decoded_message[0] == 'AT':  # 在入口处判断是否@bot本身
+            bind_info = await DataBase.query_bind_info(qq_number=decoded_message[1][1], game_type='IO')
+            if bind_info is None:
+                return '未查询到绑定信息'
+            return f'* 由于无法验证绑定信息, 不能保证查询到的用户为本人\n{await Processor.generate_message(user_id=bind_info)}'
+        if decoded_message[0] == 'ME':
+            if qq_number is None:
+                logger.error('获取QQ号失败')
+                return '获取QQ号失败, 请联系bot主人'
+            bind_info = await DataBase.query_bind_info(qq_number=qq_number, game_type='IO')
+            if bind_info is None:
+                return '未查询到绑定信息'
+            return f'* 由于无法验证绑定信息, 不能保证查询到的用户为本人\n{await Processor.generate_message(user_id=bind_info)}'
+        if decoded_message[0] == 'ID':
+            return await Processor.generate_message(user_id=decoded_message[1][1])
+        if decoded_message[0] == 'Name':
+            return await Processor.generate_message(user_name=decoded_message[1][1])
+
+    @classmethod
+    async def get_user_data(
+            cls,
+            user_name: str | None = None,
+            user_id: str | None = None
+    ) -> tuple[bool, bool, dict[str, Any]]:
+        '''获取用户数据'''
+        if user_name is not None and user_id is None:
+            user_data_url = f'https://ch.tetr.io/api/users/{user_name}'
+        elif user_name is None and user_id is not None:
+            user_data_url = f'https://ch.tetr.io/api/users/{user_id}'
+        else:
+            raise ValueError('预期外行为, 请上报GitHub')
+        return await Request.request(user_data_url)
+
+    @classmethod
+    async def get_solo_data(
+            cls,
+            user_name: str | None = None,
+            user_id: str | None = None
+    ) -> tuple[bool, bool, dict[str, Any]]:
+        '''获取Solo数据'''
+        if user_name is not None and user_id is None:
+            user_solo_url = f'https://ch.tetr.io/api/users/{user_name}/records'
+        elif user_name is None and user_id is not None:
+            user_solo_url = f'https://ch.tetr.io/api/users/{user_id}/records'
+        else:
+            raise ValueError('预期外行为, 请上报GitHub')
+        return await Request.request(user_solo_url)
+
+    @classmethod
+    async def get_user_id(cls, user_data: dict) -> str:
+        '''获取用户ID'''
+        return user_data['data']['user']['_id']
+
+    @classmethod
+    async def check_user_id(cls, user_id: str) -> tuple[bool, str]:
+        '''检查用户ID是否有效 返回值为tuple[bool, message]'''
+        user_data = await cls.get_user_data(user_id=user_id)
+        if user_data[0] is False:
+            return False, '用户信息请求失败'
+        if user_data[1] is False:
+            return False, f'用户信息请求错误:\n{user_data[2]["error"]}'
+        if user_id == user_data[2]['data']['user']['_id']:
+            return True, ''
+        raise ValueError('服务器返回的userID和用户提供的不一致, 这种情况理论上不应该发生, 以防万一还是写一下（x')
+
+    @classmethod
+    async def get_league_stats(cls, user_data: dict) -> dict[str, Any]:
+        '''获取排位统计数据'''
+        league = user_data['data']['user']['league']
+        league_stats: dict[str, Any] = {}
+        if league['gamesplayed'] != 0:
+            league_stats['PPS'] = league['pps']
+            league_stats['APM'] = league['apm']
+            league_stats['VS'] = 0 if league['vs'] is None else league['vs']
+            league_stats['Rank'] = 'Z' if league['rank'] == 'z' else league['rank'].upper(
+            )
+            if league['rating'] == -1:
+                league_stats['Rank'] = None
+            else:
+                league_stats['Rating'] = round(league['rating'], 2)
+                league_stats['Glicko'] = round(league['glicko'], 2)
+                league_stats['RD'] = round(league['rd'], 2)
+            league_stats['Standing'] = league['standing']
+            league_stats['LPM'] = round((league['pps'] * 24), 2)
+            league_stats['APL'] = round(
+                (league_stats['APM'] / league_stats['LPM']), 2)
+            league_stats['ADPM'] = round((league_stats['VS'] * 0.6), 2)
+            league_stats['ADPL'] = round(
+                (league_stats['ADPM'] / league_stats['LPM']), 2)
+        return league_stats
+
+    @classmethod
+    async def get_sprint_stats(cls, solo_data: dict) -> dict[str, Any]:
+        '''获取40L统计数据'''
+        sprint_stats = {}
+        solo = solo_data['data']['records']['40l']
+        if solo['record'] is not None:
+            sprint_stats['Time'] = round(
+                solo['record']['endcontext']['finalTime'] / 1000, 2)
+            if solo['rank'] is not None:
+                sprint_stats['Rank'] = solo['rank']
+        return sprint_stats
+
+    @classmethod
+    async def get_blitz_stats(cls, solo_data: dict) -> dict[str, Any]:
+        '''获取Blitz统计数据'''
+        blitz_stats = {}
+        blitz = solo_data['data']['records']['blitz']
+        if blitz['record'] is not None:
+            blitz_stats['Score'] = blitz['record']['endcontext']['score']
+            if blitz['rank'] is not None:
+                blitz_stats['Rank'] = blitz['rank']
+        return blitz_stats
+
+    @classmethod
+    async def generate_message(
+            cls,
+            user_name: str | None = None,
+            user_id: str | None = None
+    ) -> str:
+        '''生成消息'''
+        user_data, solo_data = await gather(
+            cls.get_user_data(user_name=user_name, user_id=user_id),
+            cls.get_solo_data(user_name=user_name, user_id=user_id)
+        )
+        if user_data[0] is False:
+            return '用户信息请求失败'
+        if user_data[1] is False:
+            return f'用户信息请求错误:\n{user_data[2]["error"]}'
+        user_name = user_data[2]['data']['user']['username'].upper()
+        league_stats = await cls.get_league_stats(user_data[2])
+        message = ''
+        if not league_stats:
+            message += f'用户 {user_name} 没有排位统计数据'
+        else:
+            if league_stats['Rank'] is None:
+                message += f'用户 {user_name} 暂未完成定级赛, 最近十场的数据:'
+            else:
+                if league_stats['Rank'] == 'Z':
+                    message += f'用户 {user_name} 暂无段位, {league_stats["Rating"]} TR'
+                else:
+                    message += f'{league_stats["Rank"]} 段用户 {user_name} {league_stats["Rating"]} TR (#{league_stats["Standing"]})'
+                message += f', 段位分 {league_stats["Glicko"]}±{league_stats["RD"]}, 最近十场的数据:'
+            message += f'\nL\'PM: {league_stats["LPM"]} ( {league_stats["PPS"]} pps )'
+            message += f'\nAPM: {league_stats["APM"]} ( x{league_stats["APL"]} )'
+            if league_stats["VS"] != 0:
+                message += f'\nADPM: {league_stats["ADPM"]} ( x{league_stats["ADPL"]} ) ( {league_stats["VS"]}vs )'
+        if solo_data[0] is False:
+            return f'{message}\nSolo统计数据请求失败'
+        if solo_data[1] is False:
+            return f'{message}\nSolo统计数据请求错误:\n{solo_data[2]["error"]}'
+        sprint_stats, blitz_stats = await gather(
+            cls.get_sprint_stats(solo_data[2]),
+            cls.get_blitz_stats(solo_data[2])
+        )
+        message += f'\n40L: {sprint_stats["Time"]}s' if 'Time' in sprint_stats else ''
+        message += f' ( #{sprint_stats["Rank"]} )' if 'Rank' in sprint_stats else ''
+        message += f'\nBlitz: {blitz_stats["Score"]}' if 'Score' in blitz_stats else ''
+        message += f' ( #{blitz_stats["Rank"]} )' if 'Rank' in blitz_stats else ''
+        return message

nonebot_plugin_tetris_stats/game_data_processor/io_data_processor/request.py

@@ -0,0 +1,155 @@
+import os
+from typing import Any
+
+import aiohttp
+from nonebot import get_driver
+from nonebot.log import logger
+from playwright.async_api import Browser, Response, async_playwright
+from ujson import JSONDecodeError, dumps, loads
+
+from ...utils.config import Config
+
+driver = get_driver()
+
+config = Config.parse_obj(get_driver().config)
+
+
+@driver.on_startup
+async def _():
+    await Request.init_cache()
+    await Request.read_cache()
+
+
+@driver.on_shutdown
+async def _():
+    await Request.close_browser()
+    await Request.write_cache()
+
+
+class Request:
+    '''网络请求相关类'''
+    _browser: Browser | None = None
+    _headers: dict | None = None
+    _cookies: dict | None = None
+
+    @classmethod
+    async def _init_playwright(cls) -> Browser:
+        '''初始化playwright'''
+        playwright = await async_playwright().start()
+        cls._browser = await playwright.firefox.launch()
+        return cls._browser
+
+    @classmethod
+    async def _get_browser(cls) -> Browser:
+        '''获取浏览器对象'''
+        return cls._browser or await cls._init_playwright()
+
+    @classmethod
+    async def _anti_cloudflare(cls, url: str) -> tuple[bool, bool, dict[str, Any]]:
+        '''用firefox硬穿五秒盾'''
+        browser = await cls._get_browser()
+        context = await browser.new_context()
+        page = await context.new_page()
+        response = await page.goto(url)
+        attempts = 0
+        while attempts < 60:
+            attempts += 1
+            text = await page.locator("body").text_content()
+            if text is None:
+                await page.wait_for_timeout(1000)
+                continue
+            if await page.title() == 'Please Wait... | Cloudflare':
+                # TODO 有无人来做一个过验证码（
+                break
+            try:
+                data = loads(text)
+            except JSONDecodeError:
+                await page.wait_for_timeout(1000)
+            else:
+                assert isinstance(response, Response)
+                cls._headers = await response.request.all_headers()
+                try:
+                    cls._cookies = {i['name']: i['value'] for i in await context.cookies()}
+                except KeyError:
+                    cls._cookies = None
+                await page.close()
+                await context.close()
+                return True, data['success'], data
+        await page.close()
+        await context.close()
+        return True, False, {'error': '绕过五秒盾失败'}
+
+    @classmethod
+    async def init_cache(cls) -> None:
+        '''初始化缓存文件'''
+        if not os.path.exists(os.path.dirname(config.cache_path)):
+            os.makedirs(os.path.dirname(config.cache_path))
+        if not os.path.exists(config.cache_path):
+            with open(file=config.cache_path, mode='w', encoding='UTF-8') as file:
+                file.write(
+                    dumps(
+                        {
+                            'headers': cls._headers,
+                            'cookies': cls._cookies
+                        }
+                    )
+                )
+
+    @classmethod
+    async def read_cache(cls) -> None:
+        '''读取缓存文件'''
+        try:
+            with open(file=config.cache_path, mode='r', encoding='UTF-8') as file:
+                json = loads(file.read())
+                cls._headers = json['headers']
+                cls._cookies = json['cookies']
+        except FileNotFoundError:
+            await cls.init_cache()
+        except PermissionError:
+            os.remove(config.cache_path)
+            await cls.init_cache()
+        except JSONDecodeError:
+            os.remove(config.cache_path)
+            await cls.init_cache()
+
+    @classmethod
+    async def write_cache(cls) -> None:
+        '''写入缓存文件'''
+        try:
+            with open(file=config.cache_path, mode='r+', encoding='UTF-8') as file:
+                file.write(
+                    dumps(
+                        {
+                            'headers': cls._headers,
+                            'cookies': cls._cookies
+                        }
+                    )
+                )
+        except FileNotFoundError:
+            await cls.init_cache()
+        except PermissionError:
+            os.remove(config.cache_path)
+            await cls.init_cache()
+        except JSONDecodeError:
+            os.remove(config.cache_path)
+            await cls.init_cache()
+
+    @classmethod
+    async def request(cls, url: str) -> tuple[bool, bool, dict[str, Any]]:
+        '''请求api'''
+        try:
+            async with aiohttp.ClientSession(cookies=cls._cookies) as session:
+                async with session.get(url, headers=cls._headers) as resp:
+                    data = await resp.json()
+                    return True, data['success'], data
+        except aiohttp.client_exceptions.ClientConnectorError as error: # type: ignore
+            logger.error(f'请求错误\n{error}')
+            return False, False, {}
+        except aiohttp.client_exceptions.ContentTypeError:  # type: ignore
+            return await cls._anti_cloudflare(url)
+
+    @classmethod
+    async def close_browser(cls) -> None:
+        '''关闭浏览器对象'''
+        if isinstance(cls._browser, Browser):
+            await cls._browser.close()

nonebot_plugin_tetris_stats/game_data_processor/top_data_processor.py

@@ -0,0 +1,196 @@
+from asyncio import gather
+from re import I
+from typing import Any
+
+import aiohttp
+from lxml import etree
+from nonebot import on_regex
+from nonebot.adapters.onebot.v11 import GROUP, MessageEvent
+from nonebot.log import logger
+from nonebot.matcher import Matcher
+from pandas import read_html
+
+from ..utils.database import DataBase
+from ..utils.message_analyzer import (
+    handle_bind_message,
+    handle_stats_query_message
+)
+
+TOPBind = on_regex(pattern=r'^top绑定|^topbind', flags=I, permission=GROUP)
+TopStats = on_regex(pattern=r'^top查|^topstats', flags=I, permission=GROUP)
+
+
+@TOPBind.handle()
+async def _(event: MessageEvent, matcher: Matcher):
+    decoded_message = await handle_bind_message(message=event.raw_message, game_type='TOP')
+    if decoded_message[0] is None:
+        await matcher.finish(decoded_message[1][0])
+    elif decoded_message[0] == 'Name':
+        user_data = await get_user_data(decoded_message[1][1])
+        if user_data[0] is False:
+            await matcher.finish('用户信息请求失败')
+        else:
+            if await check_user(user_data[1]) is False:
+                await matcher.finish('用户不存在')
+            user_name = await get_user_name(user_data[1])
+            if event.sender.user_id is None:  # 理论上是不会有None出现的, ide快乐行属于是（
+                logger.error('获取QQ号失败')
+                await matcher.finish('获取QQ号失败')
+            await matcher.finish(
+                await DataBase.write_bind_info(
+                    qq_number=event.sender.user_id,
+                    user=user_name,
+                    game_type='TOP'
+                )
+            )
+
+
+@TopStats.handle()
+async def _(event: MessageEvent, matcher: Matcher):
+    decoded_message = await handle_stats_query_message(message=event.raw_message, game_type='TOP')
+    if decoded_message[0] is None:
+        await matcher.finish(decoded_message[1][0])
+    elif decoded_message[0] == 'AT':
+        if event.is_tome() is True:
+            await matcher.finish('不能查询bot的信息')
+        bind_info = await DataBase.query_bind_info(qq_number=decoded_message[1][1], game_type='TOP')
+        if bind_info is None:
+            message = '未查询到绑定信息'
+        else:
+            message = (f'* 由于无法验证绑定信息, 不能保证查询到的用户为本人\n{await generate_message(bind_info)}')
+    elif decoded_message[0] == 'ME':
+        if event.sender.user_id is None:
+            logger.error('获取QQ号失败')
+            await matcher.finish('获取QQ号失败, 请联系bot主人')
+        bind_info = await DataBase.query_bind_info(qq_number=event.sender.user_id, game_type='TOP')
+        if bind_info is None:
+            message = '未查询到绑定信息'
+        else:
+            message = (f'* 由于无法验证绑定信息, 不能保证查询到的用户为本人\n{await generate_message(bind_info)}')
+    elif decoded_message[0] == 'Name':
+        message = await generate_message(decoded_message[1][1])
+    else:
+        raise ValueError('预期外行为, 请上报GitHub')
+    await matcher.finish(message)
+
+
+async def get_user_data(user_name: str) -> tuple[bool, str]:
+    '''获取用户信息'''
+    url = f'http://tetrisonline.pl/top/profile.php?user={user_name}'
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url) as resp:
+                return True, await resp.text()
+    except aiohttp.client_exceptions.ClientConnectorError as error: # type: ignore
+        logger.error(error)
+        return False, ''
+
+
+async def check_user(user_data: str) -> bool:
+    '''如果用户存在返回True, 如果用户不存在返回False'''
+    return user_data.find('user not found!') == -1
+
+
+async def get_user_name(user_data: str) -> str:
+    '''获取用户名'''
+    data = etree.HTML(user_data).xpath('//div[@class="mycontent"]/h1/text()')
+    if isinstance(data, list):
+        return str(data[0]).replace('\'s profile', '')
+    raise TypeError('预期外行为, 请上报GitHub')
+
+
+async def get_game_stats(user_data: str) -> dict[str, dict[str, Any]]:
+    '''获取游戏统计数据'''
+    game_stats: dict[str, Any] = {'24H': {}, 'All': {}}
+    html = etree.HTML(user_data)
+    data = html.xpath('//div[@class="mycontent"]//text()')
+    if isinstance(data, list):
+        for i in data:
+            if not isinstance(i, str):
+                i = str(i)
+            i = i.strip()
+            if i.startswith('lpm:'):
+                game_stats['24H']['LPM'] = i.replace('lpm:', '').strip()
+            if i.startswith('apm:'):
+                game_stats['24H']['APM'] = i.replace('apm:', '').strip()
+            if 'LPM' in game_stats['24H'] and 'APM' in game_stats['24H']:
+                break
+    else:
+        raise TypeError('预期外行为, 请上报GitHub')
+    # 如果没有24H统计数据
+    if game_stats['24H'].get('LPM') in [None, ''] or game_stats['24H'].get('APM') in [None, '']:
+        game_stats['24H'].pop('LPM', None)
+        game_stats['24H'].pop('APM', None)
+    else:
+        game_stats['24H']['PPS'] = round(
+            float(game_stats['24H']['LPM']) / 24, 2)
+        game_stats['24H']['APL'] = round(
+            float(game_stats['24H']['APM']) / float(game_stats['24H']['LPM']), 2)
+        game_stats['24H']['LPM'] = round(float(game_stats['24H']['LPM']), 2)
+        game_stats['24H']['APM'] = round(float(game_stats['24H']['APM']), 2)
+    table = html.xpath('//table')
+    if isinstance(table, list):
+        if isinstance(table[0], etree._Element):
+            stats_table = etree.tostring(table[0], encoding='utf-8').decode()
+            df = read_html(stats_table, encoding='utf-8', header=0)[0]
+            results = df.T.to_dict().values()
+            if results:
+                game_stats['All']['LPM'] = 0
+                game_stats['All']['APM'] = 0
+                for i in results:
+                    if isinstance(i, dict):
+                        game_stats['All']['LPM'] += i['lpm']
+                        game_stats['All']['APM'] += i['apm']
+                game_stats['All']['LPM'] = game_stats['All']['LPM'] / \
+                    len(results)
+                game_stats['All']['APM'] = game_stats['All']['APM'] / \
+                    len(results)
+                game_stats['All']['PPS'] = round(
+                    game_stats['All']['LPM'] / 24, 2)
+                game_stats['All']['APL'] = round(
+                    float(game_stats['All']['APM']) / float(game_stats['All']['LPM']), 2)
+                game_stats['All']['LPM'] = round(
+                    float(game_stats['All']['LPM']), 2)
+                game_stats['All']['APM'] = round(
+                    float(game_stats['All']['APM']), 2)
+    else:
+        raise TypeError('预期外行为, 请上报GitHub')
+    return game_stats
+
+
+async def generate_message(user_name: str) -> str:
+    '''生成消息'''
+    user_data = await get_user_data(user_name)
+    if user_data[0] is False:
+        return '用户信息请求失败'
+    if await check_user(user_data[1]) is False:
+        return '用户不存在'
+    user_name, game_stats = await gather(
+        get_user_name(user_data[1]),
+        get_game_stats(user_data[1])
+    )
+    message = ''
+    if game_stats['24H'] and game_stats['All']:
+        message += f'用户 {user_name} 24小时内统计数据为: '
+        message += f'\nL\'PM: {game_stats["24H"]["LPM"]} ( {game_stats["24H"]["PPS"]} pps )'
+        message += f'\nAPM: {game_stats["24H"]["APM"]} ( x{game_stats["24H"]["APL"]} )'
+        message += '\n历史统计数据为: '
+        message += f'\nL\'PM: {game_stats["All"]["LPM"]} ( {game_stats["All"]["PPS"]} pps )'
+        message += f'\nAPM: {game_stats["All"]["APM"]} ( x{game_stats["All"]["APL"]} )'
+    elif game_stats['24H'] and not game_stats['All']:
+        message += f'用户 {user_name} 24小时内统计数据为: '
+        message += f'\nL\'PM: {game_stats["24H"]["LPM"]} ( {game_stats["24H"]["PPS"]} pps )'
+        message += f'\nAPM: {game_stats["24H"]["APM"]} ( x{game_stats["24H"]["APL"]} )'
+        message += '\n暂无历史统计数据'
+        message += '\n( 这理论上不该存在, 如果你看到了, 请联系bot主人查看后台'
+        logger.error(f'老实说这个不算Error, 但是理论上不应该有, 如果你看到了这条日志, 我希望你能来Github发个issue（\
+user_name: {user_name}\
+user_data: {user_data}\
+game_stats: {game_stats}')
+    elif not game_stats['24H'] and game_stats['All']:
+        message += f'用户 {user_name} 暂无24小时内统计数据, 历史统计数据为: '
+        message += f'\nL\'PM: {game_stats["All"]["LPM"]} ( {game_stats["All"]["PPS"]} pps )'
+        message += f'\nAPM: {game_stats["All"]["APM"]} ( x{game_stats["All"]["APL"]} )'
+    else:
+        message += f'用户 {user_name} 暂无24小时内统计数据, 暂无历史统计数据'
+    return message

nonebot_plugin_tetris_stats/game_data_processor/tos_data_processor.py

@@ -0,0 +1,179 @@
+from asyncio import gather
+from re import I
+from typing import Any
+
+import aiohttp
+from nonebot import on_regex
+from nonebot.adapters.onebot.v11 import GROUP, MessageEvent
+from nonebot.log import logger
+from nonebot.matcher import Matcher
+
+from ..utils.message_analyzer import handle_stats_query_message
+
+TOSStats = on_regex(
+    pattern=r'^tos查|^tostats|^tosstats|^茶服查|^茶服stats',
+    flags=I,
+    permission=GROUP
+)
+
+
+@TOSStats.handle()
+async def _(event: MessageEvent, matcher: Matcher):
+    decoded_message = await handle_stats_query_message(message=event.raw_message, game_type='TOS')
+    if decoded_message[0] is None:
+        await matcher.finish(decoded_message[1][0])
+    elif decoded_message[0] == 'AT' or decoded_message[0] == 'QQ':
+        if decoded_message[1][1] == event.self_id:
+            await matcher.finish('不能查询bot的信息')
+        message = await generate_message(tea_id=decoded_message[1][1])
+    elif decoded_message[0] == 'ME':
+        message = await generate_message(tea_id=event.sender.user_id)
+    elif decoded_message[0] == 'Name':
+        message = await generate_message(user_name=decoded_message[1][1])
+    else:
+        raise ValueError('预期外行为, 请上报GitHub')
+    await matcher.finish(message)
+
+
+async def request(url: str) -> tuple[bool, bool, dict[str, Any]]:
+    '''请求api'''
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url) as resp:
+                data = await resp.json()
+                return True, data['success'], data
+    except aiohttp.client_exceptions.ClientConnectorError as error: # type: ignore
+        logger.error(f'请求错误\n{error}')
+        return False, False, {}
+
+
+async def get_user_info(
+    user_name: str | None = None,
+    tea_id: int | None = None
+) -> tuple[bool, bool, dict[str, Any]]:
+    '''获取用户信息'''
+    if user_name is not None and tea_id is None:
+        user_data_url = f'https://teatube.cn:8888/getUsernameInfo?username={user_name}'
+    elif user_name is None and tea_id is not None:
+        user_data_url = f'https://teatube.cn:8888/getTeaIdInfo?teaId={tea_id}'
+    else:
+        raise ValueError('预期外行为, 请上报GitHub')
+    return await request(user_data_url)
+
+
+async def get_user_data(
+    user_name: str | None = None,
+    tea_id: int | None = None,
+    other_parameter: str = ''
+) -> tuple[bool, bool, dict[str, Any]]:
+    '''获取用户数据'''
+    if user_name is not None and tea_id is None:
+        user_data_url = f'https://teatube.cn:8888/getProfile?id={user_name}{other_parameter}'
+    elif user_name is None and tea_id is not None:
+        user_data_url = f'https://teatube.cn:8888/getProfile?id={tea_id}{other_parameter}'
+    else:
+        raise ValueError('预期外行为, 请上报GitHub')
+    return await request(user_data_url)
+
+
+async def get_rank_stats(user_info: dict) -> dict[str, float]:
+    '''获取Rank数据'''
+    data = user_info['data']
+    rank_stats = {}
+    if int(data['rankedGames']) != 0:
+        rank_stats['Rating'] = round(float(data['ratingNow']), 2)
+        rank_stats['RD'] = round(float(data['rdNow']), 2)
+        rank_stats['Vol'] = round(float(data['volNow']), 3)
+    return rank_stats
+
+
+async def get_game_data(user_data: dict) -> dict[str, int | float]:
+    '''获取游戏数据'''
+    game_data: dict[str, int | float] = {}
+    if user_data['data'] != []:
+        weighted_total_lpm = weighted_total_apm = weighted_total_adpm = total_time = num = 0
+        for i in user_data['data']:
+            # 排除单人局和时间为0的游戏
+            if i['num_players'] == 1 or i['time'] == 0:
+                continue
+            # 茶：不计算没挖掘的局, 即使apm和lpm也如此
+            if i['dig'] is None:
+                continue
+            # 加权计算
+            time = i['time'] / 1000
+            lpm = 24 * (i['pieces'] / time)
+            apm = (i['attack'] / time) * 60
+            adpm = ((i['attack'] + i['dig']) / time) * 60
+            weighted_total_lpm += lpm * time
+            weighted_total_apm += apm * time
+            weighted_total_adpm += adpm * time
+            total_time += time
+            num += 1
+            if num == 50:
+                break
+        if num > 0:
+            game_data['NUM'] = num
+            game_data['LPM'] = round((weighted_total_lpm / total_time), 2)
+            game_data['APM'] = round((weighted_total_apm / total_time), 2)
+            game_data['ADPM'] = round((weighted_total_adpm / total_time), 2)
+            game_data['PPS'] = round((game_data['LPM'] / 24), 2)
+            game_data['APL'] = round((game_data['APM'] / game_data['LPM']), 2)
+            game_data['ADPL'] = round(
+                (game_data['ADPM'] / game_data['LPM']), 2)
+            game_data['VS'] = round((game_data['ADPM'] / 60 * 100), 2)
+        # TODO: 如果有效局数不满50, 没有无dig信息的局, 且userData['data']内有50个局, 则继续往前获取信息
+    return game_data
+
+
+async def get_pb_data(user_info: dict) -> dict[str, float | str]:
+    '''获取PB数据'''
+    pb_data: dict[str, float | str] = {}
+    data = user_info['data']
+    if int(data['PBSprint']) != 2147483647:
+        pb_data['Sprint'] = round(float(data['PBSprint']) / 1000, 2)
+    if int(data['PBMarathon']) != 0:
+        pb_data['Marathon'] = data['PBMarathon']
+    if int(data['PBChallenge']) != 0:
+        pb_data['Challenge'] = data['PBChallenge']
+    return pb_data
+
+
+async def generate_message(
+    user_name: str | None = None,
+    tea_id: int | None = None
+) -> str:
+    '''生成消息'''
+    user_info, user_data = await gather(
+        get_user_info(user_name=user_name, tea_id=tea_id),
+        get_user_data(user_name=user_name, tea_id=tea_id)
+    )
+    if user_info[0] is False:
+        return '用户信息请求失败'
+    if user_info[1] is False:
+        return f'用户信息请求错误:\n{user_info[2]["error"]}'
+    rank_stats, pb_data = await gather(
+        get_rank_stats(user_info[2]),
+        get_pb_data(user_info[2])
+    )
+    message = f'用户 {user_info[2]["data"]["name"]} ({user_info[2]["data"]["teaId"]}) '
+    if not rank_stats:
+        message += '暂无段位统计数据'
+    else:
+        message += f', 段位分 {rank_stats["Rating"]}±{rank_stats["RD"]} ({rank_stats["Vol"]}) '
+    if user_data[0] is False:
+        message = f'{message.rstrip()}\n游戏数据请求失败'
+    elif user_data[1] is False:
+        message = f'{message.rstrip()}\n游戏数据请求错误:\n{user_data[2]["error"]}'
+    else:
+        game_data = await get_game_data(user_data[2])
+        if not game_data:
+            message += ', 暂无游戏数据'
+        else:
+            message += f', 最近 {game_data["NUM"]} 局数据'
+            message += f'\nL\'PM: {game_data["LPM"]} ( {game_data["PPS"]} pps )'
+            message += f'\nAPM：{game_data["APM"]} ( x{game_data["APL"]} )'
+            message += f'\nADPM：{game_data["ADPM"]} ( x{game_data["ADPL"]} ) ( {game_data["VS"]}vs )'
+    message += f'\n40L: {pb_data["Sprint"]}s' if 'Sprint' in pb_data else ''
+    message += f'\nMarathon: {pb_data["Marathon"]}' if 'Marathon' in pb_data else ''
+    message += f'\nChallenge: {pb_data["Challenge"]}' if 'Challenge' in pb_data else ''
+    return message

nonebot_plugin_tetris_stats/utils/__init__.py

@@ -0,0 +1 @@
+from . import config, database, message_analyzer

nonebot_plugin_tetris_stats/utils/config.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+
+
+class Config(BaseModel):
+    '''配置类'''
+    cache_path: str = 'cache/nonebot_plugin_tetris_stats/cache'
+    db_path: str = 'data/nonebot_plugin_tetris_stats/data.db'

nonebot_plugin_tetris_stats/utils/database.py

@@ -0,0 +1,140 @@
+import datetime
+import os
+from asyncio import gather
+from sqlite3 import Connection, connect
+
+from nonebot import get_driver
+from nonebot.log import logger
+
+from .config import Config
+
+driver = get_driver()
+
+config = Config.parse_obj(get_driver().config)
+
+
+@driver.on_startup
+async def _():
+    await DataBase.init_db()
+
+
+@driver.on_shutdown
+async def _():
+    await DataBase.close_db()
+
+
+class DataBase():
+    '''数据库交互类'''
+    _db: Connection | None = None
+
+    @classmethod
+    async def init_db(cls) -> Connection:
+        '''初始化数据库'''
+        if not os.path.exists(os.path.dirname(config.db_path)):
+            os.makedirs(os.path.dirname(config.db_path))
+        cls._db = connect(config.db_path)
+        cursor = cls._db.cursor()
+        cursor.execute('''CREATE TABLE IF NOT EXISTS IOBIND
+                        (QQ INTEGER NOT NULL,
+                        USER TEXT NOT NULL)''')
+        cursor.execute('''CREATE TABLE IF NOT EXISTS TOPBIND
+                        (QQ INTEGER NOT NULL,
+                        USER TEXT NOT NULL)''')
+        cursor.execute('''CREATE TABLE IF NOT EXISTS IORANK
+                        (RANK VARCHAR(2) NOT NULL,
+                         TRENDING CHAR(1) NOT NULL,
+                         TRLINE FLOAT NOT NULL,
+                         PLAYERCOUNT INTEGER NOT NULL,
+                         AVGAPM FLOAT NOT NULL,
+                         AVGPPS FLOAT NOT NULL,
+                         ARGVS FLOAT NOT NULL,
+                         DATE TEXT NOT NULL)''')
+        cls._db.commit()
+        logger.info('数据库初始化完成')
+        return cls._db
+
+    @classmethod
+    async def query_rank_info_today(cls, rank: str) -> list | None:
+        '''查询段位信息'''
+        db = await cls._get_db()
+        cursor = db.cursor()
+        cursor.execute('''SELECT TRENDING, TRLINE, PLAYERCOUNT, AVGAPM, AVGPPS, ARGVS, DATE
+                        FROM IORANK
+                        WHERE RANK = ? AND DATE = ?''', (rank, datetime.date.today()))
+        result = cursor.fetchone()
+
+        if result is None:
+            return None
+
+        return list(result)
+
+    @classmethod
+    async def write_rank_info_today(cls, rank: str, trline: int, playercount: int, avgapm: float, avgpps: float, avgvs: float):
+        '''写入段位信息'''
+
+        db = await cls._get_db()
+        cursor = db.cursor()
+        cursor.execute('''SELECT TRLINE
+                        FROM IORANK
+                        WHERE RANK = ? AND DATE = ?''', (rank, datetime.date.today()))
+
+        result = cursor.fetchone()
+
+        if result is None:
+            trending = '?'
+        else:
+            if result[0] > trline:
+                trending = '↑'
+            else:
+                trending = '↓'
+
+        cursor.execute('''INSERT INTO IORANK
+                        (RANK, TRENDING, TRLINE, PLAYERCOUNT, AVGAPM, AVGPPS, ARGVS, DATE)
+                        VALUES (?,?,?,?,?,?,?,?)''',
+                       (rank, trending, trline, playercount, avgapm, avgpps, avgvs, datetime.date.today()))
+
+        db.commit()
+
+    @classmethod
+    async def _get_db(cls) -> Connection:
+        '''获取数据库对象'''
+        return cls._db or await cls.init_db()
+
+    @classmethod
+    async def query_bind_info(cls, qq_number: str | int, game_type: str) -> str | None:
+        '''查询绑定信息'''
+        db = await cls._get_db()
+        cursor = db.cursor()
+        cursor.execute(
+            f'SELECT USER FROM {game_type}BIND WHERE QQ = {qq_number}')
+        user = cursor.fetchone()
+        if user is None:
+            return None
+        return user[0]
+
+    @classmethod
+    async def write_bind_info(cls, qq_number: str | int, user: str, game_type: str) -> str:
+        '''写入绑定信息'''
+        bind_info, db = await gather(
+            cls.query_bind_info(qq_number=qq_number, game_type=game_type),
+            cls._get_db()
+        )
+        cursor = db.cursor()
+        if bind_info is not None:
+            cursor.execute(
+                f'UPDATE {game_type}BIND SET USER = ? WHERE QQ = ?', (user, qq_number))
+            message = '更新成功'
+        elif bind_info is None:
+            cursor.execute(
+                f'INSERT INTO {game_type}BIND (QQ, USER) VALUES (?, ?)', (qq_number, user))
+            message = '绑定成功'
+        else:
+            raise ValueError('预期外行为, 请上报GitHub')
+        db.commit()
+        return message
+
+    @classmethod
+    async def close_db(cls) -> None:
+        '''关闭数据库对象'''
+        if isinstance(cls._db, Connection):
+            cls._db.close()

nonebot_plugin_tetris_stats/utils/message_analyzer.py

@@ -0,0 +1,91 @@
+from re import match, sub
+
+
+async def handle_bind_message(message: str, game_type: str) -> tuple[str | None, tuple]:
+    '''返回值为tuple[gameType, tuple[message, user]]'''
+    _cmd_aliases = {'IO': ['io绑定', 'iobind'],
+                    'TOP': ['top绑定', 'topbind']}
+    # 剔除命令前缀
+    for i in _cmd_aliases[game_type]:
+        if match(rf'(?i){i}', message):
+            message = sub(rf'(?i){i}', '', message)
+            message = message.strip()
+            break
+    else:
+        raise ValueError('预期外行为, 请上报GitHub')
+    if message == '' or message.isspace():
+        return None, ('用户名为空', None)
+    return await check_name(message, game_type)
+
+
+async def handle_stats_query_message(message: str, game_type: str) -> tuple[str | None, tuple]:
+    '''返回值为tuple[gameType, tuple[message, user]]'''
+    _cmd_aliases = {'IO': ['io查', 'iostats'],
+                    'TOS': ['tos查', 'tostats', 'tosstats', '茶服查', '茶服stats'],
+                    'TOP': ['top查', 'topstats']}
+    _me = [
+        '我', '自己', '我等', '卑人', '愚', '老身', '爷', '老娘', '本姑娘', '本大爷', '鄙人', '寡人',
+        '小生', '贫僧', '本人', '孤', '吾', '俺', '咱', '私', 'me', '洒家', '在下', '偶', '人家',
+        '本小姐', '老夫', '老子', '朕', '本尊', '僕', '拙者', '妾', '儂', '自分', '吾輩', '我輩', '某',
+        '己等', '俺等', '此方', '哥', '姐', '劳资', '本宝宝', '余', '本喵',  'watashi',  'i', 'myself',
+        'self', 'oneself'
+    ]
+    # 剔除命令前缀
+    for i in _cmd_aliases[game_type]:
+        if match(rf'(?i){i}', message):
+            message = sub(rf'(?i){i}', '', message)
+            message = message.strip()
+            break
+    if message == '' or message.isspace():
+        return None, ('用户名为空', None)
+    if message.startswith('[CQ:at,qq='):
+        try:
+            user = int(str(message).split('[CQ:at,qq=')[1].split(']')[0])
+        except ValueError:
+            return None, ('QQ号码不合法', None)
+        else:
+            return 'AT', (None, user)
+    elif message in _me:
+        # 会不会有人叫本姑娘 本大爷这种或许可以成为id的名字呢
+        # TODO: 在判断是否可能是查自己的情况的时候 也去判断是否能成立为一个UserName
+        return 'ME', (None, None)
+    else:
+        return await check_name(message, game_type)
+
+
+async def handle_rank_message(message: str) -> str:
+    _cmd_aliases = ['io段位', 'iorank']
+    # 剔除命令前缀
+    for i in _cmd_aliases:
+        if match(rf'(?i){i}', message):
+            message = sub(rf'(?i){i}', '', message)
+            message = message.strip()
+            break
+    else:
+        raise ValueError('预期外行为, 请上报GitHub')
+    return message
+
+
+async def check_name(name: str, game_type: str) -> tuple[str | None, tuple]:
+    '''返回值为tuple[gameType, tuple[message, user]]'''
+    if game_type == 'IO':
+        if match(r'^[a-f0-9]{24}$', name):
+            return 'ID', (None, name)
+        if match(r'^[a-zA-Z0-9_-]{3,16}$', name):
+            return 'Name', (None, name.lower())
+        return None, ('用户名不合法', None)
+    if game_type == 'TOP':
+        if match(r'^[a-zA-Z0-9_]{1,16}$', name):
+            return 'Name', (None, name)
+        return None, ('用户名不合法', None)
+    if game_type == 'TOS':
+        if (match(r'^(?!\.)(?!com[0-9]$)(?!con$)(?!lpt[0-9]$)(?!nul$)(?!prn$)[^\-][^\+][^\|\*\?\\\s\!:<>/$"]*[^\.\|\*\?\\\s\!:<>/$"]+$', name)
+            and name.isdigit() is False
+                and 2 <= len(name) <= 18):
+            # 虽然我也不想这么长 但是似乎确实得这么长
+            # TODO 简化正则表达式
+            return 'Name', (None, name)
+        if name.isdigit() is True:
+            return 'QQ', (None, name)
+        return None, ('用户名不合法', None)
+    return None, ('游戏类型错误', None)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nonebot-plugin-tetris-stats"
-version = "0.1.0"
+version = "0.4.0.post1"
 description = "一个基于nonebot2的用于查询TETRIS相关游戏玩家数据的插件"
 authors = ["scdhh <wallfjjd@gmail.com>"]
 readme = "README.md"
@@ -9,14 +9,24 @@ repository = "https://github.com/shoucandanghehe/nonebot-plugin-tetris-stats"
 license = "MIT"
 
 [tool.poetry.dependencies]
-python = "^3.10"
-nb-cli = "^0.6.7"
+python = "^3.10,<3.12"
 nonebot-adapter-onebot = "^2.0.0-beta.1"
 aiohttp = "^3.8.1"
 asyncio = "^3.4.3"
+nonebot2 = "^2.0.0-beta.3"
+lxml = "^4.9.1"
+pandas = "^1.4.3"
+playwright = "^1.24.1"
+ujson = "^5.4.0"
+Brotli = "^1.0.9"
 
 [tool.poetry.dev-dependencies]
-mypy = "^0.950"
+mypy = "^0.991"
+autopep8 = "^2.0.0"
+pylint = "^2.15.9"
+types-ujson = "^5.5.0"
+lxml-stubs = "^0.4.0"
+pandas-stubs = "^1.5.2"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]