diff --git a/ddgs/cli.py b/ddgs/cli.py index 36cf2f3..fe0e5f9 100644 --- a/ddgs/cli.py +++ b/ddgs/cli.py @@ -1,5 +1,6 @@ """CLI tool for DDGS.""" +import contextlib import csv import json import logging @@ -10,6 +11,7 @@ import click import primp +from feedgenerator import Rss201rev2Feed # type: ignore[import-not-found] from . import __version__ from .ddgs import DDGS @@ -44,12 +46,15 @@ def _convert_tuple_to_csv(_ctx: click.Context, _param: click.Parameter, value: t def _save_data(query: str, data: list[dict[str, str]], function_name: str, filename: str | None) -> None: - filename, ext = filename.rsplit(".", 1) if filename and filename.endswith((".csv", ".json")) else (None, filename) + valid_exts = (".csv", ".json", ".rss") + filename, ext = filename.rsplit(".", 1) if filename and filename.endswith(valid_exts) else (None, filename) filename = filename if filename else f"{function_name}_{query}_{datetime.now(tz=timezone.utc):%Y%m%d_%H%M%S}" if ext == "csv": _save_csv(f"{filename}.{ext}", data) elif ext == "json": _save_json(f"{filename}.{ext}", data) + elif ext == "rss": + _save_rss(f"{filename}.{ext}", data) def _save_json(jsonfile: str | Path, data: list[dict[str, str]]) -> None: @@ -66,6 +71,45 @@ def _save_csv(csvfile: str | Path, data: list[dict[str, str]]) -> None: writer.writerows(data) +def _save_rss(rssfile: str | Path, data: list[dict[str, str]]) -> None: + """Save search results as an RSS 2.0 feed. + + Args: + rssfile: Path to save the RSS file. + data: List of search result dictionaries. + + """ + feed = Rss201rev2Feed( + title="DDGS Search Results", + link="https://github.com/deedy5/ddgs", + description="Search results from DDGS metasearch", + ) + + for result in data: + # Extract common fields with fallbacks + title = result.get("title", "No Title") + link = result.get("href") or result.get("url") or result.get("image") or "" + description = result.get("body") or result.get("content") or result.get("description") or "" + + # Get publication date if available (news uses "date", videos use "published") + pub_date = None + date_str = result.get("date") or result.get("published") + if date_str: + with contextlib.suppress(ValueError, AttributeError): + # Try to parse the date string + pub_date = datetime.fromisoformat(date_str.replace("Z", "+00:00")) + + feed.add_item( + title=title, + link=link, + description=description, + pubdate=pub_date, + ) + + with Path(rssfile).open("w", encoding="utf-8") as file: + feed.write(file, "utf-8") + + def _print_data(data: list[dict[str, str]]) -> None: if data: for i, e in enumerate(data, start=1): @@ -198,7 +242,7 @@ def version() -> str: multiple=True, callback=_convert_tuple_to_csv, ) -@click.option("-o", "--output", help="csv, json or filename.csv|json (save the results to a csv or json file)") +@click.option("-o", "--output", help="csv, json, rss or filename.csv|json|rss (save the results to a file)") @click.option("-d", "--download", is_flag=True, default=False, help="download results. -dd to set custom directory") @click.option("-dd", "--download-directory", help="Specify custom download directory") @click.option("-th", "--threads", default=10, help="download threads, default=10") @@ -295,7 +339,7 @@ def text( "--license_image", type=click.Choice(["any", "Public", "Share", "ShareCommercially", "Modify", "ModifyCommercially"]), ) -@click.option("-o", "--output", help="csv, json or filename.csv|json (save the results to a csv or json file)") +@click.option("-o", "--output", help="csv, json, rss or filename.csv|json|rss (save the results to a file)") @click.option("-d", "--download", is_flag=True, default=False, help="download results. -dd to set custom directory") @click.option("-dd", "--download-directory", help="Specify custom download directory") @click.option("-th", "--threads", default=10, help="download threads, default=10") @@ -375,7 +419,7 @@ def images( @click.option("-res", "--resolution", type=click.Choice(["high", "standart"])) @click.option("-d", "--duration", type=click.Choice(["short", "medium", "long"])) @click.option("-lic", "--license_videos", type=click.Choice(["creativeCommon", "youtube"])) -@click.option("-o", "--output", help="csv, json or filename.csv|json (save the results to a csv or json file)") +@click.option("-o", "--output", help="csv, json, rss or filename.csv|json|rss (save the results to a file)") @click.option("-pr", "--proxy", help="the proxy to send requests, example: socks5h://127.0.0.1:9150") @click.option("-v", "--verify", default=True, help="verify SSL when making the request") def videos( @@ -432,7 +476,7 @@ def videos( multiple=True, callback=_convert_tuple_to_csv, ) -@click.option("-o", "--output", help="csv, json or filename.csv|json (save the results to a csv or json file)") +@click.option("-o", "--output", help="csv, json, rss or filename.csv|json|rss (save the results to a file)") @click.option("-pr", "--proxy", help="the proxy to send requests, example: socks5h://127.0.0.1:9150") @click.option("-v", "--verify", default=True, help="verify SSL when making the request") def news( @@ -480,7 +524,7 @@ def news( multiple=True, callback=_convert_tuple_to_csv, ) -@click.option("-o", "--output", help="csv, json or filename.csv|json (save the results to a csv or json file)") +@click.option("-o", "--output", help="csv, json, rss or filename.csv|json|rss (save the results to a file)") @click.option("-pr", "--proxy", help="the proxy to send requests, example: socks5h://127.0.0.1:9150") @click.option("-v", "--verify", default=True, help="verify SSL when making the request") def books( diff --git a/pyproject.toml b/pyproject.toml index 0059a3c..7e99bb7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,6 +32,7 @@ dependencies = [ "lxml>=4.9.4", "httpx[http2,socks,brotli]>=0.28.1", # temporarily "fake-useragent>=2.2.0", + "feedgenerator>=2.0.0", ] dynamic = ["version"] diff --git a/tests/cli_test.py b/tests/cli_test.py index 9c3fb44..d94be1a 100644 --- a/tests/cli_test.py +++ b/tests/cli_test.py @@ -7,7 +7,7 @@ from click.testing import CliRunner from ddgs import DDGS, __version__ -from ddgs.cli import _download_results, _save_csv, _save_json, cli +from ddgs.cli import _download_results, _save_csv, _save_json, _save_rss, cli runner = CliRunner() TEXT_RESULTS = [] @@ -77,6 +77,19 @@ def test_save_json(tmp_path: Path) -> None: assert temp_file.exists() +@pytest.mark.dependency(depends=["test_get_text"]) +def test_save_rss(tmp_path: Path) -> None: + temp_file = tmp_path / "test_rss.rss" + _save_rss(temp_file, TEXT_RESULTS) + assert temp_file.exists() + # Verify it's valid XML by reading it + content = temp_file.read_text(encoding="utf-8") + assert "" in content + assert "" in content + + @pytest.mark.dependency(depends=["test_get_text"]) def test_text_download() -> None: pathname = pathlib.Path("text_downloads")