# 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)