Skip to content

[FEAT] AI 핀 생성·축제 적재 후 Redis GEO 캐시 동기화 #136

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
selnem merged 2 commits into from
Jun 18, 2026
Merged

[FEAT] AI 핀 생성·축제 적재 후 Redis GEO 캐시 동기화 #136

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e8d556c
feat: AI 핀 생성·축제 적재 후 Redis GEO 캐시 동기화
selnem Jun 18, 2026
621efc2
fix: 이슈 핀 GEO 캐시 동기화를 best-effort로 처리
selnem Jun 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions app/core/deps.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from __future__ import annotations

from typing import Annotated

import httpx
Expand Down Expand Up @@ -27,6 +29,7 @@
from app.services.IssueService import IssueService
from app.services.internal.IssuePinDailyRateLimitService import IssuePinDailyRateLimitService
from app.services.internal.IssuePinBackgroundRunner import IssuePinBackgroundRunner
from app.services.internal.map.PinGeoRedisPublisher import PinGeoRedisPublisher
from app.services.UserService import UserService
from app.services.ComplaintEmailService import ComplaintEmailService
from app.services.ContestPinService import ContestPinService
Expand Down Expand Up @@ -325,6 +328,8 @@ def get_issue_service(
s3_util: S3UtilDep,
background_runner: IssuePinBackgroundRunnerDep,
issue_pin_daily_rate_limit_service: IssuePinDailyRateLimitServiceDep,
location_repo: LocationRepoDep,
pin_geo_redis_publisher: PinGeoRedisPublisherDep,
) -> IssueService:
return IssueService(
vector_store_service=vector_store_service,
Expand All @@ -341,6 +346,8 @@ def get_issue_service(
s3_util=s3_util,
background_runner=background_runner,
issue_pin_daily_rate_limit_service=issue_pin_daily_rate_limit_service,
location_repo=location_repo,
pin_geo_redis_publisher=pin_geo_redis_publisher,
)


Expand All @@ -356,6 +363,18 @@ def get_async_redis_client(request: Request) -> AsyncRedis:

AsyncRedisDep = Annotated[AsyncRedis, Depends(get_async_redis_client)]


def get_pin_geo_redis_publisher(
redis_client: AsyncRedisDep,
) -> PinGeoRedisPublisher:
return PinGeoRedisPublisher(redis_client)


PinGeoRedisPublisherDep = Annotated[
PinGeoRedisPublisher,
Depends(get_pin_geo_redis_publisher),
]

CurrentUserIdDep = Annotated[str, Depends(get_current_user_id)]
OptionalUserIdDep = Annotated[str | None, Depends(get_optional_user_id)]

Expand Down Expand Up @@ -420,13 +439,17 @@ def get_festival_event_ingest_service(
pin_location_repo: PinLocationRepoDep,
pin_image_repo: PinImageRepoDep,
location_resolve_client: LocationResolveClientDep,
location_repo: LocationRepoDep,
pin_geo_redis_publisher: PinGeoRedisPublisherDep,
) -> FestivalEventIngestService:
return FestivalEventIngestService(
pin_repo=pin_repo,
event_pin_repo=event_pin_repo,
pin_location_repo=pin_location_repo,
pin_image_repo=pin_image_repo,
location_resolve_client=location_resolve_client,
location_repo=location_repo,
pin_geo_redis_publisher=pin_geo_redis_publisher,
)


Expand Down
6 changes: 6 additions & 0 deletions app/repositories/LocationRepo.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,12 @@ async def get_all_ordered(self) -> list[Location]:
)
return list(result.scalars().all())

async def get_region_by_id(self, location_id: int) -> str | None:
location = await self.get_by_id(location_id)
if location is None:
return None
return location.region

async def get_target_petition_by_location_id(self, *, location_id: int) -> int | None:
result = await self.session.execute(
select(PopulationDensity.target_petition)
Expand Down
57 changes: 51 additions & 6 deletions app/services/FestivalEventIngestService.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
from app.models.enum.PinType import PinType
from app.models.enum.ToneType import ToneType
from app.repositories.EventPinRepo import EventPinRepo
from app.repositories.LocationRepo import LocationRepo
from app.repositories.PinImageRepo import PinImageRepo
from app.repositories.PinLocationRepo import PinLocationRepo
from app.repositories.PinRepo import PinRepo
Expand Down Expand Up @@ -47,6 +48,8 @@
)
from app.services.internal.geo.LocationResolveClient import LocationResolveClient
from app.services.internal.geo.location_resolve_fields import resolve_pin_location_fields
from app.services.internal.map.PinGeoRedisPublisher import PinGeoCacheItem, PinGeoRedisPublisher
from app.utils.geo import wgs84_from_pin_point
from app.utils.festival_date_filter import current_year_festival_range, festival_overlaps_range
from app.utils.visitkorea_area import area_display_name, resolve_row_area_code, row_matches_area_filter
from rag.scripts.chunk_module import write_jsonl
Expand Down Expand Up @@ -85,12 +88,16 @@ def __init__(
pin_location_repo: PinLocationRepo,
pin_image_repo: PinImageRepo,
location_resolve_client: LocationResolveClient,
location_repo: LocationRepo,
pin_geo_redis_publisher: PinGeoRedisPublisher,
) -> None:
self._pin_repo = pin_repo
self._event_pin_repo = event_pin_repo
self._pin_location_repo = pin_location_repo
self._pin_image_repo = pin_image_repo
self._location_resolve_client = location_resolve_client
self._location_repo = location_repo
self._pin_geo_redis_publisher = pin_geo_redis_publisher

async def commit(self) -> None:
await self._pin_repo.commit()
Expand Down Expand Up @@ -369,6 +376,7 @@ async def _import_handoff_rows(
errors: list[dict[str, Any]] = []
items: list[FestivalBatchItemResult] = []
pin_ids: list[int] = []
geo_cache_items: list[PinGeoCacheItem] = []

pending_rows: list[dict[str, Any]] = []
for row in handoff_rows:
Expand Down Expand Up @@ -403,14 +411,15 @@ async def _import_handoff_rows(
try:
async with self._pin_repo.session.begin_nested():
if existing is None:
pin_id = await self._insert_festival_pin(
pin_id, cache_item = await self._insert_festival_pin(
admin_uid=admin_uid,
row=row,
festival_api_id=festival_api_id,
)
inserted_count += 1
db_ids.add(festival_api_id)
pin_ids.append(pin_id)
geo_cache_items.append(cache_item)
items.append(
FestivalBatchItemResult(
festival_api_id=festival_api_id,
Expand All @@ -419,9 +428,10 @@ async def _import_handoff_rows(
),
)
elif allow_update:
pin_id = await self._update_festival_pin(existing=existing, row=row)
pin_id, cache_item = await self._update_festival_pin(existing=existing, row=row)
updated_count += 1
pin_ids.append(pin_id)
geo_cache_items.append(cache_item)
items.append(
FestivalBatchItemResult(
festival_api_id=festival_api_id,
Expand Down Expand Up @@ -449,6 +459,7 @@ async def _import_handoff_rows(
)

await self._pin_repo.commit()
await self._pin_geo_redis_publisher.add_pins_batch(geo_cache_items)

pending_import = 0
for row in handoff_rows:
Expand Down Expand Up @@ -561,7 +572,7 @@ async def _insert_festival_pin(
admin_uid: str,
row: dict[str, Any],
festival_api_id: int,
) -> int:
) -> tuple[int, PinGeoCacheItem]:
location_fields = await self._resolve_location(row)
if location_fields is None:
raise ValueError("위치 resolve 실패")
Expand Down Expand Up @@ -596,9 +607,17 @@ async def _insert_festival_pin(
await self._pin_location_repo.save(pin_location, flush_immediately=True)

await self._replace_pin_images(pin_id=pin.pin_id, row=row)
return int(pin.pin_id)
cache_item = await self._build_festival_geo_cache_item(
pin_id=int(pin.pin_id),
location_id=location_id,
detail_address=detail_address,
pin_point=pin_point,
)
return int(pin.pin_id), cache_item

async def _update_festival_pin(self, *, existing: EventPin, row: dict[str, Any]) -> int:
async def _update_festival_pin(
self, *, existing: EventPin, row: dict[str, Any]
) -> tuple[int, PinGeoCacheItem]:
pin = existing.pin
if pin is None:
raise ValueError("연결된 pin 없음")
Expand Down Expand Up @@ -633,7 +652,33 @@ async def _update_festival_pin(self, *, existing: EventPin, row: dict[str, Any])

await self._pin_image_repo.delete_by_pin_id(pin.pin_id)
await self._replace_pin_images(pin_id=pin.pin_id, row=row)
return int(pin.pin_id)
cache_item = await self._build_festival_geo_cache_item(
pin_id=int(pin.pin_id),
location_id=location_id,
detail_address=detail_address,
pin_point=pin_point,
)
return int(pin.pin_id), cache_item

async def _build_festival_geo_cache_item(
self,
*,
pin_id: int,
location_id: int,
detail_address: str,
pin_point,
) -> PinGeoCacheItem:
Comment on lines +663 to +670

@gemini-code-assist gemini-code-assist Bot Jun 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

_build_festival_geo_cache_item 메서드의 pin_point 매개변수에 타입 어노테이션이 누락되어 있습니다. 코드의 일관성과 정적 타입 검사(MyPy 등)의 정확성을 높이기 위해 Any 또는 적절한 기하학 타입을 명시해 주는 것이 좋습니다. 이미 파일 상단에 typing.Any가 임포트되어 있으므로 이를 활용할 수 있습니다.

    async def _build_festival_geo_cache_item(\n        self,\n        *,\n        pin_id: int,\n        location_id: int,\n        detail_address: str,\n        pin_point: Any,\n    ) -> PinGeoCacheItem:

lat, lng = wgs84_from_pin_point(pin_point)
region = await self._location_repo.get_region_by_id(location_id) or ""
return PinGeoCacheItem(
pin_id=pin_id,
pin_type=PinType.FESTIVAL.name,
lat=lat,
lng=lng,
detail_address=detail_address,
region=region,
discount=None,
)

async def _replace_pin_images(self, *, pin_id: int, row: dict[str, Any]) -> None:
for spec in pin_images_for_db_row(row):
Expand Down
46 changes: 46 additions & 0 deletions app/services/IssueService.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
from app.models.enum.ToneType import ToneType
from app.repositories.CommunityRepo import CommunityRepo
from app.repositories.IssuePinRepo import IssuePinRepo
from app.repositories.LocationRepo import LocationRepo
from app.repositories.PinLikeRepo import PinLikeRepo
from app.repositories.PinImageRepo import PinImageRepo
from app.repositories.PinLocationRepo import PinLocationRepo
Expand All @@ -53,6 +54,7 @@
RateLimitSnapshot,
)
from app.services.internal.IssuePinBackgroundRunner import IssuePinBackgroundRunner
from app.services.internal.map.PinGeoRedisPublisher import PinGeoRedisPublisher
from app.services.internal.ai.IssuePinLLMService import IssuePinLLMService
from app.services.internal.ai.IssueRagPlannerService import IssueRagPlannerService
from app.services.internal.geo.LocationResolveClient import LocationResolveClient
Expand Down Expand Up @@ -104,6 +106,8 @@ def __init__(
s3_util: S3Util,
background_runner: IssuePinBackgroundRunner,
issue_pin_daily_rate_limit_service: IssuePinDailyRateLimitService,
location_repo: LocationRepo,
pin_geo_redis_publisher: PinGeoRedisPublisher,
) -> None:
self._vector_store_service = vector_store_service
self._issue_rag_planner_service = issue_rag_planner_service
Expand All @@ -119,6 +123,38 @@ def __init__(
self._user_repo = user_repo
self._background_runner = background_runner
self._rate_limit = issue_pin_daily_rate_limit_service
self._location_repo = location_repo
self._pin_geo_redis_publisher = pin_geo_redis_publisher

async def _sync_issue_pin_geo_cache(
self,
*,
pin_id: int,
pin_location: PinLocation,
) -> None:
"""commit 성공 후 Redis GEO 캐시를 best-effort로 동기화한다.

DB 커밋 이후에만 호출해야 한다. 실패해도 API 응답은 성공으로 유지하고
warning 로그만 남긴다(BE 야간 배치로 캐시 정합성 복구 가능).
"""
try:
lat, lng = wgs84_from_pin_point(pin_location.pin_point)
region = await self._location_repo.get_region_by_id(pin_location.location_id) or ""
await self._pin_geo_redis_publisher.add_pin(
pin_id=pin_id,
pin_type=PinType.ISSUE.name,
lat=lat,
lng=lng,
detail_address=pin_location.detail_address or "",
region=region,
discount=None,
)
except Exception:
logger.warning(
"issue pin geo cache sync failed pin_id=%s",
pin_id,
exc_info=True,
)

async def _build_rate_limit_quota_dict(
self,
Expand Down Expand Up @@ -753,6 +789,11 @@ async def create_issue_pin(
logger.exception("issue pin create commit failed uid=%s", uid)
raise_business_exception(ErrorCode.ISSUE_PIN_IMPORT_FAILED)

await self._sync_issue_pin_geo_cache(
pin_id=pin.pin_id,
pin_location=pin_location,
)

snapshot = await self._rate_limit.record_daily_quota_success(
RateLimitKind.CREATE,
uid=uid,
Expand Down Expand Up @@ -872,6 +913,11 @@ async def update_issue_pin(
logger.exception("issue pin update failed pin_id=%s", pin.pin_id)
raise_business_exception(ErrorCode.ISSUE_PIN_EDIT_FAILED)

await self._sync_issue_pin_geo_cache(
pin_id=pin.pin_id,
pin_location=pin_location,
)

snapshot = await self._rate_limit.record_daily_quota_success(
RateLimitKind.EDIT,
uid=uid,
Expand Down
Loading
Loading