✨ 使用新版模板 (#313)

* 🔥 删除现有模板

* ✨ 自动克隆模板仓库

* 🔥 删除 identicon 相关代码

* 🚚 修改静态文件路径

* ✨ 使用新模板进行渲染

* ✨ 每次渲染都获取一次模板, 以应对实时更新

* ✨ TETR.IO 绑定图使用新模板

* 🚚 修改网络路径

* ✨ TOP 绑定图使用新模板

* ✨ TOS 绑定图使用新模板

* 🐛 防止截图超时

* 🐛 Pydantic V1 会把 float 转换成 int

* ✏️ 模板字段名写错了

* ✨ 兼容 Pydantic V1

* ✨ TETR.IO 查询图使用新模板

* 🐛 在查询的用户没有历史记录时不去查询更多记录

nonebot_plugin_tetris_stats/utils/avatar.py

@@ -1,13 +1,10 @@
-from base64 import b64decode, b64encode
+from base64 import b64encode
 from io import BytesIO
 from typing import Literal, overload
 
 from nonebot_plugin_userinfo import UserInfo  # type: ignore[import-untyped]
 from PIL import Image
 
-from ..templates import path
-from .browser import BrowserManager
-
 
 @overload
 async def get_avatar(user: UserInfo, scheme: Literal['Data URI'], default: str | None) -> str:
@@ -53,29 +50,3 @@ async def get_avatar(user: UserInfo, scheme: Literal['Data URI', 'bytes'], defau
             raise TypeError("Can't get avatar format")
         return f'data:{Image.MIME[avatar_format]};base64,{b64encode(bot_avatar).decode()}'
     return bot_avatar
-
-
-async def generate_identicon(hash: str) -> bytes:  # noqa: A002
-    """使用 identicon 生成头像
-
-    Args:
-        hash (str): 提交给 identicon 的 hash 值
-
-    Returns:
-        bytes: identicon 生成的 svg 的二进制数据
-    """
-    browser = await BrowserManager.get_browser()
-    async with await browser.new_page() as page:
-        await page.add_script_tag(path=path / 'js/identicon.js')
-        return b64decode(
-            await page.evaluate(rf"""
-                new Identicon('{hash}', {{
-                    background: [0x08, 0x0a, 0x06, 255],
-                    margin: 0.15,
-                    size: 300,
-                    brightness: 0.48,
-                    saturation: 0.65,
-                    format: 'svg',
-                }}).toString();
-                """)
-        )

nonebot_plugin_tetris_stats/utils/host.py

@@ -2,17 +2,14 @@ from hashlib import sha256
 from ipaddress import IPv4Address, IPv6Address
 from typing import ClassVar
 
-from aiofiles import open
-from fastapi import FastAPI, Query, Response, status
-from fastapi.responses import FileResponse, HTMLResponse
+from fastapi import FastAPI, status
+from fastapi.responses import HTMLResponse
 from fastapi.staticfiles import StaticFiles
 from nonebot import get_app, get_driver
-from nonebot.log import logger
 from nonebot_plugin_localstore import get_cache_dir  # type: ignore[import-untyped]
 from pydantic import IPvAnyAddress
 
-from ..templates import path
-from .avatar import generate_identicon
+from .templates import templates_dir
 
 app = get_app()
 
@@ -43,32 +40,19 @@ class HostPage:
 
 
 app.mount(
-    '/static',
-    StaticFiles(directory=path),
-    name='static',
+    '/host/assets',
+    StaticFiles(directory=templates_dir / 'assets'),
+    name='assets',
 )
 
 
-@app.get('/host/page/{page_hash}.html', status_code=status.HTTP_200_OK)
+@app.get('/host/{page_hash}.html', status_code=status.HTTP_200_OK)
 async def _(page_hash: str) -> HTMLResponse:
     if page_hash in HostPage.pages:
         return HTMLResponse(HostPage.pages[page_hash])
     return NOT_FOUND
 
 
-@app.get('/identicon')
-async def _(md5: str = Query(regex=r'^[a-fA-F0-9]{32}$')):
-    identicon_path = cache_dir / 'identicon' / f'{md5}.svg'
-    if identicon_path.exists() is False:
-        identicon_path.parent.mkdir(parents=True, exist_ok=True)
-        result = await generate_identicon(md5)
-        async with open(identicon_path, mode='xb') as file:
-            await file.write(result)
-        return Response(result, media_type='image/svg+xml')
-    logger.debug('Identicon Cache hit!')
-    return FileResponse(identicon_path, media_type='image/svg+xml')
-
-
 def get_self_netloc() -> str:
     host: IPv4Address | IPv6Address | IPvAnyAddress = global_config.host
     if isinstance(host, IPv4Address):

nonebot_plugin_tetris_stats/utils/render.py

@@ -1,69 +1,104 @@
-from typing import Any, Literal, overload
+from datetime import datetime
+from typing import Annotated, ClassVar, Literal, overload
 
 from jinja2 import Environment, FileSystemLoader
+from nonebot.compat import PYDANTIC_V2
+from pydantic import BaseModel
 
 from ..game_data_processor.io_data_processor.typing import Rank
-from ..templates import path
-from .typing import GameType
+from .templates import templates_dir
+from .typing import Number
 
-Bind = Literal['bind.j2.html']
-Data = Literal['data.j2.html']
+if PYDANTIC_V2:
+    from pydantic import PlainSerializer
 
 env = Environment(
-    loader=FileSystemLoader(path), autoescape=True, trim_blocks=True, lstrip_blocks=True, enable_async=True
+    loader=FileSystemLoader(templates_dir), autoescape=True, trim_blocks=True, lstrip_blocks=True, enable_async=True
 )
 
 
-@overload
-async def render(
-    template: Bind,
-    *,
-    user_avatar: str,
-    state: Literal['error', 'success', 'unknown', 'unlink', 'unverified'],
-    bot_avatar: str,
-    game_type: GameType,
-    user_name: str,
-    bot_name: str,
-    command: str,
-) -> str: ...
+def format_datetime_to_timestamp(dt: datetime) -> int:
+    return int(dt.timestamp() * 1000)
+
+
+class Bind(BaseModel):
+    class People(BaseModel):
+        avatar: str
+        name: str
+
+    platform: Literal['TETR.IO', 'TOP', 'TOS']
+    status: Literal['error', 'success', 'unknown', 'unlink', 'unverified']
+    user: People
+    bot: People
+    command: str
+
+
+class TETRIOInfo(BaseModel):
+    class User(BaseModel):
+        avatar: str
+        name: str
+        bio: str | None
+
+    class Ranking(BaseModel):
+        rating: Number
+        rd: Number
+
+    class TetraLeague(BaseModel):
+        rank: Rank
+        tr: Number
+        global_rank: Number
+        pps: Number
+        lpm: Number
+        apm: Number
+        apl: Number
+        vs: Number
+        adpm: Number
+        adpl: Number
+
+    class TetraLeagueHistory(BaseModel):
+        class Data(BaseModel):
+            if PYDANTIC_V2:
+                record_at: Annotated[datetime, PlainSerializer(format_datetime_to_timestamp, return_type=int)]
+            else:
+                record_at: datetime  # type: ignore[no-redef]
+            tr: Number
+
+        data: list[Data]
+        split_interval: Number
+        min_tr: Number
+        max_tr: Number
+        offset: Number
+
+    class Radar(BaseModel):
+        app: Number
+        dsps: Number
+        dspp: Number
+        ci: Number
+        ge: Number
+
+    user: User
+    ranking: Ranking
+    tetra_league: TetraLeague
+    tetra_league_history: TetraLeagueHistory
+    radar: Radar
+    sprint: str
+    blitz: str
+
+    if not PYDANTIC_V2:
+
+        class Config:
+            json_encoders: ClassVar[dict] = {datetime: format_datetime_to_timestamp}
 
 
 @overload
-async def render(
-    template: Data,
-    *,
-    user_avatar: str,
-    user_name: str,
-    user_sign: str | None,
-    game_type: Literal['TETR.IO'],
-    ranking: str | float,
-    rd: str | float,
-    rank: Rank,
-    TR: str | float,  # noqa: N803
-    global_rank: str | int,
-    lpm: str | float,
-    pps: str | float,
-    apm: str | float,
-    apl: str | float,
-    adpm: str | float,
-    adpl: str | float,
-    vs: str | float,
-    sprint: str,
-    blitz: str,
-    data: list[list[int | float]],
-    split_value: int,
-    offset: int,
-    value_max: int,
-    value_min: int,
-    app: str | float,
-    dsps: str | float,
-    dspp: str | float,
-    ci: str | float,
-    ge: str | float,
-) -> str: ...
+async def render(render_type: Literal['binding'], data: Bind) -> str: ...
 
 
-async def render(template: Bind | Data, **kwargs: Any) -> str:
-    if kwargs['game_type'] == 'IO':
-        kwargs['game_type'] = 'TETR.IO'
-    return await env.get_template(template).render_async(**kwargs)
+@overload
+async def render(render_type: Literal['tetrio/info'], data: TETRIOInfo) -> str: ...
+
+
+async def render(render_type: Literal['binding', 'tetrio/info'], data: Bind | TETRIOInfo) -> str:
+    if PYDANTIC_V2:
+        return await env.get_template('index.html').render_async(path=render_type, data=data.model_dump_json())
+    return await env.get_template('index.html').render_async(path=render_type, data=data.json())

nonebot_plugin_tetris_stats/utils/screenshot.py

@@ -7,4 +7,5 @@ async def screenshot(url: str) -> bytes:
         await browser.new_page(no_viewport=True, viewport={'width': 0, 'height': 0}) as page,
     ):
         await page.goto(url)
+        await page.wait_for_load_state('networkidle')
         return await page.screenshot(full_page=True, type='png')

nonebot_plugin_tetris_stats/utils/templates.py

@@ -0,0 +1,60 @@
+from asyncio.subprocess import PIPE, create_subprocess_exec
+from shutil import rmtree
+
+from nonebot import get_driver
+from nonebot.log import logger
+from nonebot_plugin_localstore import get_data_dir  # type: ignore[import-untyped]
+
+driver = get_driver()
+
+templates_dir = get_data_dir('nonebot_plugin_tetris_stats') / 'templates'
+
+
+@driver.on_startup
+async def init_templates() -> None:
+    try:
+        await create_subprocess_exec('git', '--version', stdout=PIPE)
+    except FileNotFoundError as e:
+        raise RuntimeError(
+            '未找到 git, 请确保 git 已安装并在环境变量中\n安装步骤请参阅: https://git-scm.com/book/zh/v2/%E8%B5%B7%E6%AD%A5-%E5%AE%89%E8%A3%85-Git'
+        ) from e
+    if not templates_dir.exists():
+        logger.info('模板仓库不存在, 正在尝试初始化...')
+        proc = await create_subprocess_exec(
+            'git',
+            'clone',
+            '-b',
+            'gh-pages',
+            'https://github.com/A-Minos/tetris-stats-templates',
+            templates_dir,
+            '--depth=1',
+            stdout=PIPE,
+            stderr=PIPE,
+        )
+        stdout, stderr = await proc.communicate()
+        if proc.returncode != 0:
+            for i in stderr.decode().splitlines():
+                logger.error(i)
+            raise RuntimeError('初始化模板仓库失败')
+        logger.success('模板仓库初始化成功')
+        return
+    proc = await create_subprocess_exec(
+        'git', 'rev-parse', '--is-inside-work-tree', stdout=PIPE, stderr=PIPE, cwd=templates_dir
+    )
+    stdout, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        for i in stderr.decode().splitlines():
+            logger.error(i)
+        logger.warning('模板仓库状态异常, 尝试重新初始化')
+        rmtree(templates_dir)
+        await init_templates()
+        return
+    logger.info('正在更新模板仓库...')
+    proc = await create_subprocess_exec('git', 'pull', stdout=PIPE, stderr=PIPE, cwd=templates_dir)
+    stdout, stderr = await proc.communicate()
+    logger.info(stdout.decode().strip())
+    if proc.returncode != 0:
+        for i in stderr.decode().splitlines():
+            logger.error(i)
+        raise RuntimeError('更新模板仓库失败')
+    logger.success('模板仓库更新成功')

nonebot_plugin_tetris_stats/utils/typing.py

@@ -1,7 +1,7 @@
 from collections.abc import Awaitable, Callable
 from typing import Any, Literal
 
-Number = int | float
+Number = float | int
 GameType = Literal['IO', 'TOP', 'TOS']
 CommandType = Literal['bind', 'query']
 AsyncCallable = Callable[..., Awaitable[Any]]