API 크롤러 템플릿

"""
API 크롤러 템플릿
라이브러리: Requests (API 직접 호출)
용도: JSON API 엔드포인트가 발견된 경우

UV 설정:
    uv init crawler-project
    uv add requests pandas
    uv run crawler.py
"""

import requests
import pandas as pd
from typing import List, Dict, Optional, Any
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class APICrawler:
    """API 직접 호출 크롤러"""

    def __init__(self, base_url: str):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            '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',
            'Accept': 'application/json',
            'Content-Type': 'application/json',
        })
        self.data: List[Dict] = []

    def set_auth_header(self, token: str, auth_type: str = 'Bearer'):
        """인증 헤더 설정"""
        self.session.headers.update({
            'Authorization': f'{auth_type} {token}'
        })

    def set_cookie(self, cookies: Dict[str, str]):
        """쿠키 설정"""
        self.session.cookies.update(cookies)

    def fetch_api(self, endpoint: str, method: str = 'GET', params: Dict = None, data: Dict = None) -> Optional[Any]:
        """API 요청"""
        url = f"{self.base_url}{endpoint}" if not endpoint.startswith('http') else endpoint

        try:
            if method.upper() == 'GET':
                response = self.session.get(url, params=params, timeout=10)
            elif method.upper() == 'POST':
                response = self.session.post(url, json=data, params=params, timeout=10)
            else:
                logger.error(f"지원하지 않는 HTTP 메서드: {method}")
                return None

            response.raise_for_status()
            return response.json()

        except requests.RequestException as e:
            logger.error(f"API 요청 실패: {url} - {e}")
            return None
        except ValueError as e:
            logger.error(f"JSON 파싱 실패: {url} - {e}")
            return None

    def parse_response(self, response: Any) -> List[Dict]:
        """API 응답 파싱 (사이트에 맞게 수정 필요)"""
        # TODO: 실제 API 응답 구조에 맞게 수정

        # 응답이 리스트인 경우
        if isinstance(response, list):
            return response

        # 응답이 dict이고 data 키가 있는 경우
        if isinstance(response, dict):
            if 'data' in response:
                return response['data'] if isinstance(response['data'], list) else [response['data']]
            if 'items' in response:
                return response['items']
            if 'results' in response:
                return response['results']

        return [response] if response else []

    def crawl_with_pagination(self, endpoint: str, page_param: str = 'page',
                               limit_param: str = 'limit', limit: int = 20,
                               max_pages: int = 10) -> List[Dict]:
        """페이지네이션 처리 크롤링"""
        page = 1

        while page <= max_pages:
            logger.info(f"페이지 {page} 크롤링 중...")

            params = {
                page_param: page,
                limit_param: limit,
            }

            response = self.fetch_api(endpoint, params=params)

            if response is None:
                break

            items = self.parse_response(response)

            if not items:
                logger.info("더 이상 데이터가 없습니다")
                break

            self.data.extend(items)
            logger.info(f"페이지 {page}: {len(items)}개 항목 수집")

            # 마지막 페이지 체크
            if len(items) < limit:
                break

            page += 1
            time.sleep(0.5)  # API 요청 간격

        logger.info(f"총 {len(self.data)}개 항목 수집 완료")
        return self.data

    def crawl(self, endpoint: str = '', params: Dict = None) -> List[Dict]:
        """단일 API 크롤링"""
        logger.info(f"API 크롤링 중: {self.base_url}{endpoint}")

        response = self.fetch_api(endpoint, params=params)

        if response:
            self.data = self.parse_response(response)
            logger.info(f"총 {len(self.data)}개 항목 수집 완료")

        return self.data

    def save_to_csv(self, filename: str = 'output.csv'):
        """CSV 파일로 저장"""
        if not self.data:
            logger.warning("저장할 데이터가 없습니다")
            return

        df = pd.DataFrame(self.data)
        df.to_csv(filename, index=False, encoding='utf-8-sig')
        logger.info(f"CSV 저장 완료: {filename}")

    def save_to_json(self, filename: str = 'output.json'):
        """JSON 파일로 저장"""
        if not self.data:
            logger.warning("저장할 데이터가 없습니다")
            return

        df = pd.DataFrame(self.data)
        df.to_json(filename, orient='records', force_ascii=False, indent=2)
        logger.info(f"JSON 저장 완료: {filename}")


def main():
    # TODO: API 베이스 URL로 수정
    BASE_URL = "https://api.example.com"
    API_ENDPOINT = "/v1/items"

    crawler = APICrawler(BASE_URL)

    # 인증이 필요한 경우
    # crawler.set_auth_header("your-api-token")

    # 단일 API 호출
    # data = crawler.crawl(API_ENDPOINT)

    # 페이지네이션 처리
    data = crawler.crawl_with_pagination(
        endpoint=API_ENDPOINT,
        page_param='page',
        limit_param='limit',
        limit=20,
        max_pages=10
    )

    crawler.save_to_csv('output.csv')

    print(f"\n수집된 데이터: {len(data)}개")
    if data:
        print(pd.DataFrame(data).head())


if __name__ == "__main__":
    main()