diff --git a/.env.example b/.env.example index 5a9b855..0424dd8 100644 --- a/.env.example +++ b/.env.example @@ -11,6 +11,11 @@ TELEGRAM_CHAT_ID= # FRED (선택) — 경제지표 실제 발표값 FRED_API_KEY= +# SearXNG (선택) — 설정하면 RSS 외에 메타검색 뉴스도 수집 (인스턴스에 json 포맷 활성화 필요) +SEARXNG_BASE_URL= +# 검색어 CSV. 비우면 기본값(us2kr/collectors/searx.py DEFAULT_QUERIES) +US2KR_SEARXNG_QUERIES= + # 실행 정책 US2KR_RUN_OFFSET_MINUTES=75 US2KR_CATCHUP_WINDOW_HOURS=3 diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index b906896..793a13b 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -192,6 +192,70 @@ def test_build_payload_history_failure_is_a_gap(env): assert any(g.startswith("history:") for g in payload.data_gaps) +def test_build_payload_skips_searx_when_not_configured(env): + def boom(*args, **kwargs): + raise AssertionError("fetch_searx_news must not be called without SEARXNG_BASE_URL") + + env.monkeypatch.setattr(pipeline, "fetch_searx_news", boom) + payload = pipeline.build_payload(SESSION, env.settings, env.clock, env.store) + assert payload.data_gaps == [] + + +def test_build_payload_merges_searx_news(monkeypatch, tmp_path): + env = Env( + monkeypatch, + tmp_path, + SEARXNG_BASE_URL="http://searx.local:8888", + US2KR_SEARXNG_QUERIES="markets, fed", + US2KR_NEWS_MAX_ITEMS="3", + ) + rss = sample_news() # 1 item, published 17:00 ET + searx_items = [ + NewsItem( # duplicate of the RSS story by link; same timestamp -> RSS copy wins the tie + title="Stocks rally (search copy)", + link="https://example.com/1#utm", + source="Reuters", + published=datetime(2026, 8, 25, 17, 0, tzinfo=ET), + ), + NewsItem( + title="Chip stocks jump", + link="https://example.com/chips", + source="Barron's", + published=datetime(2026, 8, 25, 18, 30, tzinfo=ET), + ), + ] + + def fetch_searx_news(base_url, since, queries=None, **kwargs): + env.calls["searxng"] = env.calls.get("searxng", 0) + 1 + assert base_url == "http://searx.local:8888" + assert queries == ["markets", "fed"] + assert since == env.news_since + return searx_items + + env.monkeypatch.setattr(pipeline, "fetch_searx_news", fetch_searx_news) + payload = pipeline.build_payload(SESSION, env.settings, env.clock, env.store) + + assert env.calls["searxng"] == 1 + assert [n.title for n in payload.news] == ["Chip stocks jump", "Stocks rally"] + assert payload.data_gaps == [] + # the archived news blob is the merged list + assert "Chip stocks jump" in env.store.raw[(SESSION, "news")] + assert rss[0].title in env.store.raw[(SESSION, "news")] + + +def test_build_payload_searx_failure_keeps_rss_news(monkeypatch, tmp_path): + env = Env(monkeypatch, tmp_path, SEARXNG_BASE_URL="http://searx.local:8888") + + def boom(base_url, since, queries=None, **kwargs): + raise RuntimeError("all searxng queries failed: markets: HTTPError") + + env.monkeypatch.setattr(pipeline, "fetch_searx_news", boom) + payload = pipeline.build_payload(SESSION, env.settings, env.clock, env.store) + + assert [n.title for n in payload.news] == ["Stocks rally"] + assert any(g.startswith("searxng: RuntimeError") for g in payload.data_gaps) + + # -------------------------------------------------------------------------------- run_briefing def test_run_briefing_happy_path_saves_then_delivers(env): result = pipeline.run_briefing(SESSION, env.settings) diff --git a/tests/test_searx.py b/tests/test_searx.py new file mode 100644 index 0000000..5a38c1b --- /dev/null +++ b/tests/test_searx.py @@ -0,0 +1,151 @@ +"""Tests for us2kr.collectors.searx (no network: requests.get is mocked).""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta + +import pytest +import requests + +from us2kr.collectors import searx +from us2kr.collectors.searx import _parse_published, fetch_searx_news + +SINCE = datetime(2026, 8, 25, 12, 0, tzinfo=UTC) + + +def _result(title: str, url: str, published: str | None = None, **extra) -> dict: + result = {"title": title, "url": url, "content": extra.pop("content", "")} + if published is not None: + result["publishedDate"] = published + result.update(extra) + return result + + +class _Resp: + def __init__(self, results: list | None, status: int = 200, body: str | None = None): + self.status_code = status + self._body = body if body is not None else json.dumps({"results": results}) + + def raise_for_status(self) -> None: + if self.status_code >= 400: + raise requests.HTTPError(f"{self.status_code} error") + + def json(self): + return json.loads(self._body) + + +def _iso(dt: datetime) -> str: + return dt.isoformat() + + +def test_parse_published_variants(): + assert _parse_published("2026-08-25T14:00:00+00:00") == datetime(2026, 8, 25, 14, 0, tzinfo=UTC) + assert _parse_published("2026-08-25T14:00:00Z") == datetime(2026, 8, 25, 14, 0, tzinfo=UTC) + # naive -> assumed UTC + assert _parse_published("2026-08-25T14:00:00") == datetime(2026, 8, 25, 14, 0, tzinfo=UTC) + assert _parse_published(None) is None + assert _parse_published("") is None + assert _parse_published("yesterday-ish") is None + assert _parse_published(1234) is None + + +def test_filters_maps_and_floors_undated(mocker): + fresh = SINCE + timedelta(hours=2) + results = [ + _result("Fed pauses", "https://n.example/fed", _iso(fresh), content="
Held & steady
", source="Reuters"), + _result("Old story", "https://n.example/old", _iso(SINCE - timedelta(hours=3))), + _result("Undated wrap", "https://n.example/wrap", None, engine="bing news"), + _result("", "https://n.example/notitle", _iso(fresh)), + _result("No link", "", _iso(fresh)), + ] + get = mocker.patch("us2kr.collectors.searx.requests.get", return_value=_Resp(results)) + + items = fetch_searx_news("http://searx.local:8888", since=SINCE, queries=["markets"]) + + assert [i.title for i in items] == ["Fed pauses", "Undated wrap"] + assert items[0].summary == "Held & steady" + assert items[0].source == "Reuters" + assert items[0].published == fresh + # undated: kept, timestamped at the window floor, engine used as source fallback + assert items[1].published == SINCE + assert items[1].source == "bing news" + + params = get.call_args.kwargs["params"] + assert params["format"] == "json" + assert params["categories"] == "news" + assert params["time_range"] == "day" + assert get.call_args.args[0] == "http://searx.local:8888/search" + + +def test_naive_since_is_treated_as_utc(mocker): + fresh = SINCE + timedelta(hours=1) + mocker.patch( + "us2kr.collectors.searx.requests.get", + return_value=_Resp([_result("A", "https://n.example/a", _iso(fresh))]), + ) + items = fetch_searx_news("http://searx.local:8888", since=SINCE.replace(tzinfo=None), queries=["q"]) + assert len(items) == 1 + assert items[0].published.tzinfo is not None + + +def test_dedup_across_queries_and_max_per_query(mocker): + fresh = SINCE + timedelta(hours=1) + per_query = { + "q1": _Resp( + [ + _result("Story A", "https://n.example/a/", _iso(fresh)), + _result("Story B", "https://n.example/b", _iso(fresh)), + ] + ), + # same article as q1's Story A modulo trailing slash / fragment + "q2": _Resp([_result("Story A again", "https://n.example/a#frag", _iso(fresh))]), + } + mocker.patch( + "us2kr.collectors.searx.requests.get", + side_effect=lambda url, params, **kw: per_query[params["q"]], + ) + items = fetch_searx_news("http://searx.local:8888", since=SINCE, queries=["q1", "q2"], max_per_query=1) + # max_per_query=1 keeps only Story A from q1; q2's duplicate link is dropped + assert [i.title for i in items] == ["Story A"] + + +def test_partial_failure_is_tolerated(mocker, caplog): + fresh = SINCE + timedelta(hours=1) + per_query = { + "bad": _Resp(None, status=500), + "good": _Resp([_result("Kept", "https://n.example/k", _iso(fresh))]), + } + mocker.patch( + "us2kr.collectors.searx.requests.get", + side_effect=lambda url, params, **kw: per_query[params["q"]], + ) + items = fetch_searx_news("http://searx.local:8888", since=SINCE, queries=["bad", "good"]) + assert [i.title for i in items] == ["Kept"] + + +def test_all_queries_failing_raises(mocker): + mocker.patch("us2kr.collectors.searx.requests.get", return_value=_Resp(None, status=502)) + with pytest.raises(RuntimeError, match="all searxng queries failed"): + fetch_searx_news("http://searx.local:8888", since=SINCE, queries=["a", "b"]) + + +def test_malformed_response_counts_as_failure(mocker): + mocker.patch( + "us2kr.collectors.searx.requests.get", + return_value=_Resp(None, body=json.dumps({"unexpected": True})), + ) + with pytest.raises(RuntimeError, match="all searxng queries failed"): + fetch_searx_news("http://searx.local:8888", since=SINCE, queries=["only"]) + + +def test_default_queries_used_when_none(mocker): + calls: list[str] = [] + + def record(url, params, **kw): + calls.append(params["q"]) + return _Resp([]) + + mocker.patch("us2kr.collectors.searx.requests.get", side_effect=record) + assert fetch_searx_news("http://searx.local:8888", since=SINCE) == [] + assert calls == searx.DEFAULT_QUERIES diff --git a/us2kr/collectors/news.py b/us2kr/collectors/news.py index fd621c3..ffb76f2 100644 --- a/us2kr/collectors/news.py +++ b/us2kr/collectors/news.py @@ -125,6 +125,23 @@ def _parse_feed(source: str, content: bytes, since: datetime) -> list[NewsItem]: return items +def dedup_news(items: list[NewsItem]) -> list[NewsItem]: + """Newest first, deduplicated by normalised title and by link (fragment/trailing slash ignored).""" + seen_titles: set[str] = set() + seen_links: set[str] = set() + unique: list[NewsItem] = [] + # Stable sort newest first before dedup so the freshest duplicate is the one kept. + for item in sorted(items, key=lambda n: n.published, reverse=True): + tkey = normalize_title(item.title) + lkey = item.link.split("#", 1)[0].rstrip("/").lower() + if tkey in seen_titles or lkey in seen_links: + continue + seen_titles.add(tkey) + seen_links.add(lkey) + unique.append(item) + return unique + + def fetch_news(since: datetime, max_items: int = 40) -> list[NewsItem]: """Collect headlines published at/after `since` from all feeds, deduplicated, newest first. @@ -151,18 +168,7 @@ def fetch_news(since: datetime, max_items: int = 40) -> list[NewsItem]: if failures and len(failures) == len(FEEDS): raise RuntimeError("all news feeds failed: " + "; ".join(failures)) - seen_titles: set[str] = set() - seen_links: set[str] = set() - unique: list[NewsItem] = [] - # Stable sort newest first before dedup so the freshest duplicate is the one kept. - for item in sorted(collected, key=lambda n: n.published, reverse=True): - tkey = normalize_title(item.title) - lkey = item.link.split("#", 1)[0].rstrip("/").lower() - if tkey in seen_titles or lkey in seen_links: - continue - seen_titles.add(tkey) - seen_links.add(lkey) - unique.append(item) + unique = dedup_news(collected) log.info( "news: %d raw -> %d unique, returning %d (feeds failed: %d/%d)", diff --git a/us2kr/collectors/searx.py b/us2kr/collectors/searx.py new file mode 100644 index 0000000..19e45d4 --- /dev/null +++ b/us2kr/collectors/searx.py @@ -0,0 +1,121 @@ +"""SearXNG news collector: supplements the RSS feeds with meta-search results. + +Used only when ``SEARXNG_BASE_URL`` is configured. Each query is fetched independently so that +one failing query never spoils the others. Only if *every* query fails does +`fetch_searx_news` raise. The instance must have the JSON output format enabled +(searxng settings.yml: ``search.formats`` includes ``json``). +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime +from urllib.parse import urljoin + +import requests + +from us2kr.collectors.news import USER_AGENT, strip_html +from us2kr.models import NewsItem + +log = logging.getLogger(__name__) + +SEARX_TIMEOUT_SECONDS = 15.0 +MAX_PER_QUERY = 15 + +# Complements the fixed RSS feeds: broad market wrap + macro + the Korea/semis angle the +# briefing cares about. Override with US2KR_SEARXNG_QUERIES (CSV). +DEFAULT_QUERIES: list[str] = [ + "US stock market", + "Federal Reserve", + "semiconductor stocks", +] + + +def _parse_published(raw: object) -> datetime | None: + """ISO-8601 publishedDate -> tz-aware UTC datetime, or None when absent/garbled.""" + if not raw or not isinstance(raw, str): + return None + try: + parsed = datetime.fromisoformat(raw) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + return parsed.astimezone(UTC) + + +def _fetch_query(base_url: str, query: str) -> list[dict]: + resp = requests.get( + urljoin(base_url.rstrip("/") + "/", "search"), + params={"q": query, "format": "json", "categories": "news", "time_range": "day"}, + timeout=SEARX_TIMEOUT_SECONDS, + headers={"User-Agent": USER_AGENT}, + ) + resp.raise_for_status() + results = resp.json().get("results") + if not isinstance(results, list): # malformed server payload, not a caller type error + raise ValueError("malformed SearXNG response: no results list") # noqa: TRY004 + return results + + +def fetch_searx_news( + base_url: str, + since: datetime, + queries: list[str] | None = None, + max_per_query: int = MAX_PER_QUERY, +) -> list[NewsItem]: + """Search the news category for each query, returning items published at/after `since`. + + Results without a publishedDate are kept (time_range=day already bounds their age) but + timestamped at `since`, so they sort last and never win deduplication against a dated copy. + Raises RuntimeError only when every query failed. + """ + if since.tzinfo is None: + since = since.replace(tzinfo=UTC) + else: + since = since.astimezone(UTC) + + queries = queries or DEFAULT_QUERIES + collected: list[NewsItem] = [] + failures: list[str] = [] + seen_links: set[str] = set() # the same article routinely matches several queries + for query in queries: + try: + results = _fetch_query(base_url, query) + except Exception as exc: # noqa: BLE001 - per-query isolation + log.warning("searxng query %r failed: %s: %s", query, type(exc).__name__, str(exc)[:200]) + failures.append(f"{query}: {type(exc).__name__}") + continue + kept = undated = 0 + for result in results: + title = (result.get("title") or "").strip() + link = (result.get("url") or "").strip() + if not title or not link: + continue + lkey = link.split("#", 1)[0].rstrip("/").lower() + if lkey in seen_links: + continue + published = _parse_published(result.get("publishedDate")) + if published is None: + undated += 1 + published = since + elif published < since: + continue + seen_links.add(lkey) + collected.append( + NewsItem( + title=title, + summary=strip_html(result.get("content") or ""), + link=link, + source=result.get("source") or result.get("engine") or "SearXNG", + published=published, + ) + ) + kept += 1 + if kept >= max_per_query: + break + log.info("searxng query %r: %d kept (%d undated) of %d results", query, kept, undated, len(results)) + + if failures and len(failures) == len(queries): + raise RuntimeError("all searxng queries failed: " + "; ".join(failures)) + return collected diff --git a/us2kr/config.py b/us2kr/config.py index 0ba0790..8f818d6 100644 --- a/us2kr/config.py +++ b/us2kr/config.py @@ -38,6 +38,13 @@ class Settings(BaseSettings): # --- Optional data keys --- fred_api_key: str | None = Field(default=None, alias="FRED_API_KEY") + # --- SearXNG (optional news search; instance must enable the json output format) --- + searxng_base_url: str | None = Field(default=None, alias="SEARXNG_BASE_URL") + # Empty -> collector defaults (us2kr.collectors.searx.DEFAULT_QUERIES). + searxng_queries: Annotated[list[str], NoDecode] = Field( + default_factory=list, alias="US2KR_SEARXNG_QUERIES" + ) + # --- Paths --- data_dir: Path = Field(default=Path("data"), alias="US2KR_DATA_DIR") reports_dir: Path = Field(default=Path("reports"), alias="US2KR_REPORTS_DIR") @@ -60,7 +67,7 @@ class Settings(BaseSettings): log_level: str = Field(default="INFO", alias="US2KR_LOG_LEVEL") - @field_validator("minimax_fallback_models", "krx_extra_holidays", mode="before") + @field_validator("minimax_fallback_models", "krx_extra_holidays", "searxng_queries", mode="before") @classmethod def _split_csv(cls, v): if isinstance(v, str): diff --git a/us2kr/pipeline.py b/us2kr/pipeline.py index f083824..daaf2b2 100644 --- a/us2kr/pipeline.py +++ b/us2kr/pipeline.py @@ -16,8 +16,9 @@ from pydantic import BaseModel from us2kr.collectors.base import run_safely from us2kr.collectors.events import fetch_earnings, fetch_events -from us2kr.collectors.news import fetch_news +from us2kr.collectors.news import dedup_news, fetch_news from us2kr.collectors.prices import fetch_quotes +from us2kr.collectors.searx import fetch_searx_news from us2kr.config import KST, Settings from us2kr.deliver import build_deliverers from us2kr.llm.base import LLMError @@ -79,6 +80,20 @@ def build_payload(session: date, settings: Settings, clock: MarketClock, store: news = run_safely( "news", lambda: fetch_news(since=news_since, max_items=settings.news_max_items), [], gaps ) + if settings.searxng_base_url: + searx_news = run_safely( + "searxng", + lambda: fetch_searx_news( + settings.searxng_base_url, # type: ignore[arg-type] + since=news_since, + queries=settings.searxng_queries or None, + ), + [], + gaps, + ) + if searx_news: + # RSS first so it wins timestamp ties in dedup; cap stays at news_max_items overall. + news = dedup_news([*news, *searx_news])[: settings.news_max_items] events = run_safely("events", lambda: fetch_events(session, settings.fred_api_key), [], gaps) tickers = earnings_tickers() earnings = run_safely("earnings", lambda: fetch_earnings(session, tickers), [], gaps)