commit 0468dd4e9d5ef77fbe9e5b2dab70a6ef05cc1317 Author: Metona Team Date: Sat Aug 1 17:02:34 2026 +0800 Initial commit: SearXNG CLI Toolkit v1.6.0 Multi-instance failover, exponential-backoff retry, SQLite cache, batch mode, domain filter, cross-engine dedup, result sorting, CSV export, structured logging, enhanced Markdown conversion, 155 pytest tests, Gitea Actions CI diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..e78b297 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,39 @@ +name: CI + +on: + push: + branches: [main, master] + pull_request: + branches: [main, master] + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"] + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install pytest requests beautifulsoup4 + # tomli backport for Python 3.8-3.10 (tomllib is stdlib in 3.11+) + pip install tomli || true + + - name: Run tests + run: python -m pytest -q + + - name: Verify scripts run + run: | + python scripts/search.py --version + python scripts/fetch.py --version diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8927f81 --- /dev/null +++ b/.gitignore @@ -0,0 +1,38 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so + +# Distribution / packaging +build/ +dist/ +*.egg-info/ +*.egg + +# Virtual environments +.venv/ +venv/ +env/ + +# Testing +.pytest_cache/ +.coverage +htmlcov/ +.tox/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# OS +.DS_Store +Thumbs.db + +# Project-specific +searxng.toml +instances.txt +*.db +.cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..15f3364 --- /dev/null +++ b/README.md @@ -0,0 +1,190 @@ +# SearXNG CLI Toolkit + +[![CI](https://git.metona.cn/MetonaTeam/searxng-use-cli/actions/workflows/ci.yml/badge.svg)](https://git.metona.cn/MetonaTeam/searxng-use-cli/actions) +[![Python 3.8+](https://img.shields.io/badge/python-3.8+-blue.svg)](https://www.python.org/downloads/) +[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) + +A privacy-respecting metasearch CLI toolkit that runs against your own SearXNG instance. Zero external dependencies for search (stdlib only), optional `requests` + `beautifulsoup4` for enhanced page fetching. Works with any AI agent (Hermes, Claude Code, Codex, OpenCode, Cursor, Trae) or directly from your terminal. + +## Features + +**Search & results** +- Multi-instance failover with parallel probing +- Exponential-backoff retry on transient errors (429/5xx/connection) +- `--verify` health-check mode (reachability / JSON-API / latency / POST / engines / auth) +- Cross-engine result deduplication (default on; `--no-dedup` disables) +- Result sorting (`--sort-by {score,date,engine,none}`) +- Domain allowlist/blocklist (`--include-domain` / `--exclude-domain`) +- Batch mode (`--queries-file`) + +**Output formats** +- JSON (default), brief, urls, CSV +- Enhanced Markdown conversion (GFM tables, code blocks, blockquotes, nested lists, definition lists) +- Structured JSON error output for machine-readable failure reporting + +**Caching & config** +- SQLite result caching (`--cache-ttl`) with TTL management +- `searxng.toml` config file for defaults; `--config FILE` for explicit loading +- Instance resolution: `-i` → `SEARXNG_INSTANCE` env → config file + +**Network & auth** +- Proxy support (`--proxy`) for both search and fetch +- Auth via CLI flag, file, or env var (`--auth-bearer` / `--auth-basic` + `*-file` variants) +- Credentials-file permission warning (POSIX) + +**Engineering** +- Shared `common.py` (unified retry/charset/auth/logging) +- Structured logging (`--verbose` / `--quiet`) +- 155 unit + integration tests with pytest + +## Quick Start + +```bash +# Prerequisites: Python 3.8+ (optional: pip install requests beautifulsoup4) + +# Search against YOUR instance (instance URL is required) +python scripts/search.py -q "python asyncio tutorial" -i https://my-searxng.example.com + +# Multiple instances for failover (comma-separated) +python scripts/search.py -q "rust memory safety" \ + -i https://a.example.com,https://b.example.com --format brief + +# Search + auto-fetch top 3 result pages +python scripts/search.py -q "climate policy" -i https://my-searxng.example.com --fetch 3 + +# Fetch a web page +python scripts/fetch.py -u "https://example.com" --extract markdown + +# Skip -i via env var +export SEARXNG_INSTANCE="https://my-searxng.example.com" +python scripts/search.py -q "python asyncio tutorial" +``` + +## Installation + +```bash +git clone https://git.metona.cn/MetonaTeam/searxng-use-cli.git +cd searxng-use-cli + +# Zero deps — search.py runs on stdlib alone +python scripts/search.py --version + +# Optional: enhanced fetch quality +pip install requests beautifulsoup4 +``` + +## Usage + +### search.py — Execute SearXNG Search + +```bash +python scripts/search.py -q "your query" -i https://your-instance \ + [--format json|brief|urls|csv] \ + [--engines google,bing,brave] \ + [--time-range day|month|year|none] \ + [--language en] \ + [--sort-by score|date|engine|none] \ + [--no-dedup] \ + [--max-results 10] \ + [--fetch 3] \ + [--cache-ttl 30] \ + [--queries-file queries.txt] \ + [--include-domain example.com] \ + [--exclude-domain spam.com] \ + [--proxy http://corp:8080] \ + [--auth-bearer-file ~/.token] \ + [--verify] \ + [--verbose|-v] [--quiet] +``` + +### fetch.py — Fetch & Extract Web Page Content + +```bash +python scripts/fetch.py -u https://example.com \ + --extract text|html|markdown \ + [--encoding gbk] \ + [--max-size 5242880] \ + [--proxy http://corp:8080] \ + [--auth-bearer-file ~/.token] +``` + +## Configuration + +Instance URLs resolve in priority order: +1. `-i / --instance` (comma-separated for failover) +2. `SEARXNG_INSTANCE` environment variable +3. Config file: `./searxng.toml` → `~/.config/searxng-cli/searxng.toml` → `./instances.txt` → `~/.config/searxng-cli/instances.txt` + +Example `searxng.toml`: +```toml +[searxng] +instance = "https://my-searxng.example.com" +# or: instances = ["https://a.example.com", "https://b.example.com"] +engines = "google,bing,brave,duckduckgo,startpage,wikipedia,wikidata" +time_range = "year" +safesearch = 0 +format = "json" +cache_ttl = 30 +``` + +## Testing + +```bash +pip install pytest +pytest -q +``` + +155 tests cover: cache operations, auth resolution, domain filtering, Markdown conversion, search logic, integration flows, and logging configuration. + +## Project Structure + +``` +├── scripts/ +│ ├── search.py # Search with multi-instance failover, cache, batch, domain filter +│ ├── fetch.py # Web page fetcher with text/markdown extraction +│ ├── common.py # Shared utilities (auth, retry, charset, logging) +│ ├── cache.py # SQLite-backed result cache +│ └── _config.py # Version + User-Agent constants +├── tests/ # pytest unit + integration tests +├── .gitea/workflows/ # Gitea Actions CI +├── SKILL.md # Full skill documentation (agent-facing) +├── pytest.ini # Test configuration +└── README.md +``` + +## Defaults + +| Setting | Default | Flag | +|---------|---------|------| +| Instance | **required** | `-i` / `SEARXNG_INSTANCE` / config | +| Safe search | 0 (off) | `-s` | +| Time range | year | `-t` | +| Output format | json | `-f` | +| Engines | google,bing,brave,duckduckgo,startpage,wikipedia,wikidata | `--engines` | +| Sort | score descending | `--sort-by` | +| Dedup | on | `--no-dedup` | + +## Cross-Agent Compatibility + +These scripts are agent-agnostic — they work with any AI agent that can invoke terminal commands: + +| Agent | How to invoke | +|-------|--------------| +| Hermes | `python scripts/search.py -q "..." -i https://your-instance` | +| Claude Code | Same — call via terminal tool | +| Codex (OpenAI) | Same — call via terminal tool | +| OpenCode | Same — call via terminal tool | +| Cursor | Same — call via terminal tool | +| Trae | Same — call via terminal tool | +| Standalone (human) | Run directly in any terminal | + +Key design decisions for universal compatibility: +- Zero external dependencies (stdlib-only for `search.py`) +- Scripts self-inject their directory into `sys.path` — run from any working directory +- Stdout carries data (JSON/CSV/text), stderr carries progress/warnings +- Exit codes: 0=success, 1=fatal error, 2=no results/empty +- NO agent-specific API calls — purely CLI-based, portable across all agent platforms + +## License + +MIT diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..124b490 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,508 @@ +--- +name: searxng-use-cli +description: Use when you need to search the web via your OWN SearXNG instance (no public-instance discovery). 3 CLI scripts + a shared common.py module — execute privacy-respecting searches against a user-supplied instance (with multi-instance failover, 5xx/429 retry, auto-fetch) or via SEARXNG_INSTANCE env / config file, fetch/extract readable text or markdown from web pages. Zero-config replacement for proprietary search APIs. +version: 1.6.0 +author: Metona Team +license: MIT +platforms: [linux, macos, windows] +metadata: + skill: + tags: [search, searxng, cli, web-scraping, privacy, no-api-key] + related_skills: [] +--- + +# SearXNG CLI Toolkit + +## Overview + +SearXNG is a privacy-respecting metasearch engine that aggregates results from 70+ search services without tracking users. This skill provides three standalone Python CLI scripts — works with **any AI agent** (Hermes, Claude Code, Codex, OpenCode, Cursor, Trae, etc.) or directly from your terminal. + +**Public-instance discovery has been removed.** You must supply your own SearXNG instance URL (self-hosted or one you trust). This makes behavior deterministic and avoids depending on volatile public instances. + +**Key capabilities:** + +**Search & results** +- Multi-instance failover with parallel probing (faster failover, deterministic output order) +- Exponential-backoff retry on transient errors (429/5xx/connection) via shared `common.py` +- `--verify` health-check mode (reachability / JSON-API / latency / POST / engine list / auth status) +- Cross-engine result deduplication (default on; `--no-dedup` disables) — collapses duplicate URLs ignoring tracking params (`utm_*`, `gclid`, etc.) and fragments +- Result sorting (`--sort-by {score,date,engine,none}`; default: score descending) — applied after dedup, before `--max-results` +- Domain allowlist/blocklist (`--include-domain` / `--exclude-domain`) — case-insensitive, ignores leading `www.`, exclude wins on conflict +- Batch mode (`--queries-file`) — run multiple queries from a file in sequence, combined output + +**Output formats** +- JSON (default, rich metadata), brief (title+URL+snippet), urls (plain list), CSV (spreadsheet-ready) +- Enhanced Markdown conversion — nested ordered lists (numbered), mixed `ul`/`ol` nesting, `
`/`
`/`
` definition lists, GFM tables, fenced code blocks, blockquotes +- Structured JSON error output (in `--format json` mode) for machine-readable failure reporting + +**Caching & config** +- SQLite result caching (`--cache-ttl`) — identical queries within a TTL skip the network entirely; `--clear-cache` / `--cache-stats` manage it +- Config file (`searxng.toml`) pre-sets most flags; `--config FILE` loads a non-default config; `instances.txt` for plain URL lists +- Instance resolution priority: `-i` → `SEARXNG_INSTANCE` env → config file + +**Network & auth** +- Proxy support (`--proxy`) for both search and fetch (sets `HTTP_PROXY`/`HTTPS_PROXY`/`NO_PROXY`) +- Auth via CLI flag, file, or env var (`--auth-bearer` / `--auth-basic` + `*-file` variants) to avoid leaking secrets in shell history +- Credentials-file permission warning — `--auth-*-file` warns on stderr if the file is group/other-readable (POSIX only) + +**Engineering** +- Shared `common.py` module — unified retry/charset/auth/logging logic across both scripts +- `search.py --fetch` reuses `fetch.py`'s higher-quality text extractor (no code duplication) +- Structured logging (`--verbose` / `--quiet`) — three levels: default INFO (progress + warnings), `--verbose` DEBUG (HTTP detail, cache keys), `--quiet` WARNING (errors only). All log output to stderr; stdout reserved for data +- Engine/category whitespace normalization (`"google, bing"` → `"google,bing"`) +- `--time-range none` option to disable time filtering + +**Scripts + shared module:** +1. `search.py` — execute searches against a user-supplied instance, with multi-instance failover + exponential-backoff retry (429/5xx/connection) + auto-fetch + caching + batch + domain filtering +2. `fetch.py` — download and extract readable text or markdown from web pages +3. `common.py` — shared utilities (auth headers, charset detection, retry policy, fallback UAs, retry constants) used by both scripts +4. `cache.py` — SQLite-backed result cache (SHA-256 key, TTL, WAL mode) +5. `_config.py` — package constants (version, User-Agent) + +## Default settings + +`search.py` ships with opinionated defaults tuned for AI research: + +| Setting | Default | Flag to override | +|---------|---------|------------------| +| Instance | **required** — via `-i`, `SEARXNG_INSTANCE` env var, or config file | `-i / --instance` | +| Safe search | **0 (off)** | `-s / --safesearch {0,1,2}` | +| Time range | **year** | `-t / --time-range {day,month,year,none}` (none = disabled) | +| Output format | **json** | `-f / --format {json,brief,urls,csv}` | +| Engines | **google,bing,brave,duckduckgo,startpage,wikipedia,wikidata** | `--engines ` | + +## Quick Start + +```bash +# Prerequisites: Python 3.8+ +# Optional but recommended: +pip install requests beautifulsoup4 + +# 1. Search against YOUR instance (instance URL is required) +python scripts/search.py -q "python asyncio tutorial" -i https://my-searxng.example.com + +# 2. Multiple instances for failover (comma-separated) +python scripts/search.py -q "rust memory safety" \ + -i https://a.example.com,https://b.example.com --format brief + +# 3. Search + auto-fetch top 3 result pages in one command +python scripts/search.py -q "climate policy" -i https://my-searxng.example.com --fetch 3 + +# 4. Fetch a result page +python scripts/fetch.py -u "https://example.com" --extract text + +# 5. Skip -i by configuring the instance once (env var, current shell) +export SEARXNG_INSTANCE="https://my-searxng.example.com,https://backup.example.com" +python scripts/search.py -q "python asyncio tutorial" # -i not needed + +# 6. Or use a config file (./searxng.toml or ~/.config/searxng-cli/searxng.toml) +# [searxng] +# instance = "https://my-searxng.example.com" +# # or: instances = ["https://a.example.com", "https://b.example.com"] +# # Any flag below can also be pre-set here (engines, categories, language, +# # safesearch, time_range, method, format, timeout, max_retries, proxy, +# # cache_ttl, fetch, fetch_timeout, fetch_retries, max_size). +# # Explicit CLI flags always override config values. +# Plain list also works in ./instances.txt (one URL per line, # for comments) + +# 7. Cache results for 30 minutes (identical queries skip the network) +python scripts/search.py -q "python asyncio" -i https://s.example.com --cache-ttl 30 + +# 7b. Sort by date (newest first) or disable dedup for raw engine output +python scripts/search.py -q "ai news" -i https://s.example.com --sort-by date --no-dedup + +# 8. Batch: run queries from a file (one per line; blank/# lines skipped) +python scripts/search.py --queries-file queries.txt -i https://s.example.com --format json > batch.json + +# 9. Domain allowlist + blocklist (applied after search) +python scripts/search.py -q "rust async" -i https://s.example.com \ + --include-domain doc.rust-lang.org,wikipedia.org --exclude-domain pinterest.com + +# 10. Route through a corporate proxy (applies to search and fetch) +python scripts/search.py -q "ai news" -i https://s.example.com --proxy http://corp-proxy:8080 + +# 11. Auth from a file (avoids leaking tokens in shell history) +python scripts/search.py -q "test" -i https://private.example.com --auth-bearer-file ~/.searxng_token + +# 12. Cache management (no search performed) +python scripts/search.py --cache-stats # entry count, age, size, path +python scripts/search.py --clear-cache # delete all entries + +# 13. Export results as CSV (great for spreadsheets / data analysis) +python scripts/search.py -q "rust async" -i https://s.example.com --format csv > results.csv + +# 14. Use a specific config file (overrides auto-discovered searxng.toml) +python scripts/search.py --config ./my-config.toml -q "test" + +# 15. Control log verbosity on stderr +python scripts/search.py -q "test" -i https://s.example.com --verbose # debug detail +python scripts/search.py -q "test" -i https://s.example.com --quiet # errors only +``` + +**Dependency levels:** +| Level | Scripts | What you get | +|-------|---------|-------------| +| Zero deps (stdlib only) | `search.py`, `_config.py` | Full search + auto-fetch | +| `pip install requests` | `fetch.py` | Better HTTP (session reuse, redirect handling) | +| `pip install beautifulsoup4` | `fetch.py` | Higher-quality text extraction | + +## When to Use + +- **Web search without API keys** — programmatic search results against your own SearXNG instance +- **Privacy-conscious research** — queries routed through your instance, not ad-tech infrastructure +- **Scraping search results** — batch query multiple terms and collect structured results as JSON +- **Fetching search result pages** — follow links from search results and extract clean text + +**Don't use for:** +- High-frequency production search without a properly-scaled instance — respect your instance's rate limits +- Guaranteed uptime/accuracy — depends entirely on the instance you supply + +## Cross-Agent Compatibility + +These scripts are **agent-agnostic** — they work with any AI agent that can invoke terminal commands: + +| Agent | How to invoke | +|-------|--------------| +| **Hermes** | `python scripts/search.py -q "..." -i https://your-instance` | +| **Claude Code** | Same — call via terminal tool | +| **Codex (OpenAI)** | Same — call via terminal tool | +| **OpenCode** | Same — call via terminal tool | +| **Cursor** | Same — call via terminal tool | +| **Trae** | Same — call via terminal tool | +| **Standalone (human)** | Run directly in any terminal | + +**Key design decisions for universal compatibility:** +- Zero external dependencies (stdlib-only for `search.py`) +- Scripts inject their own directory into `sys.path`, so they run from **any** working directory +- Stdout carries data (JSON/text), stderr carries progress/warnings +- Exit codes: 0=success, 1=fatal error, 2=no results/empty +- NO agent-specific API calls or tool dependencies — purely CLI-based, portable across all agent platforms + +**Standalone usage (no agent):** +```bash +# Clone or download the scripts/ directory, then run from anywhere: +python /path/to/scripts/search.py -q "your query" -i https://your-instance +python /path/to/scripts/fetch.py -u "https://example.com" +``` + +## SearXNG Search API (Quick Reference) + +Source: [docs.searxng.org/dev/search_api.html](https://docs.searxng.org/dev/search_api.html) + +``` +GET/POST /search?q=&format=json GET /search +POST /search (form data) POST / +``` + +**Parameters** + +| Parameter | Required | Values | Description | +|-------------|----------|---------------------------|--------------------------------------------------| +| `q` | yes | string | Search query (supports engine syntax like `site:`)| +| `format` | no | `json`, `csv`, `rss` | Output format (**many instances only allow html**) | +| `categories`| no | comma-separated | e.g. `general,images,news,science` | +| `language` | no | language code | e.g. `en`, `zh-CN`, `de` | +| `pageno` | no | integer (default 1) | Result page number — **fully supported** by search.py | +| `time_range`| no | `day`, `month`, `year`, `none` | Time filter (search.py default: `year`; `none` = disabled) | +| `safesearch`| no | `0`, `1`, `2` | Safe search (search.py default: `0` = off) | +| `engines` | no | comma-separated | search.py default: `google,bing,brave,duckduckgo,startpage,wikipedia,wikidata` | + +### Authentication + +SearXNG supports `Authorization` headers for private instances (configured via `settings.yml` → `server.secret_key`). Both `search.py` and `fetch.py` accept: + +```bash +# Bearer token (most common for API-style auth) +python search.py -q "test" --auth-bearer "sk-abc123..." -i https://private-searx.example.com + +# Basic auth (base64-encoded user:pass) +python search.py -q "test" --auth-basic "admin:secret123" -i https://private-searx.example.com + +# fetch.py also supports auth for authenticated endpoints +python fetch.py -u "https://protected.example.com/page" --auth-bearer "tok_xxx" +``` + +If both are provided, Bearer takes precedence. Auth headers are forwarded to both search API calls and auto-fetched page requests. + +**Avoid leaking secrets in shell history** — credentials can be read from a file or env var instead of a CLI flag (priority: CLI flag > file > env var): + +```bash +# From a file (first non-empty, non-# line is used) +python search.py -q "test" -i https://private.example.com --auth-bearer-file ~/.searxng_token +python search.py -q "test" -i https://private.example.com --auth-basic-file ~/.searxng_auth + +# From an environment variable +export SEARXNG_BEARER_TOKEN="sk-abc123..." +export SEARXNG_BASIC_AUTH="admin:secret123" +python search.py -q "test" -i https://private.example.com # credentials picked up automatically +``` + +### JSON Response Structure + +```json +{ + "query": "searxng", + "number_of_results": 1234, + "results": [ + { + "title": "...", + "url": "https://...", + "content": "Snippet text...", + "engine": "google", + "score": 0.0, + "category": "general" + } + ], + "answers": [], + "corrections": [], + "suggestions": ["searxng docker", "searxng api"], + "infoboxes": [], + "unresponsive_engines": [["bing", "timeout"]] +} +``` + +**Key gotcha:** Many instances disable `format=json` in their settings. `search.py` auto-detects and falls back to HTML scraping. Enable `format: [html, json]` under `search.formats` in your instance's `settings.yml` to get rich JSON metadata. + +## Scripts + +All scripts live in `scripts/`; run with `python scripts/.py` from any directory. They import `_config.py` for shared constants and self-inject their own directory into `sys.path`. + +### 1. `search.py` — Execute SearXNG Search + +``` +usage: search.py [-h] [--query QUERY] [--instance URL] + [--categories CATS] [--language LANG] [--pageno N] + [--time-range {day,month,year,none}] [--safesearch {0,1,2}] + [--engines E] [--method {GET,POST}] [--max-results N] + [--format {json,brief,urls,csv}] [--snippet-len N] + [--fetch N] [--fetch-timeout SEC] [--fetch-retries N] + [--max-size BYTES] [--output FILE] [--timeout SEC] + [--retry N] [--fail-fast] [--serial] [--verify] + [--auth-bearer TOKEN] [--auth-bearer-file FILE] + [--auth-basic USER:PASS] [--auth-basic-file FILE] + [--proxy URL] [--include-domain DOMAINS] + [--exclude-domain DOMAINS] [--queries-file FILE] + [--cache-ttl MINUTES] [--clear-cache] [--cache-stats] + [--sort-by {score,date,engine,none}] [--no-dedup] + [--config FILE] [--verbose] [--quiet] [--version] +``` + +**What it does:** +1. Takes a search query and resolves one or more instance URLs (`-i`, `SEARXNG_INSTANCE`, or config file — comma-separated for failover) +2. **Multi-instance failover:** if an instance fails (429/5xx/timeout/captcha), automatically tries the next one +3. **Exponential backoff:** retries each instance up to 3 times with jitter on transient errors (429, 502, 503, 504, and connection errors) +4. **Parallel probing (default for 2+ instances):** queries every instance concurrently and returns the first *successful* result in your original instance order — this keeps output deterministic while drastically speeding up failover when an early instance is down/slow. Use `--serial` to disable. +5. Calls the SearXNG API (GET or POST) with `format=json` +6. Falls back to HTML scraping if JSON is blocked +7. HTML parser extracts results + suggestions + answers + infoboxes (full content, never truncated) +8. **Stable auto-fetch:** `--fetch 3` concurrently downloads top 3 result pages with retry, browser-UA fallback, CAPTCHA detection, and **`fetch.py`'s higher-quality text extractor** (the same engine `fetch.py` uses) +9. **Health-check mode:** `--verify` probes each instance (reachability / JSON-API support / latency / POST support / engine list / auth status) and prints a report, then exits without searching — use it to validate your instance list +10. **Result caching:** `--cache-ttl 30` stores results for 30 min; identical queries within the TTL skip the network entirely. Cache lives at `$SEARXNG_CACHE_DIR` or `~/.cache/searxng-cli/cache.db` (SQLite, WAL mode). `--clear-cache` / `--cache-stats` manage it without searching +11. **Batch mode:** `--queries-file FILE` reads one query per line (blank/`#` lines skipped) and runs them in sequence; output is a JSON array (or one block per query in brief/urls). A failed query is recorded but does not abort the batch +12. **Dedup + sort + domain filter:** After search (and cache), duplicate URLs are collapsed (default; `--no-dedup` disables), results are sorted (`--sort-by`; default: score descending), and then `--include-domain`/`--exclude-domain` filter by domain. Matching is case-insensitive and ignores a leading `www.`; when a domain is in both lists, exclude wins +13. **Proxy & auth:** `--proxy URL` routes both search and fetch through a proxy; `--auth-bearer` / `--auth-basic` (plus `*-file` variants and `SEARXNG_BEARER_TOKEN` / `SEARXNG_BASIC_AUTH` env vars) supply credentials without leaking them via shell history +14. **Config defaults:** `searxng.toml` may pre-set most flags (engines, categories, language, safesearch, time_range, method, format, sort_by, timeout, max_retries, proxy, cache_ttl, fetch, fetch_timeout, fetch_retries, max_size); explicit CLI flags always win +15. **Structured errors:** in `--format json` mode, failures print a JSON object `{"error": "...", "exit_code": N, "query": "..."}` to stdout so agents can parse them + +**Key options:** +- `--query "your search"` — **required unless** `--verify`, `--queries-file`, `--clear-cache`, or `--cache-stats` is used +- `--instance https://searx.example.org` — **required unless** `SEARXNG_INSTANCE` env var or a config file supplies it; comma-separated list enables failover +- `--queries-file FILE` — read queries from a file (one per line; blank/`#` skipped) and run them in sequence; overrides `--query` +- `--engines google,duckduckgo` — restrict to specific search engines (whitespace around commas is auto-stripped; see default list above) +- `--method POST` — use POST instead of GET (better for long queries) +- `--categories general,news` — comma-separated categories (whitespace auto-stripped) +- `--language zh-CN` — language filter +- `--pageno 1` — page number +- `--time-range {day,month,year,none}` — time filter (default: `year`; `none` disables time filtering) +- `--safesearch {0,1,2}` — safe search (default: `0` = off) +- `--max-results N` — limit number of results (applied AFTER dedup+sort, so the highest-scoring/newest items are kept) +- `--sort-by {score,date,engine,none}` — sort results (default: `score` descending; `none` preserves instance order). Applied after dedup, before `--max-results`. HTML-fallback results have no score and keep their order +- `--no-dedup` — disable cross-engine deduplication (by default, duplicate URLs — same page ignoring tracking params/fragment — are collapsed, keeping the first occurrence's engine/score) +- `--config FILE` — path to a `searxng.toml` config file; overrides the default auto-discovery (`./searxng.toml` → `~/.config/searxng-cli/searxng.toml`). Must be the first flag so its values can set defaults for other flags +- `--verbose` / `-v` — show debug-level diagnostics on stderr (HTTP request URLs, response codes, cache keys, retry detail) +- `--quiet` — suppress progress messages and retry notices on stderr; only warnings and errors are shown (no short flag: `-q` is `--query`) +- `--include-domain a.com,b.org` — allowlist; only results from these domains are kept (applied after search) +- `--exclude-domain pinterest.com` — blocklist; results from these domains are dropped (applied after search; wins over include on conflict) +- `--serial` — disable parallel multi-instance probing; search instances strictly one at a time +- `--verify` — health-check mode: verify instances and exit (no search); combine with `--format brief` for a table or `--format json` for machine-readable output +- `--format json` — full JSON (default); includes `fetched` array when `--fetch` is used; errors are emitted as JSON to stdout +- `--format brief` — title + URL + full snippet (no truncation by default; use `--snippet-len 200` to cap) +- `--format urls` — only result URLs +- `--format csv` — CSV export (title,url,engine,score,published_date,content); in batch mode (`--queries-file`), all queries merge into one CSV with a `query` column +- `--fetch N` — after search, auto-fetch full text of top N result pages (concurrent, stdlib only) +- `--fetch-timeout 10` — timeout per page fetch (default: 10s) +- `--fetch-retries 3` — max retries per page fetch +- `--max-size BYTES` — cap page size (default: unlimited; e.g. `5242880` for 5MB) +- `--retry 5` — max retries per instance (default: 3) +- `--timeout 15` — request timeout in seconds +- `--fail-fast` — use only the first instance, don't fail over to the rest +- `--proxy URL` — HTTP/HTTPS proxy for both search and fetch (e.g. `http://corp-proxy:8080`) +- `--cache-ttl MINUTES` — cache results for N minutes (default: `0` = disabled); identical queries within the TTL skip the network +- `--clear-cache` — delete all cached entries and exit (no search) +- `--cache-stats` — print cache statistics (entries, age, size, path) and exit +- `--auth-bearer TOKEN` — `Authorization: Bearer` header for private instances +- `--auth-bearer-file FILE` — read Bearer token from a file (first non-empty, non-`#` line); also honors `SEARXNG_BEARER_TOKEN` env var +- `--auth-basic USER:PASS` — `Authorization: Basic` header (auto base64-encoded) +- `--auth-basic-file FILE` — read `user:pass` from a file (first non-empty, non-`#` line); also honors `SEARXNG_BASIC_AUTH` env var +- `--version` — print version and exit + +**Completion criterion:** Outputs valid JSON with `results` array. Non-zero exit on total failure (all instances exhausted). + +### 2. `fetch.py` — Fetch & Extract Web Page Content + +``` +usage: fetch.py [-h] --url URL [--extract {text,html,markdown}] + [--timeout SEC] [--retries N] [--max-size BYTES] + [--user-agent STR] [--encoding CHARSET] + [--no-redirect] [--proxy URL] [--output FILE] + [--auth-bearer TOKEN] [--auth-bearer-file FILE] + [--auth-basic USER:PASS] [--auth-basic-file FILE] + [--verbose] [--quiet] [--version] +``` + +**What it does:** +1. Downloads a web page via HTTP GET with retry + exponential backoff +2. **Stable fetching:** retries on 429/5xx/connection errors (3x default), falls back to browser User-Agent if blocked +3. **No size limit by default** — full page content returned; use `--max-size` for a cap +4. Detects charset from HTTP headers, HTML meta tags, or UTF-8 fallback +5. Extracts readable content using tree-based parsers (stdlib or BeautifulSoup) +6. Outputs clean text, raw HTML, or properly-converted Markdown (with correct nested-link handling) + +**Key options:** +- `--url https://...` — required +- `--extract text` — clean readable text (default) +- `--extract html` — raw HTML +- `--extract markdown` — Markdown conversion (tree-based; handles nested tags, GFM tables, fenced code blocks, blockquotes, inline code, ordered/unordered/nested lists, definition lists, images, emphasis) +- `--encoding gbk` — force charset for non-UTF-8 pages (else auto-detected from HTTP header / HTML meta / UTF-8 fallback) +- `--timeout 15` — request timeout in seconds +- `--retries 3` — max retries on transient errors (429/5xx/connection); falls back to a browser User-Agent when blocked +- `--max-size BYTES` — cap page size (default: unlimited; e.g. `5242880` for 5MB) +- `--user-agent STR` — custom User-Agent header +- `--no-redirect` — do **not** follow HTTP 3xx redirects (implemented for both the requests and stdlib paths) +- `--proxy URL` — HTTP/HTTPS proxy (e.g. `http://corp-proxy:8080`); respects existing `HTTP_PROXY`/`HTTPS_PROXY` env vars when omitted +- `--output FILE` — save to file instead of stdout +- `--auth-bearer TOKEN` / `--auth-bearer-file FILE` — `Authorization: Bearer` header; file variant reads first non-empty, non-`#` line; also honors `SEARXNG_BEARER_TOKEN` env var +- `--auth-basic USER:PASS` / `--auth-basic-file FILE` — `Authorization: Basic` header (auto base64-encoded); file variant + `SEARXNG_BASIC_AUTH` env var also supported +- `--verbose` / `-v` — show debug-level diagnostics on stderr +- `--quiet` — suppress progress messages on stderr; only warnings and errors are shown +- `--version` — print version and exit + +**Extraction strategy (text mode):** +1. Strip non-content elements (script, style, nav, footer, header) +2. Extract `
`, `
`, or `` content +3. Collapse whitespace, output clean UTF-8 +4. Warn if extracted text < 500 chars (likely JS-heavy or bot-blocked) + +**Completion criterion:** Outputs page content. Non-zero exit on HTTP failure. Stderr carries warnings for low-confidence extraction. + +## Common Workflow + +```bash +INSTANCE=https://my-searxng.example.com + +# 1. Search + auto-fetch top 3 result pages (uses defaults: safesearch off, +# time_range year, json output, the 7 default engines) +python scripts/search.py -q "python asyncio tutorial" -i "$INSTANCE" --fetch 3 + +# 2. Brief format with full snippets (no truncation by default) +python scripts/search.py -q "rust memory safety" -i "$INSTANCE" --format brief --fetch 2 + +# 3. Pipe-friendly: search -> extract first URL +TOP_URL=$(python scripts/search.py -q "rust book" -i "$INSTANCE" --format urls --fail-fast | head -1) + +# 4. Override engines and use POST for a long query +python scripts/search.py -q "ai news" -i "$INSTANCE" --engines google,duckduckgo \ + --method POST --fetch 5 --fetch-timeout 15 --format json + +# 5. Failover across two instances +python scripts/search.py -q "quantum computing" -i "$INSTANCE,https://backup.example.com" +``` + +**JSON vs HTML fallback difference:** When an instance supports `format=json`, search results include rich metadata (engine name, score, category) and `fetched_source` is `"json"`. With HTML fallback, results are less structured and `fetched_source` is `"html"`. The auto-fetch feature works identically in both cases. + +## Troubleshooting + +### `--instance` is required (unless configured) +Public-instance discovery has been removed. You must supply an instance, but you have three options (in priority order): +1. `-i https://your-instance` (CLI flag, comma-separated for failover) +2. `SEARXNG_INSTANCE` environment variable (e.g. `export SEARXNG_INSTANCE="https://a,https://b"`) +3. A config file: `./searxng.toml` or `~/.config/searxng-cli/searxng.toml` (`instance = "..."` or `instances = [...]`), or `./instances.txt` / `~/.config/searxng-cli/instances.txt` (one URL per line, `#` comments) + +Self-host SearXNG (Docker: `docker run -d -p 8080:8080 searxng/searxng`) or use an instance you trust. + +### All instances return 429 / timeout +Your instance is rate-limited. Wait a few minutes, pass multiple instances via `-i a,b`, or tune your instance's limiter settings. + +### Validate your instance list with `--verify` +Before relying on a multi-instance setup, run `python scripts/search.py --verify -i a,b,c` (optionally with `--format brief` for a human-readable table). It reports each instance's reachability, JSON-API support, and latency, then exits without searching. Combine with `SEARXNG_INSTANCE` or a config file to check your persisted list in one go. + +### JSON format blocked (HTML fallback) +Some instances disable `format=json`. Enable `search.formats: [html, json]` in your instance's `settings.yml`. The HTML fallback parser handles the rest automatically. + +### Certificate errors on Windows +Some instances use Let's Encrypt certificates. Run `pip install certifi` or upgrade Python's certifi bundle. + +### fetch.py output is empty or gibberish +Try `--encoding gbk` for Chinese sites, `--encoding shift_jis` for Japanese. If the page requires JavaScript, use a headless browser instead. If a page 3xx-redirects and you passed `--no-redirect`, an empty body is expected (that's the redirect response itself). + +### Caching, proxy, domain filter, batch +- **Cache:** `--cache-ttl 30` caches results for 30 min; repeat the same query and you'll see `[cache hit]` on stderr. `--cache-stats` shows entries/age/size/path; `--clear-cache` wipes them. The cache key covers query + engines + categories + language + time_range + safesearch + pageno + method — different params get separate entries. Cache lives at `$SEARXNG_CACHE_DIR` or `~/.cache/searxng-cli/cache.db` (SQLite, WAL mode). +- **Proxy:** `--proxy http://corp-proxy:8080` sets `HTTP_PROXY`/`HTTPS_PROXY` for both search and fetch; `NO_PROXY` defaults to `localhost,127.0.0.1,::1` so local traffic stays direct. Existing proxy env vars are honored when `--proxy` is omitted. +- **Domain filter:** `--include-domain` (allowlist) and `--exclude-domain` (blocklist) run after search. Matching is case-insensitive and ignores a leading `www.`. If a domain appears in both lists, exclude wins (the result is dropped). +- **Batch:** `--queries-file FILE` runs one query per line (blank/`#` skipped) in sequence. JSON output is an array `[{"query":..., "results":...}, ...]`; a failed query becomes `{"query":..., "error":...}` but does not abort the batch. Exit 0 if any query succeeded, 1 only if all failed. +- **Config defaults:** `searxng.toml` can pre-set most flags (see the Quick Start example). Explicit CLI flags always override config values; string-valued ints from TOML are normalized automatically. + +### "ModuleNotFoundError: No module named '_config'" +Each script self-injects its directory into `sys.path`, so this should no longer occur. If it does, ensure `_config.py` sits next to `search.py` / `fetch.py` in the same `scripts/` directory. + +## Common Pitfalls + +1. **JSON format blocked.** Many instances disable `format=json`. Enable it in your instance settings for rich metadata; otherwise the HTML fallback parser is used. + +2. **Rate limiting.** `search.py` retries each instance up to 3 times with exponential backoff and jitter, then fails over to the next instance. Supply multiple instances via `-i a,b` for resilience. + +3. **fetch.py extraction quality varies.** The heuristic text extractor works well on articles and documentation but poorly on SPAs, login walls, and JavaScript-heavy pages. For JS-heavy pages, use a headless browser. + +4. **Encoding issues on Windows.** The scripts output UTF-8. On cmd.exe, run `chcp 65001` first. PowerShell handles UTF-8 natively. For non-UTF-8 pages, use `fetch.py --encoding gbk`. + +5. **POST vs GET.** Some instances handle POST differently or block it entirely. If POST search fails, try the default GET method. + +6. **Overriding defaults.** Remember the new defaults (safesearch off, time_range year, the 7-engine list). Pass the corresponding flag to change any of them per query. + +7. **Cache vs. freshness.** `--cache-ttl` returns cached results without hitting the network — fast, but stale. For time-sensitive queries (news, prices), use `--cache-ttl 0` or `--clear-cache`; the cache key includes `time_range`/`pageno` but not wall-clock time, so a `day`-range query cached at 09:00 is served as-is until the TTL expires. + +8. **Domain filter semantics.** `--include-domain`/`--exclude-domain` run *after* search and only prune the already-returned results — they do not make the instance fetch more. An overly strict allowlist can yield zero results. Exclude wins over include when a domain is in both lists. + +9. **Dedup changes output.** Dedup is on by default — duplicate URLs (ignoring `utm_*`/`gclid`/fragment, normalizing scheme/host case and param order) are collapsed. Use `--no-dedup` if you need the raw per-engine result set (e.g., comparing engine coverage). + +10. **Sort changes `--max-results` behavior.** With the default `--sort-by score`, `--max-results N` keeps the *highest-scoring* N results, not the first N in instance order. Use `--sort-by none` to preserve the original order before limiting. + +## Verification Checklist + +- [ ] `python scripts/search.py -q "test" -i --format json` returns results +- [ ] `python scripts/search.py -q "test" -i --method POST` works (optional) +- [ ] `python scripts/search.py -q "test" -i a,b` fails over across instances +- [ ] `python scripts/fetch.py -u "https://example.com"` returns readable text +- [ ] `python scripts/fetch.py -u "https://example.com" -e markdown` produces valid markdown +- [ ] `python scripts/fetch.py -u "http://" --no-redirect` does not follow the redirect +- [ ] All scripts have `--help` and `--version` +- [ ] Scripts run from any working directory (sys.path self-injection) +- [ ] Scripts exit 0 on success, non-zero on failure +- [ ] `python scripts/search.py --verify -i ` reports instance health (reachable / JSON support / latency), exit 0 +- [ ] `python scripts/search.py -q "test" -i ` shows `Dedup:` on stderr when duplicates exist; `--no-dedup` suppresses it +- [ ] `python scripts/search.py -q "test" -i --sort-by date` returns newest-first; `--sort-by none` preserves instance order +- [ ] `python scripts/search.py -q "test" -i --format csv` outputs a CSV with header `title,url,engine,score,published_date,content` +- [ ] `python scripts/search.py --config ./my-config.toml -q "test"` loads the specified config and applies its defaults +- [ ] `python scripts/search.py -q "test" -i --verbose` shows debug-level diagnostics on stderr +- [ ] `python scripts/search.py -q "test" -i --quiet` suppresses progress messages; only warnings/errors on stderr +- [ ] `python scripts/fetch.py -u --verbose` shows debug-level diagnostics on stderr +- [ ] `python scripts/search.py -q "test" -i --cache-ttl 30` then re-run → `[cache hit]` on stderr +- [ ] `python scripts/search.py --cache-stats` prints entries/size/path; `--clear-cache` reports count deleted +- [ ] `python scripts/search.py --queries-file queries.txt -i --format json` emits a JSON array +- [ ] `python scripts/search.py -q "test" -i --include-domain example.com --exclude-domain spam.com` filters as expected +- [ ] `python scripts/search.py -q "test" -i --proxy http://proxy:8080` routes through the proxy +- [ ] `python scripts/search.py -q "test" -i --auth-bearer-file ` authenticates without CLI token leakage +- [ ] A failing search with `--format json` emits `{"error":..., "exit_code":1}` to stdout +- [ ] `python scripts/fetch.py -u --proxy http://proxy:8080` routes through the proxy +- [ ] `python scripts/fetch.py -u --auth-bearer-file ` authenticates +- [ ] Stderr carries warnings; stdout carries data diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..b8f492d --- /dev/null +++ b/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +testpaths = tests +python_files = test_*.py +python_functions = test_* +addopts = -q diff --git a/scripts/_config.py b/scripts/_config.py new file mode 100644 index 0000000..5561cfb --- /dev/null +++ b/scripts/_config.py @@ -0,0 +1,11 @@ +"""Package-level constants for searxng-cli scripts. + +Import in sibling scripts with: + from _config import VERSION, USER_AGENT + +Retry settings and shared HTTP utilities now live in ``common.py`` so that +both ``search.py`` and ``fetch.py`` share one consistent implementation. +""" + +VERSION = "1.6.0" +USER_AGENT = f"searxng-cli/{VERSION}" diff --git a/scripts/cache.py b/scripts/cache.py new file mode 100644 index 0000000..6902c33 --- /dev/null +++ b/scripts/cache.py @@ -0,0 +1,162 @@ +"""SQLite-backed result cache for searxng-cli. + +Avoids re-hitting the SearXNG instance for identical queries within a +configurable TTL. Cache key is a SHA-256 of the normalized search params +(query + engines + categories + language + time_range + safesearch + +pageno + method), so different parameter combinations get separate entries. + +Storage location (in priority order): + 1. ``$SEARXNG_CACHE_DIR`` env var (directory; ``cache.db`` is created inside) + 2. ``~/.cache/searxng-cli/cache.db`` (XDG-style; on Windows this resolves + to ``C:\\Users\\\\.cache\\searxng-cli\\cache.db``) + +Uses WAL journal mode for better read concurrency. Entries expire lazily +on read; :func:`clear` removes all rows. Schema is created on first use. + +Design notes: + * Only the search result dict is cached — fetched page content is NOT, + because it is large and changes independently of the search result set. + * The cache key excludes auth headers and timeouts (transient concerns) + so two callers with the same query + params share an entry. + * All operations swallow ``sqlite3.Error`` and degrade gracefully — a + cache failure must never break a search. +""" + +import hashlib +import json +import os +import sqlite3 +import time +from pathlib import Path + +DEFAULT_CACHE_DIR = Path.home() / ".cache" / "searxng-cli" + + +def _cache_path() -> Path: + """Resolve the cache database path from env var or default location.""" + env = os.environ.get("SEARXNG_CACHE_DIR") + if env: + return Path(env) / "cache.db" + return DEFAULT_CACHE_DIR / "cache.db" + + +def _connect(path: Path) -> sqlite3.Connection: + """Open a connection with WAL mode and ensure the schema exists.""" + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(path), timeout=10) + # WAL allows concurrent readers alongside a single writer, which matters + # when --fetch spawns parallel page fetches that might also touch the cache. + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS search_cache ( + key TEXT PRIMARY KEY, + created_at REAL NOT NULL, + ttl_seconds INTEGER NOT NULL, + payload TEXT NOT NULL + ) + """ + ) + conn.commit() + return conn + + +def _make_key(params: dict) -> str: + """Build a stable cache key from search params. + + Only params that affect the result set are included; transient fields + (auth, timeout, format) are excluded so the same logical query hits the + same cache entry regardless of output formatting. + """ + # Whitelist the params that actually change what SearXNG returns. + # fmt: off + relevant = ( + "q", "categories", "language", "pageno", + "time_range", "safesearch", "engines", "method", + ) + # fmt: on + normalized = {k: params[k] for k in relevant if params.get(k)} + raw = json.dumps(normalized, sort_keys=True, ensure_ascii=False) + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +def get(params: dict, ttl_seconds: int): + """Return cached result if within TTL, else None. + + ``ttl_seconds`` is the caller's current TTL setting. If the stored + entry was written with a longer TTL, the caller's shorter TTL wins + (so reducing --cache-ttl takes effect immediately without a clear). + """ + if ttl_seconds <= 0: + return None + key = _make_key(params) + try: + with _connect(_cache_path()) as conn: + row = conn.execute( + "SELECT payload, created_at, ttl_seconds FROM search_cache " + "WHERE key = ?", + (key,), + ).fetchone() + if row is None: + return None + payload, created_at, stored_ttl = row + effective_ttl = min(ttl_seconds, stored_ttl) + if time.time() - created_at > effective_ttl: + return None + return json.loads(payload) + except sqlite3.Error: + return None + except (ValueError, json.JSONDecodeError): + # Corrupt payload — treat as miss + return None + + +def put(params: dict, result: dict, ttl_seconds: int) -> None: + """Store a result with the given TTL. Silently no-ops on TTL<=0 or error.""" + if ttl_seconds <= 0: + return + key = _make_key(params) + payload = json.dumps(result, ensure_ascii=False) + try: + with _connect(_cache_path()) as conn: + conn.execute( + "INSERT OR REPLACE INTO search_cache " + "(key, created_at, ttl_seconds, payload) VALUES (?, ?, ?, ?)", + (key, time.time(), ttl_seconds, payload), + ) + conn.commit() + except sqlite3.Error: + pass + + +def clear() -> int: + """Remove all cache entries. Returns count deleted, or 0 on error.""" + try: + with _connect(_cache_path()) as conn: + cur = conn.execute("DELETE FROM search_cache") + conn.commit() + return cur.rowcount + except sqlite3.Error: + return 0 + + +def stats() -> dict: + """Return cache statistics (entry count, age range, path).""" + path = _cache_path() + try: + with _connect(path) as conn: + row = conn.execute( + "SELECT COUNT(*), MIN(created_at), MAX(created_at) " + "FROM search_cache" + ).fetchone() + count, oldest, newest = row + return { + "entries": count or 0, + "oldest_created_at": oldest, + "newest_created_at": newest, + "path": str(path), + "size_bytes": path.stat().st_size if path.exists() else 0, + } + except sqlite3.Error as e: + return {"entries": 0, "error": str(e), "path": str(path), "size_bytes": 0} diff --git a/scripts/common.py b/scripts/common.py new file mode 100644 index 0000000..2ee4906 --- /dev/null +++ b/scripts/common.py @@ -0,0 +1,261 @@ +"""Shared utilities for searxng-cli scripts. + +This module centralizes code that was previously duplicated across +``search.py`` and ``fetch.py``: + +* ``build_auth_headers`` — construct an Authorization header from CLI flags +* ``resolve_auth_basic`` — resolve basic-auth credentials from file/env/CLI + (avoids leaving passwords in shell history) +* ``detect_charset`` — guess a response's text encoding +* ``is_retryable_error`` — unified transient-error policy (urllib + requests) +* ``FALLBACK_UAS`` — browser-like User-Agents used when blocked +* retry constants — ``RETRYABLE_STATUS``, ``RETRY_BACKOFF_BASE``, etc. +* ``setup_logging`` — shared logging configuration (--verbose/--quiet) + +Centralizing the retry policy guarantees that both scripts treat 429/5xx +as retryable and connection errors as transient, eliminating the previous +inconsistency where ``search.py`` ignored 5xx. +""" + +import logging +import sys +import urllib.error + +# Root logger for the searxng-cli package. All modules create child loggers +# via ``logging.getLogger("searxng.")`` so a single setup_logging() +# call controls them all. +_LOG = logging.getLogger("searxng") + + +def setup_logging(verbose: bool = False, quiet: bool = False) -> None: + """Configure the ``searxng`` logger hierarchy. + + * Default (no flags): ``INFO`` — progress messages, warnings, retry notices. + Matches the previous ``print(..., file=sys.stderr)`` behavior so existing + scripts and agents see no change. + * ``--verbose`` / ``-v``: ``DEBUG`` — also shows HTTP request URLs, response + status codes, cache keys, and other diagnostic detail. + * ``--quiet`` / ``-q``: ``WARNING`` — suppresses progress and retry noise; + only warnings and errors reach stderr. + + All log output goes to stderr; stdout is reserved for data (JSON/CSV/etc.). + """ + if verbose: + level = logging.DEBUG + elif quiet: + level = logging.WARNING + else: + level = logging.INFO + + _LOG.setLevel(level) + # Avoid duplicate handlers if setup_logging() is called twice (e.g. tests). + if not _LOG.handlers: + handler = logging.StreamHandler(sys.stderr) + handler.setFormatter(logging.Formatter("%(message)s")) + _LOG.addHandler(handler) + # Don't let root logger add its own handler — we own the searxng namespace. + _LOG.propagate = False + +# Retry settings (shared by both scripts) +MAX_RETRIES = 3 +RETRY_BACKOFF_BASE = 1.5 # seconds; exponential backoff + jitter + +# HTTP status codes that are worth retrying (rate limit + gateway errors) +RETRYABLE_STATUS = frozenset({429, 502, 503, 504}) + +# Browser-like UA strings for fallback when the searxng-cli UA is blocked +FALLBACK_UAS = [ + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", +] + + +def build_auth_headers(bearer_token: str = None, basic_auth: str = None) -> dict: + """Build an Authorization header dict from CLI auth flags. + + ``bearer_token``: raw Bearer token string. + ``basic_auth``: ``"username:password"`` string (base64-encoded). + + If both are provided, Bearer takes precedence (more common for APIs). + Returns a dict to merge into request headers, or an empty dict. + """ + import base64 + + headers = {} + if bearer_token: + headers["Authorization"] = f"Bearer {bearer_token}" + elif basic_auth: + encoded = base64.b64encode(basic_auth.encode("utf-8")).decode("ascii") + headers["Authorization"] = f"Basic {encoded}" + return headers + + +def _warn_file_perms(path: str) -> None: + """Warn if a credentials file is readable by group/other (POSIX only). + + On Windows the Unix permission bits in ``st_mode`` do not reflect the + actual ACL, so the check is skipped to avoid false alarms. + """ + import os + if os.name != "posix": + return + log = logging.getLogger("searxng.common") + try: + mode = os.stat(path).st_mode & 0o777 + if mode & 0o077: + log.warning( + f"Warning: credentials file '{path}' has permissions {oct(mode)} " + f"(accessible by group/other); recommend 'chmod 600' for security." + ) + except OSError: + pass + + +def resolve_auth_basic(cli_value: str = None, file_path: str = None, + env_var: str = "SEARXNG_BASIC_AUTH") -> str: + """Resolve basic-auth credentials without leaking them via shell history. + + Priority (highest wins): + 1. ``cli_value`` — explicit ``--auth-basic "user:pass"`` (convenient + but leaks into shell history; discouraged) + 2. ``file_path`` — ``--auth-basic-file FILE``; first non-empty line + is read as ``user:pass``. Recommended for shells. + 3. ``env_var`` — ``SEARXNG_BASIC_AUTH`` environment variable. + + Returns ``"user:pass"`` or ``None`` if no source provides credentials. + Raises ``RuntimeError`` if a file is specified but cannot be read. + """ + if cli_value: + return cli_value + + if file_path: + try: + from pathlib import Path + text = Path(file_path).read_text(encoding="utf-8") + _warn_file_perms(file_path) + for line in text.splitlines(): + line = line.strip() + if line and not line.startswith("#"): + return line + raise RuntimeError(f"auth file '{file_path}' contains no credentials") + except OSError as e: + raise RuntimeError(f"cannot read auth file '{file_path}': {e}") from e + + import os + return os.environ.get(env_var) + + +def resolve_auth_bearer(cli_value: str = None, file_path: str = None, + env_var: str = "SEARXNG_BEARER_TOKEN") -> str: + """Resolve a Bearer token from CLI flag, file, or environment variable. + + Mirrors :func:`resolve_auth_basic` for token-style auth. Useful for + long-lived API tokens that should not appear in shell history. + """ + if cli_value: + return cli_value + + if file_path: + try: + from pathlib import Path + text = Path(file_path).read_text(encoding="utf-8") + _warn_file_perms(file_path) + for line in text.splitlines(): + line = line.strip() + if line and not line.startswith("#"): + return line + raise RuntimeError(f"token file '{file_path}' contains no token") + except OSError as e: + raise RuntimeError(f"cannot read token file '{file_path}': {e}") from e + + import os + return os.environ.get(env_var) + + +def apply_proxy(proxy_url: str) -> None: + """Configure proxy via environment variables. + + Sets ``HTTP_PROXY`` and ``HTTPS_PROXY`` so both urllib (which reads + them via :func:`urllib.request.getproxies`) and ``requests`` (which + honors them when ``trust_env=True``, the default) pick up the proxy + without any changes to call sites. + + ``NO_PROXY`` is set to ``localhost,127.0.0.1,::1`` (if not already set) + so local traffic stays direct — matters for self-hosted SearXNG on + localhost behind a corporate proxy. + + Pass an empty string to clear the proxy env vars (rarely needed; the + default unset state already means "no proxy"). + """ + import os + if not proxy_url: + return + os.environ["HTTP_PROXY"] = proxy_url + os.environ["HTTPS_PROXY"] = proxy_url + # Keep local traffic direct unless the user has explicitly set NO_PROXY + os.environ.setdefault("NO_PROXY", "localhost,127.0.0.1,::1") + + +def detect_charset(raw: bytes, content_type: str) -> str: + """Detect charset from the Content-Type header, then an HTML meta tag. + + Falls back to UTF-8 (with replacement) if nothing reliable is found. + """ + import re + + # 1. HTTP header + if "charset=" in content_type: + charset = content_type.split("charset=")[-1].split(";")[0].strip() + try: + raw.decode(charset) + return charset + except (UnicodeDecodeError, LookupError): + pass + + # 2. HTML or + try: + head = raw[:4096].decode("ascii", errors="replace") + m = re.search(r']+charset=["\']?([a-zA-Z0-9_-]+)', head, re.IGNORECASE) + if m: + charset = m.group(1).strip() + try: + raw.decode(charset) + return charset + except (UnicodeDecodeError, LookupError): + pass + except Exception: + pass + + # 3. Fallback: UTF-8 with replacement + return "utf-8" + + +def is_retryable_error(exc: BaseException) -> bool: + """Return True if ``exc`` is a transient error worth retrying. + + Handles both the stdlib ``urllib`` errors and ``requests`` errors via + duck-typing (so this module does not need to import ``requests``): + + * ``urllib.error.HTTPError`` → retry iff status in ``RETRYABLE_STATUS`` + * ``urllib.error.URLError`` / ``OSError`` / ``TimeoutError`` → retry + (connection refused, DNS failure, timeout — all transient) + * ``requests.exceptions.HTTPError`` → retry iff ``response.status_code`` + is in ``RETRYABLE_STATUS`` + * ``requests`` connection/timeout errors (no ``.response``) → retry + """ + if isinstance(exc, urllib.error.HTTPError): + return exc.code in RETRYABLE_STATUS + if isinstance(exc, (urllib.error.URLError, OSError, TimeoutError)): + return not isinstance(exc, urllib.error.HTTPError) + + # requests.exceptions.HTTPError / RequestException (duck-typed) + resp = getattr(exc, "response", None) + status = getattr(resp, "status_code", None) + if status is not None: + return status in RETRYABLE_STATUS + if resp is None: + # requests connection/timeout error without a response -> transient + return True + return False diff --git a/scripts/fetch.py b/scripts/fetch.py new file mode 100644 index 0000000..fc00907 --- /dev/null +++ b/scripts/fetch.py @@ -0,0 +1,738 @@ +#!/usr/bin/env python3 +"""Fetch a web page and extract readable content. + +Downloads page content via HTTP GET and extracts clean, readable text. +Strips navigation, ads, scripts, and other boilerplate using heuristic rules. + +Dependencies: Python 3.8+ stdlib. Install `requests` and `beautifulsoup4` +for improved extraction quality (optional, falls back to stdlib). +""" + +import argparse +import logging +import random +import re +import sys +import time +import urllib.error +import urllib.request +from collections import namedtuple +from html.parser import HTMLParser +from pathlib import Path + +# Allow running standalone from any working directory +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from _config import USER_AGENT, VERSION +from common import ( + FALLBACK_UAS, + RETRYABLE_STATUS, + RETRY_BACKOFF_BASE, + apply_proxy, + build_auth_headers, + detect_charset, + is_retryable_error, + resolve_auth_basic, + resolve_auth_bearer, + setup_logging, +) + +logger = logging.getLogger("searxng.fetch") + + +# ----- Auth helpers ----- +# build_auth_headers is imported from common.py + +# ----- stdlib HTML-to-text extractor ----- + +class TextExtractor(HTMLParser): + """Extract visible text from HTML, skipping non-content elements.""" + + SKIP_TAGS = {"script", "style", "nav", "footer", "header", + "noscript", "iframe", "svg", "canvas", "template"} + BLOCK_TAGS = {"p", "div", "article", "section", "li", "h1", "h2", "h3", + "h4", "h5", "h6", "blockquote", "pre", "table", "tr", + "br", "hr", "main", "aside", "form", "fieldset"} + + def __init__(self): + super().__init__() + self._skip_depth = 0 + self._lines = [] + self._current_line = [] + self._block_pending = False + + def handle_starttag(self, tag, attrs): + tag_lower = tag.lower() + if tag_lower in self.SKIP_TAGS: + self._skip_depth += 1 + elif tag_lower in self.BLOCK_TAGS: + self._flush_line() + self._block_pending = True + + def handle_endtag(self, tag): + tag_lower = tag.lower() + if tag_lower in self.SKIP_TAGS and self._skip_depth > 0: + self._skip_depth -= 1 + elif tag_lower in self.BLOCK_TAGS: + self._flush_line() + + def handle_data(self, data): + if self._skip_depth > 0: + return + text = data.strip() + if text: + self._current_line.append(text) + self._block_pending = False + + def _flush_line(self): + if self._current_line: + self._lines.append(" ".join(self._current_line)) + self._current_line = [] + if self._block_pending: + self._lines.append("") + self._block_pending = False + + def get_text(self) -> str: + self._flush_line() + text = "\n".join(self._lines) + text = re.sub(r"\n{3,}", "\n\n", text) + return text.strip() + + +# ----- Enhanced extraction with BeautifulSoup (optional) ----- + +_HAS_BS4 = False +_HAS_REQUESTS = False + +try: + import requests as _requests + _HAS_REQUESTS = True +except ImportError: + pass + +try: + from bs4 import BeautifulSoup as _BeautifulSoup + _HAS_BS4 = True +except ImportError: + pass + + +def extract_with_stdlib(html_content: str) -> str: + """Extract text using stdlib HTMLParser.""" + extractor = TextExtractor() + extractor.feed(html_content) + return extractor.get_text() + + +def extract_with_bs4(html_content: str) -> str: + """Extract text using BeautifulSoup for better quality.""" + soup = _BeautifulSoup(html_content, "html.parser") + + for tag in soup(["script", "style", "nav", "footer", "header", + "noscript", "iframe", "svg", "canvas"]): + tag.decompose() + + main = (soup.find("article") or + soup.find("main") or + soup.find(role="main") or + soup.find("div", class_=re.compile(r"content|article|post|entry")) or + soup.body) + + if main is None: + main = soup + + text = main.get_text(separator="\n", strip=True) + lines = [line.strip() for line in text.split("\n") if line.strip()] + text = "\n".join(lines) + text = re.sub(r"\n{3,}", "\n\n", text) + return text + + +def extract_text(html_content: str) -> str: + """Extract readable text from HTML, preferring bs4 if available.""" + if _HAS_BS4: + try: + return extract_with_bs4(html_content) + except Exception: + pass + return extract_with_stdlib(html_content) + + +# ----- Tree-based HTML-to-Markdown converter (robust) ----- + +class MarkdownConverter(HTMLParser): + """Convert HTML to Markdown using a tag-stack approach. + + Handles nested tags in elements correctly unlike regex. Supports: + * GFM tables (```` → ``| a | b |`` with separator row) + * fenced code blocks (``
`` → triple-backtick fences)
+      * inline code (```` → backticks)
+      * blockquotes (``
`` → ``> `` prefix per line)
+      * headings, lists, images, emphasis, links
+    """
+
+    def __init__(self):
+        super().__init__()
+        self._output = []
+        self._skip_depth = 0
+        self._list_stack = []  # list of [type, counter] for ol numbering
+        self._pending_indent = ""  # preserved leading indent for li/dd
+        self._block_buffer = ""
+        self._link_href = None
+        self._link_text = []
+        self._in_link = False
+        self._in_pre = False
+        self._pre_content = []
+        self._heading_level = 0
+        self._block_empty = True
+        # Blockquote support
+        self._in_blockquote = False
+        # Table support (GFM)
+        self._in_table = False
+        self._table_rows = []  # list of (cells_list, is_header)
+        self._current_row = None
+        self._current_cell = None
+        self._in_cell = False
+
+    def _append_inline(self, text: str):
+        """Append inline markup, routing to cell/link/block buffer.
+
+        In cell context, fragments go to ``_current_cell`` so table cell text
+        accumulates correctly. In link context, they go to ``_link_text``
+        and are collapsed at ```` time. Otherwise they append to the
+        block buffer; whitespace is normalized at ``_flush_block`` time.
+        """
+        if self._in_cell:
+            self._current_cell.append(text)
+        elif self._in_link:
+            self._link_text.append(text)
+        else:
+            self._block_buffer += text
+            self._block_empty = False
+
+    def _flush_block(self):
+        t = self._block_buffer
+        # Normalize whitespace: collapse runs of spaces/tabs/newlines
+        t = re.sub(r'\s+', ' ', t).strip()
+        if t:
+            if self._in_blockquote:
+                # Prefix each line with "> " for markdown blockquote syntax
+                t = "\n".join(
+                    ("> " + line) if line.strip() else ">"
+                    for line in t.split("\n")
+                )
+            if self._pending_indent:
+                self._output.append(self._pending_indent + t)
+            else:
+                self._output.append(t)
+        self._block_buffer = ""
+        self._block_empty = True
+        self._pending_indent = ""
+
+    def _emit_table(self):
+        """Emit a GFM table from accumulated rows.
+
+        The first row becomes the header; a ``| --- | --- |`` separator row
+        follows; remaining rows become the body. Cells are padded to the
+        header width so the table renders correctly in strict GFM parsers.
+        """
+        if not self._table_rows:
+            return
+        header_row, _ = self._table_rows[0]
+        body_rows = self._table_rows[1:]
+        if not header_row:
+            return
+        ncols = len(header_row)
+        self._output.append("| " + " | ".join(header_row) + " |")
+        self._output.append("| " + " | ".join("---" for _ in range(ncols)) + " |")
+        for row, _ in body_rows:
+            # Pad short rows; truncate long rows to header width
+            while len(row) < ncols:
+                row.append("")
+            self._output.append("| " + " | ".join(row[:ncols]) + " |")
+        self._output.append("")
+
+    def handle_starttag(self, tag, attrs):
+        tag_lower = tag.lower()
+        attrs_dict = dict(attrs)
+
+        if tag_lower in ("script", "style", "nav", "footer", "header",
+                          "noscript", "iframe", "svg", "canvas", "template"):
+            self._skip_depth += 1
+            return
+
+        if self._skip_depth > 0:
+            self._skip_depth += 1
+            return
+
+        # Table structural elements are handled up-front so cell content
+        # routing (via _in_cell) takes effect before any other tag handler.
+        if tag_lower == "table":
+            self._flush_block()
+            self._in_table = True
+            self._table_rows = []
+            return
+        if self._in_table:
+            if tag_lower == "tr":
+                self._current_row = []
+                return
+            elif tag_lower in ("th", "td"):
+                self._in_cell = True
+                self._current_cell = []
+                return
+            elif tag_lower in ("thead", "tbody", "tfoot"):
+                return  # container only — rows/cells drive the output
+            # Other tags inside cells (a/strong/em/code/br) fall through
+            # to normal handling; _append_inline routes them to _current_cell.
+
+        if tag_lower in ("p", "div", "section"):
+            self._flush_block()
+        elif tag_lower == "blockquote":
+            self._flush_block()
+            self._in_blockquote = True
+        elif tag_lower == "br":
+            self._append_inline("\n")
+        elif tag_lower in ("h1", "h2", "h3", "h4", "h5", "h6"):
+            self._flush_block()
+            self._heading_level = int(tag_lower[1])
+        elif tag_lower == "pre":
+            self._flush_block()
+            self._in_pre = True
+            self._pre_content = []
+        elif tag_lower == "code":
+            # Inline code. 
 is handled by the pre path, which
+            # captures raw text verbatim — so only emit backticks when
+            # we're NOT inside a pre block.
+            if not self._in_pre:
+                self._append_inline("`")
+        elif tag_lower in ("ul", "ol"):
+            self._list_stack.append([tag_lower, 0])
+        elif tag_lower == "li":
+            self._flush_block()
+            depth = max(0, len(self._list_stack) - 1)
+            self._pending_indent = "  " * depth
+            if self._list_stack and self._list_stack[-1][0] == "ol":
+                self._list_stack[-1][1] += 1
+                marker = f"{self._list_stack[-1][1]}. "
+            else:
+                marker = "- "
+            self._block_buffer = marker
+            self._block_empty = False
+        elif tag_lower == "dt":
+            self._flush_block()
+            self._block_buffer = "**"
+            self._block_empty = False
+        elif tag_lower == "dd":
+            self._flush_block()
+            self._pending_indent = "  "
+            self._block_buffer = ""
+            self._block_empty = True
+        elif tag_lower == "a":
+            self._in_link = True
+            self._link_href = attrs_dict.get("href", "")
+            self._link_text = []
+        elif tag_lower in ("strong", "b"):
+            self._append_inline("**")
+        elif tag_lower in ("em", "i"):
+            self._append_inline("*")
+        elif tag_lower == "img":
+            alt = attrs_dict.get("alt", "")
+            src = attrs_dict.get("src", "")
+            if alt or src:
+                self._append_inline(f"![{alt}]({src})")
+
+    def handle_endtag(self, tag):
+        tag_lower = tag.lower()
+        if self._skip_depth > 0:
+            self._skip_depth -= 1
+            return
+
+        if tag_lower == "pre":
+            self._in_pre = False
+            code = "\n".join(self._pre_content)
+            self._output.append(f"```\n{code}\n```")
+            self._pre_content = []
+            self._block_empty = True
+        elif tag_lower in ("h1", "h2", "h3", "h4", "h5", "h6"):
+            prefix = "#" * self._heading_level
+            self._output.append(f"\n{prefix} {self._block_buffer.strip()}\n")
+            self._block_buffer = ""
+            self._heading_level = 0
+            self._block_empty = True
+        elif tag_lower in ("th", "td"):
+            # Collapse cell content to a single line (newlines from 

+            # would break the GFM table row). Pipe chars are escaped to
+            # avoid prematurely terminating cells.
+            cell_text = " ".join(self._current_cell).strip()
+            cell_text = re.sub(r'\s+', ' ', cell_text)
+            cell_text = cell_text.replace("|", "\\|")
+            if self._current_row is not None:
+                self._current_row.append(cell_text)
+            self._in_cell = False
+            self._current_cell = None
+        elif tag_lower == "tr":
+            if self._current_row is not None:
+                is_header = False  # GFM doesn't distinguish; first row is header
+                self._table_rows.append((self._current_row, is_header))
+            self._current_row = None
+        elif tag_lower == "table":
+            self._emit_table()
+            self._in_table = False
+            self._table_rows = []
+            self._current_row = None
+            self._current_cell = None
+            self._in_cell = False
+        elif tag_lower == "blockquote":
+            self._flush_block()
+            self._in_blockquote = False
+            self._output.append("")
+        elif tag_lower in ("p", "div", "section"):
+            self._flush_block()
+            self._output.append("")
+        elif tag_lower in ("ul", "ol"):
+            if self._list_stack:
+                self._list_stack.pop()
+            self._output.append("")
+        elif tag_lower == "li":
+            self._flush_block()
+        elif tag_lower == "dt":
+            self._block_buffer = self._block_buffer.rstrip() + "**"
+            self._flush_block()
+        elif tag_lower == "dd":
+            self._flush_block()
+        elif tag_lower == "a":
+            if self._in_link:
+                link_text = " ".join("".join(self._link_text).split())
+                if link_text and self._link_href:
+                    rendered = f"[{link_text}]({self._link_href})"
+                elif self._link_href:
+                    rendered = f"<{self._link_href}>"
+                else:
+                    rendered = ""
+                if self._in_cell:
+                    self._current_cell.append(rendered)
+                else:
+                    self._block_buffer += rendered
+                    self._block_empty = False
+                self._in_link = False
+                self._link_href = None
+                self._link_text = []
+        elif tag_lower == "code":
+            if not self._in_pre:
+                self._append_inline("`")
+        elif tag_lower in ("strong", "b"):
+            self._append_inline("**")
+        elif tag_lower in ("em", "i"):
+            self._append_inline("*")
+
+    def handle_data(self, data):
+        if self._skip_depth > 0:
+            return
+        if self._in_pre:
+            self._pre_content.append(data)
+        elif self._in_cell:
+            self._current_cell.append(data)
+        elif self._in_link:
+            self._link_text.append(data)
+        else:
+            # Don't strip — let _flush_block normalize whitespace
+            if data.strip():  # only skip purely whitespace nodes
+                self._block_buffer += data
+                self._block_empty = False
+
+    def get_markdown(self) -> str:
+        self._flush_block()
+        # Emit a table if one was left open (malformed HTML)
+        if self._in_table and self._table_rows:
+            self._emit_table()
+        text = "\n".join(self._output)
+        text = re.sub(r"\n{3,}", "\n\n", text)
+        return text.strip()
+
+
+def html_to_markdown(html_content: str) -> str:
+    """Convert HTML to Markdown using tree-based parser."""
+    # Quick strip of scripts/styles first
+    html_content = re.sub(r']*>.*?', '', html_content,
+                          flags=re.DOTALL | re.IGNORECASE)
+    html_content = re.sub(r']*>.*?', '', html_content,
+                          flags=re.DOTALL | re.IGNORECASE)
+
+    converter = MarkdownConverter()
+    converter.feed(html_content)
+    return converter.get_markdown()
+
+
+# ----- HTTP Fetch -----
+# RETRY_BACKOFF_BASE, FALLBACK_UAS, detect_charset and is_retryable_error are
+# imported from common.py (shared with search.py for a consistent retry policy).
+
+
+class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
+    """urllib handler that returns the 3xx response instead of following it."""
+
+    def http_error_302(self, req, fp, code, msg, headers):
+        return fp
+
+    http_error_301 = http_error_303 = http_error_307 = http_error_308 = http_error_302
+
+
+FetchResult = namedtuple(
+    "FetchResult",
+    ["content", "content_type", "final_url", "truncated", "user_agent"],
+)
+
+
+def fetch_url(url: str, timeout: int = 15, user_agent: str = None,
+              encoding: str = None, auth_headers: dict = None,
+              max_retries: int = 3, max_size: int = None,
+              allow_redirects: bool = True) -> "FetchResult":
+    """Fetch a URL with retry, encoding detection, UA fallback, and optional size limit.
+
+    Returns a :class:`FetchResult` namedtuple with fields:
+      ``content`` (str), ``content_type`` (str), ``final_url`` (str),
+      ``truncated`` (bool — True if ``max_size`` cut the response short),
+      ``user_agent`` (str — the UA string that succeeded; useful for logging
+      whether a fallback UA was needed).
+
+    max_size=None means unlimited (full page). Set to e.g. 5242880 for a 5MB cap.
+    allow_redirects=False stops the client from following HTTP 3xx redirects.
+    """
+    if user_agent is None:
+        user_agent = USER_AGENT
+
+    last_error = None
+    user_agents = [user_agent] + FALLBACK_UAS
+
+    for attempt in range(max_retries + 1):
+        ua = user_agents[min(attempt, len(user_agents) - 1)]
+        headers = {"User-Agent": ua}
+        if auth_headers:
+            headers.update(auth_headers)
+
+        try:
+            if _HAS_REQUESTS:
+                resp = _requests.get(url, timeout=timeout, headers=headers,
+                                     allow_redirects=allow_redirects, stream=True)
+                resp.raise_for_status()
+
+                # Read: unlimited if max_size is None, chunked with limit otherwise
+                if max_size is None:
+                    raw = resp.content
+                    truncated = False
+                else:
+                    chunks = []
+                    total = 0
+                    for chunk in resp.iter_content(chunk_size=65536, decode_unicode=False):
+                        if chunk:
+                            chunks.append(chunk)
+                            total += len(chunk)
+                            if total > max_size:
+                                break
+                    raw = b"".join(chunks)
+                    truncated = total > max_size
+
+                if encoding:
+                    content = raw.decode(encoding)
+                else:
+                    charset = detect_charset(raw, resp.headers.get("Content-Type", ""))
+                    try:
+                        content = raw.decode(charset)
+                    except (UnicodeDecodeError, LookupError):
+                        content = raw.decode("utf-8", errors="replace")
+
+                return FetchResult(content, resp.headers.get("Content-Type", ""),
+                                   resp.url, truncated, ua)
+
+            # stdlib fallback
+            req = urllib.request.Request(url, headers=headers)
+            if allow_redirects:
+                _opener = urllib.request.urlopen(req, timeout=timeout)
+            else:
+                _opener = urllib.request.build_opener(_NoRedirectHandler).open(
+                    req, timeout=timeout)
+            with _opener as resp:
+                if max_size is None:
+                    raw = resp.read()
+                    truncated = False
+                else:
+                    chunks = []
+                    total = 0
+                    while True:
+                        chunk = resp.read(65536)
+                        if not chunk:
+                            break
+                        chunks.append(chunk)
+                        total += len(chunk)
+                        if total > max_size:
+                            break
+                    raw = b"".join(chunks)
+                    truncated = total > max_size
+                content_type = resp.headers.get("Content-Type", "")
+                final_url = resp.geturl()
+
+                if encoding:
+                    charset = encoding
+                else:
+                    charset = detect_charset(raw, content_type)
+                try:
+                    content = raw.decode(charset)
+                except (UnicodeDecodeError, LookupError):
+                    content = raw.decode("utf-8", errors="replace")
+
+                return FetchResult(content, content_type, final_url, truncated, ua)
+
+        except urllib.error.HTTPError as e:
+            last_error = e
+            if is_retryable_error(e) and attempt < max_retries:
+                delay = RETRY_BACKOFF_BASE * (2 ** attempt) + random.uniform(0, 1)
+                logger.info(f"  Fetch retry {attempt+1}/{max_retries} ({e}) in {delay:.1f}s")
+                time.sleep(delay)
+                continue
+            raise RuntimeError(f"HTTP {e.code} for {url}")
+        except (urllib.error.URLError, OSError, TimeoutError) as e:
+            last_error = e
+            if attempt < max_retries:
+                delay = RETRY_BACKOFF_BASE * (2 ** attempt) + random.uniform(0, 1)
+                logger.info(f"  Fetch retry {attempt+1}/{max_retries} ({e}) in {delay:.1f}s")
+                time.sleep(delay)
+                continue
+            raise RuntimeError(f"Request failed for {url}: {e}")
+        except Exception as e:
+            # requests backend: retry only on connection errors (no response)
+            # or transient 429/5xx; do NOT retry permanent errors like 404.
+            if _HAS_REQUESTS and isinstance(e, _requests.exceptions.RequestException):
+                last_error = e
+                status = getattr(getattr(e, "response", None), "status_code", None)
+                if (status is None or status in RETRYABLE_STATUS) and attempt < max_retries:
+                    delay = RETRY_BACKOFF_BASE * (2 ** attempt) + random.uniform(0, 1)
+                    logger.info(f"  Fetch retry {attempt+1}/{max_retries} ({e}) in {delay:.1f}s")
+                    time.sleep(delay)
+                    continue
+            raise RuntimeError(f"Request failed for {url}: {e}")
+
+    raise RuntimeError(f"All {max_retries+1} attempts failed for {url}: {last_error}")
+
+
+# ----- Main -----
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Fetch a web page and extract readable content",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog=f"""searxng-cli v{VERSION}
+
+Examples:
+  %(prog)s -u https://example.com                  extract clean text
+  %(prog)s -u https://example.com -e html          raw HTML
+  %(prog)s -u https://example.com -e markdown      markdown conversion
+  %(prog)s -u https://example.com -o page.txt      save to file
+  %(prog)s -u https://example.cn -e text --encoding gbk   force charset
+        """,
+    )
+    parser.add_argument("--url", "-u", required=True, help="URL to fetch")
+    parser.add_argument("--extract", "-e", choices=["text", "html", "markdown"],
+                        default="text", help="Extraction mode (default: text)")
+    parser.add_argument("--timeout", "-t", type=int, default=15,
+                        help="Request timeout in seconds (default: 15)")
+    parser.add_argument("--retries", type=int, default=3,
+                        help="Max retries on transient errors (default: 3)")
+    parser.add_argument("--max-size", type=int, default=None, metavar="BYTES",
+                        help="Max page size in bytes (default: unlimited). Set to 5242880 for 5MB cap.")
+    parser.add_argument("--user-agent", default=None,
+                        help="Custom User-Agent header")
+    parser.add_argument("--encoding", default=None,
+                        help="Force charset for decoding (e.g. gbk, shift_jis)")
+    parser.add_argument("--no-redirect", action="store_true",
+                        help="Do not follow HTTP redirects")
+    parser.add_argument("--proxy", default=None, metavar="URL",
+                        help="HTTP/HTTPS proxy URL (e.g. http://corp-proxy:8080). "
+                             "Respects existing HTTP_PROXY/HTTPS_PROXY env vars when omitted.")
+    parser.add_argument("--output", "-o", default=None,
+                        help="Save to file instead of stdout")
+    parser.add_argument("--auth-bearer", default=None, metavar="TOKEN",
+                        help="Authorization: Bearer  for authenticated endpoints")
+    parser.add_argument("--auth-bearer-file", default=None, metavar="FILE",
+                        help="Read Bearer token from a file (first non-empty, non-# line). "
+                             "Avoids leaving tokens in shell history.")
+    parser.add_argument("--auth-basic", default=None, metavar="USER:PASS",
+                        help="Authorization: Basic base64(user:pass) for authenticated endpoints")
+    parser.add_argument("--auth-basic-file", default=None, metavar="FILE",
+                        help="Read basic auth 'user:pass' from a file (first non-empty, non-# line). "
+                             "Avoids leaving passwords in shell history. "
+                             "Env var SEARXNG_BASIC_AUTH is also honored.")
+    parser.add_argument("--verbose", "-v", action="store_true", default=False,
+                        help="Verbose output: show debug-level diagnostics on stderr")
+    parser.add_argument("--quiet", action="store_true", default=False,
+                        help="Quiet output: suppress progress messages on stderr; "
+                             "only warnings and errors are shown")
+    parser.add_argument("--version", action="version", version=f"searxng-cli v{VERSION}")
+
+    args = parser.parse_args()
+    setup_logging(verbose=args.verbose, quiet=args.quiet)
+
+    if not args.url.startswith(("http://", "https://")):
+        logger.error("Error: URL must start with http:// or https://")
+        sys.exit(1)
+
+    # Apply proxy via env vars so both urllib and requests honor it.
+    if args.proxy:
+        apply_proxy(args.proxy)
+        logger.info(f"Proxy: {args.proxy}")
+
+    try:
+        bearer_token = resolve_auth_bearer(args.auth_bearer, args.auth_bearer_file)
+        basic_auth = resolve_auth_basic(args.auth_basic, args.auth_basic_file)
+    except RuntimeError as e:
+        logger.error(f"Error: {e}")
+        sys.exit(1)
+    auth_headers = build_auth_headers(
+        bearer_token=bearer_token,
+        basic_auth=basic_auth,
+    )
+    if auth_headers:
+        auth_type = "Bearer" if bearer_token else "Basic"
+        logger.info(f"Auth: {auth_type} ***")
+
+    try:
+        result = fetch_url(
+            args.url, timeout=args.timeout, user_agent=args.user_agent,
+            encoding=args.encoding, auth_headers=auth_headers,
+            max_retries=args.retries, max_size=args.max_size,
+            allow_redirects=not args.no_redirect,
+        )
+        content, content_type, final_url = (
+            result.content, result.content_type, result.final_url,
+        )
+    except Exception as e:
+        logger.error(f"Error: {e}")
+        sys.exit(1)
+
+    if final_url != args.url:
+        logger.info(f"Redirected to: {final_url}")
+
+    is_html = ("html" in content_type.lower() or
+               content.strip().startswith(" dict:
+    """Merge multiple header dicts, with later dicts overriding earlier ones."""
+    result = {}
+    for d in header_dicts:
+        if d:
+            result.update(d)
+    return result
+
+
+# ----- HTML fallback: result extractor -----
+
+class SearXNGHTMLParser(HTMLParser):
+    """Extract search results + metadata from SearXNG's simple theme HTML.
+
+    Matches:
+      
+      
+      
...

+
+    The ``published_date`` field is populated from the ``
AB
12
" + md = html_to_markdown(html) + assert "| A | B |" in md + assert "| --- | --- |" in md + assert "| 1 | 2 |" in md + + +def test_markdown_table_pipe_escaped(): + """Pipe chars inside cells must be escaped so they don't break the row.""" + html = "
Col
a|b
" + md = html_to_markdown(html) + assert "a\\|b" in md + + +# ----- Emphasis / images ----- + +def test_markdown_strong_emphasis(): + md = html_to_markdown("

bold

") + assert "**bold**" in md + + +def test_markdown_em_italic(): + md = html_to_markdown("

italic

") + assert "*italic*" in md + + +def test_markdown_image(): + md = html_to_markdown('pic') + assert "![pic](http://x.com/a.png)" in md + + +# ---- Stripping non-content ----- + +def test_markdown_strips_script(): + md = html_to_markdown("

visible

") + assert "alert" not in md + assert "visible" in md + + +def test_markdown_strips_style(): + md = html_to_markdown("

visible

") + assert "color" not in md + assert "visible" in md + + +# ----- Stdlib text extractor ----- + +def test_extract_with_stdlib_basic(): + html = "

Hello

" + text = extract_with_stdlib(html) + assert "Hello" in text + assert "x" not in text # script content excluded + + +def test_extract_with_stdlib_skips_nav(): + html = "
content
" + text = extract_with_stdlib(html) + assert "content" in text + assert "menu" not in text + + +# ----- Ordered / nested lists ----- + +def test_markdown_ordered_list_numbered(): + md = html_to_markdown("
  1. first
  2. second
  3. third
") + assert "1. first" in md + assert "2. second" in md + assert "3. third" in md + + +def test_markdown_nested_ordered_list(): + html = "
  1. outer1
    1. inner1
    2. inner2
  2. outer2
" + md = html_to_markdown(html) + assert "1. outer1" in md + assert " 1. inner1" in md + assert " 2. inner2" in md + assert "2. outer2" in md + + +def test_markdown_nested_unordered_list(): + html = "
  • outer
    • inner
" + md = html_to_markdown(html) + assert "- outer" in md + assert " - inner" in md + + +def test_markdown_mixed_nested_list(): + """
    containing
      — markers must match list type at each depth.""" + html = "
      • item
        1. sub1
        2. sub2
      " + md = html_to_markdown(html) + assert "- item" in md + assert " 1. sub1" in md + assert " 2. sub2" in md + + +# ----- Definition lists (
      /
      /
      ) ----- + +def test_markdown_definition_list(): + html = "
      Term
      Definition
      " + md = html_to_markdown(html) + assert "**Term**" in md + assert " Definition" in md + + +def test_markdown_definition_list_multiple(): + html = "
      T1
      D1
      T2
      D2
      " + md = html_to_markdown(html) + assert "**T1**" in md + assert "**T2**" in md + assert " D1" in md + assert " D2" in md diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..30e1bad --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,133 @@ +"""Integration tests — mock urllib to test search_multi / fetch_url end-to-end. + +Covers: search_json success/HTML-fallback/404/403, search_multi serial +failover + all-fail, fetch_url stdlib-path success. + +No real network calls are made; ``urllib.request.urlopen`` is patched. +""" +import json +import urllib.error +from unittest.mock import patch, MagicMock + +from search import search_json, search_multi +import fetch as fetch_mod +from fetch import fetch_url + + +def _mock_urlopen(data: bytes, content_type="application/json"): + """Build a MagicMock that quacks like an urlopen context manager.""" + resp = MagicMock() + resp.read.return_value = data + resp.headers = {"Content-Type": content_type} + resp.__enter__.return_value = resp + resp.__exit__.return_value = None + return resp + + +# ----- search_json ----- + +def test_search_json_success(): + payload = json.dumps({"results": [{"title": "hi", "url": "https://x.com"}]}) + with patch("urllib.request.urlopen", + return_value=_mock_urlopen(payload.encode())): + r = search_json("https://s.example.com", {"q": "test", "format": "json"}) + assert r is not None + assert r["results"][0]["title"] == "hi" + + +def test_search_json_html_returns_none(): + """HTML response (not JSON) → return None (JSON unsupported).""" + with patch("urllib.request.urlopen", + return_value=_mock_urlopen(b"not json")): + r = search_json("https://s.example.com", {"q": "test", "format": "json"}) + assert r is None + + +def test_search_json_404_returns_none(): + """404 = JSON endpoint absent → None (triggers HTML fallback upstream).""" + err = urllib.error.HTTPError("url", 404, "Not Found", {}, None) + with patch("urllib.request.urlopen", side_effect=err): + r = search_json("https://s.example.com", {"q": "test", "format": "json"}) + assert r is None + + +def test_search_json_403_raises(): + """403 = auth/IP issue → must raise, not silently fall back to HTML.""" + err = urllib.error.HTTPError("url", 403, "Forbidden", {}, None) + with patch("urllib.request.urlopen", side_effect=err): + try: + search_json("https://s.example.com", {"q": "test", "format": "json"}) + assert False, "should have raised" + except urllib.error.HTTPError: + pass # expected + + +# ----- search_multi (serial failover) ----- + +def test_search_multi_serial_failover(): + """First instance fails, second succeeds → return second's results.""" + payload = json.dumps({"results": [{"title": "from-b", "url": "https://b.com"}]}) + err = urllib.error.URLError("connection refused") + with patch("urllib.request.urlopen", + side_effect=[err, _mock_urlopen(payload.encode())]): + r = search_multi( + ["https://a.example.com", "https://b.example.com"], + {"q": "test", "format": "json"}, + parallel=False, retry_per=0, + ) + assert r["results"][0]["title"] == "from-b" + + +def test_search_multi_all_fail_raises(): + """All instances fail → RuntimeError.""" + err = urllib.error.URLError("connection refused") + with patch("urllib.request.urlopen", side_effect=err): + try: + search_multi( + ["https://a.example.com", "https://b.example.com"], + {"q": "test", "format": "json"}, + parallel=False, retry_per=0, + ) + assert False, "should have raised" + except RuntimeError as e: + assert "All 2 instances failed" in str(e) + + +def test_search_multi_single_instance_success(): + """Single instance, serial mode, success → return results.""" + payload = json.dumps({"results": [{"title": "ok", "url": "https://a.com"}]}) + with patch("urllib.request.urlopen", + return_value=_mock_urlopen(payload.encode())): + r = search_multi( + ["https://a.example.com"], + {"q": "test", "format": "json"}, + parallel=False, retry_per=0, + ) + assert r["results"][0]["title"] == "ok" + + +# ----- fetch_url (stdlib path) ----- + +def test_fetch_url_stdlib_success(): + """fetch_url with stdlib path returns content + content_type.""" + html = b"

      Hello

      " + resp = _mock_urlopen(html, content_type="text/html; charset=utf-8") + with patch("urllib.request.urlopen", return_value=resp), \ + patch.object(fetch_mod, "_HAS_REQUESTS", False): + result = fetch_url("https://example.com", max_retries=0) + assert "Hello" in result.content + assert "text/html" in result.content_type + + +def test_fetch_url_stdlib_max_size_truncates(): + """max_size sets truncated=True when response exceeds the cap. + + Note: mock's read() ignores the size arg, so content length is not + accurately capped here — we only verify the truncated flag is set. + """ + html = b"" + b"x" * 200 + b"" + resp = _mock_urlopen(html, content_type="text/html") + with patch("urllib.request.urlopen", return_value=resp), \ + patch.object(fetch_mod, "_HAS_REQUESTS", False): + result = fetch_url("https://example.com", max_retries=0, max_size=50) + assert result.truncated is True diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 0000000..4f68f02 --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,120 @@ +"""Tests for the shared logging configuration (setup_logging). + +Verifies: + * Default level is INFO (matches previous print-to-stderr behavior). + * --verbose sets DEBUG. + * --quiet sets WARNING. + * All log output goes to stderr, never stdout. + * Repeated setup calls don't stack duplicate handlers. + * Child loggers (searxng.search, searxng.fetch, searxng.common) inherit + the root searxng logger's level. +""" +import logging +import sys + +from common import setup_logging + + +_ROOT = logging.getLogger("searxng") + + +def _reset_logger(): + """Clear the searxng logger so each test starts fresh.""" + _ROOT.handlers.clear() + _ROOT.setLevel(logging.NOTSET) + + +def test_setup_logging_default_info(): + _reset_logger() + setup_logging() + assert _ROOT.level == logging.INFO + + +def test_setup_logging_verbose_debug(): + _reset_logger() + setup_logging(verbose=True) + assert _ROOT.level == logging.DEBUG + + +def test_setup_logging_quiet_warning(): + _reset_logger() + setup_logging(quiet=True) + assert _ROOT.level == logging.WARNING + + +def test_setup_logging_verbose_overrides_quiet(): + """If both --verbose and --quiet are passed, verbose wins (checked first).""" + _reset_logger() + setup_logging(verbose=True, quiet=True) + assert _ROOT.level == logging.DEBUG + + +def test_setup_logging_no_duplicate_handlers(): + _reset_logger() + setup_logging() + setup_logging() + setup_logging() + assert len(_ROOT.handlers) == 1 + + +def test_setup_logging_output_to_stderr(capsys): + """Log messages must go to stderr, never stdout.""" + _reset_logger() + setup_logging() + log = logging.getLogger("searxng.search") + log.info("test message") + captured = capsys.readouterr() + assert "test message" in captured.err + assert captured.out == "" + + +def test_child_logger_inherits_level(): + """Child loggers (searxng.search, searxng.fetch, searxng.common) must + see the level set on the root searxng logger.""" + _reset_logger() + setup_logging(verbose=True) + for name in ("searxng.search", "searxng.fetch", "searxng.common"): + child = logging.getLogger(name) + assert child.getEffectiveLevel() == logging.DEBUG + + +def test_quiet_suppresses_info(capsys): + """In quiet mode, INFO messages are NOT written to stderr.""" + _reset_logger() + setup_logging(quiet=True) + log = logging.getLogger("searxng.search") + log.info("this should be hidden") + log.warning("this should be visible") + captured = capsys.readouterr() + assert "this should be hidden" not in captured.err + assert "this should be visible" in captured.err + + +def test_verbose_shows_debug(capsys): + """In verbose mode, DEBUG messages ARE written to stderr.""" + _reset_logger() + setup_logging(verbose=True) + log = logging.getLogger("searxng.fetch") + log.debug("debug detail") + captured = capsys.readouterr() + assert "debug detail" in captured.err + + +def test_default_hides_debug(capsys): + """In default (INFO) mode, DEBUG messages are NOT written to stderr.""" + _reset_logger() + setup_logging() + log = logging.getLogger("searxng.search") + log.debug("hidden debug") + log.info("visible info") + captured = capsys.readouterr() + assert "hidden debug" not in captured.err + assert "visible info" in captured.err + + +def test_propagate_disabled(): + """The searxng logger must not propagate to the root logger (avoids + duplicate output via the root handler).""" + _reset_logger() + setup_logging() + assert _ROOT.propagate is False diff --git a/tests/test_search.py b/tests/test_search.py new file mode 100644 index 0000000..3ffc0c8 --- /dev/null +++ b/tests/test_search.py @@ -0,0 +1,496 @@ +"""Tests for scripts/search.py — pure-logic functions (no network). + +Covers: parse_instances (URL normalization, lists, whitespace), +_normalize_csv, filter_results_by_domain (include/exclude, www, case, +precedence), _read_queries_file (comments, blanks, missing file), +_cfg_int (str/int/absent/bad), _build_params (param construction, +time_range=none exclusion), and _merge_headers. +""" +import pytest +from search import ( + _build_params, + _cfg_int, + _format_results, + _merge_headers, + _normalize_csv, + _read_queries_file, + deduplicate_results, + filter_results_by_domain, + load_config, + parse_instances, + sort_results, +) + + +# ----- parse_instances ----- + +def test_parse_instances_single(): + assert parse_instances("https://example.com") == ["https://example.com"] + + +def test_parse_instances_adds_https_prefix(): + assert parse_instances("example.com") == ["https://example.com"] + + +def test_parse_instances_strips_trailing_slash(): + assert parse_instances("https://example.com/") == ["https://example.com"] + + +def test_parse_instances_multiple_comma(): + assert parse_instances("a.com,b.com") == ["https://a.com", "https://b.com"] + + +def test_parse_instances_handles_whitespace(): + assert parse_instances(" a.com , b.com ") == ["https://a.com", "https://b.com"] + + +def test_parse_instances_empty_string(): + assert parse_instances("") == [] + + +def test_parse_instances_preserves_http(): + assert parse_instances("http://localhost:8080") == ["http://localhost:8080"] + + +def test_parse_instances_skips_empty_entries(): + assert parse_instances("a.com,,b.com,") == ["https://a.com", "https://b.com"] + + +# ----- _normalize_csv ----- + +def test_normalize_csv_strips_spaces(): + assert _normalize_csv("google, bing, brave") == "google,bing,brave" + + +def test_normalize_csv_drops_empty_parts(): + assert _normalize_csv("google,,bing,") == "google,bing" + + +def test_normalize_csv_empty_input(): + assert _normalize_csv("") == "" + + +# ----- filter_results_by_domain ----- + +def _results(*urls): + return {"results": [{"url": u, "title": u} for u in urls]} + + +def test_filter_no_args_returns_unchanged(): + r = _results("https://a.com/1", "https://b.com/2") + out = filter_results_by_domain(dict(r)) + assert len(out["results"]) == 2 + + +def test_filter_include_allowlist(): + r = _results("https://a.com/1", "https://b.com/2", "https://c.com/3") + out = filter_results_by_domain(dict(r), include_domains=["a.com"]) + assert len(out["results"]) == 1 + assert out["results"][0]["url"] == "https://a.com/1" + + +def test_filter_include_www_normalized(): + """www. prefix is stripped for matching, so 'a.com' matches 'www.a.com'.""" + r = _results("https://www.a.com/1", "https://b.com/2") + out = filter_results_by_domain(dict(r), include_domains=["a.com"]) + assert len(out["results"]) == 1 + assert out["results"][0]["url"] == "https://www.a.com/1" + + +def test_filter_exclude_blocklist(): + r = _results("https://a.com/1", "https://b.com/2") + out = filter_results_by_domain(dict(r), exclude_domains=["b.com"]) + assert len(out["results"]) == 1 + assert out["results"][0]["url"] == "https://a.com/1" + + +def test_filter_exclude_overrides_include(): + """When a domain is in BOTH lists, exclude wins (result dropped). + + Rationale: include filters first (allowlist), then exclude filters the + survivors (blocklist). A domain listed in both is kept by include then + removed by exclude — exclude is the more explicit "do not want" intent. + """ + r = _results("https://a.com/1") + out = filter_results_by_domain(dict(r), include_domains=["a.com"], + exclude_domains=["a.com"]) + assert len(out["results"]) == 0 + + +def test_filter_case_insensitive(): + r = _results("https://A.COM/1") + out = filter_results_by_domain(dict(r), include_domains=["a.com"]) + assert len(out["results"]) == 1 + + +def test_filter_empty_results_list(): + out = filter_results_by_domain({"results": []}, include_domains=["a.com"]) + assert out["results"] == [] + + +def test_filter_no_results_key(): + """Missing 'results' key should not raise.""" + out = filter_results_by_domain({}, include_domains=["a.com"]) + assert out == {} + + +def test_filter_multiple_include(): + r = _results("https://a.com/1", "https://b.com/2", "https://c.com/3") + out = filter_results_by_domain(dict(r), include_domains=["a.com", "c.com"]) + assert len(out["results"]) == 2 + + +# ----- _read_queries_file ----- + +def test_read_queries_file_basic(tmp_path): + f = tmp_path / "queries.txt" + f.write_text("query one\n# comment\n\nquery two\n", encoding="utf-8") + assert _read_queries_file(str(f)) == ["query one", "query two"] + + +def test_read_queries_file_missing_raises(): + with pytest.raises(RuntimeError): + _read_queries_file("nonexistent_file.txt") + + +def test_read_queries_file_all_comments(tmp_path): + f = tmp_path / "empty.txt" + f.write_text("# only comments\n# another\n", encoding="utf-8") + assert _read_queries_file(str(f)) == [] + + +def test_read_queries_file_strips_whitespace(tmp_path): + f = tmp_path / "q.txt" + f.write_text(" spaced query \n", encoding="utf-8") + assert _read_queries_file(str(f)) == ["spaced query"] + + +# ----- _cfg_int ----- + +def test_cfg_int_present_int_value(): + assert _cfg_int({"timeout": 15}, "timeout", 30) == 15 + + +def test_cfg_int_present_str_value(): + """TOML may carry the value as a string; _cfg_int must coerce.""" + assert _cfg_int({"timeout": "15"}, "timeout", 30) == 15 + + +def test_cfg_int_absent_returns_default(): + assert _cfg_int({}, "timeout", 30) == 30 + + +def test_cfg_int_bad_value_returns_default(): + assert _cfg_int({"timeout": "abc"}, "timeout", 30) == 30 + + +def test_cfg_int_none_default(): + """Defaults may be None (e.g. --retry); absent key must return None.""" + assert _cfg_int({}, "max_retries", None) is None + + +# ----- _build_params ----- + +class _Args: + """Minimal argparse.Namespace stand-in for _build_params tests.""" + + def __init__(self, **overrides): + self.categories = None + self.language = None + self.pageno = 1 + self.time_range = "year" + self.safesearch = 0 + self.engines = "google,bing" + for k, v in overrides.items(): + setattr(self, k, v) + + +def test_build_params_minimal(): + p = _build_params("hello", _Args()) + assert p["q"] == "hello" + assert p["format"] == "json" + assert "categories" not in p + assert "language" not in p + assert p["engines"] == "google,bing" + + +def test_build_params_normalizes_categories(): + p = _build_params("x", _Args(categories="general, news")) + assert p["categories"] == "general,news" + + +def test_build_params_time_range_none_excluded(): + p = _build_params("x", _Args(time_range="none")) + assert "time_range" not in p + + +def test_build_params_pageno_as_string(): + p = _build_params("x", _Args(pageno=3)) + assert p["pageno"] == "3" + + +def test_build_params_safesearch_as_string(): + p = _build_params("x", _Args(safesearch=1)) + assert p["safesearch"] == "1" + + +# ----- _merge_headers ----- + +def test_merge_headers_basic(): + assert _merge_headers({"a": 1}, {"b": 2}) == {"a": 1, "b": 2} + + +def test_merge_headers_later_overrides(): + assert _merge_headers({"a": 1}, {"a": 2}) == {"a": 2} + + +def test_merge_headers_skips_none_dicts(): + assert _merge_headers(None, {"a": 1}, None) == {"a": 1} + + +def test_merge_headers_all_none(): + assert _merge_headers(None, None) == {} + + +# ----- deduplicate_results ----- + +def test_dedup_removes_exact_duplicate_url(): + r = {"results": [ + {"url": "https://a.com/1", "engine": "google", "score": 1.0}, + {"url": "https://a.com/1", "engine": "bing", "score": 0.5}, + ]} + out = deduplicate_results(r) + assert len(out["results"]) == 1 + assert out["results"][0]["engine"] == "google" # first kept + + +def test_dedup_strips_tracking_params(): + """utm_*, gclid, fbclid, etc. are stripped before comparison.""" + r = {"results": [ + {"url": "https://a.com/page?utm_source=x&id=1"}, + {"url": "https://a.com/page?id=1&utm_medium=y"}, + ]} + out = deduplicate_results(r) + assert len(out["results"]) == 1 + + +def test_dedup_strips_fragment(): + r = {"results": [ + {"url": "https://a.com/page#section1"}, + {"url": "https://a.com/page#section2"}, + ]} + out = deduplicate_results(r) + assert len(out["results"]) == 1 + + +def test_dedup_normalizes_scheme_host_case(): + r = {"results": [ + {"url": "HTTPS://Example.COM/path"}, + {"url": "https://example.com/path"}, + ]} + out = deduplicate_results(r) + assert len(out["results"]) == 1 + + +def test_dedup_normalizes_param_order(): + r = {"results": [ + {"url": "https://a.com/p?a=1&b=2"}, + {"url": "https://a.com/p?b=2&a=1"}, + ]} + out = deduplicate_results(r) + assert len(out["results"]) == 1 + + +def test_dedup_keeps_different_urls(): + r = {"results": [ + {"url": "https://a.com/1"}, + {"url": "https://a.com/2"}, + ]} + out = deduplicate_results(r) + assert len(out["results"]) == 2 + + +def test_dedup_keeps_url_less_results(): + """Results without a URL are never deduped (kept as-is).""" + r = {"results": [ + {"title": "no url"}, + {"title": "also no url"}, + ]} + out = deduplicate_results(r) + assert len(out["results"]) == 2 + + +def test_dedup_empty_results(): + out = deduplicate_results({"results": []}) + assert out["results"] == [] + + +def test_dedup_no_results_key(): + out = deduplicate_results({}) + assert out == {} + + +# ----- sort_results ----- + +def test_sort_by_score_descending(): + r = {"results": [ + {"url": "https://a.com", "score": 0.5}, + {"url": "https://b.com", "score": 2.0}, + {"url": "https://c.com", "score": 1.0}, + ]} + out = sort_results(r, "score") + assert [x["url"] for x in out["results"]] == [ + "https://b.com", "https://c.com", "https://a.com" + ] + + +def test_sort_by_score_none_at_end(): + """Entries without score keep relative order at the end.""" + r = {"results": [ + {"url": "https://a.com", "score": 1.0}, + {"url": "https://b.com"}, # no score + {"url": "https://c.com", "score": 3.0}, + {"url": "https://d.com"}, # no score + ]} + out = sort_results(r, "score") + assert out["results"][0]["url"] == "https://c.com" + assert out["results"][1]["url"] == "https://a.com" + # no-score entries keep relative order: b before d + assert out["results"][2]["url"] == "https://b.com" + assert out["results"][3]["url"] == "https://d.com" + + +def test_sort_by_date_descending(): + r = {"results": [ + {"url": "https://a.com", "published_date": "2024-01-01"}, + {"url": "https://b.com", "published_date": "2024-06-15"}, + {"url": "https://c.com", "published_date": "2024-03-10"}, + ]} + out = sort_results(r, "date") + assert [x["url"] for x in out["results"]] == [ + "https://b.com", "https://c.com", "https://a.com" + ] + + +def test_sort_by_date_none_at_end(): + r = {"results": [ + {"url": "https://a.com", "published_date": "2024-01-01"}, + {"url": "https://b.com"}, # no date + ]} + out = sort_results(r, "date") + assert out["results"][0]["url"] == "https://a.com" + assert out["results"][1]["url"] == "https://b.com" + + +def test_sort_by_engine_ascending(): + r = {"results": [ + {"url": "https://a.com", "engine": "duckduckgo"}, + {"url": "https://b.com", "engine": "bing"}, + {"url": "https://c.com", "engine": "google"}, + ]} + out = sort_results(r, "engine") + assert [x["engine"] for x in out["results"]] == ["bing", "duckduckgo", "google"] + + +def test_sort_by_none_preserves_order(): + r = {"results": [ + {"url": "https://a.com", "score": 0.5}, + {"url": "https://b.com", "score": 2.0}, + ]} + out = sort_results(r, "none") + assert [x["url"] for x in out["results"]] == ["https://a.com", "https://b.com"] + + +def test_sort_empty_results(): + out = sort_results({"results": []}, "score") + assert out["results"] == [] + + +def test_sort_no_results_key(): + out = sort_results({}, "score") + assert out == {} + + +# ----- _format_results (csv) ----- + +def test_format_csv_basic(): + results = {"results": [ + {"title": "T1", "url": "https://a.com", "engine": "google", "score": 1.0, + "published_date": "2024-01-01", "content": "snippet one"}, + ]} + args = _Args(format="csv", snippet_len=0, fetch=0) + out = _format_results(results, args) + assert "title,url,engine,score,published_date,content" in out + assert "T1" in out + assert "https://a.com" in out + assert "google" in out + assert "1.0" in out + assert "snippet one" in out + + +def test_format_csv_multiple_rows(): + results = {"results": [ + {"title": "A", "url": "https://a.com", "engine": "google", "score": 2.0, + "published_date": "", "content": "ca"}, + {"title": "B", "url": "https://b.com", "engine": "bing", "score": 1.0, + "published_date": "2024-06-01", "content": "cb"}, + ]} + args = _Args(format="csv", snippet_len=0, fetch=0) + out = _format_results(results, args) + lines = out.strip().split("\n") + assert len(lines) == 3 # header + 2 rows + assert lines[0].startswith("title,url") + + +def test_format_csv_empty_results(): + results = {"results": []} + args = _Args(format="csv", snippet_len=0, fetch=0) + out = _format_results(results, args) + assert "title,url,engine,score,published_date,content" in out + + +def test_format_csv_missing_fields(): + """Results with missing fields → empty string in CSV, no crash.""" + results = {"results": [ + {"title": "Only Title"}, # no url, engine, score, etc. + ]} + args = _Args(format="csv", snippet_len=0, fetch=0) + out = _format_results(results, args) + assert "Only Title" in out + + +def test_format_csv_comma_in_content_escaped(): + """Commas in content are properly quoted by the csv module.""" + results = {"results": [ + {"title": "T", "url": "https://a.com", "engine": "g", "score": 1.0, + "published_date": "", "content": "has, comma"}, + ]} + args = _Args(format="csv", snippet_len=0, fetch=0) + out = _format_results(results, args) + assert '"has, comma"' in out + + +# ----- load_config (--config FILE) ----- + +def test_load_config_explicit_path(tmp_path): + f = tmp_path / "test.toml" + f.write_text( + '[searxng]\ninstance = "https://x.example.com"\ntimeout = 20\nformat = "brief"\n', + encoding="utf-8") + cfg = load_config(str(f)) + assert cfg["instance"] == "https://x.example.com" + assert cfg["timeout"] == 20 + assert cfg["format"] == "brief" + + +def test_load_config_nonexistent_returns_empty(): + cfg = load_config("/nonexistent/path/config.toml") + assert cfg == {} + + +def test_load_config_top_level_table(tmp_path): + """Config without [searxng] section — top-level keys used directly.""" + f = tmp_path / "flat.toml" + f.write_text('instance = "https://flat.example.com"\n', encoding="utf-8") + cfg = load_config(str(f)) + assert cfg["instance"] == "https://flat.example.com"