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 utf16_len(s: str) -> int: return len(s.encode("utf-16-le")) // 2 def build_message(usd: float, btc: float, fng_value: int, fng_label: str) -> tuple[str, list]: usd_emoji = "💲" btc_emoji = "💵" btc_str = f"{btc:,.0f}".replace(",", " ") prefix = f"{usd_emoji} ₽{usd:.2f} " text = f"{prefix}{btc_emoji} ${btc_str} 😱 {fng_value} ({fng_label})" entities = [ {"type": "custom_emoji", "offset": 0, "length": utf16_len(usd_emoji), "custom_emoji_id": "5229186695072590160"}, {"type": "custom_emoji", "offset": utf16_len(prefix), "length": utf16_len(btc_emoji), "custom_emoji_id": "5438436707164761151"}, ] return text, entities 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, entities: list) -> 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, "entities": entities, } 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, entities = 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, entities) 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())