o2o-site-AEO/solution/backend/crud/local_content_crud.py

from sqlalchemy import and_, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert

from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import area_contents
from common.enums import ErrorType, LocalContentType
from common.utils.gtime import GTime


class LocalContentCRUD:
    async def list(self, db, status: int | None = None, region_code: str | None = None):
        """축제·관광지·맛집·날씨 전 종류. ★ 예전엔 FESTIVAL 로 고정돼 있어 sync_region 이 받은
        관광지·맛집이 이 목록에 영영 안 보였다(admin 화면이 축제만 검수/발행하는 줄 알게 됨)."""
        conds = [area_contents.deleted == False]  # noqa: E712
        if status is not None:
            conds.append(area_contents.status == status)
        if region_code:
            conds.append(area_contents.region_code == region_code)
        return await DB_SESSION_MNG.execute(
            db, select(area_contents).where(and_(*conds)).order_by(area_contents.collected_at.desc())
        )

    async def insert(self, db, row):
        return await DB_SESSION_MNG.insert(db, row)

    async def publish(self, db, ids: list, user_id):
        return await DB_SESSION_MNG.add_with_rowcount(
            db,
            update(area_contents).where(
                area_contents.local_content_id.in_(ids), area_contents.deleted == False  # noqa: E712
            ).values(status=2, published_at=GTime.UTC(), published_by=user_id, updated_at=GTime.UTC()),
        )

    async def update(self, db, content_id, data: dict):
        return await DB_SESSION_MNG.add_with_rowcount(
            db,
            update(area_contents).where(
                area_contents.local_content_id == content_id, area_contents.deleted == False  # noqa: E712
            ).values(**data, updated_at=GTime.UTC()),
        )

    async def end(self, db, content_id):
        return await self.update(db, content_id, {"status": 3})

    async def list_keyed(self, db, region_code: str, content_type: int):
        """지역 × 종류의 외부 ID 있는 행 전부(축제·관광지·맛집). 동기화가 기존값과 비교할 때 쓴다."""
        return await DB_SESSION_MNG.execute(
            db,
            select(area_contents).where(
                area_contents.region_code == region_code,
                area_contents.content_type == content_type,
                area_contents.external_id.isnot(None),
                area_contents.deleted == False,  # noqa: E712
            ),
        )

    async def upsert_keyed(self, db, values: dict):
        """외부 ID 로 식별되는 행(축제·관광지·맛집)의 삽입/갱신.

        ★ uq_local_contents_keyed 부분 유니크 인덱스에 태운다 — 같은 지역을 두 번 동기화해도
          중복 행이 생기지 않고 기존 값만 갱신된다."""
        stmt = pg_insert(area_contents).values(**values)
        stmt = stmt.on_conflict_do_update(
            index_elements=[area_contents.region_code, area_contents.content_type, area_contents.external_id],
            index_where=and_(area_contents.deleted == False, area_contents.external_id.isnot(None)),  # noqa: E712
            set_={
                "title": stmt.excluded.title,
                "body": stmt.excluded.body,
                "source": stmt.excluded.source,
                "status": stmt.excluded.status,
                "collected_at": stmt.excluded.collected_at,
                "display_end_at": stmt.excluded.display_end_at,
                "published_at": stmt.excluded.published_at,
                "updated_at": GTime.UTC(),
            },
        )
        return await DB_SESSION_MNG.add(db, stmt)

    async def upsert_kind(self, db, values: dict):
        """지역 이야기 한 종류(가요·인물·…)의 삽입/갱신.

        ★ `uq_local_contents_kind`(region_code, kind — kind IS NOT NULL)에 태운다.
          이 표의 규약은 **한 지역에 종류당 한 벌**이다(migrations/0004). 항목마다 한 행이 아니라
          `body.items` 에 통째로 담긴다 — 사장님이 붙여넣는 같은 종류의 JSON 과 모양을 맞추기
          위해서다. 다시 생성하면 그 한 행을 덮어쓴다.
        ★ external_id 는 넣지 않는다. 넣으면 `uq_local_contents_external`(source, external_id)에도
          걸려, 종류가 다른 두 행이 같은 키로 충돌한다."""
        stmt = pg_insert(area_contents).values(**values)
        stmt = stmt.on_conflict_do_update(
            index_elements=[area_contents.region_code, area_contents.kind],
            index_where=and_(area_contents.deleted == False, area_contents.kind.isnot(None)),  # noqa: E712
            set_={
                "title": stmt.excluded.title,
                "body": stmt.excluded.body,
                "content_type": stmt.excluded.content_type,
                "source": stmt.excluded.source,
                "status": stmt.excluded.status,
                "collected_at": stmt.excluded.collected_at,
                "published_at": stmt.excluded.published_at,
                "updated_at": GTime.UTC(),
            },
        )
        return await DB_SESSION_MNG.add(db, stmt)

    async def list_kinds(self, db, region_code: str):
        """지역의 이야기 행 전부(종류당 1행). cache-aside 판단에 쓴다."""
        return await DB_SESSION_MNG.execute(
            db,
            select(area_contents).where(
                area_contents.region_code == region_code,
                area_contents.kind.isnot(None),
                area_contents.deleted == False,  # noqa: E712
            ),
        )

    async def get_weather(self, db, region_code: str):
        err, rows = await DB_SESSION_MNG.execute(
            db,
            select(area_contents).where(
                area_contents.region_code == region_code,
                area_contents.content_type == LocalContentType.WEATHER.value,
                area_contents.external_id.is_(None),
                area_contents.deleted == False,  # noqa: E712
            ).limit(1),
        )
        return err, rows[0] if rows else None

    async def upsert_weather(self, db, values: dict):
        stmt = pg_insert(area_contents).values(**values)
        stmt = stmt.on_conflict_do_update(
            index_elements=[area_contents.region_code, area_contents.content_type],
            index_where=and_(area_contents.deleted == False, area_contents.external_id.is_(None)),  # noqa: E712
            set_={
                "source": stmt.excluded.source,
                "body": stmt.excluded.body,
                "status": stmt.excluded.status,
                "collected_at": stmt.excluded.collected_at,
                "expires_at": stmt.excluded.expires_at,
                "updated_at": stmt.excluded.updated_at,
            },
        )
        return await DB_SESSION_MNG.add(db, stmt)