economic_indicator_crawler.py

"""
한국투자증권(kr.investing.com) 경제지표 수집기 - 최적화 버전
- 날짜 범위 확인
- 중복 없는 효율적 수집
- 최종 한 번만 파싱
"""

import requests
import json
import uuid
import time
from datetime import datetime, timedelta, timezone
from bs4 import BeautifulSoup
from pymongo import MongoClient

# 한국 시간대 설정
KST = timezone(timedelta(hours=9))

def now_kst():
    """한국 시간으로 현재 시간 반환"""
    return datetime.now(KST)


def get_session():
    """세션 생성"""
    session = requests.Session()
    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36",
    }

    session.get(
        "https://kr.investing.com/economic-calendar/", headers=headers, timeout=10
    )
    print("✅ 세션 생성 완료")
    return session


def make_api_call(session, data, step_name):
    """API 호출"""
    url = "https://kr.investing.com/economic-calendar/Service/getCalendarFilteredData"

    headers = {
        "accept": "*/*",
        "content-type": "application/x-www-form-urlencoded",
        "origin": "https://kr.investing.com",
        "referer": "https://kr.investing.com/economic-calendar/",
        "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36",
        "x-requested-with": "XMLHttpRequest",
    }

    response = session.post(url, data=data, headers=headers, timeout=30)
    print(f"📡 {step_name}: {response.status_code} ({len(response.text):,} 문자)")

    if response.status_code == 200:
        return response.text
    return None


def extract_pids(response_text):
    """응답에서 PID만 추출"""
    pids = []
    try:
        response_data = json.loads(response_text)
        html_content = response_data.get("data", "")

        soup = BeautifulSoup(f"<table>{html_content}</table>", "html.parser")
        rows = soup.find_all("tr", class_="js-event-item")

        for row in rows:
            event_id = row.get("id", "")
            if event_id.startswith("eventRowId_"):
                pid = f"event-{event_id.replace('eventRowId_', '')}:"
                pids.append(pid)

    except Exception as e:
        print(f"❗ PID 추출 오류: {e}")

    return pids


def extract_last_time_scope(response_text):
    """응답에서 마지막 timestamp 추출"""
    try:
        response_data = json.loads(response_text)
        html_content = response_data.get("data", "")

        # 마지막 이벤트의 timestamp 찾기
        soup = BeautifulSoup(f"<table>{html_content}</table>", "html.parser")
        date_rows = soup.find_all("td", {"class": "theDay"})

        if date_rows:
            last_date_row = date_rows[-1]
            date_id = last_date_row.get("id", "")
            if date_id and date_id.startswith("theDay"):
                # theDay1755043200 형식에서 숫자 추출
                timestamp = date_id.replace("theDay", "")
                if timestamp.isdigit():
                    return timestamp
    except:
        pass
    return "1755043200"  # 기본값


def collect_all_responses(session, start_date, end_date):
    """모든 응답 수집 (파싱하지 않고 저장만)"""
    all_responses = []
    all_pids = []  # 순서 유지를 위해 list 사용
    seen_pids = set()  # 중복 체크용
    limit_from = 0
    last_time_scope = "1755043200"

    print(f"📅 날짜 범위: {start_date} ~ {end_date}")
    print(f"🔍 날짜 확인: 시작일={start_date}, 종료일={end_date}")
    print("=" * 60)

    # 1단계: 초기 호출
    data = {
        "country[]": ["5", "11", "37", "35"],  # 미국, 한국, 중국, 일본
        "dateFrom": start_date,
        "dateTo": end_date,
        "timeZone": "88",
        "timeFilter": "timeRemain",
        "currentTab": "custom",
        "submitFilters": "1",
        "limit_from": str(limit_from),
    }

    response = make_api_call(session, data, f"Step {limit_from+1}")
    if not response:
        return []

    all_responses.append(response)
    new_pids = extract_pids(response)
    for pid in new_pids:
        if pid not in seen_pids:
            all_pids.append(pid)
            seen_pids.add(pid)

    last_time_scope = extract_last_time_scope(response)
    print(f"✅ Step 1: {len(new_pids)}개 PID, last_time_scope={last_time_scope}")

    # 2단계 이상: limit_from을 증가시키며 계속 수집
    no_new_data_count = 0
    max_steps = 10

    for step in range(1, max_steps):
        limit_from = step
        time.sleep(0.3)

        # 요청 데이터
        data = {
            "country[]": ["5", "11", "37", "35"],
            "dateFrom": start_date,
            "dateTo": end_date,
            "timeZone": "88",
            "timeFilter": "timeRemain",
            "currentTab": "custom",
            "submitFilters": "0",
            "limit_from": str(limit_from),
            "last_time_scope": last_time_scope,
            "byHandler": "true",
        }

        # 현재 단계에 맞는 PID만 추가 (실제 웹사이트처럼)
        if step == 1:  # 2단계
            # 처음 200개 PID
            for i, pid in enumerate(all_pids[:200]):
                data[f"pids[{i}]"] = pid
        elif step == 2:  # 3단계
            # 다음 200개 PID
            start_idx = 200
            end_idx = min(400, len(all_pids))
            for i, pid in enumerate(all_pids[start_idx:end_idx]):
                data[f"pids[{i}]"] = pid
            # last_time_scope 업데이트
            last_time_scope = "1755907200"  # 실제 값 사용
        else:
            # 모든 PID 추가
            for i, pid in enumerate(all_pids):
                data[f"pids[{i}]"] = pid

        response = make_api_call(session, data, f"Step {limit_from+1}")
        if not response:
            break

        # 새로운 PID 확인
        new_pids = extract_pids(response)
        new_unique = 0
        for pid in new_pids:
            if pid not in seen_pids:
                all_pids.append(pid)
                seen_pids.add(pid)
                new_unique += 1

        if new_unique > 0 or len(new_pids) > 0:
            all_responses.append(response)
            # 새 last_time_scope 추출
            new_scope = extract_last_time_scope(response)
            if new_scope != last_time_scope:
                last_time_scope = new_scope
            print(
                f"✅ Step {limit_from+1}: {new_unique}개 신규 PID (총 {len(all_pids)}개)"
            )
            no_new_data_count = 0
        else:
            no_new_data_count += 1
            print(f"⚠️ Step {limit_from+1}: 새 데이터 없음 ({no_new_data_count}/3)")

        if no_new_data_count >= 3:
            break

    print(f"\n📊 수집 완료: {len(all_responses)}개 응답, {len(all_pids)}개 고유 PID")
    return all_responses


def parse_all_responses(responses):
    """수집된 모든 응답을 한 번에 파싱"""
    all_documents = []
    current_date = now_kst().strftime("%Y-%m-%d")

    print("\n🔍 전체 응답 파싱 중...")

    for response_text in responses:
        try:
            response_data = json.loads(response_text)
            html_content = response_data.get("data", "")

            if not html_content:
                continue

            soup = BeautifulSoup(f"<table>{html_content}</table>", "html.parser")
            all_rows = soup.find_all("tr")

            for row in all_rows:
                # 날짜 행 처리
                td_with_colspan = row.find("td", {"colspan": "9", "class": "theDay"})
                if td_with_colspan:
                    date_text = td_with_colspan.text.strip()
                    import re

                    match = re.search(
                        r"(\d{4})년\s*(\d{1,2})월\s*(\d{1,2})일", date_text
                    )
                    if match:
                        year, month, day = match.groups()
                        current_date = f"{year}-{int(month):02d}-{int(day):02d}"
                    continue

                # 이벤트 행 처리
                if "js-event-item" not in row.get("class", []):
                    continue

                cells = row.find_all("td")
                if len(cells) < 7:
                    continue

                time_text = cells[0].text.strip()
                currency = cells[1].text.strip()

                # 중요도
                importance_cell = cells[2]
                importance = len(
                    importance_cell.find_all("i", class_="grayFullBullishIcon")
                )

                event_name = cells[3].text.strip()
                actual = (
                    cells[4].text.strip()
                    if cells[4].text.strip() not in ["-", ""]
                    else None
                )
                forecast = (
                    cells[5].text.strip()
                    if cells[5].text.strip() not in ["-", ""]
                    else None
                )
                previous = (
                    cells[6].text.strip()
                    if cells[6].text.strip() not in ["-", ""]
                    else None
                )

                if not event_name:
                    continue

                # 국가 매핑
                country_map = {
                    "USD": "미국",
                    "KRW": "한국",
                    "JPY": "일본",
                    "CNY": "중국",
                    "EUR": "유로존",
                    "GBP": "영국",
                }
                country = country_map.get(currency, "기타")

                doc = {
                    "economic_uuid": str(uuid.uuid4()),
                    "date": current_date,
                    "time": time_text,
                    "country": country,
                    "currency": currency,
                    "indicator_name": event_name,
                    "actual_value": actual,
                    "forecast_value": forecast,
                    "previous_value": previous,
                    "importance": importance,
                    "importance_text": (
                        "High"
                        if importance >= 3
                        else "Medium" if importance >= 2 else "Low"
                    ),
                    "source": "kr.investing.com",
                    "created_at": now_kst().isoformat(),
                }

                all_documents.append(doc)

        except Exception as e:
            print(f"❗ 응답 파싱 오류: {e}")

    # 중복 제거
    unique_docs = []
    seen_keys = set()

    for doc in all_documents:
        key = f"{doc['date']}|{doc['time']}|{doc['country']}|{doc['indicator_name']}"
        if key not in seen_keys:
            seen_keys.add(key)
            unique_docs.append(doc)

    print(
        f"✅ 파싱 완료: {len(all_documents)}개 → {len(unique_docs)}개 (중복 {len(all_documents)-len(unique_docs)}개 제거)"
    )
    return unique_docs


def save_to_mongodb(documents):
    """MongoDB에 저장 - 일일 업데이트 모드"""
    if not documents:
        print("❗ 저장할 데이터가 없습니다")
        return

    # 도커 MongoDB 컨테이너 연결
    client = MongoClient("mongodb://root:1234@mongodb:27017",
                             authSource="admin")
    db = client["crame"]
    collection = db["economic_indicator"]

    # 인덱스 생성
    try:
        # 1년 TTL 설정 (365일 * 24시간 * 3600초)
        collection.create_index("created_at", expireAfterSeconds=31536000)
        collection.create_index(
            [("date", 1), ("time", 1), ("country", 1), ("indicator_name", 1)]
        )
    except Exception as e:
        print(f"⚠️ 인덱스 생성 경고: {e}")

    saved_count = 0
    updated_count = 0
    duplicate_count = 0

    for doc in documents:
        try:
            # 중복 체크 (날짜, 시간, 국가, 지표명으로)
            filter_query = {
                "date": doc["date"],
                "time": doc["time"],
                "country": doc["country"],
                "indicator_name": doc["indicator_name"],
            }

            existing = collection.find_one(filter_query)

            if existing:
                # 기존 데이터와 비교하여 값이 다르면 업데이트
                needs_update = False

                if (
                    existing.get("actual_value") != doc.get("actual_value")
                    or existing.get("forecast_value") != doc.get("forecast_value")
                    or existing.get("previous_value") != doc.get("previous_value")
                ):
                    needs_update = True

                if needs_update:
                    # created_at은 유지하고 나머지만 업데이트
                    update_doc = doc.copy()
                    update_doc["updated_at"] = now_kst().isoformat()

                    collection.update_one(
                        {"_id": existing["_id"]}, {"$set": update_doc}
                    )
                    updated_count += 1
                else:
                    duplicate_count += 1
            else:
                # 새로운 데이터 삽입
                collection.insert_one(doc)
                saved_count += 1

        except Exception as e:
            print(f"❗ 저장 오류: {e}")
            continue

    print(f"\n📊 MongoDB 저장 결과:")
    print(f"   ✅ 신규 저장: {saved_count}개")
    print(f"   🔄 업데이트: {updated_count}개")
    print(f"   ⏭️ 중복 건너뜀: {duplicate_count}개")
    print(f"   📍 저장 위치: crame.economic_indicator")
    print(f"   ⏰ TTL 설정: 1년 (자동 만료)")


def save_json(documents):
    """JSON 파일 저장"""
    if not documents:
        return

    timestamp = now_kst().strftime("%Y%m%d_%H%M%S")
    filename = f"kr_investing_{timestamp}.json"

    with open(filename, "w", encoding="utf-8") as f:
        json.dump(documents, f, ensure_ascii=False, indent=2)

    print(f"\n💾 JSON 저장:")
    print(f"   📁 전체: {filename} ({len(documents)}개)")

    # 국가별 파일
    countries = set(doc["country"] for doc in documents)
    for country in countries:
        country_docs = [doc for doc in documents if doc["country"] == country]
        country_file = f"kr_investing_{country}_{timestamp}.json"

        with open(country_file, "w", encoding="utf-8") as f:
            json.dump(country_docs, f, ensure_ascii=False, indent=2)

        print(f"   🏳️ {country}: {country_file} ({len(country_docs)}개)")

    # 날짜별 분포 확인
    date_stats = {}
    for doc in documents:
        date = doc["date"]
        date_stats[date] = date_stats.get(date, 0) + 1

    print(f"\n📅 날짜별 분포:")
    for date in sorted(date_stats.keys()):
        print(f"   {date}: {date_stats[date]}개")


def main():
    """메인 실행 - 일일 업데이트 모드"""
    print("🚀 한국투자증권 경제지표 수집기 (최적화)")
    print("📅 일일 업데이트 모드 (어제~60일후)")
    print("=" * 60)

    now = now_kst()

    # 어제부터 60일 후까지
    yesterday = now - timedelta(days=1)
    start_date = yesterday.strftime("%Y-%m-%d")

    # 60일 후
    sixty_days_later = now + timedelta(days=60)
    end_date = sixty_days_later.strftime("%Y-%m-%d")

    print(f"📅 설정된 기간: {start_date} ~ {end_date} (어제부터 60일)")
    print(f"🌍 대상 국가: 미국, 한국, 일본, 중국\n")

    # 세션 생성
    session = get_session()

    # 모든 응답 수집
    responses = collect_all_responses(session, start_date, end_date)

    if not responses:
        print("❌ 데이터 수집 실패")
        return

    # 한 번에 파싱
    documents = parse_all_responses(responses)

    # 국가별 통계
    country_stats = {}
    for doc in documents:
        country = doc["country"]
        country_stats[country] = country_stats.get(country, 0) + 1

    print("\n🌍 국가별 현황:")
    for country, count in sorted(
        country_stats.items(), key=lambda x: x[1], reverse=True
    ):
        print(f"   {country}: {count}개")

    # MongoDB 저장
    save_to_mongodb(documents)

    print(f"\n✅ 완료!")
    print(f"   📊 총 {len(documents)}개 경제지표")
    print(
        f"   🕐 {now_kst().strftime('%Y-%m-%d %H:%M:%S')}"
    )


if __name__ == "__main__":
    main()