bot.py - master - eup/tg_curs

bot.py
288 строк9 KB
eup
Create: .env.example, .env.txt, bot.py, README.md, test_bot.py
07 июн 2026, 12:00
07 июн 2026, 12:000c87280
import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP


CBR_DAILY_URL = "https://www.cbr.ru/scripts/XML_daily.asp"
TELEGRAM_API_URL = "https://api.telegram.org/bot{token}/{method}"
RATES_CACHE_SECONDS = 60 * 60


ALIASES = {
    "RUR": "RUB",
    "РУБ": "RUB",
    "РУБЛЬ": "RUB",
    "РУБЛИ": "RUB",
    "ДОЛЛАР": "USD",
    "ДОЛЛАРЫ": "USD",
    "БАКС": "USD",
    "БАКСЫ": "USD",
    "ЕВРО": "EUR",
    "ЮАНЬ": "CNY",
    "ЮАНИ": "CNY",
    "ФУНТ": "GBP",
    "ТЕНГЕ": "KZT",
}


POPULAR_CODES = ("USD", "EUR", "CNY", "GBP", "JPY", "KZT", "TRY")


@dataclass(frozen=True)
class CurrencyRate:
    code: str
    name: str
    nominal: int
    value_rub: Decimal

    @property
    def rub_per_unit(self) -> Decimal:
        return self.value_rub / Decimal(self.nominal)


class RatesProvider:
    def __init__(self, url: str = CBR_DAILY_URL, cache_seconds: int = RATES_CACHE_SECONDS):
        self.url = url
        self.cache_seconds = cache_seconds
        self._loaded_at = 0.0
        self._rates = None
        self._date = ""

    def get_rates(self) -> tuple[dict[str, CurrencyRate], str]:
        now = time.time()
        if self._rates is not None and now - self._loaded_at < self.cache_seconds:
            return self._rates, self._date

        with urllib.request.urlopen(self.url, timeout=10) as response:
            xml_text = response.read()

        rates, date = parse_cbr_xml(xml_text)
        self._rates = rates
        self._date = date
        self._loaded_at = now
        return rates, date


def parse_cbr_xml(xml_bytes: bytes) -> tuple[dict[str, CurrencyRate], str]:
    root = ET.fromstring(xml_bytes)
    rates = {
        "RUB": CurrencyRate(
            code="RUB",
            name="Российский рубль",
            nominal=1,
            value_rub=Decimal("1"),
        )
    }

    for item in root.findall("Valute"):
        code = required_text(item, "CharCode").upper()
        name = required_text(item, "Name")
        nominal = int(required_text(item, "Nominal"))
        value = Decimal(required_text(item, "Value").replace(",", "."))
        rates[code] = CurrencyRate(code=code, name=name, nominal=nominal, value_rub=value)

    return rates, root.attrib.get("Date", "")


def required_text(node: ET.Element, tag: str) -> str:
    child = node.find(tag)
    if child is None or child.text is None:
        raise ValueError(f"CBR response does not contain {tag}")
    return child.text.strip()


def normalize_code(value: str) -> str:
    cleaned = value.strip().upper().replace(".", "")
    return ALIASES.get(cleaned, cleaned)


def format_money(value: Decimal, places: str = "0.01") -> str:
    quantized = value.quantize(Decimal(places), rounding=ROUND_HALF_UP)
    text = f"{quantized:f}"
    if "." in text:
        text = text.rstrip("0").rstrip(".")
    return text.replace(".", ",")


def convert(amount: Decimal, from_code: str, to_code: str, rates: dict[str, CurrencyRate]) -> Decimal:
    source = rates[from_code].rub_per_unit
    target = rates[to_code].rub_per_unit
    return amount * source / target


def parse_amount(value: str) -> Decimal:
    normalized = value.replace(",", ".").replace("_", "")
    return Decimal(normalized)


def build_help() -> str:
    return (
        "Привет! Я показываю курс валют и конвертирую суммы по курсу ЦБ РФ.\n\n"
        "Команды:\n"
        "/rates - популярные курсы\n"
        "/rate USD - курс одной валюты\n"
        "/convert 100 USD RUB - конвертировать сумму\n\n"
        "Можно писать и без команды: 100 usd eur"
    )


def handle_message(text: str, provider: RatesProvider) -> str:
    text = text.strip()
    if not text:
        return build_help()

    command, *args = text.split()
    command_lower = command.lower()

    if command_lower in ("/start", "/help"):
        return build_help()

    if command_lower == "/rates":
        rates, date = provider.get_rates()
        return render_popular_rates(rates, date)

    if command_lower == "/rate":
        return handle_rate(args, provider)

    if command_lower == "/convert":
        return handle_convert(args, provider)

    if len([command, *args]) == 3:
        return handle_convert([command, *args], provider)

    return "Не понял запрос. Попробуйте: /convert 100 USD RUB или /rate EUR"


def handle_rate(args: list[str], provider: RatesProvider) -> str:
    if len(args) != 1:
        return "Формат: /rate USD"

    code = normalize_code(args[0])
    rates, date = provider.get_rates()
    if code not in rates:
        return f"Валюта {code} не найдена. Используйте код вроде USD, EUR, CNY."

    rate = rates[code]
    if code == "RUB":
        return "1 RUB = 1 RUB"

    return (
        f"{rate.name} ({rate.code}), курс ЦБ РФ на {date}:\n"
        f"1 {rate.code} = {format_money(rate.rub_per_unit)} RUB"
    )


def handle_convert(args: list[str], provider: RatesProvider) -> str:
    if len(args) != 3:
        return "Формат: /convert 100 USD RUB"

    try:
        amount = parse_amount(args[0])
    except InvalidOperation:
        return "Сумма должна быть числом. Например: /convert 100 USD RUB"

    from_code = normalize_code(args[1])
    to_code = normalize_code(args[2])
    rates, date = provider.get_rates()

    missing = [code for code in (from_code, to_code) if code not in rates]
    if missing:
        return f"Валюта {', '.join(missing)} не найдена. Используйте код вроде USD, EUR, RUB."

    result = convert(amount, from_code, to_code, rates)
    return (
        f"{format_money(amount)} {from_code} = {format_money(result)} {to_code}\n"
        f"Курс ЦБ РФ на {date}"
    )


def render_popular_rates(rates: dict[str, CurrencyRate], date: str) -> str:
    lines = [f"Популярные курсы ЦБ РФ на {date}:"]
    for code in POPULAR_CODES:
        if code in rates:
            rate = rates[code]
            lines.append(f"{code}: {format_money(rate.rub_per_unit)} RUB")
    return "\n".join(lines)


def load_env_file(path: str = ".env") -> None:
    if not os.path.exists(path):
        return

    with open(path, "r", encoding="utf-8") as file:
        for raw_line in file:
            line = raw_line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, value = line.split("=", 1)
            os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))


def telegram_request(token: str, method: str, payload: dict) -> dict:
    data = urllib.parse.urlencode(payload).encode("utf-8")
    request = urllib.request.Request(
        TELEGRAM_API_URL.format(token=token, method=method),
        data=data,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )

    with urllib.request.urlopen(request, timeout=30) as response:
        return json.loads(response.read().decode("utf-8"))


def send_message(token: str, chat_id: int, text: str) -> None:
    telegram_request(
        token,
        "sendMessage",
        {
            "chat_id": chat_id,
            "text": text,
            "disable_web_page_preview": "true",
        },
    )


def run_bot() -> None:
    load_env_file()
    token = os.getenv("TELEGRAM_BOT_TOKEN")
    if not token:
        raise RuntimeError("Set TELEGRAM_BOT_TOKEN in .env or environment variables")

    provider = RatesProvider()
    offset = None
    print("Currency bot is running. Press Ctrl+C to stop.")

    while True:
        payload = {"timeout": 25}
        if offset is not None:
            payload["offset"] = offset

        try:
            updates = telegram_request(token, "getUpdates", payload)
        except (urllib.error.URLError, TimeoutError) as error:
            print(f"Telegram request failed: {error}. Retrying...")
            time.sleep(3)
            continue

        for update in updates.get("result", []):
            offset = update["update_id"] + 1
            message = update.get("message") or update.get("edited_message")
            if not message or "text" not in message:
                continue

            chat_id = message["chat"]["id"]
            try:
                answer = handle_message(message["text"], provider)
            except Exception as error:
                answer = f"Не удалось обработать запрос: {error}"
            send_message(token, chat_id, answer)


if __name__ == "__main__":
    run_bot()