admin / Bid-Sentinel

"""Shared scraper utilities: the normalized record and value/date parsers."""
from __future__ import annotations

import logging
import re
from dataclasses import dataclass, field
from datetime import datetime

logger = logging.getLogger("scraper")


@dataclass
class TenderRecord:
    """Source-agnostic representation of a scraped opportunity."""

    url: str
    title: str
    source: str
    reference: str | None = None
    description: str | None = None
    published_date: datetime | None = None
    closing_date: datetime | None = None
    value_amount: float | None = None
    value_text: str | None = None
    currency: str = "GBP"
    duration: str | None = None
    buyer: str | None = None
    matched_keyword: str | None = None
    # CPV code(s) + description(s) from the source's structured classification, if
    # any. Used by the scrape's CPV AND-filter (empty for HTML sources).
    cpv: str = ""
    extra: dict = field(default_factory=dict)


# OCDS tender.status values that mean the tender is no longer open.
_CLOSED_TENDER_STATUS = {"complete", "cancelled", "unsuccessful", "withdrawn"}
# OCDS release tags that mark award / contract notices (not open tenders).
_AWARD_TAGS = {
    "award", "awardupdate", "awardcancellation",
    "contract", "contractupdate", "contractamendment",
    "contracttermination", "contractclosure",
}


def is_award_or_closed(release: dict, tender: dict | None) -> bool:
    """True if an OCDS release is an award/contract notice or a closed/cancelled
    tender — i.e. NOT currently open to bid on.

    An explicit ``tender`` tag is trusted (kept; its closing date is checked
    separately). Otherwise award/contract tags, an award payload, or a
    closed/cancelled tender status mark it as not-open.
    """
    tender = tender or {}
    status = str(tender.get("status") or "").lower()
    if status in _CLOSED_TENDER_STATUS:
        return True
    tags = {str(t).lower() for t in (release.get("tag") or [])}
    if "tender" in tags:
        return False  # explicit tender notice — keep
    if tags & _AWARD_TAGS:
        return True
    if release.get("awards"):
        return True  # award data present, no tender tag -> award notice
    return False


def normalize_cpv(code: str | None) -> str:
    """Normalise a CPV code for lenient matching: digits only, check-digit dropped.

    CPV codes look like ``72500000-0`` (8 digits + a hyphen check digit). We keep
    just the 8 significant digits so a user's ``72500000`` matches a notice tagged
    ``72500000-0`` (and vice-versa).
    """
    if not code:
        return ""
    digits = re.sub(r"\D", "", str(code))
    # Drop the trailing check digit only when a full 9-digit form was given.
    if len(digits) == 9:
        digits = digits[:8]
    return digits


def cpv_text(tender: dict) -> str:
    """Collect CPV code + description strings from an OCDS ``tender`` object.

    Included in the text passed to ``match_keyword`` so user CPV criteria are
    matched against a release's structured classification, not just its prose.
    """
    classifications: list[dict] = []
    main = tender.get("classification")
    if isinstance(main, dict):
        classifications.append(main)
    extra = tender.get("additionalClassifications")
    if isinstance(extra, list):
        classifications.extend(c for c in extra if isinstance(c, dict))

    parts: list[str] = []
    for c in classifications:
        scheme = str(c.get("scheme", "")).upper()
        if scheme and "CPV" not in scheme:
            continue  # only CPV-scheme classifications
        cid = c.get("id")
        if cid:
            parts.append(str(cid))
            parts.append(normalize_cpv(cid))  # normalised form for lenient hits
        desc = c.get("description")
        if desc:
            parts.append(str(desc))
    return " ".join(parts)


def parse_datetime(value: str | None) -> datetime | None:
    """Best-effort ISO/date parsing. Returns None on failure (never raises)."""
    if not value:
        return None
    value = value.strip()
    candidates = (
        value,
        value.replace("Z", "+00:00"),
    )
    for cand in candidates:
        try:
            return datetime.fromisoformat(cand)
        except (ValueError, TypeError):
            continue
    for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%d %B %Y", "%d %b %Y"):
        try:
            return datetime.strptime(value, fmt)
        except ValueError:
            continue
    logger.debug("Could not parse datetime: %r", value)
    return None


_NUM_RE = re.compile(r"[-+]?\d[\d,]*\.?\d*")


def parse_value(value) -> tuple[float | None, str | None]:
    """Parse a financial value.

    Returns (amount, raw_text). amount is the first numeric found (commas
    stripped); raw_text preserves the original for ranges / "POA" / etc.
    """
    if value is None:
        return None, None
    if isinstance(value, (int, float)):
        return float(value), f"{value:,.2f}"

    text = str(value).strip()
    if not text:
        return None, None

    match = _NUM_RE.search(text.replace(",", ""))
    if match:
        try:
            return float(match.group()), text
        except ValueError:
            pass
    return None, text  # e.g. "Price on application"