src/utils/validator.py

# -*- coding: utf-8 -*-
import asyncio
import httpx
import json
import os
from typing import Dict, Any, List

async def check_url_live(client: httpx.AsyncClient, code: str, url: str, semaphore: asyncio.Semaphore) -> Dict[str, Any]:
    """단일 URL에 대해 세마포어 제어 하에 HEAD 요청을 비동기 실행합니다."""
    async with semaphore:
        try:
            # HEAD 요청으로 대역폭을 절약하고 신속히 응답 코드만 검사
            response = await client.head(url, timeout=5.0, follow_redirects=True)
            status_code = response.status_code
            is_live = (200 <= status_code < 400)
            return {"code": code, "url": url, "is_live": is_live, "status": status_code}
        except httpx.HTTPError as e:
            return {"code": code, "url": url, "is_live": False, "status": "HTTPError", "error": str(e)}
        except Exception as e:
            return {"code": code, "url": url, "is_live": False, "status": "Error", "error": str(e)}

async def run_mappings_purification(mappings_filepath: str, output_filepath: str):
    """mappings.json을 로드하고 생존 여부를 전수 검사하여 깨진 링크가 제거된 무결성 맵을 저장합니다."""
    if not os.path.exists(mappings_filepath):
        print(f"오류: 매핑 파일이 없습니다. ({mappings_filepath})")
        return
        
    with open(mappings_filepath, 'r', encoding='utf-8') as f:
        data = json.load(f)
        
    url_mapping = data.get("url_mapping", {})
    if not url_mapping:
        print("정화할 url_mapping 필드가 비어 있습니다.")
        return
        
    print(f"총 {len(url_mapping)}개의 추정 URL 비동기 생존 여부 전수 검증 시작...")
    
    # 윈도우 환경 소켓 과부하 및 서버 차단 방지를 위한 동시성 수 제어 (Semaphore 20 제한)
    semaphore = asyncio.Semaphore(20)
    
    # 모바일 클라이언트 User-Agent를 설정하여 봇 차단 우회
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    }
    
    cleaned_url_mapping = {}
    verified_count = 0
    dead_count = 0
    
    async with httpx.AsyncClient(headers=headers, verify=False) as client:
        tasks = []
        for code, url in url_mapping.items():
            tasks.append(check_url_live(client, code, url, semaphore))
            
        # 비동기 병렬 대기 및 프로그래스 트래킹
        results = await asyncio.gather(*tasks)
        
        for res in results:
            code = res["code"]
            url = res["url"]
            if res["is_live"]:
                cleaned_url_mapping[code] = url
                verified_count += 1
            else:
                dead_count += 1
                # [개선안] 깨진 링크는 공식 KCSC 메인 검색 페이지 등의 안전 지대로 포워딩 재매핑
                cleaned_url_mapping[code] = "https://www.kcsc.re.kr/Standard/Portal"
                
    # 원본 파일 백업 및 정화된 mappings_clean.json 저장
    data["url_mapping"] = cleaned_url_mapping
    
    # 통계 및 리팩토링 정보 적재 시 예외 방지 안전 조치
    if "metadata" not in data:
        data["metadata"] = {}
    if "statistics" not in data["metadata"]:
        data["metadata"]["statistics"] = {}
        
    data["metadata"]["statistics"]["verified_urls"] = verified_count
    data["metadata"]["statistics"]["estimated_urls"] = len(url_mapping) - verified_count
    data["metadata"]["last_purified"] = ""
    
    # 로컬 시간 또는 timezone 라이브러리 가비지 없이 무난히 API로부터 시간을 받거나 로컬 시간으로 처리
    try:
        response = httpx.get("http://worldtimeapi.org/api/ip", timeout=3.0)
        data["metadata"]["last_purified"] = response.json().get("datetime", "")
    except Exception:
        import datetime
        data["metadata"]["last_purified"] = datetime.datetime.now().isoformat()
        
    with open(output_filepath, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
        
    print(f"\n===== Mappings 정화 완료 요약 =====")
    print(f"  * 검증 생존한 유효 URL : {verified_count} 개")
    print(f"  * 깨진 추정 URL 우회    : {dead_count} 개 (안전 주소로 리다이렉트 완료)")
    print(f"  * 정화된 매핑 저장 완료  : {output_filepath}")