extract_articles.py

# extract_articles.py
from __future__ import annotations
import hashlib, time, datetime
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional
from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode

import requests
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter

# ---------------------------
# URL/시간 정규화
# ---------------------------
_TRACKING = {"utm_source","utm_medium","utm_campaign","utm_term","utm_content","fbclid","gclid"}

def canonical_url(u: str) -> str:
    p = urlparse(u)
    q = [(k,v) for k,v in parse_qsl(p.query, keep_blank_values=True) if k not in _TRACKING]
    p = p._replace(
        scheme=p.scheme.lower(),
        netloc=p.netloc.lower(),
        path=p.path if p.path != "/" else "",
        query=urlencode(q, doseq=True),
        fragment=""
    )
    return urlunparse(p)

def url_id(u: str) -> str:
    return hashlib.sha256(u.encode("utf-8")).hexdigest()[:16]

def iso_utc(dt: Any) -> str:
    if isinstance(dt, datetime.datetime):
        return dt.astimezone(datetime.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00","Z")
    return str(dt)

# ---------------------------
# HTTP 세션
# ---------------------------
def make_session(timeout: int = 12, retries: int = 3, backoff: float = 0.5) -> requests.Session:
    s = requests.Session()
    r = Retry(total=retries, connect=retries, read=retries,
              backoff_factor=backoff,
              status_forcelist=(429,500,502,503,504),
              allowed_methods=frozenset({"GET"}))
    ad = HTTPAdapter(max_retries=r)
    s.mount("http://", ad)
    s.mount("https://", ad)
    s.headers.update({"User-Agent": "RedFinFeedBot/1.0 (+contact:sungminwoo.devops@gmail.com)"})
    s.request_timeout = timeout  # custom
    return s

def fetch_html(url: str, s: Optional[requests.Session] = None) -> str:
    s = s or make_session()
    resp = s.get(url, timeout=getattr(s, "request_timeout", 12))
    resp.raise_for_status()
    return resp.text

# ---------------------------
# 본문 추출기
# ---------------------------
def fetch_html(url: str, s=None) -> str:
    s = s or make_session()
    resp = s.get(url, timeout=getattr(s, "request_timeout", 12))
    if resp.status_code >= 400:
        raise requests.HTTPError(f"{resp.status_code} for {url}", response=resp)
    return resp.text

def extract_article(url: str, s: Optional[requests.Session] = None) -> Dict[str, Any]:
    try:
        html = fetch_html(url, s)
    except Exception as e:
        return {"content_text": None, "engine": "http_error", "error": str(e)}
    
    # 1) trafilatura
    try:
        import trafilatura
        text = trafilatura.extract(
            html, include_comments=False, include_tables=False,
            favor_precision=True, url=url
        )
        if text:
            return {"content_text": text, "engine": "trafilatura"}
    except Exception:
        pass
    # 2) readability
    try:
        from readability import Document
        from bs4 import BeautifulSoup
        doc = Document(html)
        html2 = doc.summary(html_partial=True)
        text2 = BeautifulSoup(html2, "html.parser").get_text("\n")
        return {"content_text": text2, "engine": "readability"}
    except Exception:
        return {"content_text": None, "engine": "none"}

# ---------------------------
# 엔트리 확장(피드 1개)
# ---------------------------
def enrich_entries(entries: Iterable[Dict[str, Any]],
                   delay_sec: float = 0.2,
                   s: Optional[requests.Session] = None) -> List[Dict[str, Any]]:
    s = s or make_session()
    out: List[Dict[str, Any]] = []
    for e in entries or []:
        raw_url = e.get("link") or (e.get("links") or [{}])[0].get("href")
        if not raw_url:
            continue
        cu = canonical_url(raw_url)
        art = extract_article(cu, s)
        out.append({
            "url": cu,
            "url_id": url_id(cu),
            "title": e.get("title"),
            "published": e.get("published") or e.get("updated"),
            "published_parsed": e.get("published_parsed") or e.get("updated_parsed"),
            "content_text": art["content_text"],
            "engine": art["engine"],
        })
        if delay_sec > 0:
            time.sleep(delay_sec)
    return out

# ---------------------------
# JSONL I/O
# ---------------------------
def read_jsonl(path: str | Path) -> List[Dict[str, Any]]:
    import json
    p = Path(path)
    rows: List[Dict[str, Any]] = []
    with p.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                rows.append(json.loads(line))
    return rows

def write_jsonl(rows: Iterable[Dict[str, Any]], path: str | Path) -> Path:
    import json
    p = Path(path)
    with p.open("w", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    return p

# ---------------------------
# feeds_result.jsonl → articles.jsonl
# ---------------------------
def process_feeds_jsonl(in_jsonl: str | Path,
                        out_jsonl: str | Path,
                        delay_sec: float = 0.2) -> Path:
    feeds = read_jsonl(in_jsonl)
    s = make_session()
    all_rows: List[Dict[str, Any]] = []
    for feed in feeds:
        entries = feed.get("entries") or []
        enriched = enrich_entries(entries, delay_sec=delay_sec, s=s)
        # feed 메타 포함
        feed_meta = {
            "feed_name": feed.get("name"),
            "feed_url": feed.get("url"),
            "group": feed.get("group"),
            "fetched_at": feed.get("fetched_at") or feed.get("http", {}).get("fetched_at")
        }
        for r in enriched:
            r.update(feed_meta)
        all_rows.extend(enriched)
    return write_jsonl(all_rows, out_jsonl)

if __name__ == "__main__":
    # 예: python extract_articles.py feeds_result.jsonl articles.jsonl
    import sys
    _in = sys.argv[1] if len(sys.argv) > 1 else "feeds_result.jsonl"
    _out = sys.argv[2] if len(sys.argv) > 2 else "articles.jsonl"
    p = process_feeds_jsonl(_in, _out)
    print(f"[OK] saved -> {p}")