File size: 3,674 Bytes
d8b23ad e575626 d8b23ad e575626 d8b23ad e575626 d8b23ad |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
from pathlib import Path
from typing import List, Set, Union
import httpx
import nonebot
from aiocache import cached
from nonebot import logger
from pydantic import BaseModel, Extra
try:
import ujson as json
except ModuleNotFoundError:
import json
class PluginConfig(BaseModel, extra=Extra.ignore):
'''
Path of tarot images resource
'''
tarot_path: Path = Path(__file__).parent / "resource"
chain_reply: bool = True
nickname: Set[str] = {"Bot"}
'''
DO NOT CHANGE THIS VALUE IN ANY .ENV FILES
'''
tarot_official_themes: List[str] = ["BilibiliTarot", "TouhouTarot"]
driver = nonebot.get_driver()
tarot_config: PluginConfig = PluginConfig.parse_obj(
driver.config.dict(exclude_unset=True))
class DownloadError(Exception):
pass
class ResourceError(Exception):
def __init__(self, msg: str):
self.msg = msg
def __str__(self):
return self.msg
__repr__ = __str__
class EventsNotSupport(Exception):
pass
async def download_url(url: str) -> Union[httpx.Response, None]:
async with httpx.AsyncClient() as client:
for i in range(3):
try:
response = await client.get(url)
if response.status_code != 200:
continue
return response
except Exception:
logger.warning(
f"Error occurred when downloading {url}, {i+1}/3")
logger.warning("Abort downloading")
return None
@driver.on_startup
async def tarot_version_check() -> None:
'''
Get the latest version of tarot.json from repo
'''
if not tarot_config.tarot_path.exists():
tarot_config.tarot_path.mkdir(parents=True, exist_ok=True)
tarot_json_path = Path(__file__).parent / "tarot.json"
cur_version = 0
if tarot_json_path.exists():
with tarot_json_path.open("r", encoding="utf-8") as f:
data = json.load(f)
cur_version = data.get("version", 0)
# url: str = "https://raw.fastgit.org/MinatoAquaCrews/nonebot_plugin_tarot/master/nonebot_plugin_tarot/tarot.json"
# response = await download_url(url)
# if response is None:
# if not tarot_json_path.exists():
# logger.warning("Tarot text resource missing! Please check!")
# raise ResourceError("Missing necessary resource: tarot.json!")
# else:
# docs = response.json()
# try:
# version = docs.get("version")
# except KeyError:
# logger.warning(
# "Tarot text resource downloaded incompletely! Please check!")
# raise DownloadError
# # Update when there is a newer version
# if version > cur_version:
# with tarot_json_path.open("w", encoding="utf-8") as f:
# json.dump(docs, f, ensure_ascii=False, indent=4)
# logger.info(
# f"Updated tarot.json, version: {cur_version} -> {version}")
@cached(ttl=180)
async def get_tarot(_theme: str, _type: str, _name: str) -> Union[bytes, None]:
'''
Downloads tarot image and stores cache temporarily
if downloading failed, return None
'''
logger.info(
f"Downloading tarot image {_theme}/{_type}/{_name} from repo")
url: str = "https://raw.fastgit.org/MinatoAquaCrews/nonebot_plugin_tarot/master/nonebot_plugin_tarot/resource/" + \
f"{_theme}/{_type}/{_name}"
data = await download_url(url)
if data is None:
logger.warning(
f"Downloading tarot image {_theme}/{_type}/{_name} failed!")
return None
return data.content
|