castad/o2o-castad-backend
9
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(crawling): 업체 공식 링크 수집·저장 및 official_site_url 응답 연동

Browse Source 
- 네이버 placeDetail GraphQL base에 homepages 필드 추가, 대표(repr)
  링크 우선으로 살아있는 URL을 추출한다 (실서비스 응답 구조 확인 완료)
- marketing 테이블에 official_site_url 컬럼 추가: 크롤링 시 플레이스
  홈페이지 항목을 저장하고, 없으면 유저가 입력한 크롤링 소스 URL을 저장
- GET /video/{id}, /video/all의 official_site_url은 저장값 우선,
  컬럼 도입 전 기존 행은 place_id 기반 플레이스 URL로 폴백
- 기존 테이블에는 수동 마이그레이션 필요:
  ALTER TABLE marketing ADD COLUMN official_site_url VARCHAR(2048) NULL
  COMMENT '업체 공식 링크' AFTER place_id;

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVSeUT2pEmeHc6hG6gVXUS
This commit is contained in:
hbyang 2026-08-21 10:49:46 +09:00
parent 61bdc27cd3
commit 1ef8e82638
5 changed files with 47 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
app/home/api/routers/v1/home.py
View File

@ -417,8 +417,16 @@ async def _crawling_logic(url: str, session: AsyncSession):
)
# Step 4-3: 분석 결과 DB 저장 (industry는 Project로 흐르므로 여기엔 미저장)
# 공식 링크: 플레이스 홈페이지 항목 우선, 없으면 유저가 입력한 크롤링 소스 URL
# (컬럼 길이를 넘는 긴 검색 URL은 place_id 기반 표준 플레이스 URL로 대체)
official_site_url = scraper.official_site_url or url
if len(official_site_url) > 2048:
official_site_url = (
f"https://map.naver.com/p/entry/place/{scraper.place_id[2:]}"
)
marketing_intel = MarketingIntel(
place_id=scraper.place_id,
official_site_url=official_site_url,
intel_result=marketing_analysis.model_dump(),
)
session.add(marketing_intel)

7
app/home/models.py
View File

@ -274,6 +274,7 @@ class MarketingIntel(Base):
Attributes:
id: 고유 식별자 (자동 증가)
place_id : 데이터 소스별 식별자
official_site_url : 업체 공식 링크 (플레이스 홈페이지 항목, 없으면 크롤링 소스 URL)
intel_result : 마케팅 분석 결과물 json
created_at: 생성 일시 (자동 설정)
"""
@ -302,6 +303,12 @@ class MarketingIntel(Base):
comment="매장 소스별 고유 식별자 (네이버 크롤링 시 'nv{id}' 형식; 직접 입력 시 NULL)",
)
official_site_url: Mapped[Optional[str]] = mapped_column(
String(2048),
nullable=True,
comment="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 크롤링 소스 URL; 직접 입력 시 NULL)",
)
intel_result : Mapped[dict[str, Any]] = mapped_column(
JSON,
nullable=False,

20
app/utils/nvMapScraper.py
View File

@ -56,6 +56,10 @@ query getAccommodation($id: String!, $deviceType: String) {
microReviews
conveniences
visitorReviewsTotal
homepages {
repr { url landingUrl isDeadUrl type }
etc { url landingUrl isDeadUrl type }
}
}
menus {
name
@ -112,6 +116,7 @@ query getVisitorReviewStats($id: String!) {
self.facility_info: str | None = None
self.voted_keyword_stats: list[dict] | None = None # 키워드 투표 집계 (displayName, count)
self.menu_info: list[dict] | None = None # 메뉴 목록 (name, price, description, recommend)
self.official_site_url: str | None = None # 업체 공식 링크 (base.homepages 대표 URL)
def _get_request_headers(self) -> dict:
headers = self.DEFAULT_HEADERS.copy()
@ -297,9 +302,24 @@ query getVisitorReviewStats($id: String!) {
self.facility_info = fac_data
self.voted_keyword_stats = stats_data
self.menu_info = business.get("menus") or None
self.official_site_url = self._extract_official_site_url(self.base_info)
return
@staticmethod
def _extract_official_site_url(base_info: dict | None) -> str | None:
"""base.homepages에서 살아있는 링크 하나를 고른다 (대표 repr 우선, 그 다음 etc 순서).
네이버 플레이스의 홈페이지 항목은 자체 홈페이지 외에 인스타그램/블로그
등일 수도 있다 — 업체가 대표로 등록한 링크를 그대로 신뢰한다.
"""
homepages = (base_info or {}).get("homepages") or {}
candidates = [homepages.get("repr"), *(homepages.get("etc") or [])]
for item in candidates:
if item and item.get("url") and not item.get("isDeadUrl"):
return item["url"]
return None
async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[dict], list[dict]]:
"""직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다.

18
app/video/api/routers/v1/video.py
View File

@ -90,7 +90,8 @@ async def _get_official_site_urls(
"""프로젝트 목록에 대해 {project_id: 공식 페이지 URL(or None)}을 일괄 조회한다.
Project.marketing_intelligence(문자열로 저장된 MarketingIntel.id)를 경유해
place_id를 찾고, 이를 네이버 플레이스 URL로 변환한다.
저장된 official_site_url을 우선 사용하고, 컬럼 도입 전 기존 행은
place_id 기반 네이버 플레이스 URL로 폴백한다.
"""
m_id_by_project: dict[int, int] = {}
for p in projects:
@ -106,17 +107,18 @@ async def _get_official_site_urls(
rows = (
await session.execute(
select(MarketingIntel.id, MarketingIntel.place_id).where(
MarketingIntel.id.in_(set(m_id_by_project.values()))
)
select(
MarketingIntel.id,
MarketingIntel.place_id,
MarketingIntel.official_site_url,
).where(MarketingIntel.id.in_(set(m_id_by_project.values())))
)
).all()
place_id_by_m_id = {m_id: place_id for m_id, place_id in rows}
intel_by_m_id = {m_id: (place_id, site_url) for m_id, place_id, site_url in rows}
for project_id, m_id in m_id_by_project.items():
url_by_project[project_id] = _place_id_to_site_url(
place_id_by_m_id.get(m_id)
)
place_id, site_url = intel_by_m_id.get(m_id, (None, None))
url_by_project[project_id] = site_url or _place_id_to_site_url(place_id)
return url_by_project

4
app/video/schemas/video_schema.py
View File

@ -188,7 +188,7 @@ class VideoThumbnailItem(BaseModel):
comment_count: int = Field(..., description="댓글 수 (대댓글 포함)")
official_site_url: Optional[str] = Field(
None,
description="업체 공식 페이지 URL (크롤링된 네이버 플레이스 기준, 없으면 null)",
description="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 크롤링 소스 URL; 직접 입력 생성 영상만 null)",
)
@ -211,7 +211,7 @@ class VideoDetailResponse(BaseModel):
is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지 (비로그인은 항상 false)")
official_site_url: Optional[str] = Field(
None,
description="업체 공식 페이지 URL (크롤링된 네이버 플레이스 기준, 없으면 null)",
description="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 크롤링 소스 URL; 직접 입력 생성 영상만 null)",
)