Catalyst / admin/Bid-Sentinel 14.8 GB / 57.8 GB 40.0 GB free
Help Sign in

admin / Bid-Sentinel

public

Bid Scrape and Tracking Application with AI Capability

Code Issues Pull requests Pipelines Packages Security Insights Wiki Settings
Bid-Sentinel / bid-sentinel-v2 / backend / app / scraper / find_a_tender.py 4504 B · main 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
"""Scraper for Find a Tender Service / FTS (https://www.find-tender.service.gov.uk).

Uses the official OCDS Release Packages API. Results are filtered locally
against the cyber-security keyword list.
"""
from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone

import httpx

from app.scraper.base import (
    TenderRecord,
    cpv_text,
    is_award_or_closed,
    parse_datetime,
    parse_value,
)
from app.scraper.keywords import match_keyword

logger = logging.getLogger("scraper.find_a_tender")

SOURCE = "Find a Tender"
API_URL = "https://www.find-tender.service.gov.uk/api/1.0/ocdsReleasePackages"
HEADERS = {"User-Agent": "CyberTenderTracker/1.0 (+https://example.com)", "Accept": "application/json"}

# How far back to pull notices on each run.
LOOKBACK_DAYS = 30
# Safety cap on pagination to avoid runaway loops.
MAX_PAGES = 10


async def fetch(keywords: list[str] | None = None) -> list[TenderRecord]:
    """Fetch and filter recent FTS notices. Never raises."""
    records: list[TenderRecord] = []
    updated_from = (datetime.now(timezone.utc) - timedelta(days=LOOKBACK_DAYS)).strftime(
        "%Y-%m-%dT%H:%M:%S"
    )
    params = {"updatedFrom": updated_from, "limit": "100"}

    async with httpx.AsyncClient(timeout=30.0, headers=HEADERS, follow_redirects=True) as client:
        url: str | None = API_URL
        page = 0
        while url and page < MAX_PAGES:
            page += 1
            try:
                resp = await client.get(url, params=params if url == API_URL else None)
                resp.raise_for_status()
                data = resp.json()
            except Exception as exc:  # noqa: BLE001
                logger.warning("Find a Tender fetch failed (page %d): %s", page, exc)
                break

            for package_release in data.get("releases", []):
                rec = _parse_release(package_release, keywords)
                if rec:
                    records.append(rec)

            url = (data.get("links") or {}).get("next")

    return records


def _parse_release(release: dict, keywords: list[str] | None) -> TenderRecord | None:
    try:
        tender = release.get("tender", {}) or {}
        title = tender.get("title") or release.get("title")
        description = tender.get("description")
        if not title:
            return None

        # Only keep opportunities currently open for tender — skip award/contract
        # notices and closed/cancelled tenders (closing date is checked later).
        if is_award_or_closed(release, tender):
            return None

        # Keyword relevance is always required; the CPV AND-filter is applied
        # later in the runner using the classification captured below.
        keyword = match_keyword(title, description, keywords=keywords)
        if keyword is None:
            return None

        ocid = release.get("ocid") or release.get("id")
        # FTS canonical notice URL is built from the ocid suffix.
        notice_ref = ocid.split("-")[-1] if ocid else None
        url = (
            f"https://www.find-tender.service.gov.uk/Notice/{notice_ref}"
            if notice_ref
            else release.get("uri", "")
        )
        if not url:
            return None

        value = (tender.get("value") or {}).get("amount")
        amount, value_text = parse_value(value)
        tender_period = tender.get("tenderPeriod") or {}
        contract_period = tender.get("contractPeriod") or {}

        duration = None
        if contract_period.get("startDate") and contract_period.get("endDate"):
            duration = f"{contract_period['startDate'][:10]} to {contract_period['endDate'][:10]}"

        return TenderRecord(
            url=url,
            title=title.strip(),
            source=SOURCE,
            reference=str(ocid) if ocid else None,
            description=(description or "").strip() or None,
            published_date=parse_datetime(release.get("date")),
            closing_date=parse_datetime(tender_period.get("endDate")),
            value_amount=amount,
            value_text=value_text,
            currency=(tender.get("value") or {}).get("currency", "GBP"),
            duration=duration,
            buyer=(release.get("buyer") or {}).get("name"),
            matched_keyword=keyword,
            cpv=cpv_text(tender),
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("Failed to parse Find a Tender release: %s", exc)
        return None