mirror of
https://github.com/Redume/Shirino.git
synced 2024-12-24 01:03:46 +03:00
Бот теперь без типизации данных, конфиг теперь yaml, удаление лишний инфы в json через регулярку
This commit is contained in:
parent
1bd1bf9cb6
commit
2d9ad91f49
1 changed files with 36 additions and 83 deletions
119
main.py
119
main.py
|
@ -1,45 +1,23 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
|
||||
import string
|
||||
from typing import List, Any, Dict, Optional
|
||||
|
||||
import aiohttp
|
||||
import requests
|
||||
import re
|
||||
import logging
|
||||
import yaml
|
||||
|
||||
from aiogram import Dispatcher, types, Bot
|
||||
import aiohttp.client_exceptions
|
||||
from pydantic.v1 import BaseSettings
|
||||
|
||||
# Constants
|
||||
DDG_URL = 'https://duckduckgo.com/js/spice/currency'
|
||||
COINAPI_URL = 'https://rest.coinapi.io/v1/exchangerate'
|
||||
|
||||
config = yaml.safe_load(open("config.yaml"))
|
||||
|
||||
# ---
|
||||
|
||||
|
||||
# Config from .env
|
||||
class Settings(BaseSettings):
|
||||
debug: bool
|
||||
timeout: int = 2
|
||||
ndigits: int = 3
|
||||
coinapi_keys: List[str]
|
||||
telegram_token: str
|
||||
|
||||
class Config:
|
||||
env_file = '.env'
|
||||
env_file_encoding = 'utf-8'
|
||||
|
||||
|
||||
settings = Settings() # type: ignore
|
||||
# ---
|
||||
|
||||
|
||||
# Logging
|
||||
log = logging.getLogger('shirino')
|
||||
handler = logging.StreamHandler()
|
||||
fmt = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
|
||||
|
@ -47,29 +25,26 @@ fmt = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
|
|||
handler.setFormatter(fmt)
|
||||
log.addHandler(handler)
|
||||
|
||||
if settings.debug:
|
||||
print(config)
|
||||
|
||||
if config['debug']:
|
||||
handler.setLevel(logging.DEBUG)
|
||||
log.setLevel(logging.DEBUG)
|
||||
# ---
|
||||
|
||||
|
||||
coinapi_len = len(settings.coinapi_keys)
|
||||
coinapi_active = [0] # API key index
|
||||
coinapi_len = len(config['coinapi_keys'])
|
||||
coinapi_active = [0]
|
||||
dp = Dispatcher()
|
||||
|
||||
|
||||
class CurrencyConverter:
|
||||
def __init__(self):
|
||||
self.amount = 1.0
|
||||
self.conv_amount = 0.0
|
||||
self.from_currency = 'RUB'
|
||||
self.conv_currency = 'USD'
|
||||
|
||||
def __init__(self) -> None:
|
||||
|
||||
self.amount: float = 1.0
|
||||
self.conv_amount: float = 0.0
|
||||
self.from_currency = ''
|
||||
self.conv_currency = ''
|
||||
|
||||
def convert(self) -> None:
|
||||
"""Currency conversion"""
|
||||
|
||||
def convert(self):
|
||||
if not self.ddgapi():
|
||||
self.coinapi()
|
||||
|
||||
|
@ -77,7 +52,7 @@ class CurrencyConverter:
|
|||
point = str_amount.find(".")
|
||||
after_point = str_amount[point + 1:]
|
||||
|
||||
fnz = min( # index of first non-zero digit
|
||||
fnz = min(
|
||||
(
|
||||
after_point.index(i)
|
||||
for i in string.digits[1:]
|
||||
|
@ -87,35 +62,26 @@ class CurrencyConverter:
|
|||
)
|
||||
|
||||
if fnz == -1:
|
||||
# it is an integer like 81.0
|
||||
return
|
||||
|
||||
# how many digits should be after the point:
|
||||
# ndigits (3 by default) after first non-zero
|
||||
ndigits = fnz + settings.ndigits
|
||||
ndigits = fnz + config['ndigits']
|
||||
|
||||
self.conv_amount = round(self.conv_amount, ndigits)
|
||||
|
||||
def ddgapi(self) -> bool:
|
||||
"""Получение данных фиатной валюты через DuckDuckGo
|
||||
|
||||
e.g: https://duckduckgo.com/js/spice/currency/1/USD/RUB
|
||||
|
||||
Returns:
|
||||
`False` если валюты нет в API
|
||||
`True` если конвертация прошла успешно
|
||||
"""
|
||||
|
||||
# Запрос к API
|
||||
def ddgapi(self):
|
||||
res = requests.get(f'{DDG_URL}/{self.amount}/{self.from_currency}/{self.conv_currency}')
|
||||
data: Dict[str, Any] = json.loads(re.findall(r'(.+)\);', res.text)[0])
|
||||
data = json.loads(re.findall(r'\(\s*(.*)\s*\);$', res.text)[0])
|
||||
|
||||
del data['terms']
|
||||
del data['privacy']
|
||||
del data['timestamp']
|
||||
|
||||
log.debug(data)
|
||||
|
||||
if len(data.get('to')) == 0:
|
||||
return False
|
||||
|
||||
conv: Dict[str, str] = data.get('to')[0]
|
||||
conv = data.get('to')[0]
|
||||
conv_amount = conv.get("mid")
|
||||
|
||||
if conv_amount is None:
|
||||
|
@ -128,13 +94,7 @@ class CurrencyConverter:
|
|||
|
||||
return True
|
||||
|
||||
def coinapi(self, depth: int = coinapi_len) -> None:
|
||||
"""Получение данных с CoinAPI для получения курса криптовалюты
|
||||
|
||||
Args:
|
||||
depth (int, optional): Счетчик, защищающий от рекурсии
|
||||
"""
|
||||
|
||||
def coinapi(self, depth: int = config['coinapi_keys']):
|
||||
if depth <= 0:
|
||||
raise RecursionError('Рейтлимит на всех токенах')
|
||||
|
||||
|
@ -144,36 +104,29 @@ class CurrencyConverter:
|
|||
f'/{self.conv_currency}'
|
||||
),
|
||||
headers={
|
||||
'X-CoinAPI-Key': settings.coinapi_keys[coinapi_active[0]],
|
||||
'X-CoinAPI-Key': config['coinapi_keys'][coinapi_active[0]],
|
||||
},
|
||||
timeout=settings.timeout,
|
||||
timeout=config['timeout'],
|
||||
)
|
||||
|
||||
if resp.status_code == 429:
|
||||
log.warning('CoinAPI ratelimited, rotating token')
|
||||
rotate_token(settings.coinapi_keys, coinapi_active)
|
||||
rotate_token(config['coinapi_keys'], coinapi_active)
|
||||
self.coinapi(depth - 1)
|
||||
|
||||
data: Dict[str, Any] = resp.json()
|
||||
data = resp.json()
|
||||
rate = data.get('rate')
|
||||
if rate is None:
|
||||
raise RuntimeError('Не удалось получить курс валюты от CoinAPI')
|
||||
self.conv_amount = float(rate * self.amount)
|
||||
|
||||
|
||||
def rotate_token(lst: List[str], active: List[int]) -> None:
|
||||
"""Смена API-ключей CoinAPI при ratelimits
|
||||
|
||||
Args:
|
||||
lst (List[str]): Список ключей
|
||||
active (List[str]): Изменяемый объект с текущим ключевым индексом
|
||||
"""
|
||||
|
||||
def rotate_token(lst, active):
|
||||
active[0] = (active[0] + 1) % len(lst)
|
||||
|
||||
|
||||
async def inline_reply(result_id: str, title: str, description: str or None, inline_query: types.InlineQuery) -> None:
|
||||
article: List[Optional[types.InlineQueryResultArticle]] = [None]
|
||||
async def inline_reply(result_id: str, title: str, description: str or None, inline_query: types.InlineQuery):
|
||||
article = [None]
|
||||
article[0] = types.InlineQueryResultArticle(
|
||||
id=result_id,
|
||||
title=title,
|
||||
|
@ -191,7 +144,7 @@ async def inline_reply(result_id: str, title: str, description: str or None, inl
|
|||
|
||||
|
||||
@dp.inline_query()
|
||||
async def currency(inline_query: types.InlineQuery) -> None:
|
||||
async def currency(inline_query: types.InlineQuery):
|
||||
query = inline_query.query
|
||||
args = query.split()
|
||||
|
||||
|
@ -239,7 +192,7 @@ async def currency(inline_query: types.InlineQuery) -> None:
|
|||
|
||||
|
||||
async def main() -> None:
|
||||
bot = Bot(settings.telegram_token)
|
||||
bot = Bot(config['telegram_token'])
|
||||
await dp.start_polling(bot)
|
||||
|
||||
|
||||
|
|
Loading…
Add table
Reference in a new issue