--- name: naver-real-estate-search description: Search, compare, and monitor 대한민국 property listings from 네이버 부동산 with natural-language queries. Use when the user wants 강남 아파트 전세 시세 찾기, 특정 지역 매매/전세/월세 비교, 조건에 맞는 매물 리스트 정리, 단지 후보 찾기, 여러 단지 비교 리포트, 자연어 채팅형 부동산 브리핑, or 목표가/새 매물/가격하락 감시 초안. Supports Korean property tasks such as apartment/빌라 listing summaries, 지역명/단지명 기반 단지 후보 탐색, same-area comparison, candidate seed/candidate cache workflows, and stdout/JSON results that can be connected to Telegram or higher-level briefings. Prefer direct 단지 URL or complex ID first when rate-limited; otherwise use the natural-language wrapper and narrow to 1~3 candidate complexes before broad scans. --- # Naver Real Estate Search 네이버 부동산 기반의 **대한민국 부동산 매물 검색 / 단지 후보 탐색 / 단지 비교 / 채팅형 브리핑 / 가격 감시** 스킬이다. 핵심 원칙: - **단일 단지 우선**: URL/complex ID가 있으면 그걸 먼저 쓴다. - **후보는 좁게**: 지역 전체를 넓게 긁기보다 후보 단지를 1~3개 먼저 좁힌다. - **자연어 우선**: “잠실 리센츠 전세 30평대”, “은마와 래미안대치팰리스 비교”처럼 바로 넣는다. - **한국어 브리핑 우선**: 채팅 표면에는 짧은 한국어 해석을 먼저 주고, 구조화 데이터가 필요할 때 JSON을 붙인다. - **429 완화**: 짧은 백오프 후에도 막히면 direct URL/ID 기반 재조회 쪽으로 유도한다. ## Source dependency 이 스킬은 로컬 upstream clone을 래핑한다. - `tmp/naverland-scrapper` 재사용하는 주요 로직: - `src.core.parser.NaverURLParser` - `src.core.services.response_capture.normalize_article_payload` - `src.utils.helpers.PriceConverter` - `src.utils.helpers.get_article_url` ## Scripts ### 1) 핵심 검색 엔진 ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --self-test ``` 역할: - 자연어 파싱 - 후보 단지 탐색 - 단일 단지 조회 - 다중 단지 비교 - 동일 평형 요약 - JSON / 텍스트 출력 - candidate cache seed import/export 연결 ### 1-1) candidate seed 자동 생성 ```bash python skills/naver-real-estate-search/scripts/build_candidate_seeds.py --print-summary python skills/naver-real-estate-search/scripts/build_candidate_seeds.py --input skills/naver-real-estate-search/references/seoul-major-complexes.seed-input.json --output skills/naver-real-estate-search/references/candidate-seeds.generated.json --pause 0.1 --print-summary ``` 역할: - 서울 주요 단지 seed 입력을 받아 자동 alias 확장 - 기존 `candidate-seeds.json` / `candidate-cache.json` / 네이버 검색 HTML 링크 추출을 조합해 candidate seed 초안 생성 - 가능한 경우 단지 상세 API로 후보 검증 후 `confidence` / `verification_status` 부여 - 403/429 발생 시 `blocked_reasons`, `candidate_pool`, `evidence`에 흔적을 남겨 운영자가 후속 검증 가능하게 함 ### 1-2) generated seed preview / apply ```bash python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --json python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --only-names "리센츠,은마" --json python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --apply-target --apply-cache --json ``` 역할: - `candidate-seeds.generated.json`의 `results[]`를 읽어 **운영 승격 가능 항목**과 **수동 검수 큐**를 분리 - 기본값은 preview이며, `--apply-target`일 때만 `references/candidate-seeds.json`을 갱신 - `--apply-cache`를 함께 쓰면 accepted 항목만 `data/candidate-cache.json`에 warm-cache - `verified/weak-verified + complex_id + confidence 기준`으로 승격 후보를 고르고, 나머지는 `manual_review_queue`에 정리 - `--only-names` / `--exclude-names`로 일부 단지만 골라 검수 가능 ### 2) 채팅형 브리핑 래퍼 ```bash python skills/naver-real-estate-search/scripts/chat_real_estate.py --query "잠실 리센츠 전세 30평대" ``` 역할: - 단일 단지 결과를 더 자연스러운 한국어 브리핑으로 요약 - 여러 단지 비교 결과를 “어디가 더 낮은지 / 동일 평형이면 어디가 더 싼지 / 차이가 어느 정도인지” 중심으로 설명 - 대표 매물/링크를 채팅에 붙이기 쉽게 정리 ### 3) 가격 감시 / 새 매물 감지 ```bash python skills/naver-real-estate-search/scripts/watch_real_estate.py add --name "리센츠 전세 30평대" --query "잠실 리센츠 전세 30평대" --target-max-price 950000000 --notify-on-new --notify-on-price-drop python skills/naver-real-estate-search/scripts/watch_real_estate.py check --preview python skills/naver-real-estate-search/scripts/watch_real_estate.py check --json ``` 역할: - 로컬 JSON 파일에 watch rule 저장 - 목표가 이하 / 새 매물 / 가격하락 감지 - `last_seen` / `events` / dedupe 기반의 실사용형 감시 초안 제공 - 텔레그램/브리핑 등 상위 레이어에 바로 전달하기 쉬운 stdout JSON 구조 제공 저장 파일: - `skills/naver-real-estate-search/data/watch-rules.json` - `skills/naver-real-estate-search/data/candidate-cache.json` - `skills/naver-real-estate-search/references/candidate-seeds.json` ## Quick start ### 1) self-test ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --self-test ``` ### 2) 자연어 파싱만 확인 ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --query "잠실 리센츠랑 엘스 전세 비교 30평대" --parse-only ``` ### 3) 후보 단지만 찾기 ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --query "서울 양천구 신월동 신월시영아파트 전세" --list-candidates --json ``` ### 4) 채팅형 후보 리스트 ```bash python skills/naver-real-estate-search/scripts/chat_real_estate.py --query "대치 은마 전세" --list-candidates ``` ### 4-1) candidate-cache seed / alias 학습 ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --seed-candidate-file python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --apply-target --apply-cache --json python skills/naver-real-estate-search/scripts/search_real_estate.py --seed-candidate --complex-id 1147 --candidate-name "리센츠" --candidate-address "서울특별시 송파구 잠실동" --candidate-aliases "잠실 리센츠,잠실리센츠" python skills/naver-real-estate-search/scripts/search_real_estate.py --show-cache --query "리센츠" ``` 권장 순서: 1. `build_candidate_seeds.py`로 generated 초안을 만든다. 2. `apply_generated_seeds.py`로 preview해서 accepted / manual review를 확인한다. 3. 문제가 없을 때만 `--apply-target --apply-cache`를 실행한다. 4. direct URL/ID를 확보한 단지는 `--seed-candidate`로 수동 보강한다. 반복 조회가 필요한 실존 단지나 alias가 자주 흔들리는 단지(예: `신월시영아파트`, `답십리두산위브`)는 먼저 cache에 학습시켜 둔다. 아직 complex ID를 확보하지 못한 경우에도 `manual_review_queue`와 `seed-input`에 남겨 두면 후보 힌트 fallback에 활용된다. ### 5) 단일 단지 조회 ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --complex-id 1147 --trade-types 전세 --limit 10 ``` ### 6) direct URL로 조회 ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --url "https://new.land.naver.com/complexes/1147" --trade-types 매매,전세 --json ``` ### 7) 자연어 한 줄 브리핑 ```bash python skills/naver-real-estate-search/scripts/chat_real_estate.py --query "잠실 리센츠 전세 30평대" --limit 8 ``` ### 8) 여러 단지 비교 브리핑 ```bash python skills/naver-real-estate-search/scripts/chat_real_estate.py --query "잠실 리센츠와 잠실 엘스 전세 비교 30평대" --compare --candidate-limit 2 ``` ## Recommended workflow 1. 사용자가 **특정 단지 URL/ID**를 주면 그 값을 우선 사용한다. 2. URL/ID가 없으면 `chat_real_estate.py --query ...` 로 먼저 자연어 브리핑을 시도한다. 3. 질의가 넓거나 모호하면 `--list-candidates` 로 후보를 1~3개만 보여준다. 4. 후보가 좁혀지면 단일 단지 조회 또는 `--compare` 로 비교한다. 5. 구조화 데이터가 필요하면 같은 조건으로 `search_real_estate.py --json` 을 다시 호출한다. 6. 반복 확인이 필요하면 `watch_real_estate.py add/check` 흐름으로 가격 감시를 연결한다. ## Candidate-search guidance 현재 후보 탐색은 다음 순서로 동작한다. 1. direct complex ID / URL 우선 추출 2. 자연어에서 비교 대상 / 위치 힌트 / 거래유형 / 평형대 분리 3. 로컬 candidate cache(alias → complex_id) exact/contains 매칭 4. 캐시에 없으면 네이버 검색 결과 HTML에서 후보 complex ID 수집 5. 가능하면 단지 상세 API로 이름/주소/세대수 보강 후 캐시에 적재 6. 그래도 후보가 비면 `candidate-seeds.json`의 `manual_review_queue` + `seoul-major-complexes.seed-input.json`를 fallback reference로 조회해 **힌트 후보**를 반환 7. 이름 정규화 / alias 일치 / 지역 힌트 / 질의 토큰 / 세대수 신뢰도 기준으로 점수화 8. 점수 상위 후보만 반환 `신월시영아파트 ↔ 신월시영` 같은 축약/별칭 차이를 줄이기 위해 공백 제거 + suffix 제거 + alias 확장을 같이 사용한다. ## Output guidance ### 단일 단지 - 단지명 / 위치 - 필터 요약(거래유형, 평형대) - 건수 / 최저가 / 평균가 / 중앙값 / 최고가 - 대표 동일 평형 요약 - 대표 매물 2~3개 - 사람이 읽기 쉬운 짧은 해석 ### 여러 단지 비교 - 단지별 핵심 수치 - 전체 평균 기준 어느 단지가 더 낮은지 - 가능한 경우 동일 평형 기준 어느 단지가 더 낮은지 - 대표 매물 또는 링크 ### 가격 감시 - rule 이름 / id - 목표가 이하 매물 수 - 신규 매물 / 가격하락 감지 - stdout JSON의 `alerts`, `snapshot`, `events`, `message_preview`, `summary` 계층 활용 - 텔레그램/브리핑 레이어에는 `watch_real_estate.py check --preview` 결과를 바로 붙이고, 구조화 후처리가 필요하면 `--json`을 사용 ## References 세부 설계/429 운영 지침/감시 schema/seed 예시는 다음 파일을 참고한다. - `references/design.md` - `references/candidate-seeds.json` - `references/seoul-major-complexes.seed-input.json` - `references/candidate-seeds.generated.json` - `references/candidate-seed-builder.md` FILE:data/candidate-cache.json { "version": 3, "updated_at": 1773964931, "entries": [ { "complex_id": "1147", "name": "리센츠", "address": "서울특별시 송파구 잠실동", "household_count": 5563, "aliases": [ "리센츠", "리센츠아파트", "리센츠단지", "잠실 리센츠", "잠실리센츠", "잠실리센츠아파트", "잠실리센츠단지" ], "updated_at": 1773919558, "source": "seed-file:candidate-seeds.json", "learned_from": "seed-file:candidate-seeds.json", "note": "verified warm-cache seed example" } ] } FILE:data/watch-rules.json { "schema_version": 2, "last_checked_at": 1773919559, "rules": [], "events": [], "last_seen": {} } FILE:README.md # naver-real-estate-search 네이버 부동산 기반으로 대한민국 아파트/빌라/오피스텔 매물 조회, 단지 후보 탐색, 비교 브리핑, 가격 감시를 수행하는 OpenClaw 스킬입니다. ## 주요 개선점 - alias/후보 캐시 구조 강화 (`candidate-cache.json` v3 스타일) - candidate-cache seed/수동 학습 CLI 추가 (`--seed-candidate`, `--seed-candidate-file`, `--show-cache`) - 서울 주요 단지용 candidate seed 자동 생성기 추가 (`scripts/build_candidate_seeds.py`) - 지역명/단지명 파서 보강, cold-start 후보 탐색 품질 개선 - `신월시영아파트`, `답십리두산위브` 같은 실사용 질의에 대해 reference-seed/manual-review fallback 보강 - 429 감지 시 direct URL/complex ID 우선 흐름을 유지하는 fallback 메타 추가 - 동일 평형 기준 비교 요약 추가 - 한국어 비교 브리핑과 대표 매물 요약 개선 - watch schema 확장: `last_seen`, `events`, dedupe, 새 매물/가격하락 감지 - 상위 레이어(텔레그램/브리핑) 연동용 stdout JSON 구조 개선 ## candidate seed 운영 흐름 ### 1) generated 초안 만들기 ```bash python skills/naver-real-estate-search/scripts/build_candidate_seeds.py --print-summary python skills/naver-real-estate-search/scripts/build_candidate_seeds.py --input skills/naver-real-estate-search/references/seoul-major-complexes.seed-input.json --output skills/naver-real-estate-search/references/candidate-seeds.generated.json --pause 0.1 --print-summary ``` - 입력: `references/seoul-major-complexes.seed-input.json` - 출력: `references/candidate-seeds.generated.json` - 각 결과에는 `confidence`, `verification_status`, 자동 생성 `aliases`, `candidate_pool`, `evidence`, `blocked_reasons`가 포함됩니다. - 자동 생성 결과는 **운영 seed가 아니라 generated 초안**입니다. ### 2) preview로 승격 후보 / 검수 큐 확인 ```bash python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --json python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --only-names "리센츠,은마" --json ``` - 기본 동작은 preview입니다. 파일은 바꾸지 않습니다. - `verified/weak-verified + complex_id + confidence` 기준으로 accepted 후보를 고릅니다. - 나머지는 `manual_review_queue`로 정리할 후보로 봅니다. - 일부 단지만 볼 때는 `--only-names`, 빼고 싶을 때는 `--exclude-names`를 씁니다. ### 3) 실제 반영 ```bash python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --apply-target --apply-cache --json ``` - `--apply-target`: `references/candidate-seeds.json` 갱신 - `--apply-cache`: accepted 항목만 `data/candidate-cache.json` warm-cache - preview 없이 곧바로 apply하지 말고, 최소 한 번은 summary를 확인하는 편이 안전합니다. 현재 2026-03-20 기준 운영 검수 결과: - 운영 유지: `리센츠` - 오검출 제외: `엘스`, `트리지움` - manual review: `은마`, `래미안대치팰리스`, `아크로리버파크`, `래미안원베일리`, `목동신시가지7단지`, `신월시영아파트`, `답십리두산위브` > 실사용 안정화 메모: production complex ID를 아직 못 확보한 단지도 `manual_review_queue` + `seoul-major-complexes.seed-input.json`에 남겨 두면, `--list-candidates`에서 완전 빈 응답 대신 **reference-seed 후보 힌트**를 돌려줄 수 있습니다. ## 스크립트 ### 1) 후보 탐색 / 단일 조회 / 비교 ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --query "잠실 리센츠 전세 30평대" python skills/naver-real-estate-search/scripts/search_real_estate.py --query "잠실 리센츠와 엘스 전세 비교 30평대" --compare --json python skills/naver-real-estate-search/scripts/search_real_estate.py --query "서울 양천구 신월동 신월시영아파트 전세" --list-candidates --json ``` ### 2) 채팅형 브리핑 ```bash python skills/naver-real-estate-search/scripts/chat_real_estate.py --query "잠실 리센츠 전세 30평대" python skills/naver-real-estate-search/scripts/chat_real_estate.py --query "잠실 리센츠와 엘스 전세 비교 30평대" --compare ``` ### 3) 가격 감시 / 새 매물 감지 ```bash python skills/naver-real-estate-search/scripts/watch_real_estate.py add --name "리센츠 전세 30평대" --query "잠실 리센츠 전세 30평대" --target-max-price 950000000 --notify-on-new --notify-on-price-drop python skills/naver-real-estate-search/scripts/watch_real_estate.py check --preview python skills/naver-real-estate-search/scripts/watch_real_estate.py check --json ``` ## 저장 파일 - `data/candidate-cache.json`: 후보 단지 alias/주소/ID 캐시 - `data/watch-rules.json`: 감시 규칙 + 최근 관측 상태 + 이벤트 히스토리 - `references/candidate-seeds.json`: 초기 alias seed / warm-cache 예시 ## candidate-cache 관리 ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --seed-candidate-file python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --apply-target --apply-cache --json python skills/naver-real-estate-search/scripts/search_real_estate.py --seed-candidate --complex-id 1147 --candidate-name "리센츠" --candidate-address "서울특별시 송파구 잠실동" --candidate-aliases "잠실 리센츠,잠실리센츠" python skills/naver-real-estate-search/scripts/search_real_estate.py --show-cache --query "리센츠" ``` - 운영 seed 전체를 cache에 한 번에 반영할 때는 `search_real_estate.py --seed-candidate-file`를 씁니다. - generated 초안에서 **accepted만 선별 반영**하려면 `apply_generated_seeds.py --apply-target --apply-cache`가 더 안전합니다. - 반복 조회가 필요한 실존 단지는 먼저 cache에 seed/학습시켜 두면 alias mismatch와 cold-start 실패를 줄일 수 있습니다. - `references/candidate-seeds.json`은 **운영 검수 통과 seed만 넣는 파일**입니다. - generated 초안 검토 결과, 오검출/미검증 항목은 `manual_review_queue`로 분리하고 `entries[]`에는 넣지 않습니다. - 특히 `신월시영아파트`, `은마`, `목동신시가지7단지`처럼 alias나 숫자 단지명이 흔들리는 케이스는 direct complex URL/ID 확보 전까지 운영 seed로 올리지 않는 편이 안전합니다. ## seed 검수 기준 요약 - `verified`: 이름/주소/ID 정합성이 맞아 운영 seed에 바로 반영 가능 - `weak-verified`: 자동 근거는 있으나 사람이 한 번 더 보고 승격 - `unresolved`: complex_id 미확보 또는 근거 부족, 운영 seed 제외 - `blocked`: 403/429 등 외부 차단으로 검증 중단, direct URL/ID 수동 확보 우선 서울 주요 단지 보강 우선순위: 1. 신월시영아파트 2. 은마 / 래미안대치팰리스 3. 래미안원베일리 / 아크로리버파크 4. 목동신시가지7단지 5. 엘스 / 트리지움 ## 배포 체크리스트 1. self-test 실행 2. 대표 자연어 질의/후보 탐색/감시 check 실제 실행 3. `references/candidate-seeds.json`의 entries/manual_review_queue가 검수 결과와 일치하는지 확인 4. skill 패키징 5. GitHub tag/release 및 ClawHub publish FILE:references/candidate-seed-builder.md # candidate seed 자동 생성 메모 ## 목적 - 서울 주요 단지 seed 리스트를 입력받아 `candidate-seeds.generated.json` 초안을 만든다. - 기존 `candidate-seeds.json`, `candidate-cache.json`, `search_real_estate.py`와 연결 가능한 구조를 유지한다. - 운영용 승격 기준은 `verification_status in {verified, weak-verified}` 로 두되, **실무에서는 이름/주소/ID 정합성 검수까지 통과한 항목만 `entries[]`에 반영**한다. ## 입력/출력 ### 입력 - `references/seoul-major-complexes.seed-input.json` - shape: ```json { "seeds": [ { "name": "리센츠", "city": "서울특별시", "district": "송파구", "neighborhood": "잠실동", "aliases": ["잠실 리센츠", "잠실리센츠"] } ] } ``` ### 출력 - `references/candidate-seeds.generated.json` - 핵심 필드: - `entries[]`: 자동 검증상 승격 후보 - `results[]`: 전체 생성 결과 + evidence - `unresolved[]`: 후속 수동 검증 대상 ## 상태값 운영 기준 ### `verified` 다음을 모두 만족하는 항목으로 본다. - `complex_id`가 있음 - 상세 조회된 단지명 정규화 결과가 seed 이름과 정확히 맞거나 매우 강하게 일치함 - 주소에 구/동 힌트가 자연스럽게 맞음 - 오검출 정황이 없음 운영 액션: - `candidate-seeds.json`의 `entries[]`에 바로 반영 가능 - warm-cache baseline으로 사용 가능 ### `weak-verified` 다음을 만족하는 항목으로 본다. - `complex_id`가 있음 - 이름/주소가 대체로 맞지만 일부 힌트만 확보됐거나 429 직전 등으로 근거가 약함 - 사람이 빠르게 다시 봤을 때 오검출 가능성이 낮음 운영 액션: - 자동 반영 후보는 될 수 있지만, **서울 주요 단지 운영 seed는 사람이 한 번 더 보고 `entries[]`에 승격**하는 편이 안전 - 저장 시 note에 검수 근거를 남긴다 ### `unresolved` 다음을 의미한다. - `complex_id`를 못 찾았거나 - 후보는 있었지만 이름/주소 검증까지 못 갔거나 - broad query만으로는 신뢰할 수 있는 후보를 고르지 못함 운영 액션: - `entries[]`에는 넣지 않음 - `manual_review_queue`로 넘겨 direct URL/ID를 확보한 뒤 재검증 ### `blocked` 다음을 의미한다. - 403/429/검색 HTML 차단 등 외부 제약 때문에 자동 검증 흐름이 끊김 - 결과 부재가 곧 단지 부재를 뜻하지는 않음 운영 액션: - `entries[]`에는 넣지 않음 - 차단이 풀릴 때까지 broad query 재시도보다 direct URL/ID 수동 확보를 우선 - 운영 가치가 높은 단지(예: 신월시영아파트, 은마)는 우선 보강 큐로 올린다 ## 생성 순서 1. 입력 seed의 이름/지역을 바탕으로 alias를 자동 확장한다. 2. 기존 `references/candidate-seeds.json`의 verified baseline과 이름/alias exact 매칭을 먼저 본다. 3. `candidate-cache.json` exact/contains 매칭으로 warm-cache 힌트를 찾는다. 4. 네이버 검색 HTML에서 complex link/ID를 추출한다. 5. 가능하면 단지 상세 API로 이름/주소/세대수를 조회해 검증한다. 6. `confidence`, `verification_status`, `candidate_pool`, `evidence`, `blocked_reasons`를 남긴다. ## 운영 권장 - 생성 파일의 `entries[]`는 그대로 신뢰하지 말고, **오검출 여부를 반드시 다시 본다.** - 특히 잠실 대단지처럼 비슷한 문맥이 자주 섞이는 곳에서는 다른 complex ID로 잘못 수렴할 수 있다. - `results[]`에 complex_id가 있어도 `verification_status`가 `unverified`면 자동 투입하지 않는다. - `verification_status`가 `weak-verified`여도 서울 주요 단지는 수동 검수 후 승격한다. - 403/429가 반복되면 broad query 대신 direct complex URL/ID로 수동 보강한다. ## preview / apply 운영 절차 ### 1. generated 초안 생성 ```bash python skills/naver-real-estate-search/scripts/build_candidate_seeds.py --print-summary ``` ### 2. preview 확인 ```bash python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --json python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --only-names "리센츠,은마" --json ``` 여기서 확인할 것: - `accepted[]`에 들어간 단지가 정말 운영 승격 대상인지 - `rejected[]`의 `complex_id`가 오검출인지 - `manual_review_queue_count`가 기대와 맞는지 ### 3. 운영 반영 ```bash python skills/naver-real-estate-search/scripts/apply_generated_seeds.py --apply-target --apply-cache --json ``` 반영 결과: - `references/candidate-seeds.json`의 `entries` / `manual_review_queue` 갱신 - accepted 항목만 `data/candidate-cache.json`에 warm-cache ### 4. 수동 보강 자동 승격이 어려운 단지는 direct complex URL/ID 확보 후 다음처럼 수동 보강한다. ```bash python skills/naver-real-estate-search/scripts/search_real_estate.py --seed-candidate --complex-id --candidate-name "단지명" --candidate-address "주소" --candidate-aliases "별칭1,별칭2" ``` ## 실제 검수 요약 (2026-03-20) ### 운영 반영 - `리센츠 -> 1147`: 유지 (`verified`) ### generated에서 제외 - `엘스`: generated가 `1147`(리센츠)로 잘못 수렴해서 제외 - `트리지움`: generated가 `1147`(리센츠)로 잘못 수렴해서 제외 ### 수동 검증 대기 - `은마` - `래미안대치팰리스` - `아크로리버파크` - `래미안원베일리` - `목동신시가지7단지` - `신월시영아파트` - `답십리두산위브` ### 우선 보강 추천 순서 1. `신월시영아파트` — alias 난이도가 높고 403 영향이 커서 baseline 가치가 큼 2. `답십리두산위브` — 실사용 질의 빈도가 높고 현재 cold-start에서 빈 응답이 나기 쉬움 3. `은마`, `래미안대치팰리스` — 대치권 대표 비교 질의 대응용 4. `래미안원베일리`, `아크로리버파크` — 반포권 대표 비교 질의 대응용 5. `목동신시가지7단지` — 숫자 단지형 alias 안정화용 6. `엘스`, `트리지움` — 잠실권 분리 baseline 확정용 complex ID를 아직 못 찾았더라도 아예 버리지 말고 `manual_review_queue`와 `seed-input`에 남겨 둔다. 그러면 후보 탐색 시 reference-seed fallback으로 노출할 수 있어, 실사용에서 `0건` 대신 검수 대기 후보 힌트를 줄 수 있다. ## 한계 - 검색 HTML 구조 변화에 취약하다. - 네이버 상세 API 429 발생 시 검증 성공률이 크게 떨어진다. - warm-cache가 부족하면 broad query만으로는 서울 주요 단지 전체를 안정적으로 맞추기 어렵다. - 따라서 자동 생성은 초안/보조 도구로 보고, 운영 seed는 검증 승격 단계를 두는 것이 안전하다. FILE:references/candidate-seeds.generated.json { "kind": "naver-real-estate-candidate-seed-generation", "schema_version": 1, "generated_at": 1773965554, "source_input": "C:\\Users\\김태완\\.openclaw\\workspace\\skills\\naver-real-estate-search\\references\\seoul-major-complexes.seed-input.json", "entry_count": 1, "unresolved_count": 6, "entries": [ { "complex_id": "1147", "name": "리센츠", "address": "서울특별시 송파구 잠실동", "household_count": 5563, "aliases": [ "리센츠", "리센츠아파트", "리센츠단지", "잠실 리센츠", "잠실리센츠", "잠실리센츠아파트", "잠실리센츠단지", "서울특별시 리센츠", "서울특별시리센츠", "서울특별시리센츠아파트", "서울특별시리센츠단지", "송파구 리센츠", "송파구리센츠", "송파구리센츠아파트", "송파구리센츠단지", "잠실동 리센츠", "잠실동리센츠", "잠실동리센츠아파트", "잠실동리센츠단지", "서울 리센츠", "서울리센츠", "서울리센츠아파트", "서울리센츠단지", "송파구 잠실동 리센츠", "송파구잠실동리센츠", "송파구잠실동리센츠아파트", "송파구잠실동리센츠단지" ], "note": "verified baseline; 생성기 동작 검증용", "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "confidence": 1.0, "verification_status": "verified" } ], "unresolved": [ { "name": "은마", "district": "강남구", "neighborhood": "대치동", "aliases": [ "은마", "은마아파트", "은마단지", "대치 은마", "대치은마", "대치은마아파트", "대치은마단지", "서울특별시 은마", "서울특별시은마", "서울특별시은마아파트", "서울특별시은마단지", "강남구 은마", "강남구은마", "강남구은마아파트", "강남구은마단지", "대치동 은마", "대치동은마", "대치동은마아파트", "대치동은마단지", "서울 은마", "서울은마", "서울은마아파트", "서울은마단지", "강남구 대치동 은마", "강남구대치동은마", "강남구대치동은마아파트", "강남구대치동은마단지" ], "verification_status": "unresolved", "confidence": 0.0, "blocked_reasons": [] }, { "name": "래미안대치팰리스", "district": "강남구", "neighborhood": "대치동", "aliases": [ "래미안대치팰리스", "래미안대치팰리스아파트", "래미안대치팰리스단지", "대치 팰리스", "대치팰리스", "대치팰리스아파트", "대치팰리스단지", "대치래미안팰리스", "대치래미안팰리스아파트", "대치래미안팰리스단지", "래미안 대치팰리스", "서울특별시 래미안대치팰리스", "서울특별시래미안대치팰리스", "서울특별시래미안대치팰리스아파트", "서울특별시래미안대치팰리스단지", "강남구 래미안대치팰리스", "강남구래미안대치팰리스", "강남구래미안대치팰리스아파트", "강남구래미안대치팰리스단지", "대치동 래미안대치팰리스", "대치동래미안대치팰리스", "대치동래미안대치팰리스아파트", "대치동래미안대치팰리스단지", "서울 래미안대치팰리스", "서울래미안대치팰리스", "서울래미안대치팰리스아파트", "서울래미안대치팰리스단지", "강남구 대치동 래미안대치팰리스", "강남구대치동래미안대치팰리스", "강남구대치동래미안대치팰리스아파트" ], "verification_status": "unresolved", "confidence": 0.0, "blocked_reasons": [] }, { "name": "아크로리버파크", "district": "서초구", "neighborhood": "반포동", "aliases": [ "아크로리버파크", "아크로리버파크아파트", "아크로리버파크단지", "반포 아크로리버파크", "반포아크로리버파크", "반포아크로리버파크아파트", "반포아크로리버파크단지", "서울특별시 아크로리버파크", "서울특별시아크로리버파크", "서울특별시아크로리버파크아파트", "서울특별시아크로리버파크단지", "서초구 아크로리버파크", "서초구아크로리버파크", "서초구아크로리버파크아파트", "서초구아크로리버파크단지", "반포동 아크로리버파크", "반포동아크로리버파크", "반포동아크로리버파크아파트", "반포동아크로리버파크단지", "서울 아크로리버파크", "서울아크로리버파크", "서울아크로리버파크아파트", "서울아크로리버파크단지", "서초구 반포동 아크로리버파크", "서초구반포동아크로리버파크", "서초구반포동아크로리버파크아파트", "서초구반포동아크로리버파크단지" ], "verification_status": "unresolved", "confidence": 0.0, "blocked_reasons": [] }, { "name": "래미안원베일리", "district": "서초구", "neighborhood": "반포동", "aliases": [ "래미안원베일리", "래미안원베일리아파트", "래미안원베일리단지", "반포 원베일리", "반포원베일리", "반포원베일리아파트", "반포원베일리단지", "원베일리", "원베일리아파트", "원베일리단지", "서울특별시 래미안원베일리", "서울특별시래미안원베일리", "서울특별시래미안원베일리아파트", "서울특별시래미안원베일리단지", "서초구 래미안원베일리", "서초구래미안원베일리", "서초구래미안원베일리아파트", "서초구래미안원베일리단지", "반포동 래미안원베일리", "반포동래미안원베일리", "반포동래미안원베일리아파트", "반포동래미안원베일리단지", "서울 래미안원베일리", "서울래미안원베일리", "서울래미안원베일리아파트", "서울래미안원베일리단지", "서초구 반포동 래미안원베일리", "서초구반포동래미안원베일리", "서초구반포동래미안원베일리아파트", "서초구반포동래미안원베일리단지" ], "verification_status": "unresolved", "confidence": 0.0, "blocked_reasons": [] }, { "name": "목동신시가지7단지", "district": "양천구", "neighborhood": "목동", "aliases": [ "목동신시가지7단지", "목동신시가지7", "목동신시가지7아파트", "목동 7단지", "목동 7", "목동7", "목동7아파트", "목동7단지", "목동신시가지 7단지", "목동신시가지 7", "서울특별시 목동신시가지7단지", "서울특별시 목동신시가지7", "서울특별시목동신시가지7", "서울특별시목동신시가지7아파트", "서울특별시목동신시가지7단지", "양천구 목동신시가지7단지", "양천구 목동신시가지7", "양천구목동신시가지7", "양천구목동신시가지7아파트", "양천구목동신시가지7단지", "목동 목동신시가지7단지", "목동 목동신시가지7", "목동목동신시가지7", "목동목동신시가지7아파트", "목동목동신시가지7단지", "서울 목동신시가지7단지", "서울 목동신시가지7", "서울목동신시가지7", "서울목동신시가지7아파트", "서울목동신시가지7단지" ], "verification_status": "unresolved", "confidence": 0.0, "blocked_reasons": [] }, { "name": "신월시영아파트", "district": "양천구", "neighborhood": "신월동", "aliases": [ "신월시영아파트", "신월시영", "신월시영단지", "양천 신월시영", "양천신월시영", "양천신월시영아파트", "양천신월시영단지", "서울특별시 신월시영아파트", "서울특별시신월시영", "서울특별시신월시영아파트", "서울특별시신월시영단지", "서울특별시 신월시영", "양천구 신월시영아파트", "양천구신월시영", "양천구신월시영아파트", "양천구신월시영단지", "양천구 신월시영", "신월동 신월시영아파트", "신월동신월시영", "신월동신월시영아파트", "신월동신월시영단지", "신월동 신월시영", "서울 신월시영아파트", "서울신월시영", "서울신월시영아파트", "서울신월시영단지", "서울 신월시영", "양천구 신월동 신월시영아파트", "양천구신월동신월시영", "양천구신월동신월시영아파트" ], "verification_status": "blocked", "confidence": 0.0, "blocked_reasons": [ "네이버 검색 HTML 후보 탐색이 차단되었습니다: HTTP 403" ] } ], "results": [ { "name": "리센츠", "district": "송파구", "neighborhood": "잠실동", "seed_input": { "name": "리센츠", "city": "서울특별시", "district": "송파구", "neighborhood": "잠실동", "aliases": [ "잠실 리센츠", "잠실리센츠" ], "note": "verified baseline; 생성기 동작 검증용" }, "complex_id": "1147", "address": "서울특별시 송파구 잠실동", "household_count": 5563, "aliases": [ "리센츠", "리센츠아파트", "리센츠단지", "잠실 리센츠", "잠실리센츠", "잠실리센츠아파트", "잠실리센츠단지", "서울특별시 리센츠", "서울특별시리센츠", "서울특별시리센츠아파트", "서울특별시리센츠단지", "송파구 리센츠", "송파구리센츠", "송파구리센츠아파트", "송파구리센츠단지", "잠실동 리센츠", "잠실동리센츠", "잠실동리센츠아파트", "잠실동리센츠단지", "서울 리센츠", "서울리센츠", "서울리센츠아파트", "서울리센츠단지", "송파구 잠실동 리센츠", "송파구잠실동리센츠", "송파구잠실동리센츠아파트", "송파구잠실동리센츠단지" ], "verification_status": "verified", "confidence": 1.0, "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": "verified baseline; 생성기 동작 검증용", "candidate_pool": [ { "complex_id": "1147", "query": "리센츠", "discovery_source": "baseline-seed", "status": "verified", "confidence": 1.0, "verification_reasons": [ "baseline-seed-match" ], "complex": { "complex_id": "1147", "name": "리센츠", "address": "서울특별시 송파구 잠실동", "household_count": 5563 } } ], "evidence": [ { "query": "리센츠", "status": "baseline-seed-hit", "ids": [ "1147" ] }, { "query": "서울특별시 송파구 잠실동 리센츠 네이버 부동산", "status": "cache-hit", "source": "candidate-cache", "ids": [ "1147" ] } ], "blocked_reasons": [] }, { "name": "엘스", "district": "송파구", "neighborhood": "잠실동", "seed_input": { "name": "엘스", "city": "서울특별시", "district": "송파구", "neighborhood": "잠실동", "aliases": [ "잠실 엘스", "잠실엘스" ] }, "complex_id": "1147", "address": "", "household_count": null, "aliases": [ "엘스", "엘스아파트", "엘스단지", "잠실 엘스", "잠실엘스", "잠실엘스아파트", "잠실엘스단지", "서울특별시 엘스", "서울특별시엘스", "서울특별시엘스아파트", "서울특별시엘스단지", "송파구 엘스", "송파구엘스", "송파구엘스아파트", "송파구엘스단지", "잠실동 엘스", "잠실동엘스", "잠실동엘스아파트", "잠실동엘스단지", "서울 엘스", "서울엘스", "서울엘스아파트", "서울엘스단지", "송파구 잠실동 엘스", "송파구잠실동엘스", "송파구잠실동엘스아파트", "송파구잠실동엘스단지" ], "verification_status": "unverified", "confidence": 0.0, "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": "generated candidate seed", "candidate_pool": [ { "complex_id": "1147", "query": "서울특별시 송파구 잠실동 잠실 엘스 네이버 부동산", "discovery_source": "candidate-cache", "status": "unverified", "confidence": 0.0, "verification_reasons": [ "detail-fetch-failed:SearchError" ], "error": "네이버 부동산 API가 429(요청 제한)를 반환했습니다. 단일 단지 URL/ID를 우선 사용하거나, 후보 단지부터 1~3개로 좁혀 다시 시도해 주세요.", "complex": { "complex_id": "1147" } } ], "evidence": [ { "ids": [], "status": "ok", "query": "서울특별시 송파구 잠실동 엘스 네이버 부동산" }, { "ids": [], "status": "ok", "query": "송파구 잠실동 엘스 네이버 부동산" }, { "ids": [], "status": "ok", "query": "송파구 엘스 네이버 부동산" }, { "ids": [], "status": "ok", "query": "엘스 네이버 부동산 아파트" }, { "ids": [], "status": "ok", "query": "site:new.land.naver.com/complexes 송파구 엘스" }, { "query": "서울특별시 송파구 잠실동 잠실 엘스 네이버 부동산", "status": "cache-hit", "source": "candidate-cache", "ids": [ "1147" ] }, { "ids": [], "status": "ok", "query": "서울특별시 송파구 잠실동 잠실 엘스 네이버 부동산" }, { "query": "송파구 잠실동 잠실 엘스 네이버 부동산", "status": "cache-hit", "source": "candidate-cache", "ids": [ "1147" ] } ], "blocked_reasons": [] }, { "name": "트리지움", "district": "송파구", "neighborhood": "잠실동", "seed_input": { "name": "트리지움", "city": "서울특별시", "district": "송파구", "neighborhood": "잠실동", "aliases": [ "잠실 트리지움", "잠실트리지움" ] }, "complex_id": "1147", "address": "", "household_count": null, "aliases": [ "트리지움", "트리지움아파트", "트리지움단지", "잠실 트리지움", "잠실트리지움", "잠실트리지움아파트", "잠실트리지움단지", "서울특별시 트리지움", "서울특별시트리지움", "서울특별시트리지움아파트", "서울특별시트리지움단지", "송파구 트리지움", "송파구트리지움", "송파구트리지움아파트", "송파구트리지움단지", "잠실동 트리지움", "잠실동트리지움", "잠실동트리지움아파트", "잠실동트리지움단지", "서울 트리지움", "서울트리지움", "서울트리지움아파트", "서울트리지움단지", "송파구 잠실동 트리지움", "송파구잠실동트리지움", "송파구잠실동트리지움아파트", "송파구잠실동트리지움단지" ], "verification_status": "unverified", "confidence": 0.0, "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": "generated candidate seed", "candidate_pool": [ { "complex_id": "1147", "query": "서울특별시 송파구 잠실동 잠실 트리지움 네이버 부동산", "discovery_source": "candidate-cache", "status": "unverified", "confidence": 0.0, "verification_reasons": [ "detail-fetch-failed:SearchError" ], "error": "네이버 부동산 API가 429(요청 제한)를 반환했습니다. 단일 단지 URL/ID를 우선 사용하거나, 후보 단지부터 1~3개로 좁혀 다시 시도해 주세요.", "complex": { "complex_id": "1147" } } ], "evidence": [ { "ids": [], "status": "ok", "query": "서울특별시 송파구 잠실동 트리지움 네이버 부동산" }, { "ids": [], "status": "ok", "query": "송파구 잠실동 트리지움 네이버 부동산" }, { "ids": [], "status": "ok", "query": "송파구 트리지움 네이버 부동산" }, { "ids": [], "status": "ok", "query": "트리지움 네이버 부동산 아파트" }, { "ids": [], "status": "ok", "query": "site:new.land.naver.com/complexes 송파구 트리지움" }, { "query": "서울특별시 송파구 잠실동 잠실 트리지움 네이버 부동산", "status": "cache-hit", "source": "candidate-cache", "ids": [ "1147" ] }, { "ids": [], "status": "ok", "query": "서울특별시 송파구 잠실동 잠실 트리지움 네이버 부동산" }, { "query": "송파구 잠실동 잠실 트리지움 네이버 부동산", "status": "cache-hit", "source": "candidate-cache", "ids": [ "1147" ] } ], "blocked_reasons": [] }, { "name": "은마", "district": "강남구", "neighborhood": "대치동", "seed_input": { "name": "은마", "city": "서울특별시", "district": "강남구", "neighborhood": "대치동", "aliases": [ "은마아파트", "대치 은마" ] }, "complex_id": null, "address": "", "household_count": null, "aliases": [ "은마", "은마아파트", "은마단지", "대치 은마", "대치은마", "대치은마아파트", "대치은마단지", "서울특별시 은마", "서울특별시은마", "서울특별시은마아파트", "서울특별시은마단지", "강남구 은마", "강남구은마", "강남구은마아파트", "강남구은마단지", "대치동 은마", "대치동은마", "대치동은마아파트", "대치동은마단지", "서울 은마", "서울은마", "서울은마아파트", "서울은마단지", "강남구 대치동 은마", "강남구대치동은마", "강남구대치동은마아파트", "강남구대치동은마단지" ], "verification_status": "unresolved", "confidence": 0.0, "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": "generated candidate seed", "candidate_pool": [], "evidence": [ { "ids": [], "status": "ok", "query": "서울특별시 강남구 대치동 은마 네이버 부동산" }, { "ids": [], "status": "ok", "query": "강남구 대치동 은마 네이버 부동산" }, { "ids": [], "status": "ok", "query": "강남구 은마 네이버 부동산" }, { "ids": [], "status": "ok", "query": "은마 네이버 부동산 아파트" }, { "ids": [], "status": "ok", "query": "site:new.land.naver.com/complexes 강남구 은마" }, { "ids": [], "status": "ok", "query": "서울특별시 강남구 대치동 은마아파트 네이버 부동산" }, { "ids": [], "status": "ok", "query": "강남구 대치동 은마아파트 네이버 부동산" }, { "ids": [], "status": "ok", "query": "강남구 은마아파트 네이버 부동산" } ], "blocked_reasons": [] }, { "name": "래미안대치팰리스", "district": "강남구", "neighborhood": "대치동", "seed_input": { "name": "래미안대치팰리스", "city": "서울특별시", "district": "강남구", "neighborhood": "대치동", "aliases": [ "대치 팰리스", "대치래미안팰리스", "래미안 대치팰리스" ] }, "complex_id": null, "address": "", "household_count": null, "aliases": [ "래미안대치팰리스", "래미안대치팰리스아파트", "래미안대치팰리스단지", "대치 팰리스", "대치팰리스", "대치팰리스아파트", "대치팰리스단지", "대치래미안팰리스", "대치래미안팰리스아파트", "대치래미안팰리스단지", "래미안 대치팰리스", "서울특별시 래미안대치팰리스", "서울특별시래미안대치팰리스", "서울특별시래미안대치팰리스아파트", "서울특별시래미안대치팰리스단지", "강남구 래미안대치팰리스", "강남구래미안대치팰리스", "강남구래미안대치팰리스아파트", "강남구래미안대치팰리스단지", "대치동 래미안대치팰리스", "대치동래미안대치팰리스", "대치동래미안대치팰리스아파트", "대치동래미안대치팰리스단지", "서울 래미안대치팰리스", "서울래미안대치팰리스", "서울래미안대치팰리스아파트", "서울래미안대치팰리스단지", "강남구 대치동 래미안대치팰리스", "강남구대치동래미안대치팰리스", "강남구대치동래미안대치팰리스아파트" ], "verification_status": "unresolved", "confidence": 0.0, "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": "generated candidate seed", "candidate_pool": [], "evidence": [ { "ids": [], "status": "ok", "query": "서울특별시 강남구 대치동 래미안대치팰리스 네이버 부동산" }, { "ids": [], "status": "ok", "query": "강남구 대치동 래미안대치팰리스 네이버 부동산" }, { "ids": [], "status": "ok", "query": "강남구 래미안대치팰리스 네이버 부동산" }, { "ids": [], "status": "ok", "query": "래미안대치팰리스 네이버 부동산 아파트" }, { "ids": [], "status": "ok", "query": "site:new.land.naver.com/complexes 강남구 래미안대치팰리스" }, { "ids": [], "status": "ok", "query": "서울특별시 강남구 대치동 대치 팰리스 네이버 부동산" }, { "ids": [], "status": "ok", "query": "강남구 대치동 대치 팰리스 네이버 부동산" }, { "ids": [], "status": "ok", "query": "강남구 대치 팰리스 네이버 부동산" } ], "blocked_reasons": [] }, { "name": "아크로리버파크", "district": "서초구", "neighborhood": "반포동", "seed_input": { "name": "아크로리버파크", "city": "서울특별시", "district": "서초구", "neighborhood": "반포동", "aliases": [ "반포 아크로리버파크", "반포아크로리버파크" ] }, "complex_id": null, "address": "", "household_count": null, "aliases": [ "아크로리버파크", "아크로리버파크아파트", "아크로리버파크단지", "반포 아크로리버파크", "반포아크로리버파크", "반포아크로리버파크아파트", "반포아크로리버파크단지", "서울특별시 아크로리버파크", "서울특별시아크로리버파크", "서울특별시아크로리버파크아파트", "서울특별시아크로리버파크단지", "서초구 아크로리버파크", "서초구아크로리버파크", "서초구아크로리버파크아파트", "서초구아크로리버파크단지", "반포동 아크로리버파크", "반포동아크로리버파크", "반포동아크로리버파크아파트", "반포동아크로리버파크단지", "서울 아크로리버파크", "서울아크로리버파크", "서울아크로리버파크아파트", "서울아크로리버파크단지", "서초구 반포동 아크로리버파크", "서초구반포동아크로리버파크", "서초구반포동아크로리버파크아파트", "서초구반포동아크로리버파크단지" ], "verification_status": "unresolved", "confidence": 0.0, "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": "generated candidate seed", "candidate_pool": [], "evidence": [ { "ids": [], "status": "ok", "query": "서울특별시 서초구 반포동 아크로리버파크 네이버 부동산" }, { "ids": [], "status": "ok", "query": "서초구 반포동 아크로리버파크 네이버 부동산" }, { "ids": [], "status": "ok", "query": "서초구 아크로리버파크 네이버 부동산" }, { "ids": [], "status": "ok", "query": "아크로리버파크 네이버 부동산 아파트" }, { "ids": [], "status": "ok", "query": "site:new.land.naver.com/complexes 서초구 아크로리버파크" }, { "ids": [], "status": "ok", "query": "서울특별시 서초구 반포동 반포 아크로리버파크 네이버 부동산" }, { "ids": [], "status": "ok", "query": "서초구 반포동 반포 아크로리버파크 네이버 부동산" }, { "ids": [], "status": "ok", "query": "서초구 반포 아크로리버파크 네이버 부동산" } ], "blocked_reasons": [] }, { "name": "래미안원베일리", "district": "서초구", "neighborhood": "반포동", "seed_input": { "name": "래미안원베일리", "city": "서울특별시", "district": "서초구", "neighborhood": "반포동", "aliases": [ "반포 원베일리", "원베일리" ] }, "complex_id": null, "address": "", "household_count": null, "aliases": [ "래미안원베일리", "래미안원베일리아파트", "래미안원베일리단지", "반포 원베일리", "반포원베일리", "반포원베일리아파트", "반포원베일리단지", "원베일리", "원베일리아파트", "원베일리단지", "서울특별시 래미안원베일리", "서울특별시래미안원베일리", "서울특별시래미안원베일리아파트", "서울특별시래미안원베일리단지", "서초구 래미안원베일리", "서초구래미안원베일리", "서초구래미안원베일리아파트", "서초구래미안원베일리단지", "반포동 래미안원베일리", "반포동래미안원베일리", "반포동래미안원베일리아파트", "반포동래미안원베일리단지", "서울 래미안원베일리", "서울래미안원베일리", "서울래미안원베일리아파트", "서울래미안원베일리단지", "서초구 반포동 래미안원베일리", "서초구반포동래미안원베일리", "서초구반포동래미안원베일리아파트", "서초구반포동래미안원베일리단지" ], "verification_status": "unresolved", "confidence": 0.0, "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": "generated candidate seed", "candidate_pool": [], "evidence": [ { "ids": [], "status": "ok", "query": "서울특별시 서초구 반포동 래미안원베일리 네이버 부동산" }, { "ids": [], "status": "ok", "query": "서초구 반포동 래미안원베일리 네이버 부동산" }, { "ids": [], "status": "ok", "query": "서초구 래미안원베일리 네이버 부동산" }, { "ids": [], "status": "ok", "query": "래미안원베일리 네이버 부동산 아파트" }, { "ids": [], "status": "ok", "query": "site:new.land.naver.com/complexes 서초구 래미안원베일리" }, { "ids": [], "status": "ok", "query": "서울특별시 서초구 반포동 반포 원베일리 네이버 부동산" }, { "ids": [], "status": "ok", "query": "서초구 반포동 반포 원베일리 네이버 부동산" }, { "ids": [], "status": "ok", "query": "서초구 반포 원베일리 네이버 부동산" } ], "blocked_reasons": [] }, { "name": "목동신시가지7단지", "district": "양천구", "neighborhood": "목동", "seed_input": { "name": "목동신시가지7단지", "city": "서울특별시", "district": "양천구", "neighborhood": "목동", "aliases": [ "목동 7단지", "목동신시가지7", "목동신시가지 7단지" ] }, "complex_id": null, "address": "", "household_count": null, "aliases": [ "목동신시가지7단지", "목동신시가지7", "목동신시가지7아파트", "목동 7단지", "목동 7", "목동7", "목동7아파트", "목동7단지", "목동신시가지 7단지", "목동신시가지 7", "서울특별시 목동신시가지7단지", "서울특별시 목동신시가지7", "서울특별시목동신시가지7", "서울특별시목동신시가지7아파트", "서울특별시목동신시가지7단지", "양천구 목동신시가지7단지", "양천구 목동신시가지7", "양천구목동신시가지7", "양천구목동신시가지7아파트", "양천구목동신시가지7단지", "목동 목동신시가지7단지", "목동 목동신시가지7", "목동목동신시가지7", "목동목동신시가지7아파트", "목동목동신시가지7단지", "서울 목동신시가지7단지", "서울 목동신시가지7", "서울목동신시가지7", "서울목동신시가지7아파트", "서울목동신시가지7단지" ], "verification_status": "unresolved", "confidence": 0.0, "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": "generated candidate seed", "candidate_pool": [], "evidence": [ { "ids": [], "status": "ok", "query": "서울특별시 양천구 목동 목동신시가지7단지 네이버 부동산" }, { "ids": [], "status": "ok", "query": "양천구 목동 목동신시가지7단지 네이버 부동산" }, { "ids": [], "status": "ok", "query": "양천구 목동신시가지7단지 네이버 부동산" }, { "ids": [], "status": "ok", "query": "목동신시가지7단지 네이버 부동산 아파트" }, { "ids": [], "status": "ok", "query": "site:new.land.naver.com/complexes 양천구 목동신시가지7단지" }, { "ids": [], "status": "ok", "query": "서울특별시 양천구 목동 목동 7단지 네이버 부동산" }, { "ids": [], "status": "ok", "query": "양천구 목동 목동 7단지 네이버 부동산" }, { "ids": [], "status": "ok", "query": "양천구 목동 7단지 네이버 부동산" } ], "blocked_reasons": [] }, { "name": "신월시영아파트", "district": "양천구", "neighborhood": "신월동", "seed_input": { "name": "신월시영아파트", "city": "서울특별시", "district": "양천구", "neighborhood": "신월동", "aliases": [ "신월시영", "양천 신월시영" ], "note": "tricky alias 케이스" }, "complex_id": null, "address": "", "household_count": null, "aliases": [ "신월시영아파트", "신월시영", "신월시영단지", "양천 신월시영", "양천신월시영", "양천신월시영아파트", "양천신월시영단지", "서울특별시 신월시영아파트", "서울특별시신월시영", "서울특별시신월시영아파트", "서울특별시신월시영단지", "서울특별시 신월시영", "양천구 신월시영아파트", "양천구신월시영", "양천구신월시영아파트", "양천구신월시영단지", "양천구 신월시영", "신월동 신월시영아파트", "신월동신월시영", "신월동신월시영아파트", "신월동신월시영단지", "신월동 신월시영", "서울 신월시영아파트", "서울신월시영", "서울신월시영아파트", "서울신월시영단지", "서울 신월시영", "양천구 신월동 신월시영아파트", "양천구신월동신월시영", "양천구신월동신월시영아파트" ], "verification_status": "blocked", "confidence": 0.0, "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": "tricky alias 케이스", "candidate_pool": [], "evidence": [ { "ids": [], "status": "ok", "query": "서울특별시 양천구 신월동 신월시영아파트 네이버 부동산" }, { "ids": [], "status": "ok", "query": "양천구 신월동 신월시영아파트 네이버 부동산" }, { "ids": [], "status": "ok", "query": "양천구 신월시영아파트 네이버 부동산" }, { "ids": [], "status": "ok", "query": "신월시영아파트 네이버 부동산 아파트" }, { "ids": [], "status": "ok", "query": "site:new.land.naver.com/complexes 양천구 신월시영아파트" }, { "ids": [], "status": "ok", "query": "서울특별시 양천구 신월동 신월시영 네이버 부동산" }, { "ids": [], "status": "ok", "query": "양천구 신월동 신월시영 네이버 부동산" }, { "ids": [], "status": "ok", "query": "양천구 신월시영 네이버 부동산" } ], "blocked_reasons": [ "네이버 검색 HTML 후보 탐색이 차단되었습니다: HTTP 403" ] } ] } FILE:references/candidate-seeds.json { "schema_version": 2, "updated_at": 1773970000, "policy": { "production_entry_rule": "verification_status가 verified 또는 weak-verified 이고, complex_id/name/address 정합성이 확인된 항목만 entries에 넣는다.", "excluded_rule": "generated 결과에서 complex_id가 있더라도 이름/주소 검증이 실패했거나 429/오염된 cache에 기대는 항목은 entries에 넣지 않는다." }, "entries": [ { "complex_id": "1147", "name": "리센츠", "address": "서울특별시 송파구 잠실동", "household_count": 5563, "aliases": [ "리센츠", "잠실 리센츠", "잠실리센츠", "리센츠아파트" ], "verification_status": "verified", "note": "운영용 verified warm-cache seed" } ], "manual_review_queue": [ { "name": "엘스", "district": "송파구", "neighborhood": "잠실동", "review_status": "excluded-for-now", "reason": "generated가 complex_id 1147(리센츠)로 잘못 수렴했다. 429 환경에서 오염된 cache candidate가 끼어든 케이스로 보고 운영 seed에서 제외한다.", "next_action": "direct complex URL/ID를 확보한 뒤 verified 또는 weak-verified로 재검수한다." }, { "name": "트리지움", "district": "송파구", "neighborhood": "잠실동", "review_status": "excluded-for-now", "reason": "generated가 complex_id 1147(리센츠)로 잘못 수렴했다. 잠실 대단지 군집에서 alias 충돌 위험이 있다.", "next_action": "잠실권 단지는 URL/ID 기반 baseline을 별도로 확보해 분리 검증한다." }, { "name": "은마", "district": "강남구", "neighborhood": "대치동", "review_status": "manual-verification-needed", "reason": "generated evidence상 complex_id를 확보하지 못했다. 서울 핵심 단지라 운영 가치가 높지만 현재 자동 생성만으로는 불충분하다.", "next_action": "네이버 complex URL/ID를 직접 확보해 대치동 주소 일치 여부를 확인한 후 seed 추가를 검토한다." }, { "name": "래미안대치팰리스", "district": "강남구", "neighborhood": "대치동", "review_status": "manual-verification-needed", "reason": "자동 생성에서 complex_id 미확보. 대치권 비교 질의에 중요하지만 현재는 운영 seed로 승격 불가.", "next_action": "브랜드명/띄어쓰기 변형을 줄인 direct URL/ID baseline을 먼저 수집한다." }, { "name": "아크로리버파크", "district": "서초구", "neighborhood": "반포동", "review_status": "manual-verification-needed", "reason": "generated unresolved. 반포권 대표 단지지만 HTML 후보 추출만으로는 안정적으로 못 잡는다.", "next_action": "반포동 direct complex URL/ID로 seed를 만든 뒤 weak-verified 이상일 때만 entries로 승격한다." }, { "name": "래미안원베일리", "district": "서초구", "neighborhood": "반포동", "review_status": "manual-verification-needed", "reason": "generated unresolved. 신축/브랜드 단지라 검색량은 높지만 자동 검색 안정성은 낮다.", "next_action": "반포 원베일리/래미안원베일리 alias를 분리한 수동 검증 seed를 마련한다." }, { "name": "목동신시가지7단지", "district": "양천구", "neighborhood": "목동", "review_status": "manual-verification-needed", "reason": "숫자 단지형 이름이라 parser와 검색 결과가 흔들린다.", "next_action": "목동 7단지/목동신시가지7단지/목동신시가지 7단지 alias를 유지하되 complex_id 확정 전까지 entries에 넣지 않는다." }, { "name": "신월시영아파트", "district": "양천구", "neighborhood": "신월동", "review_status": "blocked-but-high-priority", "reason": "403 차단으로 generated가 막혔다. 다만 alias 난이도가 높아서 운영상 seed 가치가 크다.", "next_action": "신월시영 complex URL/ID를 수동 확보해 최우선으로 baseline seed에 추가한다." }, { "name": "답십리두산위브", "district": "동대문구", "neighborhood": "답십리동", "review_status": "manual-verification-needed", "reason": "실사용 질의 빈도가 높은데 현재 production seed/candidate-cache에 없음. 네이버 complex ID를 아직 수동 확정하지 못했다.", "next_action": "direct complex URL/ID를 확보하면 production seed로 승격하고, 그 전까지는 reference seed fallback으로 후보 탐색 응답을 보강한다." } ], "review_summary": { "accepted_for_production": [ "리센츠" ], "excluded_due_to_wrong_match": [ "엘스", "트리지움" ], "pending_manual_verification": [ "은마", "래미안대치팰리스", "아크로리버파크", "래미안원베일리", "목동신시가지7단지", "신월시영아파트", "답십리두산위브" ], "priority_backfill_targets": [ "신월시영아파트", "답십리두산위브", "은마", "래미안대치팰리스", "래미안원베일리", "아크로리버파크", "목동신시가지7단지", "엘스", "트리지움" ] } } FILE:references/design.md # naver-real-estate-search 설계 메모 ## 목표 - 대한민국 / 네이버 부동산 맥락에 맞는 OpenClaw 스킬 제공 - 자연어 요청을 최소 실행 가능한 검색 파라미터로 빠르게 축약 - `twbeatles/naverland-scrapper` 내부 로직을 가능한 범위에서 재사용 - 429 환경에서도 **direct URL/ID 우선 → 후보 좁히기 → 비교 → 감시** 흐름 유지 ## 이번 고도화 핵심 ### 1) 식별/안정성 - candidate cache를 단순 리스트에서 `version/entries/updated_at` 구조로 확장하고, seed/import/list 관리용 CLI를 추가 - alias 변형 생성 강화: 원문 / 공백 정리 / suffix 제거 / `아파트` 재부착 / 일부 지역 alias 확장 - 자연어 파서 강화: raw subject, location hint, direct complex ID 추출 보존 - cold-start 후보 탐색 시 `지역 + 단지명`, raw subject, cleaned query를 조합한 다중 검색어 전략 사용 - 점수화 기준 강화: 이름 정규화, alias, 질의 토큰, 주소 내 지역 힌트, 세대수 신뢰도 반영 - `신월시영아파트` 같은 케이스에서 alias cache warm-up 이후 재질문 성공률이 높아지도록 설계 - `references/candidate-seeds.json`을 통해 자주 쓰는 단지 후보를 운영 전 warm-cache 할 수 있게 설계 - 429 발생 시 메타에 상태를 남기고 direct URL/ID 우선 재시도 흐름을 유지 ### 2) 비교/출력 - 거래유형별 `min/avg/median/max` 요약 추가 - `area_summary`를 도입해 동일 평형 버킷(예: `33평`) 기준 비교 가능하게 확장 - compare 결과에 `compare_insights.trade`, `compare_insights.same_area` 계층 추가 - 한국어 브리핑에서: - 전체 평균 비교 문장 - 동일 평형 기준 비교 문장 - 가격 분산 해석 문장 - 대표 매물 bullet 정리 를 더 자연스럽게 생성 ### 3) 감시/연동 - watch schema를 `schema_version`, `rules`, `events`, `last_seen`, `last_checked_at` 구조로 확장 - rule 단위 옵션 추가: - `target_max_price` - `notify_on_new` - `notify_on_price_drop` - check 결과 stdout JSON을 상위 레이어 친화적으로 정리: - `kind` - `schema_version` - `checked_at` - `alert_count` - `alerts[]` - `message_preview` - `summary` - `last_seen` + `dedupe_key` 기반 중복 알림 억제 - snapshot에 `complex_info`, `market_summary`, `meta`를 넣어 텔레그램/브리핑 레이어가 재가공하기 쉽게 만듦 ## 후보 탐색 전략 상세 1. direct complex ID / URL / 텍스트 내 complex ID를 먼저 본다. 2. cache에서 alias exact/contains 매칭을 먼저 시도한다. 3. 실패 시 web search를 쓰되 단일 키워드 하나만 쓰지 않는다. - raw subject - cleaned query - candidate keyword - location hint - `지역 + 단지명` 조합 4. 수집한 complex ID에 대해 상세 API를 조회해 이름/주소/세대수 보강 5. 점수 상위 후보만 반환 ## 429 운영 원칙 - 먼저 짧은 backoff로 재시도한다. - 여전히 실패하면 더 넓은 탐색으로 밀어붙이지 않는다. - 사용자나 상위 레이어에 다음을 유도한다. - direct complex ID - direct URL - 후보 1~3개만 좁혀 재시도 ## watch JSON 예시 ```json { "kind": "naver-real-estate-watch-check", "schema_version": 2, "checked_at": 1760000000, "alert_count": 2, "alerts": [ { "rule": {"id": "rule-abcd1234", "name": "리센츠 전세 30평대"}, "matched_count": 1, "matched": [ { "event_type": "target_hit", "article_key": "1147:123456789", "price": 950000000, "price_text": "9억 5,000", "article_url": "https://new.land.naver.com/..." } ], "snapshot": { "complex_info": {"name": "리센츠"}, "market_summary": {"전세": {"count": 8}}, "meta": {"rate_limited": false} } } ] } ``` ## 테스트 권장 시나리오 - `--self-test` - `--parse-only` 로 자연어 파서 확인 - `--list-candidates` 로 tricky alias 확인 - direct complex ID 조회 - 비교 브리핑 확인 - watch add/check 및 두 번째 check에서 dedupe 동작 확인 ## 추가 개선 후보 - Playwright 세션 재사용 기반 429 회피력 향상 - 실거래가/전세가율 같은 파생 지표 추가 - 특정 지역 사전(alias seed) 확장 - 텔레그램 markdown-safe formatter 별도 스크립트 추가 FILE:references/seoul-major-complexes.seed-input.json { "kind": "naver-real-estate-seed-input", "schema_version": 1, "description": "서울 주요 단지 seed 생성용 입력 예시", "seeds": [ { "name": "리센츠", "city": "서울특별시", "district": "송파구", "neighborhood": "잠실동", "aliases": ["잠실 리센츠", "잠실리센츠"], "note": "verified baseline; 생성기 동작 검증용" }, { "name": "엘스", "city": "서울특별시", "district": "송파구", "neighborhood": "잠실동", "aliases": ["잠실 엘스", "잠실엘스"] }, { "name": "트리지움", "city": "서울특별시", "district": "송파구", "neighborhood": "잠실동", "aliases": ["잠실 트리지움", "잠실트리지움"] }, { "name": "은마", "city": "서울특별시", "district": "강남구", "neighborhood": "대치동", "aliases": ["은마아파트", "대치 은마"] }, { "name": "래미안대치팰리스", "city": "서울특별시", "district": "강남구", "neighborhood": "대치동", "aliases": ["대치 팰리스", "대치래미안팰리스", "래미안 대치팰리스"] }, { "name": "아크로리버파크", "city": "서울특별시", "district": "서초구", "neighborhood": "반포동", "aliases": ["반포 아크로리버파크", "반포아크로리버파크"] }, { "name": "래미안원베일리", "city": "서울특별시", "district": "서초구", "neighborhood": "반포동", "aliases": ["반포 원베일리", "원베일리"] }, { "name": "목동신시가지7단지", "city": "서울특별시", "district": "양천구", "neighborhood": "목동", "aliases": ["목동 7단지", "목동신시가지7", "목동신시가지 7단지"] }, { "name": "신월시영아파트", "city": "서울특별시", "district": "양천구", "neighborhood": "신월동", "aliases": ["신월시영", "양천 신월시영"], "note": "tricky alias 케이스" }, { "name": "답십리두산위브", "city": "서울특별시", "district": "동대문구", "neighborhood": "답십리동", "aliases": ["두산위브", "답십리 두산위브", "답십리두산위브아파트"], "note": "실사용 자연어 질의 fallback 강화용. complex ID 수동 확보 전까지 reference seed로 운영" } ] } FILE:scripts/apply_generated_seeds.py from __future__ import annotations import argparse import json import sys import time from pathlib import Path from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) import search_real_estate as sre DEFAULT_GENERATED = SCRIPT_DIR.parent / "references" / "candidate-seeds.generated.json" DEFAULT_TARGET = SCRIPT_DIR.parent / "references" / "candidate-seeds.json" DEFAULT_CACHE = SCRIPT_DIR.parent / "data" / "candidate-cache.json" DEFAULT_INCLUDE_STATUSES = ["verified", "weak-verified"] def _load_json(path: Path, default: Any) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return default def _write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def _dedupe_keep_order(values: list[str]) -> list[str]: out: list[str] = [] seen: set[str] = set() for value in values: item = str(value or "").strip() if not item or item in seen: continue out.append(item) seen.add(item) return out def _normalize_names(values: list[str] | None) -> set[str]: return {sre.normalize_complex_alias(v) for v in (values or []) if sre.normalize_complex_alias(v)} def _merge_aliases(*parts: list[str]) -> list[str]: flat: list[str] = [] for group in parts: flat.extend([str(x) for x in group if str(x).strip()]) expanded: list[str] = [] for item in flat: expanded.extend(sre.expand_alias_variants(item)) return _dedupe_keep_order(expanded) def _build_review_row(result: dict[str, Any], existing: dict[str, Any] | None = None) -> dict[str, Any]: status = str(result.get("verification_status") or "unresolved") seed = result.get("seed_input") or {} complex_id = str(result.get("complex_id") or "").strip() reason_bits = [] if result.get("blocked_reasons"): reason_bits.append("external-blocked:" + ", ".join(str(x) for x in result.get("blocked_reasons") or [])) if not complex_id: reason_bits.append("complex_id 미확보") pool = result.get("candidate_pool") or [] if complex_id and pool: top = pool[0] top_name = ((top.get("complex") or {}).get("name") or top.get("name") or "") if top_name: reason_bits.append(f"top-candidate={top_name} ({complex_id})") if not reason_bits: reason_bits.append(f"verification_status={status}") review_status = { "blocked": "blocked-but-high-priority", "unresolved": "manual-verification-needed", "unverified": "manual-verification-needed", "weak-verified": "review-before-promote", }.get(status, "manual-verification-needed") if existing and existing.get("review_status") == "excluded-for-now": review_status = "excluded-for-now" next_action = ( "direct complex URL/ID를 확보해 name/address 정합성을 다시 확인한 뒤 승격한다." if not complex_id else "preview 결과를 보고 name/address/aliases가 맞으면 --apply-target으로 승격하고, 필요 시 --apply-cache로 warm-cache 한다." ) return { "name": result.get("name") or seed.get("name"), "district": result.get("district") or seed.get("district"), "neighborhood": result.get("neighborhood") or seed.get("neighborhood"), "review_status": review_status, "reason": "; ".join(reason_bits), "next_action": next_action, "verification_status": status, "confidence": result.get("confidence"), "complex_id": complex_id or None, } def _build_entry(result: dict[str, Any], previous: dict[str, Any] | None = None) -> dict[str, Any]: seed = result.get("seed_input") or {} previous_aliases = list((previous or {}).get("aliases") or []) aliases = _merge_aliases( [str(result.get("name") or "")], list(result.get("aliases") or []), list(seed.get("aliases") or []), previous_aliases, ) note_bits = [str((previous or {}).get("note") or "").strip(), str(result.get("note") or "").strip()] note = " | ".join([x for x in note_bits if x]) return { "complex_id": str(result.get("complex_id") or ""), "name": result.get("name") or seed.get("name"), "address": result.get("address") or seed.get("address") or (previous or {}).get("address"), "household_count": result.get("household_count") or (previous or {}).get("household_count"), "aliases": aliases, "verification_status": result.get("verification_status"), "confidence": result.get("confidence"), "source": result.get("source") or "generated:web-search+detail", "learned_from": result.get("learned_from") or "apply_generated_seeds.py", "note": note or "generated seed promoted to production", } def build_plan( generated_payload: dict[str, Any], current_payload: dict[str, Any], *, include_statuses: set[str], min_confidence: float, only_names: set[str], exclude_names: set[str], ) -> dict[str, Any]: results = list(generated_payload.get("results") or []) existing_entries = list(current_payload.get("entries") or []) existing_reviews = list(current_payload.get("manual_review_queue") or []) existing_by_id = {str(row.get("complex_id") or ""): row for row in existing_entries if str(row.get("complex_id") or "").strip()} existing_reviews_by_name = {sre.normalize_complex_alias(row.get("name") or ""): row for row in existing_reviews} accepted: list[dict[str, Any]] = [] rejected: list[dict[str, Any]] = [] review_rows: list[dict[str, Any]] = [] for result in results: name = str(result.get("name") or "").strip() name_key = sre.normalize_complex_alias(name) complex_id = str(result.get("complex_id") or "").strip() status = str(result.get("verification_status") or "") confidence = float(result.get("confidence") or 0.0) if only_names and name_key not in only_names: continue if exclude_names and name_key in exclude_names: continue eligible = bool(complex_id) and status in include_statuses and confidence >= min_confidence if eligible: accepted.append(_build_entry(result, previous=existing_by_id.get(complex_id))) else: rejected.append({ "name": name, "complex_id": complex_id or None, "verification_status": status, "confidence": confidence, }) review_rows.append(_build_review_row(result, existing_reviews_by_name.get(name_key))) preserved_entries: list[dict[str, Any]] = [] accepted_ids = {str(row.get("complex_id") or "") for row in accepted} for row in existing_entries: row_id = str(row.get("complex_id") or "") if row_id and row_id not in accepted_ids: preserved_entries.append(row) merged_entries = sorted( [*preserved_entries, *accepted], key=lambda row: (str(row.get("name") or ""), str(row.get("complex_id") or "")), ) merged_reviews_by_name = {sre.normalize_complex_alias(row.get("name") or ""): row for row in existing_reviews} for row in review_rows: merged_reviews_by_name[sre.normalize_complex_alias(row.get("name") or "")] = row for row in accepted: merged_reviews_by_name.pop(sre.normalize_complex_alias(row.get("name") or ""), None) merged_reviews = sorted(merged_reviews_by_name.values(), key=lambda row: str(row.get("name") or "")) next_payload = { "schema_version": max(2, int(current_payload.get("schema_version") or 2)), "updated_at": int(time.time()), "policy": current_payload.get("policy") or { "production_entry_rule": "verification_status가 verified 또는 weak-verified 이고, complex_id/name/address 정합성이 확인된 항목만 entries에 넣는다.", "excluded_rule": "generated 결과에서 complex_id가 있더라도 이름/주소 검증이 실패했거나 차단/오염 가능성이 있으면 entries에 넣지 않는다.", }, "entries": merged_entries, "manual_review_queue": merged_reviews, "review_summary": { "accepted_for_production": [row.get("name") for row in accepted], "pending_manual_verification": [row.get("name") for row in merged_reviews], "generated_input_count": len(results), "accepted_count": len(accepted), "rejected_count": len(rejected), }, } return { "accepted": accepted, "rejected": rejected, "next_payload": next_payload, } def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="generated candidate seed preview/apply 도우미") p.add_argument("--input", default=str(DEFAULT_GENERATED), help="candidate-seeds.generated.json 경로") p.add_argument("--target", default=str(DEFAULT_TARGET), help="운영 candidate-seeds.json 경로") p.add_argument("--cache-file", default=str(DEFAULT_CACHE), help="candidate-cache.json 경로") p.add_argument("--include-statuses", default=",".join(DEFAULT_INCLUDE_STATUSES), help="승격 허용 상태 목록") p.add_argument("--min-confidence", type=float, default=0.65, help="승격 최소 confidence") p.add_argument("--only-names", default="", help="쉼표 구분 이름 필터") p.add_argument("--exclude-names", default="", help="쉼표 구분 제외 이름") p.add_argument("--apply-target", action="store_true", help="운영 candidate-seeds.json에 반영") p.add_argument("--apply-cache", action="store_true", help="accepted 항목을 candidate-cache에도 warm-cache") p.add_argument("--json", action="store_true", help="JSON으로 출력") p.add_argument("--self-test", action="store_true", help="내장 self-test 실행") return p def run_self_test() -> int: sample_generated = { "results": [ { "name": "리센츠", "district": "송파구", "neighborhood": "잠실동", "complex_id": "1147", "address": "서울특별시 송파구 잠실동", "household_count": 5563, "aliases": ["리센츠", "잠실 리센츠"], "seed_input": {"aliases": ["잠실리센츠"]}, "verification_status": "verified", "confidence": 1.0, "source": "generated:test", "learned_from": "self-test", "note": "ok", }, { "name": "은마", "district": "강남구", "neighborhood": "대치동", "complex_id": "", "aliases": ["은마", "대치 은마"], "verification_status": "blocked", "confidence": 0.0, "blocked_reasons": ["HTTP 403"], }, ] } sample_current = { "schema_version": 2, "entries": [], "manual_review_queue": [], } plan = build_plan( sample_generated, sample_current, include_statuses={"verified", "weak-verified"}, min_confidence=0.65, only_names=set(), exclude_names=set(), ) assert len(plan["accepted"]) == 1 assert plan["accepted"][0]["name"] == "리센츠" assert len(plan["next_payload"]["entries"]) == 1 assert plan["next_payload"]["entries"][0]["complex_id"] == "1147" assert len(plan["next_payload"]["manual_review_queue"]) == 1 assert plan["next_payload"]["manual_review_queue"][0]["name"] == "은마" print("SELF_TEST_OK") print(json.dumps({ "accepted": plan["accepted"], "review_queue": plan["next_payload"]["manual_review_queue"], }, ensure_ascii=False, indent=2)) return 0 def main() -> int: args = build_parser().parse_args() if args.self_test: return run_self_test() input_path = Path(args.input) target_path = Path(args.target) cache_path = Path(args.cache_file) include_statuses = {x.strip() for x in str(args.include_statuses or "").split(",") if x.strip()} only_names = _normalize_names([x.strip() for x in str(args.only_names or "").split(",") if x.strip()]) exclude_names = _normalize_names([x.strip() for x in str(args.exclude_names or "").split(",") if x.strip()]) generated_payload = _load_json(input_path, {"results": []}) current_payload = _load_json(target_path, {"schema_version": 2, "entries": [], "manual_review_queue": []}) plan = build_plan( generated_payload, current_payload, include_statuses=include_statuses, min_confidence=max(0.0, float(args.min_confidence)), only_names=only_names, exclude_names=exclude_names, ) next_payload = plan["next_payload"] cache_result: dict[str, Any] | None = None if args.apply_target: _write_json(target_path, next_payload) if args.apply_cache: previous_cache = sre.CANDIDATE_CACHE_FILE sre.CANDIDATE_CACHE_FILE = cache_path try: cache_saved = sre.seed_candidate_cache(next_payload.get("entries") or [], source=f"generated-seed:{input_path.name}") cache_result = {"saved_count": len(cache_saved), "saved": cache_saved} finally: sre.CANDIDATE_CACHE_FILE = previous_cache summary = { "input": str(input_path), "target": str(target_path), "cache_file": str(cache_path), "applied_target": bool(args.apply_target), "applied_cache": bool(args.apply_cache), "accepted_count": len(plan["accepted"]), "rejected_count": len(plan["rejected"]), "accepted": [ { "name": row.get("name"), "complex_id": row.get("complex_id"), "verification_status": row.get("verification_status"), "confidence": row.get("confidence"), } for row in plan["accepted"] ], "rejected": plan["rejected"], "manual_review_queue_count": len(next_payload.get("manual_review_queue") or []), "cache_result": cache_result, } if args.json or not (args.apply_target or args.apply_cache): print(json.dumps(summary, ensure_ascii=False, indent=2)) else: print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/build_candidate_seeds.py from __future__ import annotations import argparse import json import re import sys import time import urllib.parse from pathlib import Path from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) import search_real_estate as sre DEFAULT_INPUT = SCRIPT_DIR.parent / "references" / "seoul-major-complexes.seed-input.json" DEFAULT_OUTPUT = SCRIPT_DIR.parent / "references" / "candidate-seeds.generated.json" SEARCH_URL = "https://search.naver.com/search.naver?where=nexearch&sm=top_hty&fbm=0&ie=utf8&query={query}" COMPLEX_PATTERNS = [ r"https://new\.land\.naver\.com/complexes/(\d+)", r"https://new\.land\.naver\.com/houses/(\d+)", r"complexNo=(\d+)", ] def _load_json(path: Path, default: Any) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return default def _write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def _dedupe_keep_order(values: list[str]) -> list[str]: out: list[str] = [] seen: set[str] = set() for value in values: key = str(value or "").strip() if not key or key in seen: continue out.append(key) seen.add(key) return out def build_aliases(name: str, district: str = "", neighborhood: str = "", city: str = "서울특별시", extra: list[str] | None = None) -> list[str]: base = str(name or "").strip() variants: list[str] = [] for seed in [base, *(extra or [])]: for value in sre.expand_alias_variants(seed): variants.append(value) place_bits = [x.strip() for x in [city, district, neighborhood] if str(x or "").strip()] short_city = city.replace("특별시", "").replace("광역시", "").strip() if city else "" if short_city: place_bits.append(short_city) normalized = sre.normalize_complex_alias(base) for prefix in _dedupe_keep_order(place_bits): if base: variants.extend([f"{prefix} {base}", f"{prefix}{base}"]) if normalized: variants.extend([f"{prefix} {normalized}", f"{prefix}{normalized}"]) if district and neighborhood: variants.extend([f"{district} {neighborhood} {base}", f"{district}{neighborhood}{base}"]) return _dedupe_keep_order([v for item in variants for v in sre.expand_alias_variants(item)])[:30] def extract_ids_from_html(query: str, *, limit: int = 8) -> dict[str, Any]: url = SEARCH_URL.format(query=urllib.parse.quote(query)) try: html = sre._request_text(url) # type: ignore[attr-defined] except Exception as exc: return {"ids": [], "status": "blocked", "error": str(exc), "query": query} ids: list[str] = [] seen: set[str] = set() for pattern in COMPLEX_PATTERNS: for match in re.finditer(pattern, html): cid = match.group(1) if cid not in seen: ids.append(cid) seen.add(cid) if len(ids) >= limit: break if len(ids) >= limit: break return {"ids": ids, "status": "ok", "query": query} def verify_candidate(complex_id: str, seed: dict[str, Any]) -> dict[str, Any]: try: info = sre.fetch_complex_info(complex_id) name = str(info.get("name") or "") address = str(info.get("address") or "") score = 0.0 reasons: list[str] = [] seed_name_norm = sre.normalize_complex_alias(seed.get("name") or "") found_name_norm = sre.normalize_complex_alias(name) district = str(seed.get("district") or "") neighborhood = str(seed.get("neighborhood") or "") if seed_name_norm and found_name_norm == seed_name_norm: score += 0.55 reasons.append("name-exact") elif seed_name_norm and seed_name_norm in found_name_norm: score += 0.35 reasons.append("name-partial") if district and district in address: score += 0.15 reasons.append("district-match") if neighborhood and neighborhood in address: score += 0.15 reasons.append("neighborhood-match") household_count = info.get("household_count") try: household_count = int(household_count or 0) except Exception: household_count = 0 if household_count >= 500: score += 0.05 reasons.append("household-signal") status = "verified" if score >= 0.65 else "weak-verified" return { "status": status, "confidence": round(min(1.0, score), 3), "verification_reasons": reasons, "complex": info, } except Exception as exc: return { "status": "unverified", "confidence": 0.0, "verification_reasons": [f"detail-fetch-failed:{type(exc).__name__}"], "error": str(exc), "complex": {"complex_id": complex_id}, } def build_search_queries(seed: dict[str, Any]) -> list[str]: city = str(seed.get("city") or "서울특별시") district = str(seed.get("district") or "") neighborhood = str(seed.get("neighborhood") or "") name = str(seed.get("name") or "") aliases = [str(x) for x in (seed.get("aliases") or []) if str(x).strip()] queries: list[str] = [] for label in [name, *aliases]: label = label.strip() if not label: continue queries.extend([ f"{city} {district} {neighborhood} {label} 네이버 부동산", f"{district} {neighborhood} {label} 네이버 부동산", f"{district} {label} 네이버 부동산", f"{label} 네이버 부동산 아파트", f"site:new.land.naver.com/complexes {district} {label}", ]) return _dedupe_keep_order([q.strip() for q in queries if q.strip()])[:12] def build_entry(seed: dict[str, Any], *, pause: float = 0.6) -> dict[str, Any]: aliases = build_aliases( seed.get("name") or "", district=seed.get("district") or "", neighborhood=seed.get("neighborhood") or "", city=seed.get("city") or "서울특별시", extra=[str(x) for x in (seed.get("aliases") or []) if str(x).strip()], ) evidence: list[dict[str, Any]] = [] candidates: list[dict[str, Any]] = [] seen_ids: set[str] = set() blocked_reasons: list[str] = [] baseline_payload = _load_json(SCRIPT_DIR.parent / "references" / "candidate-seeds.json", {"entries": []}) baseline_entries = baseline_payload.get("entries") if isinstance(baseline_payload, dict) else [] seed_name_norm = sre.normalize_complex_alias(seed.get("name") or "") for row in baseline_entries or []: row_name_norm = sre.normalize_complex_alias(row.get("name") or "") row_aliases = [sre.normalize_complex_alias(x) for x in (row.get("aliases") or [])] if seed_name_norm and (seed_name_norm == row_name_norm or seed_name_norm in row_aliases): candidates.append({ "complex_id": str(row.get("complex_id") or ""), "query": seed.get("name") or "", "discovery_source": "baseline-seed", "status": "verified", "confidence": 1.0, "verification_reasons": ["baseline-seed-match"], "complex": { "complex_id": str(row.get("complex_id") or ""), "name": row.get("name"), "address": row.get("address"), "household_count": row.get("household_count"), }, }) seen_ids.add(str(row.get("complex_id") or "")) evidence.append({"query": seed.get("name") or "", "status": "baseline-seed-hit", "ids": [row.get("complex_id")]}) break def _append_candidate(cid: str, query: str, source: str) -> None: if not cid or cid in seen_ids: return seen_ids.add(cid) verified = verify_candidate(cid, seed) candidates.append({ "complex_id": cid, "query": query, "discovery_source": source, **verified, }) for query in build_search_queries(seed): try: cached = sre.search_cached_candidates(query, candidate_limit=3) except Exception as exc: cached = [] evidence.append({"query": query, "status": "cache-error", "error": str(exc)}) else: if cached: evidence.append({ "query": query, "status": "cache-hit", "source": "candidate-cache", "ids": [row.get("complex_id") for row in cached], }) for row in cached: _append_candidate(str(row.get("complex_id") or ""), query, "candidate-cache") if candidates and float(candidates[0].get("confidence") or 0.0) >= 0.8: break snapshot = extract_ids_from_html(query, limit=6) evidence.append(snapshot) if snapshot.get("status") != "ok": blocked_reasons.append(str(snapshot.get("error") or snapshot.get("status") or "blocked")) time.sleep(pause) continue for cid in snapshot.get("ids") or []: _append_candidate(cid, query, "html-extract") time.sleep(pause) time.sleep(pause) candidates.sort(key=lambda row: (-float(row.get("confidence") or 0.0), str(row.get("complex", {}).get("name") or ""), str(row.get("complex_id") or ""))) chosen = candidates[0] if candidates else None verification_status = "blocked" if (not chosen and blocked_reasons) else (chosen.get("status") if chosen else "unresolved") confidence = float(chosen.get("confidence") or 0.0) if chosen else 0.0 complex_info = (chosen or {}).get("complex") or {} return { "name": seed.get("name"), "district": seed.get("district"), "neighborhood": seed.get("neighborhood"), "seed_input": seed, "complex_id": complex_info.get("complex_id") or (chosen or {}).get("complex_id"), "address": complex_info.get("address") or seed.get("address") or "", "household_count": complex_info.get("household_count"), "aliases": aliases, "verification_status": verification_status, "confidence": round(confidence, 3), "source": "generated:web-search+detail", "learned_from": "build_candidate_seeds.py", "note": seed.get("note") or "generated candidate seed", "candidate_pool": candidates[:5], "evidence": evidence[:8], "blocked_reasons": _dedupe_keep_order(blocked_reasons)[:5], } def build_generated_payload(seed_items: list[dict[str, Any]], *, pause: float = 0.6) -> dict[str, Any]: generated = [build_entry(seed, pause=pause) for seed in seed_items] entries = [ { "complex_id": row.get("complex_id"), "name": row.get("name"), "address": row.get("address"), "household_count": row.get("household_count"), "aliases": row.get("aliases") or [], "note": row.get("note"), "source": row.get("source"), "learned_from": row.get("learned_from"), "confidence": row.get("confidence"), "verification_status": row.get("verification_status"), } for row in generated if row.get("complex_id") and row.get("verification_status") in {"verified", "weak-verified"} ] unresolved = [ { "name": row.get("name"), "district": row.get("district"), "neighborhood": row.get("neighborhood"), "aliases": row.get("aliases") or [], "verification_status": row.get("verification_status"), "confidence": row.get("confidence"), "blocked_reasons": row.get("blocked_reasons") or [], } for row in generated if not row.get("complex_id") ] return { "kind": "naver-real-estate-candidate-seed-generation", "schema_version": 1, "generated_at": int(time.time()), "source_input": str(DEFAULT_INPUT), "entry_count": len(entries), "unresolved_count": len(unresolved), "entries": entries, "unresolved": unresolved, "results": generated, } def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="네이버 부동산 candidate seed 자동 생성기") p.add_argument("--input", default=str(DEFAULT_INPUT), help="seed 입력 JSON 파일") p.add_argument("--output", default=str(DEFAULT_OUTPUT), help="생성 결과 JSON 파일") p.add_argument("--pause", type=float, default=0.7, help="요청 사이 pause seconds") p.add_argument("--print-summary", action="store_true", help="간단 요약 출력") return p def main() -> int: args = build_parser().parse_args() input_path = Path(args.input) output_path = Path(args.output) payload = _load_json(input_path, {"seeds": []}) seed_items = payload.get("seeds") if isinstance(payload, dict) else payload if not seed_items: raise SystemExit(f"seed 입력이 비어 있습니다: {input_path}") generated = build_generated_payload(list(seed_items), pause=max(0.0, float(args.pause))) generated["source_input"] = str(input_path) _write_json(output_path, generated) if args.print_summary: print(json.dumps({ "output": str(output_path), "entry_count": generated.get("entry_count"), "unresolved_count": generated.get("unresolved_count"), "resolved": [ { "name": row.get("name"), "complex_id": row.get("complex_id"), "confidence": row.get("confidence"), "verification_status": row.get("verification_status"), } for row in generated.get("results", []) ], }, ensure_ascii=False, indent=2)) else: print(str(output_path)) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/chat_real_estate.py from __future__ import annotations import argparse import json from typing import Any from search_real_estate import PriceConverter, run_query, search_complex_candidates def _fmt_price(value: int | None) -> str: return PriceConverter.to_string(value) if value else "-" def _pick_headline(meta: dict[str, Any], trade_type: str) -> str: count = int(meta.get("count") or 0) if count == 0: return f"{trade_type} 매물이 거의 안 잡힙니다." if count <= 3: return f"{trade_type} 매물이 많지는 않지만 확인 가능한 건은 있습니다." if count <= 10: return f"{trade_type} 매물이 어느 정도 보입니다." return f"{trade_type} 매물이 비교적 넉넉하게 잡힙니다." def _trend_line(meta: dict[str, Any]) -> str: min_price = meta.get("min_price") avg_price = meta.get("avg_price") max_price = meta.get("max_price") if not (min_price and avg_price and max_price): return "가격대 해석용 표본은 아직 제한적입니다." spread = max_price - min_price if spread <= max(1, avg_price * 0.05): return "가격 분산이 크지 않아 호가대가 비교적 촘촘합니다." if spread >= max(1, avg_price * 0.15): return "호가 폭이 꽤 넓어서 같은 단지 안에서도 조건 차이가 크게 보입니다." return "호가 차이는 있으나 극단적으로 벌어지지는 않습니다." def _representative_lines(items: list[dict[str, Any]], max_items: int = 3) -> list[str]: lines: list[str] = [] for row in items[:max_items]: price = row.get("매매가") or row.get("보증금") or "-" if row.get("거래유형") == "월세" and row.get("월세"): price = f"{price}/{row.get('월세')}" parts = [f"{price}", f"{row.get('면적(평)', 0)}평", row.get("층/방향", "-") or "-"] lines.append(f" · 대표 매물: {', '.join(parts)}") if row.get("특징"): lines.append(f" - 포인트: {row['특징']}") if row.get("매물URL"): lines.append(f" - 링크: {row['매물URL']}") return lines def brief_single(payload: dict[str, Any]) -> str: info = payload.get("complex_info") or {} parsed = payload.get("parsed") or {} lines = [f"{info.get('name', payload.get('selected_complex_id'))} 브리핑"] if info.get("address"): lines.append(f"- 위치: {info['address']}") filters = [] if payload.get("trade_types"): filters.append("거래유형 " + ", ".join(payload["trade_types"])) if parsed.get("min_pyeong") or parsed.get("max_pyeong"): filters.append(f"평형 {parsed.get('min_pyeong', '-')}~{parsed.get('max_pyeong', '-')}평") if filters: lines.append(f"- 필터: {' / '.join(filters)}") summary = payload.get("market_summary") or {} if not summary: lines.append("- 조건에 맞는 매물이 아직 잡히지 않았습니다.") return "\n".join(lines) for trade_type, meta in summary.items(): lines.append(f"- {trade_type}: {meta.get('count', 0)}건") lines.append(f" · {_pick_headline(meta, trade_type)}") lines.append(f" · 가격대: 최저 {_fmt_price(meta.get('min_price'))} / 평균 {_fmt_price(meta.get('avg_price'))} / 중앙값 {_fmt_price(meta.get('median_price'))} / 최고 {_fmt_price(meta.get('max_price'))}") lines.append(f" · 해석: {_trend_line(meta)}") area_rows = meta.get("area_summary") or [] if area_rows: head = area_rows[0] lines.append(f" · 동일 평형 대표: {head.get('area_key')} {head.get('count')}건, 평균 {_fmt_price(head.get('avg_price'))}") lines.extend(_representative_lines(payload.get("items", []))) return "\n".join(lines) def brief_compare(payload: dict[str, Any]) -> str: results = payload.get("results") or [] insights = payload.get("compare_insights") or {} lines = ["단지 비교 브리핑"] for result in results: info = result.get("complex_info") or {} name = info.get("name", result.get("complex_id")) address = info.get("address") or "-" lines.append(f"- {name} ({address})") summary = result.get("market_summary") or {} if not summary: lines.append(" · 조건에 맞는 매물이 거의 안 보입니다.") continue for trade_type, meta in summary.items(): lines.append(f" · {trade_type}: {meta.get('count', 0)}건 / 최저 {_fmt_price(meta.get('min_price'))} / 평균 {_fmt_price(meta.get('avg_price'))} / 중앙값 {_fmt_price(meta.get('median_price'))} / 최고 {_fmt_price(meta.get('max_price'))}") area_rows = meta.get("area_summary") or [] if area_rows: head = area_rows[0] lines.append(f" - 대표 동일 평형: {head.get('area_key')} {head.get('count')}건 / 평균 {_fmt_price(head.get('avg_price'))}") for trade_type, meta in (insights.get("trade") or {}).items(): lines.append(f"- 종합 해석 ({trade_type}): {meta['cheapest']['name']} 쪽 평균이 가장 낮고 {meta['most_expensive']['name']} 쪽이 가장 높습니다. 전체 평균 차이는 {_fmt_price(meta['gap'])} 정도입니다.") for trade_type, area_rows in (insights.get("same_area") or {}).items(): if not area_rows: continue head = area_rows[0] lines.append(f"- 동일 평형 해석 ({trade_type} {head['area_key']}): {head['cheapest']['name']} 쪽이 더 낮고 {head['most_expensive']['name']} 쪽이 더 높습니다. 같은 평형 기준 차이는 {_fmt_price(head['gap'])} 정도입니다.") if payload.get("meta", {}).get("rate_limited"): lines.append("- 참고: 현재 네이버 요청 제한이 감지돼 direct URL/complex ID 기반 조회가 더 안정적일 수 있습니다.") return "\n".join(lines) def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="네이버 부동산 채팅형 브리핑 래퍼") p.add_argument("--query", help="자연어 질의") p.add_argument("--complex-id") p.add_argument("--url") p.add_argument("--trade-types", default="") p.add_argument("--pages", type=int, default=1) p.add_argument("--limit", type=int, default=10) p.add_argument("--candidate-limit", type=int, default=3) p.add_argument("--min-pyeong", type=float) p.add_argument("--max-pyeong", type=float) p.add_argument("--list-candidates", action="store_true") p.add_argument("--compare", action="store_true") p.add_argument("--json", action="store_true") return p def main() -> int: args = build_parser().parse_args() if args.list_candidates: if not args.query: raise SystemExit("--list-candidates 는 --query 와 함께 사용하세요.") candidates = search_complex_candidates(args.query, candidate_limit=max(1, args.candidate_limit)) if args.json: print(json.dumps(candidates, ensure_ascii=False, indent=2)) else: lines = ["후보 단지"] for idx, row in enumerate(candidates, start=1): lines.append(f"- {idx}. {row.get('name')} | {row.get('address') or '-'} | ID {row.get('complex_id')} | score {row.get('match_score', '-')} | source {row.get('source', '-')}") print("\n".join(lines)) return 0 trade_types = [token.strip() for token in str(args.trade_types).split(",") if token.strip()] payload = run_query( query=args.query, complex_id=args.complex_id, url=args.url, trade_types=trade_types, pages=max(1, args.pages), limit=max(1, args.limit), candidate_limit=max(1, args.candidate_limit), min_pyeong=args.min_pyeong, max_pyeong=args.max_pyeong, compare=bool(args.compare), ) if args.json: print(json.dumps(payload, ensure_ascii=False, indent=2)) else: print(brief_compare(payload) if payload.get("compare_mode") else brief_single(payload)) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/search_real_estate.py from __future__ import annotations import argparse import json import math import re import statistics import sys import time import urllib.parse import urllib.request from dataclasses import asdict, dataclass from pathlib import Path from typing import Any WORKSPACE = Path(__file__).resolve().parents[3] UPSTREAM = WORKSPACE / "tmp" / "naverland-scrapper" SRC_ROOT = UPSTREAM / "src" if str(UPSTREAM) not in sys.path: sys.path.insert(0, str(UPSTREAM)) if str(SRC_ROOT) not in sys.path: sys.path.insert(0, str(SRC_ROOT)) UPSTREAM_IMPORT_ERROR: Exception | None = None try: from src.core.parser import NaverURLParser from src.core.services.response_capture import normalize_article_payload from src.utils.helpers import PriceConverter, get_article_url except Exception as exc: UPSTREAM_IMPORT_ERROR = exc class NaverURLParser: @staticmethod def extract_from_text(text: str) -> list[tuple[str, str]]: pairs: list[tuple[str, str]] = [] for cid in RAW_COMPLEX_ID_RE.findall(text or ""): pairs.append(("", cid)) return pairs @staticmethod def extract_complex_id(text: str | None) -> str | None: match = RAW_COMPLEX_ID_RE.search(text or "") return match.group(1) if match else None @staticmethod def fetch_complex_name(complex_id: str) -> str: _raise_missing_upstream() class PriceConverter: @staticmethod def to_int(value: Any) -> int: raw = str(value or "").strip() if not raw: return 0 digits = re.sub(r"[^0-9]", "", raw) return int(digits) if digits else 0 @staticmethod def to_string(value: Any) -> str: amount = PriceConverter.to_int(value) if amount <= 0: return "0" eok = amount // 100000000 rem = amount % 100000000 man = rem // 10000 if eok and man: return f"{eok}억 {man:,}만" if eok: return f"{eok}억" return f"{man:,}만" def get_article_url(complex_id: str, article_id: str, real_estate_type: str = "APT") -> str: article = str(article_id or "").strip() if not article: return "" return f"https://new.land.naver.com/articles/{article}?complexNo={complex_id}&realEstateType={real_estate_type}" def normalize_article_payload(*args: Any, **kwargs: Any) -> dict[str, Any]: _raise_missing_upstream() def _raise_missing_upstream() -> None: detail = f" ({UPSTREAM_IMPORT_ERROR})" if UPSTREAM_IMPORT_ERROR else "" raise RuntimeError( "필수 upstream clone(tmp/naverland-scrapper)이 없거나 불완전합니다. " "이 스킬은 해당 로컬 저장소의 src 패키지에 의존합니다. " f"경로를 확인한 뒤 다시 시도해 주세요{detail}" ) USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36" SEARCH_URL = "https://search.naver.com/search.naver?where=nexearch&sm=top_hty&fbm=0&ie=utf8&query={query}" COMPLEX_DETAIL_URL = "https://new.land.naver.com/api/complexes/{complex_id}?sameAddressGroup=false" COMPLEX_ARTICLE_URL = ( "https://new.land.naver.com/api/articles/complex/{complex_id}?" "realEstateType=APT%3AVL&tradeType={trade_codes}&tag=%3A%3A%3A%3A%3A%3A%3A%3A" "&rentPriceMin=0&rentPriceMax=900000000&priceMin=0&priceMax=900000000" "&areaMin=0&areaMax=900000000&oldBuildYears=&recentlyBuildYears=&minHouseHoldCount=" "&maxHouseHoldCount=&showArticle=false&sameAddressGroup=false&minMaintenanceCost=&maxMaintenanceCost=" "&priceType=RETAIL&directions=&page={page}&complexNo={complex_id}&buildingNos=&areaNos=&type=list&order=rank" ) TRADE_CODE_MAP = {"매매": "A1", "전세": "B1", "월세": "B2"} DEFAULT_QUERY_SUFFIX = " 네이버 부동산 아파트" DEFAULT_BACKOFFS = [1.2, 2.5, 4.0] STOPWORDS = [ "네이버 부동산", "부동산", "시세", "매물", "가격", "가격대", "얼마", "비교", "정리", "요약", "알려줘", "찾아줘", "보여줘", "조회", "검색", "추천", "빌라", "오피스텔", "실거래가", "단지", "찾기", "브리핑", "아파트좀", "해줘", "해주세요", "알림", "감시", "체크", "체크해줘", "요청", "보고", "리포트", "채팅", "래퍼", "근처", "살펴봐", "알아봐", "부탁", "정도", "위주", "기준", "기반", "정도면", "좀", "그리고", "중에서", "해석", ] TRADE_STOPWORDS = ["매매", "전세", "월세"] COMPARE_TOKENS = ["비교", "대비", "vs", "VS"] LOCATION_SUFFIXES = ("특별시", "광역시", "시", "도", "군", "구", "동", "읍", "면", "리", "가") SIMPLE_KOREAN_TOKEN_RE = re.compile(r"[가-힣A-Za-z0-9]{2,}") RAW_COMPLEX_ID_RE = re.compile(r"(?:complex(?:\s*id|no)?|단지(?:\s*id)?|id)\s*[:=#-]?\s*(\d{3,10})", re.I) WATCH_STATE_FILE = WORKSPACE / "skills" / "naver-real-estate-search" / "data" / "watch-rules.json" CANDIDATE_CACHE_FILE = WORKSPACE / "skills" / "naver-real-estate-search" / "data" / "candidate-cache.json" DEFAULT_CANDIDATE_SEED_FILE = WORKSPACE / "skills" / "naver-real-estate-search" / "references" / "candidate-seeds.json" DEFAULT_SEED_INPUT_FILE = WORKSPACE / "skills" / "naver-real-estate-search" / "references" / "seoul-major-complexes.seed-input.json" APT_SUFFIX_RE = re.compile(r"(?:아파트|맨션|타운하우스|주상복합|오피스텔|빌라)$") AREA_RANGE_RE = re.compile(r"(\d{1,2})\s*평\s*[~-]\s*(\d{1,2})\s*평") AREA_BAND_RE = re.compile(r"(\d{1,2})\s*평대") AREA_SINGLE_RE = re.compile(r"(\d{1,2})\s*평(?:형)?") NOISE_TOKENS = set(STOPWORDS + TRADE_STOPWORDS + [ "서울", "경기", "인천", "부산", "대구", "대전", "광주", "울산", "세종", "대한민국", "한국", "수도권" ]) REGION_ALIASES = { "서울": ["서울", "서울시", "서울특별시"], "부산": ["부산", "부산시", "부산광역시"], "인천": ["인천", "인천시", "인천광역시"], "대구": ["대구", "대구시", "대구광역시"], "광주": ["광주", "광주시", "광주광역시"], "대전": ["대전", "대전시", "대전광역시"], "울산": ["울산", "울산시", "울산광역시"], "세종": ["세종", "세종시", "세종특별자치시"], "경기": ["경기", "경기도"], "강원": ["강원", "강원도", "강원특별자치도"], "충북": ["충북", "충청북도"], "충남": ["충남", "충청남도"], "전북": ["전북", "전라북도", "전북특별자치도"], "전남": ["전남", "전라남도"], "경북": ["경북", "경상북도"], "경남": ["경남", "경상남도"], "제주": ["제주", "제주도", "제주특별자치도"], } class SearchError(RuntimeError): pass @dataclass class ParsedQuery: raw_query: str cleaned_query: str trade_types: list[str] min_pyeong: float | None max_pyeong: float | None compare_mode: bool candidate_keywords: list[str] location_hints: list[str] raw_subjects: list[str] direct_complex_ids: list[str] RATE_LIMIT_STATE = {"active": False, "last_error": None} def _record_rate_limit(message: str) -> None: RATE_LIMIT_STATE["active"] = True RATE_LIMIT_STATE["last_error"] = message def _request_json(url: str, *, referer: str = "https://new.land.naver.com/", backoffs: list[float] | None = None) -> Any: backoffs = DEFAULT_BACKOFFS if backoffs is None else backoffs req = urllib.request.Request(url) req.add_header("User-Agent", USER_AGENT) req.add_header("Referer", referer) req.add_header("Accept", "application/json, text/plain, */*") attempts = len(backoffs) + 1 for attempt in range(attempts): try: with urllib.request.urlopen(req, timeout=20) as response: RATE_LIMIT_STATE["active"] = False return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="ignore") if exc.code == 429 and attempt < len(backoffs): _record_rate_limit("429") time.sleep(backoffs[attempt]) continue if exc.code == 429: _record_rate_limit("429") raise SearchError( "네이버 부동산 API가 429(요청 제한)를 반환했습니다. 단일 단지 URL/ID를 우선 사용하거나, 후보 단지부터 1~3개로 좁혀 다시 시도해 주세요." ) raise SearchError(f"네이버 부동산 API 호출 실패: HTTP {exc.code} {body[:200]}") except Exception as exc: raise SearchError(f"네이버 부동산 API 호출 실패: {exc}") from exc raise SearchError("네이버 부동산 API 호출 실패") def _request_text(url: str) -> str: req = urllib.request.Request(url) req.add_header("User-Agent", USER_AGENT) req.add_header("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8") try: with urllib.request.urlopen(req, timeout=20) as response: return response.read().decode("utf-8", errors="ignore") except urllib.error.HTTPError as exc: if exc.code in {403, 429}: raise SearchError(f"네이버 검색 HTML 후보 탐색이 차단되었습니다: HTTP {exc.code}") raise def _read_json_file(path: Path, default: Any) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return default def _write_json_file(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def _read_candidate_cache() -> dict[str, Any]: data = _read_json_file(CANDIDATE_CACHE_FILE, {"version": 3, "updated_at": 0, "entries": []}) if isinstance(data, list): data = {"version": 3, "updated_at": 0, "entries": data} data.setdefault("version", 3) data.setdefault("updated_at", 0) data.setdefault("entries", []) return data def _write_candidate_cache(data: dict[str, Any]) -> None: data = dict(data) data["updated_at"] = int(time.time()) _write_json_file(CANDIDATE_CACHE_FILE, data) def normalize_keyword(text: str) -> str: value = str(text or "").strip() for token in STOPWORDS + TRADE_STOPWORDS: value = value.replace(token, " ") value = re.sub(r"$[^)]*$", " ", value) value = AREA_RANGE_RE.sub(" ", value) value = AREA_BAND_RE.sub(" ", value) value = AREA_SINGLE_RE.sub(" ", value) value = re.sub(r"[?!.]", " ", value) value = re.sub(r"\s+", " ", value).strip(" ,/") return value def normalize_complex_alias(text: str) -> str: value = normalize_keyword(text) value = APT_SUFFIX_RE.sub("", value.strip()) value = re.sub(r"\s+", "", value) return value.strip() def expand_region_aliases(value: str) -> list[str]: out: list[str] = [] for canonical, aliases in REGION_ALIASES.items(): if value in aliases or canonical == value: for alias in aliases + [canonical]: if alias not in out: out.append(alias) return out def expand_alias_variants(text: str) -> list[str]: raw = str(text or "").strip() normalized = normalize_complex_alias(raw) spaced = normalize_keyword(raw) variants: list[str] = [] for value in [raw, spaced, normalized]: value = str(value or "").strip() if value and value not in variants: variants.append(value) if normalized: for suffix in ["아파트", "", "단지"]: value = f"{normalized}{suffix}" if suffix else normalized if value and value not in variants: variants.append(value) for token in list(variants): for alias in expand_region_aliases(token): if alias not in variants: variants.append(alias) return variants[:12] def parse_trade_types(query: str) -> list[str]: hits = [trade for trade in ["매매", "전세", "월세"] if trade in query] return hits or ["전세"] def parse_pyeong_range(query: str) -> tuple[float | None, float | None]: m = AREA_BAND_RE.search(query) if m: base = float(m.group(1)) return max(0.0, base - 3), base + 3 m = AREA_RANGE_RE.search(query) if m: return float(m.group(1)), float(m.group(2)) single = AREA_SINGLE_RE.search(query) if single: base = float(single.group(1)) return max(0.0, base - 1), base + 1 return None, None def _looks_like_location(token: str) -> bool: return any(token.endswith(suffix) for suffix in LOCATION_SUFFIXES) or token in REGION_ALIASES def extract_location_hints(query: str) -> list[str]: seen: set[str] = set() results: list[str] = [] spaced = normalize_keyword(query) for token in re.split(r"\s+", spaced): token = token.strip() if len(token) < 2 or token in NOISE_TOKENS: continue if _looks_like_location(token): for alias in [token, *expand_region_aliases(token)]: if alias not in seen: results.append(alias) seen.add(alias) for token in SIMPLE_KOREAN_TOKEN_RE.findall(spaced): if token in NOISE_TOKENS or len(token) < 2: continue if token.endswith("동") or token.endswith("구") or token.endswith("시") or token in {"잠실", "대치", "반포", "신월", "목동", "상계", "중계"}: if token not in seen: results.append(token) seen.add(token) return results[:8] def split_query_subjects(query: str) -> list[str]: cleaned = normalize_keyword(query) if not cleaned: return [] parts = [part.strip() for part in re.split(r"\s*(?:,|/|\||vs\.?|대비|와|과|랑|및)\s*", cleaned) if part.strip()] return parts[:6] def split_candidate_keywords(query: str) -> list[str]: uniq: list[str] = [] seen: set[str] = set() global_locations = extract_location_hints(query) removable_locations = list(global_locations) for location in list(global_locations): removable_locations.extend(expand_region_aliases(location)) removable_locations.extend(["서울", "서울시", "서울특별시", "경기", "경기도"]) for part in split_query_subjects(query): value = part.strip() for location in removable_locations: value = value.replace(location, " ") value = normalize_keyword(value) for variant in expand_alias_variants(value): norm = normalize_complex_alias(variant) if norm and norm not in seen: uniq.append(variant) seen.add(norm) return uniq[:10] def extract_direct_complex_ids(text: str) -> list[str]: direct_complex_ids: list[str] = [] seen: set[str] = set() for _, cid in NaverURLParser.extract_from_text(text or ""): if cid and cid not in seen: direct_complex_ids.append(cid) seen.add(cid) for match in RAW_COMPLEX_ID_RE.finditer(text or ""): cid = match.group(1) if cid and cid not in seen: direct_complex_ids.append(cid) seen.add(cid) text_clean = str(text or "").strip() if re.fullmatch(r"\d{3,10}", text_clean) and text_clean not in seen: direct_complex_ids.append(text_clean) return direct_complex_ids def build_direct_lookup_payload(query: str | None, complex_id: str | None, url: str | None) -> dict[str, Any]: parts = [str(x or "") for x in [query, complex_id, url] if str(x or "").strip()] direct_ids = extract_direct_complex_ids("\n".join(parts)) chosen = str(complex_id or "").strip() or (NaverURLParser.extract_complex_id(url) if url else None) or (direct_ids[0] if direct_ids else None) return { "query": query, "explicit_complex_id": complex_id, "explicit_url": url, "detected_complex_ids": direct_ids, "selected_complex_id": chosen, "canonical_complex_url": f"https://new.land.naver.com/complexes/{chosen}" if chosen else None, } def parse_natural_query(query: str) -> ParsedQuery: min_pyeong, max_pyeong = parse_pyeong_range(query) subject_parts = split_query_subjects(query) location_hints = extract_location_hints(query) candidate_keywords = split_candidate_keywords(query) compare_mode = any(token in query for token in COMPARE_TOKENS) or len(subject_parts) >= 2 cleaned_query = normalize_keyword(query) direct_complex_ids = extract_direct_complex_ids(query) return ParsedQuery( raw_query=query, cleaned_query=cleaned_query, trade_types=parse_trade_types(query), min_pyeong=min_pyeong, max_pyeong=max_pyeong, compare_mode=compare_mode, candidate_keywords=candidate_keywords or ([cleaned_query] if cleaned_query else []), location_hints=location_hints, raw_subjects=subject_parts, direct_complex_ids=direct_complex_ids, ) def extract_complex_candidates_from_web(query: str, limit: int = 5) -> list[dict[str, str]]: html = _request_text(SEARCH_URL.format(query=urllib.parse.quote(query + DEFAULT_QUERY_SUFFIX))) ids: list[str] = [] seen: set[str] = set() patterns = [ r"https://new\.land\.naver\.com/complexes/(\d+)", r"https://new\.land\.naver\.com/houses/(\d+)", r"complexNo=(\d+)", ] for pattern in patterns: for match in re.finditer(pattern, html): cid = match.group(1) if cid not in seen: ids.append(cid) seen.add(cid) if len(ids) >= limit: break if len(ids) >= limit: break return [{"complex_id": cid, "source": "web-search", "query": query} for cid in ids] def fetch_complex_info(complex_id: str) -> dict[str, Any]: payload = _request_json(COMPLEX_DETAIL_URL.format(complex_id=complex_id)) info = payload.get("complexDetail") or payload address = " ".join(filter(None, [str(info.get("cortarAddress") or "").strip(), str(info.get("roadAddressPrefix") or "").strip()])).strip() result = { "complex_id": complex_id, "name": str(info.get("complexName") or info.get("complexNm") or f"단지_{complex_id}"), "address": address, "household_count": info.get("totalHouseHoldCount") or info.get("houseHoldCount"), "complex_url": f"https://new.land.naver.com/complexes/{complex_id}", } remember_candidate(result) return result def _tokenize_for_match(text: str) -> list[str]: normalized = normalize_keyword(text) return [token for token in re.split(r"\s+", normalized) if token and len(token) >= 2] def _score_candidate(info: dict[str, Any], keyword: str, parsed: ParsedQuery | None) -> int: score = 0 name = str(info.get("name") or "") address = str(info.get("address") or "") haystack = f"{name} {address}" normalized_name = normalize_complex_alias(name) normalized_keyword = normalize_complex_alias(keyword) query_alias = normalize_complex_alias(parsed.cleaned_query if parsed else keyword) keyword_tokens = _tokenize_for_match(keyword) query_tokens = _tokenize_for_match(parsed.cleaned_query if parsed else keyword) if normalized_keyword and normalized_keyword == normalized_name: score += 260 elif normalized_keyword and normalized_keyword in normalized_name: score += 130 if query_alias and query_alias == normalized_name: score += 180 elif query_alias and query_alias in normalized_name: score += 85 for token in keyword_tokens: if token == name: score += 120 elif token in name: score += 55 elif token in address: score += 22 elif token in haystack: score += 10 for token in query_tokens: if token in name: score += 14 elif token in address: score += 8 for alias in info.get("aliases") or []: alias_norm = normalize_complex_alias(alias) if query_alias and alias_norm == query_alias: score += 170 elif query_alias and query_alias in alias_norm: score += 70 if normalized_keyword and alias_norm == normalized_keyword: score += 140 if parsed: for location in parsed.location_hints: if location and location in address: score += 32 elif location and location in name: score += 16 if parsed.raw_subjects: own = normalize_complex_alias(parsed.raw_subjects[0]) if own and own == normalized_name: score += 50 try: household_count = int(info.get("household_count") or 0) except Exception: household_count = 0 if household_count >= 300: score += 4 if household_count >= 800: score += 4 if household_count >= 2000: score += 3 return score def remember_candidate(info: dict[str, Any], aliases: list[str] | None = None) -> dict[str, Any]: complex_id = str(info.get("complex_id") or "").strip() if not complex_id: return {} cache = _read_candidate_cache() entries = cache.get("entries") or [] by_id = {str(row.get("complex_id") or ""): dict(row) for row in entries if row.get("complex_id")} row = by_id.get(complex_id, {}) merged_aliases: list[str] = [] for value in [row.get("name"), *(row.get("aliases") or []), info.get("name"), *(aliases or [])]: for variant in expand_alias_variants(str(value or "")): if variant and variant not in merged_aliases: merged_aliases.append(variant) note_items = [str(x).strip() for x in [row.get("note"), info.get("note")] if str(x or "").strip()] row.update( { "complex_id": complex_id, "name": str(info.get("name") or row.get("name") or f"단지_{complex_id}"), "address": str(info.get("address") or row.get("address") or ""), "household_count": info.get("household_count") or row.get("household_count"), "aliases": merged_aliases[:30], "updated_at": int(time.time()), "source": info.get("source") or row.get("source") or "runtime", "learned_from": info.get("learned_from") or row.get("learned_from"), "note": note_items[-1] if note_items else None, } ) by_id[complex_id] = row rows = sorted(by_id.values(), key=lambda x: (-int(x.get("updated_at") or 0), str(x.get("name") or ""))) cache["entries"] = rows[:500] _write_candidate_cache(cache) return row def list_candidate_cache(*, limit: int = 50, keyword: str | None = None) -> list[dict[str, Any]]: rows = _read_candidate_cache().get("entries") or [] if keyword: term = normalize_complex_alias(keyword) rows = [ row for row in rows if term in normalize_complex_alias(row.get("name") or "") or any(term in normalize_complex_alias(alias) for alias in (row.get("aliases") or [])) or term in normalize_complex_alias(row.get("address") or "") ] rows = sorted(rows, key=lambda row: (-int(row.get("updated_at") or 0), str(row.get("name") or ""))) return rows[: max(1, limit)] def seed_candidate_cache(entries: list[dict[str, Any]], *, source: str = "seed-file") -> list[dict[str, Any]]: saved: list[dict[str, Any]] = [] for item in entries: complex_id = str(item.get("complex_id") or "").strip() if not complex_id: continue info = { "complex_id": complex_id, "name": item.get("name"), "address": item.get("address"), "household_count": item.get("household_count"), "source": item.get("source") or source, "learned_from": item.get("learned_from") or source, "note": item.get("note"), } saved.append(remember_candidate(info, aliases=[str(x) for x in (item.get("aliases") or []) if str(x).strip()])) return [row for row in saved if row] def seed_candidate_from_file(path: str | Path | None = None) -> dict[str, Any]: target = Path(path) if path else DEFAULT_CANDIDATE_SEED_FILE payload = _read_json_file(target, {"entries": []}) entries = payload.get("entries") if isinstance(payload, dict) else payload saved = seed_candidate_cache(entries or [], source=f"seed-file:{target.name}") return {"path": str(target), "saved_count": len(saved), "saved": saved} def _load_reference_seed_rows() -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] target = _read_json_file(DEFAULT_CANDIDATE_SEED_FILE, {"entries": [], "manual_review_queue": []}) if isinstance(target, dict): for row in target.get("entries") or []: rows.append({**row, "reference_kind": "production-entry"}) for row in target.get("manual_review_queue") or []: rows.append({**row, "reference_kind": "manual-review"}) seed_input = _read_json_file(DEFAULT_SEED_INPUT_FILE, {"seeds": []}) if isinstance(seed_input, dict): for row in seed_input.get("seeds") or []: rows.append({**row, "reference_kind": "seed-input"}) return rows def search_reference_candidates(query: str, *, candidate_limit: int = 5, parsed: ParsedQuery | None = None) -> list[dict[str, Any]]: parsed = parsed or parse_natural_query(query) terms: list[str] = [] for value in [query, parsed.cleaned_query, *parsed.candidate_keywords, *parsed.raw_subjects]: for variant in expand_alias_variants(value): norm = normalize_complex_alias(variant) if norm and norm not in terms: terms.append(norm) if not terms: return [] scored: list[dict[str, Any]] = [] seen_keys: set[str] = set() for row in _load_reference_seed_rows(): name = str(row.get("name") or "").strip() if not name: continue aliases = [str(x) for x in (row.get("aliases") or []) if str(x).strip()] bag = [name, *aliases, row.get("district") or "", row.get("neighborhood") or "", row.get("address") or ""] bag_norms = [normalize_complex_alias(x) for x in bag if str(x or "").strip()] local_score = 0 for term in terms: if term in bag_norms: local_score = max(local_score, 240) elif any(term and term in norm for norm in bag_norms): local_score = max(local_score, 170) if local_score <= 0: continue key = str(row.get("complex_id") or f"hint:{normalize_complex_alias(name)}:{row.get('reference_kind')}") if key in seen_keys: continue seen_keys.add(key) reference_score = local_score + _score_candidate( { "name": name, "address": row.get("address") or " ".join(filter(None, [row.get("district"), row.get("neighborhood")])), "household_count": row.get("household_count"), "aliases": aliases, }, query, parsed, ) cid = str(row.get("complex_id") or "") scored.append( { "complex_id": cid, "name": name, "address": row.get("address") or " ".join(filter(None, [row.get("district"), row.get("neighborhood")])), "household_count": row.get("household_count"), "aliases": aliases[:20], "match_score": reference_score, "source_term": row.get("reference_kind") or "reference-seed", "source": "reference-seed", "reference_kind": row.get("reference_kind"), "review_status": row.get("review_status"), "verification_status": row.get("verification_status"), "next_action": row.get("next_action"), "note": row.get("note") or row.get("reason"), "complex_url": f"https://new.land.naver.com/complexes/{cid}" if cid else None, } ) scored.sort(key=lambda row: (-int(row.get("match_score") or 0), str(row.get("name") or ""), str(row.get("complex_id") or ""))) return scored[:candidate_limit] def search_cached_candidates(query: str, *, candidate_limit: int = 5, parsed: ParsedQuery | None = None) -> list[dict[str, Any]]: parsed = parsed or parse_natural_query(query) cache = _read_candidate_cache().get("entries") or [] if not cache: return [] terms: list[str] = [] for value in [query, parsed.cleaned_query, *parsed.candidate_keywords, *parsed.location_hints]: for variant in expand_alias_variants(value): norm = normalize_complex_alias(variant) if norm and norm not in terms: terms.append(norm) scored: list[dict[str, Any]] = [] for row in cache: aliases = row.get("aliases") or [] alias_norms = [normalize_complex_alias(x) for x in aliases if x] alias_norms = [x for x in alias_norms if x] local_score = 0 for term in terms: if term in alias_norms: local_score = max(local_score, 300) elif any(term in alias for alias in alias_norms): local_score = max(local_score, 210) if local_score <= 0: continue score = local_score + _score_candidate({**row, "aliases": aliases}, query, parsed) cid = str(row.get("complex_id") or "") scored.append({**row, "aliases": aliases, "match_score": score, "source_term": "candidate-cache", "source": "candidate-cache", "complex_url": f"https://new.land.naver.com/complexes/{cid}" if cid else None}) scored.sort(key=lambda row: (-int(row.get("match_score") or 0), str(row.get("name") or ""), str(row.get("complex_id") or ""))) return scored[:candidate_limit] def build_search_terms(parsed: ParsedQuery) -> list[str]: terms: list[str] = [] raw_values = [*parsed.raw_subjects, parsed.cleaned_query, *parsed.candidate_keywords, *parsed.location_hints] for value in raw_values: for variant in expand_alias_variants(value): variant = str(variant or "").strip() if variant and variant not in terms: terms.append(variant) if parsed.location_hints and parsed.candidate_keywords: for loc in parsed.location_hints[:3]: for kw in parsed.candidate_keywords[:3]: combo = f"{loc} {kw}".strip() if combo and combo not in terms: terms.append(combo) return terms[:16] def resolve_complex_ids(query: str | None, complex_id: str | None, url: str | None, *, candidate_limit: int = 5) -> list[str]: results: list[str] = [] seen: set[str] = set() def _push(value: str | None): if value and value not in seen: results.append(value) seen.add(value) _push(complex_id) if url: _push(NaverURLParser.extract_complex_id(url)) if query: parsed = parse_natural_query(query) for cid in parsed.direct_complex_ids: _push(cid) if not results: for item in search_cached_candidates(query, candidate_limit=max(candidate_limit * 2, 6), parsed=parsed): _push(item.get("complex_id")) if len(results) >= candidate_limit: break if not results: for item in search_complex_candidates(query, candidate_limit=max(candidate_limit * 2, 6)): _push(item.get("complex_id")) if len(results) >= candidate_limit: break return results[:candidate_limit] def search_complex_candidates(query: str, *, candidate_limit: int = 5) -> list[dict[str, Any]]: parsed = parse_natural_query(query) merged: dict[str, dict[str, Any]] = {} def _merge_item(item: dict[str, Any]) -> None: key = str(item.get("complex_id") or "").strip() or f"hint:{normalize_complex_alias(item.get('name') or '')}:{item.get('source') or item.get('reference_kind') or 'unknown'}" prev = merged.get(key) if not prev or int(item.get("match_score") or 0) >= int(prev.get("match_score") or 0): merged[key] = dict(item) for item in search_cached_candidates(query, candidate_limit=max(candidate_limit * 2, 8), parsed=parsed): _merge_item(item) for item in search_reference_candidates(query, candidate_limit=max(candidate_limit * 2, 8), parsed=parsed): _merge_item(item) raw_ids: list[tuple[str, str]] = [] seen_ids: set[str] = {str(row.get("complex_id") or "") for row in merged.values() if str(row.get("complex_id") or "").strip()} for term in build_search_terms(parsed): try: web_items = extract_complex_candidates_from_web(term, limit=max(candidate_limit * 2, 8)) except SearchError: web_items = [] for item in web_items: cid = str(item.get("complex_id") or "").strip() if cid and cid not in seen_ids: raw_ids.append((cid, term)) seen_ids.add(cid) for cid, source_term in raw_ids: try: info = fetch_complex_info(cid) except Exception as exc: info = {"complex_id": cid, "name": f"단지_{cid}", "address": "", "error": str(exc), "source": "web-search-fallback"} score = _score_candidate(info, source_term, parsed) merged[cid] = {**info, "match_score": score, "source_term": source_term, "source": info.get("source") or "web-search", "complex_url": f"https://new.land.naver.com/complexes/{cid}"} scored = list(merged.values()) scored.sort(key=lambda row: (-int(row.get("match_score") or 0), str(row.get("name") or ""), str(row.get("complex_id") or ""))) return scored[:candidate_limit] def fetch_articles(complex_id: str, trade_types: list[str], pages: int = 1) -> list[dict[str, Any]]: trade_codes = ":".join(TRADE_CODE_MAP[t] for t in trade_types if t in TRADE_CODE_MAP) if not trade_codes: trade_codes = "A1:B1:B2" complex_name = NaverURLParser.fetch_complex_name(complex_id) items: list[dict[str, Any]] = [] for page in range(1, pages + 1): payload = _request_json(COMPLEX_ARTICLE_URL.format(complex_id=complex_id, trade_codes=trade_codes, page=page)) article_list = payload.get("articleList") or payload.get("list") or [] if not article_list: break for article in article_list: trade_type = str(article.get("tradeTypeName") or article.get("tradTpNm") or "").strip() normalized = normalize_article_payload(article, complex_name, complex_id, requested_trade_type=trade_type) normalized["매물URL"] = get_article_url(complex_id, normalized.get("매물ID", ""), normalized.get("자산유형", "APT")) normalized["complex_id"] = complex_id normalized["article_key"] = f"{complex_id}:{normalized.get('매물ID', '')}" items.append(normalized) return items def filter_items(items: list[dict[str, Any]], min_pyeong: float | None, max_pyeong: float | None, limit: int) -> list[dict[str, Any]]: filtered = [] for item in items: area = float(item.get("면적(평)") or 0) if min_pyeong is not None and area < min_pyeong: continue if max_pyeong is not None and area > max_pyeong: continue filtered.append(item) filtered.sort(key=lambda row: (row.get("거래유형", ""), PriceConverter.to_int(row.get("매매가") or row.get("보증금") or "0"), row.get("면적(평)", 0))) return filtered[:limit] def _bucket_area(value: float | int | None) -> str: try: area = float(value or 0) except Exception: area = 0.0 if area <= 0: return "미상" base = int(round(area)) return f"{base}평" def _extract_price_int(row: dict[str, Any]) -> int: return PriceConverter.to_int(row.get("매매가") or row.get("보증금") or "0") def build_market_summary(items: list[dict[str, Any]]) -> dict[str, Any]: grouped: dict[str, list[dict[str, Any]]] = {} for item in items: grouped.setdefault(item.get("거래유형", "기타"), []).append(item) summary: dict[str, Any] = {} for trade_type, rows in grouped.items(): prices = [_extract_price_int(r) for r in rows] prices = [p for p in prices if p > 0] area_groups: dict[str, list[dict[str, Any]]] = {} for row in rows: area_groups.setdefault(_bucket_area(row.get("면적(평)")), []).append(row) comparable_area_rows = [] for area_key, area_items in area_groups.items(): area_prices = [_extract_price_int(r) for r in area_items] area_prices = [p for p in area_prices if p > 0] comparable_area_rows.append({ "area_key": area_key, "count": len(area_items), "min_price": min(area_prices) if area_prices else None, "avg_price": int(sum(area_prices) / len(area_prices)) if area_prices else None, "max_price": max(area_prices) if area_prices else None, "sample_items": area_items[:2], }) comparable_area_rows.sort(key=lambda row: (-int(row.get("count") or 0), str(row.get("area_key") or ""))) summary[trade_type] = { "count": len(rows), "min_price": min(prices) if prices else None, "max_price": max(prices) if prices else None, "avg_price": int(sum(prices) / len(prices)) if prices else None, "median_price": int(statistics.median(prices)) if prices else None, "sample_items": rows[:3], "area_summary": comparable_area_rows[:5], } return summary def summarize(items: list[dict[str, Any]]) -> str: if not items: return "조건에 맞는 매물을 찾지 못했습니다." lines = [] grouped: dict[str, list[dict[str, Any]]] = {} for item in items: grouped.setdefault(item.get("거래유형", "기타"), []).append(item) for trade_type, rows in grouped.items(): lines.append(f"[{trade_type}] {len(rows)}건") prices = [_extract_price_int(r) for r in rows] prices = [p for p in prices if p > 0] if prices: lines.append(f"- 가격 범위: {PriceConverter.to_string(min(prices))} ~ {PriceConverter.to_string(max(prices))}") lines.append(f"- 평균/중앙값: {PriceConverter.to_string(int(sum(prices)/len(prices)))} / {PriceConverter.to_string(int(statistics.median(prices)))}") area_summary = build_market_summary(rows).get(trade_type, {}).get("area_summary", []) if area_summary: head = area_summary[0] lines.append(f"- 대표 동일 평형: {head.get('area_key')} {head.get('count')}건 | 평균 {PriceConverter.to_string(head.get('avg_price')) if head.get('avg_price') else '-'}") for row in rows[:5]: price = row.get("매매가") or row.get("보증금") or "-" if trade_type == "월세" and row.get("월세"): price = f"{price}/{row.get('월세')}" lines.append(f"- {row.get('단지명')} | {price} | {row.get('면적(평)', 0)}평 | {row.get('층/방향', '-') or '-'} | {row.get('매물URL', '')}") return "\n".join(lines) def _brief_price(value: int | None) -> str: return PriceConverter.to_string(value) if value else "-" def build_compare_insights(results: list[dict[str, Any]]) -> dict[str, Any]: trade_groups: dict[str, list[dict[str, Any]]] = {} area_match_groups: dict[str, dict[str, list[dict[str, Any]]]] = {} for result in results: info = result.get("complex_info", {}) name = info.get("name", result.get("complex_id")) for trade_type, meta in (result.get("market_summary") or {}).items(): trade_groups.setdefault(trade_type, []).append({"name": name, "avg_price": meta.get("avg_price"), "min_price": meta.get("min_price"), "count": meta.get("count", 0)}) for area_row in meta.get("area_summary") or []: area_match_groups.setdefault(trade_type, {}).setdefault(area_row.get("area_key") or "미상", []).append({"name": name, **area_row}) insights: dict[str, Any] = {"trade": {}, "same_area": {}} for trade_type, rows in trade_groups.items(): valid = [row for row in rows if row.get("avg_price")] if len(valid) >= 2: valid.sort(key=lambda x: x["avg_price"]) insights["trade"][trade_type] = { "cheapest": valid[0], "most_expensive": valid[-1], "gap": valid[-1]["avg_price"] - valid[0]["avg_price"], } for trade_type, area_rows in area_match_groups.items(): for area_key, rows in area_rows.items(): valid = [row for row in rows if row.get("avg_price")] if len(valid) >= 2: valid.sort(key=lambda x: x["avg_price"]) insights["same_area"].setdefault(trade_type, []).append({ "area_key": area_key, "cheapest": valid[0], "most_expensive": valid[-1], "gap": valid[-1]["avg_price"] - valid[0]["avg_price"], "participant_count": len(valid), }) for trade_type in list(insights["same_area"].keys()): insights["same_area"][trade_type].sort(key=lambda row: (-row.get("participant_count", 0), row.get("gap", 0))) return insights def summarize_comparison(results: list[dict[str, Any]]) -> str: if not results: return "비교할 단지 결과가 없습니다." lines = ["[단지 비교 브리핑]"] insights = build_compare_insights(results) for result in results: info = result.get("complex_info", {}) name = info.get("name", result.get("complex_id")) address = info.get("address") or "-" lines.append(f"- {name}") lines.append(f" · 주소: {address}") for trade_type, meta in result.get("market_summary", {}).items(): lines.append(f" · {trade_type}: {meta.get('count', 0)}건 | 최저 {_brief_price(meta.get('min_price'))} | 평균 {_brief_price(meta.get('avg_price'))} | 중앙값 {_brief_price(meta.get('median_price'))} | 최고 {_brief_price(meta.get('max_price'))}") area_rows = meta.get("area_summary") or [] if area_rows: head = area_rows[0] lines.append(f" - 대표 동일 평형: {head.get('area_key')} {head.get('count')}건 | 평균 {_brief_price(head.get('avg_price'))}") for trade_type, meta in (insights.get("trade") or {}).items(): lines.append(f"- 한줄 해석 ({trade_type}): {meta['cheapest']['name']} 쪽 평균이 가장 낮고 {meta['most_expensive']['name']} 쪽이 가장 높습니다. 전체 평균 격차는 {_brief_price(meta['gap'])} 정도입니다.") for trade_type, area_rows in (insights.get("same_area") or {}).items(): if area_rows: head = area_rows[0] lines.append(f"- 동일 평형 비교 ({trade_type} {head['area_key']}): {head['cheapest']['name']} 쪽이 더 낮고 {head['most_expensive']['name']} 쪽이 더 높습니다. 같은 평형대 기준 격차는 {_brief_price(head['gap'])} 정도입니다.") return "\n".join(lines) def run_query(*, query: str | None, complex_id: str | None, url: str | None, trade_types: list[str] | None, pages: int, limit: int, candidate_limit: int, min_pyeong: float | None, max_pyeong: float | None, compare: bool) -> dict[str, Any]: parsed = parse_natural_query(query or "") if query else None trade_types = list(trade_types or []) if not trade_types: trade_types = parsed.trade_types if parsed else ["전세"] min_pyeong = min_pyeong if min_pyeong is not None else (parsed.min_pyeong if parsed else None) max_pyeong = max_pyeong if max_pyeong is not None else (parsed.max_pyeong if parsed else None) complex_ids = resolve_complex_ids(query, complex_id, url, candidate_limit=max(1, candidate_limit)) if not complex_ids: raise SearchError("단지 ID를 찾지 못했습니다. 더 구체적인 단지명/지역명을 주거나 단지 URL/ID를 직접 넣어 주세요.") compare_mode = compare or bool(parsed and parsed.compare_mode and len(complex_ids) >= 2) target_ids = complex_ids[: max(1, candidate_limit if compare_mode else 1)] meta = {"rate_limited": bool(RATE_LIMIT_STATE.get("active")), "rate_limit_message": RATE_LIMIT_STATE.get("last_error")} if compare_mode: results = [] for cid in target_ids: items = fetch_articles(cid, trade_types, pages=max(1, pages)) items = filter_items(items, min_pyeong, max_pyeong, max(1, limit)) results.append({ "complex_id": cid, "complex_info": fetch_complex_info(cid), "trade_types": trade_types, "count": len(items), "market_summary": build_market_summary(items), "items": items[:5], }) compare_insights = build_compare_insights(results) return {"query": query, "parsed": asdict(parsed) if parsed else None, "compare_mode": True, "results": results, "compare_insights": compare_insights, "meta": meta} selected_complex_id = target_ids[0] items = fetch_articles(selected_complex_id, trade_types, pages=max(1, pages)) items = filter_items(items, min_pyeong, max_pyeong, max(1, limit)) return { "query": query, "parsed": asdict(parsed) if parsed else None, "selected_complex_id": selected_complex_id, "complex_info": fetch_complex_info(selected_complex_id), "trade_types": trade_types, "count": len(items), "market_summary": build_market_summary(items), "items": items, "meta": meta, } def run_self_test() -> int: if not UPSTREAM.exists() or UPSTREAM_IMPORT_ERROR is not None: print( "SKIP: upstream clone(tmp/naverland-scrapper) 미구성 상태라 upstream 의존 self-test는 건너뜁니다.", file=sys.stderr, ) return 0 sample = { "articleNo": "123456789", "tradeTypeName": "전세", "dealOrWarrantPrc": "12억 5,000", "area1": 84.98, "floorInfo": "12/25", "direction": "남향", "articleFeatureDesc": "역세권, 학군우수", "realEstateTypeCode": "APT", } row = normalize_article_payload(sample, "테스트아파트", "99999", requested_trade_type="전세") assert row["단지명"] == "테스트아파트" assert row["거래유형"] == "전세" assert row["보증금"] == "12억 5,000" assert row["면적(평)"] > 0 parsed = parse_natural_query("잠실 리센츠랑 엘스 전세 비교 30평대") assert parsed.compare_mode is True assert "전세" in parsed.trade_types assert parsed.min_pyeong is not None and parsed.max_pyeong is not None assert len(parsed.candidate_keywords) >= 2 assert "잠실" in parsed.location_hints tricky = parse_natural_query("서울 양천구 신월동 신월시영아파트 전세") assert tricky.cleaned_query.startswith("서울 양천구 신월동 신월시영") assert any(normalize_complex_alias(x) == "신월시영" for x in tricky.candidate_keywords) assert normalize_complex_alias("신월시영아파트") == "신월시영" direct = build_direct_lookup_payload("complex 1147 리센츠", None, None) assert direct["selected_complex_id"] == "1147" assert direct["canonical_complex_url"].endswith("/1147") backup_cache = _read_candidate_cache() remember_candidate({"complex_id": "777", "name": "신월시영아파트", "address": "서울시 양천구 신월동", "household_count": 2256}, aliases=["신월시영", "양천 신월시영"]) saved = seed_candidate_cache([ {"complex_id": "778", "name": "리센츠", "address": "서울시 송파구 잠실동", "aliases": ["잠실 리센츠", "잠실리센츠"], "household_count": 5563} ], source="self-test") cached = search_cached_candidates("신월시영아파트", candidate_limit=3) assert cached and cached[0]["complex_id"] == "777" assert saved and any(row["complex_id"] == "778" for row in saved) assert list_candidate_cache(limit=5, keyword="리센츠") ref_candidates = search_reference_candidates("서울 양천구 신월동 신월시영아파트 전세", candidate_limit=3) assert ref_candidates and normalize_complex_alias(ref_candidates[0]["name"]) == "신월시영" _write_candidate_cache(backup_cache) score = _score_candidate({"name": "잠실리센츠", "address": "서울시 송파구 잠실동", "household_count": 5563}, "리센츠", parsed) assert score > 0 summary = build_market_summary([ {"거래유형": "전세", "보증금": "10억", "면적(평)": 33.0}, {"거래유형": "전세", "보증금": "11억", "면적(평)": 33.2}, {"거래유형": "전세", "보증금": "13억", "면적(평)": 44.0}, ]) assert summary["전세"]["area_summary"][0]["area_key"] in {"33평", "44평"} compare_insights = build_compare_insights([ {"complex_info": {"name": "A"}, "market_summary": {"전세": {"avg_price": 100, "count": 2, "area_summary": [{"area_key": "33평", "avg_price": 100, "count": 2}]}}}, {"complex_info": {"name": "B"}, "market_summary": {"전세": {"avg_price": 130, "count": 2, "area_summary": [{"area_key": "33평", "avg_price": 130, "count": 2}]}}}, ]) assert compare_insights["trade"]["전세"]["gap"] == 30 print("SELF_TEST_OK") print(json.dumps({"parsed_query": asdict(parsed), "tricky_query": asdict(tricky), "cached_top": cached[:1], "sample_score": score, "compare_insights": compare_insights}, ensure_ascii=False, indent=2)[:3000]) return 0 def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="네이버 부동산 매물 검색/비교 래퍼") p.add_argument("--query", help="자연어 또는 지역/단지 키워드. 예: 잠실 리센츠 전세 30평대, 대치 은마와 래미안대치팰리스 비교") p.add_argument("--complex-id", help="네이버 부동산 단지 ID") p.add_argument("--url", help="네이버 부동산 단지/매물 URL") p.add_argument("--trade-types", default="", help="쉼표 구분 거래 유형. 비우면 query에서 추론하고, 없으면 전세") p.add_argument("--pages", type=int, default=1) p.add_argument("--limit", type=int, default=15) p.add_argument("--candidate-limit", type=int, default=3) p.add_argument("--min-pyeong", type=float) p.add_argument("--max-pyeong", type=float) p.add_argument("--list-candidates", action="store_true", help="매물 조회 대신 단지 후보만 출력") p.add_argument("--compare", action="store_true", help="후보 상위 단지들을 비교 모드로 조회") p.add_argument("--parse-only", action="store_true", help="자연어 파싱 결과만 출력") p.add_argument("--show-cache", action="store_true", help="candidate cache를 조회") p.add_argument("--resolve-direct", action="store_true", help="query/url/complex-id에서 direct ID와 canonical URL을 정리") p.add_argument("--lookup-complex", action="store_true", help="매물 조회 대신 단지 기본 정보만 direct lookup") p.add_argument("--seed-candidate-file", nargs="?", const="", help="candidate-cache에 seed할 JSON 파일 경로. 비우면 references/candidate-seeds.json") p.add_argument("--seed-candidate", action="store_true", help="단일 후보를 candidate-cache에 직접 저장") p.add_argument("--candidate-name", help="seed candidate 이름") p.add_argument("--candidate-address", help="seed candidate 주소") p.add_argument("--candidate-households", type=int, help="seed candidate 세대수") p.add_argument("--candidate-aliases", default="", help="쉼표 구분 alias 목록") p.add_argument("--candidate-note", help="seed candidate 메모") p.add_argument("--json", action="store_true") p.add_argument("--self-test", action="store_true") return p def main() -> int: args = build_parser().parse_args() if args.self_test: return run_self_test() parsed = parse_natural_query(args.query or "") if args.query else None if args.parse_only: print(json.dumps(asdict(parsed) if parsed else {}, ensure_ascii=False, indent=2)) return 0 if args.seed_candidate_file is not None: result = seed_candidate_from_file(args.seed_candidate_file or None) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if args.resolve_direct: print(json.dumps(build_direct_lookup_payload(args.query, args.complex_id, args.url), ensure_ascii=False, indent=2)) return 0 if args.seed_candidate: if not args.complex_id: raise SystemExit("--seed-candidate 는 --complex-id 와 함께 사용하세요.") aliases = [token.strip() for token in str(args.candidate_aliases or "").split(",") if token.strip()] saved = remember_candidate( { "complex_id": args.complex_id, "name": args.candidate_name, "address": args.candidate_address, "household_count": args.candidate_households, "source": "manual-seed", "learned_from": args.query or args.url or "manual-seed", "note": args.candidate_note, }, aliases=aliases, ) print(json.dumps({"saved": True, "candidate": saved}, ensure_ascii=False, indent=2)) return 0 if args.show_cache: rows = list_candidate_cache(limit=max(1, args.limit), keyword=args.query) print(json.dumps({"count": len(rows), "entries": rows}, ensure_ascii=False, indent=2)) return 0 if args.lookup_complex: payload = build_direct_lookup_payload(args.query, args.complex_id, args.url) selected = payload.get("selected_complex_id") if not selected: raise SystemExit("direct lookup용 complex ID를 찾지 못했습니다. --complex-id / --url / --query 중 하나에 direct 단서를 넣어 주세요.") payload["complex_info"] = fetch_complex_info(str(selected)) print(json.dumps(payload, ensure_ascii=False, indent=2)) return 0 trade_types = [token.strip() for token in str(args.trade_types).split(",") if token.strip()] if args.list_candidates: if not args.query: raise SystemExit("--list-candidates 는 --query 와 함께 사용하세요.") candidates = search_complex_candidates(args.query, candidate_limit=max(1, args.candidate_limit)) print(json.dumps({"query": args.query, "parsed": asdict(parsed), "candidates": candidates, "meta": {"rate_limited": RATE_LIMIT_STATE.get('active'), "rate_limit_message": RATE_LIMIT_STATE.get('last_error')}}, ensure_ascii=False, indent=2)) return 0 output = run_query( query=args.query, complex_id=args.complex_id, url=args.url, trade_types=trade_types, pages=max(1, args.pages), limit=max(1, args.limit), candidate_limit=max(1, args.candidate_limit), min_pyeong=args.min_pyeong, max_pyeong=args.max_pyeong, compare=bool(args.compare), ) if args.json: print(json.dumps(output, ensure_ascii=False, indent=2)) else: print(summarize_comparison(output.get("results", [])) if output.get("compare_mode") else summarize(output.get("items", []))) return 0 if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/watch_real_estate.py from __future__ import annotations import argparse import json import time import uuid from pathlib import Path from typing import Any from search_real_estate import WATCH_STATE_FILE, PriceConverter, run_query SCHEMA_VERSION = 2 def _load_rules() -> dict[str, Any]: if WATCH_STATE_FILE.exists(): data = json.loads(WATCH_STATE_FILE.read_text(encoding="utf-8")) if isinstance(data, dict) and "rules" in data: data.setdefault("schema_version", SCHEMA_VERSION) data.setdefault("last_checked_at", None) data.setdefault("events", []) data.setdefault("last_seen", {}) return data return {"schema_version": SCHEMA_VERSION, "last_checked_at": None, "rules": [], "events": [], "last_seen": {}} def _save_rules(data: dict[str, Any]) -> None: WATCH_STATE_FILE.parent.mkdir(parents=True, exist_ok=True) WATCH_STATE_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def _fmt_price(value: int | None) -> str: return PriceConverter.to_string(value) if value else "-" def _article_key(item: dict[str, Any]) -> str: return str(item.get("article_key") or f"{item.get('complex_id', '')}:{item.get('매물ID', '')}") def _normalize_rule(args: argparse.Namespace) -> dict[str, Any]: return { "id": getattr(args, "rule_id", None) or f"rule-{uuid.uuid4().hex[:10]}", "name": args.name, "query": args.query, "complex_id": args.complex_id, "url": args.url, "trade_types": [token.strip() for token in str(args.trade_types or "").split(",") if token.strip()], "target_max_price": args.target_max_price, "pages": args.pages, "limit": args.limit, "candidate_limit": args.candidate_limit, "min_pyeong": args.min_pyeong, "max_pyeong": args.max_pyeong, "notify_on_new": bool(args.notify_on_new), "notify_on_price_drop": bool(args.notify_on_price_drop), "enabled": True, "created_at": int(time.time()), "updated_at": int(time.time()), "notes": getattr(args, "notes", None), } def add_rule(args: argparse.Namespace) -> int: data = _load_rules() rule = _normalize_rule(args) data.setdefault("rules", []).append(rule) _save_rules(data) print(json.dumps({"saved": True, "rule": rule}, ensure_ascii=False, indent=2)) return 0 def list_rules() -> int: print(json.dumps(_load_rules(), ensure_ascii=False, indent=2)) return 0 def _make_match(rule: dict[str, Any], item: dict[str, Any], *, event_type: str, previous: dict[str, Any] | None = None) -> dict[str, Any]: price = PriceConverter.to_int(item.get("매매가") or item.get("보증금") or "0") return { "event_type": event_type, "article_key": _article_key(item), "price": price, "price_text": _fmt_price(price), "complex_id": item.get("complex_id"), "complex_name": item.get("단지명"), "article_url": item.get("매물URL"), "area_pyeong": item.get("면적(평)"), "trade_type": item.get("거래유형"), "floor_direction": item.get("층/방향"), "feature": item.get("특징"), "previous_price": previous.get("price") if previous else None, "previous_price_text": _fmt_price(previous.get("price")) if previous and previous.get("price") else None, "target_max_price": rule.get("target_max_price"), "target_max_price_text": _fmt_price(rule.get("target_max_price")), "detected_at": int(time.time()), } def _build_alert_lines(alerts: list[dict[str, Any]], *, max_matches_per_rule: int = 3) -> list[str]: lines = [] for row in alerts: rule = row.get("rule") or {} matched = row.get("matched") or [] snapshot = row.get("snapshot") or {} complex_info = snapshot.get("complex_info") or {} label = rule.get("name") or rule.get("id") or "이름없는 규칙" header = f"- {label}: {row.get('matched_count', 0)}건" if complex_info.get("name"): header += f" | {complex_info.get('name')}" if row.get("error"): header += f" | 오류: {row['error']}" lines.append(header) for item in matched[:max_matches_per_rule]: badges = [] if item.get("event_type") == "new_listing": badges.append("신규") elif item.get("event_type") == "price_drop": badges.append("가격하락") elif item.get("event_type") == "target_hit": badges.append("목표가도달") if item.get("previous_price_text"): badges.append(f"이전 {item['previous_price_text']}") badge_text = f" ({', '.join(badges)})" if badges else "" lines.append(f" · {item.get('complex_name')} {item.get('trade_type')} {item.get('price_text')} / {item.get('area_pyeong')}평{badge_text}") if item.get("article_url"): lines.append(f" - {item['article_url']}") return lines def _build_message_preview(alerts: list[dict[str, Any]], *, checked_at: int) -> str: total = sum(int(row.get("matched_count") or 0) for row in alerts) lines = [f"부동산 감시 점검 결과: {total}건 알림", f"checked_at={checked_at}"] lines.extend(_build_alert_lines(alerts)) return "\n".join(lines) def _stdout_payload(alerts: list[dict[str, Any]], *, checked_at: int) -> dict[str, Any]: preview = _build_message_preview(alerts, checked_at=checked_at) return { "kind": "naver-real-estate-watch-check", "schema_version": SCHEMA_VERSION, "checked_at": checked_at, "alert_count": sum(int(row.get("matched_count") or 0) for row in alerts), "alerts": alerts, "message_preview": preview, "summary": { "rule_count": len(alerts), "rules_with_matches": sum(1 for row in alerts if int(row.get("matched_count") or 0) > 0), "rules_with_errors": sum(1 for row in alerts if row.get("error")), }, } def check_rules(args: argparse.Namespace) -> int: data = _load_rules() last_seen = data.setdefault("last_seen", {}) checked_at = int(time.time()) alerts = [] new_events = [] for rule in data.get("rules", []): if rule.get("enabled") is False: continue try: payload = run_query( query=rule.get("query"), complex_id=rule.get("complex_id"), url=rule.get("url"), trade_types=rule.get("trade_types") or [], pages=max(1, int(rule.get("pages") or 1)), limit=max(1, int(rule.get("limit") or 10)), candidate_limit=max(1, int(rule.get("candidate_limit") or 1)), min_pyeong=rule.get("min_pyeong"), max_pyeong=rule.get("max_pyeong"), compare=False, ) except Exception as exc: alerts.append({ "rule": rule, "matched": [], "matched_count": 0, "error": str(exc), "snapshot": None, }) continue threshold = rule.get("target_max_price") matches = [] for item in payload.get("items", []): price = PriceConverter.to_int(item.get("매매가") or item.get("보증금") or "0") article_key = _article_key(item) prev = last_seen.get(article_key) is_new = prev is None is_price_drop = bool(prev and prev.get("price") and price and price < prev.get("price")) threshold_hit = bool(threshold and price and price <= threshold) emit = False event_type = None if threshold_hit: emit = True event_type = "target_hit" if is_new and rule.get("notify_on_new"): emit = True event_type = event_type or "new_listing" if is_price_drop and rule.get("notify_on_price_drop"): emit = True event_type = event_type or "price_drop" if emit: match = _make_match(rule, item, event_type=event_type or "matched", previous=prev) matches.append(match) dedupe_key = f"{rule.get('id')}::{event_type}::{article_key}::{price}" known_event_keys = {e.get('dedupe_key') for e in data.get('events', [])[-500:]} if dedupe_key not in known_event_keys: event = {"rule_id": rule.get("id"), "rule_name": rule.get("name"), "dedupe_key": dedupe_key, **match} new_events.append(event) last_seen[article_key] = { "rule_id": rule.get("id"), "price": price, "last_seen_at": checked_at, "trade_type": item.get("거래유형"), "complex_name": item.get("단지명"), "article_url": item.get("매물URL"), } alerts.append({ "rule": rule, "matched": matches, "matched_count": len(matches), "snapshot": { "selected_complex_id": payload.get("selected_complex_id"), "complex_info": payload.get("complex_info"), "market_summary": payload.get("market_summary"), "meta": payload.get("meta"), }, }) data["last_checked_at"] = checked_at data.setdefault("events", []).extend(new_events) data["events"] = data["events"][-1000:] data["last_seen"] = last_seen _save_rules(data) payload = _stdout_payload(alerts, checked_at=checked_at) if args.json: print(json.dumps(payload, ensure_ascii=False, indent=2)) elif args.preview: print(payload["message_preview"]) else: print(payload["message_preview"]) return 0 def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="네이버 부동산 가격 감시/새 매물 감지") sub = p.add_subparsers(dest="cmd", required=True) add_p = sub.add_parser("add") add_p.add_argument("--name", required=True) add_p.add_argument("--query") add_p.add_argument("--complex-id") add_p.add_argument("--url") add_p.add_argument("--trade-types", default="") add_p.add_argument("--target-max-price", type=int, help="정수 가격 기준. 예: 950000000") add_p.add_argument("--pages", type=int, default=1) add_p.add_argument("--limit", type=int, default=10) add_p.add_argument("--candidate-limit", type=int, default=1) add_p.add_argument("--min-pyeong", type=float) add_p.add_argument("--max-pyeong", type=float) add_p.add_argument("--notify-on-new", action="store_true") add_p.add_argument("--notify-on-price-drop", action="store_true") add_p.add_argument("--notes") sub.add_parser("list") check_p = sub.add_parser("check") check_p.add_argument("--json", action="store_true") check_p.add_argument("--preview", action="store_true", help="사람이 읽기 쉬운 preview만 출력") return p def main() -> int: args = build_parser().parse_args() if args.cmd == "add": return add_rule(args) if args.cmd == "list": return list_rules() if args.cmd == "check": return check_rules(args) raise SystemExit(2) if __name__ == "__main__": raise SystemExit(main())

Naver Real Estate Search

Comments (0)