Compare commits

...

7 commits

6 changed files with 71 additions and 101 deletions

View file

@ -1,5 +0,0 @@
DEBUG=false # debug logging
TIMEOUT=2 # http requests timeout
NDIGITS=3 # digits after floating point or after zeroes
COINAPI_KEYS=["key"] # coinapi keys list
TELEGRAM_TOKEN= # telegram bot token

2
.gitignore vendored
View file

@ -5,4 +5,4 @@ __pycache__
.mypy_cache .mypy_cache
venv venv
.env config.yaml

View file

@ -9,15 +9,18 @@ https://t.me/Shirino_bot
Получите токен бота в телеграме и токен CoinAPI. Получите токен бота в телеграме и токен CoinAPI.
Вставьте в файл `.env` в формате: Вставьте в файл `.env` в формате:
``` ```yaml
COINAPI_KEYS=["Токен от CoinAPI"] coinapi_keys:
TELEGRAM_TOKEN=Токен Telegram-бота - key1
- key2
- etc.
telegram_token: Токен Telegram-бота
``` ```
В .env файл ещё можно такие переменные добавить: В .env файл ещё можно такие переменные добавить:
``` ```yaml
DEBUG=false или true, включает отладочные логи debug: false или true, включает отладочные логи
TIMEOUT=таймаут для библиотеки requests, в секундах (2 по дефолту) timeout: таймаут для библиотеки requests, в секундах (2 по дефолту)
``` ```
## Хочу сделать пулл-реквест ## Хочу сделать пулл-реквест
@ -28,8 +31,11 @@ TIMEOUT=таймаут для библиотеки requests, в секундах
## Почему энв для CoinAPI -- список? ## Почему энв для CoinAPI -- список?
Можно получить несколько ключей на разные почтовые ящики Можно получить несколько ключей на разные почтовые ящики
и все ключи вписать в список: и все ключи вписать в список:
``` ```yaml
COINAPI_KEYS=["первый", "второй", "и так далее"] coinapi_keys:
- key1
- key2
- etc.
``` ```
Если вдруг один из них будет заблокирован по рейтлимиту, Если вдруг один из них будет заблокирован по рейтлимиту,

7
config_sample.yaml Normal file
View file

@ -0,0 +1,7 @@
debug: false # debug logging
timeout: 2 # http requests timeout
ndigits: 3 # digits after floating point or after zeroes
coinapi_keys:
- key
- key2 # coinapi keys list
telegram_token: # telegram bot token

128
main.py
View file

@ -1,45 +1,24 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import asyncio import asyncio
import hashlib import hashlib
import json import json
import logging
import re
import string import string
from typing import List, Any, Dict, Optional
import aiohttp
import requests import requests
import re
import logging
import yaml
from aiogram import Dispatcher, types, Bot from aiogram import Dispatcher, types, Bot
import aiohttp.client_exceptions from aiogram.filters import CommandStart
from pydantic.v1 import BaseSettings
# Constants # Constants
DDG_URL = 'https://duckduckgo.com/js/spice/currency' DDG_URL = 'https://duckduckgo.com/js/spice/currency'
COINAPI_URL = 'https://rest.coinapi.io/v1/exchangerate' 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') log = logging.getLogger('shirino')
handler = logging.StreamHandler() handler = logging.StreamHandler()
fmt = logging.Formatter('%(asctime)s %(levelname)s %(message)s') fmt = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
@ -47,29 +26,24 @@ fmt = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
handler.setFormatter(fmt) handler.setFormatter(fmt)
log.addHandler(handler) log.addHandler(handler)
if settings.debug: if config['debug']:
handler.setLevel(logging.DEBUG) handler.setLevel(logging.DEBUG)
log.setLevel(logging.DEBUG) log.setLevel(logging.DEBUG)
# ---
coinapi_len = len(settings.coinapi_keys) coinapi_len = len(config['coinapi_keys'])
coinapi_active = [0] # API key index coinapi_active = [0]
dp = Dispatcher() dp = Dispatcher()
class CurrencyConverter: 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: def convert(self):
self.amount: float = 1.0
self.conv_amount: float = 0.0
self.from_currency = ''
self.conv_currency = ''
def convert(self) -> None:
"""Currency conversion"""
if not self.ddgapi(): if not self.ddgapi():
self.coinapi() self.coinapi()
@ -77,7 +51,7 @@ class CurrencyConverter:
point = str_amount.find(".") point = str_amount.find(".")
after_point = str_amount[point + 1:] after_point = str_amount[point + 1:]
fnz = min( # index of first non-zero digit fnz = min(
( (
after_point.index(i) after_point.index(i)
for i in string.digits[1:] for i in string.digits[1:]
@ -87,35 +61,26 @@ class CurrencyConverter:
) )
if fnz == -1: if fnz == -1:
# it is an integer like 81.0
return return
# how many digits should be after the point: ndigits = fnz + config['ndigits']
# ndigits (3 by default) after first non-zero
ndigits = fnz + settings.ndigits
self.conv_amount = round(self.conv_amount, ndigits) self.conv_amount = round(self.conv_amount, ndigits)
def ddgapi(self) -> bool: def ddgapi(self):
"""Получение данных фиатной валюты через DuckDuckGo
e.g: https://duckduckgo.com/js/spice/currency/1/USD/RUB
Returns:
`False` если валюты нет в API
`True` если конвертация прошла успешно
"""
# Запрос к API
res = requests.get(f'{DDG_URL}/{self.amount}/{self.from_currency}/{self.conv_currency}') 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) log.debug(data)
if len(data.get('to')) == 0: if len(data.get('to')) == 0:
return False return False
conv: Dict[str, str] = data.get('to')[0] conv = data.get('to')[0]
conv_amount = conv.get("mid") conv_amount = conv.get("mid")
if conv_amount is None: if conv_amount is None:
@ -128,13 +93,7 @@ class CurrencyConverter:
return True return True
def coinapi(self, depth: int = coinapi_len) -> None: def coinapi(self, depth: int = config['coinapi_keys']):
"""Получение данных с CoinAPI для получения курса криптовалюты
Args:
depth (int, optional): Счетчик, защищающий от рекурсии
"""
if depth <= 0: if depth <= 0:
raise RecursionError('Рейтлимит на всех токенах') raise RecursionError('Рейтлимит на всех токенах')
@ -144,36 +103,29 @@ class CurrencyConverter:
f'/{self.conv_currency}' f'/{self.conv_currency}'
), ),
headers={ 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: if resp.status_code == 429:
log.warning('CoinAPI ratelimited, rotating token') log.warning('CoinAPI ratelimited, rotating token')
rotate_token(settings.coinapi_keys, coinapi_active) rotate_token(config['coinapi_keys'], coinapi_active)
self.coinapi(depth - 1) self.coinapi(depth - 1)
data: Dict[str, Any] = resp.json() data = resp.json()
rate = data.get('rate') rate = data.get('rate')
if rate is None: if rate is None:
raise RuntimeError('Не удалось получить курс валюты от CoinAPI') raise RuntimeError('Не удалось получить курс валюты от CoinAPI')
self.conv_amount = float(rate * self.amount) self.conv_amount = float(rate * self.amount)
def rotate_token(lst: List[str], active: List[int]) -> None: def rotate_token(lst, active):
"""Смена API-ключей CoinAPI при ratelimits
Args:
lst (List[str]): Список ключей
active (List[str]): Изменяемый объект с текущим ключевым индексом
"""
active[0] = (active[0] + 1) % len(lst) 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: async def inline_reply(result_id: str, title: str, description: str or None, inline_query: types.InlineQuery):
article: List[Optional[types.InlineQueryResultArticle]] = [None] article = [None]
article[0] = types.InlineQueryResultArticle( article[0] = types.InlineQueryResultArticle(
id=result_id, id=result_id,
title=title, title=title,
@ -191,7 +143,7 @@ async def inline_reply(result_id: str, title: str, description: str or None, inl
@dp.inline_query() @dp.inline_query()
async def currency(inline_query: types.InlineQuery) -> None: async def currency(inline_query: types.InlineQuery):
query = inline_query.query query = inline_query.query
args = query.split() args = query.split()
@ -238,8 +190,18 @@ async def currency(inline_query: types.InlineQuery) -> None:
await inline_reply(result_id, result, None, inline_query) await inline_reply(result_id, result, None, inline_query)
@dp.message(CommandStart())
async def start(message: types.Message):
await message.answer("Привет! Бот может показывать курс обмена криптовалюты и фиатной валюты. "
"Бот используется с помощью инлайн-команд: "
"`@shirino_bot 12 usd rub` или `@shirino_bot usd rub`"
"\n\nИсходный код опубликован на [Github](https://github.com/redume/shirino)",
parse_mode="markdown")
async def main() -> None: async def main() -> None:
bot = Bot(settings.telegram_token) log.debug(config)
bot = Bot(config['telegram_token'])
await dp.start_polling(bot) await dp.start_polling(bot)

View file

@ -1,4 +1,4 @@
requests==2.31.0 aiohttp~=3.9.5
aiogram==3.1.1 requests~=2.32.3
aiohttp==3.8.5 PyYAML~=6.0.1
pydantic[dotenv]==2.5.2 aiogram~=3.7.0