init: telegram currency bot with systemd timer

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"<tg-emoji emoji-id=\"5474464267233153364\">💲</tg-emoji> "
+        f"₽{usd:.2f} <tg-emoji emoji-id=\"4931893053163045869\">💲</tg-emoji> "
+        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())