From 52142b77b8777bdeaa4b8b8f319f8efb41fc8bcc Mon Sep 17 00:00:00 2001 From: selnem Date: Fri, 12 Jun 2026 20:29:36 +0900 Subject: [PATCH 1/8] =?UTF-8?q?feat:=20=EC=A0=95=EC=B1=85=20=ED=95=80=20sy?= =?UTF-8?q?nc=C2=B7=EB=B0=B0=EC=B9=98=20=EA=B0=80=EA=B3=B5=C2=B7DB=20?= =?UTF-8?q?=EC=A0=81=EC=9E=AC=20=ED=8C=8C=EC=9D=B4=ED=94=84=EB=9D=BC?= =?UTF-8?q?=EC=9D=B8=20=EB=B0=8F=20=EA=B4=80=EB=A6=AC=EC=9E=90=20API=20?= =?UTF-8?q?=EC=B6=94=EA=B0=80=20=EC=A0=95=EC=B1=85=EB=89=B4=EC=8A=A4=20?= =?UTF-8?q?=EC=88=98=EC=A7=91=E2=86=92LLM=20=EA=B0=80=EA=B3=B5=E2=86=92DB?= =?UTF-8?q?=20INSERT=20=EC=9D=BC=EA=B4=84=20=ED=8C=8C=EC=9D=B4=ED=94=84?= =?UTF-8?q?=EB=9D=BC=EC=9D=B8(sync=5Fpipeline)=EA=B3=BC=20/policy-admin=20?= =?UTF-8?q?=EC=97=94=EB=93=9C=ED=8F=AC=EC=9D=B8=ED=8A=B8(sync,=20transform?= =?UTF-8?q?-batch,=20import-batch,=20pipeline-status)=EB=A5=BC=20=EC=B6=94?= =?UTF-8?q?=EA=B0=80=ED=95=9C=EB=8B=A4.=20KST=20=EA=B8=B0=EC=A4=80=20?= =?UTF-8?q?=EC=9E=90=EB=8F=99=20sync=20=EC=8A=A4=EC=BC=80=EC=A4=84?= =?UTF-8?q?=EB=9F=AC,=20=EC=B9=B4=EB=93=9C=EB=89=B4=EC=8A=A4=20S3=20?= =?UTF-8?q?=EC=97=85=EB=A1=9C=EB=93=9C=C2=B7CardnewsImageS3=20=EC=A0=81?= =?UTF-8?q?=EC=9E=AC,=20=EC=A4=91=EB=B3=B5=20=EA=B1=B4=20=EC=8A=A4?= =?UTF-8?q?=ED=82=B5=C2=B7=ED=8C=8C=EC=9D=B4=ED=94=84=EB=9D=BC=EC=9D=B8=20?= =?UTF-8?q?=EC=A0=95=EB=A6=AC,=20Community/EventPin(policy=5Fapi=5Fid)=20?= =?UTF-8?q?=EB=AA=A8=EB=8D=B8=20=ED=99=95=EC=9E=A5=EC=9D=84=20=ED=8F=AC?= =?UTF-8?q?=ED=95=A8=ED=95=9C=EB=8B=A4.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/config.py | 59 ++- app/core/deps.py | 47 +++ app/main.py | 30 +- app/models/CardnewsImageS3.py | 40 ++ app/models/Community.py | 15 +- app/models/EventPin.py | 7 +- app/models/__init__.py | 4 + app/policy_cardnews/template/dispatch.py | 11 +- app/repositories/CardnewsImageS3Repo.py | 27 ++ app/repositories/CommunityRepo.py | 6 + app/repositories/EventPinRepo.py | 22 ++ app/repositories/UserRepo.py | 18 + app/routes/PolicyAdminRoute.py | 160 ++++++++ app/routes/PolicyPinRoute.py | 5 +- app/routes/__init__.py | 2 + app/schemas/PolicyAdminDTO.py | 57 +++ app/schemas/PolicyPinDTO.py | 2 + app/services/PolicyEventIngestService.py | 344 ++++++++++++++++++ app/services/PolicyPinService.py | 180 ++++++++- .../internal/PolicyPinSchedulerService.py | 136 +++++++ app/services/policy_cardnews.py | 165 +++++++-- app/services/policy_pin_transform.py | 234 +++++++++--- app/services/policy_pipeline_cleanup.py | 109 ++++++ app/utils/policy_news_parse.py | 19 + rag/scripts/fetch_policy_news.py | 6 + 25 files changed, 1594 insertions(+), 111 deletions(-) create mode 100644 app/models/CardnewsImageS3.py create mode 100644 app/repositories/CardnewsImageS3Repo.py create mode 100644 app/routes/PolicyAdminRoute.py create mode 100644 app/schemas/PolicyAdminDTO.py create mode 100644 app/services/PolicyEventIngestService.py create mode 100644 app/services/internal/PolicyPinSchedulerService.py create mode 100644 app/services/policy_pipeline_cleanup.py diff --git a/app/core/config.py b/app/core/config.py index 1e98d42..3ec72f9 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -186,6 +186,59 @@ class Settings(BaseSettings): alias="POLICY_SYNC_LOOKBACK_DAYS", description="배치 수집 시 오늘 포함 N일 (API 3일 제한 고려)", ) + policy_sync_interval_days: int = Field( + default=3, + ge=1, + le=30, + alias="POLICY_SYNC_INTERVAL_DAYS", + description="자동 sync 최소 간격(일)", + ) + policy_sync_schedule_hour_kst: int = Field( + default=1, + ge=0, + le=23, + alias="POLICY_SYNC_SCHEDULE_HOUR_KST", + description="정책 핀 sync 스케줄 확인 시각 (KST)", + ) + policy_admin_user_name: str = Field( + default="admin", + validation_alias=AliasChoices("POLICY_ADMIN_USER_NAME", "POLICY_ADMIN_NICKNAME"), + description="정책 핀 등록에 사용할 user.user_name", + ) + policy_sync_batch_size: int = Field( + default=5, + ge=1, + le=25, + alias="POLICY_SYNC_BATCH_SIZE", + description="sync/transform/import 1회 배치 건수", + ) + policy_transform_concurrency: int = Field( + default=3, + ge=1, + le=10, + alias="POLICY_TRANSFORM_CONCURRENCY", + description="정책 pin_content·카드뉴스 Gemini 가공 동시 호출 수", + ) + policy_prune_pipeline_after_import: bool = Field( + default=True, + alias="POLICY_PRUNE_PIPELINE_AFTER_IMPORT", + description="DB INSERT 성공 후 JSONL·로컬 카드뉴스 캐시 제거", + ) + policy_cardnews_keep_local_files: bool = Field( + default=False, + alias="POLICY_CARDNEWS_KEEP_LOCAL_FILES", + description="True면 S3 업로드 후에도 rag/output/policy_cardnews 유지", + ) + policy_sync_merge_documents: bool = Field( + default=False, + alias="POLICY_SYNC_MERGE_DOCUMENTS", + description="False면 sync 수집 시 policy_documents.jsonl을 이번 구간으로 덮어씀", + ) + policy_cardnews_s3_prefix: str = Field( + default="policy-cardnews", + alias="POLICY_CARDNEWS_S3_PREFIX", + description="정책 카드뉴스 S3 object key prefix", + ) gemini_cardnews_image_model: str = Field( default="gemini-2.5-flash-image", alias="GEMINI_CARDNEWS_IMAGE_MODEL", @@ -215,9 +268,9 @@ class Settings(BaseSettings): rag_enable_rerank: bool = Field(default=False, alias="RAG_ENABLE_RERANK") rag_vector_query_mode: str = Field(default="hybrid", alias="RAG_VECTOR_QUERY_MODE") policy_cardnews_font_dir: str = Field( - default="app/assets/fonts", + default="../assets/fonts", alias="POLICY_CARDNEWS_FONT_DIR", - description="Pretendard 등 TTF 폴더", + description="Pretendard 등 폰트 폴더 (app/policy_cardnews 기준 상대 경로)", ) # 한국관광공사 TourAPI (공공데이터포털 활용신청 키) @@ -277,7 +330,7 @@ class Settings(BaseSettings): @classmethod def _empty_string_policy_cardnews_font_dir(cls, value: object) -> object: if value == "": - return "app/assets/fonts" + return "../assets/fonts" return value @field_validator("policy_cardnews_mascot_dir", mode="before") diff --git a/app/core/deps.py b/app/core/deps.py index de34a24..3bbefe5 100644 --- a/app/core/deps.py +++ b/app/core/deps.py @@ -11,6 +11,7 @@ from app.core.exceptions import raise_business_exception from app.login.http_auth import get_current_user_id, get_optional_user_id from app.models.User import User +from app.repositories.CardnewsImageS3Repo import CardnewsImageS3Repo from app.repositories.CommunityRepo import CommunityRepo from app.repositories.ComplaintPetitionRepo import ComplaintPetitionRepo from app.repositories.DepartmentRepo import DepartmentRepo @@ -29,7 +30,9 @@ from app.services.UserService import UserService from app.services.ComplaintEmailService import ComplaintEmailService from app.services.FestivalPinService import FestivalPinService +from app.services.PolicyEventIngestService import PolicyEventIngestService from app.services.PolicyPinService import PolicyPinService +from app.services.internal.PolicyPinSchedulerService import PolicyPinSchedulerService from app.services.ComplaintPetitionService import ComplaintPetitionService from app.services.FestivalEventIngestService import FestivalEventIngestService from app.services.RagRerankService import RagRerankService @@ -519,3 +522,47 @@ def get_policy_pin_service() -> PolicyPinService: PolicyPinService, Depends(get_policy_pin_service), ] + + +def get_cardnews_image_s3_repo(session: DbSessionDep) -> CardnewsImageS3Repo: + return CardnewsImageS3Repo(session) + + +CardnewsImageS3RepoDep = Annotated[CardnewsImageS3Repo, Depends(get_cardnews_image_s3_repo)] + + +def get_policy_event_ingest_service( + pin_repo: PinRepoDep, + event_pin_repo: EventPinRepoDep, + community_repo: CommunityRepoDep, + cardnews_image_s3_repo: CardnewsImageS3RepoDep, + user_repo: UserRepoDep, +) -> PolicyEventIngestService: + return PolicyEventIngestService( + pin_repo=pin_repo, + event_pin_repo=event_pin_repo, + community_repo=community_repo, + cardnews_image_s3_repo=cardnews_image_s3_repo, + user_repo=user_repo, + ) + + +PolicyEventIngestServiceDep = Annotated[ + PolicyEventIngestService, + Depends(get_policy_event_ingest_service), +] + + +def get_policy_pin_scheduler(request: Request) -> PolicyPinSchedulerService | None: + scheduler = getattr(request.app.state, "policy_pin_scheduler", None) + if scheduler is None: + return None + if not isinstance(scheduler, PolicyPinSchedulerService): + raise RuntimeError("policy_pin_scheduler is not initialized correctly.") + return scheduler + + +PolicyPinSchedulerDep = Annotated[ + PolicyPinSchedulerService | None, + Depends(get_policy_pin_scheduler), +] diff --git a/app/main.py b/app/main.py index 4e2fcb7..c274e21 100644 --- a/app/main.py +++ b/app/main.py @@ -23,6 +23,7 @@ from app.services.RagRetrievalService import RagRetrievalService from app.services.VectorStoreService import VectorStoreService from app.services.internal.ComplaintPetitionSchedulerService import ComplaintPetitionSchedulerService +from app.services.internal.PolicyPinSchedulerService import PolicyPinSchedulerService from app.services.internal.ai.ComplaintEmailLLMService import ComplaintEmailLLMService from app.services.internal.ai.VLMService import VLMService from app.services.internal.ai.gemini_retry import parse_gemini_model_list @@ -154,7 +155,7 @@ async def lifespan(app: FastAPI): exc, ) - scheduler = None + complaint_scheduler = None if ( gemini_api_key_secret is not None and getattr(app.state, "vector_store_service", None) is not None @@ -193,24 +194,39 @@ async def lifespan(app: FastAPI): complaint_llm_service=complaint_llm_service, rag_retrieval_service=rag_retrieval_service, ) - scheduler = ComplaintPetitionSchedulerService( + complaint_scheduler = ComplaintPetitionSchedulerService( complaint_email_service=complaint_email_service, s3_util=app.state.s3_util, ) - scheduler.start() - app.state.complaint_scheduler = scheduler + complaint_scheduler.start() + app.state.complaint_scheduler = complaint_scheduler except Exception as exc: logger.warning("Complaint scheduler initialization failed: %s", exc) + policy_pin_scheduler = None + if settings.policy_news_service_key is not None and gemini_api_key_secret is not None: + try: + policy_pin_scheduler = PolicyPinSchedulerService(s3_util=app.state.s3_util) + policy_pin_scheduler.start() + app.state.policy_pin_scheduler = policy_pin_scheduler + except Exception as exc: + logger.warning("Policy pin scheduler initialization failed: %s", exc) + try: yield finally: - scheduler = getattr(app.state, "complaint_scheduler", None) - if scheduler is not None: + complaint_scheduler = getattr(app.state, "complaint_scheduler", None) + if complaint_scheduler is not None: try: - await scheduler.stop() + await complaint_scheduler.stop() except Exception as exc: logger.warning("Complaint scheduler stop failed: %s", exc) + policy_pin_scheduler = getattr(app.state, "policy_pin_scheduler", None) + if policy_pin_scheduler is not None: + try: + await policy_pin_scheduler.stop() + except Exception as exc: + logger.warning("Policy pin scheduler stop failed: %s", exc) try: await ComplaintEmailPdfService.stop_playwright_browser() except Exception as exc: diff --git a/app/models/CardnewsImageS3.py b/app/models/CardnewsImageS3.py new file mode 100644 index 0000000..dea4764 --- /dev/null +++ b/app/models/CardnewsImageS3.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from sqlalchemy import BigInteger, ForeignKey, Identity, String +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.core.database import Base +from app.models.Community import Community + + +class CardnewsImageS3(Base): + __tablename__ = "cardnews_image_s3" + + cardnews_image_s3_id: Mapped[int] = mapped_column( + "cardnews_image_s3_id", + BigInteger, + Identity(), + primary_key=True, + ) + community_id: Mapped[int] = mapped_column( + "community_id", + BigInteger, + ForeignKey("community.community_id"), + nullable=False, + ) + community: Mapped[Community] = relationship( + "Community", + foreign_keys=[community_id], + back_populates="cardnews_images", + lazy="selectin", + ) + cardnews_image_s3_key: Mapped[str] = mapped_column( + "cardnews_image_s3_key", + String(500), + nullable=False, + ) + cardnews_image_s3_url: Mapped[str] = mapped_column( + "cardnews_image_s3_url", + String(500), + nullable=False, + ) diff --git a/app/models/Community.py b/app/models/Community.py index e77af6e..29157a4 100644 --- a/app/models/Community.py +++ b/app/models/Community.py @@ -1,7 +1,7 @@ from __future__ import annotations -from sqlalchemy import BigInteger, ForeignKey, Identity -from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy import BigInteger, Float, ForeignKey, Identity, String +from sqlalchemy.orm import Mapped, mapped_column, relationship from app.models.BaseEntity import BaseEntity @@ -15,9 +15,16 @@ class Community(BaseEntity): Identity(), primary_key=True, ) - pin_id: Mapped[int | None] = mapped_column( + pin_id: Mapped[int] = mapped_column( "pin_id", BigInteger, ForeignKey("pin.pin_id"), - nullable=True, + nullable=False, + ) + community_type: Mapped[str] = mapped_column("community_type", String(50), nullable=False) + popularity: Mapped[float] = mapped_column("popularity", Float, nullable=False, default=0.0) + cardnews_images: Mapped[list["CardnewsImageS3"]] = relationship( + "CardnewsImageS3", + back_populates="community", + lazy="selectin", ) diff --git a/app/models/EventPin.py b/app/models/EventPin.py index 313e29d..f807f46 100644 --- a/app/models/EventPin.py +++ b/app/models/EventPin.py @@ -31,12 +31,17 @@ class EventPin(BaseEntity): back_populates="event_pin", lazy="selectin", ) - festival_api_id: Mapped[int] = mapped_column( + festival_api_id: Mapped[int | None] = mapped_column( "festival_api_id", BigInteger, unique=True, nullable=True, ) + policy_api_id: Mapped[int | None] = mapped_column( + "policy_api_id", + BigInteger, + nullable=True, + ) event_start_time: Mapped[datetime] = mapped_column("event_start_time", DateTime, nullable=False) event_end_time: Mapped[datetime] = mapped_column("event_end_time", DateTime, nullable=False) discount: Mapped[str | None] = mapped_column("discount", String(255), nullable=True) diff --git a/app/models/__init__.py b/app/models/__init__.py index d80d2fd..05dfb18 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -1,3 +1,5 @@ +from app.models.CardnewsImageS3 import CardnewsImageS3 +from app.models.Community import Community from app.models.ComplaintPetition import ComplaintPetition from app.models.Department import Department from app.models.DbConnectionCheck import DbConnectionCheck @@ -24,6 +26,8 @@ from app.models.enum.UserRole import UserRole __all__ = [ + "CardnewsImageS3", + "Community", "DbConnectionCheck", "ComplaintPetition", "Department", diff --git a/app/policy_cardnews/template/dispatch.py b/app/policy_cardnews/template/dispatch.py index 0e12333..b9d5e48 100644 --- a/app/policy_cardnews/template/dispatch.py +++ b/app/policy_cardnews/template/dispatch.py @@ -56,12 +56,17 @@ MASCOT_LAYOUTS = {LAYOUT_COVER, LAYOUT_CTA, LAYOUT_THREE_COL, LAYOUT_NUMBERED} _REPO_ROOT = Path(__file__).resolve().parents[3] +# JS import처럼 app/policy_cardnews 패키지 기준 상대 경로 해석 +_POLICY_CARDNEWS_DIR = Path(__file__).resolve().parents[1] def _font_dir() -> Path: from app.core.config import settings - return Path(settings.policy_cardnews_font_dir) + raw = Path(settings.policy_cardnews_font_dir) + if raw.is_absolute(): + return raw + return (_POLICY_CARDNEWS_DIR / raw).resolve() @dataclass(frozen=True) @@ -244,6 +249,10 @@ def _load_font(size: int, *, bold: bool = False, extra_bold: bool = False) -> Im path = base_dir / name if path.is_file(): return ImageFont.truetype(str(path), size=size) + logger.warning( + "Pretendard 폰트를 찾지 못해 기본 폰트로 렌더합니다 (font_dir=%s)", + base_dir, + ) return ImageFont.load_default() diff --git a/app/repositories/CardnewsImageS3Repo.py b/app/repositories/CardnewsImageS3Repo.py new file mode 100644 index 0000000..4eab67d --- /dev/null +++ b/app/repositories/CardnewsImageS3Repo.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.CardnewsImageS3 import CardnewsImageS3 +from app.repositories.BaseRepo import BaseRepo + + +class CardnewsImageS3Repo(BaseRepo[CardnewsImageS3]): + def __init__(self, session: AsyncSession) -> None: + super().__init__(session, CardnewsImageS3) + + async def list_by_community_id(self, community_id: int) -> list[CardnewsImageS3]: + result = await self.session.execute( + select(CardnewsImageS3) + .where(CardnewsImageS3.community_id == community_id) + .order_by(CardnewsImageS3.cardnews_image_s3_id.asc()), + ) + return list(result.scalars().all()) + + async def delete_by_community_id(self, community_id: int) -> int: + result = await self.session.execute( + delete(CardnewsImageS3).where(CardnewsImageS3.community_id == community_id), + ) + await self.session.flush() + return int(result.rowcount or 0) diff --git a/app/repositories/CommunityRepo.py b/app/repositories/CommunityRepo.py index 217de87..20d1897 100644 --- a/app/repositories/CommunityRepo.py +++ b/app/repositories/CommunityRepo.py @@ -19,3 +19,9 @@ async def get_community_id_by_pin_id(self, pin_id: int) -> int | None: ) value = result.scalar_one_or_none() return int(value) if value is not None else None + + async def get_by_pin_id(self, pin_id: int) -> Community | None: + result = await self.session.execute( + select(Community).where(Community.pin_id == pin_id).limit(1), + ) + return result.scalar_one_or_none() diff --git a/app/repositories/EventPinRepo.py b/app/repositories/EventPinRepo.py index 8f13ddb..4aa165d 100644 --- a/app/repositories/EventPinRepo.py +++ b/app/repositories/EventPinRepo.py @@ -39,3 +39,25 @@ async def count_festival_pins(self) -> int: ), ) return result.scalar_one() + + async def get_by_policy_api_id(self, policy_api_id: int) -> EventPin | None: + result = await self.session.execute( + select(EventPin) + .where(EventPin.policy_api_id == policy_api_id) + .options(*_EVENT_PIN_LOAD_OPTIONS), + ) + return result.scalar_one_or_none() + + async def list_policy_api_ids(self) -> set[int]: + result = await self.session.execute( + select(EventPin.policy_api_id).where(EventPin.policy_api_id.is_not(None)), + ) + return {int(row) for row in result.scalars().all() if row is not None} + + async def count_policy_pins(self) -> int: + result = await self.session.execute( + select(func.count(EventPin.event_pin_id)).where( + EventPin.policy_api_id.is_not(None), + ), + ) + return result.scalar_one() diff --git a/app/repositories/UserRepo.py b/app/repositories/UserRepo.py index b1418a0..c46d651 100644 --- a/app/repositories/UserRepo.py +++ b/app/repositories/UserRepo.py @@ -21,5 +21,23 @@ async def get_by_email(self, email: str) -> User | None: result = await self.session.execute(select(User).where(User.email == email)) return result.scalar_one_or_none() + async def get_by_nickname(self, nickname: str) -> User | None: + normalized = (nickname or "").strip() + if not normalized: + return None + result = await self.session.execute( + select(User).where(User.nickname == normalized).limit(1), + ) + return result.scalar_one_or_none() + + async def get_by_user_name(self, user_name: str) -> User | None: + normalized = (user_name or "").strip() + if not normalized: + return None + result = await self.session.execute( + select(User).where(User.user_name == normalized).limit(1), + ) + return result.scalar_one_or_none() + async def get_all_users(self) -> list[User]: return await self.get_all() diff --git a/app/routes/PolicyAdminRoute.py b/app/routes/PolicyAdminRoute.py new file mode 100644 index 0000000..1682249 --- /dev/null +++ b/app/routes/PolicyAdminRoute.py @@ -0,0 +1,160 @@ +from __future__ import annotations + +from fastapi import APIRouter, HTTPException, Query + +from app.core.codes import ErrorCode, SuccessCode +from app.core.config import settings +from app.core.deps import ( + PolicyEventIngestServiceDep, + PolicyPinSchedulerDep, + PolicyPinServiceDep, + S3UtilDep, +) +from app.core.responses import SuccessEnvelope, success_response +from app.schemas.PolicyAdminDTO import ( + PolicyImportBatchResult, + PolicyPipelineStatusResult, + PolicySyncResult, + PolicyTransformBatchResult, +) + +router = APIRouter(prefix="/policy-admin", tags=["policy-admin"]) + +_MAX_BATCH = 25 + + +def _runtime_error_response(exc: RuntimeError) -> HTTPException: + detail = str(exc) + if "GEMINI" in detail or "POLICY" in detail or "S3" in detail: + status = ErrorCode.VLM_NOT_CONFIGURED.http_status + else: + status = ErrorCode.INTERNAL_SERVER_ERROR.http_status + return HTTPException(status_code=status, detail=detail) + + +def _clamp_batch(value: int | None) -> int: + if value is None: + return settings.policy_sync_batch_size + return min(max(value, 1), _MAX_BATCH) + + +@router.post( + "/sync", + response_model=SuccessEnvelope[PolicySyncResult], + summary="정책 핀 3일 lookback 수집 → 배치 transform → 배치 DB 적재", + description=( + "오늘 포함 최근 POLICY_SYNC_LOOKBACK_DAYS(기본 3일) 구간을 수집하고, " + "신규 NewsItemId만 POLICY_SYNC_BATCH_SIZE 단위로 LLM 가공·DB INSERT합니다." + ), +) +async def sync_policy_pins( + service: PolicyPinServiceDep, + ingest_service: PolicyEventIngestServiceDep, + s3_util: S3UtilDep, + start_date: str | None = Query(default=None, min_length=8, max_length=8, description="YYYYMMDD"), + end_date: str | None = Query(default=None, min_length=8, max_length=8, description="YYYYMMDD"), + transform_limit: int | None = Query(default=None, ge=1, le=100, description="가공 최대 건수"), + batch_size: int | None = Query(default=None, ge=1, le=25, description="배치 크기"), +) -> SuccessEnvelope[PolicySyncResult]: + try: + body = await service.sync_pipeline( + ingest_service=ingest_service, + s3_util=s3_util, + start_date=start_date, + end_date=end_date, + transform_limit=transform_limit, + batch_size=_clamp_batch(batch_size), + ) + except FileNotFoundError as exc: + raise HTTPException(status_code=ErrorCode.NOT_FOUND.http_status, detail=str(exc)) from exc + except RuntimeError as exc: + raise _runtime_error_response(exc) from exc + + return success_response(result=body, success_code=SuccessCode.CREATED) + + +@router.post( + "/transform-batch", + response_model=SuccessEnvelope[PolicyTransformBatchResult], + summary="원문 JSONL 배치 가공 (신규 건만)", +) +async def transform_policy_batch( + service: PolicyPinServiceDep, + ingest_service: PolicyEventIngestServiceDep, + s3_util: S3UtilDep, + batch_size: int | None = Query(default=None, ge=1, le=25, description="이번 배치 가공 건수"), +) -> SuccessEnvelope[PolicyTransformBatchResult]: + try: + db_ids = await ingest_service._event_pin_repo.list_policy_api_ids() + body = await service.transform_batch( + batch_size=_clamp_batch(batch_size), + s3_util=s3_util, + db_policy_api_ids=db_ids, + ) + except FileNotFoundError as exc: + raise HTTPException(status_code=ErrorCode.NOT_FOUND.http_status, detail=str(exc)) from exc + except RuntimeError as exc: + raise _runtime_error_response(exc) from exc + + return success_response(result=body, success_code=SuccessCode.CREATED) + + +@router.post( + "/import-batch", + response_model=SuccessEnvelope[PolicyImportBatchResult], + summary="handoff JSONL 배치 DB 적재", +) +async def import_policy_batch( + ingest_service: PolicyEventIngestServiceDep, + batch_size: int | None = Query(default=None, ge=1, le=25, description="이번 배치 INSERT 건수"), +) -> SuccessEnvelope[PolicyImportBatchResult]: + try: + body = await ingest_service.import_handoff_batch( + import_all=False, + limit=_clamp_batch(batch_size), + ) + except FileNotFoundError as exc: + raise HTTPException(status_code=ErrorCode.NOT_FOUND.http_status, detail=str(exc)) from exc + except RuntimeError as exc: + raise _runtime_error_response(exc) from exc + + return success_response(result=body, success_code=SuccessCode.CREATED) + + +@router.get( + "/status", + response_model=SuccessEnvelope[PolicyPipelineStatusResult], + summary="정책 핀 파이프라인 상태", +) +async def policy_pipeline_status( + ingest_service: PolicyEventIngestServiceDep, +) -> SuccessEnvelope[PolicyPipelineStatusResult]: + body = await ingest_service.get_pipeline_status() + return success_response(result=body, success_code=SuccessCode.OK) + + +@router.post( + "/scheduler/run-once", + response_model=SuccessEnvelope[PolicySyncResult], + summary="정책 핀 스케줄러 즉시 1회 실행 (테스트용, 간격 무시)", +) +async def run_policy_scheduler_once( + service: PolicyPinServiceDep, + ingest_service: PolicyEventIngestServiceDep, + s3_util: S3UtilDep, + scheduler: PolicyPinSchedulerDep, +) -> SuccessEnvelope[PolicySyncResult]: + try: + if scheduler is not None: + body = await scheduler.run_once_now(force=True) + else: + body = await service.sync_pipeline( + ingest_service=ingest_service, + s3_util=s3_util, + ) + except FileNotFoundError as exc: + raise HTTPException(status_code=ErrorCode.NOT_FOUND.http_status, detail=str(exc)) from exc + except RuntimeError as exc: + raise _runtime_error_response(exc) from exc + + return success_response(result=body, success_code=SuccessCode.CREATED) diff --git a/app/routes/PolicyPinRoute.py b/app/routes/PolicyPinRoute.py index 3c3877c..bd4095e 100644 --- a/app/routes/PolicyPinRoute.py +++ b/app/routes/PolicyPinRoute.py @@ -3,7 +3,7 @@ from fastapi import APIRouter, HTTPException, Query from app.core.codes import ErrorCode, SuccessCode -from app.core.deps import PolicyPinServiceDep +from app.core.deps import PolicyPinServiceDep, S3UtilDep from app.core.responses import SuccessEnvelope, success_response from app.schemas.PolicyPinDTO import ( PolicyPinHandoffResult, @@ -76,6 +76,7 @@ async def search_policy_news( ) async def transform_policy_documents( service: PolicyPinServiceDep, + s3_util: S3UtilDep, limit: int | None = Query( default=None, ge=1, @@ -84,7 +85,7 @@ async def transform_policy_documents( ), ) -> SuccessEnvelope[PolicyPinTransformResult]: try: - body = await service.transform_and_save(limit=limit) + body = await service.transform_and_save(limit=limit, s3_util=s3_util) except FileNotFoundError as exc: raise HTTPException( status_code=ErrorCode.NOT_FOUND.http_status, diff --git a/app/routes/__init__.py b/app/routes/__init__.py index 12bed54..12d94d0 100644 --- a/app/routes/__init__.py +++ b/app/routes/__init__.py @@ -6,6 +6,7 @@ from app.routes.TestRoute import router as test_router from app.routes.VectorTestRoute import router as vector_test_router from app.routes.FestivalPinRoute import router as festival_pin_router +from app.routes.PolicyAdminRoute import router as policy_admin_router from app.routes.PolicyPinRoute import router as policy_pin_router from app.routes.FestivalAdminRoute import router as festival_admin_router from app.core.config import settings @@ -20,6 +21,7 @@ {"router": vector_test_router, "disabled_envs": {"dev", "prod"}}, {"router": festival_pin_router, "disabled_envs": {"prod"}}, {"router": policy_pin_router, "disabled_envs": {"prod"}}, + {"router": policy_admin_router, "disabled_envs": {"prod"}}, {"router": festival_admin_router, "disabled_envs": set()}, ) diff --git a/app/schemas/PolicyAdminDTO.py b/app/schemas/PolicyAdminDTO.py new file mode 100644 index 0000000..aeea162 --- /dev/null +++ b/app/schemas/PolicyAdminDTO.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from enum import Enum +from typing import Any + +from pydantic import BaseModel, Field + +from app.schemas.PolicyPinDTO import PolicyPinHandoffDTO, PolicyPinSearchResult, PolicyPinTransformResult + + +class PolicyBatchAction(str, Enum): + CREATED = "created" + SKIPPED = "skipped" + ERROR = "error" + + +class PolicyBatchItemResult(BaseModel): + policy_api_id: int | None = None + pin_title: str | None = None + action: PolicyBatchAction + message: str | None = None + + +class PolicyImportBatchResult(BaseModel): + inserted_count: int + skipped_duplicate_count: int + pending_import_count: int + error_count: int + errors: list[dict[str, Any]] = Field(default_factory=list) + items: list[PolicyBatchItemResult] = Field(default_factory=list) + pin_ids: list[int] = Field(default_factory=list) + requested_batch_size: int | None = None + + +class PolicyTransformBatchResult(PolicyPinTransformResult): + requested_batch_size: int | None = None + batches_run: int = 1 + + +class PolicyPipelineStatusResult(BaseModel): + query_start_date: str | None = None + query_end_date: str | None = None + last_sync_at: str | None = None + documents_count: int + handoff_count: int + db_policy_count: int + pending_transform_count: int + pending_import_count: int + + +class PolicySyncResult(BaseModel): + query_start_date: str + query_end_date: str + search: PolicyPinSearchResult + transform: PolicyPinTransformResult + import_result: PolicyImportBatchResult + hint: str | None = None diff --git a/app/schemas/PolicyPinDTO.py b/app/schemas/PolicyPinDTO.py index 8c4b4a5..6568388 100644 --- a/app/schemas/PolicyPinDTO.py +++ b/app/schemas/PolicyPinDTO.py @@ -88,6 +88,8 @@ class PolicyPinTransformResult(BaseModel): description="output_path JSONL에 저장된 행과 동일 (DB 전달 4필드)", ) hint: str | None = None + skipped_duplicate_count: int = 0 + pending_count: int = 0 class PolicyPinHandoffResult(BaseModel): diff --git a/app/services/PolicyEventIngestService.py b/app/services/PolicyEventIngestService.py new file mode 100644 index 0000000..e3d7621 --- /dev/null +++ b/app/services/PolicyEventIngestService.py @@ -0,0 +1,344 @@ +from __future__ import annotations + +import json +import logging +from datetime import datetime, time +from pathlib import Path +from typing import Any +from zoneinfo import ZoneInfo + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.config import settings +from app.models.CardnewsImageS3 import CardnewsImageS3 +from app.models.Community import Community +from app.models.EventPin import EventPin +from app.models.Pin import Pin +from app.models.enum.PinType import PinType +from app.models.enum.ToneType import ToneType +from app.repositories.CardnewsImageS3Repo import CardnewsImageS3Repo +from app.repositories.CommunityRepo import CommunityRepo +from app.repositories.EventPinRepo import EventPinRepo +from app.repositories.PinRepo import PinRepo +from app.repositories.UserRepo import UserRepo +from app.schemas.PolicyAdminDTO import ( + PolicyBatchAction, + PolicyBatchItemResult, + PolicyImportBatchResult, + PolicyPipelineStatusResult, +) +from app.services.policy_pipeline_cleanup import cleanup_after_policy_import +from app.services.policy_pin_transform import ( + POLICY_DOCUMENTS_PATH, + POLICY_HANDOFF_PATH, + POLICY_SYNC_META_PATH, + count_pending_transform, + load_jsonl_rows, + load_rows_by_content_id, + parse_policy_api_id, +) +from app.utils.policy_news_parse import approve_date_to_yyyymmdd, parse_policy_datetime + +logger = logging.getLogger(__name__) + +_KST = ZoneInfo("Asia/Seoul") +_IMPORT_REPORT_PATH = POLICY_HANDOFF_PATH.with_name("policy_import_batch_report.json") +_COMMUNITY_TYPE_POLICY = "POLICY" + + +def _discard_session_state_after_nested_failure(session: AsyncSession) -> None: + for obj in list(session.new): + session.expunge(obj) + for obj in list(session.dirty): + session.expire(obj) + + +def _parse_event_datetime(value: Any) -> datetime: + if value is None: + raise ValueError("event datetime 없음") + if isinstance(value, datetime): + return value.replace(tzinfo=None) + text = str(value).strip() + if not text: + raise ValueError("event datetime 비어 있음") + if len(text) == 8 and text.isdigit(): + year, month, day = int(text[:4]), int(text[4:6]), int(text[6:8]) + return datetime(year, month, day) + parsed = parse_policy_datetime(text) + if parsed is not None: + return parsed.replace(tzinfo=None) + raise ValueError(f"event datetime 파싱 실패: {value!r}") + + +def _event_window_from_row(row: dict[str, Any]) -> tuple[datetime, datetime]: + start_raw = row.get("event_start_time") or approve_date_to_yyyymmdd(row.get("approve_date")) + end_raw = row.get("event_end_time") or start_raw + start_dt = _parse_event_datetime(start_raw) + end_dt = _parse_event_datetime(end_raw) + if end_dt < start_dt: + end_dt = start_dt + start_at = datetime.combine(start_dt.date(), time.min) + end_at = datetime.combine(end_dt.date(), time.max.replace(microsecond=0)) + return start_at, end_at + + +class PolicyEventIngestService: + def __init__( + self, + *, + pin_repo: PinRepo, + event_pin_repo: EventPinRepo, + community_repo: CommunityRepo, + cardnews_image_s3_repo: CardnewsImageS3Repo, + user_repo: UserRepo, + ) -> None: + self._pin_repo = pin_repo + self._event_pin_repo = event_pin_repo + self._community_repo = community_repo + self._cardnews_image_s3_repo = cardnews_image_s3_repo + self._user_repo = user_repo + + async def commit(self) -> None: + await self._pin_repo.commit() + + async def rollback(self) -> None: + await self._pin_repo.rollback() + + async def resolve_admin_uid(self) -> str: + user_name = settings.policy_admin_user_name.strip() + user = await self._user_repo.get_by_user_name(user_name) + if user is None: + raise RuntimeError( + f"정책 핀 등록용 사용자를 찾을 수 없습니다 (user_name={user_name!r}).", + ) + return str(user.uid) + + async def import_handoff_batch( + self, + *, + admin_uid: str | None = None, + import_all: bool = True, + limit: int | None = None, + ) -> PolicyImportBatchResult: + handoff_rows = load_jsonl_rows(POLICY_HANDOFF_PATH) + if not handoff_rows: + raise FileNotFoundError( + f"핸드오프 JSONL 없음: {POLICY_HANDOFF_PATH}. transform을 먼저 실행하세요.", + ) + + uid = admin_uid or await self.resolve_admin_uid() + db_ids = await self._event_pin_repo.list_policy_api_ids() + inserted_count = 0 + skipped_duplicate_count = 0 + errors: list[dict[str, Any]] = [] + items: list[PolicyBatchItemResult] = [] + pin_ids: list[int] = [] + prune_policy_api_ids: set[int] = set() + + pending_rows: list[dict[str, Any]] = [] + for row in handoff_rows: + policy_api_id = parse_policy_api_id(row) + if policy_api_id is None: + continue + if policy_api_id in db_ids: + skipped_duplicate_count += 1 + prune_policy_api_ids.add(policy_api_id) + items.append( + PolicyBatchItemResult( + policy_api_id=policy_api_id, + pin_title=str(row.get("title") or row.get("pin_title") or ""), + action=PolicyBatchAction.SKIPPED, + message="DB에 이미 존재", + ), + ) + else: + pending_rows.append(row) + + target_rows = pending_rows if import_all else pending_rows[: (limit or len(pending_rows))] + if limit is not None and import_all: + target_rows = pending_rows[:limit] + + for row in target_rows: + policy_api_id = parse_policy_api_id(row) + if policy_api_id is None: + errors.append({"row": row, "error": "policy_api_id 없음"}) + continue + existing = await self._event_pin_repo.get_by_policy_api_id(policy_api_id) + if existing is not None: + skipped_duplicate_count += 1 + prune_policy_api_ids.add(policy_api_id) + continue + try: + async with self._pin_repo.session.begin_nested(): + pin_id = await self._insert_policy_pin(admin_uid=uid, row=row, policy_api_id=policy_api_id) + inserted_count += 1 + db_ids.add(policy_api_id) + prune_policy_api_ids.add(policy_api_id) + pin_ids.append(pin_id) + items.append( + PolicyBatchItemResult( + policy_api_id=policy_api_id, + pin_title=str(row.get("title") or row.get("pin_title") or ""), + action=PolicyBatchAction.CREATED, + ), + ) + except Exception as exc: + _discard_session_state_after_nested_failure(self._pin_repo.session) + logger.exception("policy import failed policy_api_id=%s", policy_api_id) + errors.append( + { + "policy_api_id": policy_api_id, + "pin_title": row.get("title") or row.get("pin_title"), + "error": str(exc), + }, + ) + items.append( + PolicyBatchItemResult( + policy_api_id=policy_api_id, + pin_title=str(row.get("title") or row.get("pin_title") or ""), + action=PolicyBatchAction.ERROR, + message=str(exc), + ), + ) + + await self._pin_repo.commit() + + if settings.policy_prune_pipeline_after_import and prune_policy_api_ids: + cleanup_after_policy_import(prune_policy_api_ids) + + pending_import = 0 + for row in handoff_rows: + policy_api_id = parse_policy_api_id(row) + if policy_api_id is not None and policy_api_id not in db_ids: + pending_import += 1 + + report = { + "inserted_count": inserted_count, + "skipped_duplicate_count": skipped_duplicate_count, + "errors": errors, + } + _IMPORT_REPORT_PATH.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + + effective_batch = limit if not import_all else limit + return PolicyImportBatchResult( + inserted_count=inserted_count, + skipped_duplicate_count=skipped_duplicate_count, + pending_import_count=pending_import, + error_count=len(errors), + errors=errors, + items=items, + pin_ids=pin_ids, + requested_batch_size=effective_batch, + ) + + async def _insert_policy_pin( + self, + *, + admin_uid: str, + row: dict[str, Any], + policy_api_id: int, + ) -> int: + title = str(row.get("title") or row.get("pin_title") or "").strip()[:100] + content = str(row.get("pin_content") or "").strip() + if not title or not content: + raise ValueError("title 또는 pin_content가 비어 있음") + + start_at, end_at = _event_window_from_row(row) + + pin = Pin( + uid=admin_uid, + pin_type=PinType.POLICY, + pin_title=title, + pin_content=content, + tone_type=ToneType.NONE, + like_count=0, + view_count=0, + ) + await self._pin_repo.save(pin, flush_immediately=True) + + event_pin = EventPin( + pin_id=pin.pin_id, + festival_api_id=None, + policy_api_id=policy_api_id, + event_start_time=start_at, + event_end_time=end_at, + discount=None, + ) + await self._event_pin_repo.save(event_pin, flush_immediately=True) + + community = Community( + pin_id=pin.pin_id, + community_type=_COMMUNITY_TYPE_POLICY, + popularity=0.0, + ) + await self._community_repo.save(community, flush_immediately=True) + + cardnews_images = row.get("cardnews_images") or [] + if not cardnews_images: + for url in row.get("cardnews_image_urls") or []: + url_text = str(url).strip() + if url_text: + cardnews_images.append({"key": "", "url": url_text}) + + for image in cardnews_images: + if not isinstance(image, dict): + continue + key = str(image.get("key") or "").strip() + url = str(image.get("url") or "").strip() + if not key or not url: + continue + entity = CardnewsImageS3( + community_id=community.community_id, + cardnews_image_s3_key=key, + cardnews_image_s3_url=url, + ) + await self._cardnews_image_s3_repo.save(entity, flush_immediately=True) + + return int(pin.pin_id) + + async def get_pipeline_status(self) -> PolicyPipelineStatusResult: + meta = self._load_sync_meta() + documents = load_jsonl_rows(POLICY_DOCUMENTS_PATH) + handoff_by_id = load_rows_by_content_id(POLICY_HANDOFF_PATH) + db_ids = await self._event_pin_repo.list_policy_api_ids() + + pending_transform = count_pending_transform(documents, handoff_by_id, db_policy_api_ids=db_ids) + pending_import = 0 + for row in handoff_by_id.values(): + policy_api_id = parse_policy_api_id(row) + if policy_api_id is not None and policy_api_id not in db_ids: + pending_import += 1 + + return PolicyPipelineStatusResult( + query_start_date=meta.get("query_start_date"), + query_end_date=meta.get("query_end_date"), + last_sync_at=meta.get("last_sync_at"), + documents_count=len(documents), + handoff_count=len(handoff_by_id), + db_policy_count=len(db_ids), + pending_transform_count=pending_transform, + pending_import_count=pending_import, + ) + + @staticmethod + def _load_sync_meta() -> dict[str, Any]: + if not POLICY_SYNC_META_PATH.is_file(): + return {} + try: + return json.loads(POLICY_SYNC_META_PATH.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return {} + + @staticmethod + def write_sync_meta( + *, + query_start_date: str, + query_end_date: str, + ) -> None: + meta = { + "last_sync_at": datetime.now(_KST).isoformat(timespec="seconds"), + "query_start_date": query_start_date, + "query_end_date": query_end_date, + } + POLICY_SYNC_META_PATH.parent.mkdir(parents=True, exist_ok=True) + POLICY_SYNC_META_PATH.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8") diff --git a/app/services/PolicyPinService.py b/app/services/PolicyPinService.py index d529148..0b428e6 100644 --- a/app/services/PolicyPinService.py +++ b/app/services/PolicyPinService.py @@ -8,6 +8,7 @@ from app.clients.PolicyNewsClient import PolicyNewsClient from app.core.config import settings +from app.schemas.PolicyAdminDTO import PolicyImportBatchResult, PolicySyncResult, PolicyTransformBatchResult from app.schemas.PolicyPinDTO import ( PolicyPinHandoffDTO, PolicyPinHandoffResult, @@ -15,11 +16,16 @@ PolicyPinSourceDTO, PolicyPinTransformResult, ) +from app.services.PolicyEventIngestService import PolicyEventIngestService +from app.services.policy_pipeline_cleanup import prune_pipeline_imported from app.services.policy_pin_transform import ( POLICY_DOCUMENTS_PATH, POLICY_HANDOFF_PATH, + load_rows_by_content_id, + merge_documents, transform_documents_jsonl, ) +from app.utils.S3Util import S3Util from rag.scripts.chunk_module import write_jsonl from rag.scripts.fetch_policy_news import fetch_policy_documents @@ -27,7 +33,6 @@ async def _enrich_cover_image_urls(documents: list[dict[str, Any]]) -> list[dict[str, Any]]: - # OpenAPI에 이미지 URL이 없을 때 원문 페이지에서 표지용 URL을 보충 from app.utils.policy_news_parse import enrich_cover_image_urls as resolve_cover_image_urls async def enrich_one(doc: dict[str, Any]) -> dict[str, Any]: @@ -58,6 +63,7 @@ async def search_and_save( start_date: str, end_date: str, limit: int | None = 10, + merge: bool = True, ) -> PolicyPinSearchResult: fetch_limit: int | None if limit is None: @@ -76,10 +82,15 @@ async def search_and_save( documents = await _enrich_cover_image_urls(documents) path = self.documents_path() - if documents: + path.parent.mkdir(parents=True, exist_ok=True) + if merge: + existing = load_rows_by_content_id(path) + merged = merge_documents(existing, documents) + write_jsonl(path, merged) + total_in_file = len(merged) + else: write_jsonl(path, documents) - elif not path.is_file(): - write_jsonl(path, []) + total_in_file = len(documents) pins = [PolicyPinSourceDTO.model_validate(doc) for doc in documents] hint: str | None @@ -87,13 +98,12 @@ async def search_and_save( hint = ( f"수집 0건 (API 호출 {stats.get('chunks', 0)}회). " "정책뉴스 API는 승인일 기준이며 미래 날짜 구간은 비어 있습니다. " - "최근 3일로 다시 시도하세요 (예: start_date=20260522, end_date=20260524). " - "기존 policy_documents.jsonl은 유지됩니다." + f"JSONL 총 {total_in_file}건 유지." ) else: hint = ( - f"{len(pins)}건을 {path.name}에 저장했습니다. " - "다음: POST /policy-pins/transform" + f"이번 수집 {len(pins)}건, {path.name} 총 {total_in_file}건. " + "다음: POST /policy-pins/transform 또는 /policy-admin/sync" ) return PolicyPinSearchResult( query_start_date=start_date, @@ -110,11 +120,157 @@ async def transform_and_save( *, limit: int | None = None, model: str | None = None, + s3_util: S3Util | None = None, + db_policy_api_ids: set[int] | None = None, ) -> PolicyPinTransformResult: - result = await transform_documents_jsonl(limit=limit, model=model) - if result.processed_count > 0: - return result - return result + return await transform_documents_jsonl( + limit=limit, + model=model, + s3_util=s3_util, + db_policy_api_ids=db_policy_api_ids, + merge_handoff=True, + ) + + async def transform_batch( + self, + *, + batch_size: int | None = None, + model: str | None = None, + s3_util: S3Util | None = None, + db_policy_api_ids: set[int] | None = None, + ) -> PolicyTransformBatchResult: + size = batch_size or settings.policy_sync_batch_size + result = await self.transform_and_save( + limit=size, + model=model, + s3_util=s3_util, + db_policy_api_ids=db_policy_api_ids, + ) + return PolicyTransformBatchResult( + **result.model_dump(), + requested_batch_size=size, + batches_run=1, + ) + + async def sync_pipeline( + self, + *, + ingest_service: PolicyEventIngestService, + s3_util: S3Util, + start_date: str | None = None, + end_date: str | None = None, + transform_limit: int | None = None, + batch_size: int | None = None, + ) -> PolicySyncResult: + if start_date is None or end_date is None: + start_date, end_date = self.default_date_range() + + effective_batch = batch_size or settings.policy_sync_batch_size + + if settings.policy_prune_pipeline_after_import: + db_ids_before = await ingest_service._event_pin_repo.list_policy_api_ids() + prune_pipeline_imported(db_ids_before) + + search = await self.search_and_save( + start_date=start_date, + end_date=end_date, + limit=None, + merge=settings.policy_sync_merge_documents, + ) + + total_processed = 0 + total_imported = 0 + total_skipped_import = 0 + total_skipped_transform = 0 + transform_errors: list[dict] = [] + transform_pins: list = [] + batches_run = 0 + last_transform_pending = 0 + last_import = PolicyImportBatchResult( + inserted_count=0, + skipped_duplicate_count=0, + pending_import_count=0, + error_count=0, + ) + + while True: + if transform_limit is not None and total_processed >= transform_limit: + break + + batch_limit = effective_batch + if transform_limit is not None: + batch_limit = min(effective_batch, transform_limit - total_processed) + + db_ids = await ingest_service._event_pin_repo.list_policy_api_ids() + transform_batch = await self.transform_and_save( + limit=batch_limit, + s3_util=s3_util, + db_policy_api_ids=db_ids, + ) + batches_run += 1 + total_processed += transform_batch.processed_count + total_skipped_transform += transform_batch.skipped_duplicate_count + transform_errors.extend(transform_batch.errors) + transform_pins.extend(transform_batch.pins) + last_transform_pending = transform_batch.pending_count + + if transform_batch.processed_count == 0: + break + + import_batch = await ingest_service.import_handoff_batch( + import_all=False, + limit=effective_batch, + ) + total_imported += import_batch.inserted_count + total_skipped_import += import_batch.skipped_duplicate_count + last_import = import_batch + + if transform_batch.processed_count < batch_limit: + break + + aggregated_transform = PolicyTransformBatchResult( + input_path=str(self.documents_path()), + output_path=str(self.handoff_path()), + processed_count=total_processed, + error_count=len(transform_errors), + errors=transform_errors, + pins=transform_pins, + hint=( + f"배치 {batches_run}회, 가공 {total_processed}건 " + f"(배치 크기 {effective_batch}, 동시성 {settings.policy_transform_concurrency})" + ), + skipped_duplicate_count=total_skipped_transform, + pending_count=last_transform_pending, + requested_batch_size=effective_batch, + batches_run=batches_run, + ) + import_result = PolicyImportBatchResult( + inserted_count=total_imported, + skipped_duplicate_count=total_skipped_import, + pending_import_count=last_import.pending_import_count, + error_count=last_import.error_count, + errors=last_import.errors, + items=last_import.items, + pin_ids=last_import.pin_ids, + requested_batch_size=effective_batch, + ) + + PolicyEventIngestService.write_sync_meta( + query_start_date=start_date, + query_end_date=end_date, + ) + hint = ( + f"sync 완료: 수집 {search.count}건, 가공 {total_processed}건({batches_run}배치), " + f"DB INSERT {total_imported}건." + ) + return PolicySyncResult( + query_start_date=start_date, + query_end_date=end_date, + search=search, + transform=aggregated_transform, + import_result=import_result, + hint=hint, + ) def load_from_jsonl( self, diff --git a/app/services/internal/PolicyPinSchedulerService.py b/app/services/internal/PolicyPinSchedulerService.py new file mode 100644 index 0000000..7e78622 --- /dev/null +++ b/app/services/internal/PolicyPinSchedulerService.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import asyncio +import json +import logging +from datetime import datetime, timedelta +from zoneinfo import ZoneInfo + +from app.core.config import settings +from app.core.database import AsyncSessionLocal +from app.repositories.CardnewsImageS3Repo import CardnewsImageS3Repo +from app.repositories.CommunityRepo import CommunityRepo +from app.repositories.EventPinRepo import EventPinRepo +from app.repositories.PinRepo import PinRepo +from app.repositories.UserRepo import UserRepo +from app.schemas.PolicyAdminDTO import PolicySyncResult +from app.services.PolicyEventIngestService import PolicyEventIngestService, POLICY_SYNC_META_PATH +from app.services.PolicyPinService import PolicyPinService +from app.utils.S3Util import S3Util + +logger = logging.getLogger(__name__) +_KST = ZoneInfo("Asia/Seoul") + + +class PolicyPinSchedulerService: + def __init__(self, *, s3_util: S3Util) -> None: + self._s3_util = s3_util + self._task: asyncio.Task[None] | None = None + self._pin_service = PolicyPinService() + + def start(self) -> None: + if self._task is not None and not self._task.done(): + return + self._task = asyncio.create_task(self._run_loop(), name="policy_pin_scheduler") + logger.info("정책 핀 sync 스케줄러 시작") + + async def stop(self) -> None: + if self._task is None: + return + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + finally: + self._task = None + logger.info("정책 핀 sync 스케줄러 종료") + + async def run_once_now(self, *, force: bool = False) -> PolicySyncResult: + if not force and not self._should_run_sync(): + start, end = self._pin_service.default_date_range() + from app.schemas.PolicyAdminDTO import PolicyImportBatchResult + from app.schemas.PolicyPinDTO import PolicyPinSearchResult, PolicyPinTransformResult + + return PolicySyncResult( + query_start_date=start, + query_end_date=end, + search=PolicyPinSearchResult( + query_start_date=start, + query_end_date=end, + count=0, + pins=[], + saved_documents_path=str(self._pin_service.documents_path()), + stats={}, + hint="POLICY_SYNC_INTERVAL_DAYS 미경과로 sync 생략", + ), + transform=PolicyPinTransformResult( + input_path=str(self._pin_service.documents_path()), + output_path=str(self._pin_service.handoff_path()), + processed_count=0, + error_count=0, + pins=[], + hint="sync 생략", + ), + import_result=PolicyImportBatchResult( + inserted_count=0, + skipped_duplicate_count=0, + pending_import_count=0, + error_count=0, + ), + hint="POLICY_SYNC_INTERVAL_DAYS 미경과로 sync 생략", + ) + + async with AsyncSessionLocal() as session: + ingest = PolicyEventIngestService( + pin_repo=PinRepo(session), + event_pin_repo=EventPinRepo(session), + community_repo=CommunityRepo(session), + cardnews_image_s3_repo=CardnewsImageS3Repo(session), + user_repo=UserRepo(session), + ) + return await self._pin_service.sync_pipeline( + ingest_service=ingest, + s3_util=self._s3_util, + ) + + def _should_run_sync(self) -> bool: + if not POLICY_SYNC_META_PATH.is_file(): + return True + try: + meta = json.loads(POLICY_SYNC_META_PATH.read_text(encoding="utf-8")) + last_sync_at = str(meta.get("last_sync_at") or "").strip() + if not last_sync_at: + return True + last_run = datetime.fromisoformat(last_sync_at) + if last_run.tzinfo is None: + last_run = last_run.replace(tzinfo=_KST) + elapsed = datetime.now(_KST) - last_run.astimezone(_KST) + return elapsed >= timedelta(days=settings.policy_sync_interval_days) + except (json.JSONDecodeError, ValueError): + return True + + async def _run_loop(self) -> None: + while True: + wait_seconds = self._seconds_until_next_schedule_kst() + logger.info("정책 핀 sync 스케줄러 대기 %.1fs", wait_seconds) + await asyncio.sleep(wait_seconds) + try: + result = await self.run_once_now(force=False) + logger.info( + "정책 핀 sync 실행: 가공 %d건, INSERT %d건", + result.transform.processed_count, + result.import_result.inserted_count, + ) + except Exception: + logger.exception("정책 핀 sync 스케줄 실행 실패") + + @staticmethod + def _seconds_until_next_schedule_kst() -> float: + now = datetime.now(_KST) + hour = settings.policy_sync_schedule_hour_kst + next_run = datetime.combine(now.date(), datetime.min.time(), tzinfo=_KST) + timedelta(hours=hour) + if now >= next_run: + next_run += timedelta(days=1) + delta = next_run - now + return max(delta.total_seconds(), 1.0) diff --git a/app/services/policy_cardnews.py b/app/services/policy_cardnews.py index 4c94883..a48b583 100644 --- a/app/services/policy_cardnews.py +++ b/app/services/policy_cardnews.py @@ -2,7 +2,7 @@ import logging from pathlib import Path -from typing import Any +from typing import Any, TypedDict from app.core.config import settings from app.policy_cardnews import ( @@ -16,24 +16,93 @@ build_policy_cardnews_image_prompt, build_policy_cardnews_slide_prompt, ) +from app.utils.S3Util import S3Util logger = logging.getLogger(__name__) POLICY_CARDNEWS_OUTPUT_DIR = ( Path(__file__).resolve().parents[2] / "rag" / "output" / "policy_cardnews" ) +_REPO_ROOT = Path(__file__).resolve().parents[2] -async def _generate_with_image_model( +class CardnewsS3Image(TypedDict): + key: str + url: str + + +def cleanup_local_cardnews_dir(content_id: str) -> None: + target = POLICY_CARDNEWS_OUTPUT_DIR / str(content_id or "").strip() + if not target.is_dir(): + return + import shutil + + shutil.rmtree(target, ignore_errors=True) + logger.debug("로컬 카드뉴스 디렉터리 삭제: %s", target) + + +def _maybe_cleanup_local_cardnews(content_id: str) -> None: + if not settings.policy_cardnews_keep_local_files: + cleanup_local_cardnews_dir(content_id) + + +def _slide_object_key(content_id: str, slide_no: int) -> str: + prefix = settings.policy_cardnews_s3_prefix.strip("/") + return f"{prefix}/{content_id}/slide_{slide_no:02d}.png" + + +async def _upload_slide_bytes( + s3_util: S3Util, + *, + content_id: str, + slide_no: int, + image_bytes: bytes, +) -> CardnewsS3Image: + key = _slide_object_key(content_id, slide_no) + result = await s3_util.upload_bytes( + image_bytes, + filename=f"slide_{slide_no:02d}.png", + content_type="image/png", + object_key=key, + ) + return {"key": result["key"], "url": result["url"]} + + +async def _upload_local_handoff_path( + s3_util: S3Util, + *, + content_id: str, + handoff_path: str, +) -> CardnewsS3Image: + rel = handoff_path.strip().lstrip("/") + local_path = _REPO_ROOT / rel + if not local_path.is_file(): + raise FileNotFoundError(f"카드뉴스 로컬 파일 없음: {local_path}") + slide_name = local_path.stem + slide_no = 1 + if slide_name.startswith("slide_"): + try: + slide_no = int(slide_name.split("_", 1)[1]) + except ValueError: + slide_no = 1 + image_bytes = local_path.read_bytes() + return await _upload_slide_bytes( + s3_util, + content_id=content_id, + slide_no=slide_no, + image_bytes=image_bytes, + ) + + +async def _generate_slide_bytes_with_image_model( *, image_service: PolicyCardnewsImageService, content_id: str, pin_title: str, minister: str, slides: list[dict[str, Any]], - output_dir: Path, -) -> list[str]: - saved_paths: list[str] = [] +) -> list[tuple[int, bytes]]: + out: list[tuple[int, bytes]] = [] total = len(slides) for index, slide in enumerate(slides, start=1): prompt = build_policy_cardnews_image_prompt( @@ -45,18 +114,11 @@ async def _generate_with_image_model( ) image_bytes = await image_service.generate_slide_image_bytes(prompt=prompt) slide_no = int(slide.get("slide") or index) - saved_paths.append( - save_slide_image_bytes( - contentid=content_id, - slide_no=slide_no, - image_bytes=image_bytes, - output_dir=output_dir, - ) - ) - return saved_paths + out.append((slide_no, image_bytes)) + return out -async def _generate_with_pillow( +async def _generate_local_paths_with_pillow( *, content_id: str, minister: str, @@ -77,19 +139,19 @@ async def _generate_with_pillow( ) -async def generate_cardnews_image_paths( +async def generate_cardnews_s3_images( text_llm: IssuePinLLMService, *, row: dict, easy_read_content: str, + s3_util: S3Util, output_dir: Path | None = None, image_service: PolicyCardnewsImageService | None = None, -) -> list[str]: +) -> list[CardnewsS3Image]: content_id = str(row.get("contentid") or "").strip() if not content_id: return [] - existing = [str(url).strip() for url in (row.get("cardnews_image_urls") or []) if str(url).strip()] prompt = build_policy_cardnews_slide_prompt( pin_title=str(row.get("pin_title") or ""), pin_content=easy_read_content or str(row.get("pin_content") or ""), @@ -116,17 +178,34 @@ async def generate_cardnews_image_paths( first["use_image"] = True slides[0] = first + uploaded: list[CardnewsS3Image] = [] + if settings.policy_cardnews_use_image_model: img_service = image_service or PolicyCardnewsImageService.from_settings() try: - generated = await _generate_with_image_model( + slide_bytes_list = await _generate_slide_bytes_with_image_model( image_service=img_service, content_id=content_id, pin_title=pin_title, minister=minister, slides=slides, - output_dir=target_dir, ) + for slide_no, image_bytes in slide_bytes_list: + if settings.policy_cardnews_keep_local_files: + save_slide_image_bytes( + contentid=content_id, + slide_no=slide_no, + image_bytes=image_bytes, + output_dir=target_dir, + ) + uploaded.append( + await _upload_slide_bytes( + s3_util, + content_id=content_id, + slide_no=slide_no, + image_bytes=image_bytes, + ), + ) except Exception: if not settings.policy_cardnews_pillow_fallback: raise @@ -134,7 +213,7 @@ async def generate_cardnews_image_paths( "카드뉴스 이미지 모델 생성 실패로 Pillow 폴백 contentid=%s", content_id, ) - generated = await _generate_with_pillow( + local_paths = await _generate_local_paths_with_pillow( content_id=content_id, minister=minister, slides=slides, @@ -143,8 +222,16 @@ async def generate_cardnews_image_paths( source_url=source_url, pin_content=easy_read_content or str(row.get("pin_content") or ""), ) + for path in local_paths: + uploaded.append( + await _upload_local_handoff_path( + s3_util, + content_id=content_id, + handoff_path=path, + ), + ) else: - generated = await _generate_with_pillow( + local_paths = await _generate_local_paths_with_pillow( content_id=content_id, minister=minister, slides=slides, @@ -153,5 +240,37 @@ async def generate_cardnews_image_paths( source_url=source_url, pin_content=easy_read_content or str(row.get("pin_content") or ""), ) + for path in local_paths: + uploaded.append( + await _upload_local_handoff_path( + s3_util, + content_id=content_id, + handoff_path=path, + ), + ) + + if not settings.policy_cardnews_keep_local_files: + _maybe_cleanup_local_cardnews(content_id) + return uploaded + - return existing + generated +async def generate_cardnews_image_paths( + text_llm: IssuePinLLMService, + *, + row: dict, + easy_read_content: str, + output_dir: Path | None = None, + image_service: PolicyCardnewsImageService | None = None, + s3_util: S3Util | None = None, +) -> list[str]: + """하위 호환: S3 URL 문자열 목록만 반환.""" + s3 = s3_util or S3Util() + images = await generate_cardnews_s3_images( + text_llm, + row=row, + easy_read_content=easy_read_content, + s3_util=s3, + output_dir=output_dir, + image_service=image_service, + ) + return [img["url"] for img in images if img.get("url")] diff --git a/app/services/policy_pin_transform.py b/app/services/policy_pin_transform.py index b75952e..989f6ec 100644 --- a/app/services/policy_pin_transform.py +++ b/app/services/policy_pin_transform.py @@ -1,13 +1,16 @@ from __future__ import annotations +import asyncio from pathlib import Path +from typing import Any from app.core.config import settings from app.schemas.PolicyPinDTO import PolicyPinHandoffDTO, PolicyPinTransformResult from app.services.internal.ai.IssuePinLLMService import IssuePinLLMService from app.services.internal.ai.gemini_retry import parse_gemini_model_list -from app.services.policy_cardnews import generate_cardnews_image_paths +from app.services.policy_cardnews import CardnewsS3Image, generate_cardnews_s3_images from app.services.prompts.policy_pin import build_policy_easy_read_prompt +from app.utils.S3Util import S3Util from rag.scripts.chunk_module import iter_jsonl, write_jsonl POLICY_DOCUMENTS_PATH = ( @@ -16,48 +19,91 @@ POLICY_HANDOFF_PATH = ( Path(__file__).resolve().parents[2] / "rag" / "output" / "policy_pins_for_db.jsonl" ) +POLICY_SYNC_META_PATH = ( + Path(__file__).resolve().parents[2] / "rag" / "output" / "policy_sync_meta.json" +) + + +def parse_policy_api_id(row: dict[str, Any]) -> int | None: + raw = row.get("policy_api_id") or row.get("contentid") + if raw is None: + return None + text = str(raw).strip() + if not text.isdigit(): + return None + return int(text) + + +def row_content_id(row: dict[str, Any]) -> str: + return str(row.get("contentid") or row.get("policy_api_id") or "").strip() + + +def row_raw_content(row: dict[str, Any]) -> str: + return (row.get("pin_content_raw") or row.get("pin_content") or row.get("text") or "").strip() + +def normalize_raw_for_compare(text: str) -> str: + return " ".join((text or "").split()).strip() -def append_source_link_to_pin_content(body: str, source_url: str) -> str: - """가공 본문 끝에 원문 기사 URL을 붙인다 (중복 방지).""" - text = (body or "").strip() - url = (source_url or "").strip() - if not url or not text: - return text or url - normalized = url.rstrip("/") - for line in text.splitlines(): - line = line.strip() - if line == url or line.rstrip("/") == normalized: - return text - if line.startswith("http") and normalized in line: - return text +def load_jsonl_rows(path: Path) -> list[dict[str, Any]]: + if not path.is_file(): + return [] + return [row for row in iter_jsonl(path) if isinstance(row, dict)] - return f"{text}\n\n원문 기사: {url}" + +def load_rows_by_content_id(path: Path) -> dict[str, dict[str, Any]]: + out: dict[str, dict[str, Any]] = {} + for row in load_jsonl_rows(path): + content_id = row_content_id(row) + if content_id: + out[content_id] = row + return out + + +def merge_documents( + existing: dict[str, dict[str, Any]], + new_rows: list[dict[str, Any]], +) -> list[dict[str, Any]]: + merged = dict(existing) + for row in new_rows: + content_id = row_content_id(row) + if content_id: + merged[content_id] = row + return list(merged.values()) + + +def write_handoff_map(handoff_by_id: dict[str, dict[str, Any]], path: Path | None = None) -> None: + dst = path or POLICY_HANDOFF_PATH + write_jsonl(dst, list(handoff_by_id.values())) def build_handoff_row( - source: dict, + source: dict[str, Any], *, easy_read_content: str, - cardnews_image_urls: list[str] | None = None, -) -> dict: - # DB 전달 JSONL 1행: title, pin_content, cardnews_image_urls, source_url 만 - cardnews_urls = [ - str(url).strip() - for url in ( - cardnews_image_urls - if cardnews_image_urls is not None - else (source.get("cardnews_image_urls") or []) - ) - if str(url).strip() - ] + cardnews_images: list[CardnewsS3Image] | None = None, +) -> dict[str, Any]: + content_id = row_content_id(source) + policy_api_id = parse_policy_api_id(source) + images = cardnews_images or [] + cardnews_urls = [img["url"] for img in images if str(img.get("url") or "").strip()] source_url = (source.get("source_url") or "").strip() + event_start = source.get("event_start_time") + event_end = source.get("event_end_time") + title = (source.get("pin_title") or source.get("title") or "").strip() return { - "title": (source.get("pin_title") or source.get("title") or "").strip(), - "pin_content": append_source_link_to_pin_content(easy_read_content, source_url), + "contentid": content_id, + "policy_api_id": policy_api_id, + "title": title, + "pin_content": (easy_read_content or "").strip(), + "pin_content_raw": row_raw_content(source), "cardnews_image_urls": cardnews_urls, + "cardnews_images": images, "source_url": source_url, + "event_start_time": event_start, + "event_end_time": event_end, + "approve_date": (source.get("approve_date") or "").strip(), } @@ -74,8 +120,13 @@ def build_llm_service(*, model: str | None = None) -> IssuePinLLMService: ) -async def transform_one_row(llm: IssuePinLLMService, row: dict) -> dict: - raw_content = (row.get("pin_content") or row.get("text") or "").strip() +async def transform_one_row( + llm: IssuePinLLMService, + row: dict[str, Any], + *, + s3_util: S3Util, +) -> dict[str, Any]: + raw_content = row_raw_content(row) if not raw_content: raise ValueError("pin_content가 비어 있음") @@ -87,17 +138,53 @@ async def transform_one_row(llm: IssuePinLLMService, row: dict) -> dict: approve_date=str(row.get("approve_date") or ""), ) easy_read_text = await llm.generate_pin_text(prompt=prompt) - cardnews_urls = await generate_cardnews_image_paths( - llm, + cardnews_images = await generate_cardnews_s3_images( + text_llm=llm, row=row, easy_read_content=easy_read_text, + s3_util=s3_util, ) - raw_html = (row.get("pin_content_raw") or raw_content).strip() - source = {**row, "pin_content_raw": raw_html} + source = {**row, "pin_content_raw": raw_content} return build_handoff_row( source, easy_read_content=easy_read_text, - cardnews_image_urls=cardnews_urls, + cardnews_images=cardnews_images, + ) + + +def list_pending_transform_rows( + documents: list[dict[str, Any]], + handoff_by_id: dict[str, dict[str, Any]], + *, + db_policy_api_ids: set[int] | None = None, +) -> list[dict[str, Any]]: + db_ids = db_policy_api_ids or set() + pending: list[dict[str, Any]] = [] + for row in documents: + content_id = row_content_id(row) + if not content_id: + continue + policy_id = parse_policy_api_id(row) + if policy_id is not None and policy_id in db_ids: + continue + if content_id in handoff_by_id: + continue + pending.append(row) + return pending + + +def count_pending_transform( + documents: list[dict[str, Any]], + handoff_by_id: dict[str, dict[str, Any]], + *, + db_policy_api_ids: set[int] | None = None, +) -> int: + return len( + list_pending_transform_rows( + documents, + handoff_by_id, + db_policy_api_ids=db_policy_api_ids, + ), ) @@ -107,6 +194,9 @@ async def transform_documents_jsonl( output_path: Path | None = None, limit: int | None = None, model: str | None = None, + s3_util: S3Util | None = None, + db_policy_api_ids: set[int] | None = None, + merge_handoff: bool = True, ) -> PolicyPinTransformResult: src = input_path or POLICY_DOCUMENTS_PATH dst = output_path or POLICY_HANDOFF_PATH @@ -115,36 +205,62 @@ async def transform_documents_jsonl( f"원문 JSONL 없음: {src}. 먼저 GET /policy-pins/search 또는 fetch 스크립트를 실행하세요.", ) + documents = load_jsonl_rows(src) + handoff_by_id = load_rows_by_content_id(dst) if merge_handoff else {} + pending = list_pending_transform_rows( + documents, + handoff_by_id, + db_policy_api_ids=db_policy_api_ids, + ) + llm = build_llm_service(model=model) - results: list[dict] = [] - errors: list[dict] = [] + s3 = s3_util or S3Util() + processed_rows: list[dict[str, Any]] = [] + errors: list[dict[str, Any]] = [] + skipped_duplicate_count = 0 - for row in iter_jsonl(src): - if not isinstance(row, dict): - continue - if limit is not None and len(results) >= limit: - break - - content_id = str(row.get("contentid") or "").strip() - try: - results.append(await transform_one_row(llm, row)) - except Exception as exc: - errors.append( - { + rows_to_process = pending if limit is None else pending[:limit] + concurrency = min(settings.policy_transform_concurrency, len(rows_to_process) or 1) + semaphore = asyncio.Semaphore(max(1, concurrency)) + + async def _transform_row(row: dict[str, Any]) -> tuple[str, dict[str, Any] | None, dict[str, Any] | None]: + content_id = row_content_id(row) + async with semaphore: + try: + handoff_row = await transform_one_row(llm, row, s3_util=s3) + return content_id, handoff_row, None + except Exception as exc: + return content_id, None, { "contentid": content_id, "pin_title": row.get("pin_title"), "error": str(exc), } - ) - write_jsonl(dst, results) - pins = [PolicyPinHandoffDTO.from_row(item) for item in results] + for content_id, handoff_row, err_row in await asyncio.gather( + *(_transform_row(row) for row in rows_to_process), + ): + if handoff_row is not None and content_id: + handoff_by_id[content_id] = handoff_row + processed_rows.append(handoff_row) + elif err_row is not None: + errors.append(err_row) + + for row in documents: + content_id = row_content_id(row) + if not content_id or content_id in handoff_by_id: + continue + policy_id = parse_policy_api_id(row) + if policy_id is not None and (db_policy_api_ids or set()) and policy_id in (db_policy_api_ids or set()): + skipped_duplicate_count += 1 + + write_handoff_map(handoff_by_id, dst) + pins = [PolicyPinHandoffDTO.from_row(item) for item in processed_rows] hint: str | None - if results: + if processed_rows or handoff_by_id: hint = ( - f"response.pins({len(pins)}건)와 {dst.name} 내용이 동일합니다. " - "기간 필터는 이미 search(policy_documents.jsonl) 단계에서 적용된 범위입니다." + f"이번 가공 {len(processed_rows)}건, handoff 총 {len(handoff_by_id)}건. " + "기간 필터는 search(policy_documents.jsonl) 단계에서 적용된 범위입니다." ) else: hint = "가공 성공 건이 없습니다. GEMINI_API_KEY·원문 pin_content를 확인하세요." @@ -152,9 +268,11 @@ async def transform_documents_jsonl( return PolicyPinTransformResult( input_path=str(src), output_path=str(dst), - processed_count=len(results), + processed_count=len(processed_rows), error_count=len(errors), errors=errors, pins=pins, hint=hint, + skipped_duplicate_count=skipped_duplicate_count, + pending_count=len(pending), ) diff --git a/app/services/policy_pipeline_cleanup.py b/app/services/policy_pipeline_cleanup.py new file mode 100644 index 0000000..38ffb16 --- /dev/null +++ b/app/services/policy_pipeline_cleanup.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import logging +import shutil +from pathlib import Path + +from app.services.policy_cardnews import POLICY_CARDNEWS_OUTPUT_DIR +from app.services.policy_pin_transform import ( + POLICY_DOCUMENTS_PATH, + POLICY_HANDOFF_PATH, + load_jsonl_rows, + parse_policy_api_id, + row_content_id, + write_handoff_map, +) +from rag.scripts.chunk_module import write_jsonl + +logger = logging.getLogger(__name__) + + +def prune_jsonl_by_policy_api_ids( + path: Path, + policy_api_ids: set[int], +) -> int: + if not policy_api_ids or not path.is_file(): + return 0 + + kept: list[dict] = [] + removed = 0 + for row in load_jsonl_rows(path): + policy_id = parse_policy_api_id(row) + if policy_id is not None and policy_id in policy_api_ids: + removed += 1 + continue + kept.append(row) + + if removed > 0: + write_jsonl(path, kept) + logger.info("JSONL 정리 %s: %d건 제거, %d건 유지", path.name, removed, len(kept)) + return removed + + +def prune_pipeline_imported( + policy_api_ids: set[int], + *, + documents_path: Path | None = None, + handoff_path: Path | None = None, +) -> dict[str, int]: + if not policy_api_ids: + return {"documents_removed": 0, "handoff_removed": 0} + + docs = documents_path or POLICY_DOCUMENTS_PATH + handoff = handoff_path or POLICY_HANDOFF_PATH + return { + "documents_removed": prune_jsonl_by_policy_api_ids(docs, policy_api_ids), + "handoff_removed": prune_jsonl_by_policy_api_ids(handoff, policy_api_ids), + } + + +def prune_handoff_pending_only( + *, + db_policy_api_ids: set[int], + handoff_path: Path | None = None, +) -> int: + """DB에 이미 있는 항목을 handoff에서 제거 (가공 완료·적재 완료 캐시 정리).""" + path = handoff_path or POLICY_HANDOFF_PATH + if not path.is_file() or not db_policy_api_ids: + return 0 + return prune_jsonl_by_policy_api_ids(path, db_policy_api_ids) + + +def cleanup_local_cardnews_dirs(content_ids: set[str]) -> int: + removed = 0 + for raw_id in content_ids: + content_id = str(raw_id or "").strip() + if not content_id: + continue + target = POLICY_CARDNEWS_OUTPUT_DIR / content_id + if not target.is_dir(): + continue + shutil.rmtree(target, ignore_errors=True) + removed += 1 + logger.debug("로컬 카드뉴스 삭제: %s", target) + return removed + + +def policy_api_ids_to_content_ids(policy_api_ids: set[int]) -> set[str]: + return {str(policy_id) for policy_id in policy_api_ids} + + +def cleanup_after_policy_import(policy_api_ids: set[int]) -> dict[str, int]: + stats = prune_pipeline_imported(policy_api_ids) + stats["cardnews_dirs_removed"] = cleanup_local_cardnews_dirs( + policy_api_ids_to_content_ids(policy_api_ids), + ) + return stats + + +def reset_handoff_map(handoff_path: Path | None = None) -> None: + write_handoff_map({}, handoff_path) + + +def content_ids_from_rows(rows: list[dict]) -> set[str]: + out: set[str] = set() + for row in rows: + content_id = row_content_id(row) + if content_id: + out.add(content_id) + return out diff --git a/app/utils/policy_news_parse.py b/app/utils/policy_news_parse.py index faad181..7ebca9f 100644 --- a/app/utils/policy_news_parse.py +++ b/app/utils/policy_news_parse.py @@ -4,6 +4,7 @@ import logging import re from datetime import date, datetime, timedelta +from zoneinfo import ZoneInfo from html import unescape from typing import Any from urllib.parse import urljoin, urlparse @@ -13,6 +14,7 @@ logger = logging.getLogger(__name__) +_KST = ZoneInfo("Asia/Seoul") _POLICY_DATE_FMT = "%m/%d/%Y %H:%M:%S" _POLICY_DATE_FMT_SHORT = "%m/%d/%Y" _IMG_SRC_RE = re.compile( @@ -329,6 +331,22 @@ def approve_date_to_yyyymmdd(raw: str | None) -> str | None: return parsed.strftime("%Y%m%d") +def policy_date_kst(raw: str | None) -> date | None: + parsed = parse_policy_datetime(raw) + if parsed is None: + return None + return parsed.date() + + +def is_embargo_active(embargo_raw: str | None, *, as_of: date | None = None) -> bool: + """EmbargoDate가 as_of(KST, 기본 오늘)보다 미래이면 아직 엠바고 → True.""" + embargo_date = policy_date_kst(embargo_raw) + if embargo_date is None: + return False + today = as_of if as_of is not None else datetime.now(_KST).date() + return embargo_date > today + + def local_name(tag: str) -> str: if "}" in tag: return tag.rsplit("}", 1)[-1] @@ -434,6 +452,7 @@ def build_policy_document_row(item: dict[str, str]) -> dict[str, Any] | None: "source_url": source_url, "subtitles": merge_subtitles(item), "contents_status": (item.get("ContentsStatus") or "").strip(), + "embargo_date": (item.get("EmbargoDate") or "").strip(), } diff --git a/rag/scripts/fetch_policy_news.py b/rag/scripts/fetch_policy_news.py index 2e8be1d..19e9e50 100644 --- a/rag/scripts/fetch_policy_news.py +++ b/rag/scripts/fetch_policy_news.py @@ -21,6 +21,7 @@ from app.clients.PolicyNewsClient import PolicyNewsClient from app.utils.policy_news_parse import ( build_policy_document_row, + is_embargo_active, iter_date_chunks, validate_yyyymmdd, ) @@ -45,6 +46,7 @@ async def fetch_policy_documents( "documents": 0, "skipped_duplicate": 0, "skipped_invalid": 0, + "skipped_embargo": 0, "api_errors": 0, } seen_ids: set[str] = set() @@ -83,6 +85,10 @@ async def fetch_policy_documents( continue seen_ids.add(news_id) + if is_embargo_active(item.get("EmbargoDate")): + stats["skipped_embargo"] += 1 + continue + row = build_policy_document_row(item) if row is None: stats["skipped_invalid"] += 1 From 1a29a066bd3993eba720496e59ec8ef39716fb64 Mon Sep 17 00:00:00 2001 From: selnem Date: Fri, 12 Jun 2026 23:16:01 +0900 Subject: [PATCH 2/8] =?UTF-8?q?fix:=20=EC=A0=95=EC=B1=85=20=ED=95=80=20?= =?UTF-8?q?=ED=8C=8C=EC=9D=B4=ED=94=84=EB=9D=BC=EC=9D=B8=20=EC=BD=94?= =?UTF-8?q?=EB=93=9C=EB=A6=AC=EB=B7=B0=20=EC=A7=80=EC=A0=81=204=EA=B1=B4?= =?UTF-8?q?=20=EB=B0=98=EC=98=81=20sync=5Fpipeline=20=EB=8B=A4=EB=B0=B0?= =?UTF-8?q?=EC=B9=98=20=EC=8B=A4=ED=96=89=20=EC=8B=9C=20import=20errors/it?= =?UTF-8?q?ems/pin=5Fids=EA=B0=80=20=EB=A7=88=EC=A7=80=EB=A7=89=20?= =?UTF-8?q?=EB=B0=B0=EC=B9=98=EB=A7=8C=20=EB=B0=98=EC=98=81=EB=90=98?= =?UTF-8?q?=EB=8D=98=20=EB=AC=B8=EC=A0=9C=EB=A5=BC=20=EB=88=84=EC=A0=81?= =?UTF-8?q?=ED=95=98=EB=8F=84=EB=A1=9D=20=EC=88=98=EC=A0=95=ED=95=98?= =?UTF-8?q?=EA=B3=A0,=20=EC=8A=A4=EC=BC=80=EC=A4=84=EB=9F=AC=20=EA=B0=84?= =?UTF-8?q?=EA=B2=A9=20=EB=B9=84=EA=B5=90=EC=97=90=201=EC=8B=9C=EA=B0=84?= =?UTF-8?q?=20=EB=B2=84=ED=8D=BC=EB=A5=BC=20=EC=A0=81=EC=9A=A9=ED=95=B4=20?= =?UTF-8?q?=EC=8B=A4=ED=96=89=20=EC=8B=9C=EA=B0=81=20=EC=98=A4=EC=B0=A8?= =?UTF-8?q?=EB=A1=9C=20sync=EA=B0=80=20=ED=95=98=EB=A3=A8=20=EB=B0=80?= =?UTF-8?q?=EB=A6=AC=EB=8A=94=20=ED=98=84=EC=83=81=EC=9D=84=20=EB=B0=A9?= =?UTF-8?q?=EC=A7=80=ED=95=9C=EB=8B=A4.=20=EB=8F=99=EC=9E=91=ED=95=98?= =?UTF-8?q?=EC=A7=80=20=EC=95=8A=EB=8D=98=20cardnews=5Fimage=5Furls=20fall?= =?UTF-8?q?back=EC=9D=84=20=EC=A0=9C=EA=B1=B0=ED=95=98=EA=B3=A0,=20?= =?UTF-8?q?=EC=B9=B4=EB=93=9C=EB=89=B4=EC=8A=A4=20handoff=5Fpath=20?= =?UTF-8?q?=EC=A0=88=EB=8C=80=20=EA=B2=BD=EB=A1=9C=20=EC=B2=98=EB=A6=AC?= =?UTF-8?q?=EB=A5=BC=20=EB=B3=B4=EC=99=84=ED=95=9C=EB=8B=A4.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/PolicyEventIngestService.py | 5 ----- app/services/PolicyPinService.py | 21 ++++++++++++++----- .../internal/PolicyPinSchedulerService.py | 3 ++- app/services/policy_cardnews.py | 5 +++-- 4 files changed, 21 insertions(+), 13 deletions(-) diff --git a/app/services/PolicyEventIngestService.py b/app/services/PolicyEventIngestService.py index e3d7621..25f3c3d 100644 --- a/app/services/PolicyEventIngestService.py +++ b/app/services/PolicyEventIngestService.py @@ -274,11 +274,6 @@ async def _insert_policy_pin( await self._community_repo.save(community, flush_immediately=True) cardnews_images = row.get("cardnews_images") or [] - if not cardnews_images: - for url in row.get("cardnews_image_urls") or []: - url_text = str(url).strip() - if url_text: - cardnews_images.append({"key": "", "url": url_text}) for image in cardnews_images: if not isinstance(image, dict): diff --git a/app/services/PolicyPinService.py b/app/services/PolicyPinService.py index 0b428e6..773faac 100644 --- a/app/services/PolicyPinService.py +++ b/app/services/PolicyPinService.py @@ -8,7 +8,12 @@ from app.clients.PolicyNewsClient import PolicyNewsClient from app.core.config import settings -from app.schemas.PolicyAdminDTO import PolicyImportBatchResult, PolicySyncResult, PolicyTransformBatchResult +from app.schemas.PolicyAdminDTO import ( + PolicyBatchItemResult, + PolicyImportBatchResult, + PolicySyncResult, + PolicyTransformBatchResult, +) from app.schemas.PolicyPinDTO import ( PolicyPinHandoffDTO, PolicyPinHandoffResult, @@ -184,6 +189,9 @@ async def sync_pipeline( total_skipped_transform = 0 transform_errors: list[dict] = [] transform_pins: list = [] + import_errors: list[dict] = [] + import_items: list[PolicyBatchItemResult] = [] + import_pin_ids: list[int] = [] batches_run = 0 last_transform_pending = 0 last_import = PolicyImportBatchResult( @@ -223,6 +231,9 @@ async def sync_pipeline( ) total_imported += import_batch.inserted_count total_skipped_import += import_batch.skipped_duplicate_count + import_errors.extend(import_batch.errors) + import_items.extend(import_batch.items) + import_pin_ids.extend(import_batch.pin_ids) last_import = import_batch if transform_batch.processed_count < batch_limit: @@ -248,10 +259,10 @@ async def sync_pipeline( inserted_count=total_imported, skipped_duplicate_count=total_skipped_import, pending_import_count=last_import.pending_import_count, - error_count=last_import.error_count, - errors=last_import.errors, - items=last_import.items, - pin_ids=last_import.pin_ids, + error_count=len(import_errors), + errors=import_errors, + items=import_items, + pin_ids=import_pin_ids, requested_batch_size=effective_batch, ) diff --git a/app/services/internal/PolicyPinSchedulerService.py b/app/services/internal/PolicyPinSchedulerService.py index 7e78622..9309756 100644 --- a/app/services/internal/PolicyPinSchedulerService.py +++ b/app/services/internal/PolicyPinSchedulerService.py @@ -106,7 +106,8 @@ def _should_run_sync(self) -> bool: if last_run.tzinfo is None: last_run = last_run.replace(tzinfo=_KST) elapsed = datetime.now(_KST) - last_run.astimezone(_KST) - return elapsed >= timedelta(days=settings.policy_sync_interval_days) + # 스케줄 시각 오차로 하루 밀리는 것 방지 + return elapsed >= timedelta(days=settings.policy_sync_interval_days) - timedelta(hours=1) except (json.JSONDecodeError, ValueError): return True diff --git a/app/services/policy_cardnews.py b/app/services/policy_cardnews.py index a48b583..c7df538 100644 --- a/app/services/policy_cardnews.py +++ b/app/services/policy_cardnews.py @@ -74,8 +74,9 @@ async def _upload_local_handoff_path( content_id: str, handoff_path: str, ) -> CardnewsS3Image: - rel = handoff_path.strip().lstrip("/") - local_path = _REPO_ROOT / rel + local_path = Path(handoff_path) + if not local_path.is_absolute(): + local_path = (_REPO_ROOT / handoff_path.strip().lstrip("/")).resolve() if not local_path.is_file(): raise FileNotFoundError(f"카드뉴스 로컬 파일 없음: {local_path}") slide_name = local_path.stem From 6eec2df91a549036345ff4e5dcfca2bac565a5e1 Mon Sep 17 00:00:00 2001 From: selnem Date: Fri, 12 Jun 2026 23:24:13 +0900 Subject: [PATCH 3/8] =?UTF-8?q?fix:=20import=5Fhandoff=5Fbatch=20=EB=A9=94?= =?UTF-8?q?=ED=83=80=EB=8D=B0=EC=9D=B4=ED=84=B0=20=EC=98=A4=ED=83=80=20?= =?UTF-8?q?=EC=88=98=EC=A0=95=20=EB=B0=8F=20=EC=A0=95=EC=B1=85=20=ED=8C=8C?= =?UTF-8?q?=EC=9D=B4=ED=94=84=EB=9D=BC=EC=9D=B8=20=ED=85=8C=EC=8A=A4?= =?UTF-8?q?=ED=8A=B8=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/PolicyEventIngestService.py | 2 +- tests/test_policy_admin_batch.py | 268 +++++++++++++++++++++++ 2 files changed, 269 insertions(+), 1 deletion(-) create mode 100644 tests/test_policy_admin_batch.py diff --git a/app/services/PolicyEventIngestService.py b/app/services/PolicyEventIngestService.py index 25f3c3d..deb0854 100644 --- a/app/services/PolicyEventIngestService.py +++ b/app/services/PolicyEventIngestService.py @@ -219,7 +219,7 @@ async def import_handoff_batch( } _IMPORT_REPORT_PATH.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") - effective_batch = limit if not import_all else limit + effective_batch = None if import_all else limit return PolicyImportBatchResult( inserted_count=inserted_count, skipped_duplicate_count=skipped_duplicate_count, diff --git a/tests/test_policy_admin_batch.py b/tests/test_policy_admin_batch.py new file mode 100644 index 0000000..f3f5f5c --- /dev/null +++ b/tests/test_policy_admin_batch.py @@ -0,0 +1,268 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from datetime import datetime, timedelta +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch +from zoneinfo import ZoneInfo + +import app.models # noqa: F401 — SQLAlchemy mapper registry + +from app.schemas.PolicyAdminDTO import ( + PolicyBatchAction, + PolicyBatchItemResult, + PolicyImportBatchResult, +) +from app.schemas.PolicyPinDTO import PolicyPinSearchResult, PolicyPinTransformResult +from app.services.PolicyEventIngestService import PolicyEventIngestService +from app.services.PolicyPinService import PolicyPinService +from app.services.internal.PolicyPinSchedulerService import PolicyPinSchedulerService +from app.services.policy_cardnews import _upload_local_handoff_path + +_KST = ZoneInfo("Asia/Seoul") + + +def _make_ingest_service() -> PolicyEventIngestService: + session = MagicMock() + session.begin_nested = MagicMock(return_value=MagicMock(__aenter__=AsyncMock(), __aexit__=AsyncMock())) + pin_repo = MagicMock() + pin_repo.session = session + pin_repo.commit = AsyncMock() + pin_repo.rollback = AsyncMock() + event_pin_repo = MagicMock() + event_pin_repo.list_policy_api_ids = AsyncMock(return_value=set()) + event_pin_repo.get_by_policy_api_id = AsyncMock(return_value=None) + event_pin_repo.save = AsyncMock() + community_repo = MagicMock() + community_repo.save = AsyncMock() + cardnews_repo = MagicMock() + cardnews_repo.save = AsyncMock() + user_repo = MagicMock() + user_repo.get_by_user_name = AsyncMock(return_value=MagicMock(uid="admin-uid")) + return PolicyEventIngestService( + pin_repo=pin_repo, + event_pin_repo=event_pin_repo, + community_repo=community_repo, + cardnews_image_s3_repo=cardnews_repo, + user_repo=user_repo, + ) + + +class PolicyPinSchedulerShouldRunSyncTest(unittest.TestCase): + def test_runs_when_elapsed_within_one_hour_of_interval(self) -> None: + scheduler = PolicyPinSchedulerService(s3_util=MagicMock()) + last_sync = datetime(2026, 6, 9, 1, 0, 5, tzinfo=_KST) + now = datetime(2026, 6, 12, 1, 0, 0, tzinfo=_KST) + + with ( + patch( + "app.services.internal.PolicyPinSchedulerService.POLICY_SYNC_META_PATH", + MagicMock(is_file=MagicMock(return_value=True)), + ), + patch( + "app.services.internal.PolicyPinSchedulerService.json.loads", + return_value={"last_sync_at": last_sync.isoformat()}, + ), + patch( + "app.services.internal.PolicyPinSchedulerService.datetime") as mock_dt, + patch( + "app.services.internal.PolicyPinSchedulerService.settings.policy_sync_interval_days", + 3, + ), + ): + mock_dt.now.return_value = now + mock_dt.fromisoformat = datetime.fromisoformat + self.assertTrue(scheduler._should_run_sync()) + + def test_skips_when_elapsed_more_than_one_hour_short_of_interval(self) -> None: + scheduler = PolicyPinSchedulerService(s3_util=MagicMock()) + last_sync = datetime(2026, 6, 9, 1, 0, 5, tzinfo=_KST) + now = datetime(2026, 6, 11, 23, 0, 0, tzinfo=_KST) + + with ( + patch( + "app.services.internal.PolicyPinSchedulerService.POLICY_SYNC_META_PATH", + MagicMock(is_file=MagicMock(return_value=True)), + ), + patch( + "app.services.internal.PolicyPinSchedulerService.json.loads", + return_value={"last_sync_at": last_sync.isoformat()}, + ), + patch( + "app.services.internal.PolicyPinSchedulerService.datetime") as mock_dt, + patch( + "app.services.internal.PolicyPinSchedulerService.settings.policy_sync_interval_days", + 3, + ), + ): + mock_dt.now.return_value = now + mock_dt.fromisoformat = datetime.fromisoformat + self.assertFalse(scheduler._should_run_sync()) + + +class PolicyCardnewsHandoffPathTest(unittest.IsolatedAsyncioTestCase): + async def test_upload_local_handoff_path_accepts_absolute_path(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + slide_path = Path(tmp) / "slide_01.png" + slide_path.write_bytes(b"png-bytes") + s3_util = MagicMock() + s3_util.upload_bytes = AsyncMock( + return_value={"key": "policy/cardnews/1/slide_01.png", "url": "https://cdn.example/slide_01.png"}, + ) + + result = await _upload_local_handoff_path( + s3_util, + content_id="1", + handoff_path=str(slide_path), + ) + + self.assertEqual(result["key"], "policy/cardnews/1/slide_01.png") + s3_util.upload_bytes.assert_awaited_once() + + +class PolicyImportHandoffBatchTest(unittest.IsolatedAsyncioTestCase): + async def test_requested_batch_size_none_when_import_all(self) -> None: + service = _make_ingest_service() + handoff_rows = [ + { + "contentid": "1", + "policy_api_id": 1, + "title": "정책1", + "pin_content": "본문1", + "cardnews_images": [{"key": "k1", "url": "https://cdn.example/1.png"}], + }, + ] + with ( + patch( + "app.services.PolicyEventIngestService.load_jsonl_rows", + return_value=handoff_rows, + ), + patch( + "app.services.PolicyEventIngestService.settings.policy_prune_pipeline_after_import", + False, + ), + patch.object(service, "_insert_policy_pin", AsyncMock(return_value=101)), + ): + result = await service.import_handoff_batch(import_all=True, limit=5) + + self.assertIsNone(result.requested_batch_size) + self.assertEqual(result.inserted_count, 1) + + async def test_requested_batch_size_limit_when_not_import_all(self) -> None: + service = _make_ingest_service() + handoff_rows = [ + { + "contentid": "1", + "policy_api_id": 1, + "title": "정책1", + "pin_content": "본문1", + "cardnews_images": [{"key": "k1", "url": "https://cdn.example/1.png"}], + }, + ] + with ( + patch( + "app.services.PolicyEventIngestService.load_jsonl_rows", + return_value=handoff_rows, + ), + patch( + "app.services.PolicyEventIngestService.settings.policy_prune_pipeline_after_import", + False, + ), + patch.object(service, "_insert_policy_pin", AsyncMock(return_value=101)), + ): + result = await service.import_handoff_batch(import_all=False, limit=3) + + self.assertEqual(result.requested_batch_size, 3) + + +class PolicySyncPipelineAccumulationTest(unittest.IsolatedAsyncioTestCase): + async def test_import_results_accumulate_across_batches(self) -> None: + service = PolicyPinService() + ingest = _make_ingest_service() + ingest._event_pin_repo.list_policy_api_ids = AsyncMock(return_value=set()) + + transform_batches = [ + PolicyPinTransformResult( + input_path="in.jsonl", + output_path="out.jsonl", + processed_count=2, + error_count=0, + pins=[], + pending_count=1, + ), + PolicyPinTransformResult( + input_path="in.jsonl", + output_path="out.jsonl", + processed_count=1, + error_count=0, + pins=[], + pending_count=0, + ), + ] + import_batches = [ + PolicyImportBatchResult( + inserted_count=2, + skipped_duplicate_count=0, + pending_import_count=1, + error_count=1, + errors=[{"policy_api_id": 1, "error": "e1"}], + items=[ + PolicyBatchItemResult(policy_api_id=1, action=PolicyBatchAction.CREATED), + PolicyBatchItemResult(policy_api_id=2, action=PolicyBatchAction.ERROR, message="e1"), + ], + pin_ids=[11, 12], + ), + PolicyImportBatchResult( + inserted_count=1, + skipped_duplicate_count=0, + pending_import_count=0, + error_count=0, + errors=[], + items=[PolicyBatchItemResult(policy_api_id=3, action=PolicyBatchAction.CREATED)], + pin_ids=[13], + ), + ] + + with ( + patch( + "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import", + False, + ), + patch.object( + service, + "search_and_save", + AsyncMock( + return_value=PolicyPinSearchResult( + query_start_date="20260610", + query_end_date="20260612", + count=0, + pins=[], + saved_documents_path="docs.jsonl", + stats={}, + ), + ), + ), + patch.object(service, "transform_and_save", AsyncMock(side_effect=transform_batches)), + patch.object(ingest, "import_handoff_batch", AsyncMock(side_effect=import_batches)), + patch.object(PolicyEventIngestService, "write_sync_meta"), + ): + result = await service.sync_pipeline( + ingest_service=ingest, + s3_util=MagicMock(), + start_date="20260610", + end_date="20260612", + batch_size=2, + ) + + import_result = result.import_result + self.assertEqual(import_result.inserted_count, 3) + self.assertEqual(import_result.error_count, 1) + self.assertEqual(len(import_result.items), 3) + self.assertEqual(import_result.pin_ids, [11, 12, 13]) + self.assertEqual(import_result.errors, [{"policy_api_id": 1, "error": "e1"}]) + + +if __name__ == "__main__": + unittest.main() From dc7018716e7e9031b216d020c78b07c0efb70333 Mon Sep 17 00:00:00 2001 From: selnem Date: Fri, 12 Jun 2026 23:37:38 +0900 Subject: [PATCH 4/8] =?UTF-8?q?fix:=20=EC=A0=95=EC=B1=85=20=ED=95=80=20syn?= =?UTF-8?q?c=20=EB=82=A0=EC=A7=9C=20=EA=B2=80=EC=A6=9D=C2=B7handoff=20?= =?UTF-8?q?=EC=A0=81=EC=9E=AC=20drain=C2=B7=EB=AA=A8=EB=8D=B8=20unique=20?= =?UTF-8?q?=EB=8F=99=EA=B8=B0=ED=99=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/models/EventPin.py | 1 + app/routes/PolicyAdminRoute.py | 29 +++++++- app/services/PolicyEventIngestService.py | 3 +- app/services/PolicyPinService.py | 28 ++++++-- app/services/policy_cardnews.py | 3 +- tests/test_policy_admin_batch.py | 89 ++++++++++++++++++++++++ 6 files changed, 142 insertions(+), 11 deletions(-) diff --git a/app/models/EventPin.py b/app/models/EventPin.py index f807f46..519d918 100644 --- a/app/models/EventPin.py +++ b/app/models/EventPin.py @@ -40,6 +40,7 @@ class EventPin(BaseEntity): policy_api_id: Mapped[int | None] = mapped_column( "policy_api_id", BigInteger, + unique=True, nullable=True, ) event_start_time: Mapped[datetime] = mapped_column("event_start_time", DateTime, nullable=False) diff --git a/app/routes/PolicyAdminRoute.py b/app/routes/PolicyAdminRoute.py index 1682249..69453a6 100644 --- a/app/routes/PolicyAdminRoute.py +++ b/app/routes/PolicyAdminRoute.py @@ -17,6 +17,7 @@ PolicySyncResult, PolicyTransformBatchResult, ) +from app.utils.policy_news_parse import validate_yyyymmdd router = APIRouter(prefix="/policy-admin", tags=["policy-admin"]) @@ -38,6 +39,27 @@ def _clamp_batch(value: int | None) -> int: return min(max(value, 1), _MAX_BATCH) +def _validate_optional_date_range( + start_date: str | None, + end_date: str | None, +) -> tuple[str | None, str | None]: + if start_date is None and end_date is None: + return None, None + if start_date is None or end_date is None: + raise HTTPException( + status_code=ErrorCode.BAD_REQUEST.http_status, + detail="start_date와 end_date는 함께 지정해야 합니다.", + ) + parsed_start = validate_yyyymmdd(start_date, label="start_date") + parsed_end = validate_yyyymmdd(end_date, label="end_date") + if parsed_start > parsed_end: + raise HTTPException( + status_code=ErrorCode.BAD_REQUEST.http_status, + detail="start_date는 end_date보다 이후일 수 없습니다.", + ) + return parsed_start, parsed_end + + @router.post( "/sync", response_model=SuccessEnvelope[PolicySyncResult], @@ -56,15 +78,18 @@ async def sync_policy_pins( transform_limit: int | None = Query(default=None, ge=1, le=100, description="가공 최대 건수"), batch_size: int | None = Query(default=None, ge=1, le=25, description="배치 크기"), ) -> SuccessEnvelope[PolicySyncResult]: + validated_start, validated_end = _validate_optional_date_range(start_date, end_date) try: body = await service.sync_pipeline( ingest_service=ingest_service, s3_util=s3_util, - start_date=start_date, - end_date=end_date, + start_date=validated_start, + end_date=validated_end, transform_limit=transform_limit, batch_size=_clamp_batch(batch_size), ) + except ValueError as exc: + raise HTTPException(status_code=ErrorCode.BAD_REQUEST.http_status, detail=str(exc)) from exc except FileNotFoundError as exc: raise HTTPException(status_code=ErrorCode.NOT_FOUND.http_status, detail=str(exc)) from exc except RuntimeError as exc: diff --git a/app/services/PolicyEventIngestService.py b/app/services/PolicyEventIngestService.py index deb0854..4b0328d 100644 --- a/app/services/PolicyEventIngestService.py +++ b/app/services/PolicyEventIngestService.py @@ -7,6 +7,7 @@ from typing import Any from zoneinfo import ZoneInfo +from anyio import to_thread from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings @@ -120,7 +121,7 @@ async def import_handoff_batch( import_all: bool = True, limit: int | None = None, ) -> PolicyImportBatchResult: - handoff_rows = load_jsonl_rows(POLICY_HANDOFF_PATH) + handoff_rows = await to_thread.run_sync(load_jsonl_rows, POLICY_HANDOFF_PATH) if not handoff_rows: raise FileNotFoundError( f"핸드오프 JSONL 없음: {POLICY_HANDOFF_PATH}. transform을 먼저 실행하세요.", diff --git a/app/services/PolicyPinService.py b/app/services/PolicyPinService.py index 773faac..90d13fa 100644 --- a/app/services/PolicyPinService.py +++ b/app/services/PolicyPinService.py @@ -201,6 +201,15 @@ async def sync_pipeline( error_count=0, ) + def _accumulate_import(import_batch: PolicyImportBatchResult) -> None: + nonlocal total_imported, total_skipped_import, last_import + total_imported += import_batch.inserted_count + total_skipped_import += import_batch.skipped_duplicate_count + import_errors.extend(import_batch.errors) + import_items.extend(import_batch.items) + import_pin_ids.extend(import_batch.pin_ids) + last_import = import_batch + while True: if transform_limit is not None and total_processed >= transform_limit: break @@ -229,16 +238,23 @@ async def sync_pipeline( import_all=False, limit=effective_batch, ) - total_imported += import_batch.inserted_count - total_skipped_import += import_batch.skipped_duplicate_count - import_errors.extend(import_batch.errors) - import_items.extend(import_batch.items) - import_pin_ids.extend(import_batch.pin_ids) - last_import = import_batch + _accumulate_import(import_batch) if transform_batch.processed_count < batch_limit: break + while True: + prev_pending = last_import.pending_import_count + import_batch = await ingest_service.import_handoff_batch( + import_all=False, + limit=effective_batch, + ) + _accumulate_import(import_batch) + if import_batch.pending_import_count == 0: + break + if import_batch.inserted_count == 0 and import_batch.pending_import_count >= prev_pending: + break + aggregated_transform = PolicyTransformBatchResult( input_path=str(self.documents_path()), output_path=str(self.handoff_path()), diff --git a/app/services/policy_cardnews.py b/app/services/policy_cardnews.py index c7df538..630231d 100644 --- a/app/services/policy_cardnews.py +++ b/app/services/policy_cardnews.py @@ -250,8 +250,7 @@ async def generate_cardnews_s3_images( ), ) - if not settings.policy_cardnews_keep_local_files: - _maybe_cleanup_local_cardnews(content_id) + _maybe_cleanup_local_cardnews(content_id) return uploaded diff --git a/tests/test_policy_admin_batch.py b/tests/test_policy_admin_batch.py index f3f5f5c..12d0659 100644 --- a/tests/test_policy_admin_batch.py +++ b/tests/test_policy_admin_batch.py @@ -223,6 +223,15 @@ async def test_import_results_accumulate_across_batches(self) -> None: items=[PolicyBatchItemResult(policy_api_id=3, action=PolicyBatchAction.CREATED)], pin_ids=[13], ), + PolicyImportBatchResult( + inserted_count=0, + skipped_duplicate_count=0, + pending_import_count=0, + error_count=0, + errors=[], + items=[], + pin_ids=[], + ), ] with ( @@ -263,6 +272,86 @@ async def test_import_results_accumulate_across_batches(self) -> None: self.assertEqual(import_result.pin_ids, [11, 12, 13]) self.assertEqual(import_result.errors, [{"policy_api_id": 1, "error": "e1"}]) + async def test_import_drains_backlog_when_transform_produces_zero(self) -> None: + service = PolicyPinService() + ingest = _make_ingest_service() + ingest._event_pin_repo.list_policy_api_ids = AsyncMock(return_value=set()) + + import_batches = [ + PolicyImportBatchResult( + inserted_count=2, + skipped_duplicate_count=0, + pending_import_count=1, + error_count=0, + errors=[], + items=[ + PolicyBatchItemResult(policy_api_id=1, action=PolicyBatchAction.CREATED), + PolicyBatchItemResult(policy_api_id=2, action=PolicyBatchAction.CREATED), + ], + pin_ids=[11, 12], + ), + PolicyImportBatchResult( + inserted_count=1, + skipped_duplicate_count=0, + pending_import_count=0, + error_count=0, + errors=[], + items=[PolicyBatchItemResult(policy_api_id=3, action=PolicyBatchAction.CREATED)], + pin_ids=[13], + ), + ] + import_mock = AsyncMock(side_effect=import_batches) + + with ( + patch( + "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import", + False, + ), + patch.object( + service, + "search_and_save", + AsyncMock( + return_value=PolicyPinSearchResult( + query_start_date="20260610", + query_end_date="20260612", + count=0, + pins=[], + saved_documents_path="docs.jsonl", + stats={}, + ), + ), + ), + patch.object( + service, + "transform_and_save", + AsyncMock( + return_value=PolicyPinTransformResult( + input_path="in.jsonl", + output_path="out.jsonl", + processed_count=0, + error_count=0, + pins=[], + pending_count=0, + ), + ), + ), + patch.object(ingest, "import_handoff_batch", import_mock), + patch.object(PolicyEventIngestService, "write_sync_meta"), + ): + result = await service.sync_pipeline( + ingest_service=ingest, + s3_util=MagicMock(), + start_date="20260610", + end_date="20260612", + batch_size=2, + ) + + self.assertEqual(import_mock.await_count, 2) + import_result = result.import_result + self.assertEqual(import_result.inserted_count, 3) + self.assertEqual(import_result.pending_import_count, 0) + self.assertEqual(import_result.pin_ids, [11, 12, 13]) + if __name__ == "__main__": unittest.main() From 89e470736017527296d34a328271d0192bc4d8ff Mon Sep 17 00:00:00 2001 From: selnem Date: Fri, 12 Jun 2026 23:51:05 +0900 Subject: [PATCH 5/8] =?UTF-8?q?fix:=20=EC=A0=95=EC=B1=85=20=EA=B4=80?= =?UTF-8?q?=EB=A6=AC=EC=9E=90=20API=20=EB=A6=AC=EB=B7=B0=20=EC=A7=80?= =?UTF-8?q?=EC=A0=81=20=EB=B3=B4=EC=99=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /policy-admin 접근 제어와 정책 파이프라인 캡슐화를 보강하고, 카드뉴스 정리 및 nested transaction 실패 처리의 데이터 유실 위험을 줄인다. Co-authored-by: Cursor --- app/models/Community.py | 7 ++- app/routes/PolicyAdminRoute.py | 8 +++- app/services/PolicyEventIngestService.py | 17 +++----- app/services/PolicyPinService.py | 4 +- app/services/policy_cardnews.py | 5 ++- app/services/policy_pin_transform.py | 5 ++- tests/test_policy_admin_batch.py | 55 +++++++++++++++++++++++- 7 files changed, 81 insertions(+), 20 deletions(-) diff --git a/app/models/Community.py b/app/models/Community.py index 29157a4..8123a24 100644 --- a/app/models/Community.py +++ b/app/models/Community.py @@ -21,7 +21,12 @@ class Community(BaseEntity): ForeignKey("pin.pin_id"), nullable=False, ) - community_type: Mapped[str] = mapped_column("community_type", String(50), nullable=False) + community_type: Mapped[str] = mapped_column( + "community_type", + String(50), + nullable=False, + server_default="POLICY", + ) popularity: Mapped[float] = mapped_column("popularity", Float, nullable=False, default=0.0) cardnews_images: Mapped[list["CardnewsImageS3"]] = relationship( "CardnewsImageS3", diff --git a/app/routes/PolicyAdminRoute.py b/app/routes/PolicyAdminRoute.py index 69453a6..d366318 100644 --- a/app/routes/PolicyAdminRoute.py +++ b/app/routes/PolicyAdminRoute.py @@ -5,6 +5,7 @@ from app.core.codes import ErrorCode, SuccessCode from app.core.config import settings from app.core.deps import ( + AdminUserIdDep, PolicyEventIngestServiceDep, PolicyPinSchedulerDep, PolicyPinServiceDep, @@ -73,6 +74,7 @@ async def sync_policy_pins( service: PolicyPinServiceDep, ingest_service: PolicyEventIngestServiceDep, s3_util: S3UtilDep, + _admin_uid: AdminUserIdDep, start_date: str | None = Query(default=None, min_length=8, max_length=8, description="YYYYMMDD"), end_date: str | None = Query(default=None, min_length=8, max_length=8, description="YYYYMMDD"), transform_limit: int | None = Query(default=None, ge=1, le=100, description="가공 최대 건수"), @@ -107,10 +109,11 @@ async def transform_policy_batch( service: PolicyPinServiceDep, ingest_service: PolicyEventIngestServiceDep, s3_util: S3UtilDep, + _admin_uid: AdminUserIdDep, batch_size: int | None = Query(default=None, ge=1, le=25, description="이번 배치 가공 건수"), ) -> SuccessEnvelope[PolicyTransformBatchResult]: try: - db_ids = await ingest_service._event_pin_repo.list_policy_api_ids() + db_ids = await ingest_service.get_imported_policy_api_ids() body = await service.transform_batch( batch_size=_clamp_batch(batch_size), s3_util=s3_util, @@ -131,6 +134,7 @@ async def transform_policy_batch( ) async def import_policy_batch( ingest_service: PolicyEventIngestServiceDep, + _admin_uid: AdminUserIdDep, batch_size: int | None = Query(default=None, ge=1, le=25, description="이번 배치 INSERT 건수"), ) -> SuccessEnvelope[PolicyImportBatchResult]: try: @@ -153,6 +157,7 @@ async def import_policy_batch( ) async def policy_pipeline_status( ingest_service: PolicyEventIngestServiceDep, + _admin_uid: AdminUserIdDep, ) -> SuccessEnvelope[PolicyPipelineStatusResult]: body = await ingest_service.get_pipeline_status() return success_response(result=body, success_code=SuccessCode.OK) @@ -168,6 +173,7 @@ async def run_policy_scheduler_once( ingest_service: PolicyEventIngestServiceDep, s3_util: S3UtilDep, scheduler: PolicyPinSchedulerDep, + _admin_uid: AdminUserIdDep, ) -> SuccessEnvelope[PolicySyncResult]: try: if scheduler is not None: diff --git a/app/services/PolicyEventIngestService.py b/app/services/PolicyEventIngestService.py index 4b0328d..c1b82d4 100644 --- a/app/services/PolicyEventIngestService.py +++ b/app/services/PolicyEventIngestService.py @@ -8,8 +8,6 @@ from zoneinfo import ZoneInfo from anyio import to_thread -from sqlalchemy.ext.asyncio import AsyncSession - from app.core.config import settings from app.models.CardnewsImageS3 import CardnewsImageS3 from app.models.Community import Community @@ -47,13 +45,6 @@ _COMMUNITY_TYPE_POLICY = "POLICY" -def _discard_session_state_after_nested_failure(session: AsyncSession) -> None: - for obj in list(session.new): - session.expunge(obj) - for obj in list(session.dirty): - session.expire(obj) - - def _parse_event_datetime(value: Any) -> datetime: if value is None: raise ValueError("event datetime 없음") @@ -105,6 +96,9 @@ async def commit(self) -> None: async def rollback(self) -> None: await self._pin_repo.rollback() + async def get_imported_policy_api_ids(self) -> set[int]: + return await self._event_pin_repo.list_policy_api_ids() + async def resolve_admin_uid(self) -> str: user_name = settings.policy_admin_user_name.strip() user = await self._user_repo.get_by_user_name(user_name) @@ -128,7 +122,7 @@ async def import_handoff_batch( ) uid = admin_uid or await self.resolve_admin_uid() - db_ids = await self._event_pin_repo.list_policy_api_ids() + db_ids = await self.get_imported_policy_api_ids() inserted_count = 0 skipped_duplicate_count = 0 errors: list[dict[str, Any]] = [] @@ -184,7 +178,6 @@ async def import_handoff_batch( ), ) except Exception as exc: - _discard_session_state_after_nested_failure(self._pin_repo.session) logger.exception("policy import failed policy_api_id=%s", policy_api_id) errors.append( { @@ -296,7 +289,7 @@ async def get_pipeline_status(self) -> PolicyPipelineStatusResult: meta = self._load_sync_meta() documents = load_jsonl_rows(POLICY_DOCUMENTS_PATH) handoff_by_id = load_rows_by_content_id(POLICY_HANDOFF_PATH) - db_ids = await self._event_pin_repo.list_policy_api_ids() + db_ids = await self.get_imported_policy_api_ids() pending_transform = count_pending_transform(documents, handoff_by_id, db_policy_api_ids=db_ids) pending_import = 0 diff --git a/app/services/PolicyPinService.py b/app/services/PolicyPinService.py index 90d13fa..6e21ded 100644 --- a/app/services/PolicyPinService.py +++ b/app/services/PolicyPinService.py @@ -173,7 +173,7 @@ async def sync_pipeline( effective_batch = batch_size or settings.policy_sync_batch_size if settings.policy_prune_pipeline_after_import: - db_ids_before = await ingest_service._event_pin_repo.list_policy_api_ids() + db_ids_before = await ingest_service.get_imported_policy_api_ids() prune_pipeline_imported(db_ids_before) search = await self.search_and_save( @@ -218,7 +218,7 @@ def _accumulate_import(import_batch: PolicyImportBatchResult) -> None: if transform_limit is not None: batch_limit = min(effective_batch, transform_limit - total_processed) - db_ids = await ingest_service._event_pin_repo.list_policy_api_ids() + db_ids = await ingest_service.get_imported_policy_api_ids() transform_batch = await self.transform_and_save( limit=batch_limit, s3_util=s3_util, diff --git a/app/services/policy_cardnews.py b/app/services/policy_cardnews.py index 630231d..5d699a4 100644 --- a/app/services/policy_cardnews.py +++ b/app/services/policy_cardnews.py @@ -32,7 +32,10 @@ class CardnewsS3Image(TypedDict): def cleanup_local_cardnews_dir(content_id: str) -> None: - target = POLICY_CARDNEWS_OUTPUT_DIR / str(content_id or "").strip() + normalized = str(content_id or "").strip() + if not normalized: + return + target = POLICY_CARDNEWS_OUTPUT_DIR / normalized if not target.is_dir(): return import shutil diff --git a/app/services/policy_pin_transform.py b/app/services/policy_pin_transform.py index 989f6ec..019e8a4 100644 --- a/app/services/policy_pin_transform.py +++ b/app/services/policy_pin_transform.py @@ -207,10 +207,11 @@ async def transform_documents_jsonl( documents = load_jsonl_rows(src) handoff_by_id = load_rows_by_content_id(dst) if merge_handoff else {} + db_ids = db_policy_api_ids or set() pending = list_pending_transform_rows( documents, handoff_by_id, - db_policy_api_ids=db_policy_api_ids, + db_policy_api_ids=db_ids, ) llm = build_llm_service(model=model) @@ -250,7 +251,7 @@ async def _transform_row(row: dict[str, Any]) -> tuple[str, dict[str, Any] | Non if not content_id or content_id in handoff_by_id: continue policy_id = parse_policy_api_id(row) - if policy_id is not None and (db_policy_api_ids or set()) and policy_id in (db_policy_api_ids or set()): + if policy_id is not None and policy_id in db_ids: skipped_duplicate_count += 1 write_handoff_map(handoff_by_id, dst) diff --git a/tests/test_policy_admin_batch.py b/tests/test_policy_admin_batch.py index 12d0659..169a433 100644 --- a/tests/test_policy_admin_batch.py +++ b/tests/test_policy_admin_batch.py @@ -19,7 +19,7 @@ from app.services.PolicyEventIngestService import PolicyEventIngestService from app.services.PolicyPinService import PolicyPinService from app.services.internal.PolicyPinSchedulerService import PolicyPinSchedulerService -from app.services.policy_cardnews import _upload_local_handoff_path +from app.services.policy_cardnews import _upload_local_handoff_path, cleanup_local_cardnews_dir _KST = ZoneInfo("Asia/Seoul") @@ -121,6 +121,16 @@ async def test_upload_local_handoff_path_accepts_absolute_path(self) -> None: self.assertEqual(result["key"], "policy/cardnews/1/slide_01.png") s3_util.upload_bytes.assert_awaited_once() + async def test_cleanup_local_cardnews_dir_ignores_empty_content_id(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) / "policy_cardnews" + root.mkdir() + + with patch("app.services.policy_cardnews.POLICY_CARDNEWS_OUTPUT_DIR", root): + cleanup_local_cardnews_dir("") + + self.assertTrue(root.is_dir()) + class PolicyImportHandoffBatchTest(unittest.IsolatedAsyncioTestCase): async def test_requested_batch_size_none_when_import_all(self) -> None: @@ -176,6 +186,49 @@ async def test_requested_batch_size_limit_when_not_import_all(self) -> None: self.assertEqual(result.requested_batch_size, 3) + async def test_get_imported_policy_api_ids_delegates_to_repo(self) -> None: + service = _make_ingest_service() + service._event_pin_repo.list_policy_api_ids = AsyncMock(return_value={1, 2}) + + result = await service.get_imported_policy_api_ids() + + self.assertEqual(result, {1, 2}) + service._event_pin_repo.list_policy_api_ids.assert_awaited_once() + + async def test_nested_failure_does_not_expunge_unrelated_session_state(self) -> None: + service = _make_ingest_service() + nested_ctx = MagicMock() + nested_ctx.__aenter__ = AsyncMock() + nested_ctx.__aexit__ = AsyncMock(return_value=False) + service._pin_repo.session.begin_nested = MagicMock(return_value=nested_ctx) + service._pin_repo.session.expunge = MagicMock() + service._pin_repo.session.expire = MagicMock() + handoff_rows = [ + { + "contentid": "1", + "policy_api_id": 1, + "title": "정책1", + "pin_content": "본문1", + "cardnews_images": [{"key": "k1", "url": "https://cdn.example/1.png"}], + }, + ] + with ( + patch( + "app.services.PolicyEventIngestService.load_jsonl_rows", + return_value=handoff_rows, + ), + patch( + "app.services.PolicyEventIngestService.settings.policy_prune_pipeline_after_import", + False, + ), + patch.object(service, "_insert_policy_pin", AsyncMock(side_effect=RuntimeError("boom"))), + ): + result = await service.import_handoff_batch(import_all=True) + + self.assertEqual(result.error_count, 1) + service._pin_repo.session.expunge.assert_not_called() + service._pin_repo.session.expire.assert_not_called() + class PolicySyncPipelineAccumulationTest(unittest.IsolatedAsyncioTestCase): async def test_import_results_accumulate_across_batches(self) -> None: From 1b97e63ca3188516e06c180256f88795ee6879fd Mon Sep 17 00:00:00 2001 From: selnem Date: Fri, 12 Jun 2026 23:57:56 +0900 Subject: [PATCH 6/8] =?UTF-8?q?fix:=20=EC=A0=95=EC=B1=85=20=EA=B4=80?= =?UTF-8?q?=EB=A6=AC=EC=9E=90=20=EB=A6=AC=EB=B7=B0=20=ED=9B=84=EC=86=8D=20?= =?UTF-8?q?=EC=A7=80=EC=A0=81=20=EB=B0=98=EC=98=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /policy-admin 날짜 검증 예외가 400으로 처리되도록 보완하고, Community.pin_id nullable 변경으로 기존 데이터 마이그레이션 리스크를 줄인다. --- app/models/Community.py | 4 ++-- app/routes/PolicyAdminRoute.py | 2 +- tests/test_policy_admin_batch.py | 21 +++++++++++++++++++++ 3 files changed, 24 insertions(+), 3 deletions(-) diff --git a/app/models/Community.py b/app/models/Community.py index 8123a24..9dc6d48 100644 --- a/app/models/Community.py +++ b/app/models/Community.py @@ -15,11 +15,11 @@ class Community(BaseEntity): Identity(), primary_key=True, ) - pin_id: Mapped[int] = mapped_column( + pin_id: Mapped[int | None] = mapped_column( "pin_id", BigInteger, ForeignKey("pin.pin_id"), - nullable=False, + nullable=True, ) community_type: Mapped[str] = mapped_column( "community_type", diff --git a/app/routes/PolicyAdminRoute.py b/app/routes/PolicyAdminRoute.py index d366318..7499404 100644 --- a/app/routes/PolicyAdminRoute.py +++ b/app/routes/PolicyAdminRoute.py @@ -80,8 +80,8 @@ async def sync_policy_pins( transform_limit: int | None = Query(default=None, ge=1, le=100, description="가공 최대 건수"), batch_size: int | None = Query(default=None, ge=1, le=25, description="배치 크기"), ) -> SuccessEnvelope[PolicySyncResult]: - validated_start, validated_end = _validate_optional_date_range(start_date, end_date) try: + validated_start, validated_end = _validate_optional_date_range(start_date, end_date) body = await service.sync_pipeline( ingest_service=ingest_service, s3_util=s3_util, diff --git a/tests/test_policy_admin_batch.py b/tests/test_policy_admin_batch.py index 169a433..6863fde 100644 --- a/tests/test_policy_admin_batch.py +++ b/tests/test_policy_admin_batch.py @@ -9,7 +9,9 @@ from zoneinfo import ZoneInfo import app.models # noqa: F401 — SQLAlchemy mapper registry +from fastapi import HTTPException +from app.routes.PolicyAdminRoute import sync_policy_pins from app.schemas.PolicyAdminDTO import ( PolicyBatchAction, PolicyBatchItemResult, @@ -50,6 +52,25 @@ def _make_ingest_service() -> PolicyEventIngestService: ) +class PolicyAdminRouteValidationTest(unittest.IsolatedAsyncioTestCase): + async def test_sync_invalid_date_returns_bad_request(self) -> None: + service = MagicMock() + service.sync_pipeline = AsyncMock() + + with self.assertRaises(HTTPException) as ctx: + await sync_policy_pins( + service=service, + ingest_service=MagicMock(), + s3_util=MagicMock(), + _admin_uid="admin-uid", + start_date="20261301", + end_date="20261302", + ) + + self.assertEqual(ctx.exception.status_code, 400) + service.sync_pipeline.assert_not_awaited() + + class PolicyPinSchedulerShouldRunSyncTest(unittest.TestCase): def test_runs_when_elapsed_within_one_hour_of_interval(self) -> None: scheduler = PolicyPinSchedulerService(s3_util=MagicMock()) From f1fc0a3b7094643b1c0602039668d34cfb254fbc Mon Sep 17 00:00:00 2001 From: selnem Date: Sat, 13 Jun 2026 00:03:19 +0900 Subject: [PATCH 7/8] =?UTF-8?q?fix:=20=EC=A0=95=EC=B1=85=20=ED=8C=8C?= =?UTF-8?q?=EC=9D=B4=ED=94=84=EB=9D=BC=EC=9D=B8=20N+1=20=EC=BF=BC=EB=A6=AC?= =?UTF-8?q?=C2=B7=EB=8F=99=EA=B8=B0=20I/O=20=EB=A6=AC=EB=B7=B0=20=EB=B0=98?= =?UTF-8?q?=EC=98=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handoff 적재와 sync_pipeline의 중복 DB 조회를 줄이고, 카드뉴스 파일 읽기·파이프라인 정리 작업을 스레드 풀에서 실행해 이벤트 루프 블로킹을 방지한다. --- app/services/PolicyEventIngestService.py | 5 ++-- app/services/PolicyPinService.py | 11 +++++++- app/services/policy_cardnews.py | 4 ++- tests/test_policy_admin_batch.py | 32 ++++++++++++++++++++++-- 4 files changed, 45 insertions(+), 7 deletions(-) diff --git a/app/services/PolicyEventIngestService.py b/app/services/PolicyEventIngestService.py index c1b82d4..caabe88 100644 --- a/app/services/PolicyEventIngestService.py +++ b/app/services/PolicyEventIngestService.py @@ -158,8 +158,7 @@ async def import_handoff_batch( if policy_api_id is None: errors.append({"row": row, "error": "policy_api_id 없음"}) continue - existing = await self._event_pin_repo.get_by_policy_api_id(policy_api_id) - if existing is not None: + if policy_api_id in db_ids: skipped_duplicate_count += 1 prune_policy_api_ids.add(policy_api_id) continue @@ -198,7 +197,7 @@ async def import_handoff_batch( await self._pin_repo.commit() if settings.policy_prune_pipeline_after_import and prune_policy_api_ids: - cleanup_after_policy_import(prune_policy_api_ids) + await to_thread.run_sync(cleanup_after_policy_import, prune_policy_api_ids) pending_import = 0 for row in handoff_rows: diff --git a/app/services/PolicyPinService.py b/app/services/PolicyPinService.py index 6e21ded..61fe13d 100644 --- a/app/services/PolicyPinService.py +++ b/app/services/PolicyPinService.py @@ -9,6 +9,7 @@ from app.clients.PolicyNewsClient import PolicyNewsClient from app.core.config import settings from app.schemas.PolicyAdminDTO import ( + PolicyBatchAction, PolicyBatchItemResult, PolicyImportBatchResult, PolicySyncResult, @@ -210,6 +211,13 @@ def _accumulate_import(import_batch: PolicyImportBatchResult) -> None: import_pin_ids.extend(import_batch.pin_ids) last_import = import_batch + db_ids = await ingest_service.get_imported_policy_api_ids() + + def _track_imported_ids(import_batch: PolicyImportBatchResult) -> None: + for item in import_batch.items: + if item.action == PolicyBatchAction.CREATED and item.policy_api_id is not None: + db_ids.add(item.policy_api_id) + while True: if transform_limit is not None and total_processed >= transform_limit: break @@ -218,7 +226,6 @@ def _accumulate_import(import_batch: PolicyImportBatchResult) -> None: if transform_limit is not None: batch_limit = min(effective_batch, transform_limit - total_processed) - db_ids = await ingest_service.get_imported_policy_api_ids() transform_batch = await self.transform_and_save( limit=batch_limit, s3_util=s3_util, @@ -239,6 +246,7 @@ def _accumulate_import(import_batch: PolicyImportBatchResult) -> None: limit=effective_batch, ) _accumulate_import(import_batch) + _track_imported_ids(import_batch) if transform_batch.processed_count < batch_limit: break @@ -250,6 +258,7 @@ def _accumulate_import(import_batch: PolicyImportBatchResult) -> None: limit=effective_batch, ) _accumulate_import(import_batch) + _track_imported_ids(import_batch) if import_batch.pending_import_count == 0: break if import_batch.inserted_count == 0 and import_batch.pending_import_count >= prev_pending: diff --git a/app/services/policy_cardnews.py b/app/services/policy_cardnews.py index 5d699a4..be070d7 100644 --- a/app/services/policy_cardnews.py +++ b/app/services/policy_cardnews.py @@ -4,6 +4,8 @@ from pathlib import Path from typing import Any, TypedDict +from anyio import to_thread + from app.core.config import settings from app.policy_cardnews import ( parse_cardnews_slides_json, @@ -89,7 +91,7 @@ async def _upload_local_handoff_path( slide_no = int(slide_name.split("_", 1)[1]) except ValueError: slide_no = 1 - image_bytes = local_path.read_bytes() + image_bytes = await to_thread.run_sync(local_path.read_bytes) return await _upload_slide_bytes( s3_util, content_id=content_id, diff --git a/tests/test_policy_admin_batch.py b/tests/test_policy_admin_batch.py index 6863fde..39f746b 100644 --- a/tests/test_policy_admin_batch.py +++ b/tests/test_policy_admin_batch.py @@ -216,6 +216,33 @@ async def test_get_imported_policy_api_ids_delegates_to_repo(self) -> None: self.assertEqual(result, {1, 2}) service._event_pin_repo.list_policy_api_ids.assert_awaited_once() + async def test_import_handoff_batch_skips_per_row_db_lookup(self) -> None: + service = _make_ingest_service() + service._event_pin_repo.get_by_policy_api_id = AsyncMock(return_value=None) + handoff_rows = [ + { + "contentid": "1", + "policy_api_id": 1, + "title": "정책1", + "pin_content": "본문1", + "cardnews_images": [{"key": "k1", "url": "https://cdn.example/1.png"}], + }, + ] + with ( + patch( + "app.services.PolicyEventIngestService.load_jsonl_rows", + return_value=handoff_rows, + ), + patch( + "app.services.PolicyEventIngestService.settings.policy_prune_pipeline_after_import", + False, + ), + patch.object(service, "_insert_policy_pin", AsyncMock(return_value=101)), + ): + await service.import_handoff_batch(import_all=True) + + service._event_pin_repo.get_by_policy_api_id.assert_not_awaited() + async def test_nested_failure_does_not_expunge_unrelated_session_state(self) -> None: service = _make_ingest_service() nested_ctx = MagicMock() @@ -255,7 +282,7 @@ class PolicySyncPipelineAccumulationTest(unittest.IsolatedAsyncioTestCase): async def test_import_results_accumulate_across_batches(self) -> None: service = PolicyPinService() ingest = _make_ingest_service() - ingest._event_pin_repo.list_policy_api_ids = AsyncMock(return_value=set()) + ingest.get_imported_policy_api_ids = AsyncMock(return_value=set()) transform_batches = [ PolicyPinTransformResult( @@ -345,11 +372,12 @@ async def test_import_results_accumulate_across_batches(self) -> None: self.assertEqual(len(import_result.items), 3) self.assertEqual(import_result.pin_ids, [11, 12, 13]) self.assertEqual(import_result.errors, [{"policy_api_id": 1, "error": "e1"}]) + ingest.get_imported_policy_api_ids.assert_awaited_once() async def test_import_drains_backlog_when_transform_produces_zero(self) -> None: service = PolicyPinService() ingest = _make_ingest_service() - ingest._event_pin_repo.list_policy_api_ids = AsyncMock(return_value=set()) + ingest.get_imported_policy_api_ids = AsyncMock(return_value=set()) import_batches = [ PolicyImportBatchResult( From 5f4badeb0f15adf03cde6e760fd089e39d6381b7 Mon Sep 17 00:00:00 2001 From: selnem Date: Sat, 13 Jun 2026 00:35:11 +0900 Subject: [PATCH 8/8] =?UTF-8?q?fix:=20=EC=A0=95=EC=B1=85=20sync=20?= =?UTF-8?q?=EB=A3=A8=ED=94=84=C2=B7handoff=20=EB=B9=84=EC=96=B4=20?= =?UTF-8?q?=EC=9E=88=EC=9D=84=20=EB=95=8C=20DB=20=EC=99=84=EB=A3=8C=20?= =?UTF-8?q?=EC=83=81=ED=83=9C=20=EC=9D=B8=EC=8B=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sync가 remaining_pending 기준으로 배치를 끝까지 돌고, import 후 캐시 정리로 handoff가 비어도 DB 적재 완료로 처리하도록 status·hint를 보강한다. --- app/schemas/PolicyAdminDTO.py | 6 + app/schemas/PolicyPinDTO.py | 4 + app/services/PolicyEventIngestService.py | 36 +++- app/services/PolicyPinService.py | 45 +++-- app/services/policy_pin_transform.py | 6 + tests/test_policy_admin_batch.py | 214 ++++++++++++++++++++++- 6 files changed, 294 insertions(+), 17 deletions(-) diff --git a/app/schemas/PolicyAdminDTO.py b/app/schemas/PolicyAdminDTO.py index aeea162..535a5d7 100644 --- a/app/schemas/PolicyAdminDTO.py +++ b/app/schemas/PolicyAdminDTO.py @@ -30,6 +30,7 @@ class PolicyImportBatchResult(BaseModel): items: list[PolicyBatchItemResult] = Field(default_factory=list) pin_ids: list[int] = Field(default_factory=list) requested_batch_size: int | None = None + hint: str | None = None class PolicyTransformBatchResult(PolicyPinTransformResult): @@ -46,6 +47,11 @@ class PolicyPipelineStatusResult(BaseModel): db_policy_count: int pending_transform_count: int pending_import_count: int + is_caught_up: bool = Field( + default=False, + description="미가공·미적재 건이 없어 DB와 파이프라인이 동기화된 상태", + ) + hint: str | None = None class PolicySyncResult(BaseModel): diff --git a/app/schemas/PolicyPinDTO.py b/app/schemas/PolicyPinDTO.py index 6568388..120d566 100644 --- a/app/schemas/PolicyPinDTO.py +++ b/app/schemas/PolicyPinDTO.py @@ -90,6 +90,10 @@ class PolicyPinTransformResult(BaseModel): hint: str | None = None skipped_duplicate_count: int = 0 pending_count: int = 0 + remaining_pending_count: int = Field( + default=0, + description="가공 후에도 handoff·DB에 없는 원문 건수", + ) class PolicyPinHandoffResult(BaseModel): diff --git a/app/services/PolicyEventIngestService.py b/app/services/PolicyEventIngestService.py index caabe88..6510653 100644 --- a/app/services/PolicyEventIngestService.py +++ b/app/services/PolicyEventIngestService.py @@ -116,13 +116,29 @@ async def import_handoff_batch( limit: int | None = None, ) -> PolicyImportBatchResult: handoff_rows = await to_thread.run_sync(load_jsonl_rows, POLICY_HANDOFF_PATH) + db_ids = await self.get_imported_policy_api_ids() if not handoff_rows: - raise FileNotFoundError( - f"핸드오프 JSONL 없음: {POLICY_HANDOFF_PATH}. transform을 먼저 실행하세요.", + documents = await to_thread.run_sync(load_jsonl_rows, POLICY_DOCUMENTS_PATH) + pending_transform = count_pending_transform(documents, {}, db_policy_api_ids=db_ids) + if pending_transform > 0: + raise FileNotFoundError( + f"핸드오프 JSONL 없음: {POLICY_HANDOFF_PATH}. " + f"transform을 먼저 실행하세요. (미가공 {pending_transform}건)", + ) + effective_batch = None if import_all else limit + return PolicyImportBatchResult( + inserted_count=0, + skipped_duplicate_count=0, + pending_import_count=0, + error_count=0, + requested_batch_size=effective_batch, + hint=( + f"DB에 policy 핀 {len(db_ids)}건이 있어 적재 완료 상태입니다. " + "handoff JSONL이 비어 있는 것은 import 후 캐시 정리로 정상입니다." + ), ) uid = admin_uid or await self.resolve_admin_uid() - db_ids = await self.get_imported_policy_api_ids() inserted_count = 0 skipped_duplicate_count = 0 errors: list[dict[str, Any]] = [] @@ -297,6 +313,18 @@ async def get_pipeline_status(self) -> PolicyPipelineStatusResult: if policy_api_id is not None and policy_api_id not in db_ids: pending_import += 1 + is_caught_up = pending_transform == 0 and pending_import == 0 + hint: str | None = None + if is_caught_up and len(db_ids) > 0: + hint = ( + f"DB에 policy 핀 {len(db_ids)}건 반영 완료. " + "handoff_count=0은 import 후 JSONL 캐시 정리로 정상입니다." + ) + elif pending_transform > 0: + hint = f"미가공 {pending_transform}건 — transform-batch 또는 sync를 실행하세요." + elif pending_import > 0: + hint = f"미적재 {pending_import}건 — import-batch 또는 sync를 실행하세요." + return PolicyPipelineStatusResult( query_start_date=meta.get("query_start_date"), query_end_date=meta.get("query_end_date"), @@ -306,6 +334,8 @@ async def get_pipeline_status(self) -> PolicyPipelineStatusResult: db_policy_count=len(db_ids), pending_transform_count=pending_transform, pending_import_count=pending_import, + is_caught_up=is_caught_up, + hint=hint, ) @staticmethod diff --git a/app/services/PolicyPinService.py b/app/services/PolicyPinService.py index 61fe13d..f1ab01e 100644 --- a/app/services/PolicyPinService.py +++ b/app/services/PolicyPinService.py @@ -27,6 +27,7 @@ from app.services.policy_pin_transform import ( POLICY_DOCUMENTS_PATH, POLICY_HANDOFF_PATH, + load_jsonl_rows, load_rows_by_content_id, merge_documents, transform_documents_jsonl, @@ -236,22 +237,22 @@ def _track_imported_ids(import_batch: PolicyImportBatchResult) -> None: total_skipped_transform += transform_batch.skipped_duplicate_count transform_errors.extend(transform_batch.errors) transform_pins.extend(transform_batch.pins) - last_transform_pending = transform_batch.pending_count + last_transform_pending = transform_batch.remaining_pending_count - if transform_batch.processed_count == 0: - break - - import_batch = await ingest_service.import_handoff_batch( - import_all=False, - limit=effective_batch, - ) - _accumulate_import(import_batch) - _track_imported_ids(import_batch) + if transform_batch.processed_count > 0: + import_batch = await ingest_service.import_handoff_batch( + import_all=False, + limit=effective_batch, + ) + _accumulate_import(import_batch) + _track_imported_ids(import_batch) - if transform_batch.processed_count < batch_limit: + if transform_batch.remaining_pending_count == 0: + break + if transform_batch.processed_count == 0: break - while True: + while load_jsonl_rows(self.handoff_path()): prev_pending = last_import.pending_import_count import_batch = await ingest_service.import_handoff_batch( import_all=False, @@ -299,6 +300,26 @@ def _track_imported_ids(import_batch: PolicyImportBatchResult) -> None: f"sync 완료: 수집 {search.count}건, 가공 {total_processed}건({batches_run}배치), " f"DB INSERT {total_imported}건." ) + if ( + total_processed == 0 + and total_imported == 0 + and last_transform_pending == 0 + and len(db_ids) > 0 + ): + hint = ( + f"이미 DB 반영 완료 (policy 핀 {len(db_ids)}건). " + "handoff JSONL이 비어 있는 것은 import 후 캐시 정리로 정상입니다." + ) + elif total_processed == 0 and last_transform_pending > 0: + hint = ( + f"수집 {search.count}건 저장됐으나 가공 0건입니다. " + f"미가공 {last_transform_pending}건 — GEMINI_API_KEY·transform 오류를 확인하세요." + ) + elif total_processed > 0 and last_transform_pending > 0: + hint = ( + f"sync 부분 완료: 가공 {total_processed}건, DB INSERT {total_imported}건. " + f"미가공 {last_transform_pending}건 남음 — /policy-admin/sync 재실행 또는 transform-batch." + ) return PolicySyncResult( query_start_date=start_date, query_end_date=end_date, diff --git a/app/services/policy_pin_transform.py b/app/services/policy_pin_transform.py index 019e8a4..ef2f4d3 100644 --- a/app/services/policy_pin_transform.py +++ b/app/services/policy_pin_transform.py @@ -255,6 +255,11 @@ async def _transform_row(row: dict[str, Any]) -> tuple[str, dict[str, Any] | Non skipped_duplicate_count += 1 write_handoff_map(handoff_by_id, dst) + remaining_pending = count_pending_transform( + documents, + handoff_by_id, + db_policy_api_ids=db_ids, + ) pins = [PolicyPinHandoffDTO.from_row(item) for item in processed_rows] hint: str | None @@ -276,4 +281,5 @@ async def _transform_row(row: dict[str, Any]) -> tuple[str, dict[str, Any] | Non hint=hint, skipped_duplicate_count=skipped_duplicate_count, pending_count=len(pending), + remaining_pending_count=remaining_pending, ) diff --git a/tests/test_policy_admin_batch.py b/tests/test_policy_admin_batch.py index 39f746b..40e8014 100644 --- a/tests/test_policy_admin_batch.py +++ b/tests/test_policy_admin_batch.py @@ -216,6 +216,36 @@ async def test_get_imported_policy_api_ids_delegates_to_repo(self) -> None: self.assertEqual(result, {1, 2}) service._event_pin_repo.list_policy_api_ids.assert_awaited_once() + async def test_import_handoff_batch_ok_when_handoff_empty_and_db_caught_up(self) -> None: + service = _make_ingest_service() + service._event_pin_repo.list_policy_api_ids = AsyncMock(return_value={1, 2}) + documents = [ + {"contentid": "1", "policy_api_id": 1, "pin_content": "a"}, + {"contentid": "2", "policy_api_id": 2, "pin_content": "b"}, + ] + + with patch( + "app.services.PolicyEventIngestService.load_jsonl_rows", + side_effect=[[], documents], + ): + result = await service.import_handoff_batch(import_all=False, limit=5) + + self.assertEqual(result.inserted_count, 0) + self.assertEqual(result.pending_import_count, 0) + self.assertIn("적재 완료", result.hint or "") + + async def test_import_handoff_batch_raises_when_handoff_empty_but_pending_transform(self) -> None: + service = _make_ingest_service() + service._event_pin_repo.list_policy_api_ids = AsyncMock(return_value=set()) + documents = [{"contentid": "1", "policy_api_id": 1, "pin_content": "a"}] + + with patch( + "app.services.PolicyEventIngestService.load_jsonl_rows", + side_effect=[[], documents], + ): + with self.assertRaises(FileNotFoundError): + await service.import_handoff_batch(import_all=False, limit=5) + async def test_import_handoff_batch_skips_per_row_db_lookup(self) -> None: service = _make_ingest_service() service._event_pin_repo.get_by_policy_api_id = AsyncMock(return_value=None) @@ -291,7 +321,8 @@ async def test_import_results_accumulate_across_batches(self) -> None: processed_count=2, error_count=0, pins=[], - pending_count=1, + pending_count=3, + remaining_pending_count=1, ), PolicyPinTransformResult( input_path="in.jsonl", @@ -299,7 +330,8 @@ async def test_import_results_accumulate_across_batches(self) -> None: processed_count=1, error_count=0, pins=[], - pending_count=0, + pending_count=1, + remaining_pending_count=0, ), ] import_batches = [ @@ -434,9 +466,15 @@ async def test_import_drains_backlog_when_transform_produces_zero(self) -> None: error_count=0, pins=[], pending_count=0, + remaining_pending_count=0, ), ), ), + patch.object(service, "handoff_path", return_value=Path("out.jsonl")), + patch( + "app.services.PolicyPinService.load_jsonl_rows", + return_value=[{"contentid": "1"}], + ), patch.object(ingest, "import_handoff_batch", import_mock), patch.object(PolicyEventIngestService, "write_sync_meta"), ): @@ -454,6 +492,178 @@ async def test_import_drains_backlog_when_transform_produces_zero(self) -> None: self.assertEqual(import_result.pending_import_count, 0) self.assertEqual(import_result.pin_ids, [11, 12, 13]) + async def test_transform_loop_continues_while_remaining_pending(self) -> None: + service = PolicyPinService() + ingest = _make_ingest_service() + ingest.get_imported_policy_api_ids = AsyncMock(return_value=set()) + transform_mock = AsyncMock( + side_effect=[ + PolicyPinTransformResult( + input_path="in.jsonl", + output_path="out.jsonl", + processed_count=20, + error_count=5, + pins=[], + pending_count=25, + remaining_pending_count=22, + ), + PolicyPinTransformResult( + input_path="in.jsonl", + output_path="out.jsonl", + processed_count=22, + error_count=0, + pins=[], + pending_count=22, + remaining_pending_count=0, + ), + ], + ) + import_mock = AsyncMock( + return_value=PolicyImportBatchResult( + inserted_count=1, + skipped_duplicate_count=0, + pending_import_count=0, + error_count=0, + ), + ) + + with ( + patch( + "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import", + False, + ), + patch.object( + service, + "search_and_save", + AsyncMock( + return_value=PolicyPinSearchResult( + query_start_date="20260610", + query_end_date="20260612", + count=42, + pins=[], + saved_documents_path="docs.jsonl", + stats={}, + ), + ), + ), + patch.object(service, "transform_and_save", transform_mock), + patch.object(ingest, "import_handoff_batch", import_mock), + patch.object(PolicyEventIngestService, "write_sync_meta"), + patch.object(service, "handoff_path", return_value=Path("out.jsonl")), + patch("app.services.PolicyPinService.load_jsonl_rows", return_value=[]), + ): + result = await service.sync_pipeline( + ingest_service=ingest, + s3_util=MagicMock(), + batch_size=25, + ) + + self.assertEqual(transform_mock.await_count, 2) + self.assertEqual(result.transform.processed_count, 42) + + async def test_sync_completes_when_transform_produces_zero_and_handoff_empty(self) -> None: + service = PolicyPinService() + ingest = _make_ingest_service() + ingest.get_imported_policy_api_ids = AsyncMock(return_value=set()) + import_mock = AsyncMock() + + with ( + patch( + "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import", + False, + ), + patch.object( + service, + "search_and_save", + AsyncMock( + return_value=PolicyPinSearchResult( + query_start_date="20260610", + query_end_date="20260612", + count=42, + pins=[], + saved_documents_path="docs.jsonl", + stats={}, + ), + ), + ), + patch.object( + service, + "transform_and_save", + AsyncMock( + return_value=PolicyPinTransformResult( + input_path="in.jsonl", + output_path="out.jsonl", + processed_count=0, + error_count=42, + pins=[], + pending_count=42, + remaining_pending_count=42, + ), + ), + ), + patch.object(service, "handoff_path", return_value=Path("out.jsonl")), + patch("app.services.PolicyPinService.load_jsonl_rows", return_value=[]), + patch.object(ingest, "import_handoff_batch", import_mock), + patch.object(PolicyEventIngestService, "write_sync_meta"), + ): + result = await service.sync_pipeline( + ingest_service=ingest, + s3_util=MagicMock(), + ) + + import_mock.assert_not_awaited() + self.assertIn("가공 0건", result.hint or "") + self.assertEqual(result.import_result.inserted_count, 0) + + async def test_sync_hint_when_already_in_db(self) -> None: + service = PolicyPinService() + ingest = _make_ingest_service() + ingest.get_imported_policy_api_ids = AsyncMock(return_value={1, 2, 3}) + + with ( + patch( + "app.services.PolicyPinService.settings.policy_prune_pipeline_after_import", + False, + ), + patch.object( + service, + "search_and_save", + AsyncMock( + return_value=PolicyPinSearchResult( + query_start_date="20260610", + query_end_date="20260612", + count=3, + pins=[], + saved_documents_path="docs.jsonl", + stats={}, + ), + ), + ), + patch.object( + service, + "transform_and_save", + AsyncMock( + return_value=PolicyPinTransformResult( + input_path="in.jsonl", + output_path="out.jsonl", + processed_count=0, + error_count=0, + pins=[], + remaining_pending_count=0, + ), + ), + ), + patch.object(service, "handoff_path", return_value=Path("out.jsonl")), + patch("app.services.PolicyPinService.load_jsonl_rows", return_value=[]), + patch.object(PolicyEventIngestService, "write_sync_meta"), + ): + result = await service.sync_pipeline( + ingest_service=ingest, + s3_util=MagicMock(), + ) + + self.assertIn("이미 DB 반영 완료", result.hint or "") + if __name__ == "__main__": unittest.main()