From adbcef271405e34d5da89d8c45c5d1ec2935307a Mon Sep 17 00:00:00 2001 From: Faisal Date: Mon, 1 Jun 2026 02:47:29 +0200 Subject: [PATCH] feat(enrichers): add Arabic media enrichers (Sabq, Argaam, Al Arabiya, Nitter) Adds 8 new enrichers that surface Arabic-language mentions of an Individual or a Phrase (topic) and link them into the graph as Website nodes: - individual_to_sabq / phrase_to_sabq MENTIONED_IN_SABQ - individual_to_argaam / phrase_to_argaam MENTIONED_IN_ARGAAM - individual_to_alarabiya / phrase_to_alarabiya MENTIONED_IN_ALARABIYA - individual_to_arabic_tweets / phrase_to_arabic_tweets MENTIONED_ON_TWITTER_AR Shared scrapers live under src/tools/arabic_media/ and follow the existing Tool base class. Al Arabiya uses Google News RSS filtered to alarabiya.net. Arabic Twitter uses Nitter mirrors with a Google dork fallback. XML parsing uses defusedxml to avoid XXE / billion-laughs on the RSS feed. Tests mock the underlying Tool layer; no real network calls. 19 new tests, all green alongside the existing test_registry suite. --- flowsint-enrichers/pyproject.toml | 1 + .../individual/to_alarabiya.py | 134 +++++++++++++++++ .../individual/to_arabic_tweets.py | 136 ++++++++++++++++++ .../individual/to_argaam.py | 134 +++++++++++++++++ .../flowsint_enrichers/individual/to_sabq.py | 134 +++++++++++++++++ .../src/flowsint_enrichers/phrase/__init__.py | 0 .../flowsint_enrichers/phrase/to_alarabiya.py | 127 ++++++++++++++++ .../phrase/to_arabic_tweets.py | 127 ++++++++++++++++ .../flowsint_enrichers/phrase/to_argaam.py | 127 ++++++++++++++++ .../src/flowsint_enrichers/phrase/to_sabq.py | 127 ++++++++++++++++ .../src/tools/arabic_media/__init__.py | 0 .../src/tools/arabic_media/alarabiya.py | 70 +++++++++ .../src/tools/arabic_media/argaam.py | 66 +++++++++ .../src/tools/arabic_media/nitter.py | 125 ++++++++++++++++ .../src/tools/arabic_media/sabq.py | 62 ++++++++ .../tests/enrichers/test_arabic_alarabiya.py | 70 +++++++++ .../tests/enrichers/test_arabic_argaam.py | 73 ++++++++++ .../tests/enrichers/test_arabic_sabq.py | 133 +++++++++++++++++ .../tests/enrichers/test_arabic_tweets.py | 75 ++++++++++ uv.lock | 11 ++ 20 files changed, 1732 insertions(+) create mode 100644 flowsint-enrichers/src/flowsint_enrichers/individual/to_alarabiya.py create mode 100644 flowsint-enrichers/src/flowsint_enrichers/individual/to_arabic_tweets.py create mode 100644 flowsint-enrichers/src/flowsint_enrichers/individual/to_argaam.py create mode 100644 flowsint-enrichers/src/flowsint_enrichers/individual/to_sabq.py create mode 100644 flowsint-enrichers/src/flowsint_enrichers/phrase/__init__.py create mode 100644 flowsint-enrichers/src/flowsint_enrichers/phrase/to_alarabiya.py create mode 100644 flowsint-enrichers/src/flowsint_enrichers/phrase/to_arabic_tweets.py create mode 100644 flowsint-enrichers/src/flowsint_enrichers/phrase/to_argaam.py create mode 100644 flowsint-enrichers/src/flowsint_enrichers/phrase/to_sabq.py create mode 100644 flowsint-enrichers/src/tools/arabic_media/__init__.py create mode 100644 flowsint-enrichers/src/tools/arabic_media/alarabiya.py create mode 100644 flowsint-enrichers/src/tools/arabic_media/argaam.py create mode 100644 flowsint-enrichers/src/tools/arabic_media/nitter.py create mode 100644 flowsint-enrichers/src/tools/arabic_media/sabq.py create mode 100644 flowsint-enrichers/tests/enrichers/test_arabic_alarabiya.py create mode 100644 flowsint-enrichers/tests/enrichers/test_arabic_argaam.py create mode 100644 flowsint-enrichers/tests/enrichers/test_arabic_sabq.py create mode 100644 flowsint-enrichers/tests/enrichers/test_arabic_tweets.py diff --git a/flowsint-enrichers/pyproject.toml b/flowsint-enrichers/pyproject.toml index 3abdb56ed..ec823ad3d 100644 --- a/flowsint-enrichers/pyproject.toml +++ b/flowsint-enrichers/pyproject.toml @@ -24,6 +24,7 @@ dependencies = [ "reconcrawl", "reconspread", "dnspython>=2.4,<3.0", + "defusedxml>=0.7,<0.8", ] [dependency-groups] diff --git a/flowsint-enrichers/src/flowsint_enrichers/individual/to_alarabiya.py b/flowsint-enrichers/src/flowsint_enrichers/individual/to_alarabiya.py new file mode 100644 index 000000000..b8666f754 --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/individual/to_alarabiya.py @@ -0,0 +1,134 @@ +"""Al Arabiya news enricher: Individual -> Website mentions (Google News RSS).""" + +from typing import Any, Dict, List, Optional, Union + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_types import Individual, Website + +from flowsint_enrichers.registry import flowsint_enricher +from tools.arabic_media.alarabiya import AlArabiyaTool + + +REL_LABEL = "MENTIONED_IN_ALARABIYA" + + +@flowsint_enricher +class IndividualToAlarabiyaEnricher(Enricher): + """[ALARABIYA] Searches Al Arabiya (via Google News RSS) for mentions of an individual.""" + + InputType = Individual + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + self._extracted: List[Dict[str, Any]] = [] + + @classmethod + def name(cls) -> str: + return "individual_to_alarabiya" + + @classmethod + def category(cls) -> str: + return "Individual" + + @classmethod + def key(cls) -> str: + return "full_name" + + def preprocess( + self, data: Union[List[str], List[dict], List[InputType]] + ) -> List[InputType]: + if not isinstance(data, list): + raise ValueError(f"Expected list input, got {type(data).__name__}") + cleaned: List[InputType] = [] + for item in data: + if isinstance(item, str): + parts = item.strip().split() + if len(parts) >= 2: + cleaned.append( + Individual( + first_name=parts[0], + last_name=" ".join(parts[1:]), + full_name=item, + ) + ) + elif isinstance(item, dict) and item.get("full_name"): + cleaned.append(Individual(**item)) + elif isinstance(item, Individual): + cleaned.append(item) + if not cleaned: + Logger.error( + self.sketch_id, + {"message": "[INDIVIDUAL_TO_ALARABIYA] No valid Individual inputs."}, + ) + return cleaned + + async def scan(self, data: List[InputType]) -> List[OutputType]: + tool = AlArabiyaTool() + websites: List[OutputType] = [] + self._extracted = [] + for individual in data: + query = individual.full_name + if not query: + continue + try: + hits = tool.launch(query) + except Exception as exc: + Logger.error( + self.sketch_id, + {"message": f"[ALARABIYA] error for '{query}': {exc}"}, + ) + continue + for hit in hits: + website = Website( + url=hit["url"], + title=hit.get("title"), + description=hit.get("snippet"), + ) + websites.append(website) + self._extracted.append({"individual": individual, "website": website}) + return websites + + def postprocess( + self, + results: List[OutputType], + original_input: List[InputType], + ) -> List[OutputType]: + if not self._graph_service: + return results + + seen_urls: set = set() + seen_individuals: set = set() + for item in self._extracted: + individual: Individual = item["individual"] + website: Website = item["website"] + url_str = str(website.url) + if url_str in seen_urls: + continue + seen_urls.add(url_str) + if individual.full_name not in seen_individuals: + self.create_node(individual) + seen_individuals.add(individual.full_name) + self.create_node(website) + self.create_relationship(individual, website, REL_LABEL) + self.log_graph_message( + f"Linked {individual.full_name} -[:{REL_LABEL}]-> {url_str}" + ) + return results + + +InputType = IndividualToAlarabiyaEnricher.InputType +OutputType = IndividualToAlarabiyaEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/individual/to_arabic_tweets.py b/flowsint-enrichers/src/flowsint_enrichers/individual/to_arabic_tweets.py new file mode 100644 index 000000000..ad1a260fe --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/individual/to_arabic_tweets.py @@ -0,0 +1,136 @@ +"""Arabic-language Twitter/X enricher via Nitter: Individual -> Website mentions.""" + +from typing import Any, Dict, List, Optional, Union + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_types import Individual, Website + +from flowsint_enrichers.registry import flowsint_enricher +from tools.arabic_media.nitter import NitterArabicTool + + +REL_LABEL = "MENTIONED_ON_TWITTER_AR" + + +@flowsint_enricher +class IndividualToArabicTweetsEnricher(Enricher): + """[NITTER] Searches Arabic tweets (Nitter + Google fallback) for mentions of an individual.""" + + InputType = Individual + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + self._extracted: List[Dict[str, Any]] = [] + + @classmethod + def name(cls) -> str: + return "individual_to_arabic_tweets" + + @classmethod + def category(cls) -> str: + return "Individual" + + @classmethod + def key(cls) -> str: + return "full_name" + + def preprocess( + self, data: Union[List[str], List[dict], List[InputType]] + ) -> List[InputType]: + if not isinstance(data, list): + raise ValueError(f"Expected list input, got {type(data).__name__}") + cleaned: List[InputType] = [] + for item in data: + if isinstance(item, str): + parts = item.strip().split() + if len(parts) >= 2: + cleaned.append( + Individual( + first_name=parts[0], + last_name=" ".join(parts[1:]), + full_name=item, + ) + ) + elif isinstance(item, dict) and item.get("full_name"): + cleaned.append(Individual(**item)) + elif isinstance(item, Individual): + cleaned.append(item) + if not cleaned: + Logger.error( + self.sketch_id, + { + "message": "[INDIVIDUAL_TO_ARABIC_TWEETS] No valid Individual inputs." + }, + ) + return cleaned + + async def scan(self, data: List[InputType]) -> List[OutputType]: + tool = NitterArabicTool() + websites: List[OutputType] = [] + self._extracted = [] + for individual in data: + query = individual.full_name + if not query: + continue + try: + hits = tool.launch(query) + except Exception as exc: + Logger.error( + self.sketch_id, + {"message": f"[NITTER] error for '{query}': {exc}"}, + ) + continue + for hit in hits: + website = Website( + url=hit["url"], + title=hit.get("title"), + description=hit.get("snippet"), + ) + websites.append(website) + self._extracted.append({"individual": individual, "website": website}) + return websites + + def postprocess( + self, + results: List[OutputType], + original_input: List[InputType], + ) -> List[OutputType]: + if not self._graph_service: + return results + + seen_urls: set = set() + seen_individuals: set = set() + for item in self._extracted: + individual: Individual = item["individual"] + website: Website = item["website"] + url_str = str(website.url) + if url_str in seen_urls: + continue + seen_urls.add(url_str) + if individual.full_name not in seen_individuals: + self.create_node(individual) + seen_individuals.add(individual.full_name) + self.create_node(website) + self.create_relationship(individual, website, REL_LABEL) + self.log_graph_message( + f"Linked {individual.full_name} -[:{REL_LABEL}]-> {url_str}" + ) + return results + + +InputType = IndividualToArabicTweetsEnricher.InputType +OutputType = IndividualToArabicTweetsEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/individual/to_argaam.py b/flowsint-enrichers/src/flowsint_enrichers/individual/to_argaam.py new file mode 100644 index 000000000..d2b6f30f0 --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/individual/to_argaam.py @@ -0,0 +1,134 @@ +"""Argaam Arabic financial news enricher: Individual -> Website mentions.""" + +from typing import Any, Dict, List, Optional, Union + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_types import Individual, Website + +from flowsint_enrichers.registry import flowsint_enricher +from tools.arabic_media.argaam import ArgaamTool + + +REL_LABEL = "MENTIONED_IN_ARGAAM" + + +@flowsint_enricher +class IndividualToArgaamEnricher(Enricher): + """[ARGAAM] Searches argaam.com Arabic financial news for mentions of an individual.""" + + InputType = Individual + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + self._extracted: List[Dict[str, Any]] = [] + + @classmethod + def name(cls) -> str: + return "individual_to_argaam" + + @classmethod + def category(cls) -> str: + return "Individual" + + @classmethod + def key(cls) -> str: + return "full_name" + + def preprocess( + self, data: Union[List[str], List[dict], List[InputType]] + ) -> List[InputType]: + if not isinstance(data, list): + raise ValueError(f"Expected list input, got {type(data).__name__}") + cleaned: List[InputType] = [] + for item in data: + if isinstance(item, str): + parts = item.strip().split() + if len(parts) >= 2: + cleaned.append( + Individual( + first_name=parts[0], + last_name=" ".join(parts[1:]), + full_name=item, + ) + ) + elif isinstance(item, dict) and item.get("full_name"): + cleaned.append(Individual(**item)) + elif isinstance(item, Individual): + cleaned.append(item) + if not cleaned: + Logger.error( + self.sketch_id, + {"message": "[INDIVIDUAL_TO_ARGAAM] No valid Individual inputs."}, + ) + return cleaned + + async def scan(self, data: List[InputType]) -> List[OutputType]: + tool = ArgaamTool() + websites: List[OutputType] = [] + self._extracted = [] + for individual in data: + query = individual.full_name + if not query: + continue + try: + hits = tool.launch(query) + except Exception as exc: + Logger.error( + self.sketch_id, + {"message": f"[ARGAAM] error for '{query}': {exc}"}, + ) + continue + for hit in hits: + website = Website( + url=hit["url"], + title=hit.get("title"), + description=hit.get("snippet"), + ) + websites.append(website) + self._extracted.append({"individual": individual, "website": website}) + return websites + + def postprocess( + self, + results: List[OutputType], + original_input: List[InputType], + ) -> List[OutputType]: + if not self._graph_service: + return results + + seen_urls: set = set() + seen_individuals: set = set() + for item in self._extracted: + individual: Individual = item["individual"] + website: Website = item["website"] + url_str = str(website.url) + if url_str in seen_urls: + continue + seen_urls.add(url_str) + if individual.full_name not in seen_individuals: + self.create_node(individual) + seen_individuals.add(individual.full_name) + self.create_node(website) + self.create_relationship(individual, website, REL_LABEL) + self.log_graph_message( + f"Linked {individual.full_name} -[:{REL_LABEL}]-> {url_str}" + ) + return results + + +InputType = IndividualToArgaamEnricher.InputType +OutputType = IndividualToArgaamEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/individual/to_sabq.py b/flowsint-enrichers/src/flowsint_enrichers/individual/to_sabq.py new file mode 100644 index 000000000..b14fd5b1c --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/individual/to_sabq.py @@ -0,0 +1,134 @@ +"""Sabq Arabic news enricher: Individual -> Website mentions.""" + +from typing import Any, Dict, List, Optional, Union + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_types import Individual, Website + +from flowsint_enrichers.registry import flowsint_enricher +from tools.arabic_media.sabq import SabqTool + + +REL_LABEL = "MENTIONED_IN_SABQ" + + +@flowsint_enricher +class IndividualToSabqEnricher(Enricher): + """[SABQ] Searches sabq.org Arabic news for mentions of an individual.""" + + InputType = Individual + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + self._extracted: List[Dict[str, Any]] = [] + + @classmethod + def name(cls) -> str: + return "individual_to_sabq" + + @classmethod + def category(cls) -> str: + return "Individual" + + @classmethod + def key(cls) -> str: + return "full_name" + + def preprocess( + self, data: Union[List[str], List[dict], List[InputType]] + ) -> List[InputType]: + if not isinstance(data, list): + raise ValueError(f"Expected list input, got {type(data).__name__}") + cleaned: List[InputType] = [] + for item in data: + if isinstance(item, str): + parts = item.strip().split() + if len(parts) >= 2: + cleaned.append( + Individual( + first_name=parts[0], + last_name=" ".join(parts[1:]), + full_name=item, + ) + ) + elif isinstance(item, dict) and item.get("full_name"): + cleaned.append(Individual(**item)) + elif isinstance(item, Individual): + cleaned.append(item) + if not cleaned: + Logger.error( + self.sketch_id, + {"message": "[INDIVIDUAL_TO_SABQ] No valid Individual inputs."}, + ) + return cleaned + + async def scan(self, data: List[InputType]) -> List[OutputType]: + tool = SabqTool() + websites: List[OutputType] = [] + self._extracted = [] + for individual in data: + query = individual.full_name + if not query: + continue + try: + hits = tool.launch(query) + except Exception as exc: + Logger.error( + self.sketch_id, + {"message": f"[SABQ] error for '{query}': {exc}"}, + ) + continue + for hit in hits: + website = Website( + url=hit["url"], + title=hit.get("title"), + description=hit.get("snippet"), + ) + websites.append(website) + self._extracted.append({"individual": individual, "website": website}) + return websites + + def postprocess( + self, + results: List[OutputType], + original_input: List[InputType], + ) -> List[OutputType]: + if not self._graph_service: + return results + + seen_urls: set = set() + seen_individuals: set = set() + for item in self._extracted: + individual: Individual = item["individual"] + website: Website = item["website"] + url_str = str(website.url) + if url_str in seen_urls: + continue + seen_urls.add(url_str) + if individual.full_name not in seen_individuals: + self.create_node(individual) + seen_individuals.add(individual.full_name) + self.create_node(website) + self.create_relationship(individual, website, REL_LABEL) + self.log_graph_message( + f"Linked {individual.full_name} -[:{REL_LABEL}]-> {url_str}" + ) + return results + + +InputType = IndividualToSabqEnricher.InputType +OutputType = IndividualToSabqEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/phrase/__init__.py b/flowsint-enrichers/src/flowsint_enrichers/phrase/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/flowsint-enrichers/src/flowsint_enrichers/phrase/to_alarabiya.py b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_alarabiya.py new file mode 100644 index 000000000..1d1135614 --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_alarabiya.py @@ -0,0 +1,127 @@ +"""Al Arabiya news enricher: Phrase (topic) -> Website mentions (Google News RSS).""" + +from typing import Any, Dict, List, Optional, Union + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_types import Phrase, Website + +from flowsint_enrichers.registry import flowsint_enricher +from tools.arabic_media.alarabiya import AlArabiyaTool + + +REL_LABEL = "MENTIONED_IN_ALARABIYA" + + +@flowsint_enricher +class PhraseToAlarabiyaEnricher(Enricher): + """[ALARABIYA] Searches Al Arabiya (via Google News RSS) for mentions of a topic/phrase.""" + + InputType = Phrase + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + self._extracted: List[Dict[str, Any]] = [] + + @classmethod + def name(cls) -> str: + return "phrase_to_alarabiya" + + @classmethod + def category(cls) -> str: + return "Phrase" + + @classmethod + def key(cls) -> str: + return "text" + + def preprocess( + self, data: Union[List[str], List[dict], List[InputType]] + ) -> List[InputType]: + if not isinstance(data, list): + raise ValueError(f"Expected list input, got {type(data).__name__}") + cleaned: List[InputType] = [] + for item in data: + if isinstance(item, str) and item.strip(): + cleaned.append(Phrase(text=item.strip())) + elif isinstance(item, dict) and item.get("text"): + cleaned.append(Phrase(**item)) + elif isinstance(item, Phrase): + cleaned.append(item) + if not cleaned: + Logger.error( + self.sketch_id, + {"message": "[PHRASE_TO_ALARABIYA] No valid Phrase inputs."}, + ) + return cleaned + + async def scan(self, data: List[InputType]) -> List[OutputType]: + tool = AlArabiyaTool() + websites: List[OutputType] = [] + self._extracted = [] + for phrase in data: + query = str(phrase.text) + if not query: + continue + try: + hits = tool.launch(query) + except Exception as exc: + Logger.error( + self.sketch_id, + {"message": f"[ALARABIYA] error for '{query}': {exc}"}, + ) + continue + for hit in hits: + website = Website( + url=hit["url"], + title=hit.get("title"), + description=hit.get("snippet"), + ) + websites.append(website) + self._extracted.append({"phrase": phrase, "website": website}) + return websites + + def postprocess( + self, + results: List[OutputType], + original_input: List[InputType], + ) -> List[OutputType]: + if not self._graph_service: + return results + + seen_urls: set = set() + seen_phrases: set = set() + for item in self._extracted: + phrase: Phrase = item["phrase"] + website: Website = item["website"] + url_str = str(website.url) + if url_str in seen_urls: + continue + seen_urls.add(url_str) + phrase_key = str(phrase.text) + if phrase_key not in seen_phrases: + self.create_node(phrase) + seen_phrases.add(phrase_key) + self.create_node(website) + self.create_relationship(phrase, website, REL_LABEL) + self.log_graph_message( + f"Linked phrase '{phrase.text}' -[:{REL_LABEL}]-> {url_str}" + ) + return results + + +InputType = PhraseToAlarabiyaEnricher.InputType +OutputType = PhraseToAlarabiyaEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/phrase/to_arabic_tweets.py b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_arabic_tweets.py new file mode 100644 index 000000000..a203154b6 --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_arabic_tweets.py @@ -0,0 +1,127 @@ +"""Arabic-language Twitter/X enricher via Nitter: Phrase (topic) -> Website mentions.""" + +from typing import Any, Dict, List, Optional, Union + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_types import Phrase, Website + +from flowsint_enrichers.registry import flowsint_enricher +from tools.arabic_media.nitter import NitterArabicTool + + +REL_LABEL = "MENTIONED_ON_TWITTER_AR" + + +@flowsint_enricher +class PhraseToArabicTweetsEnricher(Enricher): + """[NITTER] Searches Arabic tweets (Nitter + Google fallback) for mentions of a topic/phrase.""" + + InputType = Phrase + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + self._extracted: List[Dict[str, Any]] = [] + + @classmethod + def name(cls) -> str: + return "phrase_to_arabic_tweets" + + @classmethod + def category(cls) -> str: + return "Phrase" + + @classmethod + def key(cls) -> str: + return "text" + + def preprocess( + self, data: Union[List[str], List[dict], List[InputType]] + ) -> List[InputType]: + if not isinstance(data, list): + raise ValueError(f"Expected list input, got {type(data).__name__}") + cleaned: List[InputType] = [] + for item in data: + if isinstance(item, str) and item.strip(): + cleaned.append(Phrase(text=item.strip())) + elif isinstance(item, dict) and item.get("text"): + cleaned.append(Phrase(**item)) + elif isinstance(item, Phrase): + cleaned.append(item) + if not cleaned: + Logger.error( + self.sketch_id, + {"message": "[PHRASE_TO_ARABIC_TWEETS] No valid Phrase inputs."}, + ) + return cleaned + + async def scan(self, data: List[InputType]) -> List[OutputType]: + tool = NitterArabicTool() + websites: List[OutputType] = [] + self._extracted = [] + for phrase in data: + query = str(phrase.text) + if not query: + continue + try: + hits = tool.launch(query) + except Exception as exc: + Logger.error( + self.sketch_id, + {"message": f"[NITTER] error for '{query}': {exc}"}, + ) + continue + for hit in hits: + website = Website( + url=hit["url"], + title=hit.get("title"), + description=hit.get("snippet"), + ) + websites.append(website) + self._extracted.append({"phrase": phrase, "website": website}) + return websites + + def postprocess( + self, + results: List[OutputType], + original_input: List[InputType], + ) -> List[OutputType]: + if not self._graph_service: + return results + + seen_urls: set = set() + seen_phrases: set = set() + for item in self._extracted: + phrase: Phrase = item["phrase"] + website: Website = item["website"] + url_str = str(website.url) + if url_str in seen_urls: + continue + seen_urls.add(url_str) + phrase_key = str(phrase.text) + if phrase_key not in seen_phrases: + self.create_node(phrase) + seen_phrases.add(phrase_key) + self.create_node(website) + self.create_relationship(phrase, website, REL_LABEL) + self.log_graph_message( + f"Linked phrase '{phrase.text}' -[:{REL_LABEL}]-> {url_str}" + ) + return results + + +InputType = PhraseToArabicTweetsEnricher.InputType +OutputType = PhraseToArabicTweetsEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/phrase/to_argaam.py b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_argaam.py new file mode 100644 index 000000000..145456d14 --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_argaam.py @@ -0,0 +1,127 @@ +"""Argaam Arabic financial news enricher: Phrase (topic) -> Website mentions.""" + +from typing import Any, Dict, List, Optional, Union + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_types import Phrase, Website + +from flowsint_enrichers.registry import flowsint_enricher +from tools.arabic_media.argaam import ArgaamTool + + +REL_LABEL = "MENTIONED_IN_ARGAAM" + + +@flowsint_enricher +class PhraseToArgaamEnricher(Enricher): + """[ARGAAM] Searches argaam.com Arabic financial news for mentions of a topic/phrase.""" + + InputType = Phrase + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + self._extracted: List[Dict[str, Any]] = [] + + @classmethod + def name(cls) -> str: + return "phrase_to_argaam" + + @classmethod + def category(cls) -> str: + return "Phrase" + + @classmethod + def key(cls) -> str: + return "text" + + def preprocess( + self, data: Union[List[str], List[dict], List[InputType]] + ) -> List[InputType]: + if not isinstance(data, list): + raise ValueError(f"Expected list input, got {type(data).__name__}") + cleaned: List[InputType] = [] + for item in data: + if isinstance(item, str) and item.strip(): + cleaned.append(Phrase(text=item.strip())) + elif isinstance(item, dict) and item.get("text"): + cleaned.append(Phrase(**item)) + elif isinstance(item, Phrase): + cleaned.append(item) + if not cleaned: + Logger.error( + self.sketch_id, + {"message": "[PHRASE_TO_ARGAAM] No valid Phrase inputs."}, + ) + return cleaned + + async def scan(self, data: List[InputType]) -> List[OutputType]: + tool = ArgaamTool() + websites: List[OutputType] = [] + self._extracted = [] + for phrase in data: + query = str(phrase.text) + if not query: + continue + try: + hits = tool.launch(query) + except Exception as exc: + Logger.error( + self.sketch_id, + {"message": f"[ARGAAM] error for '{query}': {exc}"}, + ) + continue + for hit in hits: + website = Website( + url=hit["url"], + title=hit.get("title"), + description=hit.get("snippet"), + ) + websites.append(website) + self._extracted.append({"phrase": phrase, "website": website}) + return websites + + def postprocess( + self, + results: List[OutputType], + original_input: List[InputType], + ) -> List[OutputType]: + if not self._graph_service: + return results + + seen_urls: set = set() + seen_phrases: set = set() + for item in self._extracted: + phrase: Phrase = item["phrase"] + website: Website = item["website"] + url_str = str(website.url) + if url_str in seen_urls: + continue + seen_urls.add(url_str) + phrase_key = str(phrase.text) + if phrase_key not in seen_phrases: + self.create_node(phrase) + seen_phrases.add(phrase_key) + self.create_node(website) + self.create_relationship(phrase, website, REL_LABEL) + self.log_graph_message( + f"Linked phrase '{phrase.text}' -[:{REL_LABEL}]-> {url_str}" + ) + return results + + +InputType = PhraseToArgaamEnricher.InputType +OutputType = PhraseToArgaamEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/phrase/to_sabq.py b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_sabq.py new file mode 100644 index 000000000..dd705eb1e --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_sabq.py @@ -0,0 +1,127 @@ +"""Sabq Arabic news enricher: Phrase (topic) -> Website mentions.""" + +from typing import Any, Dict, List, Optional, Union + +from flowsint_core.core.enricher_base import Enricher +from flowsint_core.core.logger import Logger +from flowsint_types import Phrase, Website + +from flowsint_enrichers.registry import flowsint_enricher +from tools.arabic_media.sabq import SabqTool + + +REL_LABEL = "MENTIONED_IN_SABQ" + + +@flowsint_enricher +class PhraseToSabqEnricher(Enricher): + """[SABQ] Searches sabq.org Arabic news for mentions of a topic/phrase.""" + + InputType = Phrase + OutputType = Website + + def __init__( + self, + sketch_id: Optional[str] = None, + scan_id: Optional[str] = None, + vault=None, + params: Optional[Dict[str, Any]] = None, + ): + super().__init__( + sketch_id=sketch_id, + scan_id=scan_id, + params_schema=self.get_params_schema(), + vault=vault, + params=params, + ) + self._extracted: List[Dict[str, Any]] = [] + + @classmethod + def name(cls) -> str: + return "phrase_to_sabq" + + @classmethod + def category(cls) -> str: + return "Phrase" + + @classmethod + def key(cls) -> str: + return "text" + + def preprocess( + self, data: Union[List[str], List[dict], List[InputType]] + ) -> List[InputType]: + if not isinstance(data, list): + raise ValueError(f"Expected list input, got {type(data).__name__}") + cleaned: List[InputType] = [] + for item in data: + if isinstance(item, str) and item.strip(): + cleaned.append(Phrase(text=item.strip())) + elif isinstance(item, dict) and item.get("text"): + cleaned.append(Phrase(**item)) + elif isinstance(item, Phrase): + cleaned.append(item) + if not cleaned: + Logger.error( + self.sketch_id, + {"message": "[PHRASE_TO_SABQ] No valid Phrase inputs."}, + ) + return cleaned + + async def scan(self, data: List[InputType]) -> List[OutputType]: + tool = SabqTool() + websites: List[OutputType] = [] + self._extracted = [] + for phrase in data: + query = str(phrase.text) + if not query: + continue + try: + hits = tool.launch(query) + except Exception as exc: + Logger.error( + self.sketch_id, + {"message": f"[SABQ] error for '{query}': {exc}"}, + ) + continue + for hit in hits: + website = Website( + url=hit["url"], + title=hit.get("title"), + description=hit.get("snippet"), + ) + websites.append(website) + self._extracted.append({"phrase": phrase, "website": website}) + return websites + + def postprocess( + self, + results: List[OutputType], + original_input: List[InputType], + ) -> List[OutputType]: + if not self._graph_service: + return results + + seen_urls: set = set() + seen_phrases: set = set() + for item in self._extracted: + phrase: Phrase = item["phrase"] + website: Website = item["website"] + url_str = str(website.url) + if url_str in seen_urls: + continue + seen_urls.add(url_str) + phrase_key = str(phrase.text) + if phrase_key not in seen_phrases: + self.create_node(phrase) + seen_phrases.add(phrase_key) + self.create_node(website) + self.create_relationship(phrase, website, REL_LABEL) + self.log_graph_message( + f"Linked phrase '{phrase.text}' -[:{REL_LABEL}]-> {url_str}" + ) + return results + + +InputType = PhraseToSabqEnricher.InputType +OutputType = PhraseToSabqEnricher.OutputType diff --git a/flowsint-enrichers/src/tools/arabic_media/__init__.py b/flowsint-enrichers/src/tools/arabic_media/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/flowsint-enrichers/src/tools/arabic_media/alarabiya.py b/flowsint-enrichers/src/tools/arabic_media/alarabiya.py new file mode 100644 index 000000000..abb5d4541 --- /dev/null +++ b/flowsint-enrichers/src/tools/arabic_media/alarabiya.py @@ -0,0 +1,70 @@ +"""Al Arabiya news search tool via Google News RSS.""" + +import urllib.parse +from typing import Any, Dict, List + +import requests +from defusedxml import ElementTree as ET + +from ..base import Tool + + +class AlArabiyaTool(Tool): + GOOGLE_NEWS_RSS = "https://news.google.com/rss/search" + MAX_ITEMS = 10 + TIMEOUT = 10 + + @classmethod + def name(cls) -> str: + return "alarabiya" + + @classmethod + def version(cls) -> str: + return "1.0.0" + + @classmethod + def description(cls) -> str: + return "Search Al Arabiya news (alarabiya.net) via Google News RSS." + + @classmethod + def category(cls) -> str: + return "Arabic media" + + def launch(self, value: str, *args, **kwargs) -> List[Dict[str, Any]]: + filtered_query = f"{value} site:alarabiya.net" + params = urllib.parse.urlencode({"q": filtered_query, "hl": "en"}) + url = f"{self.GOOGLE_NEWS_RSS}?{params}" + + response = requests.get(url, timeout=self.TIMEOUT) + if response.status_code != 200: + return [] + + try: + root = ET.fromstring(response.text) + except ET.ParseError: + return [] + + channel = root.find("channel") + items = channel.findall("item") if channel is not None else [] + + results: List[Dict[str, Any]] = [] + for item in items[: self.MAX_ITEMS]: + title_el = item.find("title") + link_el = item.find("link") + desc_el = item.find("description") + pub_el = item.find("pubDate") + + link = link_el.text if link_el is not None else None + if not link: + continue + + results.append( + { + "url": link, + "title": (title_el.text if title_el is not None else value) + or value, + "snippet": (desc_el.text if desc_el is not None else "") or "", + "date": pub_el.text if pub_el is not None else None, + } + ) + return results diff --git a/flowsint-enrichers/src/tools/arabic_media/argaam.py b/flowsint-enrichers/src/tools/arabic_media/argaam.py new file mode 100644 index 000000000..aaedf2794 --- /dev/null +++ b/flowsint-enrichers/src/tools/arabic_media/argaam.py @@ -0,0 +1,66 @@ +"""Argaam Arabic financial/business news search tool.""" + +import urllib.parse +from typing import Any, Dict, List + +import requests +from bs4 import BeautifulSoup + +from ..base import Tool + + +class ArgaamTool(Tool): + ARGAAM_SEARCH = "https://www.argaam.com/ar/search" + ARGAAM_BASE = "https://www.argaam.com" + MAX_RESULTS = 10 + TIMEOUT = 10 + + @classmethod + def name(cls) -> str: + return "argaam" + + @classmethod + def version(cls) -> str: + return "1.0.0" + + @classmethod + def description(cls) -> str: + return "Search Argaam Arabic financial news for mentions of a person or topic." + + @classmethod + def category(cls) -> str: + return "Arabic media" + + def launch(self, value: str, *args, **kwargs) -> List[Dict[str, Any]]: + params = urllib.parse.urlencode({"q": value}) + url = f"{self.ARGAAM_SEARCH}?{params}" + response = requests.get(url, timeout=self.TIMEOUT) + if response.status_code != 200: + return [] + + soup = BeautifulSoup(response.text, "html.parser") + items = soup.select("article, .search-result, .news-item, .listItem") + + results: List[Dict[str, Any]] = [] + for item in items[: self.MAX_RESULTS]: + title_el = item.select_one("h2 a, h3 a, .title a, a") + if title_el is None: + continue + + title = title_el.get_text(strip=True) or value + link = title_el.get("href", "") + if link and not link.startswith("http"): + link = f"{self.ARGAAM_BASE}{link}" + if not link: + continue + + snippet_el = item.select_one("p, .excerpt, .summary") + snippet = snippet_el.get_text(strip=True) if snippet_el else title + + date_el = item.select_one(".date, time, [datetime]") + date = date_el.get_text(strip=True) if date_el else None + + results.append( + {"url": link, "title": title, "snippet": snippet, "date": date} + ) + return results diff --git a/flowsint-enrichers/src/tools/arabic_media/nitter.py b/flowsint-enrichers/src/tools/arabic_media/nitter.py new file mode 100644 index 000000000..79932b04f --- /dev/null +++ b/flowsint-enrichers/src/tools/arabic_media/nitter.py @@ -0,0 +1,125 @@ +"""Nitter (Twitter mirror) Arabic search tool with Google fallback.""" + +import urllib.parse +from typing import Any, Dict, List + +import requests +from bs4 import BeautifulSoup + +from ..base import Tool + + +NITTER_INSTANCES = ( + "https://nitter.privacydev.net", + "https://nitter.poast.org", +) +GOOGLE_SEARCH = "https://www.google.com/search" +X_DOMAINS = ("x.com", "twitter.com") +MAX_RESULTS = 10 +TIMEOUT = 10 + + +class NitterArabicTool(Tool): + @classmethod + def name(cls) -> str: + return "nitter_arabic" + + @classmethod + def version(cls) -> str: + return "1.0.0" + + @classmethod + def description(cls) -> str: + return ( + "Search Arabic tweets via Nitter mirrors with a Google dork fallback " + "(site:x.com lang:ar)." + ) + + @classmethod + def category(cls) -> str: + return "Arabic media" + + def launch(self, value: str, *args, **kwargs) -> List[Dict[str, Any]]: + results = self._search_nitter(value) + if results: + return results + return self._search_google_fallback(value) + + def _search_nitter(self, value: str) -> List[Dict[str, Any]]: + arabic_query = f"{value} lang:ar" + for instance in NITTER_INSTANCES: + params = urllib.parse.urlencode({"q": arabic_query, "f": "tweets"}) + url = f"{instance}/search?{params}" + try: + response = requests.get(url, timeout=TIMEOUT) + except requests.RequestException: + continue + if response.status_code != 200: + continue + + soup = BeautifulSoup(response.text, "html.parser") + items = soup.select(".timeline-item .tweet-link") + collected: List[Dict[str, Any]] = [] + for item in items[:MAX_RESULTS]: + href = item.get("href", "") + text = item.get_text(strip=True) or value + if not href: + continue + collected.append( + { + "url": f"https://x.com{href}", + "title": f"Arabic tweet about: {value}", + "snippet": text, + "via": instance, + } + ) + if collected: + return collected + return [] + + def _search_google_fallback(self, value: str) -> List[Dict[str, Any]]: + dork = f"{value} site:x.com lang:ar" + params = urllib.parse.urlencode({"q": dork, "num": MAX_RESULTS}) + url = f"{GOOGLE_SEARCH}?{params}" + try: + response = requests.get( + url, + headers={ + "Accept": "text/html,application/xhtml+xml", + "Accept-Language": "en-US,en;q=0.9", + }, + timeout=TIMEOUT, + ) + except requests.RequestException: + return [] + if response.status_code != 200: + return [] + + soup = BeautifulSoup(response.text, "html.parser") + links = soup.select("div.g a[href]") + collected: List[Dict[str, Any]] = [] + for tag in links: + href = tag.get("href", "") + if not href.startswith("http"): + continue + if not any(domain in href for domain in X_DOMAINS): + continue + + title = tag.get_text(strip=True) or href + snippet = "" + parent = tag.find_parent("div") + if parent: + spans = parent.find_all("span") + if spans: + snippet = spans[-1].get_text(strip=True) + collected.append( + { + "url": href, + "title": title, + "snippet": snippet or title, + "via": "google_fallback", + } + ) + if len(collected) >= MAX_RESULTS: + break + return collected diff --git a/flowsint-enrichers/src/tools/arabic_media/sabq.py b/flowsint-enrichers/src/tools/arabic_media/sabq.py new file mode 100644 index 000000000..0a1529d26 --- /dev/null +++ b/flowsint-enrichers/src/tools/arabic_media/sabq.py @@ -0,0 +1,62 @@ +"""Sabq Arabic news search tool.""" + +import urllib.parse +from typing import Any, Dict, List + +import requests +from bs4 import BeautifulSoup + +from ..base import Tool + + +class SabqTool(Tool): + SABQ_SEARCH = "https://sabq.org/search" + SABQ_BASE = "https://sabq.org" + MAX_RESULTS = 10 + TIMEOUT = 10 + + @classmethod + def name(cls) -> str: + return "sabq" + + @classmethod + def version(cls) -> str: + return "1.0.0" + + @classmethod + def description(cls) -> str: + return "Search Sabq Arabic news for mentions of a person or topic." + + @classmethod + def category(cls) -> str: + return "Arabic media" + + def launch(self, value: str, *args, **kwargs) -> List[Dict[str, Any]]: + params = urllib.parse.urlencode({"q": value}) + url = f"{self.SABQ_SEARCH}?{params}" + response = requests.get(url, timeout=self.TIMEOUT) + if response.status_code != 200: + return [] + + soup = BeautifulSoup(response.text, "html.parser") + items = soup.select("article, .search-item, .news-item") + + results: List[Dict[str, Any]] = [] + for item in items[: self.MAX_RESULTS]: + title_el = item.select_one("h2 a, h3 a, .title a, a") + if title_el is None: + continue + + title = title_el.get_text(strip=True) or value + link = title_el.get("href", "") + + if link and not link.startswith("http"): + link = f"{self.SABQ_BASE}{link}" + if not link: + continue + + snippet_el = item.select_one("p, .excerpt, .summary") + snippet = snippet_el.get_text(strip=True) if snippet_el else title + + results.append({"url": link, "title": title, "snippet": snippet}) + return results diff --git a/flowsint-enrichers/tests/enrichers/test_arabic_alarabiya.py b/flowsint-enrichers/tests/enrichers/test_arabic_alarabiya.py new file mode 100644 index 000000000..5e8f3d38c --- /dev/null +++ b/flowsint-enrichers/tests/enrichers/test_arabic_alarabiya.py @@ -0,0 +1,70 @@ +"""Tests for Al Arabiya enrichers (Individual and Phrase variants).""" + +from unittest.mock import MagicMock, patch + +import pytest + +from flowsint_enrichers.individual.to_alarabiya import IndividualToAlarabiyaEnricher +from flowsint_enrichers.phrase.to_alarabiya import PhraseToAlarabiyaEnricher +from flowsint_types import Individual, Phrase + + +FIXTURE_HITS = [ + { + "url": "https://www.alarabiya.net/saudi-today/2026/05/30/news-article", + "title": "News", + "snippet": "Breaking news.", + "date": "Sat, 30 May 2026 14:00:00 GMT", + }, +] + + +def _make_enricher(cls): + enricher = cls(sketch_id="s", scan_id="x") + enricher._graph_service = MagicMock() + return enricher + + +@pytest.mark.asyncio +async def test_individual_to_alarabiya_scan_returns_websites(): + enricher = _make_enricher(IndividualToAlarabiyaEnricher) + individual = Individual(first_name="X", last_name="Y", full_name="X Y") + with patch( + "flowsint_enrichers.individual.to_alarabiya.AlArabiyaTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([individual]) + assert len(results) == 1 + + +@pytest.mark.asyncio +async def test_phrase_to_alarabiya_scan_returns_websites(): + enricher = _make_enricher(PhraseToAlarabiyaEnricher) + phrase = Phrase(text="energy markets") + with patch( + "flowsint_enrichers.phrase.to_alarabiya.AlArabiyaTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([phrase]) + assert len(results) == 1 + + +@pytest.mark.asyncio +async def test_alarabiya_uses_correct_rel_label(): + enricher = _make_enricher(IndividualToAlarabiyaEnricher) + individual = Individual(first_name="X", last_name="Y", full_name="X Y") + with patch( + "flowsint_enrichers.individual.to_alarabiya.AlArabiyaTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([individual]) + enricher.postprocess(results, [individual]) + rel_calls = enricher._graph_service.create_relationship.call_args_list + assert rel_calls and rel_calls[0].kwargs["rel_label"] == "MENTIONED_IN_ALARABIYA" + + +def test_enricher_registration(): + from flowsint_enrichers import ENRICHER_REGISTRY + + assert ENRICHER_REGISTRY.enricher_exists("individual_to_alarabiya") + assert ENRICHER_REGISTRY.enricher_exists("phrase_to_alarabiya") diff --git a/flowsint-enrichers/tests/enrichers/test_arabic_argaam.py b/flowsint-enrichers/tests/enrichers/test_arabic_argaam.py new file mode 100644 index 000000000..0ad4e5279 --- /dev/null +++ b/flowsint-enrichers/tests/enrichers/test_arabic_argaam.py @@ -0,0 +1,73 @@ +"""Tests for Argaam enrichers (Individual and Phrase variants).""" + +from unittest.mock import MagicMock, patch + +import pytest + +from flowsint_enrichers.individual.to_argaam import IndividualToArgaamEnricher +from flowsint_enrichers.phrase.to_argaam import PhraseToArgaamEnricher +from flowsint_types import Individual, Phrase + + +FIXTURE_HITS = [ + { + "url": "https://www.argaam.com/ar/article/articledetail/id/1234567", + "title": "تقرير مالي", + "snippet": "ملخص التقرير.", + "date": "2026-05-30", + }, +] + + +def _make_enricher(cls): + enricher = cls(sketch_id="s", scan_id="x") + enricher._graph_service = MagicMock() + return enricher + + +@pytest.mark.asyncio +async def test_individual_to_argaam_scan_returns_websites(): + enricher = _make_enricher(IndividualToArgaamEnricher) + individual = Individual( + first_name="X", last_name="Y", full_name="X Y" + ) + with patch( + "flowsint_enrichers.individual.to_argaam.ArgaamTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([individual]) + assert len(results) == 1 + assert results[0].title == "تقرير مالي" + + +@pytest.mark.asyncio +async def test_phrase_to_argaam_scan_returns_websites(): + enricher = _make_enricher(PhraseToArgaamEnricher) + phrase = Phrase(text="السوق المالية") + with patch( + "flowsint_enrichers.phrase.to_argaam.ArgaamTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([phrase]) + assert len(results) == 1 + + +@pytest.mark.asyncio +async def test_individual_to_argaam_uses_correct_rel_label(): + enricher = _make_enricher(IndividualToArgaamEnricher) + individual = Individual(first_name="X", last_name="Y", full_name="X Y") + with patch( + "flowsint_enrichers.individual.to_argaam.ArgaamTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([individual]) + enricher.postprocess(results, [individual]) + rel_calls = enricher._graph_service.create_relationship.call_args_list + assert rel_calls and rel_calls[0].kwargs["rel_label"] == "MENTIONED_IN_ARGAAM" + + +def test_enricher_registration(): + from flowsint_enrichers import ENRICHER_REGISTRY + + assert ENRICHER_REGISTRY.enricher_exists("individual_to_argaam") + assert ENRICHER_REGISTRY.enricher_exists("phrase_to_argaam") diff --git a/flowsint-enrichers/tests/enrichers/test_arabic_sabq.py b/flowsint-enrichers/tests/enrichers/test_arabic_sabq.py new file mode 100644 index 000000000..697a54946 --- /dev/null +++ b/flowsint-enrichers/tests/enrichers/test_arabic_sabq.py @@ -0,0 +1,133 @@ +"""Tests for Sabq enrichers (Individual and Phrase variants).""" + +from unittest.mock import MagicMock, patch + +import pytest + +from flowsint_enrichers.individual.to_sabq import IndividualToSabqEnricher +from flowsint_enrichers.phrase.to_sabq import PhraseToSabqEnricher +from flowsint_types import Individual, Phrase + + +FIXTURE_HITS = [ + { + "url": "https://sabq.org/news/123", + "title": "خبر تجريبي", + "snippet": "مقتطف نصي تجريبي للتحقق.", + }, + { + "url": "https://sabq.org/news/456", + "title": "Second Article", + "snippet": "Second snippet.", + }, +] + + +def _make_enricher(cls): + enricher = cls(sketch_id="s", scan_id="x") + enricher._graph_service = MagicMock() + return enricher + + +@pytest.mark.asyncio +async def test_individual_to_sabq_scan_returns_websites(): + enricher = _make_enricher(IndividualToSabqEnricher) + individual = Individual( + first_name="Faisal", last_name="Aldeghaither", full_name="Faisal Aldeghaither" + ) + + with patch( + "flowsint_enrichers.individual.to_sabq.SabqTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([individual]) + + assert len(results) == 2 + assert str(results[0].url) == "https://sabq.org/news/123" + assert results[0].title == "خبر تجريبي" + assert results[0].description == "مقتطف نصي تجريبي للتحقق." + + +@pytest.mark.asyncio +async def test_individual_to_sabq_handles_tool_exception(): + enricher = _make_enricher(IndividualToSabqEnricher) + individual = Individual( + first_name="A", last_name="B", full_name="A B" + ) + with patch( + "flowsint_enrichers.individual.to_sabq.SabqTool" + ) as MockTool: + MockTool.return_value.launch.side_effect = RuntimeError("boom") + results = await enricher.scan([individual]) + assert results == [] + + +@pytest.mark.asyncio +async def test_individual_to_sabq_postprocess_creates_nodes_and_edges(): + enricher = _make_enricher(IndividualToSabqEnricher) + individual = Individual( + first_name="A", last_name="B", full_name="A B" + ) + with patch( + "flowsint_enrichers.individual.to_sabq.SabqTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([individual]) + enricher.postprocess(results, [individual]) + + gs = enricher._graph_service + # 1 individual + 2 websites = 3 nodes + assert gs.create_node_from_flowsint_type.call_count == 3 + # 2 relationships + rel_calls = gs.create_relationship.call_args_list + assert len(rel_calls) == 2 + for call in rel_calls: + assert call.kwargs["rel_label"] == "MENTIONED_IN_SABQ" + + +@pytest.mark.asyncio +async def test_individual_to_sabq_postprocess_dedupes_urls(): + enricher = _make_enricher(IndividualToSabqEnricher) + individual = Individual(first_name="A", last_name="B", full_name="A B") + dup = FIXTURE_HITS + [FIXTURE_HITS[0]] + with patch( + "flowsint_enrichers.individual.to_sabq.SabqTool" + ) as MockTool: + MockTool.return_value.launch.return_value = dup + results = await enricher.scan([individual]) + enricher.postprocess(results, [individual]) + + # Still only 2 unique relationships + assert enricher._graph_service.create_relationship.call_count == 2 + + +@pytest.mark.asyncio +async def test_phrase_to_sabq_scan_returns_websites(): + enricher = _make_enricher(PhraseToSabqEnricher) + phrase = Phrase(text="رؤية 2030") + + with patch( + "flowsint_enrichers.phrase.to_sabq.SabqTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([phrase]) + + assert len(results) == 2 + assert str(results[1].url) == "https://sabq.org/news/456" + + +@pytest.mark.asyncio +async def test_phrase_to_sabq_preprocess_accepts_string_inputs(): + enricher = _make_enricher(PhraseToSabqEnricher) + cleaned = enricher.preprocess(["رؤية 2030", " ", "Saudi Vision"]) + assert len(cleaned) == 2 + assert cleaned[0].text == "رؤية 2030" + assert cleaned[1].text == "Saudi Vision" + + +def test_enricher_registration(): + """Confirm both enrichers are registered in the global registry.""" + from flowsint_enrichers import ENRICHER_REGISTRY + + assert ENRICHER_REGISTRY.enricher_exists("individual_to_sabq") + assert ENRICHER_REGISTRY.enricher_exists("phrase_to_sabq") diff --git a/flowsint-enrichers/tests/enrichers/test_arabic_tweets.py b/flowsint-enrichers/tests/enrichers/test_arabic_tweets.py new file mode 100644 index 000000000..ce86060c0 --- /dev/null +++ b/flowsint-enrichers/tests/enrichers/test_arabic_tweets.py @@ -0,0 +1,75 @@ +"""Tests for Arabic Twitter (Nitter) enrichers.""" + +from unittest.mock import MagicMock, patch + +import pytest + +from flowsint_enrichers.individual.to_arabic_tweets import ( + IndividualToArabicTweetsEnricher, +) +from flowsint_enrichers.phrase.to_arabic_tweets import ( + PhraseToArabicTweetsEnricher, +) +from flowsint_types import Individual, Phrase + + +FIXTURE_HITS = [ + { + "url": "https://x.com/someuser/status/123", + "title": "Arabic tweet about: query", + "snippet": "Tweet body in Arabic.", + "via": "https://nitter.privacydev.net", + }, +] + + +def _make_enricher(cls): + enricher = cls(sketch_id="s", scan_id="x") + enricher._graph_service = MagicMock() + return enricher + + +@pytest.mark.asyncio +async def test_individual_to_arabic_tweets_scan(): + enricher = _make_enricher(IndividualToArabicTweetsEnricher) + individual = Individual(first_name="A", last_name="B", full_name="A B") + with patch( + "flowsint_enrichers.individual.to_arabic_tweets.NitterArabicTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([individual]) + assert len(results) == 1 + assert str(results[0].url) == "https://x.com/someuser/status/123" + + +@pytest.mark.asyncio +async def test_phrase_to_arabic_tweets_scan(): + enricher = _make_enricher(PhraseToArabicTweetsEnricher) + phrase = Phrase(text="Saudi Vision") + with patch( + "flowsint_enrichers.phrase.to_arabic_tweets.NitterArabicTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([phrase]) + assert len(results) == 1 + + +@pytest.mark.asyncio +async def test_arabic_tweets_uses_correct_rel_label(): + enricher = _make_enricher(IndividualToArabicTweetsEnricher) + individual = Individual(first_name="A", last_name="B", full_name="A B") + with patch( + "flowsint_enrichers.individual.to_arabic_tweets.NitterArabicTool" + ) as MockTool: + MockTool.return_value.launch.return_value = FIXTURE_HITS + results = await enricher.scan([individual]) + enricher.postprocess(results, [individual]) + rel_calls = enricher._graph_service.create_relationship.call_args_list + assert rel_calls and rel_calls[0].kwargs["rel_label"] == "MENTIONED_ON_TWITTER_AR" + + +def test_enricher_registration(): + from flowsint_enrichers import ENRICHER_REGISTRY + + assert ENRICHER_REGISTRY.enricher_exists("individual_to_arabic_tweets") + assert ENRICHER_REGISTRY.enricher_exists("phrase_to_arabic_tweets") diff --git a/uv.lock b/uv.lock index 0cb468da3..abcb7e9a5 100644 --- a/uv.lock +++ b/uv.lock @@ -787,6 +787,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/4e/8c/f3147f5c4b73e7550fe5f9352eaa956ae838d5c51eb58e7a25b9f3e2643b/decorator-5.2.1-py3-none-any.whl", hash = "sha256:d316bb415a2d9e2d2b3abcc4084c6502fc09240e292cd76a76afc106a1c8e04a", size = 9190, upload-time = "2025-02-24T04:41:32.565Z" }, ] +[[package]] +name = "defusedxml" +version = "0.7.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/0f/d5/c66da9b79e5bdb124974bfe172b4daf3c984ebd9c2a06e2b8a4dc7331c72/defusedxml-0.7.1.tar.gz", hash = "sha256:1bb3032db185915b62d7c6209c5a8792be6a32ab2fedacc84e01b52c51aa3e69", size = 75520, upload-time = "2021-03-08T10:59:26.269Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/07/6c/aa3f2f849e01cb6a001cd8554a88d4c77c5c1a31c95bdf1cf9301e6d9ef4/defusedxml-0.7.1-py2.py3-none-any.whl", hash = "sha256:a352e7e428770286cc899e2542b6cdaedb2b4953ff269a210103ec58f6198a61", size = 25604, upload-time = "2021-03-08T10:59:24.45Z" }, +] + [[package]] name = "dnspython" version = "2.8.0" @@ -1243,6 +1252,7 @@ name = "flowsint-enrichers" version = "1.2.8" source = { editable = "flowsint-enrichers" } dependencies = [ + { name = "defusedxml" }, { name = "dnspython" }, { name = "flowsint-core" }, { name = "flowsint-types" }, @@ -1275,6 +1285,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "defusedxml", specifier = ">=0.7,<0.8" }, { name = "dnspython", specifier = ">=2.4,<3.0" }, { name = "flowsint-core", editable = "flowsint-core" }, { name = "flowsint-types", editable = "flowsint-types" },