auto-stock/backend/app/services/news_sentiment.py

"""뉴스 수집 + 감성 분석 엔진 (news_sentiment.py)

아키텍처:
  - 수집 소스 3개: 네이버 금융(종목별), 네이버 시장 뉴스, NewsAPI.org(선택)
  - 감성 분석 2단계: (1) 키워드 사전 기반 빠른 분석 → (2) AI 배치 분석 (선택)
  - Redis 캐시: TTL 30분 (settings.NEWS_CACHE_TTL)
  - DB 저장: db_worker 큐 경유 (CLAUDE.md 3.2 준수)

공개 API:
  - get_stock_news_score(stock_code) → float (-1.0 ~ +1.0)
  - get_market_news_score()          → float (-1.0 ~ +1.0)
  - collect_stock_news(stock_code)   → list[dict]
  - collect_market_news()            → list[dict]
"""
import asyncio
import logging
import re
import time
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional, Tuple

import httpx
from bs4 import BeautifulSoup

from app.core.config import settings
from app.core.redis import setex_json, get_json
from app.services.db_worker import db_task_queue
from app.core.timeutil import now_kst, KST

from app.core.log_throttle import warn_throttled  # noqa: E402

logger = logging.getLogger(__name__)

# ═════════════════════════════════════════════════════════════════════════════
# 1. 한국어 금융 감성 키워드 사전
# ═════════════════════════════════════════════════════════════════════════════

# (키워드, 가중치) 튜플 — 가중치 절대값이 클수록 강한 신호
_POSITIVE_KEYWORDS: List[Tuple[str, float]] = [
    # 실적 관련 (강한 긍정)
    ("실적호전", 0.8), ("어닝서프라이즈", 0.9), ("깜짝실적", 0.9),
    ("흑자전환", 0.85), ("사상최대", 0.8), ("최대실적", 0.8),
    ("영업이익증가", 0.7), ("매출증가", 0.6), ("순이익증가", 0.7),
    ("실적개선", 0.6), ("호실적", 0.7),
    # 투자의견 (중간 긍정)
    ("목표가상향", 0.7), ("투자의견상향", 0.7), ("매수추천", 0.6),
    ("컨센서스상회", 0.65), ("상향조정", 0.6), ("목표주가상향", 0.7),
    # 사업/계약 (중간 긍정)
    ("수주", 0.6), ("대형계약", 0.7), ("신규수주", 0.65),
    ("수출증가", 0.5), ("해외진출", 0.5), ("사업확장", 0.5),
    ("신사업", 0.4), ("협력", 0.3), ("제휴", 0.3), ("MOU", 0.3),
    # 제약/바이오 (강한 긍정)
    ("신약승인", 0.9), ("FDA승인", 0.95), ("임상성공", 0.85),
    ("기술이전", 0.7), ("임상3상", 0.6),
    # 주주환원 (중간 긍정)
    ("배당확대", 0.5), ("배당증가", 0.5), ("자사주매입", 0.6),
    ("자사주소각", 0.65), ("주주환원", 0.5),
    # 일반 긍정
    ("강세", 0.3), ("상승", 0.2), ("반등", 0.3), ("회복", 0.3),
    ("돌파", 0.3), ("신고가", 0.4), ("급등", 0.35),
]

_NEGATIVE_KEYWORDS: List[Tuple[str, float]] = [
    # 실적 관련 (강한 부정)
    ("실적악화", -0.8), ("어닝쇼크", -0.9), ("적자전환", -0.85),
    ("적자확대", -0.8), ("영업손실", -0.7), ("매출감소", -0.6),
    ("순손실", -0.7), ("실적부진", -0.6),
    # 투자의견 (중간 부정)
    ("목표가하향", -0.7), ("투자의견하향", -0.7), ("매도추천", -0.6),
    ("하향조정", -0.6), ("목표주가하향", -0.7), ("컨센서스하회", -0.65),
    # 기업 리스크 (강한 부정)
    ("유상증자", -0.7), ("감자", -0.8), ("무상감자", -0.85),
    ("상장폐지", -0.95), ("관리종목", -0.9), ("영업정지", -0.9),
    ("횡령", -0.85), ("배임", -0.8), ("분식회계", -0.9),
    ("소송", -0.5), ("과징금", -0.5), ("리콜", -0.6),
    ("부도", -0.95), ("워크아웃", -0.9), ("법정관리", -0.9),
    # 규제/정책 (중간 부정)
    ("규제강화", -0.4), ("공매도", -0.3), ("대주주매도", -0.5),
    ("지분매각", -0.4), ("블록딜", -0.35),
    # 제약/바이오 (강한 부정)
    ("임상실패", -0.9), ("FDA거부", -0.9), ("임상중단", -0.8),
    # 일반 부정
    ("하락", -0.2), ("급락", -0.4), ("폭락", -0.5),
    ("약세", -0.3), ("부진", -0.3), ("위기", -0.4),
    ("불확실", -0.2), ("우려", -0.2),
]

# 전처리된 룩업용 — O(1) 탐색
_POS_DICT: Dict[str, float] = {kw: w for kw, w in _POSITIVE_KEYWORDS}
_NEG_DICT: Dict[str, float] = {kw: w for kw, w in _NEGATIVE_KEYWORDS}

# HTTP 헤더 (네이버 크롤링)
_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    )
}

# 수집 0건이 이 횟수만큼 연속되면 error + 텔레그램 경보 (getattr: settings 에 없으면 기본값)
_EMPTY_ALERT_DEFAULT = 5
# 0건 결과 단기 캐시(초) — 점수 0.0 을 정상 캐시하지 않고, 재크롤링 폭주만 막는다
_EMPTY_NEGATIVE_TTL = 300

# 수집 건강 상태 (프로세스 내 메모리). /health 등에서 get_news_collect_health() 로 조회.
_health: Dict[str, Dict[str, Any]] = {
    "stock": {"consecutive_empty": 0, "last_success_at": None, "alerted": False},
    "market": {"consecutive_empty": 0, "last_success_at": None, "alerted": False},
}


def get_news_collect_health() -> Dict[str, Any]:
    """마지막 뉴스 수집 성공 시각과 연속 0건 횟수 (종목/시장 구분)."""
    out: Dict[str, Any] = {}
    for kind, st in _health.items():
        ts = st["last_success_at"]
        out[kind] = {
            "consecutive_empty": st["consecutive_empty"],
            "last_success_at": ts.isoformat() if ts else None,
        }
    return out


async def _record_collect_result(kind: str, count: int) -> None:
    """수집 결과를 기록하고 연속 0건이 임계에 도달하면 error + 텔레그램 경보."""
    st = _health[kind]
    if count > 0:
        st["consecutive_empty"] = 0
        st["last_success_at"] = now_kst()
        st["alerted"] = False
        return
    st["consecutive_empty"] += 1
    threshold = int(getattr(settings, "NEWS_EMPTY_ALERT_THRESHOLD", _EMPTY_ALERT_DEFAULT))
    if st["consecutive_empty"] < threshold or st["alerted"]:
        return
    st["alerted"] = True
    last = st["last_success_at"].isoformat() if st["last_success_at"] else "없음"
    msg = (
        f"[NEWS] {kind} 뉴스 수집 {st['consecutive_empty']}회 연속 0건 - "
        f"악재 차단/축소 필터가 동작하지 않음 (마지막 성공: {last})"
    )
    logger.error(msg)
    try:
        from app.services.telegram_notifier import telegram_notifier, _escape_markdown_v2
        await telegram_notifier.send_message(_escape_markdown_v2(msg))
    except Exception as e:
        logger.warning(f"[NEWS] 텔레그램 경보 전송 실패: {e}")


def _to_kst_naive(dt: datetime) -> datetime:
    """aware 값은 KST 로 변환 후 tz 제거, naive 는 KST 로 간주하고 그대로 반환."""
    if dt.tzinfo is not None:
        return dt.astimezone(KST).replace(tzinfo=None)
    return dt


def _parse_iso_to_kst(value: str) -> datetime:
    """ISO 문자열('...Z' 포함) → KST naive. 실패 시 ValueError."""
    return _to_kst_naive(datetime.fromisoformat(value.replace("Z", "+00:00")))



def _parse_naver_json_datetime(raw: Any) -> datetime:
    """네이버 모바일 JSON 의 datetime → KST naive.

    'YYYYMMDDHHMM[SS]' 는 KST 로 간주(오프셋 없음), ISO 문자열은 오프셋을 존중해 KST 로 변환.
    파싱 실패 시 현재 KST.
    """
    dt_raw = str(raw or "").strip()
    for fmt in ("%Y%m%d%H%M", "%Y%m%d%H%M%S"):
        try:
            return datetime.strptime(dt_raw, fmt)
        except ValueError:
            continue
    try:
        return _parse_iso_to_kst(dt_raw)
    except Exception:
        return now_kst()


# ═════════════════════════════════════════════════════════════════════════════
# 2. 감성 분석 함수
# ═════════════════════════════════════════════════════════════════════════════

def analyze_sentiment_keyword(title: str, content: str = "") -> Tuple[float, str]:
    """키워드 사전 기반 감성 점수 산출 (1단계).

    Args:
        title:   뉴스 제목
        content: 뉴스 본문 (선택, 제목에 비해 가중치 낮음)

    Returns:
        (score, label)
        score: -1.0 ~ +1.0 (클램프 적용)
        label: "positive" | "negative" | "neutral"
    """
    text = (title or "").replace(" ", "") + (content or "").replace(" ", "")
    if not text:
        return 0.0, "neutral"

    total_score = 0.0
    match_count = 0

    # 제목 매칭 — 가중치 1.5배 (제목이 핵심 정보)
    title_clean = (title or "").replace(" ", "")
    for kw, weight in _POSITIVE_KEYWORDS:
        if kw in title_clean:
            total_score += weight * 1.5
            match_count += 1
    for kw, weight in _NEGATIVE_KEYWORDS:
        if kw in title_clean:
            total_score += weight * 1.5  # weight 자체가 음수
            match_count += 1

    # 본문 매칭 — 가중치 0.5배
    if content:
        content_clean = content.replace(" ", "")
        for kw, weight in _POSITIVE_KEYWORDS:
            if kw in content_clean:
                total_score += weight * 0.5
                match_count += 1
        for kw, weight in _NEGATIVE_KEYWORDS:
            if kw in content_clean:
                total_score += weight * 0.5
                match_count += 1

    if match_count == 0:
        return 0.0, "neutral"

    # 평균 후 클램프
    avg_score = total_score / match_count
    clamped = max(-1.0, min(1.0, avg_score))

    if clamped > 0.1:
        label = "positive"
    elif clamped < -0.1:
        label = "negative"
    else:
        label = "neutral"

    return round(clamped, 3), label


def _compute_aggregate_score(articles: List[Dict[str, Any]]) -> float:
    """뉴스 리스트에서 종합 감성 점수 산출.

    최신 뉴스일수록 가중치가 높은 시간 감쇠 적용.
    - 24시간 이내: 가중치 1.0
    - 24~48시간: 가중치 0.6
    - 48시간 이후: 가중치 0.3
    """
    if not articles:
        return 0.0

    now = now_kst()
    weighted_sum = 0.0
    weight_total = 0.0

    for art in articles:
        score = art.get("sentiment_score", 0.0)
        if score is None:
            score = 0.0

        pub_at = art.get("published_at")
        if isinstance(pub_at, str):
            try:
                pub_at = _parse_iso_to_kst(pub_at)
            except Exception:
                pub_at = now

        if isinstance(pub_at, datetime):
            hours_ago = (now - _to_kst_naive(pub_at)).total_seconds() / 3600
        else:
            hours_ago = 24  # 파싱 실패 시 기본 24시간

        if hours_ago <= 24:
            time_weight = 1.0
        elif hours_ago <= 48:
            time_weight = 0.6
        else:
            time_weight = 0.3

        weighted_sum += score * time_weight
        weight_total += time_weight

    if weight_total == 0:
        return 0.0

    return round(max(-1.0, min(1.0, weighted_sum / weight_total)), 3)


# ═════════════════════════════════════════════════════════════════════════════
# 3. 뉴스 수집 (네이버 금융 — 종목별)
# ═════════════════════════════════════════════════════════════════════════════

async def _collect_stock_news_json(stock_code: str, limit: int) -> List[Dict[str, Any]]:
    """m.stock.naver.com 종목 뉴스 JSON API (1차 소스). 실패/0건이면 빈 리스트.

    응답 스키마는 외부 호출 금지로 실측하지 못했다(추정): 그룹 리스트
    [{"items":[{"officeId","articleId","officeName","datetime":"YYYYMMDDHHMM","title"}]}]
    또는 items 리스트 자체. 두 형태 모두 방어적으로 파싱한다.
    """
    url = f"https://m.stock.naver.com/api/news/stock/{stock_code}?pageSize={limit}&page=1"
    async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
        resp = await client.get(url, headers=_HEADERS)
    if resp.status_code != 200:
        logger.warning(f"[NEWS] 네이버 모바일 종목뉴스 HTTP {resp.status_code}: {stock_code}")
        return []

    data = resp.json()
    groups = data if isinstance(data, list) else (data.get("items") or data.get("newsList") or [])
    raw_items: List[Dict[str, Any]] = []
    for g in groups:
        if isinstance(g, dict) and isinstance(g.get("items"), list):
            raw_items.extend(i for i in g["items"] if isinstance(i, dict))
        elif isinstance(g, dict):
            raw_items.append(g)

    articles: List[Dict[str, Any]] = []
    for it in raw_items:
        title = (it.get("title") or "").strip()
        if not title:
            continue
        office, art = it.get("officeId"), it.get("articleId")
        link = f"https://n.news.naver.com/mnews/article/{office}/{art}" if office and art else ""
        published_at = _parse_naver_json_datetime(it.get("datetime") or it.get("dt"))
        score, label = analyze_sentiment_keyword(title)
        articles.append({
            "title": title,
            "source": it.get("officeName") or "네이버금융",
            "url": link,
            "stock_codes": [stock_code],
            "sentiment": label,
            "sentiment_score": score,
            "published_at": published_at,
        })
        if len(articles) >= limit:
            break
    return articles


async def collect_stock_news(
    stock_code: str,
    limit: int = 0,
) -> List[Dict[str, Any]]:
    """종목별 뉴스를 수집하고 감성 분석을 수행한다.

    1차: m.stock.naver.com JSON API, 2차(폴백): finance.naver.com HTML 크롤링.

    Args:
        stock_code: 종목 코드 (예: "005930")
        limit:      수집할 최대 뉴스 개수 (0 이면 settings.NEWS_MAX_ARTICLES)

    Returns:
        [{"title", "source", "url", "published_at", "sentiment", "sentiment_score"}, ...]
    """
    if limit <= 0:
        limit = settings.NEWS_MAX_ARTICLES

    try:
        articles = await _collect_stock_news_json(stock_code, limit)
        if articles:
            return articles
    except httpx.TimeoutException:
        logger.warning(f"[NEWS] 네이버 모바일 종목뉴스 타임아웃: {stock_code}")
    except Exception as e:
        logger.warning(f"[NEWS] 네이버 모바일 종목뉴스 오류 ({stock_code}): {e}")

    return await _collect_stock_news_html(stock_code, limit)


async def _collect_stock_news_html(stock_code: str, limit: int) -> List[Dict[str, Any]]:
    """finance.naver.com HTML 크롤링 (폴백)."""
    url = f"https://finance.naver.com/item/news_news.naver?code={stock_code}&page=1"
    articles: List[Dict[str, Any]] = []

    try:
        async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
            resp = await client.get(url, headers=_HEADERS)

        if resp.status_code != 200:
            logger.warning(f"[NEWS] 네이버 종목뉴스 HTTP {resp.status_code}: {stock_code}")
            return []

        soup = BeautifulSoup(resp.text, "html.parser")

        for row in soup.select("table.type5 tbody tr"):
            title_tag = row.select_one(".title a")
            if not title_tag:
                continue

            title = title_tag.get_text(strip=True)
            href = title_tag.get("href", "")
            link = ("https://finance.naver.com" + href) if href else ""
            date_tag = row.select_one(".date")
            date_str = date_tag.get_text(strip=True) if date_tag else ""
            source_tag = row.select_one(".info")
            source = source_tag.get_text(strip=True) if source_tag else "네이버금융"

            published_at = _parse_naver_date(date_str)
            score, label = analyze_sentiment_keyword(title)

            articles.append({
                "title": title,
                "source": source,
                "url": link,
                "stock_codes": [stock_code],
                "sentiment": label,
                "sentiment_score": score,
                "published_at": published_at,
            })

            if len(articles) >= limit:
                break

    except httpx.TimeoutException:
        logger.warning(f"[NEWS] 네이버 종목뉴스 타임아웃: {stock_code}")
    except Exception as e:
        logger.error(f"[NEWS] 네이버 종목뉴스 크롤링 오류 ({stock_code}): {e}")

    return articles


# ═════════════════════════════════════════════════════════════════════════════
# 4. 뉴스 수집 (네이버 시장 뉴스 — 코스피/코스닥)
# ═════════════════════════════════════════════════════════════════════════════

# 시장 뉴스 JSON 후보 엔드포인트 (실측 불가 - 추정). 순서대로 시도, 첫 비어있지 않은 결과 사용.
_MARKET_NEWS_JSON_URLS = (
    "https://m.stock.naver.com/api/news/mainnews?pageSize={limit}&page=1",
    "https://m.stock.naver.com/api/news/category/domestic?pageSize={limit}&page=1",
)


def _extract_news_items(data: Any) -> List[Dict[str, Any]]:
    """다양한 JSON 형태(list / {items|newsList|news} / 그룹 중첩)에서 기사 dict 목록 추출."""
    if isinstance(data, dict):
        groups = data.get("items") or data.get("newsList") or data.get("news") or []
    else:
        groups = data
    out: List[Dict[str, Any]] = []
    if not isinstance(groups, list):
        return out
    for g in groups:
        if isinstance(g, dict) and isinstance(g.get("items"), list):
            out.extend(i for i in g["items"] if isinstance(i, dict))
        elif isinstance(g, dict):
            out.append(g)
    return out


async def _collect_market_news_json(limit: int) -> List[Dict[str, Any]]:
    """m.stock.naver.com 시장(주요) 뉴스 JSON (1차 소스). 실패/0건이면 빈 리스트.

    스키마·경로는 외부 호출 금지로 실측하지 못했다(추정). 후보 URL 을 순차 시도한다.
    """
    articles: List[Dict[str, Any]] = []
    async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
        for tmpl in _MARKET_NEWS_JSON_URLS:
            url = tmpl.format(limit=limit)
            try:
                resp = await client.get(url, headers=_HEADERS)
                if resp.status_code != 200:
                    logger.warning(f"[NEWS] 시장뉴스 JSON HTTP {resp.status_code}: {url}")
                    continue
                raw_items = _extract_news_items(resp.json())
            except Exception as e:
                logger.warning(f"[NEWS] 시장뉴스 JSON 오류 ({url}): {e}")
                continue
            for it in raw_items:
                title = (it.get("title") or it.get("titleFull") or "").strip()
                if not title:
                    continue
                office, art = it.get("officeId"), it.get("articleId")
                link = f"https://n.news.naver.com/mnews/article/{office}/{art}" if office and art else ""
                score, label = analyze_sentiment_keyword(title)
                articles.append({
                    "title": title,
                    "source": it.get("officeName") or "네이버금융",
                    "url": link,
                    "stock_codes": [],
                    "sentiment": label,
                    "sentiment_score": score,
                    "published_at": _parse_naver_json_datetime(it.get("datetime") or it.get("dt")),
                })
                if len(articles) >= limit:
                    return articles
            if articles:
                return articles
    return articles


async def collect_market_news(limit: int = 0) -> List[Dict[str, Any]]:
    """시장 뉴스 수집 + 감성 분석.

    1차: m.stock.naver.com JSON, 2차(폴백): finance.naver.com HTML 크롤링.
    """
    if limit <= 0:
        limit = settings.NEWS_MAX_ARTICLES

    try:
        articles = await _collect_market_news_json(limit)
        if articles:
            return articles
    except Exception as e:
        logger.warning(f"[NEWS] 시장뉴스 JSON 경로 실패, HTML 폴백: {e}")

    return await _collect_market_news_html(limit)


async def _collect_market_news_html(limit: int) -> List[Dict[str, Any]]:
    """네이버 금융 시장 뉴스(코스피/코스닥) HTML 크롤링 (폴백).

    Returns:
        [{"title", "source", "url", "published_at", "sentiment", "sentiment_score"}, ...]
    """
    if limit <= 0:
        limit = settings.NEWS_MAX_ARTICLES

    urls = [
        "https://finance.naver.com/news/mainnews.naver",
        "https://finance.naver.com/news/news_list.naver?mode=LSS2D&section_id=101&section_id2=258",
    ]
    articles: List[Dict[str, Any]] = []

    for page_url in urls:
        if len(articles) >= limit:
            break
        try:
            async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
                resp = await client.get(page_url, headers=_HEADERS)

            if resp.status_code != 200:
                continue

            soup = BeautifulSoup(resp.text, "html.parser")

            # 네이버 증권 메인뉴스 / 뉴스리스트 파싱
            for link_tag in soup.select("a[href*='/news/news_read.naver']"):
                title = link_tag.get_text(strip=True)
                if not title or len(title) < 5:
                    continue

                href = link_tag.get("href", "")
                full_url = ("https://finance.naver.com" + href) if href.startswith("/") else href

                score, label = analyze_sentiment_keyword(title)

                articles.append({
                    "title": title,
                    "source": "네이버금융",
                    "url": full_url,
                    "stock_codes": [],  # 시장 뉴스는 특정 종목 없음
                    "sentiment": label,
                    "sentiment_score": score,
                    "published_at": now_kst(),  # 파싱 불가 시 현재 시간
                })

                if len(articles) >= limit:
                    break

        except Exception as e:
            logger.warning(f"[NEWS] 시장뉴스 크롤링 오류: {e}")

    return articles


# ═════════════════════════════════════════════════════════════════════════════
# 5. 뉴스 수집 (NewsAPI.org — 글로벌, 선택적)
# ═════════════════════════════════════════════════════════════════════════════

async def collect_newsapi_news(
    query: str = "한국 주식 OR KOSPI OR KOSDAQ",
    limit: int = 0,
) -> List[Dict[str, Any]]:
    """NewsAPI.org에서 글로벌 뉴스 수집. NEWS_API_KEY 미설정 시 빈 리스트 반환.

    Args:
        query: 검색 키워드
        limit: 최대 수집 개수

    Returns:
        [{"title", "source", "url", "published_at", "sentiment", "sentiment_score"}, ...]
    """
    if not settings.NEWS_API_KEY:
        return []

    if limit <= 0:
        limit = min(settings.NEWS_MAX_ARTICLES, 20)

    api_url = "https://newsapi.org/v2/everything"
    params = {
        "q": query,
        "language": "ko",
        "sortBy": "publishedAt",
        "pageSize": limit,
        "apiKey": settings.NEWS_API_KEY,
    }

    articles: List[Dict[str, Any]] = []

    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            resp = await client.get(api_url, params=params)

        if resp.status_code != 200:
            logger.warning(f"[NEWS] NewsAPI HTTP {resp.status_code}")
            return []

        data = resp.json()
        for item in data.get("articles", []):
            title = item.get("title", "")
            content = item.get("description", "") or ""
            source = (item.get("source") or {}).get("name", "NewsAPI")
            url = item.get("url", "")
            pub_str = item.get("publishedAt", "")

            try:
                published_at = _parse_iso_to_kst(pub_str)
            except Exception:
                published_at = now_kst()

            score, label = analyze_sentiment_keyword(title, content)

            articles.append({
                "title": title,
                "source": source,
                "url": url,
                "stock_codes": [],
                "sentiment": label,
                "sentiment_score": score,
                "published_at": published_at,
            })

    except httpx.TimeoutException:
        logger.warning("[NEWS] NewsAPI 타임아웃")
    except Exception as e:
        logger.error(f"[NEWS] NewsAPI 오류: {e}")

    return articles


# ═════════════════════════════════════════════════════════════════════════════
# 6. 스코어 조회 (캐시 우선 → 실시간 수집 폴백)
# ═════════════════════════════════════════════════════════════════════════════

async def get_stock_news_score(stock_code: str) -> float:
    """종목별 뉴스 감성 점수를 반환한다 (-1.0 ~ +1.0).

    - Redis 캐시 히트 시 즉시 반환 (TTL: NEWS_CACHE_TTL)
    - 캐시 미스 시 실시간 크롤링 → 점수 산출 → 캐시 저장
    - 크롤링 실패 시 0.0 (중립) 반환
    """
    if not settings.NEWS_ENABLED:
        return 0.0

    cache_key = f"news_score:stock:{stock_code}"

    # 1. Redis 캐시 조회
    cached = await get_json(cache_key)
    if cached is not None:
        return float(cached)

    # 1-b. 최근 0건 판정(단기 음성 캐시) — 재크롤링 폭주만 막는다. 점수 캐시가 아니다.
    if await get_json(f"news_score:stock_empty:{stock_code}") is not None:
        return 0.0

    # 2. 실시간 크롤링
    try:
        articles = await collect_stock_news(stock_code)
        await _record_collect_result("stock", len(articles))
        if not articles:
            # '데이터 없음' 은 중립 점수 0.0 으로 캐시하지 않는다(악재 필터 무력화를 가리므로)
            await setex_json(f"news_score:stock_empty:{stock_code}", _EMPTY_NEGATIVE_TTL, 1)
            return 0.0
        score = _compute_aggregate_score(articles)

        # 캐시 저장
        await setex_json(cache_key, settings.NEWS_CACHE_TTL, score)

        # DB 저장 (db_worker 큐 경유)
        await _save_news_to_db(articles)

        return score
    except Exception as e:
        logger.error(f"[NEWS_SCORE] {stock_code} 점수 산출 오류: {e}")
        return 0.0


async def get_market_news_score() -> float:
    """시장 전체 뉴스 감성 점수를 반환한다 (-1.0 ~ +1.0).

    - Redis 캐시 키: "news_score:market"
    - 캐시 미스 시 시장뉴스 + NewsAPI 수집 → 점수 산출
    """
    if not settings.NEWS_ENABLED:
        return 0.0

    cache_key = "news_score:market"

    cached = await get_json(cache_key)
    if cached is not None:
        return float(cached)

    # 최근 0건 판정(단기 음성 캐시) — 재크롤링 폭주만 막는다. 점수 캐시가 아니다.
    if await get_json("news_score:market_empty") is not None:
        return 0.0

    try:
        # 시장 뉴스 + NewsAPI 병렬 수집
        market_task = collect_market_news()
        api_task = collect_newsapi_news()
        market_news, api_news = await asyncio.gather(market_task, api_task)

        all_articles = market_news + api_news
        await _record_collect_result("market", len(all_articles))
        if not all_articles:
            await setex_json("news_score:market_empty", _EMPTY_NEGATIVE_TTL, 1)
            return 0.0  # 데이터 없음 — 중립 점수를 캐시하지 않는다
        score = _compute_aggregate_score(all_articles)

        await setex_json(cache_key, settings.NEWS_CACHE_TTL, score)
        await _save_news_to_db(all_articles)

        return score
    except Exception as e:
        logger.error(f"[NEWS_SCORE] 시장 뉴스 점수 산출 오류: {e}")
        return 0.0


async def get_stock_news_summary(stock_code: str) -> Dict[str, Any]:
    """종목 뉴스 요약 정보 반환 (API 엔드포인트용).

    Returns:
        {
            "stock_code": str,
            "score": float,
            "article_count": int,
            "latest_articles": list,
            "sentiment_distribution": {"positive": int, "negative": int, "neutral": int}
        }
    """
    articles = await collect_stock_news(stock_code)
    score = _compute_aggregate_score(articles)

    dist = {"positive": 0, "negative": 0, "neutral": 0}
    for a in articles:
        label = a.get("sentiment", "neutral")
        if label in dist:
            dist[label] += 1

    # 최신 5개만 반환 (토큰 절약)
    latest = articles[:5] if settings.TOKEN_SAVE_MODE else articles[:10]

    return {
        "stock_code": stock_code,
        "score": score,
        "article_count": len(articles),
        "latest_articles": [
            {
                "title": a["title"],
                "source": a.get("source", ""),
                "sentiment": a.get("sentiment", "neutral"),
                "sentiment_score": a.get("sentiment_score", 0.0),
            }
            for a in latest
        ],
        "sentiment_distribution": dist,
    }


# ═════════════════════════════════════════════════════════════════════════════
# 7. DB 저장 (db_worker 큐 경유)
# ═════════════════════════════════════════════════════════════════════════════

async def _save_news_to_db(articles: List[Dict[str, Any]]) -> None:
    """뉴스 리스트를 db_worker 큐에 일괄 전달.

    CLAUDE.md 3.2: 실시간 계층에서 직접 DB 접근 금지 → 큐 경유.
    """
    if not articles:
        return

    for article in articles:
        try:
            await db_task_queue.put({
                "type": "insert_news",
                "data": {
                    "title": (article.get("title") or "")[:500],
                    "content": article.get("content"),
                    "source": (article.get("source") or "")[:100],
                    "url": article.get("url"),
                    "stock_codes": article.get("stock_codes", []),
                    "sentiment": article.get("sentiment", "neutral"),
                    "sentiment_score": article.get("sentiment_score", 0.0),
                    "published_at": article.get("published_at"),
                },
            })
        except Exception as e:
            warn_throttled(logger, "news_sentiment:558", f"[NEWS_DB] 큐 전달 오류: {e}")


# ═════════════════════════════════════════════════════════════════════════════
# 8. 유틸리티
# ═════════════════════════════════════════════════════════════════════════════

def _parse_naver_date(date_str: str) -> datetime:
    """네이버 금융 날짜 문자열을 datetime으로 변환.

    형식: "2026.03.24 09:15" 또는 "2026.03.24"
    """
    if not date_str:
        return now_kst()

    date_str = date_str.strip()
    for fmt in ("%Y.%m.%d %H:%M", "%Y.%m.%d"):
        try:
            return datetime.strptime(date_str, fmt)
        except ValueError:
            continue

    return now_kst()


# ═════════════════════════════════════════════════════════════════════════════
# 9. 뉴스 수집 스케줄러 함수 (schedulers.py에서 호출)
# ═════════════════════════════════════════════════════════════════════════════

async def news_collect_cycle() -> Dict[str, Any]:
    """한 번의 뉴스 수집 사이클 실행.

    - 시장 뉴스 수집 + 점수 산출
    - 활성 포트폴리오 종목 뉴스 수집 (별도 호출 필요 시 확장)

    Returns:
        {"market_score": float, "articles_collected": int}
    """
    market_score = await get_market_news_score()

    logger.info(
        f"[NEWS_CYCLE] 시장 뉴스 수집 완료: market_score={market_score:.3f}"
    )

    return {
        "market_score": market_score,
    }