""" Web Fetch Module - Platform-agnostic web content fetching. This module provides URL extraction, web content fetching, and source citation building without platform-specific dependencies. """ import re import logging from enum import Enum from typing import Optional, Dict, List, Callable, Awaitable # Heavy crawler dependencies are optional. Install via: # pip install 'aborist[crawler]' try: from miniuri import Uri except ImportError as e: # pragma: no cover raise ImportError( "crawler module requires extras: pip install 'aborist[crawler]'" ) from e # `core.keywords` lived in agents.ai.unturf.com and detected fetch-intent # from chat-bot messages. Aborist has no chat surface, so the verbatim # lift drops those triggers. Callers asking "is this content a fetch # request?" should adapt at the application layer instead. def has_fresh_fetch_trigger(content: str) -> bool: # pragma: no cover raise NotImplementedError( "chat-bot fetch triggers were dropped during the aborist lift; " "callers should detect fetch intent at the application layer" ) def has_web_fetch_trigger(content: str) -> bool: # pragma: no cover raise NotImplementedError( "chat-bot fetch triggers were dropped during the aborist lift; " "callers should detect fetch intent at the application layer" ) logger = logging.getLogger(__name__) class URIContentType(Enum): """Content type categories for URIs.""" PAGE = "page" # HTML, text pages - fetch and extract PDF = "pdf" # PDF documents - fetch and extract IMAGE = "image" # Images (png, jpg, gif, webp) - use vision SVG = "svg" # SVG - convert to PNG then vision VIDEO = "video" # Video - skip or extract metadata AUDIO = "audio" # Audio - skip or transcribe UNKNOWN = "unknown" # Unknown - try fetch, fallback to vision # Content-type mapping CONTENT_TYPE_MAP = { # Pages 'text/html': URIContentType.PAGE, 'application/xhtml+xml': URIContentType.PAGE, 'text/plain': URIContentType.PAGE, 'text/markdown': URIContentType.PAGE, 'application/json': URIContentType.PAGE, 'application/xml': URIContentType.PAGE, 'text/xml': URIContentType.PAGE, # PDF 'application/pdf': URIContentType.PDF, # Images 'image/png': URIContentType.IMAGE, 'image/jpeg': URIContentType.IMAGE, 'image/gif': URIContentType.IMAGE, 'image/webp': URIContentType.IMAGE, 'image/bmp': URIContentType.IMAGE, 'image/tiff': URIContentType.IMAGE, # SVG (special handling) 'image/svg+xml': URIContentType.SVG, # Video 'video/mp4': URIContentType.VIDEO, 'video/webm': URIContentType.VIDEO, 'video/ogg': URIContentType.VIDEO, # Audio 'audio/mpeg': URIContentType.AUDIO, 'audio/ogg': URIContentType.AUDIO, 'audio/wav': URIContentType.AUDIO, } async def detect_uri_content_type(url: str, session=None) -> URIContentType: """ Detect content type of URI using HEAD request. Args: url: URI to check session: Optional aiohttp session to reuse Returns: URIContentType enum value """ import aiohttp try: uri = Uri(url) path = uri.path or '' # Make HEAD request close_session = False if session is None: session = aiohttp.ClientSession() close_session = True try: async with session.head( url, allow_redirects=True, timeout=aiohttp.ClientTimeout(total=10), headers={'User-Agent': 'uncloseai.com/1.42 (ethical web crawler; +https://uncloseai.com)'} ) as response: content_type = response.headers.get('Content-Type', '') # Strip charset etc: "text/html; charset=utf-8" -> "text/html" mime_type = content_type.split(';')[0].strip().lower() logger.info(f"HEAD {url} -> {response.status}, Content-Type: {mime_type}") if mime_type in CONTENT_TYPE_MAP: return CONTENT_TYPE_MAP[mime_type] # Fallback heuristics based on extension if path.endswith('.pdf'): return URIContentType.PDF elif path.endswith(('.png', '.jpg', '.jpeg', '.gif', '.webp')): return URIContentType.IMAGE elif path.endswith('.svg'): return URIContentType.SVG # Default to page for unknown return URIContentType.PAGE finally: if close_session: await session.close() except Exception as e: logger.warning(f"Failed to detect content type for {url}: {e}") return URIContentType.UNKNOWN def extract_all_uris_from_content(content: str) -> List[str]: """ Extract ALL URIs from message content using regex. This function bypasses LLM tool calling to prevent URI hallucination. It extracts all URIs with protocol first, then any domain-like patterns. URIs inside code blocks are excluded. Args: content: Message content to extract URIs from Returns: List of extracted URI strings (may be empty) Examples: >>> extract_all_uris_from_content("Check https://a.com and https://b.com/file.css") ['https://a.com', 'https://b.com/file.css'] >>> extract_all_uris_from_content("```python\\nurl = 'https://ignore.me'\\n```\\nVisit https://real.com") ['https://real.com'] """ # Strip Matrix mentions first (e.g., @user:matrix.org) to avoid matching them as URIs content_cleaned = re.sub(r'@[\w.-]+:[\w.-]+', '', content) # Remove code blocks (``` ... ```) to avoid extracting URIs from code content_cleaned = re.sub(r'```[\s\S]*?```', '', content_cleaned) # Also remove inline code (`...`) content_cleaned = re.sub(r'`[^`]+`', '', content_cleaned) uris = [] # Find all URIs with protocol for match in re.finditer(r'https?://[^\s<>]+', content_cleaned): uri = match.group(0) if uri not in uris: uris.append(uri) logger.info(f"Extracted URI with protocol: {uri}") # Find URIs without protocol (e.g., "example.com", "media.unturf.com/test") # Exclude common file extensions that look like TLDs file_extensions = {'html', 'htm', 'css', 'js', 'jsx', 'tsx', 'ts', 'json', 'xml', 'svg', 'png', 'jpg', 'jpeg', 'gif', 'webp', 'ico', 'pdf', 'txt', 'md', 'py', 'java', 'cpp', 'hpp', 'go', 'rs', 'rb', 'php', 'sh', 'bash', 'yaml', 'yml', 'toml', 'ini', 'conf', 'log', 'csv', 'sql'} for match in re.finditer(r'(?:www\.)?(?:[a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}(?:/[^\s<>]*)?', content_cleaned): matched = match.group(0) # Check if this is followed by :number (error line reference like "file.html:98:35") end_pos = match.end() if end_pos < len(content_cleaned) and content_cleaned[end_pos] == ':': # Check if followed by digits remaining = content_cleaned[end_pos+1:] if remaining and remaining[0].isdigit(): logger.debug(f"Skipping error reference pattern: {matched}") continue # Extract the TLD (last part after final dot, before any path) domain_part = matched.split('/')[0] # Get just the domain tld = domain_part.split('.')[-1].lower() # Skip if it looks like a filename (TLD is a file extension) if tld in file_extensions: logger.debug(f"Skipping filename-like pattern: {matched}") continue uri = f"https://{matched}" # Check it wasn't already captured with protocol if uri not in uris and not any(uri in existing for existing in uris): uris.append(uri) logger.info(f"Extracted URI without protocol, added https://: {uri}") if not uris: logger.debug("No URIs found in content") return uris def extract_url_from_content(content: str) -> Optional[str]: """ Extract first URL from message content using regex. This function bypasses LLM tool calling to prevent URL hallucination. It first tries to match URLs with protocol (http:// or https://), then falls back to matching domain-like patterns and prepending https://. Args: content: Message content to extract URL from Returns: Extracted URL string or None if no URL found Examples: >>> extract_url_from_content("Check out https://example.com") 'https://example.com' >>> extract_url_from_content("Visit example.com") 'https://example.com' >>> extract_url_from_content("What about media.unturf.com/test") 'https://media.unturf.com/test' """ uris = extract_all_uris_from_content(content) return uris[0] if uris else None def detect_fresh_fetch_intent(content: str) -> bool: """ Detect if user wants to bypass cache and fetch fresh content. Uses centralized multilingual keywords (fresh, refresh, reload, update, etc.) Args: content: Message content to analyze Returns: True if fresh fetch is requested, False otherwise """ if has_fresh_fetch_trigger(content): logger.info("Fresh fetch intent detected via centralized keywords") return True return False def build_sources_footer(source_urls: List[Dict[str, str]], format: str = "markdown") -> str: """ Build sources footer from source URL metadata. Args: source_urls: List of dicts with 'url' and 'title' keys format: Output format - "markdown" (Discord), "plain" (IRC/Matrix), or "html" Returns: Formatted sources section Example: >>> sources = [{'url': 'https://example.com', 'title': 'Example Page'}] >>> build_sources_footer(sources) '**Sources:**\\n- **Example Page**: \\n' >>> build_sources_footer(sources, format="plain") 'Sources: Example Page (https://example.com)' """ if not source_urls: return "" if format == "plain": # Single line, no markdown, no emoji - for IRC and Matrix parts = [] for source in source_urls: title = source.get('title', 'Untitled') url = source.get('url', '') if url: parts.append(f"{title} ({url})") return "Sources: " + ", ".join(parts) if parts else "" else: # Markdown format - for Discord footer = "**Sources:**\n" for source in source_urls: title = source.get('title', 'Untitled') url = source.get('url', '') if url: # Use angle brackets to suppress Discord embeds footer += f"- **{title}**: <{url}>\n" return footer async def fetch_and_cache( url: str, web_cache_manager, keywords: Optional[List[str]] = None, keyword_variations: Optional[List[str]] = None, depth: int = 2, fresh: bool = False, progress_callback: Optional[Callable[[str], Awaitable[None]]] = None, user_query: Optional[str] = None ) -> Optional[Dict]: """ Fetch and cache web content using the web cache manager. This is a thin wrapper around web_cache_manager.get_or_fetch() that provides a platform-agnostic interface. Args: url: URL to fetch web_cache_manager: WebCacheManager instance keywords: Primary keywords for relevance scoring keyword_variations: Keyword variations/synonyms depth: Crawl depth (0 = single page, 1+ = follow links) fresh: If True, bypass cache and fetch fresh progress_callback: Optional async callback for progress updates user_query: Original user query for context Returns: Dict with keys: url, title, extracted_text, fetched_at, crawl_depth, source_urls, crawl_tree, token_count, from_cache Returns None if fetch failed """ try: result = await web_cache_manager.get_or_fetch( url=url, depth=depth, force_refresh=fresh, query_keywords=keywords if keywords else None, keyword_variations=keyword_variations if keyword_variations else None, progress_callback=progress_callback, user_query=user_query ) if result: logger.info(f"Web fetch successful: {url} ({len(result.get('source_urls', []))} pages)") else: logger.warning(f"Web fetch failed: {url}") return result except Exception as e: logger.error(f"Error fetching {url}: {e}", exc_info=True) return None def is_web_fetch_request(content: str) -> bool: """ Detect if message content is a web fetch request. Requires BOTH a URL AND a fetch keyword (fetch, scrape, crawl). Just having a URL is not enough - user must explicitly request fetch. Args: content: Message content to analyze Returns: True if this appears to be a web fetch request """ url = extract_url_from_content(content) if not url: return False return has_web_fetch_trigger(content) def extract_domain(url: str) -> Optional[str]: """ Extract domain from URL using miniuri. Args: url: URL to extract domain from Returns: Domain string or None if extraction fails Example: >>> extract_domain("https://example.com/path") 'example.com' """ try: uri = Uri(url) return uri.authority or None except Exception as e: logger.warning(f"Failed to extract domain from {url}: {e}") # Fallback to urlparse from urllib.parse import urlparse try: return urlparse(url).netloc or None except: return None def format_progress_message( url: str, task: str = "", keywords: Optional[List[str]] = None, variations: Optional[List[str]] = None, stems: Optional[List[str]] = None, format_instruction: str = "", depth: int = 2, cache_mode: str = "default", status: str = "", crawl_delay: Optional[float] = None ) -> str: """ Format a detailed progress message for web fetch operations. Args: url: URL being fetched task: Extracted task description keywords: Primary keywords variations: Keyword variations stems: Keyword stems format_instruction: Format instructions depth: Crawl depth cache_mode: "default" or "fresh" status: Current status message crawl_delay: Optional crawl delay in seconds Returns: Formatted progress message """ task_display = task if task else 'none' keywords_display = ', '.join(keywords) if keywords else 'none' variations_display = ', '.join(variations) if variations else 'none' stems_display = ', '.join(stems) if stems else 'none' format_display = format_instruction if format_instruction else 'default' msg = f"🔍 **Searching** <{url}>\n" msg += f"**Task:** {task_display}\n" msg += f"**Keywords:** {keywords_display}\n" msg += f"**Variations:** {variations_display}\n" msg += f"**Stems:** {stems_display}\n" msg += f"**Format:** {format_display}\n" msg += f"**Depth:** {depth}\n" msg += f"**Mode:** {cache_mode}" if crawl_delay and crawl_delay > 2.0: msg += f"\n**Crawl delay:** {crawl_delay}s (from robots.txt)" if status: msg += f"\n\n{status}" return msg def format_answering_message(num_pages: int) -> str: """ Format a message for the answering phase after fetch completes. Args: num_pages: Number of pages fetched Returns: Formatted answering message """ return f"💭 **Answering your research...**\nAnalyzing {num_pages} page(s)" def generate_keyword_stems(keywords: List[str], variations: List[str]) -> List[str]: """ Generate stems from keywords and variations using simple suffix stripping. This is a copy of the function from bot.py to avoid circular imports. Args: keywords: Primary keywords variations: Keyword variations Returns: List of unique stems (excluding duplicates from keywords/variations) """ def simple_stem(word: str) -> str: """Simple suffix-stripping stemmer for common English suffixes""" word = word.lower() # Common suffixes in order of longest first suffixes = ['ings', 'ing', 'ed', 'es', 's', 'ly', 'er', 'est', 'tion', 'sion', 'ness', 'ment', 'able', 'ible'] for suffix in suffixes: if word.endswith(suffix) and len(word) > len(suffix) + 2: # Keep at least 3 chars in stem return word[:-len(suffix)] return word stems = [] # Lowercase all keywords and variations for comparison keywords_lower = [kw.lower() for kw in keywords] variations_lower = [v.lower() for v in variations] all_words = keywords + variations for word in all_words: # Split phrases into individual words and stem each separately individual_words = word.split() for individual_word in individual_words: stem = simple_stem(individual_word) # Only add if: # 1. Stem is different from original word (stemming actually changed it) # 2. Stem is not already in keywords (case-insensitive) # 3. Stem is not already in variations (case-insensitive) if (stem != individual_word.lower() and stem not in keywords_lower and stem not in variations_lower): stems.append(stem) # Deduplicate stems return list(dict.fromkeys(stems))