Add search enhancements: JS fallback, deduplication, filtering, and encoding detection

- Add browserless JS fallback for pages that require JavaScript rendering
- Add URL deduplication with normalization (strip www/amp/m, tracking params)
- Add regex filter_query parameter to filter search results server-side
- Add user_location support for localized search results
- Add snippet enrichment with query-relevant window extraction
- Add auto-encoding detection via charset_normalizer for non-UTF-8 pages
- Add max_content_length override per extract request
- Expand SearXNG engines and configure additional search backends
This commit is contained in:
Максим Рябов 2026-04-10 22:39:21 +03:00
parent 92e92cd7f6
commit f686fdfe0b
13 changed files with 541 additions and 42 deletions

View file

@ -26,6 +26,13 @@ class ExtractionConfig(BaseModel):
max_content_length: int = 50000
domain_concurrency: int = 2
domain_semaphore_max_size: int = 1000
snippet_enrichment: bool = False
snippet_enrichment_count: int = 3
js_fallback: bool = False
browserless_url: str = "ws://browserless:3000"
js_timeout: int = 15
js_wait_for: str = "networkidle"
js_max_concurrent: int = 3
user_agents: list[str] = [
"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",

View file

@ -9,6 +9,8 @@ from fastapi.middleware.cors import CORSMiddleware
from slowapi import _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from charset_normalizer import from_bytes
from app.config import settings
from app.logging_setup import setup_logging
from app.middleware import RequestIDMiddleware, TimingMiddleware
@ -23,6 +25,12 @@ from app.services.search_backend import create_search_backend
setup_logging()
logger = structlog.get_logger(__name__)
def _detect_encoding(content: bytes) -> str:
"""Auto-detect encoding when Content-Type header lacks charset."""
result = from_bytes(content).best()
return result.encoding if result else "utf-8"
TOOL_SCHEMA = {
"tools": [
{
@ -82,6 +90,10 @@ TOOL_SCHEMA = {
"enum": ["day", "week", "month", "year"],
"description": "Filter by time range",
},
"filter_query": {
"type": "string",
"description": "Regex pattern to filter results — only results whose title, URL, or content match are returned",
},
},
},
},
@ -131,6 +143,7 @@ async def lifespan(app: FastAPI):
limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
timeout=httpx.Timeout(settings.extraction.timeout + 5),
proxy=proxy_url,
default_encoding=_detect_encoding,
)
# Cache

View file

@ -37,8 +37,15 @@ class ContentFormat(str, Enum):
# --- Search ---
class UserLocation(BaseModel):
country: str = Field(default="", description="Country code (e.g. 'US', 'RU', 'DE')")
city: str = Field(default="", description="City name")
region: str = Field(default="", description="Region or state")
timezone: str = Field(default="", description="IANA timezone (e.g. 'Europe/Moscow')")
class SearchRequest(BaseModel):
query: str = Field(..., description="The search query")
query: str = Field(..., min_length=1, max_length=2000, description="The search query")
search_depth: SearchDepth = Field(
default=SearchDepth.basic,
description="basic = fast snippets only, advanced = fetch and extract full content",
@ -51,6 +58,8 @@ class SearchRequest(BaseModel):
include_domains: list[str] = Field(default_factory=list, description="Only include results from these domains")
exclude_domains: list[str] = Field(default_factory=list, description="Exclude results from these domains")
time_range: Optional[TimeRange] = Field(default=None, description="Filter results by time range")
filter_query: Optional[str] = Field(default=None, max_length=500, description="Regex pattern to filter results — only results whose title, URL, or content match are returned")
user_location: Optional[UserLocation] = Field(default=None, description="User location for localizing search results")
class SearchResult(BaseModel):
@ -80,6 +89,7 @@ class ExtractRequest(BaseModel):
urls: list[str] = Field(..., min_length=1, max_length=20, description="URLs to extract content from")
extract_depth: ExtractDepth = Field(default=ExtractDepth.basic, description="Extraction depth")
format: ContentFormat = Field(default=ContentFormat.markdown, description="Output format")
max_content_length: Optional[int] = Field(default=None, ge=100, le=500000, description="Max chars per extracted page (overrides server default)")
class ExtractResult(BaseModel):

View file

@ -56,6 +56,7 @@ async def _do_extract(request: Request, body: ExtractRequest) -> ExtractResponse
if urls_to_fetch:
extractions = await extractor.extract_urls(
urls_to_fetch, output_format=body.format.value,
max_content_length=body.max_content_length,
)
to_cache: list[tuple[str, str]] = []
for extraction in extractions:

View file

@ -1,7 +1,9 @@
import asyncio
import hashlib
import json
import re
import time
from urllib.parse import parse_qs, urlparse
import structlog
from fastapi import APIRouter, Depends, HTTPException, Request
@ -20,6 +22,163 @@ logger = structlog.get_logger(__name__)
router = APIRouter()
_COUNTRY_LANG_MAP = {
"RU": "ru-RU", "US": "en-US", "GB": "en-GB", "DE": "de-DE", "FR": "fr-FR",
"ES": "es-ES", "IT": "it-IT", "PT": "pt-PT", "BR": "pt-BR", "NL": "nl-NL",
"PL": "pl-PL", "CZ": "cs-CZ", "JP": "ja-JP", "KR": "ko-KR", "CN": "zh-CN",
"TW": "zh-TW", "TR": "tr-TR", "UA": "uk-UA", "KZ": "kk-KZ", "BY": "be-BY",
"IN": "hi-IN", "AR": "es-AR", "MX": "es-MX", "SE": "sv-SE", "NO": "no-NO",
"FI": "fi-FI", "DK": "da-DK",
}
_STRIP_HOST_PREFIXES = ("www.", "amp.", "m.", "mobile.")
_STRIP_QUERY_PARAMS = {"utm_source", "utm_medium", "utm_campaign", "utm_term",
"utm_content", "ref", "fbclid", "gclid", "yclid"}
def _normalize_url(url: str) -> str:
"""Normalize URL for deduplication.
Strips scheme, common host prefixes (www/amp/m/mobile), trailing slash,
and tracking query parameters (UTM, fbclid, etc.).
"""
p = urlparse(url)
host = p.hostname or ""
for prefix in _STRIP_HOST_PREFIXES:
if host.startswith(prefix):
host = host[len(prefix):]
break
path = p.path.rstrip("/")
# Strip tracking query params
if p.query:
qs = parse_qs(p.query, keep_blank_values=True)
cleaned = {k: v for k, v in qs.items() if k.lower() not in _STRIP_QUERY_PARAMS}
if cleaned:
query_str = "&".join(f"{k}={v[0]}" for k, v in sorted(cleaned.items()))
return f"{host}{path}?{query_str}"
return f"{host}{path}"
def _deduplicate_results(results: list) -> list:
"""Remove duplicate URLs, keeping the first (highest-scored) occurrence."""
seen: set[str] = set()
deduped = []
for r in results:
key = _normalize_url(r.url)
if key not in seen:
seen.add(key)
deduped.append(r)
return deduped
def _find_relevant_snippet(query: str, extracted_text: str, max_length: int = 500) -> str:
"""Find the most query-relevant window in extracted text.
Tokenizes the query into keywords, scores sliding windows of the text
by keyword hit count, and returns the best window. Falls back to the
beginning of the text if no keywords match.
"""
if not extracted_text or len(extracted_text) <= max_length:
return extracted_text or ""
# Build keyword set from query (words >= 3 chars, lowered)
keywords = {w.lower() for w in re.findall(r'\w+', query) if len(w) >= 3}
if not keywords:
return extracted_text[:max_length]
text_lower = extracted_text.lower()
best_score = 0
best_start = 0
# Slide a window across the text, stepping by 100 chars for speed
step = 100
for start in range(0, max(1, len(extracted_text) - max_length + 1), step):
window = text_lower[start:start + max_length]
score = sum(1 for kw in keywords if kw in window)
if score > best_score:
best_score = score
best_start = start
if score == len(keywords):
break # all keywords found, stop early
snippet = extracted_text[best_start:best_start + max_length]
# Trim to last word boundary to avoid cutting mid-word
if best_start + max_length < len(extracted_text):
last_space = snippet.rfind(" ")
if last_space > max_length // 2:
snippet = snippet[:last_space]
return snippet
async def _enrich_snippets(
query: str,
results: list[SearchResult],
extractor,
enrich_count: int,
cache=None,
) -> None:
"""Enrich short snippets with extracted page content (query-relevant window).
Checks the extract cache first to avoid redundant HTTP requests.
If all initial extractions fail, retries with the next batch of URLs.
"""
enrich_count = min(enrich_count, len(results))
async def _enrich_batch(batch: list[SearchResult]) -> int:
urls = [r.url for r in batch]
# Check cache first
cached: dict[str, str | None] = {}
if cache:
cached = await cache.get_extract_batch(urls)
need_extract = [url for url in urls if not cached.get(url)]
extractions_map: dict[str, str | None] = {url: cached.get(url) for url in urls}
if need_extract:
fresh = await extractor.extract_urls(need_extract)
for extraction in fresh:
if extraction.success and extraction.content:
extractions_map[extraction.url] = extraction.content
enriched = 0
for result in batch:
content = extractions_map.get(result.url)
if content and len(result.content) < 80:
result.content = _find_relevant_snippet(query, content)
enriched += 1
return enriched
enriched = await _enrich_batch(results[:enrich_count])
# Auto-retry: if no enrichments succeeded, try the next batch
if enriched == 0 and len(results) > enrich_count:
remaining = results[enrich_count:]
retry_count = min(enrich_count, len(remaining))
await _enrich_batch(remaining[:retry_count])
def _safe_regex_filter(pattern_str: str, results: list[SearchResult]) -> list[SearchResult]:
"""Filter results by regex, compiling with a size guard against ReDoS."""
if len(pattern_str) > 200:
logger.warning("filter_query_too_long", length=len(pattern_str))
return results
try:
pattern = re.compile(pattern_str, re.IGNORECASE)
except re.error:
logger.warning("invalid_filter_query", filter_query=pattern_str)
return results
return [
r for r in results
if pattern.search(r.title) or pattern.search(r.url) or pattern.search(r.content)
]
def _country_to_searxng_lang(country: str) -> str:
code = country.upper().strip()
if lang := _COUNTRY_LANG_MAP.get(code):
return lang
# Fallback: use country code as language (e.g. "ru" -> "ru")
return code.lower()
def _params_hash(req: SearchRequest) -> str:
data = req.model_dump(exclude={"query"})
return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()[:12]
@ -56,41 +215,74 @@ async def _do_search(request: Request, body: SearchRequest) -> SearchResponse:
elapsed = time.perf_counter() - start
return SearchResponse(**{**cached, "response_time": round(elapsed, 3)})
# Overfetch when reranking is enabled to give the reranker a larger candidate pool
fetch_count = body.max_results
if settings.rerank.enabled:
fetch_count = min(body.max_results * 3, 40)
# Derive SearXNG language code from user_location country
language = None
if body.user_location and body.user_location.country:
language = _country_to_searxng_lang(body.user_location.country)
# Query search backend with graceful degradation
try:
backend_resp = await backend.search(
body.query,
max_results=body.max_results,
max_results=fetch_count,
topic=body.topic.value,
time_range=body.time_range.value if body.time_range else None,
include_domains=body.include_domains or None,
exclude_domains=body.exclude_domains or None,
include_images=body.include_images,
language=language,
)
except Exception as e:
logger.error("search_backend_failed", error=str(e))
stale = await cache.get_search(body.query, ph)
stale = await cache.get_search(body.query, ph, allow_stale=True)
if stale:
elapsed = time.perf_counter() - start
logger.info("serving_stale_cache", query=body.query)
return SearchResponse(**{**stale, "response_time": round(elapsed, 3)})
raise HTTPException(503, "Search service unavailable")
# Build results
# Build results, separate infobox entries, and deduplicate
results: list[SearchResult] = []
infobox_results: list[SearchResult] = []
for raw in backend_resp.results:
results.append(
SearchResult(
title=raw.title, url=raw.url, content=raw.snippet,
score=raw.score, raw_content=None,
)
sr = SearchResult(
title=raw.title, url=raw.url, content=raw.snippet,
score=raw.score, raw_content=None,
)
if raw.is_infobox:
infobox_results.append(sr)
else:
results.append(sr)
results = _deduplicate_results(results)
# Rerank results
# Rerank only non-infobox results, then prepend infobox
if reranker and settings.rerank.enabled and results:
result_dicts = [r.model_dump() for r in results]
reranked = reranker.rerank(body.query, result_dicts, top_k=body.max_results)
reranked = reranker.rerank(body.query, result_dicts, top_k=max(body.max_results, settings.rerank.top_k))
results = [SearchResult(**r) for r in reranked]
results = infobox_results + results
# Dynamic filtering: keep only results matching the filter regex
if body.filter_query and results:
results = _safe_regex_filter(body.filter_query, results)
# Snippet enrichment: extract content for top-N results in basic mode
if (
body.search_depth.value == "basic"
and not body.include_raw_content
and settings.extraction.snippet_enrichment
and results
):
await _enrich_snippets(
body.query, results, extractor,
settings.extraction.snippet_enrichment_count,
cache=cache,
)
# Advanced depth: fetch and extract content
if body.search_depth.value == "advanced" or body.include_raw_content:

View file

@ -10,6 +10,12 @@ from app.auth import verify_api_key
from app.config import settings
from app.models.schemas import SearchRequest, SearchResult
from app.rate_limit import limiter
from app.routers.search import (
_country_to_searxng_lang,
_deduplicate_results,
_enrich_snippets,
_safe_regex_filter,
)
logger = structlog.get_logger(__name__)
router = APIRouter()
@ -26,30 +32,75 @@ async def search_stream(
start = time.perf_counter()
backend = request.app.state.search_backend
extractor = request.app.state.extractor
cache = request.app.state.cache
# Overfetch when reranking is enabled
fetch_count = body.max_results
if settings.rerank.enabled:
fetch_count = min(body.max_results * 3, 40)
# Derive SearXNG language code from user_location country
language = None
if body.user_location and body.user_location.country:
language = _country_to_searxng_lang(body.user_location.country)
# Search
try:
backend_resp = await backend.search(
body.query,
max_results=body.max_results,
max_results=fetch_count,
topic=body.topic.value,
time_range=body.time_range.value if body.time_range else None,
include_domains=body.include_domains or None,
exclude_domains=body.exclude_domains or None,
include_images=body.include_images,
language=language,
)
except Exception as e:
yield {"event": "error", "data": json.dumps({"error": str(e)})}
return
# Stream each result
# Build results, separate infobox entries, and deduplicate
results = []
infobox_results = []
for raw in backend_resp.results:
result = SearchResult(
sr = SearchResult(
title=raw.title, url=raw.url, content=raw.snippet,
score=raw.score, raw_content=None,
)
results.append(result)
if raw.score == 1.0 and results == [] and raw.snippet and len(raw.snippet) > 100:
infobox_results.append(sr)
else:
results.append(sr)
results = _deduplicate_results(results)
# Rerank only non-infobox results, then prepend infobox
reranker = request.app.state.reranker
if reranker and settings.rerank.enabled and results:
result_dicts = [r.model_dump() for r in results]
reranked = reranker.rerank(body.query, result_dicts, top_k=body.max_results)
results = [SearchResult(**r) for r in reranked]
results = infobox_results + results
# Dynamic filtering
if body.filter_query and results:
results = _safe_regex_filter(body.filter_query, results)
# Snippet enrichment for basic mode (shared logic with sync endpoint)
if (
body.search_depth.value == "basic"
and not body.include_raw_content
and settings.extraction.snippet_enrichment
and results
):
await _enrich_snippets(
body.query, results, extractor,
settings.extraction.snippet_enrichment_count,
cache=cache,
)
# Stream each result
for result in results:
yield {"event": "result", "data": json.dumps(result.model_dump())}
# Stream images
@ -62,15 +113,15 @@ async def search_stream(
# If advanced, stream extractions as they complete
if body.search_depth.value == "advanced" or body.include_raw_content:
tasks = {
url: asyncio.create_task(extractor.extract_url(r.url))
for r, url in ((r, r.url) for r in results)
asyncio.create_task(extractor.extract_url(r.url)): r.url
for r in results
}
for url, task in tasks.items():
extraction = await task
for coro in asyncio.as_completed(tasks):
extraction = await coro
if extraction.success:
yield {
"event": "extraction",
"data": json.dumps({"url": url, "raw_content": extraction.content}),
"data": json.dumps({"url": extraction.url, "raw_content": extraction.content}),
}
# AI answer generation (streamed)

View file

@ -2,6 +2,7 @@ from __future__ import annotations
import hashlib
import json
import time
from typing import Any, Optional
import redis.asyncio as redis
@ -47,15 +48,25 @@ class CacheService:
# --- Search cache ---
async def get_search(self, query: str, params_hash: str) -> Optional[dict[str, Any]]:
async def get_search(
self, query: str, params_hash: str, *, allow_stale: bool = False,
) -> Optional[dict[str, Any]]:
if not self.enabled or not self._redis:
return None
key = self._hash_key("search", f"{query}|{params_hash}")
try:
data = await self._redis.get(key)
if data:
logger.debug("cache_hit", type="search", query=query)
return json.loads(data)
parsed = json.loads(data)
fresh_until = parsed.pop("_fresh_until", 0)
is_fresh = time.time() < fresh_until
if is_fresh:
logger.debug("cache_hit", type="search", query=query)
return parsed
if allow_stale:
logger.debug("cache_hit_stale", type="search", query=query)
return parsed
return None
except Exception as e:
logger.warning("cache_read_error", error=str(e))
return None
@ -64,8 +75,10 @@ class CacheService:
if not self.enabled or not self._redis:
return
key = self._hash_key("search", f"{query}|{params_hash}")
# Store with fresh_until timestamp; Redis TTL = 3x search_ttl for stale serving
envelope = {**data, "_fresh_until": time.time() + self.search_ttl}
try:
await self._redis.setex(key, self.search_ttl, json.dumps(data))
await self._redis.setex(key, self.search_ttl * 3, json.dumps(envelope))
except Exception as e:
logger.warning("cache_write_error", error=str(e))

View file

@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import gzip
import random
import re
from collections import OrderedDict
@ -38,6 +39,7 @@ class ContentExtractor:
self._domain_concurrency = config.extraction.domain_concurrency
self._max_domains = config.extraction.domain_semaphore_max_size
self._global_semaphore = asyncio.Semaphore(config.extraction.max_concurrent)
self._js_semaphore = asyncio.Semaphore(config.extraction.js_max_concurrent)
def _get_domain_semaphore(self, url: str) -> asyncio.Semaphore:
domain = urlparse(url).netloc
@ -55,27 +57,37 @@ class ContentExtractor:
return {
"User-Agent": ua,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Language": "ru,en-US;q=0.9,en;q=0.8,*;q=0.5",
"Accept-Encoding": "gzip, deflate, br",
"DNT": "1",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
}
async def extract_url(self, url: str, *, output_format: str = "markdown") -> ExtractionResult:
async def extract_url(self, url: str, *, output_format: str = "markdown", max_content_length: int | None = None) -> ExtractionResult:
domain_sem = self._get_domain_semaphore(url)
async with self._global_semaphore, domain_sem:
return await self._fetch_and_extract(url, output_format)
return await self._fetch_and_extract(url, output_format, max_content_length)
async def _fetch_and_extract(self, url: str, output_format: str) -> ExtractionResult:
async def _fetch_and_extract(self, url: str, output_format: str, max_content_length: int | None = None) -> ExtractionResult:
try:
resp = await self._fetch_url(url)
html = resp.text
max_len = max_content_length or self.config.extraction.max_content_length
content_type = resp.headers.get("content-type", "")
if "application/pdf" in content_type:
extracted = self._extract_pdf(resp.content, url, output_format)
if not extracted:
return ExtractionResult(url=url, error="Failed to extract PDF content")
if len(extracted) > max_len:
extracted = extracted[:max_len] + "\n\n[Content truncated]"
return ExtractionResult(url=url, content=extracted)
if "text/html" not in content_type and "application/xhtml" not in content_type:
return ExtractionResult(url=url, error=f"Non-HTML content type: {content_type}")
html = self._decode_response(resp)
# Tier 1: trafilatura
extracted = self._extract_trafilatura(html, url, output_format)
@ -84,10 +96,14 @@ class ContentExtractor:
logger.info("trafilatura_fallback", url=url)
extracted = self._extract_readability(html, output_format)
# Tier 3: JS rendering fallback via Browserless
if not extracted and self.config.extraction.js_fallback:
logger.info("js_fallback", url=url)
extracted = await self._extract_js(url, output_format)
if not extracted:
return ExtractionResult(url=url, error="No content could be extracted")
max_len = self.config.extraction.max_content_length
if len(extracted) > max_len:
extracted = extracted[:max_len] + "\n\n[Content truncated]"
@ -119,6 +135,51 @@ class ContentExtractor:
resp.raise_for_status()
return resp
@staticmethod
def _decode_response(resp: httpx.Response) -> str:
"""Decode response with charset detection and decompression fallback.
httpx uses Content-Type charset or default_encoding (set on the client).
If the response is still compressed (e.g. brotli library missing),
attempt manual decompression. If the decoded text still contains
replacement characters, re-detect encoding from raw bytes.
"""
raw = resp.content
content_encoding = resp.headers.get("content-encoding", "").lower()
# Safety net: if httpx didn't decompress (e.g. missing brotli package),
# detect by checking for compressed magic bytes or content-encoding header.
if content_encoding == "br" and raw[:4] != b"<" and raw[:4] != b"\xef\xbb\xbf":
try:
import brotli
raw = brotli.decompress(raw)
except Exception:
logger.warning("brotli_decompress_failed", url=resp.url)
elif content_encoding == "gzip" and raw[:2] == b"\x1f\x8b":
try:
raw = gzip.decompress(raw)
except Exception:
logger.warning("gzip_decompress_failed", url=resp.url)
# Decode bytes to text
charset = resp.charset_encoding
if charset:
try:
text = raw.decode(charset)
except (UnicodeDecodeError, LookupError):
text = None
else:
if "\ufffd" not in text:
return text
# Fallback: auto-detect encoding
from charset_normalizer import from_bytes
detected = from_bytes(raw).best()
if detected:
return str(detected)
return raw.decode("utf-8", errors="replace")
def _extract_trafilatura(self, html: str, url: str, output_format: str) -> str | None:
extracted = trafilatura.extract(
html,
@ -167,6 +228,63 @@ class ContentExtractor:
lines.append(text)
return "\n".join(lines)
async def extract_urls(self, urls: list[str], *, output_format: str = "markdown") -> list[ExtractionResult]:
tasks = [self.extract_url(url, output_format=output_format) for url in urls]
def _extract_pdf(self, raw: bytes, url: str, output_format: str) -> str | None:
"""Extract text content from a PDF document."""
try:
import pymupdf
except ImportError:
logger.warning("pymupdf_not_installed", url=url)
return None
try:
doc = pymupdf.open(stream=raw, filetype="pdf")
pages = []
for page in doc:
pages.append(page.get_text())
doc.close()
text = "\n\n".join(pages).strip()
if not text:
return None
if output_format == "markdown":
return f"# PDF Document\n\nSource: {url}\n\n---\n\n{text}"
return text
except Exception:
logger.exception("pdf_extraction_failed", url=url)
return None
async def _extract_js(self, url: str, output_format: str) -> str | None:
"""Tier 3: render page with headless browser via Browserless, then extract."""
try:
from playwright.async_api import async_playwright
except ImportError:
logger.warning("playwright_not_installed", url=url)
return None
cfg = self.config.extraction
try:
async with self._js_semaphore:
async with async_playwright() as p:
browser = await p.chromium.connect_over_cdp(cfg.browserless_url)
try:
context = await browser.new_context()
page = await context.new_page()
try:
await page.goto(
url,
wait_until=cfg.js_wait_for,
timeout=cfg.js_timeout * 1000,
)
html = await page.content()
finally:
await context.close()
finally:
await browser.close()
return self._extract_trafilatura(html, url, output_format)
except Exception:
logger.exception("js_extraction_failed", url=url)
return None
async def extract_urls(self, urls: list[str], *, output_format: str = "markdown", max_content_length: int | None = None) -> list[ExtractionResult]:
tasks = [self.extract_url(url, output_format=output_format, max_content_length=max_content_length) for url in urls]
return await asyncio.gather(*tasks)

View file

@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import math
from abc import ABC, abstractmethod
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
@ -22,6 +23,7 @@ class RawSearchResult:
url: str
snippet: str
score: float = 0.0
is_infobox: bool = False
@dataclass
@ -48,6 +50,7 @@ class SearchBackend(ABC):
include_domains: Optional[list[str]] = None,
exclude_domains: Optional[list[str]] = None,
include_images: bool = False,
language: Optional[str] = None,
) -> BackendSearchResponse:
...
@ -67,6 +70,7 @@ class SearXNGBackend(SearchBackend):
include_domains: Optional[list[str]] = None,
exclude_domains: Optional[list[str]] = None,
include_images: bool = False,
language: Optional[str] = None,
) -> BackendSearchResponse:
effective_query = query
if include_domains:
@ -77,7 +81,7 @@ class SearXNGBackend(SearchBackend):
categories = "news" if topic == "news" else "general"
searxng_time_range = time_range if time_range in ("day", "week", "month", "year") else None
tasks = [self._web_search_with_retry(effective_query, max_results, categories, searxng_time_range)]
tasks = [self._web_search_with_retry(effective_query, max_results, categories, searxng_time_range, language)]
if include_images:
tasks.append(self._image_search_with_retry(query, max_results=5))
@ -101,12 +105,13 @@ class SearXNGBackend(SearchBackend):
return BackendSearchResponse(results=web_results, images=images)
async def _web_search_with_retry(
self, query: str, max_results: int, categories: str, time_range: Optional[str]
self, query: str, max_results: int, categories: str, time_range: Optional[str],
language: Optional[str] = None,
) -> list[RawSearchResult]:
@search_circuit
async def _do():
return await retry_async(
lambda: self._web_search(query, max_results, categories, time_range)
lambda: self._web_search(query, max_results, categories, time_range, language)
)
return await _do()
@ -117,26 +122,52 @@ class SearXNGBackend(SearchBackend):
return await _do()
async def _web_search(
self, query: str, max_results: int, categories: str, time_range: Optional[str]
self, query: str, max_results: int, categories: str, time_range: Optional[str],
language: Optional[str] = None,
) -> list[RawSearchResult]:
params: dict = {"q": query, "format": "json", "categories": categories}
if time_range:
params["time_range"] = time_range
if language:
params["language"] = language
resp = await self.client.get(f"{self.base_url}/search", params=params)
resp.raise_for_status()
data = resp.json()
results: list[RawSearchResult] = []
for item in data.get("results", [])[:max_results]:
# Parse infoboxes — rich structured data that SearXNG sometimes returns
for box in data.get("infoboxes", [])[:1]:
box_content = box.get("content", "")
box_urls = box.get("urls", [])
box_url = box_urls[0].get("url", "") if box_urls else ""
if box_content and box_url:
results.append(
RawSearchResult(
title=box.get("infobox", ""),
url=box_url,
snippet=box_content[:500],
score=1.0,
is_infobox=True,
)
)
raw_items = data.get("results", [])[:max_results]
for item in raw_items:
raw_score = item.get("score", 0)
score = min(1.0, raw_score) if raw_score <= 1.0 else min(1.0, raw_score / 10.0)
# Preserve score spread: clamp small values, compress large ones logarithmically
if raw_score <= 1.0:
score = min(1.0, raw_score)
else:
score = min(1.0, math.log1p(raw_score) / math.log1p(10))
score = round(score, 4)
results.append(
RawSearchResult(
title=item.get("title", ""),
url=item.get("url", ""),
snippet=item.get("content", ""),
score=round(score, 4),
score=score,
)
)
return results
@ -169,6 +200,7 @@ class DuckDuckGoBackend(SearchBackend):
include_domains: Optional[list[str]] = None,
exclude_domains: Optional[list[str]] = None,
include_images: bool = False,
language: Optional[str] = None,
) -> BackendSearchResponse:
effective_query = query
if include_domains:

View file

@ -14,6 +14,13 @@ extraction:
max_content_length: 50000
domain_concurrency: 2
domain_semaphore_max_size: 1000
snippet_enrichment: true
snippet_enrichment_count: 2
js_fallback: false # Enable JS rendering fallback via Browserless (requires 'js' profile)
browserless_url: "ws://browserless:3000"
js_timeout: 15
js_wait_for: "networkidle" # load | networkidle | domcontentloaded
js_max_concurrent: 3
user_agents:
- "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"
@ -49,7 +56,7 @@ rate_limit:
extract_rate: "30/minute"
rerank:
enabled: false
enabled: true
model: "ms-marco-MiniLM-L-12-v2"
top_k: 5

View file

@ -3,7 +3,7 @@ services:
image: searxng/searxng:latest
container_name: orio-search-searxng
ports:
- "8080:8080"
- "${SEARXNG_PORT:-8080}:8080"
volumes:
- ./searxng/settings.yml:/etc/searxng/settings.yml:ro
restart: unless-stopped
@ -29,6 +29,27 @@ services:
limits:
memory: 256M
browserless:
image: browserless/chrome:latest
container_name: orio-search-browserless
profiles: ["js"]
ports:
- "${BROWSERLESS_PORT:-3000}:3000"
environment:
- MAX_CONCURRENT_SESSIONS=5
- CONNECTION_TIMEOUT=15000
- DEFAULT_LAUNCH_ARGS=["--no-sandbox","--disable-gpu"]
restart: unless-stopped
healthcheck:
test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://localhost:3000/"]
interval: 30s
timeout: 5s
retries: 3
deploy:
resources:
limits:
memory: 1G
ollama:
image: ollama/ollama:latest
container_name: orio-search-ollama
@ -47,7 +68,7 @@ services:
build: .
container_name: orio-search-api
ports:
- "8000:8000"
- "${API_PORT:-8000}:8000"
volumes:
- ./config.yaml:/app/config.yaml:ro
depends_on:

View file

@ -1,6 +1,8 @@
fastapi==0.115.6
uvicorn[standard]==0.34.0
httpx==0.28.1
brotli==1.1.0
charset-normalizer==3.4.1
trafilatura==2.0.0
redis[hiredis]==5.2.1
pyyaml==6.0.2
@ -15,6 +17,8 @@ flashrank==0.2.9
readability-lxml==0.8.1
circuitbreaker==2.0.0
openai>=1.12.0
playwright>=1.40.0
pymupdf>=1.24.0
# Dev / Testing
pytest==8.3.4

View file

@ -8,6 +8,36 @@ search:
formats:
- html
- json
default_lang: "auto"
general:
enable_metrics: false
engines:
# Приоритетные движки
- name: google
weight: 1.2
disabled: false
- name: yandex
weight: 1.1
disabled: false
- name: brave
weight: 0.7
disabled: false
- name: duckduckgo
weight: 0.8
disabled: false
- name: wikipedia
weight: 0.6
disabled: false
# Отключённые движки
- name: bing
disabled: true
- name: yahoo
disabled: true