# src/analyzer/chat/chat_tools.py from __future__ import annotations import asyncio import logging import re import difflib from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional from ..config import load_config from ..llm_client import LLMClient from ..prompt_templates import build_prompt from ..context_builder import build_context_with_supporting from ..utils.errors import DataLoadError, ValidationError, LLMError from ..utils.text import clean, to_number from ..utils.dates import parse_date, format_date from ..summarizer_optimized import ( # NEW: Optimized caching + batch processing SummaryCache, summarize_grants_async, extract_minimal_context, ) # Optional: try to load hybrid search index (requires scikit-learn) try: from ..search.hybrid_index import load_index, search_by_grant_id HAS_SEARCH_INDEX = True except ImportError: HAS_SEARCH_INDEX = False load_index = None search_by_grant_id = None # --------------------------------------------------------------------- # Utility helpers (moved to utils modules) # --------------------------------------------------------------------- # Date parsing: use utils.dates.parse_date() and format_date() # Text normalization: use utils.text.clean() # Number parsing: use utils.text.to_number() # Keep backward compatibility wrappers _parse_date = parse_date _fmt_date = format_date # Helper to normalize complex objects to searchable text def _norm(s: Any) -> str: """Normalize any object to searchable text string.""" if s is None: return "" if isinstance(s, (list, tuple, set)): return " ".join(_norm(x) for x in s if x) if isinstance(s, dict): return " ".join(_norm(v) for v in s.values() if v) return clean(str(s)) # Use utils.text.clean() for final normalization # Fuzzy matching helper - supports partial/typo matches def _fuzzy_match(keyword: str, text: str, threshold: float = 0.6) -> bool: """ Check if keyword matches text using fuzzy matching. Returns True if: - Exact substring match (e.g., "ai" in "agentic ai") - Fuzzy word match with high similarity (e.g., "agent" vs "agentic" @ 86%) - Any word in text starts with the keyword Args: keyword: The search keyword text: The text to search in threshold: Minimum similarity score (0-1) for fuzzy match Returns: True if match found, False otherwise """ keyword_lower = keyword.lower() text_lower = text.lower() # Exact substring match (fastest, most common case) if keyword_lower in text_lower: return True # Split into words and try fuzzy matching on individual words text_words = text_lower.split() keyword_words = keyword_lower.split() for kw_word in keyword_words: # Check if any text word starts with the keyword word for text_word in text_words: # Allow leading punctuation in text_word clean_text_word = re.sub(r'^[^a-z0-9]+', '', text_word) if clean_text_word.startswith(kw_word): return True # Fuzzy match single words (e.g., "agent" vs "agentic") ratio = difflib.SequenceMatcher(None, kw_word, clean_text_word).ratio() if ratio >= threshold: return True return False # --------------------------------------------------------------------- # Main ChatTools class # --------------------------------------------------------------------- @dataclass class ChatTools: current: List[Dict[str, Any]] past: List[Dict[str, Any]] def __init__(self, current: List[Dict[str, Any]], past: List[Dict[str, Any]]) -> None: self.current = current or [] self.past = past or [] self.cfg = load_config() try: self.client = LLMClient(self.cfg) except Exception as e: logging.warning("LLMClient init failed: %s", e) self.client = None # Try to load hybrid index for enrichment (optional - requires scikit-learn) self.support_idx = None if HAS_SEARCH_INDEX: idx_path = Path("data/index/hybrid_index.pkl") try: self.support_idx = load_index() if idx_path.exists() else None except Exception as e: logging.warning("Could not load search index: %s", e) self.support_idx = None # Initialize cache for summaries (NEW: Optimized caching) self.summary_cache = SummaryCache(ttl_seconds=3600) # ----------------------------------------------------------------- # Status calculation (NEW) # ----------------------------------------------------------------- def _calculate_grant_status(self, grant: Dict[str, Any]) -> str: """ Calculate grant status based on open_date and close_date. Returns: "upcoming", "open", or "closed" """ today = datetime.now() close_date = _parse_date(grant.get("close_date") or grant.get("deadline")) open_date = _parse_date(grant.get("open_date")) # If we can't parse dates, assume open if not close_date: return "unknown" # If deadline has passed, it's closed if close_date < today: return "closed" # If hasn't opened yet, it's upcoming if open_date and open_date > today: return "upcoming" # Otherwise it's open return "open" # ----------------------------------------------------------------- # Listing grants # ----------------------------------------------------------------- def list_grants( self, keyword: Optional[str] = None, max_award: Optional[float] = None, audience: Optional[str] = None, status: Optional[str] = None, limit: Optional[int] = None, ) -> List[Dict[str, Any]]: """ List all grants (or filtered subset) sorted by deadline. Args: keyword: Filter by keyword in title/description max_award: Filter by maximum funding ceiling audience: Filter by audience type (not yet implemented) status: Filter by status - "open", "closed", "upcoming", or None for all limit: Max results to return. If None, returns ALL matching grants. Returns: List of grant dicts sorted by deadline, with status field included """ kw = (keyword or "").lower() results = [] for r in self.current: txt = _norm(r) # Use fuzzy matching instead of exact substring match if kw and not _fuzzy_match(kw, txt): continue if max_award is not None: ma = to_number(r.get("max_award") or r.get("funding_max")) if ma and ma > max_award: continue # Calculate status based on dates grant_status = self._calculate_grant_status(r) # Filter by status if specified if status is not None and grant_status != status: continue results.append( { "id": r.get("id") or r.get("competition_id"), "title": r.get("title") or "(untitled)", "deadline": r.get("deadline") or r.get("close_date") or "n/a", "status": grant_status, # NEW: Include status field } ) results.sort(key=lambda x: _parse_date(x.get("deadline")) or datetime.max) # KEY FIX: Return ALL results if limit is None, not hardcoded 5 if limit is None: return results return results[:limit] # ----------------------------------------------------------------- # Search past winners # ----------------------------------------------------------------- def search_past_winners( self, keyword: Optional[str] = None, competition: Optional[str] = None, limit: Optional[int] = None, ) -> List[Dict[str, Any]]: """ Search past winners by keyword or competition name. Args: keyword: Filter by keyword in project title, organization, or description competition: Filter by competition/programme name limit: Max results to return. If None, returns ALL matching winners. Returns: List of past winner records sorted by year (newest first) """ kw = (keyword or "").lower() comp = (competition or "").lower() results = [] for r in self.past: # Filter by keyword (searches project title, org name, and description) if kw: searchable_text = _norm({ "project_title": r.get("project_title"), "lead_org": r.get("lead_org"), "participant_name": r.get("participant_name"), "public_description": r.get("public_description"), "abstract": r.get("abstract"), }) # Use fuzzy matching instead of exact substring match if not _fuzzy_match(kw, searchable_text): continue # Filter by competition name if comp: comp_text = _norm({ "competition": r.get("competition"), "competition_title": r.get("competition_title"), "programme_title": r.get("programme_title"), }) # Use fuzzy matching instead of exact substring match if not _fuzzy_match(comp, comp_text): continue # Extract key fields for display results.append({ "project_title": r.get("project_title") or "(untitled)", "lead_org": r.get("lead_org") or r.get("participant_name") or "(unknown)", "competition": r.get("competition_title") or r.get("competition") or "(unknown)", "award_amount": r.get("award_amount"), "year": r.get("year") or r.get("project_start_date"), "abstract": r.get("public_description") or r.get("abstract"), }) # Sort by year (newest first) results.sort( key=lambda x: _parse_date(x.get("year")) or datetime.min, reverse=True ) # Return all results if limit is None if limit is None: return results return results[:limit] # ----------------------------------------------------------------- # Retrieve a single grant # ----------------------------------------------------------------- def get_grant(self, gid: str) -> Dict[str, Any]: gid = str(gid).replace("competition-", "").replace("grant-", "").strip().lower() for coll in (self.current, self.past): for r in coll: rid = str(r.get("id") or r.get("competition_id") or "").lower() if rid.replace("competition-", "") == gid: return r raise KeyError(f"Grant not found: {gid}") # ----------------------------------------------------------------- # Batch Summarize Multiple Grants (NEW - Parallelized) # ----------------------------------------------------------------- async def summarize_grants_batch( self, grant_ids: List[str], include_supporting: bool = False, batch_size: int = 5, ): """ Batch summarize multiple grants efficiently using parallel processing. This method: 1. Resolves grant IDs to grant objects 2. Processes them in parallel batches (5 per batch by default) 3. Caches results for future use 4. Yields results as they complete (parallelized) Args: grant_ids: List of grant IDs to summarize include_supporting: If True, include supporting materials (slower) batch_size: Number of grants per batch (default 5) Yields: Dict with grant_id, title, summary_md as each completes """ import asyncio # Resolve all grant IDs to actual grant objects grants_to_summarize = [] for gid in grant_ids: try: grant = self.get_grant(gid) grants_to_summarize.append(grant) except KeyError: logging.warning(f"Grant not found: {gid}") continue if not grants_to_summarize: logging.warning("No valid grants found to summarize") return logging.info( f"📦 Starting batch summarization of {len(grants_to_summarize)} grants " f"(batch_size={batch_size})" ) # Use the optimized async batch processing function try: results = await summarize_grants_async( grants_to_summarize, past_winners=self.past, client=self.client, cache=self.summary_cache, batch_size=batch_size, ) # Yield each result as it's ready for result in results: yield result except Exception as e: logging.error(f"Batch summarization failed: {e}") raise async def get_all_grant_summaries(self, batch_size: int = 5): """ Get summaries of ALL grants in a single efficient batch operation. This method: 1. Extracts all grant IDs from current database 2. Summarizes them all in parallel batches 3. Returns all results formatted for display Args: batch_size: Number of grants per batch (default 5) Yields: Dict with grant_id, title, summary_md as each completes """ # Get all grant IDs all_grants = self.list_grants(limit=None) # Get ALL grants all_grant_ids = [g["id"] for g in all_grants] if not all_grant_ids: logging.warning("No grants found in database") return logging.info(f"📦 Getting summaries for ALL {len(all_grant_ids)} grants in batch") # Use batch summarization with all IDs async for result in self.summarize_grants_batch(all_grant_ids, batch_size=batch_size): yield result # ----------------------------------------------------------------- # Summarize a grant # ----------------------------------------------------------------- def summarize_grant(self, gid: str, include_supporting: bool = True) -> Dict[str, Any]: """ Summarize a grant using LLM with caching. Args: gid: Grant ID include_supporting: If True, include supporting PDFs and materials in context Returns: Dict with summary_md, title, id """ row = self.get_grant(gid) title = row.get("title", "(untitled)") grant_id = row.get("id") or gid # NEW: Check cache first cached_summary = self.summary_cache.get(row) if cached_summary: logging.info("📦 Cache HIT for grant %s", grant_id) return { "summary_md": cached_summary, "title": title, "id": grant_id, } # Use enhanced context builder that includes supporting materials if include_supporting: try: context = build_context_with_supporting(row, k=5) except Exception as e: logging.warning("Failed to build context with supporting materials: %s", e) # Fallback to basic context context = self._build_basic_context(row) else: context = self._build_basic_context(row) if not self.client or not self.client.is_ready(): result = { "summary_md": f"LLM unavailable — context excerpt:\n\n{context[:1000]}", "title": title, "id": grant_id, } self.summary_cache.set(row, result["summary_md"]) return result payload = build_prompt("openai", context) try: text = self.client.chat(payload["messages"], max_tokens=1200, temperature=0.25) logging.info("✅ Generated summary for grant %s", grant_id) except Exception as e: text = f"LLM error: {e}\n\n{context[:800]}" logging.error("❌ Failed to summarize %s: %s", grant_id, e) # NEW: Cache the summary self.summary_cache.set(row, text) return {"summary_md": text, "title": title, "id": grant_id} # ----------------------------------------------------------------- # Helper method for basic context (without supporting materials) # ----------------------------------------------------------------- def _build_basic_context(self, row: Dict[str, Any]) -> str: """Build basic grant context without supporting materials.""" title = row.get("title", "(untitled)") url = row.get("url") or row.get("source_url") or "" parts = [ f"TITLE: {title}", f"ID: {row.get('id') or row.get('competition_id')}", f"URL: {url}", f"DEADLINE: {_fmt_date(row.get('deadline') or row.get('close_date'))}", ] for k in ( "summary", "overview", "scope", "eligibility", "funding", "dates", "how_to_apply", "supporting_information", ): v = row.get(k) if v: parts.append(f"{k.upper()}:\n{_norm(v)}") return "\n".join(parts) # ----------------------------------------------------------------- # Compare two grants (deterministic) # ----------------------------------------------------------------- def compare_grants(self, grant_id_a: str, grant_id_b: str) -> dict: """ Deterministic comparison: - Loads both grant records and any supporting index data - Builds a factual side-by-side table from structured fields - Optionally adds a short insight section from the LLM """ # ---------------- enrich ---------------- def enrich(gid: str) -> dict: base = self.get_grant(gid) row = dict(base) if self.support_idx: hits = search_by_grant_id( self.support_idx, gid.replace("competition-", "").replace("grant-", ""), k=5 ) # Adapt to new format: [(doc_dict, score), ...] if hits: doc, score = hits[0] meta = doc.get("meta", {}) for k, v in meta.items(): if v and k not in row: row[k] = v row["_support_text"] = doc.get("text", "") return row A = enrich(grant_id_a) B = enrich(grant_id_b) def getf(d: dict, key: str) -> str: v = d.get(key) or d.get(key.replace("_", " ")) or "" return str(v).strip() if v not in (None, "", "n/a") else "—" def get_funding(d: dict, field: str) -> str: """Extract funding amount from nested structure.""" funding = d.get("funding", {}) if isinstance(funding, dict): val = funding.get(field) if val is not None: try: return f"{float(val):,.0f}" except (ValueError, TypeError): pass # Fallback to flat field return getf(d, f"funding_{field}") # ---------------- table fields ---------------- fields = [ ("Title", getf(A, "title"), getf(B, "title")), ("Open date", getf(A, "open_date"), getf(B, "open_date")), ("Close date", getf(A, "close_date"), getf(B, "close_date")), ( "Funding per project", f"£{get_funding(A, 'min')}–£{get_funding(A, 'max')}", f"£{get_funding(B, 'min')}–£{get_funding(B, 'max')}", ), ("Total pot", f"£{get_funding(A, 'total_pot')}", f"£{get_funding(B, 'total_pot')}"), ( "Duration (months)", f"{getf(A, 'duration_min')}–{getf(A, 'duration_max')}", f"{getf(B, 'duration_min')}–{getf(B, 'duration_max')}", ), ] # Build text-based comparison instead of table comparison_lines = ["### Grant A: " + getf(A, "title")] for name, va, vb in fields: comparison_lines.append(f"**{name}:** {va}") comparison_lines.append("\n### Grant B: " + getf(B, "title")) for name, va, vb in fields: comparison_lines.append(f"**{name}:** {vb}") # ---------------- optional insight ---------------- context_text = "" if "_support_text" in A: context_text += "\n\n[Grant A Supporting Text]\n" + A["_support_text"][:1500] if "_support_text" in B: context_text += "\n\n[Grant B Supporting Text]\n" + B["_support_text"][:1500] insight = "" if self.client and self.client.is_ready() and context_text.strip(): try: prompt = ( "Given the factual comparison and context below, write 3-5 bullet points " "highlighting *meaningful differences* that matter to SMEs (funding size, " "duration, eligibility, etc.). Do not restate identical facts.\n\n" + "\n".join(comparison_lines) + "\n\n" + context_text ) insight = self.client.summarize(prompt) except Exception as e: logging.warning("compare_grants insight failed: %s", e) md = [ "### Comparison", "\n".join(comparison_lines), ] if insight: md += ["\n### Key differences", insight] return {"comparison_md": "\n".join(md)} # ----------------------------------------------------------------- # Deadlines overview # ----------------------------------------------------------------- def deadlines_overview(self, n: Optional[int] = None) -> List[Dict[str, Any]]: """ Get upcoming grant deadlines sorted by date. If n is None, returns all deadlines. Otherwise returns top n. """ rows = [] for r in self.current: d = _parse_date(r.get("deadline") or r.get("close_date")) if not d: continue status = self._calculate_grant_status(r) rows.append( { "id": r.get("id") or r.get("competition_id"), "title": r.get("title") or "(untitled)", "deadline": d.strftime("%Y-%m-%d %H:%M:%S"), "status": status, # NEW: Include status } ) rows.sort(key=lambda x: _parse_date(x["deadline"]) or datetime.max) # Default to 5 if not specified (for backward compat with UI) if n is None: n = 5 return rows[:n] # ----------------------------------------------------------------- # Analyze company for grant matching # ----------------------------------------------------------------- def analyze_company_for_grants(self, company_url: str, limit: int = 3) -> Dict[str, Any]: """ Fetch a company website and analyze which grants would be most suitable. Args: company_url: URL of the company website to analyze limit: Number of grant recommendations to return Returns: Dict with company analysis and recommended grants """ import urllib.request from html.parser import HTMLParser # Improved HTML to text parser that handles scripts/styles class HTMLTextExtractor(HTMLParser): def __init__(self): super().__init__() self.text = [] self.skip_tags = set() def handle_starttag(self, tag, attrs): # Skip script, style, noscript tags if tag in ('script', 'style', 'noscript', 'svg'): self.skip_tags.add(tag) def handle_endtag(self, tag): self.skip_tags.discard(tag) def handle_data(self, data): # Only add text if not in skip tags if not self.skip_tags: stripped = data.strip() if stripped and len(stripped) > 3: # Filter out single chars self.text.append(stripped) def get_text(self): return ' '.join(self.text) try: # Fetch the website logging.info(f"Fetching company website: {company_url}") # Add scheme if missing if not company_url.startswith(('http://', 'https://')): company_url = 'https://' + company_url # Set a realistic user agent to avoid blocks headers = { 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' } req = urllib.request.Request(company_url, headers=headers) # Fetch with timeout with urllib.request.urlopen(req, timeout=15) as response: html = response.read().decode('utf-8', errors='ignore') # Extract text from HTML parser = HTMLTextExtractor() parser.feed(html) company_text = parser.get_text() # Clean and truncate (keep first 4000 chars for analysis) company_text = ' '.join(company_text.split())[:4000] logging.info(f"Extracted {len(company_text)} chars from {company_url}") # Debug: log first 200 chars logging.debug(f"First 200 chars: {company_text[:200]}") except Exception as e: logging.error(f"Failed to fetch company website: {e}") return { "error": f"Could not fetch website: {str(e)}", "company_url": company_url, "recommendations": [] } # Use LLM to analyze company and match with grants if not self.client or not self.client.is_ready(): return { "error": "LLM not available for analysis", "company_url": company_url, "recommendations": [] } try: # Get list of available grants available_grants = [] for r in self.current[:20]: # Limit to 20 grants for context available_grants.append({ "id": r.get("id") or r.get("competition_id"), "title": r.get("title", "(untitled)"), "summary": r.get("summary", "")[:200], "scope": r.get("scope", "")[:200], "funding_max": r.get("funding_max") or r.get("max_award"), "deadline": r.get("deadline") or r.get("close_date") }) # Build analysis prompt grants_context = "\n".join([ f"- {g['id']}: {g['title']} (max funding: £{g['funding_max']}, deadline: {g['deadline']})" for g in available_grants ]) prompt = f"""Analyze this company website and recommend the most suitable grants. COMPANY WEBSITE TEXT: {company_text} AVAILABLE GRANTS: {grants_context} Based on the company's activities, industry, and apparent needs, which grants would be most suitable? Provide your answer in this format: COMPANY ANALYSIS: [Brief 2-3 sentence analysis of what the company does] RECOMMENDED GRANTS: 1. [Grant ID]: [Grant Title] - Why: [1-2 sentence explanation of fit] 2. [Grant ID]: [Grant Title] - Why: [1-2 sentence explanation of fit] 3. [Grant ID]: [Grant Title] - Why: [1-2 sentence explanation of fit] """ # Get LLM analysis analysis_text = self.client.summarize(prompt) # Extract recommended grant IDs from the response recommended_ids = [] for line in analysis_text.split('\n'): # Look for patterns like "1. competition-2313:" or "- 2313:" match = re.search(r'(?:competition-)?(\d{4})', line) if match: gid = match.group(1) if gid not in recommended_ids: recommended_ids.append(gid) if len(recommended_ids) >= limit: break # Get full details for recommended grants recommendations = [] for gid in recommended_ids: try: grant = self.get_grant(gid) recommendations.append({ "id": grant.get("id") or grant.get("competition_id"), "title": grant.get("title"), "deadline": grant.get("deadline") or grant.get("close_date"), "funding_max": grant.get("funding_max") or grant.get("max_award"), }) except KeyError: continue return { "company_url": company_url, "analysis": analysis_text, "recommendations": recommendations } except Exception as e: logging.error(f"Failed to analyze company: {e}") return { "error": f"Analysis failed: {str(e)}", "company_url": company_url, "recommendations": [] }