diff --git a/docs/sources/available-enrichers.mdx b/docs/sources/available-enrichers.mdx index 93abe656..978a03cf 100644 --- a/docs/sources/available-enrichers.mdx +++ b/docs/sources/available-enrichers.mdx @@ -113,6 +113,9 @@ Tools/APIs: SIRENE (SireneTool) — see [INSEE Sirene API](https://api.insee.fr/ **org_to_asn**: Find ASNs associated with an organization name. Tools/Pivots: [asnmap](https://github.com/projectdiscovery/asnmap) (CLI), [jq](https://jqlang.github.io/jq/) (CLI) +**org_to_news**: Find news coverage of an organization, as Website nodes with the article title and snippet. +Tools/APIs: [Serply API](https://serply.io/docs) + ### Phone **phone_to_infos**: Probe phone footprint across services (demo modules) and normalize number. Tools/APIs: ignorant modules (Amazon, Snapchat, Instagram), [httpx](https://github.com/projectdiscovery/httpx) diff --git a/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py b/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py new file mode 100644 index 00000000..a402c3ee --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/organization/to_news.py @@ -0,0 +1,308 @@ +import os +from typing import Any, Dict, List, Optional + +import requests + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_core.core.vault import VaultProtocol +from flowsint_enrichers.registry import flowsint_enricher +from flowsint_types.organization import Organization +from flowsint_types.website import Website + +SEARCH_ENDPOINT = "https://api.serply.io/v1/search/" +# `num` is an approximate cap of about ten rows per call, not an exact count, so +# the window size is fixed and surplus rows are trimmed against max_results. +PAGE_SIZE = 10 +DEFAULT_MAX_RESULTS = 20 +# The news vertical is selected with Google's own `tbm`, off the same endpoint. +NEWS_VERTICAL = "nws" +# Recency maps onto Google's `tbs=qdr:`. +RECENCY = {"d": "qdr:d", "w": "qdr:w", "m": "qdr:m", "y": "qdr:y"} +DEFAULT_LANGUAGE = "en" +DEFAULT_REGION = "us" + + +@flowsint_enricher +class OrgToNewsEnricher(Enricher): + """[Serply] Find news coverage of an organization. + + Runs the organization name as an exact-phrase query against the news + vertical and emits one Website per article. + Expects a `SERPLY_API_KEY` vault secret. + """ + + # Define types as class attributes - base class handles schema generation automatically + InputType = Organization + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault: Optional[VaultProtocol] = None, + params: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + **kwargs, + ) + + @classmethod + def name(cls) -> str: + return "org_to_news" + + @classmethod + def category(cls) -> str: + return "Organization" + + @classmethod + def key(cls) -> str: + return "name" + + @classmethod + def required_params(cls) -> bool: + return True + + @classmethod + def get_params_schema(cls) -> List[Dict[str, Any]]: + """Declare required parameters for this enricher""" + return [ + { + "name": "SERPLY_API_KEY", + "type": "vaultSecret", + "description": "Your Serply API key, from serply.io.", + "required": True, + }, + { + "name": "time_range", + "type": "select", + "description": "Only keep coverage published within this window.", + "required": False, + "default": "any", + "options": [ + {"label": "Any time", "value": "any"}, + {"label": "Past 24 hours", "value": "d"}, + {"label": "Past week", "value": "w"}, + {"label": "Past month", "value": "m"}, + {"label": "Past year", "value": "y"}, + ], + }, + { + "name": "language", + "type": "string", + "description": f"Language of the coverage, as a two-letter code. Default: {DEFAULT_LANGUAGE}", + "required": False, + "default": DEFAULT_LANGUAGE, + }, + { + "name": "region", + "type": "string", + "description": f"Market the coverage is ranked for, as a two-letter country code. Default: {DEFAULT_REGION}", + "required": False, + "default": DEFAULT_REGION, + }, + { + "name": "max_results", + "type": "number", + "description": f"Maximum articles per organization. Default: {DEFAULT_MAX_RESULTS}", + "required": False, + }, + ] + + def _max_results(self) -> int: + limit = self.params.get("max_results") + if limit is None: + return DEFAULT_MAX_RESULTS + try: + return max(1, int(limit)) + except (TypeError, ValueError): + Logger.warn( + self.sketch_id, + { + "message": f"(OrgToNews) Invalid max_results '{limit}', falling back to {DEFAULT_MAX_RESULTS}." + }, + ) + return DEFAULT_MAX_RESULTS + + def _base_query_params(self, query: str) -> Dict[str, str | int]: + """Build the parameters shared by every window of one scan. + + Language and market are pinned on purpose. Left unset, the index infers + them from where the call originates, which for a server means the + vertical answers an English query with articles in whatever language + that host looks like it wants. + """ + query_params: Dict[str, str | int] = { + "q": query, + "num": PAGE_SIZE, + "tbm": NEWS_VERTICAL, + "hl": (self.params.get("language") or DEFAULT_LANGUAGE).strip(), + "gl": (self.params.get("region") or DEFAULT_REGION).strip(), + } + + time_range = (self.params.get("time_range") or "").strip() + if time_range in RECENCY: + query_params["tbs"] = RECENCY[time_range] + elif time_range and time_range != "any": + Logger.warn( + self.sketch_id, + { + "message": f"(OrgToNews) Unknown time_range '{time_range}', searching all of time instead." + }, + ) + + return query_params + + def _fetch_window( + self, query: str, start: int, api_key: str + ) -> List[Dict[str, Any]]: + """Fetch one result window. `start` is the only offset the API honours.""" + query_params = self._base_query_params(query) + query_params["start"] = start + + api_request = requests.get( + SEARCH_ENDPOINT, + params=query_params, + headers={ + "X-Api-Key": api_key, + "Accept": "application/json", + "User-Agent": "FlowsInt-Enricher", + }, + timeout=30, + ) + + if api_request.status_code != 200: + Logger.error( + self.sketch_id, + { + "message": f"(OrgToNews) Search failed for '{query}' (HTTP {api_request.status_code}): {api_request.text}" + }, + ) + return [] + + return api_request.json().get("results") or [] + + async def scan(self, data: List[InputType]) -> List[OutputType]: + results: List[OutputType] = [] + + api_key = self.get_secret("SERPLY_API_KEY", os.getenv("SERPLY_API_KEY")) + max_results = self._max_results() + + for org in data: + org_name = str(org.name).strip() if org.name else "" + if not org_name: + Logger.warn( + self.sketch_id, + { + "message": "(OrgToNews) Skipping an organization with no name to search on." + }, + ) + continue + + try: + # Quoting keeps a multi-word name together, so a two-word + # company does not match every article carrying both words. + query = f'"{org_name}"' + articles: List[Website] = [] + seen_links: set[str] = set() + start = 0 + + while len(articles) < max_results: + rows = self._fetch_window(query, start, api_key) + if not rows: + break + + before = len(articles) + for row in rows: + link = row.get("link") + if not link or link in seen_links: + continue + seen_links.add(link) + + try: + article = Website( + url=link, + title=row.get("title"), + description=row.get("description"), + ) + except Exception as e: + # The vertical occasionally returns links that are + # not addressable as an HTTP URL; skip those rows. + Logger.warn( + self.sketch_id, + { + "message": f"(OrgToNews) Skipping unusable result '{link}': {e}" + }, + ) + continue + + # Carry the source name through to postprocess for graph + # wiring, the way domain_to_dns threads its source domain. + setattr(article, "_source_org_name", org_name) + articles.append(article) + + if len(articles) >= max_results: + break + + # A window that adds nothing new means the result set is + # exhausted (or the offset stopped moving), so stop paging + # rather than spend a credit per duplicate window. + if len(articles) == before: + break + start += PAGE_SIZE + + if not articles: + Logger.info( + self.sketch_id, + {"message": f"(OrgToNews) No coverage found for '{org_name}'."}, + ) + + results.extend(articles) + + except Exception as e: + Logger.error( + self.sketch_id, + { + "message": f"(OrgToNews) Exception while querying {org_name}: {e}" + }, + ) + + return results + + def postprocess( + self, results: List[OutputType], input_data: Optional[List[InputType]] = None + ) -> List[OutputType]: + if not self._graph_service: + return results + + for article in results: + source_org_name = getattr(article, "_source_org_name", None) + if not source_org_name: + continue + + org = Organization(name=source_org_name) + self.create_node(org) + self.create_node(article) + + # An article covering an organization mentions it; it is not a site + # the organization owns, so this is not the HAS_WEBSITE edge. + self.create_relationship(org, article, "MENTIONED_IN") + self.log_graph_message( + f"(OrgToNews) {source_org_name} -> {str(article.url)}" + ) + + # Clean up the temporary attribute used to thread context. + delattr(article, "_source_org_name") + + return results + + +# Make types available at module level for easy access +InputType = OrgToNewsEnricher.InputType +OutputType = OrgToNewsEnricher.OutputType diff --git a/flowsint-enrichers/tests/enrichers/test_org_to_news.py b/flowsint-enrichers/tests/enrichers/test_org_to_news.py new file mode 100644 index 00000000..bded399a --- /dev/null +++ b/flowsint-enrichers/tests/enrichers/test_org_to_news.py @@ -0,0 +1,295 @@ +import pytest + +from flowsint_enrichers import ENRICHER_REGISTRY +from flowsint_enrichers.organization.to_news import ( + PAGE_SIZE, + OrgToNewsEnricher, +) +from flowsint_types.organization import Organization +from flowsint_types.website import Website + + +# --------------------------------------------------------------------------- +# Registry wiring +# --------------------------------------------------------------------------- +def test_org_to_news_is_registered(): + enricher = ENRICHER_REGISTRY.get_enricher("org_to_news", "123", "123") + assert enricher.name() == "org_to_news" + + +def test_org_to_news_metadata(): + assert OrgToNewsEnricher.category() == "Organization" + assert OrgToNewsEnricher.key() == "name" + assert OrgToNewsEnricher.input_schema()["type"] == "Organization" + assert OrgToNewsEnricher.output_schema()["type"] == "Website" + + +def test_api_key_is_declared_as_a_required_vault_secret(): + schema = {p["name"]: p for p in OrgToNewsEnricher.get_params_schema()} + assert schema["SERPLY_API_KEY"]["type"] == "vaultSecret" + assert schema["SERPLY_API_KEY"]["required"] is True + assert schema["time_range"]["required"] is False + assert [o["value"] for o in schema["time_range"]["options"]] == [ + "any", + "d", + "w", + "m", + "y", + ] + + +def test_enricher_declares_that_it_requires_params(): + assert OrgToNewsEnricher.required_params() is True + + +# --------------------------------------------------------------------------- +# scan() - HTTP layer mocked, no API key and no network needed +# --------------------------------------------------------------------------- +class _FakeResponse: + def __init__(self, payload, status_code=200): + self._payload = payload + self.status_code = status_code + self.text = str(payload) + + def json(self): + return self._payload + + +class _FakeRequests: + """Stands in for the `requests` module inside the enricher.""" + + def __init__(self, windows, status_code=200): + self._windows = windows + self._status_code = status_code + self.calls = [] + + def get(self, url, params=None, headers=None, timeout=None): + self.calls.append({"url": url, "params": params, "headers": headers}) + index = len(self.calls) - 1 + rows = self._windows[index] if index < len(self._windows) else [] + return _FakeResponse({"results": rows}, self._status_code) + + +def _row(position, link, title="Title", description="Description"): + return { + "position": position, + "title": title, + "link": link, + "description": description, + } + + +def _enricher(fake, monkeypatch, params=None): + monkeypatch.setattr( + "flowsint_enrichers.organization.to_news.requests", fake, raising=True + ) + return OrgToNewsEnricher( + sketch_id="s", + scan_id="t", + graph_service=None, + params={"SERPLY_API_KEY": "test-key", **(params or {})}, + ) + + +@pytest.mark.asyncio +async def test_scan_maps_articles_to_websites(monkeypatch): + fake = _FakeRequests( + [[_row(1, "https://news.example.com/acme-raises", "Acme raises", "Snippet")]] + ) + enricher = _enricher(fake, monkeypatch) + + results = await enricher.scan([Organization(name="Acme Corp")]) + + assert len(results) == 1 + article = results[0] + assert isinstance(article, Website) + assert str(article.url) == "https://news.example.com/acme-raises" + assert article.title == "Acme raises" + assert article.description == "Snippet" + + params = fake.calls[0]["params"] + # Quoting keeps a multi-word name together instead of matching both words + # anywhere on the page. + assert params["q"] == '"Acme Corp"' + # The news vertical is selected with Google's own `tbm`. + assert params["tbm"] == "nws" + assert fake.calls[0]["headers"]["X-Api-Key"] == "test-key" + + +@pytest.mark.asyncio +async def test_language_and_region_are_pinned_by_default(monkeypatch): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch) + + await enricher.scan([Organization(name="Acme Corp")]) + + # Left unset these are inferred from where the call originates, which for a + # server answers an English query with articles in another language. + assert fake.calls[0]["params"]["hl"] == "en" + assert fake.calls[0]["params"]["gl"] == "us" + + +@pytest.mark.asyncio +async def test_language_and_region_are_overridable(monkeypatch): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch, {"language": "fr", "region": "fr"}) + + await enricher.scan([Organization(name="Acme Corp")]) + + assert fake.calls[0]["params"]["hl"] == "fr" + assert fake.calls[0]["params"]["gl"] == "fr" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "time_range,expected", + [("d", "qdr:d"), ("w", "qdr:w"), ("m", "qdr:m"), ("y", "qdr:y")], +) +async def test_time_range_maps_onto_the_recency_filter( + monkeypatch, time_range, expected +): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch, {"time_range": time_range}) + + await enricher.scan([Organization(name="Acme Corp")]) + + assert fake.calls[0]["params"]["tbs"] == expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize("time_range", ["any", "", "since-tuesday"]) +async def test_unfiltered_time_ranges_send_no_recency_filter(monkeypatch, time_range): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch, {"time_range": time_range}) + + await enricher.scan([Organization(name="Acme Corp")]) + + # An unrecognised value searches all of time rather than failing the scan. + assert "tbs" not in fake.calls[0]["params"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("name", ["", " ", None]) +async def test_scan_skips_an_organization_without_a_name(monkeypatch, name): + fake = _FakeRequests([[_row(1, "https://news.example.com/a")]]) + enricher = _enricher(fake, monkeypatch) + + # `Organization.name` is typed Any, so it can arrive empty or unset from an + # upstream pivot. Nothing to search on, so no credit is spent. + assert await enricher.scan([Organization(name=name)]) == [] + assert fake.calls == [] + + +@pytest.mark.asyncio +async def test_scan_pages_with_start_and_dedupes_links(monkeypatch): + first = [_row(1, "https://a.example.com/"), _row(2, "https://b.example.com/")] + second = [_row(1, "https://a.example.com/"), _row(2, "https://c.example.com/")] + fake = _FakeRequests([first, second]) + enricher = _enricher(fake, monkeypatch, {"max_results": "4"}) + + results = await enricher.scan([Organization(name="Acme Corp")]) + + assert [str(a.url) for a in results] == [ + "https://a.example.com/", + "https://b.example.com/", + "https://c.example.com/", + ] + # `start` is the only offset the API honours, so it is what advances. + assert [c["params"]["start"] for c in fake.calls[:2]] == [0, PAGE_SIZE] + # Every window stays on the news vertical, not just the first. + assert all(c["params"]["tbm"] == "nws" for c in fake.calls) + + +@pytest.mark.asyncio +async def test_scan_stops_when_a_window_adds_nothing_new(monkeypatch): + repeated = [_row(1, "https://a.example.com/")] + fake = _FakeRequests([repeated] * 10) + enricher = _enricher(fake, monkeypatch, {"max_results": "50"}) + + results = await enricher.scan([Organization(name="Acme Corp")]) + + # One window of duplicates ends the loop instead of refetching forever. + assert len(results) == 1 + assert len(fake.calls) == 2 + + +@pytest.mark.asyncio +async def test_scan_trims_surplus_rows_to_max_results(monkeypatch): + rows = [_row(i, f"https://p{i}.example.com/") for i in range(1, 6)] + fake = _FakeRequests([rows]) + enricher = _enricher(fake, monkeypatch, {"max_results": "2"}) + + results = await enricher.scan([Organization(name="Acme Corp")]) + assert len(results) == 2 + + +@pytest.mark.asyncio +async def test_scan_returns_nothing_on_api_error(monkeypatch): + fake = _FakeRequests([[]], status_code=403) + enricher = _enricher(fake, monkeypatch) + + assert await enricher.scan([Organization(name="Acme Corp")]) == [] + + +@pytest.mark.asyncio +async def test_scan_skips_rows_without_a_usable_link(monkeypatch): + rows = [_row(1, ""), _row(2, "not-a-url"), _row(3, "https://ok.example.com/")] + fake = _FakeRequests([rows]) + enricher = _enricher(fake, monkeypatch) + + results = await enricher.scan([Organization(name="Acme Corp")]) + assert [str(a.url) for a in results] == ["https://ok.example.com/"] + + +# --------------------------------------------------------------------------- +# postprocess() - graph service mocked +# --------------------------------------------------------------------------- +class _FakeGraphService: + def __init__(self): + self.nodes = [] + self.relationships = [] + self.messages = [] + + def create_node_from_flowsint_type(self, node_obj): + self.nodes.append(node_obj) + + def create_relationship(self, from_obj, to_obj, rel_label): + self.relationships.append((from_obj, to_obj, rel_label)) + + def log_graph_message(self, message): + self.messages.append(message) + + +def _article(url, org_name="Acme Corp"): + article = Website(url=url) + setattr(article, "_source_org_name", org_name) + return article + + +def test_postprocess_relates_each_article_back_to_the_organization(): + graph = _FakeGraphService() + enricher = OrgToNewsEnricher(sketch_id="s", scan_id="t", graph_service=graph) + article = _article("https://news.example.com/acme-raises") + + assert enricher.postprocess([article]) == [article] + org = Organization(name="Acme Corp") + assert graph.nodes == [org, article] + # Coverage is a mention, not a site the organization owns, so this is not + # the HAS_WEBSITE edge domain_to_website writes. + assert graph.relationships == [(org, article, "MENTIONED_IN")] + # The attribute used to thread context does not leak onto the node. + assert not hasattr(article, "_source_org_name") + + +def test_postprocess_keeps_each_article_with_its_own_organization(): + graph = _FakeGraphService() + enricher = OrgToNewsEnricher(sketch_id="s", scan_id="t", graph_service=graph) + first = _article("https://news.example.com/acme", "Acme Corp") + second = _article("https://news.example.com/globex", "Globex") + + enricher.postprocess([first, second]) + + assert [(rel[0].name, str(rel[1].url)) for rel in graph.relationships] == [ + ("Acme Corp", "https://news.example.com/acme"), + ("Globex", "https://news.example.com/globex"), + ]