scripts/test_mastercard_collector.py

import asyncio
import json
import sys
from dataclasses import asdict
from datetime import date
from typing import Any

from app.collectors.providers.mastercard_foundation import (
    MastercardFoundationCollector,
    MastercardFoundationCollectorError,
)


def serialize_value(
    value: Any,
) -> Any:
    if isinstance(value, date):
        return value.isoformat()

    if isinstance(value, list):
        return [
            serialize_value(item)
            for item in value
        ]

    if isinstance(value, tuple):
        return [
            serialize_value(item)
            for item in value
        ]

    if isinstance(value, dict):
        return {
            key: serialize_value(item)
            for key, item in value.items()
        }

    return value


async def main() -> None:
    collector = MastercardFoundationCollector()

    print(
        "[collector-check] "
        "Starting Mastercard Foundation check...",
        flush=True,
    )

    opportunities = await collector.collect()

    if not opportunities:
        raise RuntimeError(
            "Mastercard Foundation returned "
            "no open opportunity."
        )

    for opportunity in opportunities:
        payload = serialize_value(
            asdict(opportunity)
        )

        # Le contenu HTML transformé en texte peut être
        # volumineux : on ne l’affiche pas dans les logs.
        raw_content = payload.pop(
            "raw_content",
            "",
        )

        payload["raw_content_length"] = len(
            raw_content
        )

        print(
            json.dumps(
                payload,
                indent=2,
                ensure_ascii=False,
            ),
            flush=True,
        )

    print(
        (
            "[collector-check] "
            "Mastercard Foundation check passed: "
            f"{len(opportunities)} "
            "opportunity collected."
        ),
        flush=True,
    )


if __name__ == "__main__":
    try:
        asyncio.run(main())

    except MastercardFoundationCollectorError as error:
        print(
            (
                "[collector-check] "
                f"Collection failed: {error}"
            ),
            file=sys.stderr,
            flush=True,
        )

        raise SystemExit(1) from error

    except Exception as error:
        print(
            (
                "[collector-check] "
                "Unexpected failure: "
                f"{type(error).__name__}: {error}"
            ),
            file=sys.stderr,
            flush=True,
        )

        raise SystemExit(1) from error