A-Minos/nonebot-plugin-tetris-stats
1
0
Fork 0
mirror of https://github.com/A-Minos/nonebot-plugin-tetris-stats.git synced 2026-03-05 05:36:54 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

✨ 支持设置代理 (#407)

Browse Source 
* ➕ 添加依赖 yarl

* ➕ 添加依赖 msgspec

* ➖ 移除依赖 ujson

* ♻️ 重构 request 使其支持分别设置代理

* ♻️ 重构 resource 接口

* ⚡️ 不再重复获取 Config

* ♻️ 使用 yarl 替换 urllib.parse

* ⚡️ 给 get_self_netloc 加个 cache

* ✨ request 使用 proxy

* ✨ 更新模板使用 proxy

* 🐛 修复删除 ujson 依赖后 迁移脚本报错的bug
This commit is contained in:
呵呵です
2024-08-20 07:37:51 +08:00
committed by GitHub
parent 414345ae5c
commit 259b38fda5
21 changed files with 258 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
nonebot_plugin_tetris_stats/config/config.py
View File

@@ -1,3 +1,4 @@
from nonebot import get_plugin_config
from nonebot_plugin_localstore import get_cache_dir, get_data_dir
from pydantic import BaseModel, Field
@@ -5,10 +6,22 @@ CACHE_PATH = get_cache_dir('nonebot_plugin_tetris_stats')
DATA_PATH = get_data_dir('nonebot_plugin_tetris_stats')
class Proxy(BaseModel):
main: str | None = None
github: str | None = None
tetrio: str | None = None
tos: str | None = None
top: str | None = None
class ScopedConfig(BaseModel):
request_timeout: float = 30.0
screenshot_quality: float = 2
proxy: Proxy = Field(default_factory=Proxy)
class Config(BaseModel):
tetris: ScopedConfig = Field(default_factory=ScopedConfig)
config = get_plugin_config(Config)

3
nonebot_plugin_tetris_stats/config/migrations/3c25a5a8c050_refactor_historical.py
View File

@@ -19,7 +19,6 @@ from sqlalchemy import desc, select
from sqlalchemy.dialects import sqlite
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session
from ujson import dumps, loads
if TYPE_CHECKING:
from collections.abc import Sequence
@@ -31,6 +30,8 @@ depends_on: str | Sequence[str] | None = None
def migrate_old_data() -> None:
from json import dumps, loads
Base = automap_base() # noqa: N806
Base.prepare(autoload_with=op.get_bind())
OldHistoricalData = Base.classes.nonebot_plugin_tetris_stats_historicaldata # noqa: N806

3
nonebot_plugin_tetris_stats/config/migrations/6c3206f90cc3_add_field.py
View File

@@ -13,7 +13,6 @@ from typing import TYPE_CHECKING
from alembic import op
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session
from ujson import dumps, loads
if TYPE_CHECKING:
from collections.abc import Sequence
@@ -27,6 +26,7 @@ depends_on: str | Sequence[str] | None = None
def upgrade(name: str = '') -> None:
if name:
return
from json import dumps, loads
Base = automap_base() # noqa: N806
connection = op.get_bind()
@@ -50,6 +50,7 @@ def upgrade(name: str = '') -> None:
def downgrade(name: str = '') -> None:
if name:
return
from json import dumps, loads
Base = automap_base() # noqa: N806
connection = op.get_bind()

10
nonebot_plugin_tetris_stats/games/tetrio/api/cache.py
View File

@@ -6,25 +6,29 @@ from weakref import WeakValueDictionary
from aiocache import Cache as ACache # type: ignore[import-untyped]
from nonebot.compat import type_validate_json
from nonebot.log import logger
from yarl import URL
from ....config.config import config
from ....utils.request import Request
from .schemas.base import FailedModel, SuccessModel
UTC = timezone.utc
request = Request(config.tetris.proxy.tetrio or config.tetris.proxy.main)
class Cache:
cache = ACache(ACache.MEMORY)
task: ClassVar[WeakValueDictionary[str, Lock]] = WeakValueDictionary()
task: ClassVar[WeakValueDictionary[URL, Lock]] = WeakValueDictionary()
@classmethod
async def get(cls, url: str) -> bytes:
async def get(cls, url: URL) -> bytes:
lock = cls.task.setdefault(url, Lock())
async with lock:
if (cached_data := await cls.cache.get(url)) is not None:
logger.debug(f'{url}: Cache hit!')
return cached_data
response_data = await Request.request(url)
response_data = await request.request(url)
parsed_data: SuccessModel | FailedModel = type_validate_json(SuccessModel | FailedModel, response_data) # type: ignore[arg-type]
if isinstance(parsed_data, SuccessModel):
await cls.cache.add(

23
nonebot_plugin_tetris_stats/games/tetrio/api/player.py
View File

@@ -7,7 +7,6 @@ from nonebot.compat import type_validate_json
from ....db import anti_duplicate_add
from ....utils.exception import RequestError
from ....utils.request import splice_url
from ..constant import BASE_URL, USER_ID, USER_NAME
from .cache import Cache
from .models import TETRIOHistoricalData
@@ -88,12 +87,7 @@ class Player:
@property
def _request_user_parameter(self) -> str:
if self.user_id is not None:
return self.user_id
if self.user_name is not None:
return self.user_name.lower()
msg = 'Invalid user'
raise ValueError(msg)
return self.user_id or cast(str, self.user_name).lower()
@property
async def user(self) -> User:
@@ -117,7 +111,7 @@ class Player:
async def get_info(self) -> UserInfoSuccess:
"""Get User Info"""
if self._user_info is None:
raw_user_info = await Cache.get(splice_url([BASE_URL, 'users/', f'{self._request_user_parameter}']))
raw_user_info = await Cache.get(BASE_URL / 'users' / self._request_user_parameter)
user_info: UserInfo = type_validate_json(UserInfo, raw_user_info) # type: ignore[arg-type]
if isinstance(user_info, FailedModel):
msg = f'用户信息请求错误:\n{user_info.error}'
@@ -147,7 +141,7 @@ class Player:
async def get_summaries(self, summaries_type: Summaries) -> SummariesModel:
if summaries_type not in self._summaries:
raw_summaries = await Cache.get(
splice_url([BASE_URL, 'users/', f'{self._request_user_parameter}/', 'summaries/', summaries_type])
BASE_URL / 'users' / self._request_user_parameter / 'summaries' / summaries_type
)
summaries: SummariesModel | FailedModel = type_validate_json(
self.__SUMMARIES_MAPPING[summaries_type] | FailedModel, # type: ignore[arg-type]
@@ -217,16 +211,7 @@ class Player:
async def get_records(self, mode_type: RecordModeType, records_type: RecordType) -> RecordsSoloSuccessModel:
if (record_key := RecordKey(mode_type, records_type)) not in self._records:
raw_records = await Cache.get(
splice_url(
[
BASE_URL,
'users/',
f'{self._request_user_parameter}/',
'records/',
f'{mode_type}/',
records_type,
]
)
BASE_URL / 'users' / self._request_user_parameter / 'records' / mode_type / records_type,
)
records: RecordsSoloSuccessModel | FailedModel = type_validate_json(SoloRecord, raw_records) # type: ignore[arg-type]
if isinstance(records, FailedModel):

20
nonebot_plugin_tetris_stats/games/tetrio/api/tetra_league.py
View File

@@ -1,27 +1,26 @@
from typing import Literal, NamedTuple, TypedDict, overload
from urllib.parse import urlencode
from typing import Literal, NamedTuple, overload
from msgspec import Struct, to_builtins
from nonebot.compat import type_validate_json
from ....utils.exception import RequestError
from ....utils.request import splice_url
from ..constant import BASE_URL
from .cache import Cache
from .schemas.base import FailedModel
from .schemas.tetra_league import TetraLeague, TetraLeagueSuccess
class Parameter(TypedDict, total=False):
after: float
before: float
limit: int
country: str
class Parameter(Struct, omit_defaults=True):
after: float | None = None
before: float | None = None
limit: int | None = None
country: str | None = None
async def leaderboard(parameter: Parameter | None = None) -> TetraLeagueSuccess:
league: TetraLeague = type_validate_json(
TetraLeague, # type: ignore[arg-type]
(await Cache.get(splice_url([BASE_URL, 'users/lists/league', f'?{urlencode(parameter or {})}']))),
(await Cache.get(BASE_URL / 'users/lists/league' % to_builtins(parameter))),
)
if isinstance(league, FailedModel):
msg = f'排行榜数据请求错误:\n{league.error}'
@@ -45,8 +44,9 @@ async def full_export(*, with_original: Literal[True]) -> FullExport: ...
async def full_export(*, with_original: bool) -> TetraLeagueSuccess | FullExport:
full: TetraLeague = type_validate_json(
TetraLeague, # type: ignore[arg-type]
(data := await Cache.get(splice_url([BASE_URL, 'users/lists/league/all']))),
(data := await Cache.get(BASE_URL / 'users/lists/league/all')),
)
if isinstance(full, FailedModel):
msg = f'排行榜数据请求错误:\n{full.error}'
raise RequestError(msg)

7
nonebot_plugin_tetris_stats/games/tetrio/bind.py
View File

@@ -1,5 +1,4 @@
from hashlib import md5
from urllib.parse import urlencode
from arclet.alconna import Arg, ArgFlag
from nonebot_plugin_alconna import Args, Subcommand
@@ -9,6 +8,7 @@ from nonebot_plugin_session import EventSession
from nonebot_plugin_session_orm import get_session_persist_id # type: ignore[import-untyped]
from nonebot_plugin_user import User
from nonebot_plugin_userinfo import BotUserInfo, UserInfo
from yarl import URL
from ...db import BindStatus, create_or_update_bind, trigger
from ...utils.host import HostPage, get_self_netloc
@@ -67,7 +67,10 @@ async def _(nb_user: User, account: Player, event_session: EventSession, bot_inf
platform='TETR.IO',
status='unknown',
user=People(
avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": avatar_revision})}'
avatar=str(
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}')
% {'revision': avatar_revision}
)
if (avatar_revision := (await account.avatar_revision)) is not None and avatar_revision != 0
else Avatar(type='identicon', hash=md5(user.ID.encode()).hexdigest()), # noqa: S324
name=user.name.upper(),

4
nonebot_plugin_tetris_stats/games/tetrio/constant.py
View File

@@ -1,11 +1,13 @@
from re import compile
from typing import Literal
from yarl import URL
from .api.typing import ValidRank
GAME_TYPE: Literal['IO'] = 'IO'
BASE_URL = 'https://ch.tetr.io/api/'
BASE_URL = URL('https://ch.tetr.io/api/')
RANK_PERCENTILE: dict[ValidRank, float] = {
'x': 1,

10
nonebot_plugin_tetris_stats/games/tetrio/query.py
View File

@@ -2,7 +2,6 @@ from asyncio import gather
from datetime import datetime, timedelta, timezone
from hashlib import md5
from typing import TYPE_CHECKING, TypeVar
from urllib.parse import urlencode
from arclet.alconna import Arg, ArgFlag
from nonebot import get_driver
@@ -16,6 +15,7 @@ from nonebot_plugin_session_orm import get_session_persist_id # type: ignore[im
from nonebot_plugin_user import User as NBUser
from nonebot_plugin_user import get_user
from sqlalchemy import select
from yarl import URL
from ...db import query_bind_info, trigger
from ...utils.host import HostPage, get_self_netloc
@@ -199,10 +199,14 @@ async def make_query_image_v2(player: Player) -> bytes:
id=user.ID,
name=user.name.upper(),
bio=user_info.data.bio,
banner=f'http://{netloc}/host/resource/tetrio/banners/{user.ID}?{urlencode({"revision": banner_revision})}'
banner=str(
URL(f'http://{netloc}/host/resource/tetrio/banners/{user.ID}') % {'revision': banner_revision}
)
if banner_revision is not None and banner_revision != 0
else None,
avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": avatar_revision})}'
avatar=str(
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
)
if avatar_revision is not None and avatar_revision != 0
else Avatar(
type='identicon',

6
nonebot_plugin_tetris_stats/games/tetrio/record/blitz.py
View File

@@ -1,7 +1,6 @@
from asyncio import gather
from datetime import timedelta
from hashlib import md5
from urllib.parse import urlencode
from nonebot.adapters import Event
from nonebot.matcher import Matcher
@@ -11,6 +10,7 @@ from nonebot_plugin_orm import get_session
from nonebot_plugin_session import EventSession
from nonebot_plugin_session_orm import get_session_persist_id # type: ignore[import-untyped]
from nonebot_plugin_user import get_user
from yarl import URL
from ....db import query_bind_info, trigger
from ....utils.exception import RecordNotFoundError
@@ -94,7 +94,9 @@ async def make_blitz_image(player: Player) -> bytes:
user=User(
id=user.ID,
name=user.name.upper(),
avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": avatar_revision})}'
avatar=str(
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
)
if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0
else Avatar(
type='identicon',

6
nonebot_plugin_tetris_stats/games/tetrio/record/sprint.py
View File

@@ -1,7 +1,6 @@
from asyncio import gather
from datetime import timedelta
from hashlib import md5
from urllib.parse import urlencode
from nonebot.adapters import Event
from nonebot.matcher import Matcher
@@ -11,6 +10,7 @@ from nonebot_plugin_orm import get_session
from nonebot_plugin_session import EventSession
from nonebot_plugin_session_orm import get_session_persist_id # type: ignore[import-untyped]
from nonebot_plugin_user import get_user
from yarl import URL
from ....db import query_bind_info, trigger
from ....utils.exception import RecordNotFoundError
@@ -95,7 +95,9 @@ async def make_sprint_image(player: Player) -> bytes:
user=User(
id=user.ID,
name=user.name.upper(),
avatar=f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}?{urlencode({"revision": avatar_revision})}'
avatar=str(
URL(f'http://{netloc}/host/resource/tetrio/avatars/{user.ID}') % {'revision': avatar_revision}
)
if (avatar_revision := (await player.avatar_revision)) is not None and avatar_revision != 0
else Avatar(
type='identicon',

9
nonebot_plugin_tetris_stats/games/top/api/player.py
View File

@@ -1,13 +1,13 @@
from contextlib import suppress
from datetime import datetime, timezone
from io import StringIO
from urllib.parse import urlencode
from lxml import etree
from pandas import read_html
from ....config.config import config
from ....db import anti_duplicate_add
from ....utils.request import Request, splice_url
from ....utils.request import Request
from ..constant import BASE_URL, USER_NAME
from .models import TOPHistoricalData
from .schemas.user import User
@@ -15,6 +15,8 @@ from .schemas.user_profile import Data, UserProfile
UTC = timezone.utc
request = Request(config.tetris.proxy.top or config.tetris.proxy.main)
class Player:
def __init__(self, *, user_name: str, trust: bool = False) -> None:
@@ -35,8 +37,7 @@ class Player:
async def get_profile(self) -> UserProfile:
"""获取用户信息"""
if self._user_profile is None:
url = splice_url([BASE_URL, 'profile.php', f'?{urlencode({"user":self.user_name})}'])
raw_user_profile = await Request.request(url, is_json=False)
raw_user_profile = await request.request(BASE_URL / 'profile.php' % {'user': self.user_name}, is_json=False)
self._user_profile = self._parse_profile(raw_user_profile)
await anti_duplicate_add(
TOPHistoricalData(

4
nonebot_plugin_tetris_stats/games/top/constant.py
View File

@@ -1,8 +1,10 @@
from re import compile
from typing import Literal
from yarl import URL
GAME_TYPE: Literal['TOP'] = 'TOP'
BASE_URL = 'http://tetrisonline.pl/top/'
BASE_URL = URL('http://tetrisonline.pl/top/')
USER_NAME = compile(r'^[a-zA-Z0-9_]{1,16}$')

52
nonebot_plugin_tetris_stats/games/tos/api/player.py
View File

@@ -1,13 +1,14 @@
from datetime import datetime, timezone
from typing import overload
from urllib.parse import urlencode
from typing import cast, overload
from httpx import TimeoutException
from nonebot.compat import type_validate_json
from yarl import URL
from ....config.config import config
from ....db import anti_duplicate_add
from ....utils.exception import RequestError
from ....utils.request import Request, splice_url
from ....utils.request import Request
from ..constant import BASE_URL, USER_NAME
from .models import TOSHistoricalData
from .schemas.user import User
@@ -16,6 +17,8 @@ from .schemas.user_profile import UserProfile
UTC = timezone.utc
request = Request(config.tetris.proxy.tos or config.tetris.proxy.main)
class Player:
@overload
@@ -56,29 +59,14 @@ class Player:
async def get_info(self) -> UserInfoSuccess:
"""获取用户信息"""
if self._user_info is None:
if self.teaid is not None:
url = [
splice_url(
[
i,
'getTeaIdInfo',
f'?{urlencode({"teaId":self.teaid})}',
]
)
for i in BASE_URL
]
else:
url = [
splice_url(
[
i,
'getUsernameInfo',
f'?{urlencode({"username":self.user_name})}',
]
)
for i in BASE_URL
]
raw_user_info = await Request.failover_request(url, failover_code=[502], failover_exc=(TimeoutException,))
path = str(
URL('getTeaIdInfo') % {'teaId': self.teaid}
if self.teaid is not None
else URL('getUsernameInfo') % {'username': cast(str, self.user_name)}
)
raw_user_info = await request.failover_request(
[i / path for i in BASE_URL], failover_code=[502], failover_exc=(TimeoutException,)
)
user_info: UserInfo = type_validate_json(UserInfo, raw_user_info) # type: ignore[arg-type]
if not isinstance(user_info, UserInfoSuccess):
msg = f'用户信息请求错误:\n{user_info.error}'
@@ -98,17 +86,11 @@ class Player:
"""获取用户数据"""
if other_parameter is None:
other_parameter = {}
params = urlencode(dict(sorted(other_parameter.items())))
params = (URL('') % dict(sorted(other_parameter.items()))).human_repr()
if self._user_profile.get(params) is None:
raw_user_profile = await Request.failover_request(
raw_user_profile = await request.failover_request(
[
splice_url(
[
i,
'getProfile',
f'?{urlencode({"id":self.teaid or self.user_name,**other_parameter})}',
]
)
i / 'getProfile' % {'id': self.teaid or cast(str, self.user_name), **other_parameter}
for i in BASE_URL
],
failover_code=[502],

6
nonebot_plugin_tetris_stats/games/tos/constant.py
View File

@@ -1,11 +1,13 @@
from re import compile
from typing import Literal
from yarl import URL
GAME_TYPE: Literal['TOS'] = 'TOS'
BASE_URL = {
'https://teatube.cn:8888/',
'http://cafuuchino1.studio26f.org:19970',
URL('https://teatube.cn:8888/'),
URL('http://cafuuchino1.studio26f.org:19970'),
}
USER_NAME = compile(

33
nonebot_plugin_tetris_stats/utils/host.py
View File

@@ -1,16 +1,20 @@
from functools import cache
from hashlib import sha256
from ipaddress import IPv4Address, IPv6Address
from pathlib import Path as FilePath
from typing import TYPE_CHECKING, ClassVar, Literal
from fastapi import FastAPI, Path, status
from aiofiles import open
from fastapi import BackgroundTasks, FastAPI, Path, status
from fastapi.responses import FileResponse, HTMLResponse, Response
from fastapi.staticfiles import StaticFiles
from nonebot import get_app, get_driver
from nonebot.log import logger
from yarl import URL
from ..config.config import CACHE_PATH
from ..games.tetrio.api.cache import request
from .image import img_to_png
from .request import Request
from .templates import TEMPLATES_DIR
if TYPE_CHECKING:
@@ -22,6 +26,7 @@ driver = get_driver()
global_config = driver.config
BASE_URL = URL('https://tetr.io/user-content/')
if not isinstance(app, FastAPI):
msg = '本插件需要 FastAPI 驱动器才能运行'
@@ -63,20 +68,30 @@ def _(page_hash: str) -> HTMLResponse:
@app.get('/host/resource/tetrio/{resource_type}/{user_id}', status_code=status.HTTP_200_OK)
async def _(
resource_type: Literal['avatars', 'banners'], revision: int, user_id: str = Path(regex=r'^[a-f0-9]{24}$')
resource_type: Literal['avatars', 'banners'],
revision: int,
background_tasks: BackgroundTasks,
user_id: str = Path(regex=r'^[a-f0-9]{24}$'),
) -> Response:
if not (path := CACHE_PATH / 'tetrio' / resource_type / f'{user_id}_{revision}.png').exists():
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(
img_to_png(
await Request.request(
f'https://tetr.io/user-content/{resource_type}/{user_id}.jpg?rv={revision}', is_json=False
)
image = img_to_png(
await request.request(
BASE_URL / resource_type / f'{user_id}.jpg' % {'rv': revision},
is_json=False,
)
)
background_tasks.add_task(write_cache, path=path, data=image)
return Response(content=image, media_type='image/png')
return FileResponse(path)
async def write_cache(path: FilePath, data: bytes) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
async with open(path, 'wb') as file:
await file.write(data)
@cache
def get_self_netloc() -> str:
host: IPv4Address | IPv6Address | IPvAnyAddress = global_config.host
if isinstance(host, IPv4Address):

161
nonebot_plugin_tetris_stats/utils/request.py
View File

@@ -1,54 +1,79 @@
from collections.abc import Sequence
from http import HTTPStatus
from urllib.parse import urljoin, urlparse
from typing import Any
from aiofiles import open
from httpx import AsyncClient, HTTPError
from nonebot import get_driver, get_plugin_config
from msgspec import DecodeError, Struct, json
from nonebot import get_driver
from nonebot.log import logger
from playwright.async_api import Response
from ujson import JSONDecodeError, dumps, loads
from yarl import URL
from ..config.config import CACHE_PATH, Config
from ..config.config import CACHE_PATH, config
from .browser import BrowserManager
from .exception import RequestError
driver = get_driver()
config = get_plugin_config(Config)
@driver.on_startup
async def _():
await Request.init_cache()
await Request.read_cache()
class CloudflareCache(Struct):
headers: dict[str, Any] | None = None
cookies: dict[str, Any] | None = None
@driver.on_shutdown
async def _():
await Request.write_cache()
encoder = json.Encoder()
decoder = json.Decoder()
def splice_url(url_list: list[str]) -> str:
url = ''
if len(url_list):
url = url_list.pop(0)
for i in url_list:
url = urljoin(url, i)
return url
class AntiCloudflare:
cache_decoder = json.Decoder(type=CloudflareCache)
def __init__(self, domain_suffix: str) -> None:
self.domain_suffix = domain_suffix
self.cache_path = CACHE_PATH / f'{self.domain_suffix}_cloudflare_cache.json'
self._headers: dict | None = None
self._cookies: dict | None = None
self.read_cache()
class Request:
"""网络请求相关类"""
def read_cache(self) -> None:
"""读取缓存文件"""
try:
cache: CloudflareCache = self.cache_decoder.decode(self.cache_path.read_text(encoding='UTF-8'))
self._headers = cache.headers
self._cookies = cache.cookies
except (OSError, DecodeError):
self.cache_path.unlink()
self.write_cache()
_CACHE_FILE = CACHE_PATH / 'cloudflare_cache.json'
_headers: dict | None = None
_cookies: dict | None = None
def write_cache(self) -> None:
"""写入缓存文件"""
self.cache_path.write_bytes(json.encode(CloudflareCache(headers=self.headers, cookies=self.cookies)))
@classmethod
async def _anti_cloudflare(cls, url: str) -> bytes:
@property
def headers(self) -> dict | None:
return self._headers
@headers.setter
def headers(self, value: dict | None) -> None:
self._headers = value
self.write_cache()
@property
def cookies(self) -> dict | None:
return self._cookies
@cookies.setter
def cookies(self, value: dict | None) -> None:
self._cookies = value
self.write_cache()
async def __call__(self, url: str, proxy: str | None = None) -> bytes:
"""用firefox硬穿五秒盾"""
browser = await BrowserManager.get_browser()
async with await browser.new_context() as context, await context.new_page() as page:
async with (
await browser.new_context(proxy={'server': proxy} if proxy is not None else None) as context,
await context.new_page() as page,
):
response = await page.goto(url)
attempts = 0
while attempts < 60: # noqa: PLR2004
@@ -61,84 +86,68 @@ class Request:
logger.warning('疑似触发了 Cloudflare 的验证码')
break
try:
loads(text)
except JSONDecodeError:
decoder.decode(text)
except DecodeError:
await page.wait_for_timeout(1000)
else:
if not isinstance(response, Response):
msg = 'api请求失败'
raise RequestError(msg)
cls._headers = await response.request.all_headers()
self.headers = await response.request.all_headers()
try:
cls._cookies = {
self.cookies = {
name: value
for i in await context.cookies()
if (name := i.get('name')) is not None and (value := i.get('value')) is not None
}
except KeyError:
cls._cookies = None
self.cookies = None
return await response.body()
msg = '绕过五秒盾失败'
raise RequestError(msg)
@classmethod
async def init_cache(cls) -> None:
"""初始化缓存文件"""
if not cls._CACHE_FILE.exists():
async with open(file=cls._CACHE_FILE, mode='w', encoding='UTF-8') as file:
await file.write(dumps({'headers': cls._headers, 'cookies': cls._cookies}))
@classmethod
async def read_cache(cls) -> None:
"""读取缓存文件"""
try:
async with open(file=cls._CACHE_FILE, mode='r', encoding='UTF-8') as file:
json = loads(await file.read())
except FileNotFoundError:
await cls.init_cache()
except (PermissionError, JSONDecodeError):
cls._CACHE_FILE.unlink()
await cls.init_cache()
else:
cls._headers = json['headers']
cls._cookies = json['cookies']
class Request:
"""网络请求相关类"""
@classmethod
async def write_cache(cls) -> None:
"""写入缓存文件"""
try:
async with open(file=cls._CACHE_FILE, mode='r+', encoding='UTF-8') as file:
await file.write(dumps({'headers': cls._headers, 'cookies': cls._cookies}))
except FileNotFoundError:
await cls.init_cache()
except (PermissionError, JSONDecodeError):
cls._CACHE_FILE.unlink()
await cls.init_cache()
def __init__(self, proxy: str | None) -> None:
self.proxy = proxy
self.anti_cloudflares: dict[str, AntiCloudflare] = {}
@classmethod
async def request(cls, url: str, *, is_json: bool = True) -> bytes:
async def request(
self,
url: URL,
*,
is_json: bool = True,
enable_anti_cloudflare: bool = False,
) -> bytes:
"""请求api"""
if (anti_cloudflare := self.anti_cloudflares.get(url.host or '')) is not None:
cookies = anti_cloudflare.cookies
headers = anti_cloudflare.headers
else:
cookies = None
headers = None
try:
async with AsyncClient(cookies=cls._cookies, timeout=config.tetris.request_timeout) as session:
response = await session.get(url, headers=cls._headers)
async with AsyncClient(cookies=cookies, timeout=config.tetris.request_timeout, proxy=self.proxy) as session:
response = await session.get(str(url), headers=headers)
if response.status_code != HTTPStatus.OK:
msg = f'请求错误 code: {response.status_code} {HTTPStatus(response.status_code).phrase}\n{response.text}'
raise RequestError(msg, status_code=response.status_code)
if is_json:
loads(response.content)
decoder.decode(response.content)
return response.content
except HTTPError as e:
msg = f'请求错误 \n{e!r}'
raise RequestError(msg) from e
except JSONDecodeError:
if urlparse(url).netloc.lower().endswith('tetr.io'):
return await cls._anti_cloudflare(url)
except DecodeError: # 由于捕获的是 DecodeError 所以一定是 is_json = True
if enable_anti_cloudflare and url.host is not None:
return await self.anti_cloudflares.setdefault(url.host, AntiCloudflare(url.host))(str(url), self.proxy)
raise
@classmethod
async def failover_request(
cls,
urls: Sequence[str],
self,
urls: Sequence[URL],
*,
failover_code: Sequence[int],
failover_exc: tuple[type[BaseException], ...],
@@ -148,7 +157,7 @@ class Request:
for i in urls:
logger.debug(f'尝试请求 {i}')
try:
return await cls.request(i, is_json=is_json)
return await self.request(i, is_json=is_json)
except RequestError as e:
if e.status_code in failover_code: # 如果状态码在 failover_code 中, 则继续尝试下一个URL
error_list.append(e)

5
nonebot_plugin_tetris_stats/utils/screenshot.py
View File

@@ -1,13 +1,10 @@
from nonebot import get_plugin_config
from playwright.async_api import TimeoutError, ViewportSize
from ..config.config import Config
from ..config.config import config
from .browser import BrowserManager
from .retry import retry
from .time_it import time_it
config = get_plugin_config(Config)
@retry(exception_type=TimeoutError, reply='截图失败, 重试中')
@time_it

6
nonebot_plugin_tetris_stats/utils/templates.py
View File

@@ -13,7 +13,7 @@ from nonebot.permission import SUPERUSER
from nonebot_plugin_alconna import Alconna, Args, Option, on_alconna
from rich.progress import Progress
from ..config.config import CACHE_PATH, DATA_PATH
from ..config.config import CACHE_PATH, DATA_PATH, config
driver = get_driver()
@@ -24,7 +24,7 @@ alc = on_alconna(Alconna('更新模板', Option('--revision', Args['revision', s
async def download_templates(tag: str) -> Path:
logger.info(f'开始下载模板 {tag}')
async with AsyncClient() as client:
async with AsyncClient(proxy=config.tetris.proxy.github or config.tetris.proxy.main) as client:
if tag == 'latest':
logger.info('目标为 latest, 正在获取最新版本号')
tag = (
@@ -105,7 +105,7 @@ async def init_templates(tag: str) -> bool:
async def check_tag(tag: str) -> bool:
async with AsyncClient() as client:
async with AsyncClient(proxy=config.tetris.proxy.github or config.tetris.proxy.main) as client:
return (
await client.get(f'https://github.com/A-Minos/tetris-stats-templates/releases/tag/{tag}')
).status_code != HTTPStatus.NOT_FOUND