admin / Bid-Sentinel

"""Optional AI bolt-on: Claude Haiku analysis of opportunities.

Toggleable and degradable: when the toggle is off, the API key is missing, or a
call fails, callers transparently fall back to the deterministic engine in
``app.analysis``. Uses the official Anthropic SDK (low-cost model by default).
"""
from __future__ import annotations

import json
import logging
import re

from app import analysis
from app.config import settings

logger = logging.getLogger("ai")

# JSON shape we ask the model to return.
_SCHEMA_HINT = (
    '{"summary": string (<=2 sentences), '
    '"fit_score": integer 0-100, '
    '"fit_reason": string (one line), '
    '"certifications": string[], '
    '"security_clearances": string[], '
    '"technical_requirements": string[] (max 5)}'
)


def is_available() -> bool:
    """True if an API key is configured (AI can actually run)."""
    return bool(settings.anthropic_api_key.strip())


def _build_system(capability_terms: list[str]) -> str:
    profile = ", ".join(capability_terms) if capability_terms else "(no profile provided)"
    return (
        "You are an analyst for a UK cyber-security bid team. Analyse a public or "
        "private sector tender notice and respond with ONLY a JSON object, no prose, "
        "no markdown fences, matching exactly this shape:\n"
        f"{_SCHEMA_HINT}\n\n"
        "Guidance:\n"
        "- summary: a concise plain-English summary of what is being procured.\n"
        "- fit_score: how well this matches the supplier's capabilities below "
        "(0 = no fit, 100 = perfect fit); fit_reason: one short sentence explaining it.\n"
        "- certifications: required/expected certifications & accreditations using "
        "canonical UK names (e.g. 'Cyber Essentials Plus', 'ISO 27001', 'CREST').\n"
        "- security_clearances: e.g. 'SC Clearance', 'DV Clearance', 'BPSS'.\n"
        "- technical_requirements: the most important technical requirements, max 5, "
        "each a short phrase.\n"
        "Use empty arrays when nothing applies. Do not invent requirements.\n\n"
        f"Supplier capabilities: {profile}"
    )


def _parse_json(raw: str) -> dict | None:
    if not raw:
        return None
    match = re.search(r"\{.*\}", raw, re.S)
    if not match:
        return None
    try:
        return json.loads(match.group(0))
    except (ValueError, TypeError):
        return None


async def _ai_call(text: str, capability_terms: list[str]) -> dict | None:
    try:
        from anthropic import AsyncAnthropic

        client = AsyncAnthropic(api_key=settings.anthropic_api_key)
        resp = await client.messages.create(
            model=settings.ai_model,
            max_tokens=1024,
            system=_build_system(capability_terms),
            messages=[{"role": "user", "content": f"Tender notice:\n\n{(text or '')[:8000]}"}],
        )
        raw = "".join(getattr(b, "text", "") for b in resp.content if getattr(b, "type", "") == "text")
        data = _parse_json(raw)
        if not data:
            logger.warning("AI returned unparseable output; falling back")
            return None

        certs = []
        for c in [*(data.get("certifications") or []), *(data.get("security_clearances") or [])]:
            c = str(c).strip()
            if c and c not in certs:
                certs.append(c)

        score = data.get("fit_score")
        try:
            score = max(0, min(100, int(score)))
        except (TypeError, ValueError):
            score = None

        techs = [str(x).strip() for x in (data.get("technical_requirements") or []) if str(x).strip()][:5]

        return {
            "required_certs": ", ".join(certs)[:1024],
            "fit_score": score,
            "fit_matched": str(data.get("fit_reason") or "").strip()[:1024],
            "tech_requirements": "\n".join(techs),
            "ai_summary": str(data.get("summary") or "").strip()[:2000],
        }
    except Exception as exc:  # noqa: BLE001 - any failure -> deterministic fallback
        logger.warning("AI analysis failed (%s); falling back to deterministic", exc)
        return None


async def suggest_capabilities(text: str) -> list[str] | None:
    """Extract a company's service capabilities from a service-overview document.

    Returns a list of short capability phrases, or None on failure (caller falls
    back to deterministic keyphrase extraction).
    """
    if not is_available():
        return None
    try:
        from anthropic import AsyncAnthropic

        client = AsyncAnthropic(api_key=settings.anthropic_api_key)
        system = (
            "You extract a company's cyber-security service capabilities from a "
            "service-overview document. Respond with ONLY a JSON object, no prose, no "
            'markdown fences: {"capabilities": string[]}. Each item is a short noun '
            "phrase naming a service the company offers (e.g. 'Managed Detection and "
            "Response', 'Penetration Testing', 'Cloud Security', 'SOC Monitoring'). "
            "Return 5-25 de-duplicated items. Do not invent services not in the text."
        )
        resp = await client.messages.create(
            model=settings.ai_model,
            max_tokens=800,
            system=system,
            messages=[{"role": "user", "content": f"Document:\n\n{(text or '')[:12000]}"}],
        )
        raw = "".join(getattr(b, "text", "") for b in resp.content if getattr(b, "type", "") == "text")
        data = _parse_json(raw)
        if not data:
            return None
        seen, out = set(), []
        for c in data.get("capabilities") or []:
            c = str(c).strip()
            if c and c.lower() not in seen:
                seen.add(c.lower())
                out.append(c)
        return out[:30] or None
    except Exception as exc:  # noqa: BLE001
        logger.warning("AI capability extraction failed (%s); falling back", exc)
        return None


async def sales_angle(context_text: str) -> str | None:
    """Write a concise sales angle for a client organisation. None on failure."""
    if not is_available():
        return None
    try:
        from anthropic import AsyncAnthropic

        client = AsyncAnthropic(api_key=settings.anthropic_api_key)
        system = (
            "You are a UK cyber-security bid strategist. Given data about a client "
            "organisation's procurement activity and the supplier's capabilities, "
            "write a punchy 2-3 sentence sales angle: why the supplier is a strong "
            "fit, the specific gap or need to lead with, and any missing "
            "certification to mitigate (partner/roadmap). Plain text only, no "
            "preamble, no bullet points, no markdown."
        )
        resp = await client.messages.create(
            model=settings.ai_model,
            max_tokens=300,
            system=system,
            messages=[{"role": "user", "content": context_text}],
        )
        text = "".join(getattr(b, "text", "") for b in resp.content if getattr(b, "type", "") == "text").strip()
        return text or None
    except Exception as exc:  # noqa: BLE001
        logger.warning("AI sales angle failed (%s); falling back", exc)
        return None


async def analyze_opportunity(text: str, capability_terms: list[str], use_ai: bool) -> dict:
    """Return the analysis fields for one opportunity (AI if enabled, else rules)."""
    if use_ai and is_available():
        result = await _ai_call(text, capability_terms)
        if result is not None:
            return result
    fields = analysis.analyze(text, capability_terms)
    fields["ai_summary"] = ""
    return fields