From 4ce8d99352a8412ffc45a5c7d35b47cdfdfd9b65 Mon Sep 17 00:00:00 2001 From: Dimitris Marakomihelakis Date: Wed, 27 May 2026 20:27:48 +0300 Subject: [PATCH] feature: add keywords parameter and filter by its matching #979 (#2702) * feature: add keywords parameter and filter by its matching #979 * fixes & nits --- README.md | 4 + maigret/checking.py | 40 +++++-- maigret/maigret.py | 9 ++ maigret/notify.py | 35 ++++-- maigret/resources/db_meta.json | 2 +- maigret/result.py | 31 ++++++ tests/test_cli.py | 1 + tests/test_keyword_filtering.py | 190 ++++++++++++++++++++++++++++++++ 8 files changed, 293 insertions(+), 19 deletions(-) create mode 100644 tests/test_keyword_filtering.py diff --git a/README.md b/README.md index 3e43b59..a24188c 100644 --- a/README.md +++ b/README.md @@ -209,6 +209,10 @@ maigret user --tags photo,dating # search on sites marked with tag us maigret user --tags us +# highlight sites whose page also mentions specific keywords +maigret user --keywords python rust +# keyword-matched sites are shown with "[++]" in bright green + # search for three usernames on all available sites maigret user1 user2 user3 -a diff --git a/maigret/checking.py b/maigret/checking.py index 8ad0554..c9cf80f 100644 --- a/maigret/checking.py +++ b/maigret/checking.py @@ -28,7 +28,7 @@ from . import errors from .activation import ParsingActivator, import_aiohttp_cookies from .errors import CheckError from .executors import AsyncioQueueGeneratorExecutor -from .result import MaigretCheckResult, MaigretCheckStatus +from .result import MaigretCheckResult, MaigretCheckStatus, KeywordMatchStatus from .sites import MaigretDatabase, MaigretSite from .types import QueryOptions, QueryResultWrapper from .utils import ascii_data_display, get_random_user_agent, is_plausible_username @@ -697,6 +697,26 @@ def process_site_result( logger.debug(presense_flag) break + + # Keyword detection logic + keywords = results_info.get("keywords", []) + keyword_match_status = None + + if keywords and html_text: + keywords_found = [] + for keyword in keywords: + if keyword.lower() in html_text.lower(): + keywords_found.append(keyword) + + if keywords_found: + keyword_match_status = KeywordMatchStatus.KEYWORD_FOUND + logger.debug(f"Keywords found in {site.name}: {keywords_found}") + else: + keyword_match_status = KeywordMatchStatus.KEYWORDS_NOT_FOUND + logger.debug(f"No keywords found in {site.name}") + else: + keyword_match_status = KeywordMatchStatus.NO_KEYWORDS + def build_result(status, **kwargs): return MaigretCheckResult( username, @@ -705,20 +725,17 @@ def process_site_result( status, query_time=response_time, tags=fulltags, + keywords=keywords, + keyword_match_status=keyword_match_status, **kwargs, ) if check_error: logger.warning(check_error) - result = MaigretCheckResult( - username, - site_name, - url, + result = build_result( MaigretCheckStatus.UNKNOWN, - query_time=response_time, error=check_error, context=str(check_error), - tags=fulltags, ) elif check_type == "message": # Checks if the error message is in the HTML @@ -781,6 +798,7 @@ def make_site_result( # Record URL of main site and username results_site["site"] = site results_site["username"] = username + results_site["keywords"] = kwargs.get('keywords', []) results_site["parsing_enabled"] = options["parsing"] results_site["url_main"] = site.url_main results_site["cookies"] = ( @@ -951,8 +969,9 @@ def make_site_result( async def check_site_for_username( site, username, options: QueryOptions, logger, query_notify, *args, **kwargs ) -> Tuple[str, QueryResultWrapper]: + keywords = kwargs.get('keywords') default_result = make_site_result( - site, username, options, logger, retry=kwargs.get('retry') + site, username, options, logger, retry=kwargs.get('retry'), keywords=keywords ) # future = default_result.get("future") # if not future: @@ -1046,6 +1065,7 @@ async def maigret( retries=0, check_domains=False, cloudflare_bypass: Optional[Dict[str, Any]] = None, + keywords=None, *args, **kwargs, ) -> QueryResultWrapper: @@ -1070,6 +1090,9 @@ async def maigret( Default is 100. no_progressbar -- Displaying of ASCII progressbar during scanner. cookies -- Filename of a cookie jar file to use for each request. + keywords -- List of keywords to search for in HTML content. + Default is None. + *args, **kwargs -- Additional arguments. Return Value: Dictionary containing results from report. Key of dictionary is the name @@ -1175,6 +1198,7 @@ async def maigret( { 'default': (sitename, default_result), 'retry': retries - attempts + 1, + 'keywords': keywords, }, ) diff --git a/maigret/maigret.py b/maigret/maigret.py index 68289a5..43fd6c0 100755 --- a/maigret/maigret.py +++ b/maigret/maigret.py @@ -328,6 +328,14 @@ def setup_arguments_parser(settings: Settings): default='', help="Specify tags to exclude from search (blacklist).", ) + filter_group.add_argument( + "--keywords", + nargs='+', + metavar='KEYWORD', + dest="keywords", + default=[], + help="Specify keywords to search for in HTML content. Sites containing both username AND any keyword get special highlighting. e.g. --keywords tech python", + ) filter_group.add_argument( "--site", action="append", @@ -853,6 +861,7 @@ async def main(): retries=args.retries, check_domains=args.with_domains, cloudflare_bypass=cf_bypass_config, + keywords=getattr(args, 'keywords', []) ) if not args.ai: diff --git a/maigret/notify.py b/maigret/notify.py index e48aa78..6e19881 100644 --- a/maigret/notify.py +++ b/maigret/notify.py @@ -7,7 +7,7 @@ import sys from colorama import Fore, Style, init -from .result import MaigretCheckStatus +from .result import MaigretCheckStatus, KeywordMatchStatus from .utils import get_dict_ascii_tree @@ -253,15 +253,30 @@ class QueryNotifyPrint(QueryNotify): # Output to the terminal is desired. if result.status == MaigretCheckStatus.CLAIMED: - color = Fore.BLUE if is_similar else Fore.GREEN - status = "?" if is_similar else "+" - notify = self.make_terminal_notify( - status, - result.site_name, - color, - color, - result.site_url_user + ids_data_text, - ) + # Check if this is a keyword match + if (result.keyword_match_status == KeywordMatchStatus.KEYWORD_FOUND and + result.keywords): + # Keyword-context match: site contains username + at least one keyword + color = Fore.LIGHTGREEN_EX + status = "++" + notify = self.make_terminal_notify( + status, + result.site_name, + color, + color, + result.site_url_user + ids_data_text, + ) + else: + # Normal claimed site + color = Fore.BLUE if is_similar else Fore.GREEN + status = "?" if is_similar else "+" + notify = self.make_terminal_notify( + status, + result.site_name, + color, + color, + result.site_url_user + ids_data_text, + ) elif result.status == MaigretCheckStatus.AVAILABLE: if not self.print_found_only: notify = self.make_terminal_notify( diff --git a/maigret/resources/db_meta.json b/maigret/resources/db_meta.json index 4e129a4..d6faf4f 100644 --- a/maigret/resources/db_meta.json +++ b/maigret/resources/db_meta.json @@ -1,6 +1,6 @@ { "version": 1, - "updated_at": "2026-05-25T13:29:37Z", + "updated_at": "2026-05-27T12:49:06Z", "sites_count": 3158, "min_maigret_version": "0.6.1", "data_sha256": "000f3174949a442f1f881fcc05c0869fee522b0414b6c093dfc0bdaa41303ec2", diff --git a/maigret/result.py b/maigret/result.py index 5346d6c..d911a8d 100644 --- a/maigret/result.py +++ b/maigret/result.py @@ -5,6 +5,26 @@ This module defines various objects for recording the results of queries. from enum import Enum +class KeywordMatchStatus(Enum): + """Keyword Match Status Enumeration. + + Describes the status of keyword matching for a given site. + """ + + NO_KEYWORDS = "No Keywords" + KEYWORD_FOUND = "Keyword Found" + KEYWORDS_NOT_FOUND = "Keywords Not Found" + + def __str__(self): + """Convert Object To String. + + Keyword Arguments: + self -- This object. + + Return Value: + Nicely formatted string to get information about this object. + """ + return self.value class MaigretCheckStatus(Enum): """Query Status Enumeration. @@ -45,6 +65,8 @@ class MaigretCheckResult: context=None, error=None, tags=[], + keywords=None, + keyword_match_status=None ): """ Keyword Arguments: @@ -67,6 +89,11 @@ class MaigretCheckResult: Default of None. ids_data -- Extracted from website page info about other usernames and inner ids. + keywords -- List of keywords to search for in page content. + Default of None. + keyword_match_status -- Enumeration of type KeywordMatchStatus() + indicating keyword matching status. + Default of None. Return Value: Nothing. @@ -81,6 +108,8 @@ class MaigretCheckResult: self.ids_data = ids_data self.tags = tags self.error = error + self.keywords = keywords or [] + self.keyword_match_status = keyword_match_status or KeywordMatchStatus.NO_KEYWORDS def json(self): return { @@ -90,6 +119,8 @@ class MaigretCheckResult: "status": str(self.status), "ids": self.ids_data or {}, "tags": self.tags, + "keywords": self.keywords, + "keyword_match_status": str(self.keyword_match_status) } def is_found(self): diff --git a/tests/test_cli.py b/tests/test_cli.py index e040a73..4c6b8ee 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -54,6 +54,7 @@ DEFAULT_ARGS: Dict[str, Any] = { 'no_autoupdate': False, 'force_update': False, 'cloudflare_bypass': False, + 'keywords': [], } diff --git a/tests/test_keyword_filtering.py b/tests/test_keyword_filtering.py new file mode 100644 index 0000000..044974a --- /dev/null +++ b/tests/test_keyword_filtering.py @@ -0,0 +1,190 @@ +from unittest.mock import Mock + +from maigret.errors import CheckError +from maigret.notify import QueryNotifyPrint +from maigret.result import MaigretCheckStatus, MaigretCheckResult, KeywordMatchStatus +from maigret.sites import MaigretSite +from maigret.checking import process_site_result + + +def test_keyword_match_status_enum(): + assert KeywordMatchStatus.NO_KEYWORDS.value == "No Keywords" + assert KeywordMatchStatus.KEYWORD_FOUND.value == "Keyword Found" + assert KeywordMatchStatus.KEYWORDS_NOT_FOUND.value == "Keywords Not Found" + assert str(KeywordMatchStatus.NO_KEYWORDS) == "No Keywords" + assert str(KeywordMatchStatus.KEYWORD_FOUND) == "Keyword Found" + assert str(KeywordMatchStatus.KEYWORDS_NOT_FOUND) == "Keywords Not Found" + + +def test_result_default_keyword_fields(): + result = MaigretCheckResult( + username="test", + site_name="SITE", + site_url_user="http://example.com/test", + status=MaigretCheckStatus.CLAIMED, + ) + assert result.keywords == [] + assert result.keyword_match_status == KeywordMatchStatus.NO_KEYWORDS + + +def test_result_with_keywords_no_match(): + result = MaigretCheckResult( + username="test", + site_name="SITE", + site_url_user="http://example.com/test", + status=MaigretCheckStatus.CLAIMED, + keywords=["nothing"], + keyword_match_status=KeywordMatchStatus.KEYWORDS_NOT_FOUND, + ) + assert result.keywords == ["nothing"] + assert result.keyword_match_status == KeywordMatchStatus.KEYWORDS_NOT_FOUND + + +def test_result_with_keywords_match(): + result = MaigretCheckResult( + username="test", + site_name="SITE", + site_url_user="http://example.com/test", + status=MaigretCheckStatus.CLAIMED, + keywords=["tech", "python"], + keyword_match_status=KeywordMatchStatus.KEYWORD_FOUND, + ) + assert result.keywords == ["tech", "python"] + assert result.keyword_match_status == KeywordMatchStatus.KEYWORD_FOUND + assert result.is_found() is True + + +def test_result_json_includes_keywords(): + result = MaigretCheckResult( + username="test", + site_name="SITE", + site_url_user="http://example.com/test", + status=MaigretCheckStatus.CLAIMED, + keywords=["tech"], + keyword_match_status=KeywordMatchStatus.KEYWORD_FOUND, + ) + data = result.json() + assert data["keywords"] == ["tech"] + assert data["keyword_match_status"] == "Keyword Found" + + +def test_notify_claimed_keyword_match(): + n = QueryNotifyPrint(color=False) + result = MaigretCheckResult( + username="test", + site_name="SITE", + site_url_user="http://example.com/test", + status=MaigretCheckStatus.CLAIMED, + keywords=["tech"], + keyword_match_status=KeywordMatchStatus.KEYWORD_FOUND, + ) + assert n.update(result) == "[++] SITE: http://example.com/test" + + +def test_notify_claimed_no_keywords(): + n = QueryNotifyPrint(color=False) + result = MaigretCheckResult( + username="test", + site_name="SITE", + site_url_user="http://example.com/test", + status=MaigretCheckStatus.CLAIMED, + keywords=[], + keyword_match_status=KeywordMatchStatus.NO_KEYWORDS, + ) + assert n.update(result) == "[+] SITE: http://example.com/test" + + +def test_notify_claimed_keywords_not_found(): + n = QueryNotifyPrint(color=False) + result = MaigretCheckResult( + username="test", + site_name="SITE", + site_url_user="http://example.com/test", + status=MaigretCheckStatus.CLAIMED, + keywords=["nonexistent"], + keyword_match_status=KeywordMatchStatus.KEYWORDS_NOT_FOUND, + ) + assert n.update(result) == "[+] SITE: http://example.com/test" + + +def test_notify_available(): + n = QueryNotifyPrint(color=False) + result = MaigretCheckResult( + username="test", + site_name="SITE", + site_url_user="http://example.com/test", + status=MaigretCheckStatus.AVAILABLE, + ) + assert n.update(result) == "[-] SITE: Not found!" + + +def test_notify_unknown(): + n = QueryNotifyPrint(color=False) + result = MaigretCheckResult( + username="test", + site_name="SITE", + site_url_user="http://example.com/test", + status=MaigretCheckStatus.UNKNOWN, + ) + result.error = CheckError("Type", "Reason") + assert n.update(result) == "[?] SITE: Type error: Reason" + + +# --------------------------------------------------------------------------- +# Integration tests — exercise the keyword detection block inside +# process_site_result with real ``html_text`` and a MaigretSite object. +# --------------------------------------------------------------------------- + +def _make_site(data_overrides=None): + base = { + "url": "https://x/{username}", + "urlMain": "https://x", + "checkType": "status_code", + "usernameClaimed": "a", + "usernameUnclaimed": "b", + } + if data_overrides: + base.update(data_overrides) + return MaigretSite("TestSite", base) + + +def test_process_site_result_no_keywords_yields_no_keywords(): + site = _make_site({"checkType": "status_code"}) + info = {"username": "a", "parsing_enabled": False, "url_user": "https://x/a", "keywords": []} + out = process_site_result(("python developer", 200, None), Mock(), Mock(), info, site) + assert out["status"].keyword_match_status == KeywordMatchStatus.NO_KEYWORDS + + +def test_process_site_result_keyword_found_in_html(): + site = _make_site({"checkType": "status_code"}) + info = {"username": "a", "parsing_enabled": False, "url_user": "https://x/a", "keywords": ["python"]} + out = process_site_result(("I love python programming", 200, None), Mock(), Mock(), info, site) + assert out["status"].keyword_match_status == KeywordMatchStatus.KEYWORD_FOUND + + +def test_process_site_result_keyword_not_found_in_html(): + site = _make_site({"checkType": "status_code"}) + info = {"username": "a", "parsing_enabled": False, "url_user": "https://x/a", "keywords": ["python"]} + out = process_site_result(("I love rust programming", 200, None), Mock(), Mock(), info, site) + assert out["status"].keyword_match_status == KeywordMatchStatus.KEYWORDS_NOT_FOUND + + +def test_process_site_result_keyword_case_insensitive(): + site = _make_site({"checkType": "status_code"}) + info = {"username": "a", "parsing_enabled": False, "url_user": "https://x/a", "keywords": ["Python"]} + out = process_site_result(("I love python programming", 200, None), Mock(), Mock(), info, site) + assert out["status"].keyword_match_status == KeywordMatchStatus.KEYWORD_FOUND + + +def test_process_site_result_empty_html_yields_no_keywords(): + site = _make_site({"checkType": "status_code"}) + info = {"username": "a", "parsing_enabled": False, "url_user": "https://x/a", "keywords": ["python"]} + out = process_site_result(("", 200, None), Mock(), Mock(), info, site) + assert out["status"].keyword_match_status == KeywordMatchStatus.NO_KEYWORDS + + +def test_process_site_result_keywords_one_of_many_found(): + site = _make_site({"checkType": "status_code"}) + info = {"username": "a", "parsing_enabled": False, "url_user": "https://x/a", "keywords": ["rust", "python"]} + out = process_site_result(("I love rust programming", 200, None), Mock(), Mock(), info, site) + assert out["status"].keyword_match_status == KeywordMatchStatus.KEYWORD_FOUND