209 lines
6.9 KiB
Python
209 lines
6.9 KiB
Python
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=\"5229186695072590160\">💲</tg-emoji> "
|
|
f"₽{usd:.2f} <tg-emoji emoji-id=\"5438436707164761151\">💵</tg-emoji> "
|
|
f"${btc:,.0f}".replace(",", " ") + f" 😱 {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())
|