commit 11b93825d9f77f69122ca0f873bd83a97d2b83a2 Author: irondru Date: Wed Jul 1 16:47:54 2026 +0300 init: telegram currency bot with systemd timer diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..d7bae84 --- /dev/null +++ b/.env.example @@ -0,0 +1,4 @@ +TELEGRAM_BOT_TOKEN=123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11 +TELEGRAM_CHAT_ID=@your_channel_username +TELEGRAM_MESSAGE_ID=123 +LOG_FILE=bot_update_pinned_post.log diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ef303be --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +.env +.venv/ +__pycache__/ +*.pyc +*.log +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..0079aab --- /dev/null +++ b/README.md @@ -0,0 +1,84 @@ +# Telegram Пин-бот курса + +Этот проект обновляет один закреплённый пост в Telegram-канале раз в час. В посте показываются: +- курс USD/RUB от ЦБ РФ +- курс BTC/USD от CoinGecko +- индекс страха/жадности от Alternative.me + +## Установка + +1. Установите Python 3.10+ на VPS. +2. Перейдите в папку проекта: + ```bash + cd ~/Agents/tg_update_currency + ``` +3. Создайте виртуальное окружение: + ```bash + python3 -m venv .venv + source .venv/bin/activate + ``` +4. Установите зависимости: + ```bash + pip install -r requirements.txt + ``` + +## Настройка .env + +1. Получите токен у @BotFather: + - Запустите @BotFather в Telegram. + - Выберите /newbot и следуйте инструкциям. + - Скопируйте токен вида `123456:ABC-DEF...`. +2. Узнайте `CHAT_ID` канала: + - Добавьте бота в канал как администратора. + - Используйте `@getidsbot` или `@username_to_id_bot` в Telegram. + - Если канал имеет публичный username, можно указать `@your_channel_username`. +3. Узнайте `MESSAGE_ID` закреплённого сообщения: + - Вручную создайте и закрепите пост в канале. + - Если бот администратор, вызовите `https://api.telegram.org/bot/getChat?chat_id=@your_channel_username`. + - В ответе найдите поле `pinned_message.message_id`. + - Либо получите ID предыдущим ботом, когда публиковали сообщение. +4. Создайте файл `.env` рядом с `.env.example` и заполните значения. + +Пример `.env`: +```ini +TELEGRAM_BOT_TOKEN=123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11 +TELEGRAM_CHAT_ID=@your_channel_username +TELEGRAM_MESSAGE_ID=123 +``` + +## Запуск вручную + +```bash +source .venv/bin/activate +python bot_update_pinned_post.py +``` + +Если всё настроено верно, бот отредактирует закреплённый пост. + +## Настройка cron + +1. Откройте редактор cron: + ```bash + crontab -e + ``` +2. Добавьте задачу: + ```cron + 0 * * * * cd /Users/irondru/Yandex.Disk.localized/Agents/tg_update_currency && /Users/irondru/Yandex.Disk.localized/Agents/tg_update_currency/.venv/bin/python bot_update_pinned_post.py >> /Users/irondru/Yandex.Disk.localized/Agents/tg_update_currency/cron.log 2>&1 + ``` +3. Сохраните и выйдите. + +Проверить cron можно командой: +```bash +crontab -l +``` + +## Проверка + +- Запустите скрипт вручную и проверьте, что закреплённый пост обновился. +- Если один из API недоступен, скрипт ничего не изменит и запишет ошибку в лог. +- Убедитесь, что бот добавлен в канал как администратор с правом редактировать сообщения. + +## Логирование + +По умолчанию лог записывается в `bot_update_pinned_post.log`. +В лог не пишутся токены, только время, значения и ошибки. diff --git a/bot_update_pinned_post.py b/bot_update_pinned_post.py new file mode 100644 index 0000000..507bf37 --- /dev/null +++ b/bot_update_pinned_post.py @@ -0,0 +1,208 @@ +import json +import logging +import os +import socket +import ssl +from datetime import datetime +from urllib.parse import urlparse + +import httpx +from dotenv import load_dotenv + +load_dotenv() + +TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") +TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID") +TELEGRAM_MESSAGE_ID = os.getenv("TELEGRAM_MESSAGE_ID") +LOG_FILE = os.getenv("LOG_FILE", "bot_update_pinned_post.log") + +logging.basicConfig( + filename=LOG_FILE, + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +for logger_name in ("httpx", "httpcore", "anyio"): + logger = logging.getLogger(logger_name) + logger.setLevel(logging.WARNING) + logger.propagate = False + + +class FetchError(Exception): + pass + + +def fetch_usd_rub() -> float: + url = "https://www.cbr-xml-daily.ru/daily_json.js" + try: + with httpx.Client(timeout=10.0) as client: + response = client.get(url) + response.raise_for_status() + data = response.json() + usd_value = data["Valute"]["USD"]["Value"] + return round(usd_value, 2) + except Exception as exc: + raise FetchError(f"USD/RUB fetch failed: {exc}") from exc + + +def fetch_btc_usd() -> float: + url = "https://api.coingecko.com/api/v3/simple/price" + params = {"ids": "bitcoin", "vs_currencies": "usd"} + try: + with httpx.Client(timeout=10.0) as client: + response = client.get(url, params=params) + response.raise_for_status() + data = response.json() + btc_value = data["bitcoin"]["usd"] + return round(float(btc_value), 2) + except Exception as exc: + raise FetchError(f"BTC/USD fetch failed: {exc}") from exc + + +def translate_fng_label(label: str) -> str: + mapping = { + "extreme fear": "сильный страх", + "fear": "страх", + "neutral": "нейтрально", + "greed": "жадность", + "extreme greed": "сильная жадность", + } + return mapping.get(label.lower(), label) + + +def fetch_fear_and_greed() -> tuple[int, str]: + url = "https://api.alternative.me/fng/" + params = {"limit": 1, "format": "json"} + try: + with httpx.Client(timeout=10.0) as client: + response = client.get(url, params=params) + response.raise_for_status() + data = response.json() + item = data["data"][0] + value = int(item["value"]) + label = translate_fng_label(item["value_classification"]) + return value, label + except Exception as exc: + raise FetchError(f"Fear and greed fetch failed: {exc}") from exc + + +def build_message(usd: float, btc: float, fng_value: int, fng_label: str) -> str: + return ( + f"💲 " + f"₽{usd:.2f} 💲 " + f"${btc:.2f} 😱 {fng_value} ({fng_label})" + ) + + +def post_telegram_ipv4(url: str, payload: dict, timeout: float = 10.0) -> dict: + parsed = urlparse(url) + if parsed.scheme != "https": + raise ValueError("Telegram URL must use https") + + host = parsed.hostname or "api.telegram.org" + port = parsed.port or 443 + path = parsed.path or "/" + if parsed.query: + path = f"{path}?{parsed.query}" + + body = json.dumps(payload) + request = ( + f"POST {path} HTTP/1.1\r\n" + f"Host: {host}\r\n" + "Content-Type: application/json\r\n" + f"Content-Length: {len(body.encode('utf-8'))}\r\n" + "Connection: close\r\n" + "\r\n" + f"{body}" + ) + + ctx = ssl.create_default_context() + addrinfos = socket.getaddrinfo(host, port, family=socket.AF_INET, type=socket.SOCK_STREAM) + if not addrinfos: + raise FetchError("Telegram IPv4 resolution failed") + + last_exc = None + response_bytes = b"" + for family, socktype, proto, _, sockaddr in addrinfos: + try: + with socket.socket(family, socktype, proto) as sock: + sock.settimeout(timeout) + sock.connect(sockaddr) + with ctx.wrap_socket(sock, server_hostname=host) as ssock: + ssock.sendall(request.encode("utf-8")) + while True: + chunk = ssock.recv(4096) + if not chunk: + break + response_bytes += chunk + break + except Exception as exc: + last_exc = exc + continue + + if not response_bytes: + raise FetchError(f"Telegram IPv4 POST failed: {last_exc}") from last_exc + + parts = response_bytes.split(b"\r\n\r\n", 1) + if len(parts) != 2: + raise FetchError("Telegram IPv4 response parse failed") + + status_line = parts[0].split(b"\r\n", 1)[0].decode("utf-8", errors="ignore") + body_text = parts[1].decode("utf-8", errors="ignore") + if not status_line.startswith("HTTP/1.1 200"): + raise FetchError(f"Telegram IPv4 HTTP error: {status_line} {body_text}") + + try: + return json.loads(body_text) + except json.JSONDecodeError as exc: + raise FetchError(f"Telegram IPv4 response JSON decode failed: {exc}") from exc + + +def edit_pinned_message(text: str) -> None: + if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID or not TELEGRAM_MESSAGE_ID: + raise FetchError("Missing TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID or TELEGRAM_MESSAGE_ID in environment") + + url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/editMessageText" + payload = { + "chat_id": TELEGRAM_CHAT_ID, + "message_id": int(TELEGRAM_MESSAGE_ID), + "text": text, + "parse_mode": "HTML", + } + try: + with httpx.Client(timeout=10.0) as client: + response = client.post(url, json=payload) + response.raise_for_status() + data = response.json() + if not data.get("ok"): + raise FetchError(f"Telegram editMessageText failed: {data}") + except Exception as exc: + if isinstance(exc, OSError) or "Network is unreachable" in str(exc): + data = post_telegram_ipv4(url, payload) + if not data.get("ok"): + raise FetchError(f"Telegram editMessageText failed: {data}") + else: + raise FetchError(f"Telegram editMessageText failed: {exc}") from exc + + +def main() -> int: + try: + usd = fetch_usd_rub() + btc = fetch_btc_usd() + fng_value, fng_label = fetch_fear_and_greed() + message = build_message(usd, btc, fng_value, fng_label) + + logging.info("Fetched values: USD/RUB=%s BTC/USD=%s FNG=%s (%s)", usd, btc, fng_value, fng_label) + edit_pinned_message(message) + logging.info("Pinned message updated successfully") + return 0 + except FetchError as exc: + logging.error("Update aborted: %s", exc) + return 1 + except Exception as exc: + logging.exception("Unexpected error") + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..02f8db9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +httpx==0.27.0 +python-dotenv==1.1.0