{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "4e3a145b",
   "metadata": {},
   "source": [
    "# TG"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "264e843e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# ignore\n",
    "import os\n",
    "\n",
    "def load_env(file_path):\n",
    "    \"\"\"Загружает переменные окружения из файла .env с поддержкой булевых значений.\"\"\"\n",
    "    with open(file_path) as f:\n",
    "        for line in f:\n",
    "            line = line.strip()\n",
    "            if not line or line.startswith('#'):\n",
    "                continue\n",
    "            \n",
    "            try:\n",
    "                key, value = line.split('=', 1)\n",
    "                key = key.strip()\n",
    "                value = value.strip('\"\\' ')\n",
    "                \n",
    "                # Преобразование строковых значений в булевые\n",
    "                if value.lower() in ['true', '1', 'yes']:\n",
    "                    value = True\n",
    "                elif value.lower() in ['false', '0', 'no']:\n",
    "                    value = False\n",
    "                \n",
    "                # Проверка, что ключ не пустой\n",
    "                if key:\n",
    "                    os.environ[key] = str(value)  # Храним как строку\n",
    "\n",
    "            except ValueError:\n",
    "                print(f\"Ошибка: неверный формат в строке '{line}'\")\n",
    "\n",
    "try:\n",
    "    load_env('env')\n",
    "except:\n",
    "    load_env('env_example')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dd08738b",
   "metadata": {},
   "source": [
    "## Class"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "49f00c66",
   "metadata": {},
   "outputs": [],
   "source": [
    "from config import *\n",
    "from io import BytesIO\n",
    "from PIL import Image\n",
    "import httpx\n",
    "import os\n",
    "from celery import Celery\n",
    "import asyncio\n",
    "import logging\n",
    "import hashlib\n",
    "from pathlib import Path\n",
    "from images import *\n",
    "import json\n",
    "from typing import List, Optional\n",
    "import asyncio\n",
    "import json\n",
    "from datetime import datetime, time, timedelta, timezone\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "63cb06af",
   "metadata": {},
   "outputs": [],
   "source": [
    "Image.MAX_IMAGE_PIXELS = 50_000_000\n",
    "TMP_DIR = Path(\"/tmp\")\n",
    "TMP_DIR.mkdir(parents=True, exist_ok=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a5250bc6",
   "metadata": {},
   "outputs": [],
   "source": [
    "celery_app_tg = Celery('tg', broker=f'amqps://admin:{RABBITMQ_PASS}@{RABBITMQ_HOST}:{RABBITMQ_PORT}/tg', task_track_started=True )\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b628baff",
   "metadata": {},
   "outputs": [],
   "source": [
    "async def fetch_bytes(client: httpx.AsyncClient, url: str) -> bytes:\n",
    "    async with client.stream(\"GET\", url, timeout=60000) as resp:\n",
    "        resp.raise_for_status()\n",
    "        chunks = []\n",
    "        async for chunk in resp.aiter_bytes():\n",
    "            if chunk:\n",
    "                chunks.append(chunk)\n",
    "        return b\"\".join(chunks)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2514ce65",
   "metadata": {},
   "outputs": [],
   "source": [
    "async def new_channel_reg(chat_id):\n",
    "    async with httpx.AsyncClient(http2=True, http1=False, timeout=30.0, verify=ctx) as client:\n",
    "        res = await client.post(f\"https://api.telegram.org/bot{BOT_TOKEN}/getChat\", data={'chat_id':chat_id})\n",
    "        channel_data = res.json()['result']\n",
    "        channel_name = channel_data['title']\n",
    "        file_id = channel_data['photo']['big_file_id']\n",
    "        if file_id:\n",
    "            res = await client.post(f\"https://api.telegram.org/bot{BOT_TOKEN}/getFile\", data={'file_id': file_id})\n",
    "            file_path = res.json()['result']['file_path']\n",
    "            img = await fetch_bytes(client, f\"https://api.telegram.org/file/bot{BOT_TOKEN}/{file_path}\")\n",
    "            new_tg = (channel_name, BytesIO(img))\n",
    "            print(channel_name)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d978426e",
   "metadata": {},
   "outputs": [],
   "source": [
    "def _cache_path(url: str) -> Path:\n",
    "    h = hashlib.sha256(url.encode()).hexdigest()\n",
    "    return TMP_DIR / f\"{h}.img\"\n",
    "\n",
    "\n",
    "REPLACE_FROM = \"avtopost.kultmost.ru\"\n",
    "REPLACE_TO = \"kultmost.spiritofweb.ru\"\n",
    "\n",
    "\n",
    "async def _fetch_or_load(client: httpx.AsyncClient, url: str) -> bytes:\n",
    "    cache_file = _cache_path(url)\n",
    "    if cache_file.is_file():\n",
    "        return cache_file.read_bytes()\n",
    "\n",
    "    last_exc: Exception | None = None\n",
    "    for attempt in range(2):\n",
    "        try:\n",
    "            resp = await client.get(url, timeout=30)\n",
    "            resp.raise_for_status()\n",
    "            data = resp.content\n",
    "\n",
    "            temp_name = f\"{cache_file}.tmp\"\n",
    "            Path(temp_name).write_bytes(data)\n",
    "            os.replace(temp_name, cache_file)\n",
    "            return data\n",
    "        except Exception as e:\n",
    "            last_exc = e\n",
    "            if attempt == 0:\n",
    "                url = url.replace(REPLACE_FROM, REPLACE_TO)\n",
    "\n",
    "    raise last_exc  # type: ignore\n",
    "\n",
    "\n",
    "async def get_images(urls: list[str], sign_avatar: bool = True, quality: int = 95):\n",
    "    async with httpx.AsyncClient(http2=True, http1=False, timeout=30.0, verify=True) as client:\n",
    "        bytes_list = await asyncio.gather(\n",
    "            *(_fetch_or_load(client, u) for u in urls)\n",
    "        )\n",
    "\n",
    "    if sign_avatar:\n",
    "        avatar_bytes = bytes_list[0]\n",
    "        photo_bytes = bytes_list[1:]\n",
    "        logo = Image.open(BytesIO(avatar_bytes)).convert(\"RGB\")\n",
    "        logo.thumbnail((200, 200), Image.Resampling.LANCZOS)\n",
    "    else:\n",
    "        photo_bytes = bytes_list\n",
    "\n",
    "    images: list = []\n",
    "    MAX_SIZE = (3000, 3000)  # Telegram-friendly + memory safe\n",
    "\n",
    "    for b in photo_bytes:\n",
    "        img = Image.open(BytesIO(b)).convert(\"RGB\")\n",
    "        \n",
    "        # <<< CRITICAL: Resize early >>>\n",
    "        img.thumbnail(MAX_SIZE, Image.Resampling.LANCZOS)\n",
    "        \n",
    "        if sign_avatar:\n",
    "            img = insert_logo(img, logo, quality=quality)  # now receives already resized image\n",
    "        else:\n",
    "            bio = BytesIO()\n",
    "            img.save(bio, format=\"JPEG\", quality=quality, optimize=True)\n",
    "            bio.seek(0)\n",
    "            img = bio\n",
    "        \n",
    "        images.append(img)\n",
    "\n",
    "    return images"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "03c429cb",
   "metadata": {},
   "outputs": [],
   "source": [
    "def build_media(images_len: int, text: str) -> list:\n",
    "    media = []\n",
    "    for i in range(images_len):\n",
    "        if i == 0:\n",
    "            media.append({\n",
    "                \"type\": \"photo\",\n",
    "                \"media\": f\"attach://file{i}\",\n",
    "                \"caption\": text,\n",
    "                \"parse_mode\": \"HTML\",\n",
    "            })\n",
    "        else:\n",
    "            media.append({\n",
    "                \"type\": \"photo\",\n",
    "                \"media\": f\"attach://file{i}\",\n",
    "            })\n",
    "    return media"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "719eb531",
   "metadata": {},
   "outputs": [],
   "source": [
    "def is_msk_night_window() -> bool:\n",
    "    now_msk_dt = datetime.now(timezone.utc) + MSK_OFFSET\n",
    "    now_msk = now_msk_dt.time()\n",
    "\n",
    "    start = time(22, 0)\n",
    "    end = time(8, 0)\n",
    "\n",
    "    return (now_msk >= start) or (now_msk < end)\n",
    "\n",
    "async def send_tg(urls: list, text: str, chat_id: str, sign_avatar: bool = True, max_retries: int = 4):\n",
    "    url = f\"https://api.telegram.org/bot{BOT_TOKEN}/sendMediaGroup\"\n",
    "\n",
    "    plans = [\n",
    "        {\"limit\": None, \"webp\": False, \"webp_quality\": 90},\n",
    "        {\"limit\": 85,   \"webp\": False, \"webp_quality\": 85},\n",
    "        {\"limit\": None, \"webp\": True,  \"webp_quality\": 90},\n",
    "        {\"limit\": None, \"webp\": True,  \"webp_quality\": 80},\n",
    "    ][:max_retries]\n",
    "\n",
    "    last_exc: Optional[Exception] = None\n",
    "\n",
    "    night = is_msk_night_window()\n",
    "\n",
    "    for attempt, plan in enumerate(plans, start=1):\n",
    "        try:\n",
    "            if plan[\"limit\"] is None:\n",
    "                images: List[BytesIO] = await get_images(urls, sign_avatar)\n",
    "            else:\n",
    "                images = await get_images(urls, sign_avatar, plan[\"limit\"])\n",
    "\n",
    "            if plan.get(\"webp\", False):\n",
    "                images_webp: List[BytesIO] = []\n",
    "                for bio in images:\n",
    "                    images_webp.append(await convert_bio_to_webp(bio, quality=plan[\"webp_quality\"]))\n",
    "                images = images_webp\n",
    "\n",
    "            media = build_media(len(images), text)\n",
    "\n",
    "            multipart = [\n",
    "                (\"chat_id\", (None, str(chat_id))),\n",
    "                (\"media\", (None, json.dumps(media))),\n",
    "            ]\n",
    "\n",
    "            if night:\n",
    "                multipart.append((\"disable_notification\", (None, \"true\")))\n",
    "\n",
    "            for i, bio in enumerate(images):\n",
    "                bio.seek(0)\n",
    "                ext = \"webp\" if plan.get(\"webp\", False) else \"jpg\"\n",
    "                filename = f\"image{i}.{ext}\"\n",
    "                content = bio.read()\n",
    "                content_type = f\"image/{ext}\"\n",
    "                multipart.append((f\"file{i}\", (filename, content, content_type)))\n",
    "\n",
    "            async with httpx.AsyncClient(http2=True, timeout=60.0) as client:\n",
    "                resp = await client.post(url, files=multipart)\n",
    "\n",
    "            if resp.status_code >= 300:\n",
    "                raise RuntimeError(f\"Telegram HTTP {resp.status_code}: {resp.text}\")\n",
    "\n",
    "            return resp.json()\n",
    "\n",
    "        except Exception as e:\n",
    "            last_exc = e\n",
    "            if attempt == len(plans):\n",
    "                break\n",
    "            await asyncio.sleep(0.8 * attempt)\n",
    "\n",
    "    raise last_exc\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fc96bf02",
   "metadata": {},
   "outputs": [],
   "source": [
    "def delete_message(chat_id: int, message_id: int) -> bool:\n",
    "    API_URL = f\"https://api.telegram.org/bot{BOT_TOKEN}/deleteMessage\"\n",
    "    payload = {\n",
    "        \"chat_id\": chat_id,\n",
    "        \"message_id\": message_id,\n",
    "    }\n",
    "    # Telegram expects a POST request with JSON body\n",
    "    with httpx.Client() as client:\n",
    "        response = client.post(API_URL, json=payload, timeout=10)\n",
    "    # The API returns JSON like {\"ok\":true,\"result\":true}\n",
    "    data = response.json()\n",
    "    return data.get(\"ok\") and data.get(\"result\") is True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cc5733db",
   "metadata": {},
   "outputs": [],
   "source": [
    "@celery_app_tg.task(\n",
    "    bind=True,\n",
    "    rate_limit='60/m',\n",
    "    name='tg.delete_message',\n",
    "    retry_backoff=True,\n",
    "    retry_backoff_max=600,\n",
    "    retry_jitter=True,\n",
    "    max_retries=20,\n",
    "    soft_time_limit=600,\n",
    "    time_limit=700,\n",
    ")\n",
    "def delete_from_tg(self, message_id: str, social: dict):\n",
    "    try:\n",
    "        chat_id = social.get('screen_name')\n",
    "        logging.info(f\"TG удаление публикации {chat_id} {message_id}\")\n",
    "        result = asyncio.run(delete_message(chat_id, message_id))\n",
    "        # result['doc'] = social\n",
    "        # httpx.post(f'{FULL_PATH_DOMAIN}/tg', json={'item': item, 'result': result}, headers={\"Authorization\": f\"bearer {ATOKEN}\"})\n",
    "    except Exception as exc:\n",
    "        raise self.retry(exc=exc, countdown=30)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b6f354d7",
   "metadata": {},
   "outputs": [],
   "source": [
    "@celery_app_tg.task(\n",
    "    bind=True,\n",
    "    rate_limit='60/m',\n",
    "    name='tg.send_tg',\n",
    "    retry_backoff=True,\n",
    "    retry_backoff_max=600,\n",
    "    retry_jitter=True,\n",
    "    max_retries=20,\n",
    "    soft_time_limit=600,\n",
    "    time_limit=700,\n",
    ")\n",
    "def notify(self, item: dict, text: str, urls: list, social: dict, sign_avatar: bool = True):\n",
    "    try:\n",
    "        chat_id = social.get('screen_name')\n",
    "        logging.info(f\"TG начало публикации {item}\")\n",
    "        result = asyncio.run(send_tg(urls, text, chat_id, sign_avatar))\n",
    "        result['doc'] = social\n",
    "        httpx.post(f'{FULL_PATH_DOMAIN}/tg', json={'item': item, 'result': result}, headers={\"Authorization\": f\"bearer {ATOKEN}\"})\n",
    "    except Exception as exc:\n",
    "        raise self.retry(exc=exc, countdown=30)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8e84bff5",
   "metadata": {},
   "outputs": [],
   "source": [
    "@celery_app_tg.task(\n",
    "    rate_limit='60/m',\n",
    "    name='tg.new_channel_reg',\n",
    "    retry_kwargs={'max_retries': None, 'countdown': 30},\n",
    "    retry_backoff=True,\n",
    ")\n",
    "def new_channel(self, chat_id: str):\n",
    "    try:\n",
    "        logging.info(f\"TG новый канал {chat_id}\")\n",
    "        asyncio.run(new_channel_reg(chat_id))\n",
    "    except Exception as exc:\n",
    "        raise self.retry(exc=exc, countdown=30)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8810213f",
   "metadata": {},
   "source": [
    "## Run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b9773867",
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import os\n",
    "\n",
    "IGNORE = ['build.ipynb', 'data.ipynb']\n",
    "\n",
    "for _filename in [i for i in os.listdir() if 'ipynb' in i and i not in IGNORE]:\n",
    "    filename_without_extension = os.path.splitext(os.path.basename(_filename))[0]\n",
    "\n",
    "    with open(f'{filename_without_extension}.ipynb', 'r') as f:\n",
    "        data = f.read()\n",
    "\n",
    "    main = json.loads(data)\n",
    "    app = []\n",
    "    for cell in main['cells']:\n",
    "        if '## Run' in cell['source']:\n",
    "            break\n",
    "        if '# ignore\\n' in cell['source']:\n",
    "            continue\n",
    "        app.append(''.join(cell['source']))\n",
    "\n",
    "    with open(f'{filename_without_extension}.py', 'w') as f:\n",
    "        f.write('\\n\\n'.join(app))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "540c3f83",
   "metadata": {},
   "outputs": [],
   "source": [
    "!celery -A tg.celery_app_tg worker --loglevel=info"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "62a866f8",
   "metadata": {},
   "source": [
    "## Test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "89ec68ec",
   "metadata": {},
   "outputs": [],
   "source": [
    "a=\"\"\"2645\"\"\".split('\\n')\n",
    "posts = map(int,a)\n",
    "import httpx\n",
    "import os\n",
    "# Bot token – keep it secret!\n",
    "BOT_TOKEN = '6623665792:AAHBDDMIFHlYVZL5tkMSmpNaHfJLE-qy2tw'\n",
    "API_URL = f\"https://api.telegram.org/bot{BOT_TOKEN}/deleteMessage\"\n",
    "\n",
    "\n",
    "def delete_message(chat_id: int, message_id: int) -> bool:\n",
    "    payload = {\n",
    "        \"chat_id\": chat_id,\n",
    "        \"message_id\": message_id,\n",
    "    }\n",
    "    # Telegram expects a POST request with JSON body\n",
    "    with httpx.Client() as client:\n",
    "        response = client.post(API_URL, json=payload, timeout=10)\n",
    "    # The API returns JSON like {\"ok\":true,\"result\":true}\n",
    "    data = response.json()\n",
    "    return data.get(\"ok\") and data.get(\"result\") is True\n",
    "\n",
    "\n",
    "\n",
    "# Example usage\n",
    "if __name__ == \"__main__\":\n",
    "    # Replace with real IDs\n",
    "    chat_id = '@kultmost'\n",
    "    for message_id in posts:\n",
    "        if delete_message(chat_id, message_id):\n",
    "            print(\"Message deleted.\")\n",
    "        else:\n",
    "            print(\"Failed to delete message.\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "416f454c",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.14.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}