from __future__ import annotations

import asyncio
import json
import mimetypes
import os
import re
import shutil
import sys
import tempfile
from datetime import datetime, time, timedelta
from pathlib import Path
from typing import Any
from urllib.parse import quote
from zoneinfo import ZoneInfo

import aiohttp
from loguru import logger

from ..config import settings
from ..constants import (
    PUBLICATION_STATUS_FAILED,
    PUBLICATION_STATUS_PENDING,
    PUBLICATION_STATUS_PUBLISHED,
    WORKER_MAX_POSTER,
)
from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
from ..text_utils import build_publication_text

LOCAL_TZ = ZoneInfo("Asia/Yekaterinburg")
MAX_MESSAGE_LIMIT = 4000
MAX_MEDIA_ITEMS = 10
MAX_VIDEO_BYTES = 250 * 1024 * 1024
PROJECT_ROOT = Path(__file__).resolve().parents[3]
UPLOAD_ROOT = PROJECT_ROOT / "uploads"


class MAXAPIError(RuntimeError):
    def __init__(self, message: str, status: int | None = None, code: str = "") -> None:
        super().__init__(message)
        self.status = status
        self.code = code


def valid_time(value: str) -> bool:
    return bool(re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", str(value or "").strip()))


def normalize_schedule(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        value = []
    rows: list[dict[str, Any]] = []
    for idx, item in enumerate(value):
        if not isinstance(item, dict):
            continue
        time_value = str(item.get("time") or "").strip()
        if not valid_time(time_value):
            continue
        count = max(0, min(20, int(item.get("count") or 0)))
        if count < 1:
            continue
        rows.append(
            {
                "id": str(item.get("id") or f"slot-{idx + 1}").strip()[:80],
                "time": time_value,
                "count": count,
                "enabled": bool(item.get("enabled", True)),
            }
        )
    return sorted(rows, key=lambda row: row["time"])


def split_message_chunks(text: str, limit: int) -> list[str]:
    text = str(text or "").strip()
    if not text:
        return []
    chunks: list[str] = []
    while len(text) > limit:
        split_at = text.rfind("\n", 0, limit)
        if split_at < limit // 2:
            split_at = text.rfind(" ", 0, limit)
        if split_at < limit // 2:
            split_at = limit
        chunks.append(text[:split_at].strip())
        text = text[split_at:].strip()
    if text:
        chunks.append(text)
    return chunks


def local_upload_path(url: str) -> Path | None:
    raw = str(url or "").strip()
    prefix = "/uploads/"
    if not raw.startswith(prefix):
        return None
    path = (UPLOAD_ROOT / raw.removeprefix(prefix)).resolve()
    try:
        path.relative_to(UPLOAD_ROOT.resolve())
    except ValueError:
        return None
    return path if path.is_file() else None


class MAXAPIClient:
    def __init__(self, token: str, base_url: str, timeout_sec: int, max_attempts: int, retry_backoff_max_sec: int) -> None:
        self.token = token
        self.base_url = base_url.rstrip("/")
        self.timeout_sec = timeout_sec
        self.max_attempts = max_attempts
        self.retry_backoff_max_sec = retry_backoff_max_sec
        self.session: aiohttp.ClientSession | None = None

    async def __aenter__(self) -> "MAXAPIClient":
        timeout = aiohttp.ClientTimeout(total=self.timeout_sec)
        self.session = aiohttp.ClientSession(timeout=timeout, headers={"Authorization": self.token})
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        if self.session:
            await self.session.close()

    async def request(self, method: str, path: str, **kwargs) -> dict[str, Any]:
        if not self.session:
            raise RuntimeError("MAX session is not initialized")
        url = f"{self.base_url}{path}"
        last_error = ""
        for attempt in range(1, self.max_attempts + 1):
            try:
                async with self.session.request(method, url, **kwargs) as resp:
                    text = await resp.text()
                    try:
                        data = json.loads(text) if text else {}
                    except json.JSONDecodeError:
                        data = {"raw": text}
                    if 200 <= resp.status < 300:
                        return data
                    retry_after = resp.headers.get("Retry-After")
                    message = data.get("message") if isinstance(data, dict) else text
                    code = data.get("code") if isinstance(data, dict) else ""
                    last_error = f"MAX API {resp.status} {code}: {message or text}"
                    if resp.status == 429 and retry_after:
                        await asyncio.sleep(float(retry_after) + 0.5)
                        continue
                    if code == "attachment.not.ready" and attempt < self.max_attempts:
                        await asyncio.sleep(min(2 ** attempt, self.retry_backoff_max_sec))
                        continue
                    if resp.status < 500:
                        raise MAXAPIError(last_error, resp.status, str(code or ""))
            except Exception as exc:
                last_error = str(exc)
                if attempt >= self.max_attempts:
                    raise
            await asyncio.sleep(min(2 ** attempt, self.retry_backoff_max_sec))
        raise RuntimeError(last_error or "MAX API request failed")

    async def send_message(self, chat_id: int, text: str, attachments: list[dict[str, Any]] | None = None) -> dict[str, Any]:
        payload: dict[str, Any] = {"text": text, "notify": True}
        if attachments:
            payload["attachments"] = attachments
        return await self.request("POST", f"/messages?chat_id={chat_id}", json=payload)

    async def get_message(self, message_id: str) -> dict[str, Any]:
        return await self.request("GET", f"/messages/{quote(message_id, safe='')}")

    async def get_video_info(self, token: str) -> dict[str, Any]:
        return await self.request("GET", f"/videos/{quote(token, safe='')}")

    async def get_upload_url(self, media_type: str) -> dict[str, Any]:
        upload_type = "video" if media_type == "video" else "image"
        return await self.request("POST", f"/uploads?type={upload_type}")

    async def upload_file(self, upload_url: str, path: Path) -> dict[str, Any]:
        if not self.session:
            raise RuntimeError("MAX session is not initialized")
        form = aiohttp.FormData()
        content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
        with path.open("rb") as fh:
            form.add_field("data", fh, filename=path.name, content_type=content_type)
            async with self.session.post(upload_url, data=form) as resp:
                text = await resp.text()
        try:
            data = json.loads(text) if text else {}
        except json.JSONDecodeError:
            data = {"raw": text}
        if resp.status < 200 or resp.status >= 300:
            raise RuntimeError(f"MAX upload {resp.status}: {data}")
        return data

    async def react(self, path_template: str, message_id: str, reaction: str) -> None:
        path = path_template.format(message_id=message_id)
        payload = {"reaction": reaction}
        await self.request("POST", path, json=payload)


class MAXPoster:
    def __init__(self) -> None:
        self.pool = None
        self.heartbeat = HeartbeatReporter(WORKER_MAX_POSTER, 30)
        self.token = ""
        self.chat_id = 0
        self.api_base_url = "https://platform-api2.max.ru"
        self.schedule: list[dict[str, Any]] = []
        self.interval_sec = 60
        self.message_limit = MAX_MESSAGE_LIMIT
        self.media_max_items = MAX_MEDIA_ITEMS
        self.media_process_delay_sec = 3.0
        self.video_ready_attempts = 5
        self.video_ready_delay_sec = 10.0
        self.max_attempts = 3
        self.retry_backoff_max_sec = 30
        self.send_delay_sec = 1.0
        self.recent_window = 20
        self.category_repeat_penalty = 3.0
        self.source_repeat_penalty = 4.0
        self.auto_reaction_enabled = True
        self.auto_reaction = "👍"
        self.reaction_path_template = "/messages/{message_id}/reactions"
        self.dry_run = False

    async def init(self) -> None:
        self.pool = await get_pool()

    async def reload_settings(self) -> None:
        self.token = str(await fetch_setting("max_poster_bot_token", "") or "").strip()
        self.chat_id = int(str(await fetch_setting("max_poster_chat_id", "0") or "0").strip())
        self.api_base_url = str(await fetch_setting("max_poster_api_base_url", self.api_base_url) or self.api_base_url).strip()
        self.schedule = normalize_schedule(await fetch_setting("max_poster_schedule_json", []))
        self.interval_sec = max(10, await fetch_int_setting("max_poster_interval_sec", 60))
        self.message_limit = max(512, min(MAX_MESSAGE_LIMIT, await fetch_int_setting("max_poster_message_limit", MAX_MESSAGE_LIMIT)))
        self.media_max_items = max(0, min(MAX_MEDIA_ITEMS, await fetch_int_setting("max_poster_media_max_items", MAX_MEDIA_ITEMS)))
        self.media_process_delay_sec = max(0.0, await fetch_float_setting("max_poster_media_process_delay_sec", 3.0))
        self.video_ready_attempts = max(1, await fetch_int_setting("max_poster_video_ready_attempts", 5))
        self.video_ready_delay_sec = max(1.0, await fetch_float_setting("max_poster_video_ready_delay_sec", 10.0))
        self.max_attempts = max(1, await fetch_int_setting("max_poster_max_attempts", 3))
        self.retry_backoff_max_sec = max(1, await fetch_int_setting("max_poster_retry_backoff_max_sec", 30))
        self.send_delay_sec = max(0.0, await fetch_float_setting("max_poster_send_delay_sec", 1.0))
        self.recent_window = max(1, await fetch_int_setting("max_poster_recent_window", 20))
        self.category_repeat_penalty = max(0.0, await fetch_float_setting("max_poster_category_repeat_penalty", 3.0))
        self.source_repeat_penalty = max(0.0, await fetch_float_setting("max_poster_source_repeat_penalty", 4.0))
        self.auto_reaction_enabled = await fetch_bool_setting("max_poster_auto_reaction_enabled", True)
        self.auto_reaction = str(await fetch_setting("max_poster_auto_reaction", "👍") or "👍").strip() or "👍"
        self.reaction_path_template = str(await fetch_setting("max_poster_reaction_path_template", self.reaction_path_template) or "").strip()
        self.dry_run = await fetch_bool_setting("max_poster_dry_run", False)
        if not self.token:
            raise RuntimeError("max_poster_bot_token is empty")
        if not self.chat_id:
            raise RuntimeError("max_poster_chat_id is empty")
        logger.info("MAX poster config: chat={} schedule={} dry_run={}", self.chat_id, self.schedule, self.dry_run)

    async def due_slots(self) -> list[dict[str, Any]]:
        now = datetime.now(LOCAL_TZ)
        today = now.date()
        due: list[dict[str, Any]] = []
        for row in self.schedule:
            if not row.get("enabled"):
                continue
            hour, minute = [int(part) for part in str(row["time"]).split(":", 1)]
            slot_start = datetime.combine(today, time(hour=hour, minute=minute), tzinfo=LOCAL_TZ)
            slot_end = slot_start + timedelta(minutes=10)
            if not (slot_start <= now < slot_end):
                continue
            inserted = await self.pool.fetchrow(
                """
                INSERT INTO publication_runs(poster, schedule_id, scheduled_for, scheduled_time, planned_count, status)
                VALUES($1, $2, $3, $4, $5, 'started')
                ON CONFLICT (poster, schedule_id, scheduled_for) DO NOTHING
                RETURNING id
                """,
                "max",
                row["id"],
                today,
                row["time"],
                int(row["count"]),
            )
            if inserted:
                due.append({**row, "run_id": int(inserted["id"]), "date": today})
        return due

    async def load_recent(self) -> list[dict[str, Any]]:
        rows = await self.pool.fetch(
            """
            SELECT rp.final_category_tag, rp.rewrite_category_tag, rp.final_source_tag, rp.rewrite_source_tag
            FROM post_publications pp
            JOIN raw_posts rp ON rp.id=pp.raw_post_id
            WHERE pp.platform='max'
              AND pp.status=$1
              AND pp.target_id=$2
            ORDER BY pp.published_at DESC NULLS LAST, pp.id DESC
            LIMIT $3::int
            """,
            PUBLICATION_STATUS_PUBLISHED,
            str(self.chat_id),
            self.recent_window,
        )
        return [dict(row) for row in rows]

    async def load_candidates(self, limit: int) -> list[dict[str, Any]]:
        rows = await self.pool.fetch(
            """
            SELECT rp.*, s.name AS source_name, s.tag AS source_tag
            FROM raw_posts rp
            JOIN sources s ON s.id=rp.source_id
            LEFT JOIN post_publications pp ON pp.raw_post_id=rp.id AND pp.platform='max'
            WHERE rp.status='storage_ready'
              AND rp.rewrite_status='ready'
              AND COALESCE(rp.editorial_status, 'review') IN ('accepted', 'published', 'publish_failed')
              AND COALESCE(pp.status, $1) = $1
              AND EXISTS (
                SELECT 1
                FROM raw_post_media rpm
                WHERE rpm.raw_post_id=rp.id
                  AND COALESCE(rpm.editor_hidden, FALSE)=FALSE
                  AND rpm.media_type IN ('photo', 'video')
                  AND COALESCE(rpm.original_url, '') <> ''
              )
            ORDER BY COALESCE(rp.reviewed_at, rp.edited_at, rp.rewritten_at, rp.created_at) ASC,
                     rp.id ASC
            LIMIT $2::int
            """,
            PUBLICATION_STATUS_PENDING,
            max(limit, 50),
        )
        return [dict(row) for row in rows]

    def choose_posts(self, candidates: list[dict[str, Any]], recent: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
        selected: list[dict[str, Any]] = []
        recent_categories = [str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "") for row in recent]
        recent_sources = [str(row.get("final_source_tag") or row.get("rewrite_source_tag") or "") for row in recent]
        pool = list(candidates)
        while pool and len(selected) < count:
            best_idx = 0
            selected_categories = [str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "") for row in selected]
            selected_sources = [str(row.get("final_source_tag") or row.get("rewrite_source_tag") or row.get("source_tag") or "") for row in selected]
            previous_category = selected_categories[-1] if selected_categories else (recent_categories[0] if recent_categories else "")
            previous_source = selected_sources[-1] if selected_sources else (recent_sources[0] if recent_sources else "")
            best_rank: tuple[int, float, int] | None = None
            for idx, post in enumerate(pool):
                category = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or "")
                source = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
                immediate_repeats = int(bool(category) and category == previous_category) + int(bool(source) and source == previous_source)
                repeat_penalty = (
                    (recent_categories.count(category) + selected_categories.count(category)) * self.category_repeat_penalty
                    + (recent_sources.count(source) + selected_sources.count(source)) * self.source_repeat_penalty
                )
                rank = (immediate_repeats, repeat_penalty, idx)
                if best_rank is None or rank < best_rank:
                    best_idx = idx
                    best_rank = rank
            selected.append(pool.pop(best_idx))
        return selected

    async def load_media(self, raw_post_id: int) -> list[dict[str, Any]]:
        rows = await self.pool.fetch(
            """
            SELECT *
            FROM raw_post_media
            WHERE raw_post_id=$1
              AND COALESCE(editor_hidden, FALSE)=FALSE
              AND media_type IN ('photo', 'video')
              AND COALESCE(original_url, '') <> ''
            ORDER BY sort_order ASC, id ASC
            LIMIT $2::int
            """,
            raw_post_id,
            self.media_max_items,
        )
        return [dict(row) for row in rows]

    def build_text(self, post: dict[str, Any]) -> str:
        category_tag = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or post.get("final_category") or post.get("rewrite_category") or "")
        source_tag = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
        return build_publication_text(post.get("final_text") or post.get("rewritten_text") or "", category_tag, source_tag)

    async def download_media_to_temp(self, session: aiohttp.ClientSession, item: dict[str, Any]) -> Path | None:
        media_type = str(item.get("media_type") or "")
        file_id = str(item.get("tg_file_id") or item.get("storage_attachment_id") or "").strip()
        if media_type == "video" and file_id:
            return await self.download_telegram_file_to_temp(session, file_id, ".mp4")
        url = str(item.get("original_url") or "").strip()
        local_path = local_upload_path(url)
        if local_path:
            return local_path
        if not url.startswith(("http://", "https://")):
            return None
        suffix = Path(url.split("?", 1)[0]).suffix[:12] or (".mp4" if item.get("media_type") == "video" else ".jpg")
        tmp = tempfile.NamedTemporaryFile(prefix="max-poster-", suffix=suffix, delete=False)
        path = Path(tmp.name)
        tmp.close()
        total = 0
        try:
            async with session.get(url) as resp:
                resp.raise_for_status()
                with path.open("wb") as fh:
                    async for chunk in resp.content.iter_chunked(1024 * 512):
                        total += len(chunk)
                        if total > MAX_VIDEO_BYTES:
                            raise RuntimeError("media file is larger than MAX video limit")
                        fh.write(chunk)
            return path
        except Exception:
            path.unlink(missing_ok=True)
            raise

    async def download_telegram_file_to_temp(self, session: aiohttp.ClientSession, file_id: str, suffix: str) -> Path:
        token = settings.tg_bot_token
        if not token:
            raise RuntimeError("TG_BOT_TOKEN is empty; cannot download stored video")
        api_base = str(await fetch_setting("local_bot_api_url", settings.local_bot_api_url) or "").strip().rstrip("/")
        if not api_base:
            api_base = "https://api.telegram.org"
        get_file_url = f"{api_base}/bot{token}/getFile"
        async with session.post(get_file_url, json={"file_id": file_id}) as resp:
            data = await resp.json(content_type=None)
        if not data.get("ok") or not isinstance(data.get("result"), dict):
            raise RuntimeError(f"Telegram getFile failed: {data}")
        file_path = str(data["result"].get("file_path") or "").strip()
        if not file_path:
            raise RuntimeError(f"Telegram getFile did not return file_path: {data}")
        local_file = Path(file_path)
        if local_file.is_absolute() and local_file.is_file():
            if local_file.stat().st_size > MAX_VIDEO_BYTES:
                raise RuntimeError("telegram local video file is larger than MAX video limit")
            tmp = tempfile.NamedTemporaryFile(prefix="max-poster-tg-", suffix=suffix, delete=False)
            path = Path(tmp.name)
            tmp.close()
            try:
                shutil.copyfile(local_file, path)
                return path
            except Exception:
                path.unlink(missing_ok=True)
                raise
        download_url = f"{api_base}/file/bot{token}/{file_path}"
        tmp = tempfile.NamedTemporaryFile(prefix="max-poster-tg-", suffix=suffix, delete=False)
        path = Path(tmp.name)
        tmp.close()
        total = 0
        try:
            async with session.get(download_url) as resp:
                resp.raise_for_status()
                with path.open("wb") as fh:
                    async for chunk in resp.content.iter_chunked(1024 * 1024):
                        total += len(chunk)
                        if total > MAX_VIDEO_BYTES:
                            raise RuntimeError("telegram video file is larger than MAX video limit")
                        fh.write(chunk)
            return path
        except Exception:
            path.unlink(missing_ok=True)
            raise

    async def upload_media(self, client: MAXAPIClient, item: dict[str, Any]) -> dict[str, Any] | None:
        media_type = str(item.get("media_type") or "")
        if media_type not in {"photo", "video"}:
            return None
        temp_path: Path | None = None
        downloaded = False
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=180)) as session:
            temp_path = await self.download_media_to_temp(session, item)
            downloaded = bool(temp_path and not local_upload_path(str(item.get("original_url") or "")))
        if not temp_path:
            return None
        try:
            upload_info = await client.get_upload_url("video" if media_type == "video" else "image")
            upload_url = str(upload_info.get("url") or "").strip()
            if not upload_url:
                raise RuntimeError(f"MAX upload URL is empty: {upload_info}")
            payload = await client.upload_file(upload_url, temp_path)
            if media_type == "video" and upload_info.get("token") and not payload.get("token"):
                payload["token"] = upload_info["token"]
            has_image_token = bool(payload.get("photos")) if isinstance(payload.get("photos"), dict) else False
            if not payload.get("token") and not has_image_token:
                raise RuntimeError(f"MAX upload did not return token: {payload}")
            return {"type": "video" if media_type == "video" else "image", "payload": payload}
        finally:
            if downloaded and temp_path:
                temp_path.unlink(missing_ok=True)

    def message_id_from_response(self, data: dict[str, Any]) -> str:
        candidates = [
            data.get("id"),
            data.get("mid"),
            (data.get("message") or {}).get("id") if isinstance(data.get("message"), dict) else None,
            (data.get("message") or {}).get("mid") if isinstance(data.get("message"), dict) else None,
        ]
        message = data.get("message") if isinstance(data.get("message"), dict) else {}
        body = message.get("body") if isinstance(message.get("body"), dict) else {}
        candidates.extend([body.get("id"), body.get("mid"), body.get("message_id")])
        return next((str(value) for value in candidates if value), "")

    def message_url_from_response(self, data: dict[str, Any]) -> str | None:
        candidates = [data.get("url")]
        message = data.get("message") if isinstance(data.get("message"), dict) else {}
        body = message.get("body") if isinstance(message.get("body"), dict) else {}
        candidates.extend([message.get("url"), body.get("url")])
        return next((str(value).strip() for value in candidates if str(value or "").strip()), None)

    async def send_message_waiting_for_media(
        self,
        client: MAXAPIClient,
        text: str,
        attachments: list[dict[str, Any]],
    ) -> dict[str, Any]:
        has_video = any(item.get("type") == "video" for item in attachments)
        attempts = self.video_ready_attempts if has_video else 1
        delay = self.video_ready_delay_sec
        for attempt in range(1, attempts + 1):
            try:
                return await client.send_message(self.chat_id, text, attachments=attachments or None)
            except MAXAPIError as exc:
                if exc.code != "attachment.not.ready" or attempt >= attempts:
                    raise
                logger.warning(
                    "MAX attachment is not ready, retry send in {}s attempt={}/{}",
                    delay,
                    attempt + 1,
                    attempts,
                )
                await asyncio.sleep(delay)
                delay = min(delay * 2, self.retry_backoff_max_sec * 4)
        raise RuntimeError("MAX attachment readiness retry exhausted")

    def video_tokens(self, attachments: list[dict[str, Any]]) -> list[str]:
        tokens: list[str] = []
        for item in attachments:
            if item.get("type") != "video":
                continue
            payload = item.get("payload") if isinstance(item.get("payload"), dict) else {}
            token = str(payload.get("token") or "").strip()
            if token:
                tokens.append(token)
        return tokens

    async def wait_for_videos(self, client: MAXAPIClient, attachments: list[dict[str, Any]]) -> None:
        tokens = self.video_tokens(attachments)
        if not tokens:
            return
        delay = self.video_ready_delay_sec
        for attempt in range(1, self.video_ready_attempts + 1):
            pending: list[str] = []
            for token in tokens:
                info = await client.get_video_info(token)
                if not info.get("urls"):
                    pending.append(token)
            if not pending:
                return
            if attempt >= self.video_ready_attempts:
                raise RuntimeError(f"MAX video is not ready after {attempt} attempts")
            logger.warning(
                "MAX video is not ready, retry info in {}s attempt={}/{} pending={}",
                delay,
                attempt + 1,
                self.video_ready_attempts,
                len(pending),
            )
            await asyncio.sleep(delay)
            delay = min(delay * 2, self.retry_backoff_max_sec * 4)

    async def send_post(self, post: dict[str, Any]) -> tuple[list[str], str | None, list[dict[str, Any]], str | None]:
        text = self.build_text(post)
        chunks = split_message_chunks(text, self.message_limit)
        media_rows = await self.load_media(int(post["id"])) if self.media_max_items else []
        message_ids: list[str] = []
        attachments: list[dict[str, Any]] = []
        first_url: str | None = None
        reaction_error: str | None = None
        async with MAXAPIClient(
            self.token,
            self.api_base_url,
            timeout_sec=max(30, await fetch_int_setting("max_poster_timeout_sec", 120)),
            max_attempts=self.max_attempts,
            retry_backoff_max_sec=self.retry_backoff_max_sec,
        ) as client:
            for item in media_rows:
                try:
                    attachment = await self.upload_media(client, item)
                    if attachment:
                        attachments.append(attachment)
                except Exception as exc:
                    logger.warning("MAX media skipped raw_post={} media_id={}: {}", post["id"], item.get("id"), exc)
                if len(attachments) >= self.media_max_items:
                    break
            if self.media_process_delay_sec and attachments:
                await asyncio.sleep(self.media_process_delay_sec)
            if self.dry_run:
                return ["dry-run"], None, attachments, None
            await self.wait_for_videos(client, attachments)
            first_text = chunks[0] if chunks else ""
            first = await self.send_message_waiting_for_media(client, first_text, attachments)
            message = first.get("message") if isinstance(first.get("message"), dict) else first
            first_url = self.message_url_from_response(message) if isinstance(message, dict) else None
            first_message_id = self.message_id_from_response(first)
            if first_message_id:
                message_ids.append(first_message_id)
                if not first_url:
                    try:
                        first_url = self.message_url_from_response(await client.get_message(first_message_id))
                    except Exception as exc:
                        logger.warning("MAX message URL lookup failed message={}: {}", first_message_id, exc)
                if self.auto_reaction_enabled and self.reaction_path_template:
                    try:
                        await client.react(self.reaction_path_template, first_message_id, self.auto_reaction)
                    except Exception as exc:
                        reaction_error = str(exc)
                        logger.warning("MAX auto reaction failed message={}: {}", first_message_id, exc)
            for chunk in chunks[1:]:
                if self.send_delay_sec:
                    await asyncio.sleep(self.send_delay_sec)
                result = await client.send_message(self.chat_id, chunk)
                message_id = self.message_id_from_response(result)
                if message_id:
                    message_ids.append(message_id)
        return message_ids, first_url, attachments, reaction_error

    async def upsert_publication(
        self,
        post_id: int,
        status: str,
        external_id: str | None = None,
        url: str | None = None,
        error: str | None = None,
        attachments: list[dict[str, Any]] | None = None,
    ) -> None:
        await self.pool.execute(
            """
            INSERT INTO post_publications(
              raw_post_id, platform, status, target_id, external_id, url, error, attachments_json, published_at, updated_at
            )
            VALUES($1, 'max', $2, $3, $4, $5, $6, $7::jsonb, CASE WHEN $2=$8 THEN NOW() ELSE NULL END, NOW())
            ON CONFLICT (raw_post_id, platform) DO UPDATE
            SET status=EXCLUDED.status,
                target_id=EXCLUDED.target_id,
                external_id=EXCLUDED.external_id,
                url=EXCLUDED.url,
                error=EXCLUDED.error,
                attachments_json=EXCLUDED.attachments_json,
                published_at=CASE WHEN EXCLUDED.status=$8 THEN COALESCE(EXCLUDED.published_at, NOW()) ELSE post_publications.published_at END,
                updated_at=NOW()
            """,
            post_id,
            status,
            str(self.chat_id),
            external_id,
            url,
            error[:1000] if error else None,
            json.dumps(attachments or [], ensure_ascii=False),
            PUBLICATION_STATUS_PUBLISHED,
        )

    async def process_slot(self, slot: dict[str, Any]) -> int:
        count = int(slot["count"])
        recent = await self.load_recent()
        candidates = await self.load_candidates(max(50, count * 10))
        posts = self.choose_posts(candidates, recent, count)
        published = 0
        for post in posts:
            await self.heartbeat.beat(self.pool, status="publishing", meta={"post_id": int(post["id"]), "slot": slot["id"]}, force=True)
            try:
                message_ids, url, attachments, reaction_error = await self.send_post(post)
                await self.upsert_publication(
                    int(post["id"]),
                    PUBLICATION_STATUS_PUBLISHED,
                    ",".join(message_ids),
                    url,
                    error=None,
                    attachments=attachments,
                )
                published += 1
                logger.info("MAX published raw_post={} messages={} attachments={}", post["id"], message_ids, len(attachments))
            except Exception as exc:
                await self.upsert_publication(int(post["id"]), PUBLICATION_STATUS_FAILED, error=str(exc))
                logger.exception("MAX publish failed raw_post={}: {}", post["id"], exc)
            if self.send_delay_sec:
                await asyncio.sleep(self.send_delay_sec)
        await self.pool.execute(
            """
            UPDATE publication_runs
            SET published_count=$2,
                status=$3,
                error=$4,
                updated_at=NOW()
            WHERE id=$1
            """,
            int(slot["run_id"]),
            published,
            "done" if published == count else "partial",
            None if published == count else f"published {published} of {count}",
        )
        return published

    async def run_once(self) -> bool:
        await self.reload_settings()
        enabled = await is_worker_enabled(self.pool, WORKER_MAX_POSTER)
        app_enabled = await fetch_bool_setting("max_poster_enabled", False)
        if not enabled or not app_enabled:
            await self.heartbeat.beat(self.pool, status="disabled", force=True)
            return False
        slots = await self.due_slots()
        if not slots:
            await self.heartbeat.beat(self.pool, status="idle")
            return False
        total = 0
        for slot in slots:
            total += await self.process_slot(slot)
        await self.heartbeat.beat(self.pool, status="idle", meta={"published": total}, force=True)
        return bool(total)

    async def run_loop(self) -> None:
        await self.init()
        logger.info("{} started pid={}", WORKER_MAX_POSTER, os.getpid())
        while True:
            try:
                await self.run_once()
            except Exception as exc:
                logger.exception("MAX poster loop error: {}", exc)
            await asyncio.sleep(self.interval_sec)


async def main() -> None:
    logger.remove()
    logger.add(sys.stdout, level=settings.log_level)
    worker = MAXPoster()
    await worker.run_loop()


if __name__ == "__main__":
    asyncio.run(main())
