274 lines
8.7 KiB
Python
274 lines
8.7 KiB
Python
# `tg.py` generate from `tg.ipynb` by `build.ipynb`
|
|
# 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}:{RABBITMQ_PORT}/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"
|
|
|
|
|
|
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
|
|
|
|
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
|
|
|
|
|
|
def delete_message(chat_id: int, message_id: int) -> bool:
|
|
API_URL = f"https://api.telegram.org/bot{BOT_TOKEN}/deleteMessage"
|
|
payload = {
|
|
"chat_id": chat_id,
|
|
"message_id": message_id,
|
|
}
|
|
# Telegram expects a POST request with JSON body
|
|
with httpx.Client() as client:
|
|
response = client.post(API_URL, json=payload, timeout=10)
|
|
# The API returns JSON like {"ok":true,"result":true}
|
|
data = response.json()
|
|
return data.get("ok") and data.get("result") is True
|
|
|
|
@celery_app_tg.task(
|
|
bind=True,
|
|
rate_limit='60/m',
|
|
name='tg.delete_message',
|
|
retry_backoff=True,
|
|
retry_backoff_max=600,
|
|
retry_jitter=True,
|
|
max_retries=20,
|
|
soft_time_limit=600,
|
|
time_limit=700,
|
|
)
|
|
def delete_from_tg(self, message_id: str, social: dict):
|
|
try:
|
|
chat_id = social.get('screen_name')
|
|
logging.info(f"TG удаление публикации {chat_id} {message_id}")
|
|
result = asyncio.run(delete_message(chat_id, message_id))
|
|
# 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(
|
|
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)
|