diff --git a/app/config.py b/app/config.py index a11f949..6ae0caa 100644 --- a/app/config.py +++ b/app/config.py @@ -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", diff --git a/app/main.py b/app/main.py index d8beee5..7ddd002 100644 --- a/app/main.py +++ b/app/main.py @@ -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 diff --git a/app/models/schemas.py b/app/models/schemas.py index 6f5e3f6..3b0e87e 100644 --- a/app/models/schemas.py +++ b/app/models/schemas.py @@ -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): diff --git a/app/routers/extract.py b/app/routers/extract.py index c7d34a3..31a9469 100644 --- a/app/routers/extract.py +++ b/app/routers/extract.py @@ -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: diff --git a/app/routers/search.py b/app/routers/search.py index 13835dd..d2bc2f5 100644 --- a/app/routers/search.py +++ b/app/routers/search.py @@ -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: diff --git a/app/routers/search_stream.py b/app/routers/search_stream.py index 3999300..d0aa089 100644 --- a/app/routers/search_stream.py +++ b/app/routers/search_stream.py @@ -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) diff --git a/app/services/cache.py b/app/services/cache.py index a34a47a..311772a 100644 --- a/app/services/cache.py +++ b/app/services/cache.py @@ -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)) diff --git a/app/services/extractor.py b/app/services/extractor.py index 284c1cb..7204e15 100644 --- a/app/services/extractor.py +++ b/app/services/extractor.py @@ -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) diff --git a/app/services/search_backend.py b/app/services/search_backend.py index 8092850..8c2da8e 100644 --- a/app/services/search_backend.py +++ b/app/services/search_backend.py @@ -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: diff --git a/config.yaml b/config.yaml index 30e2119..2b53c49 100644 --- a/config.yaml +++ b/config.yaml @@ -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 diff --git a/docker-compose.yml b/docker-compose.yml index 487f363..d8e0dfa 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -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: diff --git a/requirements.txt b/requirements.txt index b2c8f78..8df0c85 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/searxng/settings.yml b/searxng/settings.yml index ae1772c..b755658 100644 --- a/searxng/settings.yml +++ b/searxng/settings.yml @@ -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