Spaces:
Sleeping
Sleeping
| """ | |
| context_builder.py — prepares text context for the LLM summarizer | |
| Takes structured JSON grant records + optional past winners and builds a single | |
| text block for each grant, suitable as LLM input. | |
| Responsibilities: | |
| - Extract key text fields from each grant JSON (title, description, sections) | |
| - Summarize/flatten them into a readable context string | |
| - Optionally include supporting documents (HTML sections + PDF text extracts) | |
| - Optionally include a few relevant past-winner snippets (if any exist) | |
| Public API | |
| ---------- | |
| build_context(grant: dict, past_winners: list[dict] | None = None, include_supporting: bool = False) -> str | |
| build_context_with_supporting(grant: dict, k: int = 5, past_winners: list[dict] | None = None) -> str | |
| """ | |
| from __future__ import annotations | |
| from typing import Any, Dict, List, Optional, Tuple | |
| import re | |
| import json | |
| from pathlib import Path | |
| # ----------------------------- Text utilities --------------------------------- | |
| def _clean(s: Any) -> str: | |
| return re.sub(r"\s+", " ", str(s or "")).strip() | |
| def _maybe(k: str, v: Any) -> str: | |
| if not v: | |
| return "" | |
| return f"{k}: {_clean(v)}\n" | |
| # ----------------------------- Context builder -------------------------------- | |
| def build_context(grant: Dict[str, Any], past_winners: Optional[List[Dict[str, Any]]] = None) -> str: | |
| """ | |
| Flatten a grant JSON object + optional past winners into a readable context string. | |
| CRITICAL: All URLs are provided in FULL form (https://...), never partial paths. | |
| The model receives complete, actionable URLs directly from the snapshot JSON. | |
| """ | |
| lines: List[str] = [] | |
| # --- Basic info --- | |
| title = grant.get("title") or grant.get("name") or grant.get("competition_title") | |
| if title: | |
| lines.append(f"TITLE: {_clean(title)}") | |
| # --- FULL URL: Always provide complete https:// URLs, never relative paths --- | |
| url = grant.get("url") or grant.get("link") | |
| if url: | |
| lines.append(f"URL: {url}") # Example: "https://apply-for-innovation-funding.service.gov.uk/competition/2276/overview/..." | |
| # --- Funding information --- | |
| funding = grant.get("funding") or {} | |
| if isinstance(funding, dict): | |
| max_funding = funding.get("max") | |
| if max_funding: | |
| lines.append(f"FUNDING: Up to £{max_funding:,}") | |
| min_funding = funding.get("min") | |
| if min_funding and min_funding != max_funding: | |
| lines.append(f"MINIMUM FUNDING: £{min_funding:,}") | |
| else: | |
| funding_text = grant.get("funding_amount") or grant.get("amount") or grant.get("funding_rate") | |
| if funding_text: | |
| lines.append(f"FUNDING: {_clean(funding_text)}") | |
| # --- Dates --- | |
| deadline = grant.get("close_date") or grant.get("deadline") | |
| if deadline: | |
| lines.append(f"DEADLINE: {_clean(deadline)}") | |
| open_date = grant.get("open_date") | |
| if open_date: | |
| lines.append(f"OPENS: {_clean(open_date)}") | |
| # --- Core text sections from snapshot --- | |
| sections = grant.get("sections") or {} | |
| if sections: | |
| # Process snapshot sections in order: summary, eligibility, scope, dates, how_to_apply | |
| section_order = [ | |
| "summary_raw", "eligibility_raw", "scope_raw", | |
| "dates_raw", "how_to_apply_raw", "supporting_information_raw" | |
| ] | |
| for section_key in section_order: | |
| v = sections.get(section_key) | |
| if v and _clean(v): # Only include non-empty sections | |
| section_name = section_key.replace("_raw", "").replace("_", " ").upper() | |
| lines.append(f"\n{section_name}:\n{_clean(v)}") | |
| else: | |
| # Fallback to common text fields | |
| desc = grant.get("description") or grant.get("summary") or grant.get("scope") | |
| if desc: | |
| lines.append(f"\nDESCRIPTION:\n{_clean(desc)}") | |
| # --- Optional extras --- | |
| eligibility = grant.get("eligibility") | |
| if eligibility: | |
| lines.append(f"\nELIGIBILITY:\n{_clean(eligibility)}") | |
| scope = grant.get("scope") | |
| if scope: | |
| lines.append(f"\nSCOPE:\n{_clean(scope)}") | |
| # --- Contact Information --- | |
| lines.append("\n--- CONTACT INFORMATION ---") | |
| lines.append("Email: support@iuk.ukri.org") | |
| lines.append("Phone: 0300 321 4357") | |
| lines.append("Hours: 9am-12pm, 2pm-5pm, Monday-Friday (excluding bank holidays)") | |
| # --- Past winners summary --- | |
| if past_winners: | |
| lines.append("\n--- RELATED PAST WINNERS ---") | |
| for w in past_winners[:5]: # limit to top 5 to avoid overloading tokens | |
| snippet_parts: List[str] = [] | |
| snippet_parts.append(_maybe("Project", w.get("project_title"))) | |
| snippet_parts.append(_maybe("Organisation", w.get("lead_org"))) | |
| snippet_parts.append(_maybe("Award", w.get("award_amount"))) | |
| snippet_parts.append(_maybe("Competition", w.get("competition"))) | |
| abs_ = _clean(w.get("abstract")) | |
| if abs_: | |
| snippet_parts.append(f"Abstract: {abs_[:400]}{'…' if len(abs_)>400 else ''}\n") | |
| lines.append("".join(snippet_parts)) | |
| # --- Return --- | |
| context_text = "\n".join(lines).strip() | |
| return context_text | |
| # ----------------------------- Supporting documents loader -------------------------------- | |
| def _load_supporting_docs_jsonl(path: Optional[str] = None) -> Dict[str, List[Dict[str, str]]]: | |
| """ | |
| Load the supporting documents JSONL file and index by grant_id. | |
| Returns: dict[grant_id] -> list of supporting docs | |
| """ | |
| if path is None: | |
| path = "data/supporting_jsonl/docs.jsonl" | |
| try: | |
| p = Path(path) | |
| if not p.exists(): | |
| return {} | |
| indexed = {} | |
| with open(p, "r", encoding="utf-8") as f: | |
| for line in f: | |
| if not line.strip(): | |
| continue | |
| try: | |
| doc = json.loads(line) | |
| gid = doc.get("grant_id", "").replace("competition-", "") | |
| if gid: | |
| if gid not in indexed: | |
| indexed[gid] = [] | |
| indexed[gid].append(doc) | |
| except json.JSONDecodeError: | |
| continue | |
| return indexed | |
| except Exception: | |
| return {} | |
| def get_supporting_docs_for_grant(grant_id: str, k: int = 5, doc_types: Optional[List[str]] = None) -> List[Dict[str, str]]: | |
| """ | |
| Retrieve supporting documents for a grant. | |
| Args: | |
| grant_id: Grant ID (with or without "competition-" prefix) | |
| k: Number of documents to return | |
| doc_types: Filter by document type ("supporting_html", "supporting_pdf", etc.) | |
| Returns: List of documents with extracted text | |
| """ | |
| # Load cache on first use (could be cached module-level) | |
| cache = _load_supporting_docs_jsonl() | |
| # Normalize grant ID | |
| gid = str(grant_id).replace("competition-", "").strip() | |
| docs = cache.get(gid, []) | |
| # Filter by type if requested | |
| if doc_types: | |
| docs = [d for d in docs if d.get("doc_type") in doc_types] | |
| # Return top k | |
| return docs[:k] | |
| def build_context_with_supporting( | |
| grant: Dict[str, Any], | |
| k: int = 5, | |
| pdf_only: bool = False, | |
| past_winners: Optional[List[Dict[str, Any]]] = None | |
| ) -> str: | |
| """ | |
| Build context including supporting documents (PDFs + HTML sections). | |
| Args: | |
| grant: Grant dict with id/competition_id field (or will extract from URL) | |
| k: Number of supporting docs to include | |
| pdf_only: If True, only include PDF documents | |
| past_winners: Optional past winners for comparison | |
| Returns: Context string with supporting materials embedded | |
| """ | |
| # Start with base context | |
| lines = [build_context(grant, past_winners)] | |
| # Extract grant ID from multiple sources | |
| gid = grant.get("id") or grant.get("competition_id") | |
| # If not found, try extracting from URL | |
| if not gid: | |
| url = grant.get("url") or "" | |
| match = re.search(r"/competition/(\d+)", url) | |
| if match: | |
| gid = match.group(1) | |
| if not gid: | |
| return lines[0] | |
| # Filter document types | |
| doc_types = ["supporting_pdf"] if pdf_only else ["supporting_pdf", "supporting_html"] | |
| # Load supporting docs | |
| supporting = get_supporting_docs_for_grant(str(gid), k=k, doc_types=doc_types) | |
| if supporting: | |
| lines.append("\n" + "="*80) | |
| lines.append("SUPPORTING MATERIALS & PDF CONTENT:") | |
| lines.append("="*80) | |
| for i, doc in enumerate(supporting, 1): | |
| doc_type = doc.get("doc_type", "unknown") | |
| section = doc.get("section", "Supporting Info") | |
| text = doc.get("text", "") | |
| # Truncate long text but keep it substantial | |
| if len(text) > 2000: | |
| text = text[:2000] + "\n[... truncated ...]" | |
| lines.append(f"\n[{i}] {section.upper()} ({doc_type})") | |
| lines.append("-" * 60) | |
| lines.append(text) | |
| return "\n".join(lines) | |
| # Self-test | |
| if __name__ == "__main__": | |
| fake_grant = { | |
| "title": "AI Battery Research Program", | |
| "funding_amount": "up to £1M", | |
| "deadline": "2025-12-17", | |
| "sections": { | |
| "summary_raw": "Funding for early-stage AI-driven battery optimization.", | |
| "scope_raw": "Projects must demonstrate significant improvement in energy density.", | |
| }, | |
| } | |
| fake_winners = [ | |
| { | |
| "project_title": "BatteryX AI", | |
| "lead_org": "EnergyAI Ltd", | |
| "award_amount": "£500,000", | |
| "competition": "Battery Innovation 2023", | |
| "abstract": "Developed machine learning models for lithium-ion battery efficiency.", | |
| } | |
| ] | |
| print(build_context(fake_grant, fake_winners)) |