SupplyAgent/api/services/aigc_article_html.py

"""Read the latest article visualization HTML from the external AIGC database."""

from __future__ import annotations

from functools import lru_cache
from typing import Any

from sqlalchemy import create_engine, text
from sqlalchemy.pool import QueuePool

from supply_infra.config import get_infra_settings


@lru_cache(maxsize=1)
def _get_aigc_readonly_engine() -> Any:
    settings = get_infra_settings()
    if not all(
        (
            settings.aigc_readonly_mysql_host,
            settings.aigc_readonly_mysql_user,
            settings.aigc_readonly_mysql_database,
        )
    ):
        raise RuntimeError("AIGC read-only MySQL connection is not configured")
    return create_engine(
        settings.aigc_readonly_mysql_url,
        poolclass=QueuePool,
        pool_size=2,
        max_overflow=0,
        pool_timeout=10,
        pool_recycle=1800,
        pool_pre_ping=True,
        connect_args={
            "connect_timeout": 10,
            "read_timeout": 30,
            "write_timeout": 30,
        },
    )


def get_latest_article_html(channel_content_id: str) -> str | None:
    statement = text(
        """
        SELECT t2.html
        FROM aigc_task_input_usage t1
        JOIN aigc_task_callback_data t2
          ON t2.task_instance_id = t1.task_instance_id
        WHERE t1.biz_unique_id = :channel_content_id
        ORDER BY t2.id DESC
        LIMIT 1
        """
    )
    with _get_aigc_readonly_engine().connect() as connection:
        value = connection.scalar(
            statement,
            {"channel_content_id": channel_content_id},
        )
    if value is None:
        return None
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="replace")
    return str(value)