Shirino/main.py

249 lines
6 KiB
Python
Raw Normal View History

2023-05-30 13:47:06 +03:00
#!/usr/bin/env python3
2023-03-08 13:31:45 +03:00
import json
import hashlib
2023-05-30 13:33:41 +03:00
import logging
import string
2023-05-30 13:33:41 +03:00
from typing import Any, Dict, List, Optional
2023-03-08 13:31:45 +03:00
import aiohttp.client_exceptions
2023-05-30 13:33:41 +03:00
import requests
2023-03-08 13:31:45 +03:00
from pydantic.v1 import BaseSettings
2023-03-08 13:31:45 +03:00
from aiogram import Bot, Dispatcher, types # type: ignore
2023-03-08 13:31:45 +03:00
import asyncio
2023-03-08 13:31:45 +03:00
2023-05-30 13:33:41 +03:00
# Constants
DDG_URL = 'https://duckduckgo.com/js/spice/currency'
COINAPI_URL = 'https://rest.coinapi.io/v1/exchangerate'
# ---
2023-05-30 13:47:06 +03:00
2023-05-30 13:33:41 +03:00
# Config from .env
class Settings(BaseSettings):
debug: bool
2023-05-30 13:47:06 +03:00
timeout: int = 2
ndigits: int = 3
2023-06-01 16:39:46 +03:00
coinapi_keys: List[str]
2023-05-30 13:33:41 +03:00
telegram_token: str
2023-05-30 13:33:41 +03:00
class Config:
env_file = '.env'
env_file_encoding = 'utf-8'
2023-05-30 13:47:06 +03:00
settings = Settings() # type: ignore
2023-05-30 13:33:41 +03:00
# ---
2023-05-30 13:47:06 +03:00
2023-05-30 13:33:41 +03:00
# Logging
log = logging.getLogger('shirino')
handler = logging.StreamHandler()
fmt = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
2023-03-08 13:31:45 +03:00
2023-05-30 13:33:41 +03:00
handler.setFormatter(fmt)
log.addHandler(handler)
2023-03-08 13:31:45 +03:00
2023-05-30 13:33:41 +03:00
if settings.debug:
handler.setLevel(logging.DEBUG)
log.setLevel(logging.DEBUG)
# ---
2023-05-30 13:47:06 +03:00
2023-06-01 16:39:46 +03:00
coinapi_len = len(settings.coinapi_keys)
coinapi_active = [0] # API key index
dp = Dispatcher()
2023-03-08 13:31:45 +03:00
2023-05-30 13:33:41 +03:00
class CurrencyConverter:
2023-05-30 13:47:06 +03:00
2023-05-30 13:33:41 +03:00
def __init__(self) -> None:
2023-05-30 13:47:06 +03:00
self.amount: float = 1.0
self.conv_amount: float = 0.0
2023-05-30 13:33:41 +03:00
self.from_currency = ''
self.conv_currency = ''
2023-05-30 13:47:06 +03:00
2023-05-30 13:33:41 +03:00
def convert(self) -> None:
"""Currency conversion"""
if not self.ddgapi():
self.coinapi()
2023-05-30 13:47:06 +03:00
str_amount = f'{self.conv_amount}'
point = str_amount.find('.')
after_point = str_amount[point + 1:]
fnz = min( # index of first non-zero digit
(
after_point.index(i)
for i in string.digits[1:]
if i in after_point
),
default=-1,
)
if fnz == -1:
# it is an integer like 81.0
return
2023-06-01 16:07:20 +03:00
# how many digits should be after the point:
# ndigits (3 by default) after first non-zero
ndigits = fnz + settings.ndigits
2023-06-01 16:07:20 +03:00
self.conv_amount = round(self.conv_amount, ndigits)
2023-05-30 13:33:41 +03:00
def ddgapi(self) -> bool:
"""Get data from DuckDuckGo's currency API
2023-05-30 13:47:06 +03:00
2023-05-30 13:33:41 +03:00
Returns:
`False` if the currency does not exist,
`True` on successful conversion
"""
# API request
resp = requests.get(
2023-05-30 13:47:06 +03:00
(
f'{DDG_URL}/{self.amount}/{self.from_currency}'
f'/{self.conv_currency}'
),
timeout=settings.timeout,
2023-05-30 13:33:41 +03:00
)
log.debug(resp.text)
# Parsing JSON data
data: Dict[str, Any] = json.loads(
2023-05-30 13:47:06 +03:00
resp.text
.replace('ddg_spice_currency(', '')
2023-05-30 13:33:41 +03:00
.replace(');', '')
)
log.debug(data)
# If the currency does not exist
error = data.get('to')
if len(error) == 0:
2023-05-30 13:33:41 +03:00
return False
# Otherwise
conv: Dict[str, str] = data.get('to')[0]
conv_amount = conv.get('mid')
if not conv_amount:
raise RuntimeError('Ошибка при конвертации через DDG')
self.conv_amount = float(conv_amount)
2023-05-30 13:33:41 +03:00
log.debug(conv)
return True
2023-05-30 13:47:06 +03:00
2023-06-01 16:39:46 +03:00
def coinapi(self, depth: int = coinapi_len) -> None:
"""Get data from CoinAPI (for cryptocurrencies)
Args:
depth (int, optional): Counter protecting from infinite recursion
"""
if depth <= 0:
raise RecursionError('Рейтлимит на всех токенах')
2023-05-30 13:33:41 +03:00
resp = requests.get(
(
f'{COINAPI_URL}/{self.from_currency}'
f'/{self.conv_currency}'
),
headers={
2023-06-01 16:39:46 +03:00
'X-CoinAPI-Key': settings.coinapi_keys[coinapi_active[0]],
2023-05-30 13:33:41 +03:00
},
2023-05-30 13:47:06 +03:00
timeout=settings.timeout,
2023-05-30 13:33:41 +03:00
)
2023-06-01 16:39:46 +03:00
if resp.status_code == 429:
log.warning('CoinAPI ratelimited, rotating token')
rotate_token(settings.coinapi_keys, coinapi_active)
self.coinapi(depth - 1)
2023-05-30 13:33:41 +03:00
data: Dict[str, Any] = resp.json()
rate = data.get('rate')
if rate is None:
raise RuntimeError('Не удалось получить курс валюты от CoinAPI')
self.conv_amount = float(rate * self.amount)
2023-03-08 13:31:45 +03:00
2023-06-01 16:39:46 +03:00
def rotate_token(lst: List[str], active: List[int]) -> None:
"""Rotates API key to prevent ratelimits
Args:
lst (List[str]): Keys list
active (List[str]): Mutable object with current key index
"""
active[0] = (active[0] + 1) % len(lst)
@dp.inline_query()
async def currency(inline_query: types.InlineQuery) -> None:
2023-03-08 13:31:45 +03:00
2023-05-30 13:33:41 +03:00
query = inline_query.query
article: List[Optional[types.InlineQueryResultArticle]] = [None]
2023-03-08 13:31:45 +03:00
2023-05-30 13:33:41 +03:00
text = query.split()
len_ = len(text)
2023-03-08 13:31:45 +03:00
2023-05-30 13:33:41 +03:00
result_id = hashlib.md5(query.encode()).hexdigest()
conv = CurrencyConverter()
2023-03-08 13:31:45 +03:00
try:
2023-05-30 13:33:41 +03:00
if len_ == 3:
conv.amount = float(text[0])
conv.from_currency = text[1].upper()
conv.conv_currency = text[2].upper()
conv.convert()
elif len_ == 2:
conv.from_currency = text[0].upper()
conv.conv_currency = text[1].upper()
conv.convert()
2023-03-08 13:31:45 +03:00
else:
2023-05-30 13:33:41 +03:00
raise ValueError('Надо 2 или 3 аргумента')
2023-05-30 13:33:41 +03:00
result = (
f'{conv.amount} {conv.from_currency} = '
f'{conv.conv_amount} {conv.conv_currency}'
2023-05-30 13:33:41 +03:00
)
except aiohttp.client_exceptions.ClientError:
result = "Рейт-лимит от api telegram, попробуйте позже"
2023-05-30 13:33:41 +03:00
except Exception as ex:
2023-05-30 13:49:37 +03:00
result = f'{type(ex).__name__}: {ex}'
article[0] = types.InlineQueryResultArticle(
2023-03-08 13:31:45 +03:00
id=result_id,
title=result,
input_message_content=types.InputTextMessageContent(
2023-05-30 13:33:41 +03:00
message_text=result,
),
)
await inline_query.answer(
article, # type: ignore
2023-05-30 13:33:41 +03:00
cache_time=1,
is_personal=True,
)
2023-03-08 13:31:45 +03:00
async def main() -> None:
bot = Bot(settings.telegram_token)
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())