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 / ocds_api.py 4558 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
"""Generic OCDS (Open Contracting) JSON API scraper.

UK procurement data is widely published as OCDS release packages. Where a portal
exposes such a JSON feed, prefer it over HTML scraping — it's structured and
reliable. This helper builds an ``async fetch(keywords)`` for any OCDS endpoint;
point a portal at it in ``portals.py`` by giving its ``api_url``.

(Find a Tender and Contracts Finder keep their own bespoke OCDS modules because
their URL/notice-link conventions differ; this is the reusable path for others.)
"""
from __future__ import annotations

import logging

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.ocds")

HEADERS = {
    "User-Agent": "CyberTenderTracker/1.0 (+https://example.com)",
    "Accept": "application/json",
}


def make_scraper(source_name: str, api_url: str, params: dict | None = None, max_pages: int = 5):
    """Build an ``async fetch(keywords)`` for an OCDS release-package endpoint."""

    async def fetch(keywords: list[str] | None = None) -> list[TenderRecord]:
        records: list[TenderRecord] = []
        try:
            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
                    resp = await client.get(url, params=params if url == api_url else None)
                    resp.raise_for_status()
                    data = resp.json()
                    for release in data.get("releases", []):
                        rec = _parse_release(release, source_name, keywords)
                        if rec:
                            records.append(rec)
                    url = (data.get("links") or {}).get("next")
        except Exception as exc:  # noqa: BLE001 - never abort the whole run
            logger.warning("%s: OCDS fetch failed: %s", source_name, exc)
        logger.info("%s: OCDS scraper found %d candidate(s)", source_name, len(records))
        return records

    return fetch


def _parse_release(release: dict, source_name: str, 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.
        matched = match_keyword(title, description, keywords=keywords)
        if matched is None:
            return None

        ocid = release.get("ocid") or release.get("id")
        url = release.get("uri") or (tender.get("documents") or [{}])[0].get("url", "")
        if not url and ocid:
            url = f"ocds:{ocid}"
        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()[:1024],
            source=source_name,
            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=matched,
            cpv=cpv_text(tender),
        )
    except Exception as exc:  # noqa: BLE001
        logger.debug("%s: could not parse release: %s", source_name, exc)
        return None