import json import re import pathlib import asyncio from urllib.parse import urlparse from datetime import datetime, UTC from typing import List, Optional, Tuple, Dict from playwright.async_api import async_playwright from bs4 import BeautifulSoup, Tag, NavigableString import typer # --------------------------------------------------------------------------- # HELPERS # --------------------------------------------------------------------------- EXPECTED_TITLES = [ "Summary", "Eligibility", "Scope", "Dates", "How to apply", "Supporting information", ] # Accept close-enough labels and alias them to canonical 6 SECTION_ALIASES = { "who can apply": "Eligibility", "who’s eligible": "Eligibility", "who is eligible": "Eligibility", "applicant eligibility": "Eligibility", "what we ask you": "How to apply", "apply": "How to apply", "application process": "How to apply", "supporting info": "Supporting information", "key dates": "Dates", "timeline": "Dates", "competition dates": "Dates", "overview": "Summary", "summary": "Summary", "scope": "Scope", } def canonical_label(label: str) -> Optional[str]: l = label.strip().lower() for t in EXPECTED_TITLES: if l == t.lower(): return t return SECTION_ALIASES.get(l, None) def norm_key(label: str) -> str: return re.sub(r"[^\w\s-]", "", label).strip().lower().replace(" ", "_") + "_raw" def parse_deeplink(url: str): u = urlparse(url) m = re.search(r"/competition/(\d+)/overview/([0-9a-f-]{8,})", u.path, re.I) if not m: raise ValueError("Expected: .../competition/{id}/overview/{uuid}") return f"{u.scheme}://{u.netloc}", m.group(1), m.group(2) def get_competition_nav_anchors(soup: BeautifulSoup) -> List[tuple[str, str]]: anchors: List[tuple[str, str]] = [] headings = soup.find_all(["h2", "h3", "h4"], string=lambda s: isinstance(s, str) and "competition sections" in s.lower()) nav_root: Optional[Tag] = None for h in headings: for sib in h.next_siblings: if isinstance(sib, Tag) and sib.name in ("nav", "ul", "ol", "div"): nav_root = sib break if nav_root: break if not nav_root: for candidate in soup.find_all("nav"): if candidate.find("a", href=True): nav_root = candidate break if nav_root: seen = set() for a in nav_root.find_all("a", href=True): href = a.get("href", "") if not href.startswith("#"): continue frag = href[1:].strip() raw = (a.get_text(" ", strip=True) or "").strip() if not frag or not raw: continue canon = canonical_label(raw) or raw if canon in EXPECTED_TITLES and frag not in seen: anchors.append((canon, frag)) seen.add(frag) if not anchors: anchors = [ ("Summary", "summary"), ("Eligibility", "eligibility"), ("Scope", "scope"), ("Dates", "dates"), ("How to apply", "how-to-apply"), ("Supporting information", "supporting-information"), ] return anchors # ----------------------------- # Footer / cookie / consent trimmer # ----------------------------- _FOOTER_STOPS = [ "Need help with this service?", "Support links", "GOV.UK uses cookies", "Create one update function for each consent parameter", "© Crown copyright", "All content is available under the Open Government Licence", ] def trim_footer(text: str) -> str: if not text: return text for marker in _FOOTER_STOPS: i = text.find(marker) if i != -1: return text[:i].rstrip() return text def _strip_boilerplate(soup: BeautifulSoup): selectors = [ "#global-cookie-message", ".cookie-banner", "#ccc-notify", "#onetrust-banner-sdk", "footer", ".govuk-footer", ".govuk-prototype-kit-warning", ] for sel in selectors: for el in soup.select(sel): el.decompose() # --------------------------------------------------------------------------- # TITLE # --------------------------------------------------------------------------- def clean_title(raw: str) -> str: if not raw: return raw raw = raw.strip() return re.sub(r"^\s*Funding competition\s+", "", raw, flags=re.I).strip() def extract_between_ids(soup: BeautifulSoup, start_id: str, end_id: Optional[str]) -> str: start = soup.find(id=start_id) if not start: return "" out_chunks: List[str] = [] for el in start.next_elements: if isinstance(el, Tag): if end_id and el.get("id") == end_id: break if el.name in ("script", "style", "noscript"): continue if isinstance(el, NavigableString): txt = el.strip() if txt: out_chunks.append(txt) text = " ".join(out_chunks) text = re.sub(r"\s+", " ", text).strip() text = trim_footer(text) return text # --------------------------------------------------------------------------- # PARSING # --------------------------------------------------------------------------- def slice_by_anchors(html: str) -> dict: soup = BeautifulSoup(html, "html.parser") _strip_boilerplate(soup) anchors = get_competition_nav_anchors(soup) ids_in_order = [aid for _, aid in anchors] id_to_next = {ids_in_order[i]: (ids_in_order[i + 1] if i + 1 < len(ids_in_order) else None) for i in range(len(ids_in_order))} out = {} for label, start_id in anchors: next_id = id_to_next.get(start_id) key = norm_key(label) out[key] = extract_between_ids(soup, start_id, next_id) return out # --------------------------------------------------------------------------- # MAIN SCRAPER # --------------------------------------------------------------------------- async def fetch_sections_from_overview(url: str) -> tuple[str, dict]: scheme_host, comp_id, uuid = parse_deeplink(url) overview_url = f"{scheme_host}/competition/{comp_id}/overview/{uuid}" async with async_playwright() as pw: browser = await pw.chromium.launch(headless=True, args=["--disable-dev-shm-usage"]) context = await browser.new_context( user_agent=("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"), locale="en-GB", timezone_id="Europe/London", ) page = await context.new_page() page.set_default_timeout(30000) for attempt in range(2): try: await page.goto(overview_url, wait_until="domcontentloaded") await page.wait_for_load_state("networkidle") break except Exception: if attempt == 1: raise html = await page.content() await context.close() await browser.close() return html, slice_by_anchors(html) # --------------------------------------------------------------------------- # DATE PARSING — SINGLE-LINE CHUNKS # --------------------------------------------------------------------------- # tokens _DATE_WORD = r"(?:\d{1,2}\s+[A-Za-z]{3,9}\s+\d{4}|[A-Za-z]{3,9}\s+\d{1,2},?\s+\d{4})" _TIME_WORD = r"(?:\d{1,2}:\d{2}\s*[ap]m|\d{1,2}\s*[ap]m)" _DATE_ONLY_RX = re.compile(_DATE_WORD, re.I) _TIME_RX = re.compile(_TIME_WORD, re.I) # e.g. "9 to 20 March 2026", "9–20 March 2026", "9 - 20 March 2026" _DATE_RANGE_RX = re.compile(r"(\d{1,2})\s*(?:to|-|–)\s*(\d{1,2})\s+([A-Za-z]{3,9})\s+(\d{4})", re.I) _MONTHS = {m.lower(): i for i, m in enumerate( ["January","February","March","April","May","June","July","August","September","October","November","December"], 1 )} _EXCLUDE_SENTENCE_CUES = ["briefing event", "briefing", "webinar", "register to attend", "register", "info session"] _LABEL_RULES = [ ("opens", ["competition opens", "opens"]), ("closes", ["competition closes", "closes", "deadline"]), ("notify", ["applicants notified", "applicants will be notified", "notification"]), ("project_start", ["project start from", "project starts from", "project start date", "project start"]), ("assessment", ["interview", "assessment", "panel"]), ("results", ["results published", "winners announced"]), ("eligibility_cutoff", ["eligibility closes", "registration closes"]), ("info_session", ["briefing", "webinar", "register"]), ] def _classify_label(sent_lower: str) -> str: for norm, cues in _LABEL_RULES: if any(c in sent_lower for c in cues): return norm return "other" def _parse_single_date(token: str, time_hint: Optional[str]) -> Optional[str]: token = token.strip() m_comma = re.match(r"([A-Za-z]{3,9})\s+(\d{1,2}),?\s+(\d{4})", token) if m_comma: month_name, day, year = m_comma.groups() else: m = re.match(r"(\d{1,2})\s+([A-Za-z]{3,9})\s+(\d{4})", token) if not m: return None day, month_name, year = m.groups() month = _MONTHS.get(month_name.lower()) if not month: return None if time_hint: t = time_hint.lower().replace(" ", "") mm = re.match(r"(\d{1,2})(?::(\d{2}))?([ap]m)", t) if mm: hh = int(mm.group(1)) mins = int(mm.group(2) or 0) ampm = mm.group(3) if ampm == "pm" and hh != 12: hh += 12 if ampm == "am" and hh == 12: hh = 0 return f"{int(year):04d}-{month:02d}-{int(day):02d}T{hh:02d}:{mins:02d}:00" return f"{int(year):04d}-{month:02d}-{int(day):02d}" def parse_dates_singleline(text: str) -> List[dict]: """ Split the Dates section into *one milestone per line/chunk*: chunk := from each DATE token up to the next DATE token (or end). Keeps the entire chunk in label_raw. """ milestones: List[dict] = [] if not text: return milestones # find all date token positions matches = list(_DATE_ONLY_RX.finditer(text)) if not matches: return milestones spans = [] for i, m in enumerate(matches): start = m.start() end = matches[i + 1].start() if i + 1 < len(matches) else len(text) spans.append((start, end)) for (start, end) in spans: chunk = text[start:end].strip() if not chunk: continue low = chunk.lower() excluded = any(k in low for k in _EXCLUDE_SENTENCE_CUES) # primary date + optional time in this chunk first_date = _DATE_ONLY_RX.search(chunk) time_hint_match = _TIME_RX.search(chunk) iso = _parse_single_date(first_date.group(0), time_hint_match.group(0) if time_hint_match else None) if first_date else None # optional same-month day range inside the chunk r = _DATE_RANGE_RX.search(chunk) date_end_iso = None if r: d1, d2, mon_name, year = r.groups() month = _MONTHS.get(mon_name.lower()) if month: date_end_iso = f"{int(year):04d}-{month:02d}-{int(d2):02d}" # if the start of the range equals first_date, keep iso as start; # otherwise we still keep iso from first_date (which begins the chunk) if iso: milestones.append({ "label_raw": chunk, "label_norm": _classify_label(low), "date_iso": iso, "date_iso_end": date_end_iso, "has_time": bool(time_hint_match), "excluded_from_open_close": excluded, }) return milestones def pick_open_close_from_milestones(milestones: List[dict]) -> Tuple[Optional[str], Optional[str]]: open_iso = close_iso = None for m in milestones: if m["excluded_from_open_close"]: continue if m["label_norm"] == "opens" and open_iso is None: open_iso = m["date_iso"].split("T")[0] if m["label_norm"] == "closes" and close_iso is None: close_iso = m["date_iso"] if open_iso is None: for m in milestones: if m["label_norm"] == "opens": open_iso = m["date_iso"].split("T")[0] break if close_iso is None: for m in milestones: if m["label_norm"] == "closes": close_iso = m["date_iso"] break return open_iso, close_iso def derive_aux_dates(milestones: List[dict]) -> Tuple[Optional[str], Optional[str]]: notify = project_start_from = None for m in milestones: if notify is None and m["label_norm"] == "notify": notify = m["date_iso"].split("T")[0] if project_start_from is None and m["label_norm"] == "project_start": project_start_from = m["date_iso"].split("T")[0] if notify and project_start_from: break return notify, project_start_from # --------------------------------------------------------------------------- # FUNDING / COMPENSATION PARSING # --------------------------------------------------------------------------- _MONEY_TOKEN = re.compile(r"(£|\bGBP\s*)([\d,]+(?:\.\d+)?)(?:\s*(million|m|billion|bn|k))?", re.I) def _money_to_int(sign: str, num_str: str, mag: Optional[str]) -> int: val = float(num_str.replace(",", "")) if mag: m = mag.lower() if m in ("million", "m"): val *= 1_000_000 elif m in ("billion", "bn"): val *= 1_000_000_000 elif m in ("k",): val *= 1_000 return int(round(val)) _TOTAL_CUES = [ "total prize fund", "total prize pot", "total funding available", "available in total", "total pot", "prize fund", "funding pot", "overall budget", "total budget", "total allocation", "in total across", "total amount available", ] _AWARD_CUES = [ "per project", "each project", "you can apply for", "can apply for", "apply for up to", "grant of up to", "awards of up to", "awards between", "awards of between", "fund between", "we will fund", "we can fund", "project costs between", "total eligible project costs between", "your project must have total costs between", "maximum grant", "minimum grant", "maximum funding", "minimum funding", "up to", "no more than", "at least", "grant funding request", "eligible grant funding", "eligible grant", "funding request must be between", ] _EXCLUDE_CUES = [ "market", "industry", "global", "worldwide", "valuation", "addressable", "gdp", "economy", "sector value", "turnover", "revenue", "jobs", "headcount", ] _RANGE_PATTERNS = [ re.compile(rf"(?:between|from)\s+{_MONEY_TOKEN.pattern}\s+(?:and|to)\s+{_MONEY_TOKEN.pattern}", re.I), re.compile(rf"{_MONEY_TOKEN.pattern}\s*(?:to|-)\s*{_MONEY_TOKEN.pattern}", re.I), ] _MAX_PATTERNS = [ re.compile(rf"(?:up to|no more than|max(?:imum)?(?:\s+grant|\s+funding)?)\s+{_MONEY_TOKEN.pattern}", re.I), ] _MIN_PATTERNS = [ re.compile(rf"(?:at least|minimum(?:\s+grant|\s+funding)?)\s+{_MONEY_TOKEN.pattern}", re.I), ] def _contains_any(text: str, cues: List[str]) -> bool: low = text.lower() return any(c in low for c in cues) def _is_excluded_sentence(sent: str) -> bool: return _contains_any(sent, _EXCLUDE_CUES) def _split_sentences_generic(text: str) -> List[str]: parts = re.split(r"(?:\n+|(?<=[\.\!\?])\s+)", text) return [p.strip() for p in parts if p and p.strip()] def _find_total_pot(text: str) -> Optional[int]: if not text: return None best = None for sent in _split_sentences_generic(text): if _is_excluded_sentence(sent): continue if _contains_any(sent, _TOTAL_CUES): vals = [] for m in _MONEY_TOKEN.finditer(sent): _, num_str, mag = m.groups() vals.append(_money_to_int("£", num_str, mag)) if vals: v = max(vals) best = v if best is None or v > best else best return best def _find_award_range(text: str) -> Tuple[Optional[int], Optional[int]]: if not text: return None, None # Strong: explicit ranges in a sentence that has award cues for sent in _split_sentences_generic(text): if _is_excluded_sentence(sent): continue if not _contains_any(sent, _AWARD_CUES): continue for rx in _RANGE_PATTERNS: m = rx.search(sent) if not m: continue monies = list(_MONEY_TOKEN.finditer(m.group(0))) if len(monies) >= 2: v1 = _money_to_int(*("£", monies[-2].group(2), monies[-2].group(3))) v2 = _money_to_int(*("£", monies[-1].group(2), monies[-1].group(3))) lo, hi = sorted([v1, v2]) return lo, hi # Next: max-only / min-only with cues chosen_min = None chosen_max = None for sent in _split_sentences_generic(text): if _is_excluded_sentence(sent): continue if not _contains_any(sent, _AWARD_CUES): continue if chosen_max is None: for rx in _MAX_PATTERNS: m = rx.search(sent) if m: money = _MONEY_TOKEN.search(m.group(0)) if money: chosen_max = _money_to_int(*("£", money.group(2), money.group(3))) break if chosen_min is None: for rx in _MIN_PATTERNS: m = rx.search(sent) if m: money = _MONEY_TOKEN.search(m.group(0)) if money: chosen_min = _money_to_int(*("£", money.group(2), money.group(3))) break if chosen_min is not None and chosen_max is not None: break return chosen_min, chosen_max # NEW: funding rates & duration # Pattern 1: Separate rates for micro/small, medium, and large _RATE_LINE_FULL = re.compile( r"up to\s*(\d{1,3})%\s*if you are a\s*(?:micro|small).*?up to\s*(\d{1,3})%\s*if you are a\s*medium.*?up to\s*(\d{1,3})%\s*if you are a\s*large", re.I | re.S, ) # Pattern 2: Combined rate for micro/small/medium and separate for large _RATE_LINE_COMBINED = re.compile( r"up to\s*(\d{1,3})%\s*if you are a\s*(?:micro,?\s*small\s*or\s*medium|micro,?\s*small,?\s*or\s*medium).*?up to\s*(\d{1,3})%\s*if you are a\s*large", re.I | re.S, ) # Pattern 3: KTP style - simplified to handle various wordings _RATE_LINE_KTP = re.compile( r"large\s+compan.*?(\d{1,3})%.*?(?:SME|small|medium).*?(\d{1,3})%", re.I | re.S, ) def _find_funding_rates(text: str) -> Optional[dict]: if not text: return None # Try full pattern (separate rates for all three categories) m = _RATE_LINE_FULL.search(text) if m: small, medium, large = map(int, m.groups()) return {"micro_small": small, "medium": medium, "large": large} # Try combined pattern (micro/small/medium together, large separate) m = _RATE_LINE_COMBINED.search(text) if m: small_medium, large = map(int, m.groups()) return {"micro_small": small_medium, "medium": small_medium, "large": large} # Try KTP style pattern (large first, then SME - simplified) m = _RATE_LINE_KTP.search(text) if m: large_rate, sme_rate = map(int, m.groups()) return {"micro_small": sme_rate, "medium": sme_rate, "large": large_rate} return None _DURATION_RX_RANGE = re.compile(r"(?:last|must\s+be)\s+between\s+(\d{1,3})\s*(?:and|to|–|-)\s*(\d{1,3})\s+months", re.I) _DURATION_RX_UPTO = re.compile(r"last\s+up\s+to\s+(\d{1,3}|one|two|three|four|five|six|seven|eight|nine|ten|eleven|twelve)\s+months?", re.I) _DURATION_RX_MAX = re.compile(r"(?:maximum|max)\s+(?:of\s+)?(\d{1,3}|one|two|three|four|five|six|seven|eight|nine|ten|eleven|twelve)\s+months?", re.I) _WORD_TO_NUM = { "one": 1, "two": 2, "three": 3, "four": 4, "five": 5, "six": 6, "seven": 7, "eight": 8, "nine": 9, "ten": 10, "eleven": 11, "twelve": 12, "thirteen": 13, "fourteen": 14, "fifteen": 15, "sixteen": 16, "seventeen": 17, "eighteen": 18, "nineteen": 19, "twenty": 20, "twenty-four": 24, "thirty-six": 36 } def _parse_month_number(s: str) -> int: """Convert month string (digit or word) to integer.""" s = s.strip().lower() if s.isdigit(): return int(s) return _WORD_TO_NUM.get(s, 0) def _find_duration_months(text: str) -> Tuple[Optional[int], Optional[int]]: if not text: return None, None # Try "between X and Y months" m = _DURATION_RX_RANGE.search(text) if m: lo, hi = map(int, m.groups()) return (lo if lo <= hi else hi), (hi if hi >= lo else lo) # Try "up to X months" (returns None, X) m = _DURATION_RX_UPTO.search(text) if m: num = _parse_month_number(m.group(1)) if num > 0: return None, num # Try "maximum X months" (returns None, X) m = _DURATION_RX_MAX.search(text) if m: num = _parse_month_number(m.group(1)) if num > 0: return None, num return None, None _PER_MONTH_COST_RX = re.compile(r"(?:costs?\s+(?:are\s+)?typically|eligible\s+costs?)\s+(?:are\s+)?[£$]?\s*([\d,]+)\s*per\s+month", re.I) def _find_per_month_cost(text: str) -> Optional[int]: """Extract per-month cost pattern like '£8,500 per month'.""" if not text: return None m = _PER_MONTH_COST_RX.search(text) if m: return _money_to_int("£", m.group(1), None) return None def extract_funding(sections: dict, duration_min: Optional[int] = None, duration_max: Optional[int] = None) -> dict: summary = sections.get("summary_raw", "") or "" support = sections.get("supporting_information_raw", "") or "" scope = sections.get("scope_raw", "") or "" eligibility = sections.get("eligibility_raw", "") or "" all_text = " ".join(v for v in sections.values() if isinstance(v, str) and v) total_pot = _find_total_pot(summary) or _find_total_pot(support) or _find_total_pot(all_text) min_award = max_award = None for candidate in (summary, eligibility, support, scope, all_text): lo, hi = _find_award_range(candidate) if lo is not None or hi is not None: if lo is not None: min_award = lo if hi is not None: max_award = hi break # If no explicit min/max found, check for per-month costs and calculate from duration if min_award is None and max_award is None: per_month = _find_per_month_cost(summary) or _find_per_month_cost(eligibility) or _find_per_month_cost(all_text) if per_month and (duration_min or duration_max): if duration_min: min_award = per_month * duration_min if duration_max: max_award = per_month * duration_max rates = None for candidate in (eligibility, support, all_text): rates = _find_funding_rates(candidate or "") if rates: break return {"min": min_award, "max": max_award, "total_pot": total_pot, "rates": rates} # --------------------------------------------------------------------------- # MAIN # --------------------------------------------------------------------------- async def _main_async(url: str): m = re.search(r"/competition/(\d+)", url) if not m: raise ValueError(f"Could not extract competition ID from URL: {url}") grant_id = m.group(1) slug = f"competition-{grant_id}" out_path = pathlib.Path("data/snapshots") / f"{slug}.json" out_path.parent.mkdir(parents=True, exist_ok=True) html, sections = await fetch_sections_from_overview(url) soup = BeautifulSoup(html, "html.parser") h1 = soup.find("h1") raw_title = h1.get_text(" ", strip=True) if h1 else "" title = clean_title(raw_title) # Dates -> single-line chunks dates_text = sections.get("dates_raw", "") or "" milestones = parse_dates_singleline(dates_text) # Derive open/close from milestones (with exclusions for briefing lines) open_date, close_date = pick_open_close_from_milestones(milestones) # If either missing, try scanning all text but still as single-line chunks if not open_date or not close_date: all_text = " ".join(v for v in sections.values() if isinstance(v, str) and v) extra = parse_dates_singleline(all_text) # merge de-duped by (label_raw, date_iso, date_iso_end) seen = {(m["label_raw"], m["date_iso"], m.get("date_iso_end")) for m in milestones} for m2 in extra: key = (m2["label_raw"], m2["date_iso"], m2.get("date_iso_end")) if key not in seen: milestones.append(m2) seen.add(key) od2, cd2 = pick_open_close_from_milestones(milestones) open_date = open_date or od2 close_date = close_date or cd2 notify_date, project_start_from = derive_aux_dates(milestones) # Duration months (must be extracted before funding for per-month calculations) dur_min, dur_max = _find_duration_months(sections.get("eligibility_raw", "") or "") if dur_min is None and dur_max is None: all_text_for_duration = " ".join(v for v in sections.values() if isinstance(v, str) and v) dur_min, dur_max = _find_duration_months(all_text_for_duration) duration_months = {"min": dur_min, "max": dur_max} # Funding (pass duration for per-month cost calculations) funding = extract_funding(sections, dur_min, dur_max) snapshot = { "id": f"competition-{grant_id}", "competition_id": grant_id, "url": url, "title": title, "programme": "", "round": "", "open_date": open_date, "close_date": close_date, "notify_date": notify_date, "project_start_from": project_start_from, "funding": funding, "duration_months": duration_months, "sections": sections, "pdfs": [], "summaries": {}, "extracted": {"milestones": milestones}, "wonky": {"score": 0.0, "reasons": []}, "prev_round_refs": [], "diff_summary": "", "history_stats": {}, "created_at": datetime.now(UTC).isoformat(), "updated_at": datetime.now(UTC).isoformat(), } out_path.write_text(json.dumps(snapshot, indent=2), encoding="utf-8") print(f"Snapshot saved to {out_path}") def main(url: str = typer.Argument(..., help="IFS overview URL e.g. .../competition/{id}/overview/{uuid}")): asyncio.run(_main_async(url)) if __name__ == "__main__": typer.run(main)