Spaces:
Runtime error
Runtime error
File size: 1,660 Bytes
5dd5b6a |
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 |
import asyncio
import concurrent.futures
import json
from typing import Iterator
import concurrent
import pytz
from datetime import datetime
WORDS = "trigger.json"
tz = pytz.timezone("Asia/Shanghai")
def get_words() -> dict:
with open(WORDS, "r", encoding="utf-8") as f:
words = json.load(f)
return words
def mapping_cog(cog_name: str) -> str:
return "".join(["_" + i.lower() if i.isupper() else i for i in cog_name]).lstrip(
"_"
)
def now(tz=tz, fmt="%Y-%m-%d %H:%M:%S") -> str:
return datetime.now(tz).strftime(fmt)
def get_time(dt: datetime, tz=tz, fmt="%Y-%m-%d %H:%M:%S") -> str:
return dt.astimezone(tz).strftime(fmt)
async def async_iter(response: Iterator):
loop = asyncio.get_event_loop()
def safe_next(iterator: Iterator):
try:
return next(iterator), False
except StopIteration:
return None, True
try:
iterator = iter(response)
while True:
item, done = await loop.run_in_executor(None, safe_next, iterator)
if done:
break
yield item
except Exception as e:
print(e)
async def async_do(func, *args, **kwargs):
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, func, *args, **kwargs)
async def async_do_thread(func, *args, **kwargs):
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(func, *args, **kwargs)
try:
return await asyncio.wrap_future(future)
except Exception as e:
print(e)
|