Python e Robocorp

blupantheon-automation/
├── conda.yaml
├── robot.yaml
├── requirements.txt
├── src/
│   ├── __init__.py
│   ├── config.py
│   ├── client.py
│   ├── signature.py
│   ├── receiver.py
│   ├── tasks.py
│   └── logging_setup.py
└── tests/
    ├── test_signature.py
    └── test_client.py

import os
from dataclasses import dataclass, field


@dataclass(frozen=True)
class Settings:
    base_url: str
    api_key: str
    project_id: str
    webhook_secrets: tuple[str, ...]
    timeout_seconds: float
    max_retries: int
    tolerance_seconds: int = 300

    @staticmethod
    def from_env() -> "Settings":
        api_key = os.environ.get("BP_API_KEY")
        if not api_key:
            raise RuntimeError("BP_API_KEY non impostata")

        secrets = tuple(
            value
            for value in (
                os.environ.get("BP_WEBHOOK_SECRET"),
                os.environ.get("BP_WEBHOOK_SECRET_NEXT"),
            )
            if value
        )
        if not secrets:
            raise RuntimeError("Nessun segreto webhook configurato")

        return Settings(
            base_url=os.environ.get("BP_API_BASE_URL", "https://api.blupantheon.com"),
            api_key=api_key,
            project_id=os.environ.get("BP_PROJECT_ID", "bp_proj_7f3a9c21"),
            webhook_secrets=secrets,
            timeout_seconds=float(os.environ.get("BP_TIMEOUT_MS", "15000")) / 1000,
            max_retries=int(os.environ.get("BP_MAX_RETRIES", "3")),
        )


settings = Settings.from_env()

import uuid
from typing import Any

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from .config import settings
from .logging_setup import logger


class BluPantheonError(RuntimeError):
    def __init__(self, status_code: int, body: str) -> None:
        super().__init__(f"Errore {status_code}: {body}")
        self.status_code = status_code
        self.body = body


class BluPantheonClient:
    def __init__(self) -> None:
        self._session = requests.Session()

        retry = Retry(
            total=settings.max_retries,
            connect=settings.max_retries,
            read=settings.max_retries,
            status=settings.max_retries,
            backoff_factor=0.5,
            status_forcelist=(429, 500, 502, 503, 504),
            allowed_methods=frozenset(["GET", "POST", "PATCH", "PUT", "DELETE"]),
            respect_retry_after_header=True,
            raise_on_status=False,
        )
        adapter = HTTPAdapter(max_retries=retry, pool_maxsize=10)
        self._session.mount("https://", adapter)

        self._session.headers.update(
            {
                "Authorization": f"Bearer {settings.api_key}",
                "X-BP-Project": settings.project_id,
                "Accept": "application/json",
                "User-Agent": "acme-rpa/1.0",
            }
        )

    def _request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, Any] | None = None,
        json_body: dict[str, Any] | None = None,
        idempotency_key: str | None = None,
    ) -> dict[str, Any]:
        headers: dict[str, str] = {}
        if json_body is not None:
            headers["Content-Type"] = "application/json"
        if idempotency_key:
            headers["Idempotency-Key"] = idempotency_key

        url = f"{settings.base_url}{path}"
        response = self._session.request(
            method,
            url,
            params=params,
            json=json_body,
            headers=headers,
            timeout=settings.timeout_seconds,
        )

        remaining = response.headers.get("X-RateLimit-Remaining")
        if remaining is not None and int(remaining) < 10:
            logger.warning(
                "Limite di frequenza vicino alla soglia",
                extra={"remaining": remaining, "path": path},
            )

        if not response.ok:
            logger.error(
                "Chiamata fallita",
                extra={
                    "method": method,
                    "path": path,
                    "status": response.status_code,
                },
            )
            raise BluPantheonError(response.status_code, response.text)

        if response.status_code == 204 or not response.content:
            return {}
        return response.json()

    def get_session(self, session_id: str) -> dict[str, Any]:
        return self._request("GET", f"/identity/v1/sessions/{session_id}")

    def list_bookings(self, since: str, limit: int = 50) -> dict[str, Any]:
        return self._request(
            "GET",
            "/identity/v1/bookings",
            params={"since": since, "limit": limit},
        )

    def add_session_note(
        self,
        session_id: str,
        body: str,
        idempotency_key: str | None = None,
    ) -> dict[str, Any]:
        return self._request(
            "POST",
            f"/identity/v1/sessions/{session_id}/notes",
            json_body={"body": body, "visibility": "internal", "author": "python-rpa"},
            idempotency_key=idempotency_key or str(uuid.uuid4()),
        )

    def upsert_client(
        self,
        payload: dict[str, Any],
        idempotency_key: str,
    ) -> dict[str, Any]:
        return self._request(
            "POST",
            "/identity/v1/clients",
            json_body=payload,
            idempotency_key=idempotency_key,
        )

    def close(self) -> None:
        self._session.close()

import requests
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential_jitter,
)

from .config import settings


class RetryableError(RuntimeError):
    pass


@retry(
    retry=retry_if_exception_type((RetryableError, requests.ConnectionError, requests.Timeout)),
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=60),
    reraise=True,
)
def get_with_retry(session: requests.Session, path: str) -> dict:
    response = session.get(
        f"{settings.base_url}{path}",
        timeout=settings.timeout_seconds,
    )
    if response.status_code == 429 or response.status_code >= 500:
        raise RetryableError(f"Stato ritentabile: {response.status_code}")
    response.raise_for_status()
    return response.json()

import hashlib
import hmac
import time
from dataclasses import dataclass

from .config import settings

_HEX_LENGTH = 64


@dataclass(frozen=True)
class VerificationResult:
    valid: bool
    reason: str
    http_status: int


def _parse_header(header: str) -> tuple[int | None, list[str]]:
    timestamp: int | None = None
    signatures: list[str] = []

    for segment in header.split(","):
        key, separator, value = segment.partition("=")
        if not separator:
            continue
        key = key.strip()
        value = value.strip()
        if key == "t":
            try:
                timestamp = int(value)
            except ValueError:
                timestamp = None
        elif key == "v1" and len(value) == _HEX_LENGTH:
            signatures.append(value)

    return timestamp, signatures


def verify_signature(
    raw_body: bytes,
    signature_header: str | None,
    now: int | None = None,
) -> VerificationResult:
    if not signature_header:
        return VerificationResult(False, "header-mancante", 401)

    now = now if now is not None else int(time.time())
    timestamp, signatures = _parse_header(signature_header)

    if timestamp is None:
        return VerificationResult(False, "timestamp-non-valido", 400)
    if abs(now - timestamp) > settings.tolerance_seconds:
        return VerificationResult(False, "timestamp-fuori-tolleranza", 400)
    if not signatures:
        return VerificationResult(False, "firma-assente", 401)

    signed_payload = f"{timestamp}.".encode("utf-8") + raw_body

    for secret in settings.webhook_secrets:
        expected = hmac.new(
            secret.encode("utf-8"),
            signed_payload,
            hashlib.sha256,
        ).hexdigest()
        for candidate in signatures:
            if hmac.compare_digest(candidate, expected):
                return VerificationResult(True, "ok", 202)

    return VerificationResult(False, "firma-non-corrispondente", 401)

import json
import sqlite3
import time
from contextlib import closing

from fastapi import BackgroundTasks, FastAPI, Request, Response

from .logging_setup import logger
from .signature import verify_signature
from .tasks import handle_event

DB_PATH = "deliveries.db"

app = FastAPI(title="Blu Pantheon Receiver")


def init_db() -> None:
    with closing(sqlite3.connect(DB_PATH)) as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS deliveries (
                delivery_id TEXT PRIMARY KEY,
                event_id    TEXT NOT NULL,
                event_type  TEXT NOT NULL,
                received_at INTEGER NOT NULL
            )
            """
        )
        connection.commit()


def remember_delivery(delivery_id: str, event_id: str, event_type: str) -> bool:
    """Restituisce True se la consegna è nuova, False se già vista."""
    with closing(sqlite3.connect(DB_PATH)) as connection:
        try:
            connection.execute(
                "INSERT INTO deliveries VALUES (?, ?, ?, ?)",
                (delivery_id, event_id, event_type, int(time.time())),
            )
            connection.commit()
            return True
        except sqlite3.IntegrityError:
            return False


@app.on_event("startup")
def on_startup() -> None:
    init_db()


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/webhooks/blupantheon")
async def receive(request: Request, background_tasks: BackgroundTasks) -> Response:
    raw_body = await request.body()
    result = verify_signature(raw_body, request.headers.get("X-BP-Signature"))

    if not result.valid:
        logger.warning(
            "Webhook rifiutato",
            extra={"reason": result.reason, "delivery_id": request.headers.get("X-BP-Delivery")},
        )
        return Response(
            content=json.dumps({"error": result.reason}),
            status_code=result.http_status,
            media_type="application/json",
        )

    delivery_id = request.headers.get("X-BP-Delivery", "")
    event = json.loads(raw_body)

    if not remember_delivery(delivery_id, event["id"], event["event"]):
        logger.info("Consegna duplicata ignorata", extra={"delivery_id": delivery_id})
        return Response(status_code=202)

    background_tasks.add_task(handle_event, event, delivery_id)

    logger.info(
        "Webhook accettato",
        extra={
            "delivery_id": delivery_id,
            "event_id": event["id"],
            "event_type": event["event"],
            "trace_id": event.get("data", {}).get("traceId"),
        },
    )
    return Response(status_code=202)

uvicorn src.receiver:app --host 0.0.0.0 --port 8000 --workers 2

from typing import Any

from .client import BluPantheonClient, BluPantheonError
from .logging_setup import logger


def handle_event(event: dict[str, Any], delivery_id: str) -> None:
    event_type = event["event"]
    handlers = {
        "session.handover.requested": _handle_handover,
        "booking.confirmed": _handle_booking,
        "client.created": _handle_client,
        "document.classified": _handle_document,
    }

    handler = handlers.get(event_type)
    if handler is None:
        logger.info("Evento non gestito", extra={"event_type": event_type})
        return

    try:
        handler(event, delivery_id)
    except BluPantheonError as error:
        logger.error(
            "Errore API durante l elaborazione",
            extra={
                "event_type": event_type,
                "delivery_id": delivery_id,
                "status": error.status_code,
            },
        )
        raise
    except Exception:
        logger.exception(
            "Errore inatteso durante l elaborazione",
            extra={"event_type": event_type, "delivery_id": delivery_id},
        )
        raise


def _handle_handover(event: dict[str, Any], delivery_id: str) -> None:
    data = event["data"]
    ticket_number = _open_ticket_in_gestionale(
        subject=f"Handover sessione {data['sessionId']}",
        description=data.get("summary", ""),
        priority=data.get("priority", "media"),
    )

    client = BluPantheonClient()
    try:
        client.add_session_note(
            data["sessionId"],
            f"Ticket aperto nel gestionale: {ticket_number}",
            idempotency_key=delivery_id,
        )
    finally:
        client.close()


def _handle_booking(event: dict[str, Any], delivery_id: str) -> None:
    data = event["data"]
    logger.info(
        "Prenotazione confermata",
        extra={"reference": data["reference"], "start_at": data["startAt"]},
    )


def _handle_client(event: dict[str, Any], delivery_id: str) -> None:
    data = event["data"]
    logger.info("Cliente creato", extra={"client_id": data["clientId"]})


def _handle_document(event: dict[str, Any], delivery_id: str) -> None:
    data = event["data"]
    logger.info(
        "Documento classificato",
        extra={
            "document_id": data["documentId"],
            "category": data["classification"]["category"],
        },
    )


def _open_ticket_in_gestionale(subject: str, description: str, priority: str) -> str:
    """Sostituisci con l integrazione reale verso il gestionale."""
    logger.info("Apertura ticket", extra={"subject": subject, "priority": priority})
    return "PRT-2026-004512"

import json
import logging
import sys

RESERVED = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__.keys()) | {
    "message",
    "asctime",
}


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        for key, value in record.__dict__.items():
            if key not in RESERVED:
                payload[key] = value
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)


handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())

logger = logging.getLogger("blupantheon")
logger.setLevel(logging.INFO)
logger.addHandler(handler)
logger.propagate = False

channels:
  - conda-forge
dependencies:
  - python=3.11.9
  - pip=24.0
  - pip:
      - requests==2.32.3
      - urllib3==2.2.2
      - tenacity==8.5.0
      - fastapi==0.115.0
      - uvicorn==0.30.6
      - rpaframework==28.6.0
      - pytest==8.3.2

tasks:
  Elabora coda eventi:
    shell: python -m src.tasks_runner

  Riconciliazione oraria:
    shell: python -m src.reconcile

condaConfigFile: conda.yaml
artifactsDir: output

PATH:
  - .
PYTHONPATH:
  - .

ignoreFiles:
  - .gitignore

import os

from RPA.Robocorp.Vault import Vault


def load_secrets_into_env() -> None:
    """Carica i segreti dal Vault nelle variabili d ambiente del processo."""
    vault = Vault()
    secret = vault.get_secret("blupantheon")

    os.environ["BP_API_KEY"] = secret["api_key"]
    os.environ["BP_WEBHOOK_SECRET"] = secret["webhook_secret"]
    os.environ["BP_PROJECT_ID"] = secret["project_id"]

rcc run --task "Elabora coda eventi"
rcc task testrun

import hashlib
import hmac
import time

import pytest

from src.signature import verify_signature

SECRET = "whsec_xxxxxxxxxxxxxxxx"
BODY = b'{"id":"evt_test","event":"webhook.test","data":{}}'


def build_header(timestamp: int, body: bytes = BODY, secret: str = SECRET) -> str:
    signed = f"{timestamp}.".encode("utf-8") + body
    digest = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest()
    return f"t={timestamp},v1={digest}"


def test_firma_valida():
    now = int(time.time())
    result = verify_signature(BODY, build_header(now), now)
    assert result.valid is True
    assert result.reason == "ok"


def test_header_mancante():
    result = verify_signature(BODY, None, int(time.time()))
    assert result.valid is False
    assert result.reason == "header-mancante"


def test_timestamp_fuori_tolleranza():
    now = int(time.time())
    old = now - 600
    result = verify_signature(BODY, build_header(old), now)
    assert result.valid is False
    assert result.reason == "timestamp-fuori-tolleranza"


def test_corpo_alterato():
    now = int(time.time())
    header = build_header(now)
    result = verify_signature(BODY + b" ", header, now)
    assert result.valid is False
    assert result.reason == "firma-non-corrispondente"


def test_segreto_errato():
    now = int(time.time())
    header = build_header(now, secret="whsec_sbagliato")
    result = verify_signature(BODY, header, now)
    assert result.valid is False
    assert result.reason == "firma-non-corrispondente"


@pytest.mark.parametrize(
    "header",
    ["", "t=abc,v1=xyz", "v1=abcdef", "t=1767225600"],
)
def test_header_malformati(header):
    result = verify_signature(BODY, header or None, int(time.time()))
    assert result.valid is False

python -m pytest tests/ -v

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY src/ ./src/

RUN useradd --create-home --uid 10001 appuser
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
  CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

CMD ["uvicorn", "src.receiver:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]