From 3e690b0e05994e36f4ca48735b2ec29c42890fdf9e1a28c9457d9e98a517a896 Mon Sep 17 00:00:00 2001 From: Vladislav Date: Tue, 30 Jun 2026 19:13:54 -0400 Subject: [PATCH] =?UTF-8?q?Bot=20commit:=20=D0=94=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=20=D1=84=D0=B0=D0=B9=D0=BB=20tg.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tg.py | 243 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 243 insertions(+) create mode 100644 tg.py diff --git a/tg.py b/tg.py new file mode 100644 index 0000000..862f313 --- /dev/null +++ b/tg.py @@ -0,0 +1,243 @@ +# TG + +## Class + +from config import * +from io import BytesIO +from PIL import Image +import httpx +import os +from celery import Celery +import asyncio +import logging +import hashlib +from pathlib import Path +from images import * +import json +from typing import List, Optional +import asyncio +import json +from datetime import datetime, time, timedelta, timezone + + +Image.MAX_IMAGE_PIXELS = 50_000_000 +TMP_DIR = Path("/tmp") +TMP_DIR.mkdir(parents=True, exist_ok=True) + +celery_app_tg = Celery('tg', broker=f'amqps://admin:{RABBITMQ_PASS}@{RABBITMQ_HOST}/tg', task_track_started=True ) + + +async def fetch_bytes(client: httpx.AsyncClient, url: str) -> bytes: + async with client.stream("GET", url, timeout=60000) as resp: + resp.raise_for_status() + chunks = [] + async for chunk in resp.aiter_bytes(): + if chunk: + chunks.append(chunk) + return b"".join(chunks) + + +async def new_channel_reg(chat_id): + async with httpx.AsyncClient(http2=True, http1=False, timeout=30.0, verify=ctx) as client: + res = await client.post(f"https://api.telegram.org/bot{BOT_TOKEN}/getChat", data={'chat_id':chat_id}) + channel_data = res.json()['result'] + channel_name = channel_data['title'] + file_id = channel_data['photo']['big_file_id'] + if file_id: + res = await client.post(f"https://api.telegram.org/bot{BOT_TOKEN}/getFile", data={'file_id': file_id}) + file_path = res.json()['result']['file_path'] + img = await fetch_bytes(client, f"https://api.telegram.org/file/bot{BOT_TOKEN}/{file_path}") + new_tg = (channel_name, BytesIO(img)) + print(channel_name) + + +def _cache_path(url: str) -> Path: + h = hashlib.sha256(url.encode()).hexdigest() + return TMP_DIR / f"{h}.img" + + +REPLACE_FROM = "avtopost.kultmost.ru" +REPLACE_TO = "kultmost.spiritofweb.ru" + + +async def _fetch_or_load(client: httpx.AsyncClient, url: str) -> bytes: + cache_file = _cache_path(url) + if cache_file.is_file(): + return cache_file.read_bytes() + + last_exc: Exception | None = None + for attempt in range(2): + try: + resp = await client.get(url, timeout=30) + resp.raise_for_status() + data = resp.content + + temp_name = f"{cache_file}.tmp" + Path(temp_name).write_bytes(data) + os.replace(temp_name, cache_file) + return data + except Exception as e: + last_exc = e + if attempt == 0: + url = url.replace(REPLACE_FROM, REPLACE_TO) + + raise last_exc # type: ignore + + +async def get_images(urls: list[str], sign_avatar: bool = True, quality: int = 95): + async with httpx.AsyncClient(http2=True, http1=False, timeout=30.0, verify=True) as client: + bytes_list = await asyncio.gather( + *(_fetch_or_load(client, u) for u in urls) + ) + + if sign_avatar: + avatar_bytes = bytes_list[0] + photo_bytes = bytes_list[1:] + logo = Image.open(BytesIO(avatar_bytes)).convert("RGB") + logo.thumbnail((200, 200), Image.Resampling.LANCZOS) + else: + photo_bytes = bytes_list + + images: list = [] + MAX_SIZE = (3000, 3000) # Telegram-friendly + memory safe + + for b in photo_bytes: + img = Image.open(BytesIO(b)).convert("RGB") + + # <<< CRITICAL: Resize early >>> + img.thumbnail(MAX_SIZE, Image.Resampling.LANCZOS) + + if sign_avatar: + img = insert_logo(img, logo, quality=quality) # now receives already resized image + else: + bio = BytesIO() + img.save(bio, format="JPEG", quality=quality, optimize=True) + bio.seek(0) + img = bio + + images.append(img) + + return images + +def build_media(images_len: int, text: str) -> list: + media = [] + for i in range(images_len): + if i == 0: + media.append({ + "type": "photo", + "media": f"attach://file{i}", + "caption": text, + "parse_mode": "HTML", + }) + else: + media.append({ + "type": "photo", + "media": f"attach://file{i}", + }) + return media + +def is_msk_night_window() -> bool: + now_msk_dt = datetime.now(timezone.utc) + MSK_OFFSET + now_msk = now_msk_dt.time() + + start = time(22, 0) + end = time(8, 0) + + return (now_msk >= start) or (now_msk < end) + +async def send_tg(urls: list, text: str, chat_id: str, sign_avatar: bool = True, max_retries: int = 4): + url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMediaGroup" + + plans = [ + {"limit": None, "webp": False, "webp_quality": 90}, + {"limit": 85, "webp": False, "webp_quality": 85}, + {"limit": None, "webp": True, "webp_quality": 90}, + {"limit": None, "webp": True, "webp_quality": 80}, + ][:max_retries] + + last_exc: Optional[Exception] = None + + night = is_msk_night_window() + + for attempt, plan in enumerate(plans, start=1): + try: + if plan["limit"] is None: + images: List[BytesIO] = await get_images(urls, sign_avatar) + else: + images = await get_images(urls, sign_avatar, plan["limit"]) + + if plan.get("webp", False): + images_webp: List[BytesIO] = [] + for bio in images: + images_webp.append(await convert_bio_to_webp(bio, quality=plan["webp_quality"])) + images = images_webp + + media = build_media(len(images), text) + + multipart = [ + ("chat_id", (None, str(chat_id))), + ("media", (None, json.dumps(media))), + ] + + if night: + multipart.append(("disable_notification", (None, "true"))) + + for i, bio in enumerate(images): + bio.seek(0) + ext = "webp" if plan.get("webp", False) else "jpg" + filename = f"image{i}.{ext}" + content = bio.read() + content_type = f"image/{ext}" + multipart.append((f"file{i}", (filename, content, content_type))) + + async with httpx.AsyncClient(http2=True, timeout=60.0) as client: + resp = await client.post(url, files=multipart) + + if resp.status_code >= 300: + raise RuntimeError(f"Telegram HTTP {resp.status_code}: {resp.text}") + + return resp.json() + + except Exception as e: + last_exc = e + if attempt == len(plans): + break + await asyncio.sleep(0.8 * attempt) + + raise last_exc + + +@celery_app_tg.task( + bind=True, + rate_limit='60/m', + name='tg.send_tg', + retry_backoff=True, + retry_backoff_max=600, + retry_jitter=True, + max_retries=20, + soft_time_limit=600, + time_limit=700, +) +def notify(self, item: dict, text: str, urls: list, social: dict, sign_avatar: bool = True): + try: + chat_id = social.get('screen_name') + logging.info(f"TG начало публикации {item}") + result = asyncio.run(send_tg(urls, text, chat_id, sign_avatar)) + result['doc'] = social + httpx.post(f'{FULL_PATH_DOMAIN}/tg', json={'item': item, 'result': result}, headers={"Authorization": f"bearer {ATOKEN}"}) + except Exception as exc: + raise self.retry(exc=exc, countdown=30) + + +@celery_app_tg.task( + rate_limit='60/m', + name='tg.new_channel_reg', + retry_kwargs={'max_retries': None, 'countdown': 30}, + retry_backoff=True, +) +def new_channel(self, chat_id: str): + try: + logging.info(f"TG новый канал {chat_id}") + asyncio.run(new_channel_reg(chat_id)) + except Exception as exc: + raise self.retry(exc=exc, countdown=30)