--- name: korea-domestic-flights description: Search, compare, and monitor 대한민국 domestic flights using a Playwright-backed airfare workflow. Use when the user wants 한국 국내선 항공권 최저가 조회, 김포-제주/부산-제주 같은 노선 비교, 편도/왕복 검색, 날짜 범위 최저가 탐색, 여러 국내 목적지 비교, 시간대 선호(늦은 출발/늦은 복귀) 반영, 국내선 운임 브리핑, 목적지+날짜 범위 최적 조합 탐색, or 목표가 이하 가격 감시/가격 알림 저장·점검. Accept common Korean airport names like 김포, 제주, 부산, 청주 and natural-language dates like 오늘/내일/모레/이번주말/내일부터 3일. Prefer this skill for Korean domestic airfare tasks; do not use it for international flights. --- # Korea Domestic Flights Use this skill for **대한민국 국내선 전용 항공권 검색**. Current scope: - 국내선 편도 검색 - 국내선 왕복 검색 - 날짜 범위 최저가 탐색 - 날짜별 가격 캘린더/히트맵 요약 - 다중 목적지 비교 - 다중 목적지 + 날짜 범위 최적 조합 탐색 - 다중 목적지 + 날짜 범위 비교에서도 목적지별 가격 캘린더 요약 제공 - JSON 출력 - 더 자연스러운 한국어 브리핑 출력 - 한글 공항명 입력 지원 - `오늘/내일/모레/이번주말/내일부터 3일` 같은 간단한 자연어 날짜 지원 - `2026-03-25~2026-03-30`, `20260325~20260330`, `2026-03-25부터 2026-03-30` 같은 명시 날짜 범위 지원 - 채팅 친화 래퍼 제공 - 시간대 필터(오전/오후/저녁, 출발 N시 이후, 복귀 N시 이후, 너무 이른 비행 제외) - 최저가 외 시간 선호 기반 추천(예: 늦은 시간대 기준 추천) - 왕복/날짜범위 검색에서 귀환편 시간 조건 반영 - 왕복 범위/조합 검색에서 가격만이 아닌 왕복 시간 균형 추천 제공 - 추천 결과에 가격 차이·시간 조건·왕복 균형 기반 추천 사유 설명 제공 - 목표가 기반 가격 감시 규칙 저장/목록/삭제/점검 - 가격 감시 규칙에 시간대 조건 저장/점검 지원 - 가격 알림 중복 방지(dedupe) - 단일/다중 목적지 가격 감시 - 알림 메시지 포맷 커스터마이즈 - cron/브리핑 연동을 염두에 둔 JSON 저장 포맷 - Windows 작업 스케줄러 등록 초안 스크립트 Do not use it for 국제선. ## Source dependency This skill wraps the local project clone at: - `tmp/Scraping-flight-information` Main reused entry points: - `scraping.searcher.FlightSearcher` - `scraping.parallel.ParallelSearcher` If the clone or its dependencies are missing, searches will fail. ## Scripts ### 1) Single-route domestic search ```bash python skills/korea-domestic-flights/scripts/search_domestic.py --origin 김포 --destination 제주 --departure 내일 --human ``` 시간 조건 포함: ```bash python skills/korea-domestic-flights/scripts/search_domestic.py --origin 김포 --destination 제주 --departure 내일 --time-pref "출발 10시 이후" --prefer late --human ``` Round trip: ```bash python skills/korea-domestic-flights/scripts/search_domestic.py --origin GMP --destination CJU --departure 2026-03-25 --return-date 2026-03-28 --human ``` ### 2) Date-range cheapest-day search ```bash python skills/korea-domestic-flights/scripts/search_date_range.py --origin 김포 --destination 제주 --date-range "내일부터 3일" --human ``` Explicit range: ```bash python skills/korea-domestic-flights/scripts/search_date_range.py --origin 김포 --destination 제주 --start-date 내일 --end-date 2026-03-30 --human ``` Round-trip-style date scan with fixed return offset: ```bash python skills/korea-domestic-flights/scripts/search_date_range.py --origin 김포 --destination 제주 --date-range "다음주말" --return-offset 2 --time-pref "복귀 18시 이후" --human ``` ### 3) Multi-destination comparison ```bash python skills/korea-domestic-flights/scripts/search_multi_destination.py --origin 김포 --destinations 제주,부산,여수 --departure 내일 --human ``` Round trip comparison: ```bash python skills/korea-domestic-flights/scripts/search_multi_destination.py --origin GMP --destinations CJU,PUS,RSU --departure 2026-03-25 --return-date 2026-03-28 --human ``` ### 4) Multi-destination + date-range best-combo search ```bash python skills/korea-domestic-flights/scripts/search_destination_date_matrix.py --origin 김포 --destinations 제주,부산 --date-range "내일부터 2일" --human ``` Round-trip offset scan across destinations: ```bash python skills/korea-domestic-flights/scripts/search_destination_date_matrix.py --origin 김포 --destinations 제주,부산 --date-range "다음주말" --return-offset 2 --human ``` ### 5) Chat-friendly wrapper Use this first when the user is chatting naturally and you want the easiest invocation. Single route: ```bash python skills/korea-domestic-flights/scripts/chat_search.py --origin 김포 --destination 제주 --when 내일 ``` Date range: ```bash python skills/korea-domestic-flights/scripts/chat_search.py --origin 김포 --destination 제주 --when "내일부터 3일" ``` Multi-destination + range: ```bash python skills/korea-domestic-flights/scripts/chat_search.py --origin 김포 --destinations 제주,부산 --when "내일부터 2일" --return-offset 1 --time-pref "출발 10시 이후, 늦은 시간 선호" ``` Add `--json` when structured output is needed; otherwise it defaults to a human-readable Korean briefing. ### 0) Hybrid diagnostics smoke/live checks Fixture-based regression/smoke check: ```bash python skills/korea-domestic-flights/scripts/hybrid_smoke_check.py ``` Shallow environment-only check: ```bash python skills/korea-domestic-flights/scripts/hybrid_live_dry_run.py ``` Optional shallow live probe (non-brittle, no fare assertion): ```bash python skills/korea-domestic-flights/scripts/hybrid_live_dry_run.py --probe ``` ### 6) Price alert / watch rules Store a single-date alert: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py add --origin 김포 --destination 제주 --departure 내일 --target-price 70000 --label "김포-제주 내일 특가" ``` Store a date-range alert: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py add --origin 김포 --destination 제주 --date-range "내일부터 3일" --target-price 80000 --label "김포-제주 3일 범위 감시" ``` Store a time-filtered alert: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py add --origin 김포 --destination 제주 --date-range "다음주말" --return-offset 2 --target-price 150000 --time-pref "복귀 18시 이후, 늦은 시간 선호" --label "주말 늦복 왕복 감시" ``` Store a multi-destination watch: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py add --origin 김포 --destinations 제주,부산,여수 --departure 내일 --target-price 90000 --label "김포 출발 내일 다중 목적지 감시" ``` Store a round-trip range alert with fixed return offset: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py add --origin 김포 --destination 제주 --date-range "다음주말" --return-offset 2 --target-price 150000 --label "주말 왕복 특가" ``` Store a custom message format: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py add --origin 김포 --destinations 제주,부산 --date-range "내일부터 3일" --target-price 85000 --message-template "[특가감시] {best_destination_label} {departure_date} {observed_price} / 기준 {target_price}" ``` Check all active rules: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py check ``` List saved rules: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py list ``` Remove a rule: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py remove --rule-id kdf-1234abcd ``` When a rule matches, `check` prints a human-readable Korean alert message to stdout so an upper-layer cron/briefing flow can forward it directly. ## Parameters `search_domestic.py` - `--origin`: 출발 공항 코드 또는 한글 공항명 - `--destination`: 도착 공항 코드 또는 한글 공항명 - `--departure`: 출발일 (`YYYY-MM-DD`, `YYYYMMDD`, `내일`, `모레`, `이번주말`, `다음주 금요일` 등) - `--return-date`: 귀국일 (선택) - `--adults`: 성인 수, 기본값 `1` - `--cabin`: `ECONOMY|BUSINESS|FIRST` - `--max-results`: 최대 결과 수 - `--time-pref`: 자연어 시간 조건 (`오전`, `오후`, `저녁`, `출발 10시 이후`, `복귀 18시 이후`, `너무 이른 비행 제외 8시`, `늦은 시간 선호` 등) - `--depart-after`, `--return-after`, `--exclude-early-before`: 옵션 기반 시간 필터 - `--prefer`: `late|morning|afternoon|evening` 시간 선호 추천 - `--human`: 자연스러운 한국어 브리핑 출력 (`최저가` / `시간대 추천` / `왕복 균형 추천` 중심 구획형 요약) `search_date_range.py` - `--origin`: 출발 공항 코드 또는 한글 공항명 - `--destination`: 도착 공항 코드 또는 한글 공항명 - `--start-date`: 범위 시작일 - `--end-date`: 범위 종료일 - `--date-range`: 자연어 범위 (`내일부터 3일`, `이번주말`, `2026-03-25~2026-03-30`) - `--return-offset`: 왕복 탐색용 귀국 오프셋 일수 - `--adults`: 성인 수 - `--cabin`: `ECONOMY|BUSINESS|FIRST` - `--time-pref`, `--depart-after`, `--return-after`, `--exclude-early-before`, `--prefer`: 시간 필터/시간 선호 추천 - 시간 조건이 있으면 전체 날짜를 병렬로 빠르게 훑은 뒤 저가 후보·인접 날짜·범위 커버리지 앵커를 함께 상세 재검증하는 하이브리드 모드로 전환됨 - 상세 재검증 후 시간 조건 일치 결과가 너무 적거나, 시간조건 탈락/빈결과 유사 패턴이 강하면 fallback 후보 확장을 한 번 더 수행할 수 있음 - `summary.search_metadata` / 최상위 `search_metadata` / `logs` 에 하이브리드 여부, 전체 스캔 수, 초기/추가 재검증 수, fallback 여부, 시간 조건 요약과 `refine_diagnostics`(시간조건 탈락 / usable match 없음 / 빠른스캔-상세 불일치 빈결과 / 출발·복귀 시간정보 부족 / 가격정보 부족 분류, 샘플, user/developer 힌트, `ranked_reasons`, `dominant_reason_code`, `primary_interpretation`)가 기록됨 - fallback 판단 요약은 `fallback_decision` / `fallback_reason_codes` 에 구조화되어 남음 - 결과 요약에는 날짜별 가격 캘린더/히트맵(`summary.price_calendar`)이 포함됨 - `--human`: 자연스러운 한국어 브리핑 출력 `search_multi_destination.py` - `--origin`: 출발 공항 코드 또는 한글 공항명 - `--destinations`: 쉼표로 구분한 여러 목적지 (코드 또는 한글) - `--departure`: 출발일 - `--return-date`: 귀국일 (선택) - `--adults`: 성인 수 - `--cabin`: `ECONOMY|BUSINESS|FIRST` - `--time-pref`, `--depart-after`, `--return-after`, `--exclude-early-before`, `--prefer`: 시간 필터/시간 선호 추천 - `--human`: 목적지 간 가격 차이와 추천이 포함된 한국어 브리핑 출력 `search_destination_date_matrix.py` - `--origin`: 출발 공항 코드 또는 한글 공항명 - `--destinations`: 쉼표로 구분한 여러 목적지 - `--start-date`, `--end-date`: 날짜 범위 지정 - `--date-range`: 자연어 날짜 범위 지정 - `--return-offset`: 출발일 기준 귀국 오프셋 일수 - `--adults`: 성인 수 - `--cabin`: `ECONOMY|BUSINESS|FIRST` - `--time-pref`, `--depart-after`, `--return-after`, `--exclude-early-before`, `--prefer`: 시간 필터/시간 선호 추천 - 시간 조건이 있으면 전체 조합을 목적지별 병렬 스캔으로 먼저 좁힌 뒤 저가 조합·목적지별 후보·인접 날짜 조합·커버리지 앵커를 함께 상세 재검증하는 하이브리드 모드로 전환됨 - 상세 재검증 후 시간 조건 일치 조합이 너무 적거나, 시간조건 탈락/빈결과 유사 패턴이 강하면 fallback 후보 확장을 한 번 더 수행할 수 있음 - `summary.search_metadata` / 최상위 `search_metadata` / `logs` 에 하이브리드 여부, 전체 스캔 수, 초기/추가 재검증 수, fallback 여부, 시간 조건 요약과 `refine_diagnostics`(시간조건 탈락 / usable match 없음 / scraper-empty 유사 분류 / 시간·가격 정보 완전누락·부분누락, 샘플, `human_hint`/`developer_hint`, `extraction_summary`, `ranked_reasons`, `dominant_reason_code`, `primary_interpretation`)가 기록됨 - fallback 판단 요약은 `fallback_decision` / `fallback_reason_codes` 에 구조화되어 남음 - `--human`: 전체 최적 조합 + 목적지별 베스트 브리핑 출력 `chat_search.py` - `--origin`: 출발 공항명/코드 - `--destination`: 단일 목적지 - `--destinations`: 다중 목적지 - `--when`: 자연어 날짜/날짜범위 - `--departure`: 명시적 출발일 - `--return-date`: 명시적 귀국일 - `--return-offset`: 날짜범위 왕복 오프셋. `chat_search.py`에서는 다중 목적지 + 명시적 `--departure` 와 함께 써도 단일일 매트릭스 왕복 탐색으로 라우팅됨 - `--time-pref`: 자연어 시간 선호/필터 입력 - `--depart-after`, `--return-after`, `--exclude-early-before`, `--prefer`: 옵션 기반 시간 필터 - `--json`: JSON 출력, 생략 시 사람이 읽기 쉬운 브리핑 출력 `price_alerts.py` - `add`: 감시 규칙 저장 - `--origin`: 출발 공항 - `--destination` 또는 `--destinations`: 단일/다중 목적지 - `--departure`: 단일 날짜 감시 - `--return-date`: 왕복 귀국일 - `--date-range`: 날짜 범위 감시 - `--return-offset`: 날짜 범위 왕복 감시 시 귀국 오프셋 - `--adults`: 성인 수 - `--cabin`: `ECONOMY|BUSINESS|FIRST` - `--target-price`: 목표가(원) - `--label`: 사람이 읽을 이름 - `--time-pref`, `--depart-after`, `--return-after`, `--exclude-early-before`, `--prefer`: 시간대 조건/선호 저장 - `--message-template`: 커스텀 알림 포맷 - `--store`: 저장 파일 경로 오버라이드 - 동일 조건+목표가 규칙은 fingerprint 기준으로 중복 저장되지 않음 - `list`: 저장된 규칙 목록 출력 - `check`: 활성 규칙 점검, 목표가 충족 시 한국어 알림 출력 - 동일한 결과는 `notify.dedupe_key` 기준으로 재알림 억제 - `--no-dedupe` 로 강제 재출력 가능 - `render`: 마지막 `last_result` 를 현재 템플릿으로 미리보기 - `remove`: 규칙 삭제 ## References Read these only when needed: - `references/domestic-airports.md`: 국내 공항 코드/이름 매핑 - `references/price-alerts-schema.md`: 가격 감시 JSON 저장 포맷, dedupe, 다중 목적지 감시, 메시지 템플릿, cron/스케줄러 연결 힌트 ## Operational notes - This skill depends on a working Playwright browser environment. - If browser init fails, install or repair Chromium/Chrome/Edge in the source repo environment. - The provider site DOM may change; if results suddenly disappear, the upstream scraper may need maintenance. - For stable chat use, prefer `chat_search.py` or `--human` summaries unless structured JSON is explicitly needed. - Prefer domestic routes only; if the user asks for ICN-NRT or any overseas route, do not use this skill. FILE:README.md # korea-domestic-flights 대한민국 **국내선 항공권 검색·비교·날짜 범위 탐색·가격 감시**를 위한 OpenClaw 스킬입니다. 이 스킬은 Playwright 기반 항공권 검색 흐름을 감싸서 다음 같은 작업을 처리합니다. - 김포-제주, 부산-제주 같은 **국내선 단일 노선 검색** - **왕복 검색** 및 시간대 조건 반영 - **날짜 범위 최저가 탐색** - **다중 목적지 비교** - **다중 목적지 + 날짜 범위 최적 조합 탐색** - **가격 캘린더/히트맵 요약** - **목표가 기반 가격 감시 규칙 저장/점검** > 국제선에는 사용하지 않습니다. --- ## 핵심 기능 ### 1. 단일 노선 검색 - 편도/왕복 검색 - 한글 공항명 입력 지원 (`김포`, `제주`, `부산` 등) - 시간대 조건 반영 (`출발 10시 이후`, `복귀 18시 이후` 등) - 추천 사유 설명 출력 ### 2. 날짜 범위 최저가 탐색 - `내일부터 3일`, `이번주말`, `다음주말`, `2026-03-25~2026-03-30` 같은 자연어/명시 범위 입력 지원 - 날짜별 가격 캘린더 제공 - 왕복 검색 시 균형 추천 제공 - 시간 조건이 있을 때는 빠른 전체 스캔 후 저가 후보 + 인접 날짜 + 범위 커버리지 앵커를 함께 상세 재검증하는 하이브리드 최적화 적용 - 시간 조건 하이브리드에서 상세 재검증 결과가 너무 적거나 시간조건 탈락/빈결과 유사 패턴이 강하면 fallback 후보 확장을 한 번 더 수행해 false no-result / false-best 위험을 줄임 ### 3. 다중 목적지 비교 - 예: 김포 출발로 제주/부산/여수 중 어디가 가장 유리한지 비교 - 목적지별 최저가/추천/가격 차이 설명 제공 ### 4. 목적지 + 날짜 범위 매트릭스 탐색 - 여러 목적지와 여러 날짜 조합을 한 번에 비교 - 최적 조합 + 목적지별 가격 캘린더 제공 ### 5. 가격 감시 / 알림 - 목표가 이하로 내려오면 알림 메시지 생성 - 단일 목적지 / 다중 목적지 / 날짜 범위 감시 지원 - 시간대 조건 포함 규칙 저장 가능 - dedupe(중복 억제) 지원 --- ## 빠른 예시 ### 단일 검색 ```bash python skills/korea-domestic-flights/scripts/search_domestic.py --origin 김포 --destination 제주 --departure 내일 --human ``` ### 날짜 범위 검색 ```bash python skills/korea-domestic-flights/scripts/search_date_range.py --origin 김포 --destination 제주 --date-range "내일부터 3일" --human ``` ### 다중 목적지 + 날짜 범위 검색 ```bash python skills/korea-domestic-flights/scripts/search_destination_date_matrix.py --origin 김포 --destinations 제주,부산 --date-range "다음주말" --return-offset 2 --human ``` ### 시간 조건 포함 가격 감시 규칙 저장 ```bash python skills/korea-domestic-flights/scripts/price_alerts.py add --origin 김포 --destination 제주 --date-range "다음주말" --return-offset 2 --target-price 150000 --time-pref "복귀 18시 이후, 늦은 시간 선호" --label "주말 늦복 왕복 감시" ``` --- ## 출력 특징 이 스킬은 단순 최저가만 보여주지 않고, 가능하면 아래 정보도 같이 제공합니다. - 추천 사유 - 시간대 추천 - 왕복 균형 추천 - 날짜별 가격 캘린더 - 목적지별 가격 캘린더 - 사람용 출력에서는 `최저가`, `시간대 추천`, `왕복 균형 추천` 같은 구획을 나눠 더 읽기 쉽게 표시 - 사람용 출력에서는 너무 길어지지 않도록 캘린더를 일부만 미리 보여주고 나머지 일수는 축약 표시 예를 들어: - **추천:** 이번 조건에서는 부산(PUS) 조합이 가장 유리합니다. - **추천 사유:** 2위보다 더 저렴하고, 시간 조건에도 맞습니다. - **왕복 균형 추천:** 아주 약간 더 비싸더라도 시간대가 더 무난한 조합을 별도로 제안할 수 있습니다. --- ## 의존성 이 스킬은 다음 로컬 소스 저장소를 감쌉니다. - `tmp/Scraping-flight-information` 필요 조건: - Playwright/브라우저 실행 가능 환경 - upstream 검색 로직이 정상 동작할 것 환경이 깨졌거나 사이트 DOM이 바뀌면 결과가 없거나 오류가 날 수 있습니다. --- ## 현재 확인된 동작 상태 최근 점검 기준: - 모든 주요 스크립트 `py_compile` 통과 - `price_alerts.py add/list/remove` 동작 확인 - `chat_search.py`를 통한 다중 목적지+날짜 범위 JSON 검색 동작 확인 - `chat_search.py`에서 다중 목적지 + 명시적 출발일 + `--return-offset` 조합이 날짜 매트릭스로 올바르게 라우팅되도록 보정 - 다중 목적지+날짜 범위 검색에서 목적지별 `price_calendar` 출력 확인 - `references/hybrid-smoke-fixtures.json` 기반 회귀/스모크 진단 케이스 확인 - `hybrid_live_dry_run.py`로 환경 전용 또는 얕은 live probe 점검 가능 --- ## 한계 / 주의점 - 실제 검색은 외부 사이트 상태와 브라우저 환경에 영향을 받습니다. - 날짜 범위/다중 목적지/왕복 검색은 실행 시간이 길 수 있지만, 시간 조건이 있는 범위/매트릭스 검색은 하이브리드 최적화로 전체 조합을 먼저 빠르게 훑은 뒤 저가 후보·인접 날짜·커버리지 앵커를 함께 상세 검색합니다. - 상세 재검증 후 시간 조건 일치 결과가 너무 적으면 fallback 후보 확장을 추가로 수행할 수 있습니다. - JSON `summary.search_metadata` / 최상위 `search_metadata` 와 `logs` 에 하이브리드 사용 여부, 초기/추가 재검증 수, fallback 여부, 시간 조건 요약과 `refine_diagnostics`(시간조건 탈락 / usable match 없음 / 빠른스캔-상세 불일치 빈결과 / 시간·가격 정보 완전누락/부분누락 분류, 샘플, `human_hint`/`developer_hint`, `extraction_summary`, `ranked_reasons`, `dominant_reason_code`, `primary_interpretation`)가 기록됩니다. - fallback 판단은 `fallback_decision` / `fallback_reason_codes` 로 구조화되어 남아 extraction incompleteness 우세인지 genuine time-filter rejection 우세인지 구분하기 쉽게 했습니다. - 사람용 출력에서는 필요할 때만 한 줄짜리 `참고:` 진단 힌트를 덧붙이고, 더 자세한 디버그성 힌트와 커버리지 수치는 JSON 메타데이터에만 남겨서 사람용 출력이 시끄러워지지 않게 유지합니다. - 국제선은 범위 밖입니다. 자세한 사용법은 `SKILL.md`를 참고하세요. FILE:references/domestic-airports.md # 대한민국 국내선 주요 공항 코드 - `GMP` — 김포 - `CJU` — 제주 - `PUS` — 부산(김해) - `TAE` — 대구 - `CJJ` — 청주 - `KWJ` — 광주 - `RSU` — 여수 - `USN` — 울산 - `HIN` — 사천 - `KPO` — 포항경주 - `YNY` — 양양 - `MWX` — 무안 - `SEL` — 서울(검색 컨텍스트용 묶음 코드로 쓰일 수 있음) ## 한글 입력 별칭 예시 - `김포` → `GMP` - `제주`, `제주도` → `CJU` - `부산`, `김해` → `PUS` - `청주` → `CJJ` - `광주` → `KWJ` - `여수` → `RSU` - `울산` → `USN` - `사천`, `진주` → `HIN` - `포항`, `포항경주` → `KPO` - `양양` → `YNY` - `무안` → `MWX` - `서울` → `SEL` ## 날짜 입력 예시 - `오늘` - `내일` - `모레` - `2026-03-25` - `20260325` - `2026/03/25` ## 추천 자연어 예시 - 김포에서 제주 가는 내일 최저가 찾아줘 - 부산 제주 왕복 항공권 요약해줘 - 청주 제주 3일 범위로 최저가 비교해줘 - 김포 제주 왕복 2박 기준 날짜별 최저가 찾아줘 - 김포 출발로 제주, 부산, 여수 중 어디가 제일 싼지 비교해줘 FILE:references/hybrid-smoke-fixtures.json { "date_range_cases": [ { "name": "broad-hit-empty-detail-vs-time-rejection", "broad_rows": [ {"departure_date": "2026-03-28", "price": 90000}, {"departure_date": "2026-03-29", "price": 95000}, {"departure_date": "2026-03-30", "price": 98000}, {"departure_date": "2026-03-31", "price": 101000} ], "detailed_rows": [ {"departure_date": "2026-03-28", "broad_price": 90000, "raw_option_count": 4, "priced_option_count": 4, "departure_time_count": 4, "return_time_count": 0, "time_pref_valid_count": 0, "search_stage": "refine", "diagnostic_reason": "broad_candidate_time_rejected", "diagnostic_detail": {"hint": "시간 조건 미충족"}}, {"departure_date": "2026-03-29", "broad_price": 95000, "raw_option_count": 0, "priced_option_count": 0, "departure_time_count": 0, "return_time_count": 0, "time_pref_valid_count": 0, "search_stage": "refine", "diagnostic_reason": "detail_empty_after_broad_hit", "diagnostic_detail": {"hint": "빠른 스캔 대비 상세 빈결과"}} ], "expect": { "dominant_reason": "detail_empty_after_broad_hit", "primary_interpretation": "extraction_incomplete", "fallback_triggered": true, "required_reasons": ["broad_candidate_time_rejected", "detail_empty_after_broad_hit"], "required_fallback_reason_prefixes": ["coverage.time_pref_shortfall", "signal.high_extraction_incomplete_ratio"] } }, { "name": "pure-time-filter-rejection", "broad_rows": [ {"departure_date": "2026-04-04", "price": 87000}, {"departure_date": "2026-04-05", "price": 91000}, {"departure_date": "2026-04-06", "price": 94000} ], "detailed_rows": [ {"departure_date": "2026-04-04", "broad_price": 87000, "raw_option_count": 5, "priced_option_count": 5, "departure_time_count": 5, "return_time_count": 0, "time_pref_valid_count": 0, "search_stage": "refine", "diagnostic_reason": "broad_candidate_time_rejected", "diagnostic_detail": {"hint": "시간 조건 미충족"}}, {"departure_date": "2026-04-05", "broad_price": 91000, "raw_option_count": 4, "priced_option_count": 4, "departure_time_count": 4, "return_time_count": 0, "time_pref_valid_count": 0, "search_stage": "refine", "diagnostic_reason": "detailed_no_usable_time_filter_match", "diagnostic_detail": {"hint": "usable match 없음"}} ], "expect": { "dominant_reason": "broad_candidate_time_rejected", "primary_interpretation": "time_filter_rejection", "fallback_triggered": true, "required_reasons": ["broad_candidate_time_rejected", "detailed_no_usable_time_filter_match"], "required_fallback_reason_prefixes": ["signal.high_time_filter_rejection_ratio", "signal.time_filter_dominant:"] } } ], "matrix_cases": [ { "name": "mixed-matrix-return-time-and-price-gaps", "broad_rows": [ {"destination": "CJU", "departure_date": "2026-03-28", "price": 90000}, {"destination": "CJU", "departure_date": "2026-03-29", "price": 97000}, {"destination": "PUS", "departure_date": "2026-03-28", "price": 88000}, {"destination": "PUS", "departure_date": "2026-03-29", "price": 92000} ], "detailed_rows": [ {"destination": "CJU", "departure_date": "2026-03-28", "broad_price": 90000, "raw_option_count": 3, "priced_option_count": 3, "departure_time_count": 3, "return_time_count": 1, "has_return_time_constraint": true, "time_pref_valid_count": 0, "search_stage": "refine", "diagnostic_reason": "detail_partial_return_times", "diagnostic_detail": {"hint": "복귀 시간 정보 부분 누락"}}, {"destination": "PUS", "departure_date": "2026-03-28", "broad_price": 88000, "raw_option_count": 2, "priced_option_count": 2, "departure_time_count": 2, "return_time_count": 2, "has_return_time_constraint": true, "time_pref_valid_count": 1, "search_stage": "refine", "time_pref_match": true, "diagnostic_reason": "detailed_match_with_time_pref", "diagnostic_detail": {"hint": "시간 조건 일치"}}, {"destination": "PUS", "departure_date": "2026-03-29", "broad_price": 92000, "raw_option_count": 4, "priced_option_count": 2, "departure_time_count": 4, "return_time_count": 4, "has_return_time_constraint": false, "time_pref_valid_count": 0, "search_stage": "refine", "diagnostic_reason": "detail_sparse_price_data", "diagnostic_detail": {"hint": "가격 정보 일부 누락"}} ], "expect": { "required_reasons": ["detailed_match_with_time_pref", "detail_partial_return_times", "detail_sparse_price_data"], "fallback_triggered": true, "required_fallback_reason_prefixes": ["coverage.time_pref_shortfall", "signal.high_extraction_incomplete_ratio"], "extraction_summary": { "partial_return_time_rows": 1, "sparse_price_rows": 1 } } } ] } FILE:references/price-alerts-schema.md # 가격 감시 JSON 포맷 이 스킬의 가격 감시 기능은 기본적으로 `skills/korea-domestic-flights/price-alert-rules.json` 파일을 사용한다. OpenClaw cron/브리핑과 연결하기 쉽게 다음 원칙으로 설계한다. - 파일 1개에 여러 감시 규칙을 저장한다. - 각 규칙은 사람이 읽기 쉬운 `label` 과 안정적인 `id` 를 가진다. - **동일 조건 중복 저장 방지**를 위해 `fingerprint` 를 저장한다. - 마지막 점검 시각(`last_checked_at`)과 마지막 결과(`last_result`)를 함께 저장한다. - **동일 최저가/동일 항공편 조건의 재알림 방지**를 위해 `notify.dedupe_key` 를 저장한다. - **알림 문구 커스터마이즈**를 위해 `notify.message_template` 를 저장한다. - 점검 스크립트는 목표가 충족 시 한국어 알림 메시지를 stdout 으로 출력하므로 cron/briefing 파이프에 바로 연결하기 쉽다. - 향후 알림 채널 확장이 가능하도록 `notify` 필드를 남겨 둔다. ## 스키마 개요 ```json { "version": 2, "timezone": "Asia/Seoul", "updated_at": "2026-03-19T18:40:00+09:00", "rules": [ { "id": "kdf-a1b2c3d4", "enabled": true, "label": "김포→제주,부산 주말 특가 감시", "fingerprint": "{...canonical json...}", "query": { "origin": "GMP", "destination": null, "destinations": ["CJU", "PUS"], "departure": null, "return_date": null, "date_range": { "start_date": "2026-03-21", "end_date": "2026-03-23" }, "return_offset": 0, "adults": 1, "cabin": "ECONOMY", "trip_type": "one_way" }, "target_price_krw": 70000, "created_at": "2026-03-19T18:35:00+09:00", "last_checked_at": "2026-03-19T18:39:12+09:00", "last_result": { "matched": false, "observed_price_krw": 81200, "best_option": { "destination": "CJU", "destination_label": "제주(CJU)", "departure_date": "2026-03-22", "price": 81200, "airline": "제주항공" }, "search_type": "destination_date_matrix" }, "notify": { "channel": "stdout", "dedupe_key": "{...canonical json...}", "last_sent_at": "2026-03-19T18:39:13+09:00", "message_template": "[특가] {label} {best_destination_label} {observed_price}" }, "meta": { "source": "price_alerts.py", "notes": "아침 비행 우선" } } ] } ``` ## 필드 설명 - `version`: 저장 포맷 버전 - `timezone`: 날짜/시간 기준 타임존 - `updated_at`: 파일 마지막 갱신 시각 - `rules[]`: 감시 규칙 배열 - `rules[].id`: 안정적인 규칙 식별자 - `rules[].enabled`: 감시 활성 여부 - `rules[].label`: 사람이 읽는 규칙 이름 - `rules[].fingerprint`: 조건 중복 저장 방지용 canonical signature - `rules[].query.origin`: IATA 코드 기준 저장 - `rules[].query.destination`: 단일 목적지용 - `rules[].query.destinations`: 다중 목적지용 배열 - `rules[].query.departure`, `return_date`: 단일 날짜 감시용 - `rules[].query.date_range`: 날짜 범위 감시용 - `rules[].query.return_offset`: 날짜 범위 왕복 감시에서 귀국일 오프셋 - `rules[].query.adults`, `cabin`: 검색 조건 - `rules[].target_price_krw`: 목표가 - `rules[].last_result`: 마지막 점검 결과 캐시 - `rules[].notify.channel`: 현재는 `stdout` 고정, 향후 확장 대비 - `rules[].notify.dedupe_key`: 최근 발송된 알림 fingerprint - `rules[].notify.last_sent_at`: 마지막 알림 발송 시각 - `rules[].notify.message_template`: 사용자 정의 메시지 템플릿 - `rules[].meta.notes`: 사용자 메모 ## 다중 목적지 감시 동작 - `destinations` 가 2개 이상이면 `price_alerts.py check` 가 자동으로 다중 목적지 검색 스크립트를 사용한다. - 단일 날짜 감시는 `search_multi_destination.py` - 날짜 범위 감시는 `search_destination_date_matrix.py` - 알림은 **가장 저렴한 목적지/날짜 조합 1건** 기준으로 발생한다. ## 중복 방지 동작 ### 1) 규칙 저장 단계 dedupe 동일한 검색 조건 + 목표가 조합으로 `add` 하면 새 규칙을 저장하지 않고 기존 규칙 ID 를 안내한다. ### 2) 알림 발송 단계 dedupe `check` 중 목표가를 만족하더라도 다음 값이 같으면 같은 알림으로 간주하고 재출력하지 않는다. - 규칙 ID - 검색 타입 - 관측 최저가 - 최적 목적지 - 출발일/귀국일 - 항공사 - 출발/도착 시각 강제로 다시 출력하려면: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py check --no-dedupe ``` ## 메시지 템플릿 변수 `add --message-template` 로 커스텀 알림 포맷을 저장할 수 있다. 자주 쓸 만한 변수: - `{label}` - `{route}` - `{origin_label}` - `{destinations_label}` - `{best_destination_label}` - `{target_price}` / `{target_price_krw}` - `{observed_price}` / `{observed_price_krw}` - `{difference_krw}` - `{departure_date}` / `{return_date}` - `{date_text}` - `{airline}` - `{departure_time}` / `{arrival_time}` - `{cabin_label}` - `{status_line}` 예시: ```bash python skills/korea-domestic-flights/scripts/price_alerts.py add \ --origin 김포 \ --destinations 제주,부산 \ --date-range "내일부터 3일" \ --target-price 80000 \ --message-template "[특가감시] {best_destination_label} {departure_date} {observed_price} / 기준 {target_price}" ``` ## 추천 cron 연결 방식 정기 점검에서 다음처럼 실행하면 된다. ```bash python skills/korea-domestic-flights/scripts/price_alerts.py check ``` 동작 방식: - 목표가 미충족이면 `점검 완료: N개 규칙 확인, 목표가 충족 알림 없음` 출력 - 목표가 충족이면 규칙별 한국어 알림 메시지를 stdout 으로 출력 - 동일 알림은 `notify.dedupe_key` 기준으로 억제 - 일부 규칙 점검 오류가 있으면 stderr 에 오류를 남기고 JSON 저장 파일에는 `last_result.error` 를 기록 즉, 상위 OpenClaw 브리핑/cron 레이어에서는 stdout 을 메시지 본문으로 사용하고, stderr 또는 종료 코드를 장애 감지에 활용하면 된다. ## Windows 작업 스케줄러 초안 Windows 환경에서는 다음 스크립트 초안을 사용할 수 있다. ```powershell powershell -ExecutionPolicy Bypass -File skills/korea-domestic-flights/scripts/register_price_alerts_task.ps1 -IntervalMinutes 30 ``` 주의: - 이 스크립트는 **작업 등록만** 돕는다. - 실제 알림 전송은 상위 OpenClaw cron/브리핑 파이프가 stdout 을 받아 전달하는 방식으로 연결해야 한다. FILE:scripts/chat_search.py #!/usr/bin/env python3 import argparse import subprocess import sys from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from common_cli import parse_date_range_text, time_preference_cli_args def run_script(script_name: str, extra_args: list[str]) -> int: command = [sys.executable, str(SCRIPT_DIR / script_name), *extra_args] return subprocess.call(command) def time_args(args) -> list[str]: return time_preference_cli_args({ "time_pref": args.time_pref, "depart_after": args.depart_after, "return_after": args.return_after, "exclude_early_before": args.exclude_early_before, "prefer": args.prefer, }) def main(): parser = argparse.ArgumentParser(description="Chat-friendly wrapper for Korea domestic flight search") parser.add_argument("--origin", required=True, help="예: 김포") parser.add_argument("--destination", help="단일 목적지") parser.add_argument("--destinations", help="여러 목적지, 예: 제주,부산,여수") parser.add_argument("--when", help="단일 날짜 또는 날짜 범위. 예: 내일, 이번주말, 내일부터 3일") parser.add_argument("--departure") parser.add_argument("--return-date") parser.add_argument("--return-offset", type=int, default=0) parser.add_argument("--adults", type=int, default=1) parser.add_argument("--cabin", default="ECONOMY", choices=["ECONOMY", "BUSINESS", "FIRST"]) parser.add_argument("--time-pref", help="예: 오전, 저녁, 출발 10시 이후, 복귀 18시 이후, 너무 이른 비행 제외 8시, 늦은 시간 선호") parser.add_argument("--depart-after") parser.add_argument("--return-after") parser.add_argument("--exclude-early-before") parser.add_argument("--prefer", choices=["late", "morning", "afternoon", "evening"]) parser.add_argument("--json", action="store_true", help="JSON 출력") args = parser.parse_args() human = [] if args.json else ["--human"] extra_time_args = time_args(args) has_multi_dest = bool(args.destinations and "," in args.destinations or (args.destinations and args.destination)) destinations_value = args.destinations or args.destination if not destinations_value: raise SystemExit("--destination 또는 --destinations 가 필요합니다.") if args.when and not args.departure: start_dt, end_dt = parse_date_range_text(args.when) inferred_single_day = start_dt == end_dt if has_multi_dest and not inferred_single_day: return run_script( "search_destination_date_matrix.py", [ "--origin", args.origin, "--destinations", destinations_value, "--start-date", start_dt.strftime("%Y-%m-%d"), "--end-date", end_dt.strftime("%Y-%m-%d"), "--return-offset", str(args.return_offset), "--adults", str(args.adults), "--cabin", args.cabin, *extra_time_args, *human, ], ) if has_multi_dest: return run_script( "search_multi_destination.py", [ "--origin", args.origin, "--destinations", destinations_value, "--departure", start_dt.strftime("%Y-%m-%d"), *(["--return-date", args.return_date] if args.return_date else []), "--adults", str(args.adults), "--cabin", args.cabin, *extra_time_args, *human, ], ) if not inferred_single_day: return run_script( "search_date_range.py", [ "--origin", args.origin, "--destination", destinations_value, "--start-date", start_dt.strftime("%Y-%m-%d"), "--end-date", end_dt.strftime("%Y-%m-%d"), "--return-offset", str(args.return_offset), "--adults", str(args.adults), "--cabin", args.cabin, *extra_time_args, *human, ], ) return run_script( "search_domestic.py", [ "--origin", args.origin, "--destination", destinations_value, "--departure", start_dt.strftime("%Y-%m-%d"), *(["--return-date", args.return_date] if args.return_date else []), "--adults", str(args.adults), "--cabin", args.cabin, *extra_time_args, *human, ], ) if has_multi_dest and args.departure and args.return_offset > 0 and not args.return_date: return run_script( "search_destination_date_matrix.py", [ "--origin", args.origin, "--destinations", destinations_value, "--start-date", args.departure, "--end-date", args.departure, "--return-offset", str(args.return_offset), "--adults", str(args.adults), "--cabin", args.cabin, *extra_time_args, *human, ], ) if has_multi_dest and args.departure: return run_script( "search_multi_destination.py", [ "--origin", args.origin, "--destinations", destinations_value, "--departure", args.departure, *( ["--return-date", args.return_date] if args.return_date else []), "--adults", str(args.adults), "--cabin", args.cabin, *extra_time_args, *human, ], ) if args.departure: return run_script( "search_domestic.py", [ "--origin", args.origin, "--destination", destinations_value, "--departure", args.departure, *(["--return-date", args.return_date] if args.return_date else []), "--adults", str(args.adults), "--cabin", args.cabin, *extra_time_args, *human, ], ) raise SystemExit("날짜 정보가 없습니다. --when 또는 --departure 를 제공하세요.") if __name__ == "__main__": main() FILE:scripts/common_cli.py from __future__ import annotations import re from dataclasses import dataclass from datetime import datetime, timedelta from typing import Iterable, List, Sequence AIRPORT_NAMES = { "GMP": "김포", "CJU": "제주", "PUS": "부산", "TAE": "대구", "CJJ": "청주", "KWJ": "광주", "RSU": "여수", "USN": "울산", "HIN": "사천", "KPO": "포항경주", "YNY": "양양", "MWX": "무안", "SEL": "서울", } AIRPORT_ALIASES = { "김포": "GMP", "제주": "CJU", "제주도": "CJU", "부산": "PUS", "김해": "PUS", "대구": "TAE", "청주": "CJJ", "광주": "KWJ", "여수": "RSU", "울산": "USN", "사천": "HIN", "진주": "HIN", "포항": "KPO", "포항경주": "KPO", "양양": "YNY", "무안": "MWX", "서울": "SEL", "gimpo": "GMP", "jeju": "CJU", "busan": "PUS", "daegu": "TAE", "cheongju": "CJJ", "gwangju": "KWJ", "yeosu": "RSU", "ulsan": "USN", "sacheon": "HIN", "pohang": "KPO", "yangyang": "YNY", "muan": "MWX", "seoul": "SEL", } WEEKDAY_ALIASES = { "월": 0, "월요일": 0, "화": 1, "화요일": 1, "수": 2, "수요일": 2, "목": 3, "목요일": 3, "금": 4, "금요일": 4, "토": 5, "토요일": 5, "일": 6, "일요일": 6, } TIME_BUCKETS = { "새벽": (0, 5), "아침": (6, 10), "오전": (6, 11), "점심": (11, 13), "오후": (12, 17), "저녁": (18, 21), "밤": (20, 23), "야간": (20, 23), "늦은": (18, 23), } def airport_label(code: str) -> str: code = (code or "").upper() return f"{AIRPORT_NAMES.get(code, code)}({code})" if code else "" def normalize_airport(value: str) -> str: if not value: raise ValueError("공항 값이 비어 있습니다.") raw = value.strip() upper = raw.upper() if upper in AIRPORT_NAMES: return upper lowered = raw.lower() if lowered in AIRPORT_ALIASES: return AIRPORT_ALIASES[lowered] if raw in AIRPORT_ALIASES: return AIRPORT_ALIASES[raw] raise ValueError(f"지원하지 않는 공항 입력입니다: {value}") def _base_today() -> datetime: return datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) def _parse_month_day(raw: str, today: datetime) -> datetime | None: match = re.fullmatch(r"(\d{1,2})[./\-월\s]+(\d{1,2})(?:일)?", raw) if not match: return None month = int(match.group(1)) day = int(match.group(2)) year = today.year candidate = datetime(year, month, day) if candidate < today: candidate = datetime(year + 1, month, day) return candidate def _parse_relative_days(raw: str, today: datetime) -> datetime | None: match = re.fullmatch(r"(\d+)\s*(?:일 뒤|일후|days? later)", raw) if match: return today + timedelta(days=int(match.group(1))) match = re.fullmatch(r"(\d+)\s*(?:주 뒤|주후)", raw) if match: return today + timedelta(days=7 * int(match.group(1))) return None def _next_weekday(today: datetime, weekday: int, week_offset: int = 0) -> datetime: days_ahead = (weekday - today.weekday()) % 7 candidate = today + timedelta(days=days_ahead + week_offset * 7) if week_offset == 0 and candidate < today: candidate += timedelta(days=7) return candidate def _parse_weekday(raw: str, today: datetime) -> datetime | None: raw = raw.strip() for prefix, offset in (("이번주 ", 0), ("이번 주 ", 0), ("다음주 ", 1), ("다음 주 ", 1), ("오는 ", 0)): if raw.startswith(prefix): tail = raw[len(prefix):].strip() if tail in WEEKDAY_ALIASES: return _next_weekday(today, WEEKDAY_ALIASES[tail], offset) if raw in WEEKDAY_ALIASES: return _next_weekday(today, WEEKDAY_ALIASES[raw], 0) if raw in ("주말", "이번주말", "이번 주말"): return _next_weekday(today, 5, 0) if raw in ("다음주말", "다음 주말"): return _next_weekday(today, 5, 1) return None def parse_flexible_date(value: str) -> datetime: raw = value.strip().lower() today = _base_today() mapping = { "today": 0, "오늘": 0, "tomorrow": 1, "내일": 1, "day after tomorrow": 2, "모레": 2, "글피": 3, } if raw in mapping: return today + timedelta(days=mapping[raw]) relative = _parse_relative_days(raw, today) if relative: return relative weekday = _parse_weekday(raw, today) if weekday: return weekday month_day = _parse_month_day(raw.replace(" ", " "), today) if month_day: return month_day for fmt in ("%Y-%m-%d", "%Y%m%d", "%Y.%m.%d", "%Y/%m/%d"): try: return datetime.strptime(value, fmt) except ValueError: pass raise ValueError(f"지원하지 않는 날짜 형식입니다: {value}") def parse_date_range_text(value: str) -> tuple[datetime, datetime]: raw = value.strip().lower() today = _base_today() m = re.fullmatch(r"(.+?)부터\s*(\d+)일", raw) if m: start = parse_flexible_date(m.group(1)) days = int(m.group(2)) return start, start + timedelta(days=max(days - 1, 0)) if raw in ("이번주말", "이번 주말", "주말"): start = _next_weekday(today, 5, 0) return start, start + timedelta(days=1) if raw in ("다음주말", "다음 주말"): start = _next_weekday(today, 5, 1) return start, start + timedelta(days=1) explicit_patterns = [ r"^\s*(\d{4}[-./]\d{1,2}[-./]\d{1,2}|\d{8})\s*~\s*(\d{4}[-./]\d{1,2}[-./]\d{1,2}|\d{8})\s*$", r"^\s*(\d{4}[-./]\d{1,2}[-./]\d{1,2}|\d{8})\s*to\s*(\d{4}[-./]\d{1,2}[-./]\d{1,2}|\d{8})\s*$", r"^\s*(\d{4}[-./]\d{1,2}[-./]\d{1,2}|\d{8})\s*부터\s*(\d{4}[-./]\d{1,2}[-./]\d{1,2}|\d{8})\s*(?:까지)?\s*$", ] for pattern in explicit_patterns: match = re.fullmatch(pattern, value, re.IGNORECASE) if match: start = parse_flexible_date(match.group(1).strip()) end = parse_flexible_date(match.group(2).strip()) return start, end parts = re.split(r"\s*(?:~|부터|to)\s*", value, maxsplit=1) if len(parts) == 2 and all(part.strip() for part in parts): start = parse_flexible_date(parts[0].strip()) end = parse_flexible_date(parts[1].strip()) return start, end single = parse_flexible_date(value) return single, single def pretty_date(value: datetime) -> str: return value.strftime("%Y-%m-%d") def compact_date(value: datetime) -> str: return value.strftime("%Y%m%d") def cabin_label(code: str) -> str: return { "ECONOMY": "이코노미", "BUSINESS": "비즈니스", "FIRST": "일등석", }.get((code or "").upper(), code or "") def format_price(value: int | float | None) -> str: return f"{int(value or 0):,}원" def format_time_or_fallback(value: str | None, fallback: str = "시간 정보 없음") -> str: text = str(value or "").strip() return text or fallback def join_nonempty(parts: Sequence[str | None], sep: str = " · ") -> str: return sep.join([str(part) for part in parts if part]) def add_section(lines: list[str], title: str, body: Sequence[str | None]) -> None: items = [str(item) for item in body if item] if not items: return if lines: lines.append("") lines.append(f"[{title}]") lines.extend(items) def summarize_price_gap(best_price: int, next_price: int | None) -> str | None: if not best_price or not next_price or next_price <= best_price: return None gap = next_price - best_price ratio = round((gap / best_price) * 100) return f"2위보다 {gap:,}원 저렴해 가성비가 좋습니다{f' (약 {ratio}% 차이)' if ratio >= 5 else ''}." def recommendation_line(subject: str, best_price: int, next_price: int | None = None) -> str: gap_text = summarize_price_gap(best_price, next_price) if gap_text: return f"추천: 이번 조건에서는 {subject}이(가) 가장 유리합니다. {gap_text}" return f"추천: 이번 조건에서는 {subject}이(가) 가장 무난한 최저가 선택입니다." def explain_recommendation(subject: str, best_price: int, next_price: int | None = None, reasons: Sequence[str] | None = None) -> str: parts: list[str] = [f"추천 사유: {subject} 기준"] if best_price > 0: parts.append(f"최저가 {format_price(best_price)}") gap_text = summarize_price_gap(best_price, next_price) if gap_text: parts.append(gap_text) if reasons: parts.extend([reason for reason in reasons if reason]) return " · ".join(parts) + "." def bullet_rank_lines(items: Sequence[dict], label_key: str, price_key: str, detail_builder=None, limit: int = 5) -> List[str]: lines: List[str] = [] for idx, item in enumerate(items[:limit], start=1): label = item.get(label_key, "옵션") price = item.get(price_key, 0) if price and price > 0: detail = detail_builder(item) if detail_builder else "" suffix = f" · {detail}" if detail else "" lines.append(f"{idx}. {label} · {format_price(price)}{suffix}") else: lines.append(f"{idx}. {label} · 결과 없음") return lines def build_price_calendar(rows: Sequence[dict], date_key: str = "departure_date", price_key: str = "price") -> list[dict]: available_prices = [int(item.get(price_key, 0) or 0) for item in rows if int(item.get(price_key, 0) or 0) > 0] if not rows: return [] cheapest = min(available_prices) if available_prices else 0 median = sorted(available_prices)[len(available_prices) // 2] if available_prices else 0 calendar = [] for item in rows: price = int(item.get(price_key, 0) or 0) if price <= 0: band = "unavailable" badge = "⚪" note = "결과 없음" elif cheapest and price == cheapest: band = "best" badge = "🟢" note = "최저가" elif median and price <= median: band = "good" badge = "🟡" note = "양호" else: band = "high" badge = "🔴" note = "상대적으로 높음" calendar.append({ "date": item.get(date_key), "price": price, "band": band, "badge": badge, "label": f"{item.get(date_key)} {badge} {format_price(price) if price > 0 else '결과 없음'} · {note}", }) return calendar def unique_codes(values: Iterable[str]) -> list[str]: seen = set() output = [] for value in values: if value not in seen: seen.add(value) output.append(value) return output def parse_time_to_minutes(value: str | None) -> int | None: if not value: return None raw = str(value).strip() match = re.fullmatch(r"(\d{1,2})(?::(\d{2}))?", raw) if not match: return None hour = int(match.group(1)) minute = int(match.group(2) or 0) if 0 <= hour <= 23 and 0 <= minute <= 59: return hour * 60 + minute return None @dataclass class TimePreference: depart_min: int | None = None depart_max: int | None = None return_min: int | None = None return_max: int | None = None exclude_before_depart: int | None = None prefer: str | None = None raw: str = "" def active(self) -> bool: return any( value is not None for value in [self.depart_min, self.depart_max, self.return_min, self.return_max, self.exclude_before_depart] ) or bool(self.prefer) def describe(self) -> str | None: parts: list[str] = [] if self.depart_min is not None: parts.append(f"출발 {format_minutes(self.depart_min)} 이후") if self.depart_max is not None: parts.append(f"출발 {format_minutes(self.depart_max)} 이전") if self.return_min is not None: parts.append(f"복귀 {format_minutes(self.return_min)} 이후") if self.return_max is not None: parts.append(f"복귀 {format_minutes(self.return_max)} 이전") if self.exclude_before_depart is not None: parts.append(f"너무 이른 비행 제외({format_minutes(self.exclude_before_depart)} 이전 제외)") if self.prefer: prefer_map = { "late": "늦은 시간대 선호", "morning": "오전 선호", "afternoon": "오후 선호", "evening": "저녁 선호", } parts.append(prefer_map.get(self.prefer, self.prefer)) return " · ".join(parts) if parts else None def _set_min(current: int | None, value: int) -> int: return value if current is None else max(current, value) def _set_max(current: int | None, value: int) -> int: return value if current is None else min(current, value) def _split_time_segments(text: str) -> list[str]: if not text: return [] parts = [part.strip() for part in re.split(r"\s*(?:,|/|\||;| 그리고 | and )\s*", text) if part.strip()] return parts or [text.strip()] def _segment_scope(segment: str) -> str: if any(keyword in segment for keyword in ["복귀", "귀환", "오는편", "오는 편", "리턴"]): return "return" return "depart" def _apply_bucket(pref: TimePreference, scope: str, start_hour: int, end_hour: int) -> None: start_minutes = start_hour * 60 end_minutes = end_hour * 60 + 59 if scope == "return": pref.return_min = _set_min(pref.return_min, start_minutes) pref.return_max = _set_max(pref.return_max, end_minutes) else: pref.depart_min = _set_min(pref.depart_min, start_minutes) pref.depart_max = _set_max(pref.depart_max, end_minutes) def _normalize_time_ranges(pref: TimePreference, raw: str) -> None: if pref.depart_min is not None and pref.depart_max is not None and pref.depart_min > pref.depart_max: if "출발" in raw and "이후" in raw: pref.depart_max = None elif "출발" in raw and "이전" in raw: pref.depart_min = None if pref.return_min is not None and pref.return_max is not None and pref.return_min > pref.return_max: if any(token in raw for token in ["복귀", "귀환", "오는편", "오는 편"]) and "이후" in raw: pref.return_max = None elif any(token in raw for token in ["복귀", "귀환", "오는편", "오는 편"]) and "이전" in raw: pref.return_min = None def apply_time_overrides(pref: TimePreference, *, depart_after: str | None = None, return_after: str | None = None, exclude_early_before: str | None = None, prefer: str | None = None) -> TimePreference: if depart_after: minutes = parse_time_to_minutes(str(depart_after)) if minutes is None: raise ValueError(f"지원하지 않는 출발 시간 형식입니다: {depart_after}") pref.depart_min = _set_min(pref.depart_min, minutes) if return_after: minutes = parse_time_to_minutes(str(return_after)) if minutes is None: raise ValueError(f"지원하지 않는 복귀 시간 형식입니다: {return_after}") pref.return_min = _set_min(pref.return_min, minutes) if exclude_early_before: minutes = parse_time_to_minutes(str(exclude_early_before)) if minutes is None: raise ValueError(f"지원하지 않는 제외 시간 형식입니다: {exclude_early_before}") pref.exclude_before_depart = _set_min(pref.exclude_before_depart, minutes) if prefer: pref.prefer = prefer return pref def parse_time_preference_args(args) -> TimePreference: return apply_time_overrides( parse_time_preference_text(getattr(args, "time_pref", None)), depart_after=getattr(args, "depart_after", None), return_after=getattr(args, "return_after", None), exclude_early_before=getattr(args, "exclude_early_before", None), prefer=getattr(args, "prefer", None), ) def time_preference_cli_args(time_pref: dict | None) -> list[str]: tp = time_pref or {} args: list[str] = [] if tp.get("time_pref"): args.extend(["--time-pref", str(tp["time_pref"])]) if tp.get("depart_after"): args.extend(["--depart-after", str(tp["depart_after"])]) if tp.get("return_after"): args.extend(["--return-after", str(tp["return_after"])]) if tp.get("exclude_early_before"): args.extend(["--exclude-early-before", str(tp["exclude_early_before"])]) if tp.get("prefer"): args.extend(["--prefer", str(tp["prefer"])]) return args def describe_time_preference_payload(time_pref: dict | None) -> str | None: tp = time_pref or {} pref = apply_time_overrides( parse_time_preference_text(tp.get("time_pref")), depart_after=tp.get("depart_after"), return_after=tp.get("return_after"), exclude_early_before=tp.get("exclude_early_before"), prefer=tp.get("prefer"), ) return pref.describe() def format_minutes(value: int | None) -> str: if value is None: return "" hour = value // 60 minute = value % 60 return f"{hour:02d}:{minute:02d}" def parse_time_preference_text(text: str | None) -> TimePreference: pref = TimePreference(raw=text or "") if not text: return pref normalized = str(text).strip().lower() normalized = normalized.replace("시 이후", "시이후").replace("시 이전", "시이전").replace("시 전", "시이전") for segment in _split_time_segments(normalized): scope = _segment_scope(segment) for key, (start_hour, end_hour) in TIME_BUCKETS.items(): if key in segment: prefer_only = f"{key} 선호" in segment or f"{key}시간 선호" in segment or f"{key} 시간 선호" in segment if not prefer_only: _apply_bucket(pref, scope, start_hour, end_hour) if "늦은 시간" in segment or "늦게" in segment: pref.prefer = pref.prefer or "late" elif "오전 선호" in segment: pref.prefer = pref.prefer or "morning" elif "오후 선호" in segment: pref.prefer = pref.prefer or "afternoon" elif "저녁 선호" in segment: pref.prefer = pref.prefer or "evening" for pattern, target in [ (r"출발\s*(\d{1,2})(?::(\d{2}))?\s*시?이후", "depart_min"), (r"출발\s*(\d{1,2})(?::(\d{2}))?\s*시?이전", "depart_max"), (r"복귀\s*(\d{1,2})(?::(\d{2}))?\s*시?이후", "return_min"), (r"귀환\s*(\d{1,2})(?::(\d{2}))?\s*시?이후", "return_min"), (r"오는편\s*(\d{1,2})(?::(\d{2}))?\s*시?이후", "return_min"), (r"오는 편\s*(\d{1,2})(?::(\d{2}))?\s*시?이후", "return_min"), (r"복귀\s*(\d{1,2})(?::(\d{2}))?\s*시?이전", "return_max"), (r"귀환\s*(\d{1,2})(?::(\d{2}))?\s*시?이전", "return_max"), (r"오는편\s*(\d{1,2})(?::(\d{2}))?\s*시?이전", "return_max"), (r"오는 편\s*(\d{1,2})(?::(\d{2}))?\s*시?이전", "return_max"), (r"너무\s*이른\s*비행\s*제외.*?(\d{1,2})(?::(\d{2}))?\s*시", "exclude_before_depart"), (r"(\d{1,2})(?::(\d{2}))?\s*시\s*이전\s*비행\s*제외", "exclude_before_depart"), ]: match = re.search(pattern, segment) if match: minutes = int(match.group(1)) * 60 + int(match.group(2) or 0) current = getattr(pref, target) if target.endswith("_max"): setattr(pref, target, _set_max(current, minutes) if current is not None else minutes) elif target == "exclude_before_depart": setattr(pref, target, _set_min(current, minutes) if current is not None else minutes) else: setattr(pref, target, _set_min(current, minutes) if current is not None else minutes) _normalize_time_ranges(pref, segment) return pref def _within_range(value_minutes: int | None, min_minutes: int | None, max_minutes: int | None) -> bool: if value_minutes is None: return False if min_minutes is not None and value_minutes < min_minutes: return False if max_minutes is not None and value_minutes > max_minutes: return False return True def _score_time_preference(item: dict, pref: TimePreference) -> int: depart = parse_time_to_minutes(item.get("departure_time")) ret = parse_time_to_minutes(item.get("return_departure_time")) score = 0 if pref.prefer == "late": score += depart or 0 score += (ret or 0) // 2 elif pref.prefer == "morning": score -= abs((depart or 12 * 60) - 9 * 60) elif pref.prefer == "afternoon": score -= abs((depart or 12 * 60) - 15 * 60) elif pref.prefer == "evening": score -= abs((depart or 12 * 60) - 19 * 60) return score def filter_and_rank_by_time_preference(items: Sequence[dict], pref: TimePreference) -> tuple[list[dict], list[dict]]: if not pref.active(): return list(items), sorted(list(items), key=lambda x: x.get("price", 0) if x.get("price", 0) > 0 else 10**12) filtered = [] for item in items: depart = parse_time_to_minutes(item.get("departure_time")) ret = parse_time_to_minutes(item.get("return_departure_time")) if pref.exclude_before_depart is not None and (depart is None or depart < pref.exclude_before_depart): continue if (pref.depart_min is not None or pref.depart_max is not None) and not _within_range(depart, pref.depart_min, pref.depart_max): continue if item.get("is_round_trip") and (pref.return_min is not None or pref.return_max is not None) and not _within_range(ret, pref.return_min, pref.return_max): continue filtered.append(item) ranked = sorted( filtered, key=lambda x: ( -_score_time_preference(x, pref), x.get("price", 0) if x.get("price", 0) > 0 else 10**12, ), ) filtered.sort(key=lambda x: x.get("price", 0) if x.get("price", 0) > 0 else 10**12) return filtered, ranked def choose_preferred_option(items: Sequence[dict], pref: TimePreference) -> dict | None: _, ranked = filter_and_rank_by_time_preference(items, pref) return ranked[0] if ranked else None def choose_balanced_round_trip_option(items: Sequence[dict], pref: TimePreference | None = None) -> dict | None: candidates = [item for item in items if (item.get("price", 0) or 0) > 0] if not candidates: return None candidates = sorted(candidates, key=lambda x: x.get("price", 0)) cheapest_price = candidates[0].get("price", 0) or 0 price_cap = int(cheapest_price * 1.15) if cheapest_price else 0 pool = [item for item in candidates if (item.get("price", 0) or 0) <= price_cap][:5] or candidates[:3] def score(item: dict) -> tuple: price = int(item.get("price", 0) or 0) depart = parse_time_to_minutes(item.get("departure_time")) or 0 ret = parse_time_to_minutes(item.get("return_departure_time")) or 0 time_score = _score_time_preference(item, pref or TimePreference()) if pref else (depart // 2 + ret) return ( -time_score, price, ) return sorted(pool, key=score)[0] if pool else candidates[0] def round_trip_balance_recommendation(balanced: dict | None, cheapest: dict | None, pref: TimePreference | None = None) -> str | None: if not balanced: return None label = balanced.get("airline") or "옵션" depart = format_time_or_fallback(balanced.get("departure_time")) ret = format_time_or_fallback(balanced.get("return_departure_time")) if cheapest and cheapest.get("price", 0) and balanced.get("price", 0): gap = int(balanced.get("price", 0)) - int(cheapest.get("price", 0)) gap_text = "최저가와 동일 가격" if gap == 0 else (f"최저가 대비 {gap:,}원 추가" if gap > 0 else f"최저가보다 {-gap:,}원 저렴") else: gap_text = "가격 비교 정보 없음" reason = pref.describe() if pref and pref.describe() else "왕복 시간 균형" return f"왕복 균형 추천: {reason} 기준으로는 {label} 가는편 {depart}, 오는편 {ret} 조합이 무난합니다 ({gap_text})." def build_best_option_reasons(best: dict | None, next_price: int | None = None, pref: TimePreference | None = None) -> list[str]: if not best: return [] reasons: list[str] = [] airline = best.get("airline") depart = best.get("departure_time") arrival = best.get("arrival_time") if airline: if depart and arrival: reasons.append(f"항공사 {airline}, 가는편 {depart}→{arrival}") else: reasons.append(f"항공사 {airline}") if best.get("return_departure_time") or best.get("return_arrival_time"): ret_depart = format_time_or_fallback(best.get("return_departure_time")) ret_arrive = best.get("return_arrival_time") if ret_arrive: reasons.append(f"오는편 {ret_depart}→{ret_arrive}") else: reasons.append(f"오는편 {ret_depart} 출발") if pref and pref.describe(): reasons.append(f"시간 조건 '{pref.describe()}' 반영") price = int(best.get("price", 0) or best.get("cheapest_price", 0) or 0) if next_price and price and next_price > price: reasons.append(f"차상위 대비 {next_price - price:,}원 저렴") return reasons def build_balanced_option_reasons(balanced: dict | None, cheapest: dict | None = None, pref: TimePreference | None = None) -> list[str]: if not balanced: return [] reasons: list[str] = [] depart = format_time_or_fallback(balanced.get("departure_time")) ret = format_time_or_fallback(balanced.get("return_departure_time")) reasons.append(f"왕복 시간대 {depart} / {ret} 조합") if pref and pref.describe(): reasons.append(f"시간 선호 '{pref.describe()}'에 더 잘 맞음") price = int(balanced.get("price", 0) or 0) cheapest_price = int((cheapest or {}).get("price", 0) or 0) if price and cheapest_price: gap = price - cheapest_price if gap == 0: reasons.append("최저가와 동일한 가격") elif gap > 0: reasons.append(f"최저가 대비 추가 비용 {gap:,}원 수준") return reasons def time_preference_recommendation(preferred: dict | None, cheapest: dict | None, pref: TimePreference) -> str | None: if not pref.active() or not preferred: return None subject = preferred.get("airline", "옵션") depart = format_time_or_fallback(preferred.get("departure_time")) price = preferred.get("price", 0) if cheapest and cheapest is not preferred and cheapest.get("price", 0) > 0 and price > 0: gap = price - cheapest.get("price", 0) gap_text = f"최저가 대비 {gap:,}원 추가" if gap > 0 else "최저가와 동일 가격" else: gap_text = "최저가와 같은 옵션" detail = pref.describe() or "시간 선호" if preferred.get("is_round_trip") and preferred.get("return_departure_time"): return f"시간대 추천: {detail} 기준으로는 {subject} 가는편 {depart}, 오는편 {format_time_or_fallback(preferred.get('return_departure_time'))} 옵션이 적합합니다 ({gap_text})." return f"시간대 추천: {detail} 기준으로는 {subject} {depart} 출발 옵션이 적합합니다 ({gap_text})." FILE:scripts/hybrid_live_dry_run.py #!/usr/bin/env python3 import argparse import json import sys from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from common_cli import normalize_airport def main(): parser = argparse.ArgumentParser(description="Shallow live dry-run / DOM drift sanity check for korea-domestic-flights") parser.add_argument("--origin", default="김포") parser.add_argument("--destination", default="제주") parser.add_argument("--departure", default="내일", help="Currently informational only; live probe uses broad date-range scanner.") parser.add_argument("--probe", action="store_true", help="Actually attempt a 1-day broad scan. Without this flag, only environment/import checks run.") args = parser.parse_args() workspace = Path(__file__).resolve().parents[3] repo_path = workspace / "tmp" / "Scraping-flight-information" report = { "status": "ok", "mode": "live_probe" if args.probe else "env_only", "checks": [], "repo_path": str(repo_path), "probe": None, } report["checks"].append({"name": "repo_exists", "ok": repo_path.exists()}) if not repo_path.exists(): report["status"] = "degraded" print(json.dumps(report, ensure_ascii=False, indent=2)) sys.exit(0) sys.path.insert(0, str(repo_path)) try: from scraping.parallel import ParallelSearcher except Exception as exc: report["status"] = "degraded" report["checks"].append({"name": "import_parallel_searcher", "ok": False, "error": str(exc)}) print(json.dumps(report, ensure_ascii=False, indent=2)) sys.exit(0) report["checks"].append({"name": "import_parallel_searcher", "ok": True}) if not args.probe: print(json.dumps(report, ensure_ascii=False, indent=2)) return try: origin = normalize_airport(args.origin) destination = normalize_airport(args.destination) except Exception as exc: report["status"] = "degraded" report["probe"] = {"ok": False, "error": f"invalid route input: {exc}"} print(json.dumps(report, ensure_ascii=False, indent=2)) sys.exit(0) searcher = ParallelSearcher() probe_logs = [] try: raw = searcher.search_date_range( origin=origin, destination=destination, dates=[], return_offset=0, adults=1, cabin_class="ECONOMY", progress_callback=lambda msg: probe_logs.append(str(msg)), ) report["probe"] = { "ok": isinstance(raw, dict), "route": f"{origin}-{destination}", "raw_type": type(raw).__name__, "keys": list(raw.keys())[:5], "log_preview": probe_logs[:10], "note": "Empty-date broad scan probe completed. This validates import + shallow execution path without asserting fare availability.", } except TypeError as exc: report["status"] = "degraded" report["probe"] = { "ok": False, "route": f"{origin}-{destination}", "error": str(exc), "note": "ParallelSearcher.search_date_range signature or call contract may have drifted.", } except Exception as exc: report["status"] = "degraded" report["probe"] = { "ok": False, "route": f"{origin}-{destination}", "error": str(exc), "log_preview": probe_logs[:10], "note": "Live probe failed; inspect upstream scraper/browser environment or DOM drift.", } finally: close_fn = getattr(searcher, "close", None) if callable(close_fn): try: close_fn() except Exception: pass print(json.dumps(report, ensure_ascii=False, indent=2)) if __name__ == "__main__": main() FILE:scripts/hybrid_observability.py from __future__ import annotations from typing import Callable, Iterable EMPTY_LIKE_REASONS = { "not_attempted", "broad_only", "detailed_match", "detailed_match_with_time_pref", } REASON_LABELS = { "not_attempted": "미시도", "broad_only": "빠른 스캔 전용", "detailed_match": "상세 검색 일치", "detailed_match_with_time_pref": "시간 조건 일치", "broad_candidate_time_rejected": "시간 조건 탈락", "detailed_no_usable_time_filter_match": "usable match 없음", "detail_empty_after_broad_hit": "빠른 스캔 가격 있었지만 상세 결과 비어 있음", "detail_empty_no_broad_signal": "상세 결과 비어 있음", "detail_missing_departure_times": "출발 시간 정보 부족", "detail_partial_departure_times": "출발 시간 정보 부분 누락", "detail_missing_return_times": "복귀 시간 정보 부족", "detail_partial_return_times": "복귀 시간 정보 부분 누락", "detail_missing_price_data": "가격 정보 부족", "detail_sparse_price_data": "가격 정보 일부 누락", } REASON_CODES = { "not_attempted": "diagnostic.not_attempted", "broad_only": "diagnostic.broad_only", "detailed_match": "success.detail_match", "detailed_match_with_time_pref": "success.time_pref_match", "broad_candidate_time_rejected": "filter.time_pref_rejected", "detailed_no_usable_time_filter_match": "filter.no_usable_match", "detail_empty_after_broad_hit": "extraction.detail_empty_after_broad_hit", "detail_empty_no_broad_signal": "extraction.detail_empty_no_broad_signal", "detail_missing_departure_times": "extraction.missing_departure_times", "detail_partial_departure_times": "extraction.partial_departure_times", "detail_missing_return_times": "extraction.missing_return_times", "detail_partial_return_times": "extraction.partial_return_times", "detail_missing_price_data": "extraction.missing_price_data", "detail_sparse_price_data": "extraction.sparse_price_data", } REASON_CATEGORIES = { "not_attempted": "diagnostic", "broad_only": "diagnostic", "detailed_match": "success", "detailed_match_with_time_pref": "success", "broad_candidate_time_rejected": "time_filter", "detailed_no_usable_time_filter_match": "time_filter", "detail_empty_after_broad_hit": "extraction", "detail_empty_no_broad_signal": "extraction", "detail_missing_departure_times": "extraction", "detail_partial_departure_times": "extraction", "detail_missing_return_times": "extraction", "detail_partial_return_times": "extraction", "detail_missing_price_data": "extraction", "detail_sparse_price_data": "extraction", } REASON_PRIORITY = { "detail_empty_after_broad_hit": 100, "detail_missing_return_times": 95, "detail_missing_departure_times": 94, "detail_missing_price_data": 93, "detail_partial_return_times": 92, "detail_partial_departure_times": 91, "detail_sparse_price_data": 90, "broad_candidate_time_rejected": 70, "detailed_no_usable_time_filter_match": 69, "detailed_match_with_time_pref": 10, "detailed_match": 9, "broad_only": 2, "not_attempted": 1, } def short_reason_label(reason: str) -> str: return REASON_LABELS.get(reason, reason) def reason_code(reason: str | None) -> str | None: if not reason: return None return REASON_CODES.get(reason, f"unknown.{reason}") def reason_category(reason: str | None) -> str | None: if not reason: return None return REASON_CATEGORIES.get(reason, "unknown") def classify_refine_row(row: dict | None) -> str: if not row: return "not_attempted" explicit = str(row.get("diagnostic_reason") or "").strip() if explicit: return explicit if row.get("search_stage") == "broad_only": return "broad_only" raw_count = int(row.get("raw_option_count") or 0) valid_count = int(row.get("time_pref_valid_count") or 0) broad_price = int(row.get("broad_price") or 0) if valid_count > 0: return "detailed_match_with_time_pref" if row.get("time_pref_match") else "detailed_match" if raw_count <= 0: return "detail_empty_after_broad_hit" if broad_price > 0 else "detail_empty_no_broad_signal" if broad_price > 0: return "broad_candidate_time_rejected" return "detailed_no_usable_time_filter_match" def _detail_hint(row: dict) -> str | None: detail = row.get("diagnostic_detail") or {} hint = str(detail.get("hint") or "").strip() return hint or None def _rank_reasons(counts: dict[str, int]) -> list[dict]: ranked = [] for reason, count in counts.items(): ranked.append({ "reason": reason, "reason_code": reason_code(reason), "label": short_reason_label(reason), "category": reason_category(reason), "count": count, "priority": REASON_PRIORITY.get(reason, 0), }) ranked.sort(key=lambda item: (-item["priority"], -item["count"], item["reason"])) return ranked def build_refine_diagnostics( broad_rows: Iterable[dict], detailed_rows: Iterable[dict], *, key_fn: Callable[[dict], object], label_fn: Callable[[dict], str], sample_limit: int = 5, ) -> dict: broad_rows = list(broad_rows) detailed_rows = list(detailed_rows) broad_map = {key_fn(row): row for row in broad_rows} detailed_map = {key_fn(row): row for row in detailed_rows} counts: dict[str, int] = {} samples: dict[str, list[str]] = {} hint_counts: dict[str, int] = {} hint_samples: list[str] = [] extraction_totals = { "rows": 0, "raw_options": 0, "priced_options": 0, "departure_time_options": 0, "return_time_options": 0, "missing_price_rows": 0, "sparse_price_rows": 0, "missing_departure_time_rows": 0, "partial_departure_time_rows": 0, "missing_return_time_rows": 0, "partial_return_time_rows": 0, } for key, broad in broad_map.items(): detailed = detailed_map.get(key) if detailed is None: merged = dict(broad) merged.setdefault("search_stage", "broad_only") else: merged = dict(broad) merged.update(detailed) reason = classify_refine_row(merged) counts[reason] = counts.get(reason, 0) + 1 samples.setdefault(reason, []) sample_label = label_fn(merged) hint = _detail_hint(merged) if hint: sample_label = f"{sample_label} ({hint})" hint_counts[hint] = hint_counts.get(hint, 0) + 1 if len(hint_samples) < sample_limit and hint not in hint_samples: hint_samples.append(hint) if len(samples[reason]) < sample_limit: samples[reason].append(sample_label) if detailed is not None: raw_option_count = int(merged.get("raw_option_count") or 0) priced_option_count = int(merged.get("priced_option_count") or 0) departure_time_count = int(merged.get("departure_time_count") or 0) return_time_count = int(merged.get("return_time_count") or 0) extraction_totals["rows"] += 1 extraction_totals["raw_options"] += raw_option_count extraction_totals["priced_options"] += priced_option_count extraction_totals["departure_time_options"] += departure_time_count extraction_totals["return_time_options"] += return_time_count if raw_option_count > 0: if priced_option_count <= 0: extraction_totals["missing_price_rows"] += 1 elif priced_option_count < raw_option_count: extraction_totals["sparse_price_rows"] += 1 if departure_time_count <= 0: extraction_totals["missing_departure_time_rows"] += 1 elif departure_time_count < raw_option_count: extraction_totals["partial_departure_time_rows"] += 1 if bool(merged.get("has_return_time_constraint")): if return_time_count <= 0: extraction_totals["missing_return_time_rows"] += 1 elif return_time_count < raw_option_count: extraction_totals["partial_return_time_rows"] += 1 attempted = len(detailed_map) broad_available = sum(1 for row in broad_rows if (row.get("price") or 0) > 0) success = counts.get("detailed_match_with_time_pref", 0) rejected = counts.get("broad_candidate_time_rejected", 0) + counts.get("detailed_no_usable_time_filter_match", 0) extraction_incomplete = sum( counts.get(reason, 0) for reason in ( "detail_empty_after_broad_hit", "detail_empty_no_broad_signal", "detail_missing_departure_times", "detail_partial_departure_times", "detail_missing_return_times", "detail_partial_return_times", "detail_missing_price_data", "detail_sparse_price_data", ) ) empty_like = ( counts.get("detail_empty_after_broad_hit", 0) + counts.get("detail_empty_no_broad_signal", 0) ) remaining_available = max(0, broad_available - attempted) attempted_with_broad = max(1, sum(1 for row in detailed_rows if (row.get("broad_price") or 0) > 0)) extraction_summary = { **extraction_totals, "price_coverage_ratio": extraction_totals["priced_options"] / max(1, extraction_totals["raw_options"]), "departure_time_coverage_ratio": extraction_totals["departure_time_options"] / max(1, extraction_totals["raw_options"]), "return_time_coverage_ratio": extraction_totals["return_time_options"] / max(1, extraction_totals["raw_options"]), } summary_bits = [] if success: summary_bits.append(f"시간 조건 일치 {success}건") if counts.get("broad_candidate_time_rejected"): summary_bits.append(f"시간 조건 탈락 {counts['broad_candidate_time_rejected']}건") if counts.get("detailed_no_usable_time_filter_match"): summary_bits.append(f"usable match 없음 {counts['detailed_no_usable_time_filter_match']}건") if counts.get("detail_empty_after_broad_hit"): summary_bits.append(f"빠른 스캔가 있었지만 상세 비어 있음 {counts['detail_empty_after_broad_hit']}건") if counts.get("detail_empty_no_broad_signal"): summary_bits.append(f"상세 빈결과 {counts['detail_empty_no_broad_signal']}건") if counts.get("detail_missing_departure_times"): summary_bits.append(f"출발 시간 정보 부족 {counts['detail_missing_departure_times']}건") if counts.get("detail_partial_departure_times"): summary_bits.append(f"출발 시간 부분 누락 {counts['detail_partial_departure_times']}건") if counts.get("detail_missing_return_times"): summary_bits.append(f"복귀 시간 정보 부족 {counts['detail_missing_return_times']}건") if counts.get("detail_partial_return_times"): summary_bits.append(f"복귀 시간 부분 누락 {counts['detail_partial_return_times']}건") if counts.get("detail_missing_price_data"): summary_bits.append(f"가격 정보 부족 {counts['detail_missing_price_data']}건") if counts.get("detail_sparse_price_data"): summary_bits.append(f"가격 정보 일부 누락 {counts['detail_sparse_price_data']}건") if extraction_totals["partial_departure_time_rows"] and not counts.get("detail_partial_departure_times"): summary_bits.append(f"출발 시간 일부 누락 후보 {extraction_totals['partial_departure_time_rows']}건") if extraction_totals["partial_return_time_rows"] and not counts.get("detail_partial_return_times"): summary_bits.append(f"복귀 시간 일부 누락 후보 {extraction_totals['partial_return_time_rows']}건") if extraction_totals["sparse_price_rows"] and not counts.get("detail_sparse_price_data"): summary_bits.append(f"가격 일부 누락 후보 {extraction_totals['sparse_price_rows']}건") if not summary_bits: summary_bits.append("상세 진단 데이터 없음") ranked_reasons = _rank_reasons(counts) dominant_reason = ranked_reasons[0]["reason"] if ranked_reasons else None user_hint = None developer_hint = None if counts.get("detail_empty_after_broad_hit"): user_hint = "일부 날짜/조합은 빠른 스캔 가격이 있었지만 상세 단계에서 빈결과가 나와 추가 후보를 다시 확인했습니다." developer_hint = "broad/detail 불일치 빈도가 있어 upstream DOM 변화나 상세 페이지 추출 불안정 가능성을 점검하세요." elif counts.get("detail_missing_return_times") or counts.get("detail_partial_return_times"): if counts.get("detail_missing_return_times"): user_hint = "일부 왕복 후보는 복귀 시간 정보가 부족해 시간 조건 판단에서 제외됐습니다." else: user_hint = "일부 왕복 후보는 복귀 시간 정보가 들쭉날쭉해 시간 조건 판단 신뢰도가 낮았습니다." developer_hint = "return_departure_time 추출 커버리지와 왕복 결과 shape 변화를 점검하세요." elif counts.get("detail_missing_departure_times") or counts.get("detail_partial_departure_times"): if counts.get("detail_missing_departure_times"): user_hint = "일부 후보는 출발 시간 정보가 부족해 시간 조건 판단에서 제외됐습니다." else: user_hint = "일부 후보는 출발 시간 정보가 부분 누락되어 시간 조건 판단 신뢰도가 낮았습니다." developer_hint = "departure_time 추출 커버리지와 시간 셀렉터 안정성을 점검하세요." elif counts.get("detail_missing_price_data") or counts.get("detail_sparse_price_data"): user_hint = "일부 상세 후보는 가격 정보가 불완전해 제외됐습니다." developer_hint = "price 추출 커버리지와 결과 카드별 가격 필드 누락 비율을 점검하세요." elif counts.get("broad_candidate_time_rejected"): user_hint = "빠른 스캔 최저가 중 일부는 요청한 시간 조건과 맞지 않아 제외됐습니다." human_hint = user_hint return { "counts": counts, "reason_codes": {reason: reason_code(reason) for reason in counts}, "reason_categories": {reason: reason_category(reason) for reason in counts}, "ranked_reasons": ranked_reasons, "samples": samples, "summary_text": ", ".join(summary_bits), "broad_available": broad_available, "attempted": attempted, "success": success, "rejected": rejected, "extraction_incomplete": extraction_incomplete, "empty_like": empty_like, "remaining_available": remaining_available, "rejection_ratio": rejected / attempted_with_broad, "empty_like_ratio": empty_like / max(1, attempted), "extraction_incomplete_ratio": extraction_incomplete / max(1, attempted), "dominant_reason": dominant_reason, "dominant_reason_code": reason_code(dominant_reason), "dominant_reason_category": reason_category(dominant_reason), "dominant_reason_label": short_reason_label(dominant_reason) if dominant_reason else None, "primary_interpretation": "extraction_incomplete" if reason_category(dominant_reason) == "extraction" else ("time_filter_rejection" if reason_category(dominant_reason) == "time_filter" else reason_category(dominant_reason)), "hint_counts": hint_counts, "hint_samples": hint_samples, "human_hint": human_hint, "user_hint": user_hint, "developer_hint": developer_hint, "extraction_summary": extraction_summary, } def choose_fallback_plan(diag: dict, *, minimum_target: int, hard_cap: int, pad: int) -> dict: broad_available = int(diag.get("broad_available") or 0) success = int(diag.get("success") or 0) remaining_available = int(diag.get("remaining_available") or 0) rejected = int(diag.get("rejected") or 0) extraction_incomplete = int(diag.get("extraction_incomplete") or 0) empty_like = int(diag.get("empty_like") or 0) rejection_ratio = float(diag.get("rejection_ratio") or 0.0) empty_like_ratio = float(diag.get("empty_like_ratio") or 0.0) extraction_incomplete_ratio = float(diag.get("extraction_incomplete_ratio") or 0.0) dominant_reason = str(diag.get("dominant_reason") or "") dominant_category = str(diag.get("dominant_reason_category") or "") target = min(max(1, minimum_target), broad_available) if broad_available else 0 shortfall = max(0, target - success) reasons = [] if shortfall > 0: reasons.append("coverage.time_pref_shortfall") if success == 0 and rejected > 0: reasons.append("coverage.zero_success_with_rejections") if remaining_available > 0 and rejection_ratio >= 0.5: reasons.append("signal.high_time_filter_rejection_ratio") if remaining_available > 0 and extraction_incomplete > 0 and extraction_incomplete_ratio >= 0.35: reasons.append("signal.high_extraction_incomplete_ratio") if remaining_available > 0 and empty_like > 0 and empty_like_ratio >= 0.35: reasons.append("signal.high_empty_like_ratio") if remaining_available > 0 and dominant_reason == "detail_empty_after_broad_hit": reasons.append("signal.broad_detail_disagreement") if remaining_available > 0 and dominant_category == "extraction": reasons.append(f"signal.extraction_dominant:{reason_code(dominant_reason)}") if remaining_available > 0 and dominant_category == "time_filter": reasons.append(f"signal.time_filter_dominant:{reason_code(dominant_reason)}") triggered = bool(reasons) and remaining_available > 0 limit = 0 if triggered: extra_pad = pad if dominant_reason == "detail_empty_after_broad_hit": extra_pad += 1 if dominant_category == "extraction": extra_pad += 1 limit = min(remaining_available, max(shortfall + extra_pad, min(hard_cap, remaining_available))) return { "triggered": triggered, "target": target, "shortfall": shortfall, "limit": limit, "reasons": reasons, "primary_reason": reasons[0] if reasons else None, "dominant_reason": dominant_reason or None, "dominant_reason_code": reason_code(dominant_reason), "dominant_reason_category": dominant_category or None, } FILE:scripts/hybrid_smoke_check.py #!/usr/bin/env python3 import json from pathlib import Path from hybrid_observability import build_refine_diagnostics, choose_fallback_plan FIXTURE_PATH = Path(__file__).resolve().parents[1] / "references" / "hybrid-smoke-fixtures.json" def _load_fixtures(): return json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) def _assert_reason_prefixes(actual_reasons, expected_prefixes): for prefix in expected_prefixes: assert any(reason.startswith(prefix) for reason in actual_reasons), f"missing fallback reason prefix: {prefix}" def _run_case(case, *, key_fn, label_fn, minimum_target=2, hard_cap=4, pad=2): diag = build_refine_diagnostics( case["broad_rows"], case["detailed_rows"], key_fn=key_fn, label_fn=label_fn, ) plan = choose_fallback_plan(diag, minimum_target=minimum_target, hard_cap=hard_cap, pad=pad) expect = case.get("expect", {}) for reason in expect.get("required_reasons", []): assert diag["counts"].get(reason), f"missing expected reason: {reason}" if "dominant_reason" in expect: assert diag["dominant_reason"] == expect["dominant_reason"], f"unexpected dominant_reason: {diag['dominant_reason']}" if "primary_interpretation" in expect: assert diag["primary_interpretation"] == expect["primary_interpretation"], f"unexpected interpretation: {diag['primary_interpretation']}" if "fallback_triggered" in expect: assert plan["triggered"] is expect["fallback_triggered"], f"unexpected fallback trigger state: {plan['triggered']}" _assert_reason_prefixes(plan["reasons"], expect.get("required_fallback_reason_prefixes", [])) for key, value in expect.get("extraction_summary", {}).items(): assert diag["extraction_summary"].get(key) == value, f"unexpected extraction_summary[{key}]: {diag['extraction_summary'].get(key)}" assert diag["ranked_reasons"], "ranked reasons should not be empty" if diag.get("dominant_reason"): assert diag.get("dominant_reason_code"), "dominant reason code missing" assert diag.get("dominant_reason_category"), "dominant reason category missing" return {"diagnostics": diag, "fallback_plan": plan} def run_date_range_cases(): fixtures = _load_fixtures() outputs = {} for case in fixtures.get("date_range_cases", []): outputs[case["name"]] = _run_case( case, key_fn=lambda row: row["departure_date"], label_fn=lambda row: row["departure_date"], ) return outputs def run_matrix_cases(): fixtures = _load_fixtures() outputs = {} for case in fixtures.get("matrix_cases", []): outputs[case["name"]] = _run_case( case, key_fn=lambda row: (row["destination"], row["departure_date"]), label_fn=lambda row: f"{row['destination']} {row['departure_date']}", ) return outputs if __name__ == "__main__": print(json.dumps({ "fixture_path": str(FIXTURE_PATH), "date_range_cases": run_date_range_cases(), "matrix_cases": run_matrix_cases(), }, ensure_ascii=False, indent=2)) FILE:scripts/price_alerts.py #!/usr/bin/env python3 from __future__ import annotations import argparse import json import subprocess import sys import uuid from datetime import datetime from pathlib import Path from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent SKILL_DIR = SCRIPT_DIR.parent DEFAULT_STORE = SKILL_DIR / "price-alert-rules.json" KST_LABEL = "Asia/Seoul" DEFAULT_MESSAGE_TEMPLATE = """[국내선 가격 알림] {label} - 노선: {route} - 조건: 성인 {adults}명 · {cabin_label} - 목표가: {target_price} - 확인된 최저가: {observed_price} - 일정: {date_text}{best_destination_line}{airline_line}{time_line} - 상태: {status_line}""" if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from common_cli import airport_label, cabin_label, describe_time_preference_payload, format_price, normalize_airport, parse_date_range_text, parse_flexible_date, pretty_date, time_preference_cli_args, unique_codes def now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") def load_store(path: Path) -> dict[str, Any]: if not path.exists(): return { "version": 2, "timezone": KST_LABEL, "updated_at": now_iso(), "rules": [], } data = json.loads(path.read_text(encoding="utf-8")) data.setdefault("version", 2) data.setdefault("timezone", KST_LABEL) data.setdefault("rules", []) for rule in data["rules"]: rule.setdefault("notify", {}) rule["notify"].setdefault("channel", "stdout") rule["notify"].setdefault("dedupe_key", None) rule["notify"].setdefault("last_sent_at", None) rule["notify"].setdefault("message_template", None) return data def save_store(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) data["updated_at"] = now_iso() path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def _parse_destinations(args) -> list[str]: if args.destinations: return unique_codes([normalize_airport(x.strip()) for x in args.destinations.split(",") if x.strip()]) if args.destination: return [normalize_airport(args.destination)] raise ValueError("--destination 또는 --destinations 가 필요합니다.") def canonical_signature(payload: dict[str, Any]) -> str: return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def make_rule(args) -> dict[str, Any]: origin = normalize_airport(args.origin) destinations = _parse_destinations(args) if args.date_range: start_dt, end_dt = parse_date_range_text(args.date_range) departure = None return_date = None date_range = { "start_date": pretty_date(start_dt), "end_date": pretty_date(end_dt), } else: departure = pretty_date(parse_flexible_date(args.departure)) return_date = pretty_date(parse_flexible_date(args.return_date)) if args.return_date else None date_range = None trip_type = "round_trip" if return_date or args.return_offset > 0 else "one_way" destination_label = airport_label(destinations[0]) if len(destinations) == 1 else ", ".join(airport_label(code) for code in destinations) date_label = f"{date_range['start_date']}~{date_range['end_date']}" if date_range else departure label = args.label or f"{airport_label(origin)}→{destination_label} {date_label}" time_preference = { "time_pref": args.time_pref, "depart_after": args.depart_after, "return_after": args.return_after, "exclude_early_before": args.exclude_early_before, "prefer": args.prefer, } fingerprint_payload = { "origin": origin, "destinations": destinations, "departure": departure, "return_date": return_date, "date_range": date_range, "return_offset": args.return_offset, "adults": args.adults, "cabin": args.cabin, "trip_type": trip_type, "target_price_krw": args.target_price, "time_preference": time_preference, } return { "id": args.rule_id or f"kdf-{uuid.uuid4().hex[:8]}", "enabled": True, "label": label, "fingerprint": canonical_signature(fingerprint_payload), "query": { "origin": origin, "destination": destinations[0] if len(destinations) == 1 else None, "destinations": destinations, "departure": departure, "return_date": return_date, "date_range": date_range, "return_offset": args.return_offset, "adults": args.adults, "cabin": args.cabin, "trip_type": trip_type, "time_preference": time_preference, }, "target_price_krw": args.target_price, "created_at": now_iso(), "last_checked_at": None, "last_result": None, "notify": { "channel": "stdout", "dedupe_key": None, "last_sent_at": None, "message_template": args.message_template or None, }, "meta": { "source": "price_alerts.py", "notes": args.notes or "", }, } def describe_rule(rule: dict[str, Any]) -> str: q = rule["query"] destinations = q.get("destinations") or ([q["destination"]] if q.get("destination") else []) destination_text = ", ".join(airport_label(code) for code in destinations) if q.get("date_range"): date_text = f"{q['date_range']['start_date']}~{q['date_range']['end_date']}" if q.get("return_offset"): date_text += f" (귀국 +{q['return_offset']}일)" else: date_text = q["departure"] if q.get("return_date"): date_text += f" ~ {q['return_date']}" state = "ON" if rule.get("enabled", True) else "OFF" template_flag = "사용자 템플릿" if rule.get("notify", {}).get("message_template") else "기본 템플릿" time_pref = q.get("time_preference") or {} time_pref_text = describe_time_preference_payload(time_pref) time_pref_line = f"\n- 시간 조건: {time_pref_text}" if time_pref_text else "" return ( f"[{state}] {rule['id']} | {rule['label']}\n" f"- 노선: {airport_label(q['origin'])} → {destination_text}\n" f"- 일정: {date_text}\n" f"- 조건: 성인 {q['adults']}명 · {cabin_label(q['cabin'])} · 목표가 {format_price(rule['target_price_krw'])}{time_pref_line}\n" f"- 알림: {template_flag} · 마지막 dedupe_key={rule.get('notify', {}).get('dedupe_key') or '없음'}" ) def run_search(script_name: str, params: list[str]) -> dict[str, Any]: command = [sys.executable, str(SCRIPT_DIR / script_name), *params] result = subprocess.run(command, capture_output=True, text=True, encoding="utf-8") if result.returncode != 0: raise RuntimeError(result.stdout or result.stderr or f"검색 스크립트 실패: {script_name}") try: return json.loads(result.stdout) except json.JSONDecodeError as exc: raise RuntimeError(f"검색 결과를 JSON으로 해석하지 못했습니다: {exc}\n{result.stdout[:800]}") from exc def check_rule(rule: dict[str, Any]) -> dict[str, Any]: q = rule["query"] destinations = q.get("destinations") or ([q["destination"]] if q.get("destination") else []) common = [ "--origin", q["origin"], "--adults", str(q["adults"]), "--cabin", q["cabin"], ] tp = q.get("time_preference") or {} common.extend(time_preference_cli_args(tp)) if len(destinations) > 1 and q.get("date_range"): payload = run_search( "search_destination_date_matrix.py", [ *common, "--destinations", ",".join(destinations), "--start-date", q["date_range"]["start_date"], "--end-date", q["date_range"]["end_date"], "--return-offset", str(q.get("return_offset", 0)), ], ) best = payload.get("summary", {}).get("best_combo") matched = bool(best and best.get("price", 0) and best["price"] <= rule["target_price_krw"]) return { "matched": matched, "observed_price_krw": best.get("price", 0) if best else 0, "best_option": best, "search_type": "destination_date_matrix", "raw_summary": payload.get("summary"), } if len(destinations) > 1: payload = run_search( "search_multi_destination.py", [ *common, "--destinations", ",".join(destinations), "--departure", q["departure"], *(["--return-date", q["return_date"]] if q.get("return_date") else []), ], ) best = payload.get("summary", {}).get("best_option") matched = bool(best and best.get("cheapest_price", 0) and best["cheapest_price"] <= rule["target_price_krw"]) return { "matched": matched, "observed_price_krw": best.get("cheapest_price", 0) if best else 0, "best_option": { "destination": best.get("destination") if best else None, "destination_label": best.get("destination_label") if best else None, "price": best.get("cheapest_price", 0) if best else 0, "airline": best.get("airline") if best else None, "departure_time": best.get("departure_time") if best else None, "arrival_time": best.get("arrival_time") if best else None, "departure_date": q.get("departure"), "return_date": q.get("return_date"), }, "search_type": "multi_destination", "raw_summary": payload.get("summary"), } destination = destinations[0] single_common = [*common, "--destination", destination] if q.get("date_range"): payload = run_search( "search_date_range.py", [ *single_common, "--start-date", q["date_range"]["start_date"], "--end-date", q["date_range"]["end_date"], "--return-offset", str(q.get("return_offset", 0)), ], ) best = payload.get("summary", {}).get("best_date") matched = bool(best and best.get("price", 0) and best["price"] <= rule["target_price_krw"]) return { "matched": matched, "observed_price_krw": best.get("price", 0) if best else 0, "best_option": best, "search_type": "date_range", "raw_summary": payload.get("summary"), } payload = run_search( "search_domestic.py", [ *single_common, "--departure", q["departure"], *(["--return-date", q["return_date"]] if q.get("return_date") else []), ], ) best = payload.get("cheapest") matched = bool(best and best.get("price", 0) and best["price"] <= rule["target_price_krw"]) return { "matched": matched, "observed_price_krw": best.get("price", 0) if best else 0, "best_option": best, "search_type": "single_date", "raw_summary": payload.get("summary"), } def _safe_format(template: str, context: dict[str, Any]) -> str: class SafeDict(dict): def __missing__(self, key): return "" return template.format_map(SafeDict({k: "" if v is None else v for k, v in context.items()})).strip() def build_notification_context(rule: dict[str, Any], result: dict[str, Any]) -> dict[str, Any]: q = rule["query"] best = result.get("best_option") or {} destinations = q.get("destinations") or ([q["destination"]] if q.get("destination") else []) route = f"{airport_label(q['origin'])} → {', '.join(airport_label(code) for code in destinations)}" observed = result.get("observed_price_krw", 0) target = rule["target_price_krw"] diff = target - observed departure_date = best.get("departure_date") or q.get("departure") or (q.get("date_range") or {}).get("start_date") return_date = best.get("return_date") or q.get("return_date") if q.get("date_range"): date_text = f"최저가 날짜 {departure_date}" if return_date: date_text += f" ~ {return_date}" else: date_text += f" (탐색 범위 {q['date_range']['start_date']}~{q['date_range']['end_date']})" else: date_text = departure_date or "" if return_date: date_text += f" ~ {return_date}" best_destination_label = best.get("destination_label") or airport_label(best.get("destination")) if best.get("destination") else "" status_line = f"목표가 충족 ({diff:,}원 여유)" if diff >= 0 else f"목표가 초과 ({abs(diff):,}원 초과)" best_destination_line = f"\n- 최적 목적지: {best_destination_label}" if best_destination_label else "" airline_line = f"\n- 항공사: {best.get('airline')}" if best.get("airline") else "" time_line = "" if best.get("departure_time") and best.get("arrival_time"): time_line = f"\n- 시간: {best['departure_time']} → {best['arrival_time']}" return { "rule_id": rule["id"], "label": rule["label"], "route": route, "origin": q["origin"], "origin_label": airport_label(q["origin"]), "destinations": ",".join(destinations), "destinations_label": ", ".join(airport_label(code) for code in destinations), "best_destination": best.get("destination") or "", "best_destination_label": best_destination_label, "adults": q["adults"], "cabin": q["cabin"], "cabin_label": cabin_label(q["cabin"]), "target_price": format_price(target), "target_price_krw": target, "observed_price": format_price(observed), "observed_price_krw": observed, "difference_krw": diff, "departure_date": departure_date or "", "return_date": return_date or "", "date_text": date_text, "airline": best.get("airline") or "", "departure_time": best.get("departure_time") or "", "arrival_time": best.get("arrival_time") or "", "search_type": result.get("search_type") or "", "status_line": status_line, "best_destination_line": best_destination_line, "airline_line": airline_line, "time_line": time_line, } def compute_dedupe_key(rule: dict[str, Any], result: dict[str, Any]) -> str: best = result.get("best_option") or {} payload = { "rule_id": rule["id"], "search_type": result.get("search_type"), "observed_price_krw": result.get("observed_price_krw", 0), "destination": best.get("destination"), "departure_date": best.get("departure_date"), "return_date": best.get("return_date"), "airline": best.get("airline"), "departure_time": best.get("departure_time"), "arrival_time": best.get("arrival_time"), } return canonical_signature(payload) def build_notification(rule: dict[str, Any], result: dict[str, Any]) -> str: template = rule.get("notify", {}).get("message_template") or DEFAULT_MESSAGE_TEMPLATE context = build_notification_context(rule, result) return _safe_format(template, context) def command_add(args) -> int: store_path = Path(args.store) data = load_store(store_path) rule = make_rule(args) if any(item["id"] == rule["id"] for item in data["rules"]): raise SystemExit(f"이미 같은 id 규칙이 있습니다: {rule['id']}") duplicate = next((item for item in data["rules"] if item.get("fingerprint") == rule["fingerprint"]), None) if duplicate: raise SystemExit(f"중복 규칙입니다. 기존 규칙 id={duplicate['id']} label={duplicate['label']}") data["rules"].append(rule) save_store(store_path, data) print("규칙 저장 완료") print(describe_rule(rule)) print(f"저장 파일: {store_path}") return 0 def command_list(args) -> int: data = load_store(Path(args.store)) rules = data.get("rules", []) if not rules: print("저장된 가격 감시 규칙이 없습니다.") return 0 for idx, rule in enumerate(rules, start=1): print(f"{idx}. {describe_rule(rule)}") return 0 def command_check(args) -> int: store_path = Path(args.store) data = load_store(store_path) matched_messages: list[str] = [] checked = 0 failures: list[str] = [] suppressed = 0 for rule in data.get("rules", []): if not rule.get("enabled", True): continue if args.rule_id and rule["id"] != args.rule_id: continue checked += 1 try: result = check_rule(rule) rule["last_checked_at"] = now_iso() rule["last_result"] = result if result["matched"]: dedupe_key = compute_dedupe_key(rule, result) if args.no_dedupe or rule.get("notify", {}).get("dedupe_key") != dedupe_key: matched_messages.append(build_notification(rule, result)) rule.setdefault("notify", {})["dedupe_key"] = dedupe_key rule["notify"]["last_sent_at"] = now_iso() else: suppressed += 1 except Exception as exc: failures.append(f"{rule['id']}: {exc}") rule["last_checked_at"] = now_iso() rule["last_result"] = {"matched": False, "error": str(exc)} save_store(store_path, data) if failures: print("[점검 오류]", file=sys.stderr) for item in failures: print(f"- {item}", file=sys.stderr) if matched_messages: print("\n\n".join(matched_messages)) if suppressed: print(f"\n[dedupe] 동일 알림 {suppressed}건은 재전송하지 않았습니다.", file=sys.stderr) return 0 if checked == 0: print("점검할 활성 규칙이 없습니다.") else: msg = f"점검 완료: {checked}개 규칙 확인, 목표가 충족 알림 없음" if suppressed: msg += f" (중복 억제 {suppressed}건)" print(msg) return 0 if not failures else 1 def command_remove(args) -> int: store_path = Path(args.store) data = load_store(store_path) before = len(data.get("rules", [])) data["rules"] = [rule for rule in data.get("rules", []) if rule["id"] != args.rule_id] after = len(data["rules"]) if before == after: raise SystemExit(f"삭제할 규칙을 찾지 못했습니다: {args.rule_id}") save_store(store_path, data) print(f"규칙 삭제 완료: {args.rule_id}") return 0 def command_render(args) -> int: data = load_store(Path(args.store)) rule = next((item for item in data.get("rules", []) if item["id"] == args.rule_id), None) if not rule: raise SystemExit(f"규칙을 찾지 못했습니다: {args.rule_id}") if not rule.get("last_result"): raise SystemExit("아직 last_result 가 없습니다. 먼저 check 를 실행하세요.") print(build_notification(rule, rule["last_result"])) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="대한민국 국내선 가격 감시 규칙 저장/점검 도구") parser.add_argument("--store", default=str(DEFAULT_STORE), help="규칙 JSON 저장 파일 경로") sub = parser.add_subparsers(dest="command", required=True) add = sub.add_parser("add", help="감시 규칙 추가") add.add_argument("--rule-id", help="직접 지정할 규칙 ID") add.add_argument("--label", help="사람이 보기 쉬운 규칙 이름") add.add_argument("--origin", required=True) add_dest = add.add_mutually_exclusive_group(required=True) add_dest.add_argument("--destination", help="단일 목적지") add_dest.add_argument("--destinations", help="쉼표 구분 다중 목적지") add.add_argument("--departure", help="단일 출발일") add.add_argument("--return-date", help="왕복 귀국일") add.add_argument("--date-range", help="날짜 범위. 예: 내일부터 3일, 2026-03-20~2026-03-22") add.add_argument("--return-offset", type=int, default=0, help="날짜 범위 감시에서 출발일 기준 귀국일 오프셋") add.add_argument("--adults", type=int, default=1) add.add_argument("--cabin", default="ECONOMY", choices=["ECONOMY", "BUSINESS", "FIRST"]) add.add_argument("--target-price", type=int, required=True, help="목표가(원)") add.add_argument("--time-pref", help="자연어 시간 조건. 예: 저녁, 출발 10시 이후, 복귀 18시 이후, 늦은 시간 선호") add.add_argument("--depart-after", help="출발 N시 이후. 예: 10, 10:30") add.add_argument("--return-after", help="복귀 N시 이후. 예: 18, 18:30") add.add_argument("--exclude-early-before", help="이 시간 이전 출발 제외. 예: 8, 08:30") add.add_argument("--prefer", choices=["late", "morning", "afternoon", "evening"], help="시간대 선호 추천") add.add_argument("--notes", help="메모") add.add_argument("--message-template", help="알림 메시지 포맷. 예: '[특가] {label} {observed_price} / {date_text}'") sub.add_parser("list", help="저장 규칙 목록") check = sub.add_parser("check", help="저장 규칙 점검") check.add_argument("--rule-id", help="특정 규칙만 점검") check.add_argument("--no-dedupe", action="store_true", help="중복 알림 억제를 끄고 이번 점검에서는 항상 출력") remove = sub.add_parser("remove", help="규칙 삭제") remove.add_argument("--rule-id", required=True) render = sub.add_parser("render", help="마지막 결과를 현재 템플릿으로 미리보기") render.add_argument("--rule-id", required=True) return parser def main() -> int: parser = build_parser() args = parser.parse_args() if args.command == "add": if not args.date_range and not args.departure: raise SystemExit("add 에서는 --departure 또는 --date-range 가 필요합니다.") return command_add(args) if args.command == "list": return command_list(args) if args.command == "check": return command_check(args) if args.command == "remove": return command_remove(args) if args.command == "render": return command_render(args) raise SystemExit("알 수 없는 명령입니다.") if __name__ == "__main__": raise SystemExit(main()) FILE:scripts/search_date_range.py #!/usr/bin/env python3 import argparse import json import math import sys from dataclasses import asdict, is_dataclass from datetime import timedelta from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from common_cli import ( add_section, airport_label, build_balanced_option_reasons, build_best_option_reasons, build_price_calendar, cabin_label, choose_balanced_round_trip_option, explain_recommendation, filter_and_rank_by_time_preference, format_price, format_time_or_fallback, join_nonempty, normalize_airport, parse_date_range_text, parse_flexible_date, parse_time_preference_args, pretty_date, recommendation_line, round_trip_balance_recommendation, time_preference_recommendation, ) from hybrid_observability import build_refine_diagnostics, choose_fallback_plan def build_dates(start_date, end_date): days = [] current = start_date while current <= end_date: days.append(current) current += timedelta(days=1) return days def _normalize(item): if is_dataclass(item): return asdict(item) if hasattr(item, "__dict__"): return dict(item.__dict__) return {"value": str(item)} def _empty_row(dep, ret, price=0, airline=""): return { "departure_date": dep, "return_date": ret, "price": price, "airline": airline, "departure_time": "", "return_departure_time": "", "preferred_option": None, "time_recommendation": None, "search_stage": "broad_only", "time_pref_match": None, "raw_option_count": 0, "time_pref_valid_count": 0, "broad_price": price, "diagnostic_reason": "broad_only", "diagnostic_detail": {}, } def _candidate_detail(row): detail = { "departure_date": row["departure_date"], "price": row.get("price", 0), "reason": row.get("candidate_reason", "broad_rank"), } if row.get("return_date"): detail["return_date"] = row["return_date"] return detail def _choose_refine_dates(dates, broad_rows): available = [row for row in broad_rows if row.get("price", 0) and row.get("price", 0) > 0] if not available: return [], [] index_by_date = {pretty_date(d): idx for idx, d in enumerate(dates)} chosen_indexes = set() selected = [] seen_dates = set() def add_date_label(date_label, reason): if date_label in seen_dates: return idx = index_by_date.get(date_label) if idx is None: return seen_dates.add(date_label) chosen_indexes.add(idx) selected.append({ "departure_date": date_label, "price": next((row["price"] for row in available if row["departure_date"] == date_label), 0), "candidate_reason": reason, }) available_by_price = sorted(available, key=lambda x: (x["price"], x["departure_date"])) base_count = min(len(available_by_price), max(8, min(len(dates), math.ceil(len(dates) * 0.6)))) for row in available_by_price[:base_count]: add_date_label(row["departure_date"], "cheap_broad_rank") for row in available_by_price[: min(4, len(available_by_price))]: idx = index_by_date.get(row["departure_date"]) if idx is None: continue for neighbor in (idx - 1, idx + 1): if 0 <= neighbor < len(dates): add_date_label(pretty_date(dates[neighbor]), "neighbor_of_cheap") spread_positions = sorted({0, len(dates) - 1, len(dates) // 2, len(dates) // 3, (len(dates) * 2) // 3}) for pos in spread_positions: if 0 <= pos < len(dates): add_date_label(pretty_date(dates[pos]), "coverage_anchor") for idx in sorted(chosen_indexes): row = next((item for item in selected if item["departure_date"] == pretty_date(dates[idx])), None) if row is None: selected.append({ "departure_date": pretty_date(dates[idx]), "price": next((item["price"] for item in available if item["departure_date"] == pretty_date(dates[idx])), 0), "candidate_reason": "selected", }) return [selected, [dates[idx] for idx in sorted(chosen_indexes)]] def _build_fallback_dates(dates, broad_rows, attempted_dates, limit, diagnostics=None): if limit <= 0: return [] diagnostics = diagnostics or {} attempted_labels = {pretty_date(d) for d in attempted_dates} index_by_date = {pretty_date(d): idx for idx, d in enumerate(dates)} broad_by_date = {row["departure_date"]: row for row in broad_rows} available = sorted( [row for row in broad_rows if row.get("price", 0) and row.get("price", 0) > 0 and row["departure_date"] not in attempted_labels], key=lambda x: (x["price"], x["departure_date"]), ) fallback = [] seen = set() dominant_reason = diagnostics.get("dominant_reason") def push(date_label, reason): if date_label in attempted_labels or date_label in seen or len(fallback) >= limit: return idx = index_by_date.get(date_label) if idx is None: return seen.add(date_label) fallback.append({ "date": dates[idx], "detail": { "departure_date": date_label, "price": next((row["price"] for row in broad_rows if row["departure_date"] == date_label), 0), "reason": reason, }, }) if dominant_reason == "detail_empty_after_broad_hit": empty_labels = [] for sample in diagnostics.get("samples", {}).get("detail_empty_after_broad_hit", []): label = str(sample).split(" (", 1)[0] if label in index_by_date: empty_labels.append(label) for label in empty_labels: idx = index_by_date.get(label) if idx is None: continue for neighbor in (idx - 2, idx - 1, idx + 1, idx + 2): if 0 <= neighbor < len(dates): push(pretty_date(dates[neighbor]), "fallback_neighbor_of_empty_detail") if dominant_reason in {"detail_missing_departure_times", "detail_missing_return_times"}: for row in available[: min(limit, 4)]: idx = index_by_date.get(row["departure_date"]) if idx is None: continue for neighbor in (idx - 1, idx + 1): if 0 <= neighbor < len(dates): push(pretty_date(dates[neighbor]), "fallback_neighbor_of_missing_time") for row in available: push(row["departure_date"], "fallback_broad_rank") idx = index_by_date.get(row["departure_date"]) if idx is None: continue for neighbor in (idx - 1, idx + 1): if 0 <= neighbor < len(dates): candidate_label = pretty_date(dates[neighbor]) if candidate_label in broad_by_date and (broad_by_date[candidate_label].get("price") or 0) > 0: push(candidate_label, "fallback_neighbor") if len(fallback) >= limit: break return fallback def _diagnose_refine_failure(raw_results, filtered, broad_price, has_return_time_constraint): raw_count = len(raw_results) priced_count = sum(1 for item in raw_results if int(item.get("price", 0) or 0) > 0) depart_time_count = sum(1 for item in raw_results if str(item.get("departure_time") or "").strip()) return_time_count = sum(1 for item in raw_results if str(item.get("return_departure_time") or "").strip()) detail = { "raw_option_count": raw_count, "priced_option_count": priced_count, "departure_time_count": depart_time_count, "return_time_count": return_time_count, "has_return_time_constraint": bool(has_return_time_constraint), } if raw_count > 0: detail["price_coverage_ratio"] = round(priced_count / raw_count, 3) detail["departure_time_coverage_ratio"] = round(depart_time_count / raw_count, 3) detail["return_time_coverage_ratio"] = round(return_time_count / raw_count, 3) if filtered: return "detailed_match_with_time_pref", {**detail, "hint": "시간 조건 일치"} if not raw_results: if broad_price > 0: return "detail_empty_after_broad_hit", {**detail, "hint": "빠른 스캔 대비 상세 빈결과"} return "detail_empty_no_broad_signal", {**detail, "hint": "상세 빈결과"} if priced_count <= 0: return "detail_missing_price_data", {**detail, "hint": "가격 정보 없음"} if 0 < priced_count < raw_count: return "detail_sparse_price_data", {**detail, "hint": "가격 정보 일부 누락"} if depart_time_count <= 0: return "detail_missing_departure_times", {**detail, "hint": "출발 시간 정보 부족"} if 0 < depart_time_count < raw_count: return "detail_partial_departure_times", {**detail, "hint": "출발 시간 정보 부분 누락"} if has_return_time_constraint and return_time_count <= 0: return "detail_missing_return_times", {**detail, "hint": "복귀 시간 정보 부족"} if has_return_time_constraint and 0 < return_time_count < raw_count: return "detail_partial_return_times", {**detail, "hint": "복귀 시간 정보 부분 누락"} if broad_price > 0: return "broad_candidate_time_rejected", {**detail, "hint": "시간 조건 미충족"} return "detailed_no_usable_time_filter_match", {**detail, "hint": "usable match 없음"} def _refine_single_date(searcher, origin, destination, dep, ret, args, time_pref, logs, stage, broad_price=0): results = searcher.search( origin=origin, destination=destination, departure_date=dep, return_date=ret, adults=args.adults, cabin_class=args.cabin, max_results=20, progress_callback=lambda msg, dep=dep, stage=stage: logs.append(f"[{stage} {dep}] {msg}"), background_mode=False, ) raw_results = [_normalize(item) for item in results] filtered, preferred_ranked = filter_and_rank_by_time_preference(raw_results, time_pref) cheapest = filtered[0] if filtered else None preferred = preferred_ranked[0] if preferred_ranked else None diagnostic_reason, diagnostic_detail = _diagnose_refine_failure( raw_results, filtered, broad_price, has_return_time_constraint=bool(ret and (time_pref.return_min is not None or time_pref.return_max is not None)), ) return { "departure_date": dep, "return_date": ret, "price": cheapest.get("price", 0) if cheapest else 0, "airline": cheapest.get("airline", "") if cheapest else "", "departure_time": cheapest.get("departure_time", "") if cheapest else "", "return_departure_time": cheapest.get("return_departure_time", "") if cheapest else "", "preferred_option": preferred, "time_recommendation": time_preference_recommendation(preferred, cheapest, time_pref), "search_stage": stage, "time_pref_match": bool(cheapest), "raw_option_count": len(raw_results), "priced_option_count": diagnostic_detail.get("priced_option_count", 0), "departure_time_count": diagnostic_detail.get("departure_time_count", 0), "return_time_count": diagnostic_detail.get("return_time_count", 0), "has_return_time_constraint": diagnostic_detail.get("has_return_time_constraint", False), "time_pref_valid_count": len(filtered), "broad_price": broad_price, "diagnostic_reason": diagnostic_reason, "diagnostic_detail": diagnostic_detail, } def main(): parser = argparse.ArgumentParser(description="Search Korean domestic flights across date ranges") parser.add_argument("--origin", required=True, help="예: GMP 또는 김포") parser.add_argument("--destination", required=True, help="예: CJU 또는 제주") parser.add_argument("--start-date", help="예: 2026-03-25, 내일") parser.add_argument("--end-date", help="예: 2026-03-30") parser.add_argument("--date-range", help="예: 내일부터 3일, 이번주말, 2026-03-25~2026-03-30") parser.add_argument("--return-offset", type=int, default=0) parser.add_argument("--adults", type=int, default=1) parser.add_argument("--cabin", default="ECONOMY", choices=["ECONOMY", "BUSINESS", "FIRST"]) parser.add_argument("--time-pref") parser.add_argument("--depart-after") parser.add_argument("--return-after") parser.add_argument("--exclude-early-before") parser.add_argument("--prefer", choices=["late", "morning", "afternoon", "evening"]) parser.add_argument("--human", action="store_true") args = parser.parse_args() try: origin = normalize_airport(args.origin) destination = normalize_airport(args.destination) if args.date_range: start_dt, end_dt = parse_date_range_text(args.date_range) elif args.start_date and args.end_date: start_dt = parse_flexible_date(args.start_date) end_dt = parse_flexible_date(args.end_date) else: raise ValueError("start/end-date 또는 --date-range 중 하나를 제공해야 합니다.") except ValueError as exc: print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False, indent=2)) sys.exit(1) if end_dt < start_dt: print(json.dumps({"status": "error", "message": "end-date must be after or equal to start-date"}, ensure_ascii=False, indent=2)) sys.exit(1) dates = build_dates(start_dt, end_dt) if len(dates) > 30: print(json.dumps({"status": "error", "message": "date range must be 30 days or less"}, ensure_ascii=False, indent=2)) sys.exit(1) workspace = Path(__file__).resolve().parents[3] repo_path = workspace / "tmp" / "Scraping-flight-information" if not repo_path.exists(): print(json.dumps({"status": "error", "message": "Source repository clone not found.", "expected": str(repo_path)}, ensure_ascii=False, indent=2)) sys.exit(1) sys.path.insert(0, str(repo_path)) time_pref = parse_time_preference_args(args) logs = [] normalized = [] search_metadata = { "strategy": "parallel", "time_preference_active": time_pref.active(), "time_preference_summary": time_pref.describe(), "broad_scan_dates": len(dates), "refined_dates": 0, "refined_candidates": [], "broad_scan_available": 0, "fallback_refined_dates": 0, "fallback_candidates": [], "refine_attempted_dates": 0, "refine_success_dates": 0, "fallback_triggered": False, "fallback_reason": None, "fallback_trigger_reasons": [], "fallback_reason_codes": [], "fallback_decision": None, "refine_diagnostics": None, "diagnostic_hint": None, "developer_diagnostic_hint": None, } if time_pref.active(): try: from scraping.parallel import ParallelSearcher from scraping.searcher import FlightSearcher except Exception as exc: print(json.dumps({"status": "error", "message": "Failed to import hybrid search components.", "details": str(exc)}, ensure_ascii=False, indent=2)) sys.exit(1) search_metadata["strategy"] = "hybrid_parallel_then_detailed" parallel_searcher = ParallelSearcher() logs.append("hybrid mode: broad parallel scan started") broad_raw = parallel_searcher.search_date_range( origin=origin, destination=destination, dates=[d.strftime("%Y%m%d") for d in dates], return_offset=args.return_offset, adults=args.adults, cabin_class=args.cabin, progress_callback=lambda msg: logs.append(f"[broad] {msg}"), ) broad_rows = [] for d in dates: dep = pretty_date(d) ret = pretty_date(d + timedelta(days=args.return_offset)) if args.return_offset > 0 else None price, airline = broad_raw.get(d.strftime("%Y%m%d"), (0, "N/A")) broad_rows.append(_empty_row(dep, ret, price=price, airline=airline)) refine_candidate_details, refine_dates = _choose_refine_dates(dates, broad_rows) search_metadata["broad_scan_available"] = len([row for row in broad_rows if row["price"] and row["price"] > 0]) search_metadata["refined_dates"] = len(refine_dates) search_metadata["refined_candidates"] = refine_candidate_details logs.append( f"hybrid mode: broad scan complete, available={search_metadata['broad_scan_available']}, refining={search_metadata['refined_dates']}" ) detailed_by_date = {} attempted_dates = list(refine_dates) searcher = FlightSearcher() try: broad_price_by_dep = {row["departure_date"]: row.get("price", 0) for row in broad_rows} for d in refine_dates: dep = pretty_date(d) ret = pretty_date(d + timedelta(days=args.return_offset)) if args.return_offset > 0 else None detailed_by_date[dep] = _refine_single_date( searcher, origin, destination, dep, ret, args, time_pref, logs, "refine", broad_price=broad_price_by_dep.get(dep, 0), ) successful_details = [row for row in detailed_by_date.values() if row.get("time_pref_match")] search_metadata["refine_attempted_dates"] = len(detailed_by_date) search_metadata["refine_success_dates"] = len(successful_details) diagnostics = build_refine_diagnostics( broad_rows, detailed_by_date.values(), key_fn=lambda row: row["departure_date"], label_fn=lambda row: row["departure_date"], ) search_metadata["refine_diagnostics"] = diagnostics search_metadata["diagnostic_hint"] = diagnostics.get("human_hint") search_metadata["developer_diagnostic_hint"] = diagnostics.get("developer_hint") logs.append(f"hybrid refine diagnostics: {diagnostics['summary_text']}") fallback_plan = choose_fallback_plan( diagnostics, minimum_target=max(2, math.ceil(len(dates) * 0.35)), hard_cap=min(6, len(dates)), pad=2, ) search_metadata["fallback_decision"] = fallback_plan search_metadata["fallback_reason_codes"] = list(fallback_plan.get("reasons", [])) if fallback_plan["triggered"]: fallback_dates = _build_fallback_dates(dates, broad_rows, attempted_dates, fallback_plan["limit"], diagnostics) if fallback_dates: search_metadata["fallback_triggered"] = True search_metadata["fallback_trigger_reasons"] = fallback_plan["reasons"] search_metadata["fallback_reason"] = ( f"후보 확장: {diagnostics['summary_text']}" ) search_metadata["fallback_refined_dates"] = len(fallback_dates) search_metadata["fallback_candidates"] = [item["detail"] for item in fallback_dates] logs.append( f"hybrid mode: fallback refine triggered, reason={fallback_plan['primary_reason']}, extra={len(fallback_dates)}" ) for item in fallback_dates: dep = pretty_date(item["date"]) ret = pretty_date(item["date"] + timedelta(days=args.return_offset)) if args.return_offset > 0 else None detailed_by_date[dep] = _refine_single_date( searcher, origin, destination, dep, ret, args, time_pref, logs, "fallback", broad_price=broad_price_by_dep.get(dep, 0), ) search_metadata["refine_attempted_dates"] = len(detailed_by_date) search_metadata["refine_success_dates"] = len([row for row in detailed_by_date.values() if row.get("time_pref_match")]) diagnostics = build_refine_diagnostics( broad_rows, detailed_by_date.values(), key_fn=lambda row: row["departure_date"], label_fn=lambda row: row["departure_date"], ) search_metadata["refine_diagnostics"] = diagnostics search_metadata["diagnostic_hint"] = diagnostics.get("human_hint") search_metadata["developer_diagnostic_hint"] = diagnostics.get("developer_hint") logs.append(f"hybrid refine diagnostics (after fallback): {diagnostics['summary_text']}") finally: try: searcher.close() except Exception: pass for broad in broad_rows: normalized.append(detailed_by_date.get(broad["departure_date"], broad)) else: try: from scraping.parallel import ParallelSearcher except Exception as exc: print(json.dumps({"status": "error", "message": "Failed to import parallel searcher.", "details": str(exc)}, ensure_ascii=False, indent=2)) sys.exit(1) searcher = ParallelSearcher() raw = searcher.search_date_range( origin=origin, destination=destination, dates=[d.strftime("%Y%m%d") for d in dates], return_offset=args.return_offset, adults=args.adults, cabin_class=args.cabin, progress_callback=lambda msg: logs.append(str(msg)), ) for d in dates: key = d.strftime("%Y%m%d") price, airline = raw.get(key, (0, "N/A")) normalized.append({ "departure_date": pretty_date(d), "return_date": pretty_date(d + timedelta(days=args.return_offset)) if args.return_offset > 0 else None, "price": price, "airline": airline, "departure_time": "", "return_departure_time": "", "preferred_option": None, "time_recommendation": None, "search_stage": "parallel", "time_pref_match": None, "raw_option_count": 0, "time_pref_valid_count": 0, "diagnostic_detail": {}, }) search_metadata["broad_scan_available"] = len([item for item in normalized if item["price"] and item["price"] > 0]) available = [item for item in normalized if item["price"] and item["price"] > 0] available.sort(key=lambda x: x["price"]) cheapest = available[0] if available else None second_price = available[1]["price"] if len(available) > 1 else None price_calendar = build_price_calendar(normalized, date_key="departure_date", price_key="price") balanced_option = choose_balanced_round_trip_option(available, time_pref) if args.return_offset > 0 else None summary = { "headline": ( f"{airport_label(origin)} → {airport_label(destination)} 날짜범위 최저가 {format_price(cheapest['price'])}" if cheapest else f"{airport_label(origin)} → {airport_label(destination)} 날짜범위 검색 결과가 없습니다." ), "range": f"{pretty_date(start_dt)} ~ {pretty_date(end_dt)}", "trip_type": "왕복 범위검색" if args.return_offset > 0 else "편도 범위검색", "search_strategy": search_metadata["strategy"], "search_metadata": search_metadata, "best_date": cheapest, "top_dates": available[:5], "price_calendar": price_calendar, "balanced_round_trip": balanced_option, "diagnostic_hint": search_metadata.get("diagnostic_hint"), "recommendation": recommendation_line( f"{cheapest['departure_date']}{f'~{cheapest['return_date']}' if cheapest and cheapest['return_date'] else ''}", cheapest["price"], second_price, ) if cheapest else None, "recommendation_explained": explain_recommendation( f"{cheapest['departure_date']}{f'~{cheapest['return_date']}' if cheapest and cheapest['return_date'] else ''}", int(cheapest["price"] or 0), second_price, build_best_option_reasons(cheapest, second_price, time_pref), ) if cheapest else None, "time_recommendation": next((item.get("time_recommendation") for item in available if item.get("time_recommendation")), None), "balanced_recommendation": round_trip_balance_recommendation(balanced_option, cheapest, time_pref) if balanced_option else None, "balanced_recommendation_explained": explain_recommendation( f"{balanced_option['departure_date']}~{balanced_option['return_date']}", int(balanced_option["price"] or 0), int(cheapest["price"] or 0) if cheapest else None, build_balanced_option_reasons(balanced_option, cheapest, time_pref), ) if balanced_option else None, } if args.human: def date_row_text(item): date_text = f"{item['departure_date']} ~ {item['return_date']}" if item.get('return_date') else item['departure_date'] time_bits = [] if item.get('return_date'): if item.get('departure_time'): time_bits.append(f"가는편 {format_time_or_fallback(item.get('departure_time'))}") if item.get('return_departure_time'): time_bits.append(f"오는편 {format_time_or_fallback(item.get('return_departure_time'))}") elif item.get('departure_time'): time_bits.append(format_time_or_fallback(item.get('departure_time'))) return join_nonempty([ date_text, format_price(item['price']), item.get('airline') or None, join_nonempty(time_bits), ]) lines = [summary["headline"]] lines.append(f"범위: {summary['range']}") lines.append(f"조건: 성인 {args.adults}명 · {cabin_label(args.cabin)}") if args.return_offset > 0: lines.append(f"왕복 기준: 출발일 + {args.return_offset}일 귀국") if time_pref.describe(): lines.append(f"시간 조건: {time_pref.describe()}") if search_metadata["strategy"].startswith("hybrid"): hybrid_text = f"검색 방식: 하이브리드(빠른 전체 스캔 {search_metadata['broad_scan_dates']}일 → 상세 재검증 {search_metadata['refined_dates']}일" if search_metadata.get("fallback_refined_dates"): hybrid_text += f" + fallback {search_metadata['fallback_refined_dates']}일" hybrid_text += ")" lines.append(hybrid_text) if search_metadata.get("diagnostic_hint"): lines.append(f"참고: {search_metadata['diagnostic_hint']}") add_section(lines, "최저가", [ f"최저가 날짜: {date_row_text(cheapest)}" if cheapest else None, summary.get("recommendation"), summary.get("recommendation_explained"), ]) add_section(lines, "시간대 추천", [summary.get("time_recommendation")]) add_section(lines, "왕복 균형 추천", [summary.get("balanced_recommendation"), summary.get("balanced_recommendation_explained")]) add_section(lines, "상위 날짜", [f"{idx}. {date_row_text(item)}" for idx, item in enumerate(summary.get("top_dates", []), start=1)]) if summary["price_calendar"]: calendar_rows = summary["price_calendar"] preview = calendar_rows[:10] add_section(lines, "가격 캘린더", [*(item["label"] for item in preview), f"… 외 {len(calendar_rows) - len(preview)}일" if len(calendar_rows) > len(preview) else None]) print("\n".join(lines)) return print(json.dumps({ "status": "success", "query": { "origin": origin, "destination": destination, "start_date": pretty_date(start_dt), "end_date": pretty_date(end_dt), "return_offset": args.return_offset, "adults": args.adults, "cabin": args.cabin, "time_preference": time_pref.describe(), }, "summary": summary, "results": normalized, "logs": logs, "search_metadata": search_metadata, }, ensure_ascii=False, indent=2)) if __name__ == "__main__": main() FILE:scripts/search_destination_date_matrix.py #!/usr/bin/env python3 import argparse import json import math import sys from dataclasses import asdict, is_dataclass from datetime import timedelta from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from common_cli import ( add_section, airport_label, build_balanced_option_reasons, build_best_option_reasons, build_price_calendar, cabin_label, choose_balanced_round_trip_option, explain_recommendation, filter_and_rank_by_time_preference, format_price, format_time_or_fallback, join_nonempty, normalize_airport, parse_date_range_text, parse_flexible_date, parse_time_preference_args, pretty_date, recommendation_line, round_trip_balance_recommendation, unique_codes, ) from hybrid_observability import build_refine_diagnostics, choose_fallback_plan def build_dates(start_date, end_date): days = [] current = start_date while current <= end_date: days.append(current) current += timedelta(days=1) return days def _normalize(item): if is_dataclass(item): return asdict(item) if hasattr(item, "__dict__"): return dict(item.__dict__) return {"value": str(item)} def _broad_row(destination, dep, ret, price=0, airline=""): return { "destination": destination, "destination_label": airport_label(destination), "departure_date": dep, "return_date": ret, "price": price, "airline": airline, "departure_time": "", "return_departure_time": "", "preferred_option": None, "search_stage": "broad_only", "time_pref_match": None, "raw_option_count": 0, "time_pref_valid_count": 0, "broad_price": price, "diagnostic_reason": "broad_only", "diagnostic_detail": {}, } def _candidate_detail(row): detail = { "destination": row["destination"], "departure_date": row["departure_date"], "price": row.get("price", 0), "reason": row.get("candidate_reason", "broad_rank"), } if row.get("return_date"): detail["return_date"] = row["return_date"] return detail def _choose_refine_combos(destinations, dates, broad_rows): available = [row for row in broad_rows if row.get("price", 0) and row.get("price", 0) > 0] if not available: return [] rows_by_destination = { destination: sorted( [row for row in available if row["destination"] == destination], key=lambda x: (x["price"], x["departure_date"]), ) for destination in destinations } date_index = {pretty_date(d): idx for idx, d in enumerate(dates)} chosen = [] seen = set() def add_row(row, reason): key = (row["destination"], row["departure_date"]) if key in seen: return seen.add(key) enriched = dict(row) enriched["candidate_reason"] = reason chosen.append(enriched) per_destination_budget = max(3, min(6, math.ceil(len(dates) * 0.5))) for destination in destinations: rows = rows_by_destination.get(destination, []) for row in rows[:per_destination_budget]: add_row(row, "cheap_within_destination") for row in rows[: min(3, len(rows))]: idx = date_index.get(row["departure_date"]) if idx is None: continue for neighbor in (idx - 1, idx + 1): if 0 <= neighbor < len(dates): neighbor_label = pretty_date(dates[neighbor]) neighbor_row = next((item for item in rows if item["departure_date"] == neighbor_label), None) if neighbor_row: add_row(neighbor_row, "neighbor_of_destination_candidate") globally_sorted = sorted(available, key=lambda x: (x["price"], x["destination"], x["departure_date"])) global_budget = min(len(globally_sorted), max(len(destinations) * 4, math.ceil(len(available) * 0.35))) for row in globally_sorted[:global_budget]: add_row(row, "cheap_global_rank") anchor_positions = sorted({0, len(dates) - 1, len(dates) // 2}) for destination in destinations: rows = {row["departure_date"]: row for row in rows_by_destination.get(destination, [])} for pos in anchor_positions: if 0 <= pos < len(dates): anchor_label = pretty_date(dates[pos]) if anchor_label in rows: add_row(rows[anchor_label], "coverage_anchor") return chosen def _build_fallback_combos(destinations, dates, broad_rows, attempted_keys, limit, diagnostics=None): if limit <= 0: return [] diagnostics = diagnostics or {} date_index = {pretty_date(d): idx for idx, d in enumerate(dates)} broad_index = {(row["destination"], row["departure_date"]): row for row in broad_rows} available = sorted( [row for row in broad_rows if row.get("price", 0) and row.get("price", 0) > 0 and (row["destination"], row["departure_date"]) not in attempted_keys], key=lambda x: (x["price"], x["destination"], x["departure_date"]), ) fallback = [] seen = set() dominant_reason = diagnostics.get("dominant_reason") def push(row, reason): key = (row["destination"], row["departure_date"]) if key in attempted_keys or key in seen or len(fallback) >= limit: return seen.add(key) enriched = dict(row) enriched["candidate_reason"] = reason fallback.append(enriched) if dominant_reason == "detail_empty_after_broad_hit": for sample in diagnostics.get("samples", {}).get("detail_empty_after_broad_hit", []): label = str(sample).split(" (", 1)[0] parts = label.split() if len(parts) != 2: continue destination, dep = parts idx = date_index.get(dep) if idx is None: continue for neighbor in (idx - 1, idx + 1): if 0 <= neighbor < len(dates): key = (destination, pretty_date(dates[neighbor])) row = broad_index.get(key) if row and (row.get("price") or 0) > 0: push(row, "fallback_neighbor_of_empty_detail") for destination in destinations: destination_rows = [row for row in available if row["destination"] == destination] if destination_rows: preferred_reason = "fallback_best_per_destination" if dominant_reason in {"detail_missing_departure_times", "detail_missing_return_times"}: preferred_reason = "fallback_best_per_destination_after_missing_time" push(destination_rows[0], preferred_reason) for row in available: push(row, "fallback_global_rank") idx = date_index.get(row["departure_date"]) if idx is not None: for neighbor in (idx - 1, idx + 1): if 0 <= neighbor < len(dates): neighbor_key = (row["destination"], pretty_date(dates[neighbor])) neighbor_row = broad_index.get(neighbor_key) if neighbor_row and (neighbor_row.get("price") or 0) > 0: push(neighbor_row, "fallback_neighbor") if len(fallback) >= limit: break return fallback def _diagnose_refine_failure(raw_results, filtered, broad_price, has_return_time_constraint): raw_count = len(raw_results) priced_count = sum(1 for item in raw_results if int(item.get("price", 0) or 0) > 0) depart_time_count = sum(1 for item in raw_results if str(item.get("departure_time") or "").strip()) return_time_count = sum(1 for item in raw_results if str(item.get("return_departure_time") or "").strip()) detail = { "raw_option_count": raw_count, "priced_option_count": priced_count, "departure_time_count": depart_time_count, "return_time_count": return_time_count, "has_return_time_constraint": bool(has_return_time_constraint), } if raw_count > 0: detail["price_coverage_ratio"] = round(priced_count / raw_count, 3) detail["departure_time_coverage_ratio"] = round(depart_time_count / raw_count, 3) detail["return_time_coverage_ratio"] = round(return_time_count / raw_count, 3) if filtered: return "detailed_match_with_time_pref", {**detail, "hint": "시간 조건 일치"} if not raw_results: if broad_price > 0: return "detail_empty_after_broad_hit", {**detail, "hint": "빠른 스캔 대비 상세 빈결과"} return "detail_empty_no_broad_signal", {**detail, "hint": "상세 빈결과"} if priced_count <= 0: return "detail_missing_price_data", {**detail, "hint": "가격 정보 없음"} if 0 < priced_count < raw_count: return "detail_sparse_price_data", {**detail, "hint": "가격 정보 일부 누락"} if depart_time_count <= 0: return "detail_missing_departure_times", {**detail, "hint": "출발 시간 정보 부족"} if 0 < depart_time_count < raw_count: return "detail_partial_departure_times", {**detail, "hint": "출발 시간 정보 부분 누락"} if has_return_time_constraint and return_time_count <= 0: return "detail_missing_return_times", {**detail, "hint": "복귀 시간 정보 부족"} if has_return_time_constraint and 0 < return_time_count < raw_count: return "detail_partial_return_times", {**detail, "hint": "복귀 시간 정보 부분 누락"} if broad_price > 0: return "broad_candidate_time_rejected", {**detail, "hint": "시간 조건 미충족"} return "detailed_no_usable_time_filter_match", {**detail, "hint": "usable match 없음"} def _refine_combo(searcher, origin, row, args, time_pref, logs, stage, broad_price=0): destination = row["destination"] dep = row["departure_date"] ret = row["return_date"] results = searcher.search( origin=origin, destination=destination, departure_date=dep, return_date=ret, adults=args.adults, cabin_class=args.cabin, max_results=20, progress_callback=lambda msg, dest=destination, dep=dep, stage=stage: logs.append(f"[{stage} {dest} {dep}] {msg}"), background_mode=False, ) raw_results = [_normalize(item) for item in results] filtered, preferred_ranked = filter_and_rank_by_time_preference(raw_results, time_pref) cheapest = filtered[0] if filtered else None diagnostic_reason, diagnostic_detail = _diagnose_refine_failure( raw_results, filtered, broad_price, has_return_time_constraint=bool(ret and (time_pref.return_min is not None or time_pref.return_max is not None)), ) return { "destination": destination, "destination_label": airport_label(destination), "departure_date": dep, "return_date": ret, "price": cheapest.get("price", 0) if cheapest else 0, "airline": cheapest.get("airline", "") if cheapest else "", "departure_time": cheapest.get("departure_time", "") if cheapest else "", "return_departure_time": cheapest.get("return_departure_time", "") if cheapest else "", "preferred_option": preferred_ranked[0] if preferred_ranked else None, "search_stage": stage, "time_pref_match": bool(cheapest), "raw_option_count": len(raw_results), "priced_option_count": diagnostic_detail.get("priced_option_count", 0), "departure_time_count": diagnostic_detail.get("departure_time_count", 0), "return_time_count": diagnostic_detail.get("return_time_count", 0), "has_return_time_constraint": diagnostic_detail.get("has_return_time_constraint", False), "time_pref_valid_count": len(filtered), "broad_price": broad_price, "diagnostic_reason": diagnostic_reason, "diagnostic_detail": diagnostic_detail, } def main(): parser = argparse.ArgumentParser(description="Search combined destination/date ranges for Korean domestic flights") parser.add_argument("--origin", required=True) parser.add_argument("--destinations", required=True, help="쉼표 구분 목적지 목록") parser.add_argument("--start-date") parser.add_argument("--end-date") parser.add_argument("--date-range", help="예: 내일부터 5일, 2026-03-25~2026-03-30") parser.add_argument("--return-offset", type=int, default=0) parser.add_argument("--adults", type=int, default=1) parser.add_argument("--cabin", default="ECONOMY", choices=["ECONOMY", "BUSINESS", "FIRST"]) parser.add_argument("--time-pref") parser.add_argument("--depart-after") parser.add_argument("--return-after") parser.add_argument("--exclude-early-before") parser.add_argument("--prefer", choices=["late", "morning", "afternoon", "evening"]) parser.add_argument("--human", action="store_true") args = parser.parse_args() try: origin = normalize_airport(args.origin) destinations = unique_codes([normalize_airport(x.strip()) for x in args.destinations.split(",") if x.strip()]) if args.date_range: start_dt, end_dt = parse_date_range_text(args.date_range) elif args.start_date and args.end_date: start_dt = parse_flexible_date(args.start_date) end_dt = parse_flexible_date(args.end_date) else: raise ValueError("start/end-date 또는 --date-range 중 하나를 제공해야 합니다.") except ValueError as exc: print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False, indent=2)) sys.exit(1) if end_dt < start_dt: print(json.dumps({"status": "error", "message": "end-date must be after or equal to start-date"}, ensure_ascii=False, indent=2)) sys.exit(1) dates = build_dates(start_dt, end_dt) if len(dates) * len(destinations) > 90: print(json.dumps({"status": "error", "message": "검색 조합 수는 90개 이하로 제한됩니다."}, ensure_ascii=False, indent=2)) sys.exit(1) workspace = Path(__file__).resolve().parents[3] repo_path = workspace / "tmp" / "Scraping-flight-information" if not repo_path.exists(): print(json.dumps({"status": "error", "message": "Source repository clone not found.", "expected": str(repo_path)}, ensure_ascii=False, indent=2)) sys.exit(1) sys.path.insert(0, str(repo_path)) time_pref = parse_time_preference_args(args) logs = [] destination_rows = [] combos = [] search_metadata = { "strategy": "parallel", "time_preference_active": time_pref.active(), "time_preference_summary": time_pref.describe(), "broad_scan_destinations": len(destinations), "broad_scan_dates": len(dates), "broad_scan_combos": len(destinations) * len(dates), "broad_scan_available": 0, "refined_combos": 0, "refined_candidates": [], "fallback_refined_combos": 0, "fallback_candidates": [], "refine_attempted_combos": 0, "refine_success_combos": 0, "fallback_triggered": False, "fallback_reason": None, "fallback_trigger_reasons": [], "fallback_reason_codes": [], "fallback_decision": None, "refine_diagnostics": None, "diagnostic_hint": None, "developer_diagnostic_hint": None, } if time_pref.active(): try: from scraping.parallel import ParallelSearcher from scraping.searcher import FlightSearcher except Exception as exc: print(json.dumps({"status": "error", "message": "Failed to import hybrid search components.", "details": str(exc)}, ensure_ascii=False, indent=2)) sys.exit(1) search_metadata["strategy"] = "hybrid_parallel_then_detailed" broad_rows = [] parallel_searcher = ParallelSearcher() for destination in destinations: logs.append(f"[broad] matrix search start: {destination}") raw = parallel_searcher.search_date_range( origin=origin, destination=destination, dates=[d.strftime("%Y%m%d") for d in dates], return_offset=args.return_offset, adults=args.adults, cabin_class=args.cabin, progress_callback=lambda msg, dest=destination: logs.append(f"[broad {dest}] {msg}"), ) for d in dates: key = d.strftime("%Y%m%d") price, airline = raw.get(key, (0, "N/A")) broad_rows.append(_broad_row( destination=destination, dep=pretty_date(d), ret=pretty_date(d + timedelta(days=args.return_offset)) if args.return_offset > 0 else None, price=price, airline=airline, )) search_metadata["broad_scan_available"] = len([row for row in broad_rows if row["price"] and row["price"] > 0]) refine_rows = _choose_refine_combos(destinations, dates, broad_rows) search_metadata["refined_combos"] = len(refine_rows) search_metadata["refined_candidates"] = [_candidate_detail(row) for row in refine_rows] logs.append( f"hybrid mode: broad scan complete, available={search_metadata['broad_scan_available']}, refining={search_metadata['refined_combos']}" ) detailed_map = {} attempted_keys = {(row["destination"], row["departure_date"]) for row in refine_rows} searcher = FlightSearcher() try: broad_price_by_key = {(row["destination"], row["departure_date"]): row.get("price", 0) for row in broad_rows} for row in refine_rows: key = (row["destination"], row["departure_date"]) detailed_map[key] = _refine_combo(searcher, origin, row, args, time_pref, logs, "refine", broad_price=broad_price_by_key.get(key, 0)) successful = [row for row in detailed_map.values() if row.get("time_pref_match")] search_metadata["refine_attempted_combos"] = len(detailed_map) search_metadata["refine_success_combos"] = len(successful) diagnostics = build_refine_diagnostics( broad_rows, detailed_map.values(), key_fn=lambda row: (row["destination"], row["departure_date"]), label_fn=lambda row: f"{row['destination']} {row['departure_date']}", ) search_metadata["refine_diagnostics"] = diagnostics search_metadata["diagnostic_hint"] = diagnostics.get("human_hint") search_metadata["developer_diagnostic_hint"] = diagnostics.get("developer_hint") logs.append(f"hybrid refine diagnostics: {diagnostics['summary_text']}") fallback_plan = choose_fallback_plan( diagnostics, minimum_target=max(len(destinations), math.ceil(search_metadata["broad_scan_combos"] * 0.18)), hard_cap=min(len(destinations) * 3, max(6, len(destinations) + len(dates))), pad=len(destinations), ) search_metadata["fallback_decision"] = fallback_plan search_metadata["fallback_reason_codes"] = list(fallback_plan.get("reasons", [])) if fallback_plan["triggered"]: fallback_rows = _build_fallback_combos(destinations, dates, broad_rows, attempted_keys, fallback_plan["limit"], diagnostics) if fallback_rows: search_metadata["fallback_triggered"] = True search_metadata["fallback_trigger_reasons"] = fallback_plan["reasons"] search_metadata["fallback_reason"] = ( f"후보 확장: {diagnostics['summary_text']}" ) search_metadata["fallback_refined_combos"] = len(fallback_rows) search_metadata["fallback_candidates"] = [_candidate_detail(row) for row in fallback_rows] logs.append( f"hybrid mode: fallback refine triggered, reason={fallback_plan['primary_reason']}, extra={len(fallback_rows)}" ) for row in fallback_rows: key = (row["destination"], row["departure_date"]) detailed_map[key] = _refine_combo(searcher, origin, row, args, time_pref, logs, "fallback", broad_price=broad_price_by_key.get(key, 0)) search_metadata["refine_attempted_combos"] = len(detailed_map) search_metadata["refine_success_combos"] = len([row for row in detailed_map.values() if row.get("time_pref_match")]) diagnostics = build_refine_diagnostics( broad_rows, detailed_map.values(), key_fn=lambda row: (row["destination"], row["departure_date"]), label_fn=lambda row: f"{row['destination']} {row['departure_date']}", ) search_metadata["refine_diagnostics"] = diagnostics search_metadata["diagnostic_hint"] = diagnostics.get("human_hint") search_metadata["developer_diagnostic_hint"] = diagnostics.get("developer_hint") logs.append(f"hybrid refine diagnostics (after fallback): {diagnostics['summary_text']}") finally: try: searcher.close() except Exception: pass all_rows = [] for row in broad_rows: key = (row["destination"], row["departure_date"]) all_rows.append(detailed_map.get(key, row)) for destination in destinations: rows = [row for row in all_rows if row["destination"] == destination] combos.extend(rows) available_rows = [row for row in rows if row["price"] and row["price"] > 0] available_rows.sort(key=lambda x: x["price"]) destination_rows.append({ "destination": destination, "destination_label": airport_label(destination), "search_strategy": search_metadata["strategy"], "best_option": available_rows[0] if available_rows else None, "top_dates": available_rows[:3], "price_calendar": build_price_calendar(rows, date_key="departure_date", price_key="price"), }) else: try: from scraping.parallel import ParallelSearcher except Exception as exc: print(json.dumps({"status": "error", "message": "Failed to import parallel searcher.", "details": str(exc)}, ensure_ascii=False, indent=2)) sys.exit(1) searcher = ParallelSearcher() for destination in destinations: logs.append(f"matrix search start: {destination}") raw = searcher.search_date_range( origin=origin, destination=destination, dates=[d.strftime("%Y%m%d") for d in dates], return_offset=args.return_offset, adults=args.adults, cabin_class=args.cabin, progress_callback=lambda msg, dest=destination: logs.append(f"[{dest}] {msg}"), ) rows = [] for d in dates: key = d.strftime("%Y%m%d") price, airline = raw.get(key, (0, "N/A")) row = { "destination": destination, "destination_label": airport_label(destination), "departure_date": pretty_date(d), "return_date": pretty_date(d + timedelta(days=args.return_offset)) if args.return_offset > 0 else None, "price": price, "airline": airline, "departure_time": "", "return_departure_time": "", "preferred_option": None, "search_stage": "parallel", "time_pref_match": None, "raw_option_count": 0, "time_pref_valid_count": 0, "diagnostic_detail": {}, } rows.append(row) combos.append(row) available_rows = [row for row in rows if row["price"] and row["price"] > 0] available_rows.sort(key=lambda x: x["price"]) destination_rows.append({ "destination": destination, "destination_label": airport_label(destination), "search_strategy": search_metadata["strategy"], "best_option": available_rows[0] if available_rows else None, "top_dates": available_rows[:3], "price_calendar": build_price_calendar(rows, date_key="departure_date", price_key="price"), }) search_metadata["broad_scan_available"] = len([row for row in combos if row["price"] and row["price"] > 0]) ranked_combos = sorted([row for row in combos if row["price"] and row["price"] > 0], key=lambda x: x["price"]) best = ranked_combos[0] if ranked_combos else None balanced_combo = choose_balanced_round_trip_option(ranked_combos, time_pref) if args.return_offset > 0 else None second_price = ranked_combos[1]["price"] if len(ranked_combos) > 1 else None destination_rows.sort(key=lambda x: x["best_option"]["price"] if x["best_option"] else 10**12) summary = { "headline": ( f"{airport_label(origin)} 출발 최적 조합은 {best['destination_label']} {best['departure_date']} {format_price(best['price'])}" if best else f"{airport_label(origin)} 출발 목적지+날짜 범위 검색 결과가 없습니다." ), "range": f"{pretty_date(start_dt)} ~ {pretty_date(end_dt)}", "search_strategy": search_metadata["strategy"], "search_metadata": search_metadata, "best_combo": best, "balanced_round_trip": balanced_combo, "top_combos": ranked_combos[:7], "by_destination": destination_rows, "diagnostic_hint": search_metadata.get("diagnostic_hint"), "recommendation": recommendation_line( f"{best['destination_label']} / {best['departure_date']}{f'~{best['return_date']}' if best and best['return_date'] else ''}", best["price"], second_price, ) if best else None, "recommendation_explained": explain_recommendation( f"{best['destination_label']} / {best['departure_date']}{f'~{best['return_date']}' if best and best['return_date'] else ''}", int(best["price"] or 0), second_price, build_best_option_reasons(best, second_price, time_pref), ) if best else None, "balanced_recommendation": round_trip_balance_recommendation(balanced_combo, best, time_pref) if balanced_combo else None, "balanced_recommendation_explained": explain_recommendation( f"{balanced_combo['destination_label']} / {balanced_combo['departure_date']}{f'~{balanced_combo['return_date']}' if balanced_combo and balanced_combo['return_date'] else ''}", int(balanced_combo["price"] or 0), int(best["price"] or 0) if best else None, build_balanced_option_reasons(balanced_combo, best, time_pref), ) if balanced_combo else None, } if args.human: def combo_text(item): date_text = f"{item['departure_date']} ~ {item['return_date']}" if item.get('return_date') else item['departure_date'] time_bits = [] if item.get('departure_time'): time_bits.append(f"가는편 {format_time_or_fallback(item.get('departure_time'))}") if item.get('return_departure_time'): time_bits.append(f"오는편 {format_time_or_fallback(item.get('return_departure_time'))}") return join_nonempty([ item.get('destination_label') or None, date_text, format_price(item['price']), item.get('airline') or None, join_nonempty(time_bits), ]) lines = [summary["headline"]] lines.append(f"범위: {summary['range']}") lines.append(f"조건: 출발 {airport_label(origin)} · 목적지 {len(destinations)}곳 · 성인 {args.adults}명 · {cabin_label(args.cabin)}") if args.return_offset > 0: lines.append(f"왕복 기준: 출발일 + {args.return_offset}일 귀국") if time_pref.describe(): lines.append(f"시간 조건: {time_pref.describe()}") if search_metadata["strategy"].startswith("hybrid"): hybrid_text = f"검색 방식: 하이브리드(빠른 전체 스캔 {search_metadata['broad_scan_combos']}조합 → 상세 재검증 {search_metadata['refined_combos']}조합" if search_metadata.get("fallback_refined_combos"): hybrid_text += f" + fallback {search_metadata['fallback_refined_combos']}조합" hybrid_text += ")" lines.append(hybrid_text) if search_metadata.get("diagnostic_hint"): lines.append(f"참고: {search_metadata['diagnostic_hint']}") add_section(lines, "최저가", [ f"최적 조합: {combo_text(best)}" if best else None, summary.get("recommendation"), summary.get("recommendation_explained"), ]) add_section(lines, "왕복 균형 추천", [summary.get("balanced_recommendation"), summary.get("balanced_recommendation_explained")]) add_section(lines, "목적지별 베스트", [ (f"{idx}. {item['destination_label']} · {combo_text(best_option).replace(item['destination_label'] + ' · ', '', 1)}" if best_option else f"{idx}. {item['destination_label']} · 결과 없음") for idx, item in enumerate(destination_rows[:5], start=1) for best_option in [item.get('best_option')] ]) if destination_rows: calendar_lines = [] for item in destination_rows[:5]: calendar_lines.append(f"- {item['destination_label']}") calendar_rows = item.get("price_calendar", []) preview = calendar_rows[:7] calendar_lines.extend(f" {entry['label']}" for entry in preview) if len(calendar_rows) > len(preview): calendar_lines.append(f" … 외 {len(calendar_rows) - len(preview)}일") add_section(lines, "목적지별 가격 캘린더", calendar_lines) add_section(lines, "전체 상위 조합", [f"{idx}. {combo_text(item)}" for idx, item in enumerate(ranked_combos[:7], start=1)]) print("\n".join(lines)) return print(json.dumps({ "status": "success", "query": { "origin": origin, "destinations": destinations, "start_date": pretty_date(start_dt), "end_date": pretty_date(end_dt), "return_offset": args.return_offset, "adults": args.adults, "cabin": args.cabin, "time_preference": time_pref.describe(), }, "summary": summary, "results": ranked_combos, "matrix": destination_rows, "logs": logs, "search_metadata": search_metadata, }, ensure_ascii=False, indent=2)) if __name__ == "__main__": main() FILE:scripts/search_domestic.py #!/usr/bin/env python3 import argparse import json import sys from dataclasses import asdict, is_dataclass from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from common_cli import ( add_section, airport_label, build_best_option_reasons, cabin_label, explain_recommendation, filter_and_rank_by_time_preference, format_price, format_time_or_fallback, join_nonempty, normalize_airport, parse_flexible_date, parse_time_preference_args, pretty_date, recommendation_line, time_preference_recommendation, ) def normalize_result(item): if is_dataclass(item): data = asdict(item) elif hasattr(item, "__dict__"): data = dict(item.__dict__) else: data = {"value": str(item)} return { "airline": data.get("airline", ""), "price": data.get("price", 0), "departure_time": data.get("departure_time", ""), "arrival_time": data.get("arrival_time", ""), "stops": data.get("stops", 0), "source": data.get("source", ""), "return_departure_time": data.get("return_departure_time", ""), "return_arrival_time": data.get("return_arrival_time", ""), "return_stops": data.get("return_stops", 0), "is_round_trip": data.get("is_round_trip", False), "outbound_price": data.get("outbound_price", 0), "return_price": data.get("return_price", 0), "return_airline": data.get("return_airline", ""), "confidence": data.get("confidence", 0), "extraction_source": data.get("extraction_source", ""), } def option_text(item): airline = item.get('airline', '') if item.get("is_round_trip"): return join_nonempty([ f"{airline}/{item.get('return_airline') or airline}" if airline else None, f"총 {format_price(item.get('price', 0))}", f"가는편 {join_nonempty([format_time_or_fallback(item.get('departure_time')), item.get('arrival_time')], '→')}", f"오는편 {join_nonempty([format_time_or_fallback(item.get('return_departure_time')), item.get('return_arrival_time')], '→')}", ]) return join_nonempty([ airline or None, format_price(item.get('price', 0)), join_nonempty([format_time_or_fallback(item.get('departure_time')), item.get('arrival_time')], '→'), ]) def build_summary(query, normalized, preferred, time_pref): route = f"{airport_label(query['origin'])} → {airport_label(query['destination'])}" trip_type = "왕복" if query.get("return_date") else "편도" if not normalized: return { "headline": f"{route} {trip_type} 검색 결과가 없습니다.", "route": route, "trip_type": trip_type, "cheapest_text": None, "top_options": [], "recommendation": None, "time_preference_recommendation": None, "preferred_option": preferred, } cheapest = normalized[0] second_price = normalized[1].get("price", 0) if len(normalized) > 1 else None return { "headline": f"{route} {trip_type} 최저가 {format_price(cheapest.get('price', 0))}", "route": route, "trip_type": trip_type, "cheapest_text": option_text(cheapest), "top_options": [option_text(item) for item in normalized[:3]], "recommendation": recommendation_line(option_text(cheapest), cheapest.get("price", 0), second_price), "recommendation_explained": explain_recommendation( option_text(cheapest), int(cheapest.get("price", 0) or 0), second_price, build_best_option_reasons(cheapest, second_price, time_pref), ), "time_preference_recommendation": time_preference_recommendation(preferred, cheapest, time_pref), "preferred_option": preferred, } def format_human(summary, query, count, time_pref): lines = [summary["headline"]] meta = [f"조건: 성인 {query['adults']}명 · {cabin_label(query['cabin'])} · 결과 {count}건"] if query.get("return_date"): meta.append(f"일정: {query['departure']} ~ {query['return_date']}") else: meta.append(f"일정: {query['departure']}") if time_pref.describe(): meta.append(f"시간 조건: {time_pref.describe()}") lines.extend(meta) add_section(lines, "최저가", [summary.get("cheapest_text"), summary.get("recommendation"), summary.get("recommendation_explained")]) add_section(lines, "시간대 추천", [summary.get("time_preference_recommendation")]) add_section(lines, "상위 옵션", [f"{idx}. {item}" for idx, item in enumerate(summary.get("top_options", []), start=1)]) return "\n".join(lines) def main(): parser = argparse.ArgumentParser(description="Search Korean domestic flights") parser.add_argument("--origin", required=True, help="예: GMP 또는 김포") parser.add_argument("--destination", required=True, help="예: CJU 또는 제주") parser.add_argument("--departure", required=True, help="예: 2026-03-25, 20260325, 내일") parser.add_argument("--return-date", dest="return_date", help="예: 2026-03-28, 모레") parser.add_argument("--adults", type=int, default=1) parser.add_argument("--cabin", default="ECONOMY", choices=["ECONOMY", "BUSINESS", "FIRST"]) parser.add_argument("--max-results", type=int, default=20) parser.add_argument("--time-pref", help="예: 오전, 저녁, 출발 10시 이후, 복귀 18시 이후, 너무 이른 비행 제외 8시") parser.add_argument("--depart-after", help="출발 N시 이후. 예: 10, 10:30") parser.add_argument("--return-after", help="복귀 N시 이후. 예: 18, 18:30") parser.add_argument("--exclude-early-before", help="이 시간 이전 출발 제외. 예: 8, 08:30") parser.add_argument("--prefer", choices=["late", "morning", "afternoon", "evening"], help="시간대 선호 추천") parser.add_argument("--human", action="store_true") args = parser.parse_args() workspace = Path(__file__).resolve().parents[3] repo_path = workspace / "tmp" / "Scraping-flight-information" if not repo_path.exists(): print(json.dumps({"status": "error", "message": "Source repository clone not found.", "expected": str(repo_path)}, ensure_ascii=False, indent=2)) sys.exit(1) sys.path.insert(0, str(repo_path)) try: from scraping.searcher import FlightSearcher except Exception as exc: print(json.dumps({"status": "error", "message": "Failed to import flight searcher.", "details": str(exc), "repo": str(repo_path)}, ensure_ascii=False, indent=2)) sys.exit(1) try: origin = normalize_airport(args.origin) destination = normalize_airport(args.destination) departure = pretty_date(parse_flexible_date(args.departure)) return_date = pretty_date(parse_flexible_date(args.return_date)) if args.return_date else None except ValueError as exc: print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False, indent=2)) sys.exit(1) time_pref = parse_time_preference_args(args) logs = [] def progress(msg): logs.append(str(msg)) query = { "origin": origin, "destination": destination, "departure": departure, "return_date": return_date, "adults": args.adults, "cabin": args.cabin, "max_results": args.max_results, "time_preference": time_pref.describe(), } searcher = FlightSearcher() try: results = searcher.search( origin=query["origin"], destination=query["destination"], departure_date=query["departure"], return_date=query["return_date"], adults=query["adults"], cabin_class=query["cabin"], max_results=query["max_results"], progress_callback=progress, background_mode=False, ) all_results = [normalize_result(item) for item in results] normalized, preferred_ranked = filter_and_rank_by_time_preference(all_results, time_pref) preferred = preferred_ranked[0] if preferred_ranked else None cheapest = normalized[0] if normalized else None summary = build_summary(query, normalized, preferred, time_pref) if args.human: print(format_human(summary, query, len(normalized), time_pref)) return print(json.dumps({ "status": "success", "query": query, "count": len(normalized), "summary": summary, "cheapest": cheapest, "preferred_option": preferred, "results": normalized, "all_results_before_time_filter": all_results, "logs": logs, }, ensure_ascii=False, indent=2)) finally: try: searcher.close() except Exception: pass if __name__ == "__main__": main() FILE:scripts/search_multi_destination.py #!/usr/bin/env python3 import argparse import json import sys from dataclasses import asdict, is_dataclass from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from common_cli import ( add_section, airport_label, build_best_option_reasons, cabin_label, bullet_rank_lines, explain_recommendation, filter_and_rank_by_time_preference, format_price, format_time_or_fallback, join_nonempty, normalize_airport, parse_flexible_date, parse_time_preference_args, pretty_date, recommendation_line, time_preference_recommendation, unique_codes, ) def normalize_result(item): if is_dataclass(item): return asdict(item) if hasattr(item, "__dict__"): return dict(item.__dict__) return {"value": str(item)} def main(): parser = argparse.ArgumentParser(description="Compare Korean domestic flight fares across multiple destinations") parser.add_argument("--origin", required=True) parser.add_argument("--destinations", required=True, help="Comma-separated destinations, e.g. CJU,PUS,RSU or 제주,부산,여수") parser.add_argument("--departure", required=True) parser.add_argument("--return-date") parser.add_argument("--adults", type=int, default=1) parser.add_argument("--cabin", default="ECONOMY", choices=["ECONOMY", "BUSINESS", "FIRST"]) parser.add_argument("--time-pref") parser.add_argument("--depart-after") parser.add_argument("--return-after") parser.add_argument("--exclude-early-before") parser.add_argument("--prefer", choices=["late", "morning", "afternoon", "evening"]) parser.add_argument("--human", action="store_true") args = parser.parse_args() workspace = Path(__file__).resolve().parents[3] repo_path = workspace / "tmp" / "Scraping-flight-information" if not repo_path.exists(): print(json.dumps({"status": "error", "message": "Source repository clone not found.", "expected": str(repo_path)}, ensure_ascii=False, indent=2)) sys.exit(1) sys.path.insert(0, str(repo_path)) try: from scraping.parallel import ParallelSearcher except Exception as exc: print(json.dumps({"status": "error", "message": "Failed to import parallel searcher.", "details": str(exc)}, ensure_ascii=False, indent=2)) sys.exit(1) try: origin = normalize_airport(args.origin) destinations = unique_codes([normalize_airport(x.strip()) for x in args.destinations.split(",") if x.strip()]) departure = pretty_date(parse_flexible_date(args.departure)) return_date = pretty_date(parse_flexible_date(args.return_date)) if args.return_date else None except ValueError as exc: print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False, indent=2)) sys.exit(1) time_pref = parse_time_preference_args(args) logs = [] def progress(msg): logs.append(str(msg)) searcher = ParallelSearcher() raw = searcher.search_multiple_destinations( origin=origin, destinations=destinations, departure_date=departure.replace("-", ""), return_date=return_date.replace("-", "") if return_date else None, adults=args.adults, cabin_class=args.cabin, progress_callback=progress, ) normalized = [] for dest in destinations: raw_results = [normalize_result(item) for item in (raw.get(dest, []) or [])] filtered, preferred_ranked = filter_and_rank_by_time_preference(raw_results, time_pref) cheapest = filtered[0] if filtered else None preferred = preferred_ranked[0] if preferred_ranked else None normalized.append({ "destination": dest, "destination_label": airport_label(dest), "count": len(filtered), "raw_count": len(raw_results), "cheapest_price": cheapest.get("price", 0) if cheapest else 0, "airline": cheapest.get("airline", "") if cheapest else "", "departure_time": cheapest.get("departure_time", "") if cheapest else "", "arrival_time": cheapest.get("arrival_time", "") if cheapest else "", "preferred_price": preferred.get("price", 0) if preferred else 0, "preferred_airline": preferred.get("airline", "") if preferred else "", "preferred_departure_time": preferred.get("departure_time", "") if preferred else "", "preferred_return_departure_time": preferred.get("return_departure_time", "") if preferred else "", "preferred_option": preferred, "time_recommendation": time_preference_recommendation(preferred, cheapest, time_pref), }) ranked = sorted(normalized, key=lambda x: x["cheapest_price"] if x["cheapest_price"] > 0 else 10**12) best = ranked[0] if ranked and ranked[0]["cheapest_price"] > 0 else None second_price = ranked[1]["cheapest_price"] if len(ranked) > 1 and ranked[1]["cheapest_price"] > 0 else None summary = { "headline": ( f"{airport_label(origin)} 출발 다중 목적지 최저가 {format_price(best['cheapest_price'])}" if best else f"{airport_label(origin)} 출발 다중 목적지 검색 결과가 없습니다." ), "best_option": best, "ranked_destinations": ranked, "recommendation": recommendation_line(best["destination_label"], best["cheapest_price"], second_price) if best else None, "recommendation_explained": explain_recommendation( best["destination_label"], int(best["cheapest_price"] or 0), second_price, build_best_option_reasons({ "airline": best.get("airline"), "departure_time": best.get("departure_time"), "arrival_time": best.get("arrival_time"), "cheapest_price": best.get("cheapest_price"), "price": best.get("cheapest_price"), }, second_price, time_pref), ) if best else None, } if args.human: def destination_detail(item): return join_nonempty([ item.get('airline') or None, join_nonempty([ format_time_or_fallback(item.get('departure_time')) if item.get('departure_time') else None, item.get('arrival_time') or None, ], '→'), ]) lines = [summary["headline"]] lines.append(f"조건: 출발 {airport_label(origin)} · 출발일 {departure} · 성인 {args.adults}명 · {cabin_label(args.cabin)}") if return_date: lines.append(f"왕복 일정: {departure} ~ {return_date}") if time_pref.describe(): lines.append(f"시간 조건: {time_pref.describe()}") add_section(lines, "최저가", [ join_nonempty([ f"최적 목적지: {best['destination_label']}" if best else None, format_price(best['cheapest_price']) if best else None, destination_detail(best) if best else None, ]), summary.get("recommendation"), summary.get("recommendation_explained"), ]) late_pref = next((item["time_recommendation"] for item in ranked if item.get("time_recommendation")), None) add_section(lines, "시간대 추천", [late_pref]) add_section(lines, "목적지 비교", bullet_rank_lines(ranked, "destination_label", "cheapest_price", destination_detail, limit=5)) print("\n".join(lines)) return print(json.dumps({ "status": "success", "query": { "origin": origin, "destinations": destinations, "departure": departure, "return_date": return_date, "adults": args.adults, "cabin": args.cabin, "time_preference": time_pref.describe(), }, "summary": summary, "results": ranked, "logs": logs, }, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()

Korea Domestic Flights

Comments (0)