feat: add Chevening opportunity collector

Dockerfile

@@ -19,4 +19,4 @@ USER appuser
 
 EXPOSE 7860
 
-CMD ["sh", "-c", "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port ${PORT:-7860}"]
+CMD ["sh", "-c", "alembic upgrade head && python -m scripts.check_chevening_collector && uvicorn app.main:app --host 0.0.0.0 --port ${PORT:-7860}"]

app/collectors/providers/chevening.py

@@ -0,0 +1,270 @@
+import re
+from datetime import date, datetime
+
+import httpx
+from bs4 import BeautifulSoup
+
+from app.collectors.base import BaseOpportunityCollector
+from app.collectors.types import CollectedOpportunity
+
+
+class CheveningCollectorError(RuntimeError):
+    """Erreur produite pendant la collecte de Chevening."""
+
+
+class CheveningCollector(BaseOpportunityCollector):
+    source_slug = "chevening"
+
+    TIMELINE_URL = (
+        "https://www.chevening.org/"
+        "scholarships/application-timeline/"
+    )
+
+    SCHOLARSHIPS_URL = (
+        "https://www.chevening.org/scholarships/"
+    )
+
+    REQUEST_TIMEOUT = 30.0
+
+    def __init__(
+        self,
+        client: httpx.AsyncClient | None = None,
+    ) -> None:
+        self._external_client = client
+
+    async def collect(
+        self,
+    ) -> list[CollectedOpportunity]:
+        html = await self._fetch_timeline_page()
+        page_text = self._extract_page_text(html)
+
+        opening_date = self._extract_opening_date(
+            page_text
+        )
+        deadline = self._extract_deadline(page_text)
+
+        if deadline is None:
+            raise CheveningCollectorError(
+                "La date limite Chevening n'a pas "
+                "été trouvée sur la page officielle."
+            )
+
+        academic_year = self._build_academic_year(
+            opening_date=opening_date,
+            deadline=deadline,
+        )
+
+        title = (
+            f"Chevening Scholarships {academic_year}"
+        )
+
+        return [
+            CollectedOpportunity(
+                source_slug=self.source_slug,
+                title=title,
+                official_url=self.TIMELINE_URL,
+                raw_content=page_text,
+                organization_name="Chevening",
+                category="scholarship",
+                language="en",
+                publication_date=opening_date,
+                deadline=deadline,
+                application_url=self.SCHOLARSHIPS_URL,
+            )
+        ]
+
+    async def _fetch_timeline_page(self) -> str:
+        headers = {
+            "User-Agent": (
+                "VerifPulseOpportunityCollector/1.0 "
+                "(https://verifpulse-platform.vercel.app)"
+            ),
+            "Accept": (
+                "text/html,application/xhtml+xml,"
+                "application/xml;q=0.9,*/*;q=0.8"
+            ),
+            "Accept-Language": "en-GB,en;q=0.9",
+        }
+
+        if self._external_client is not None:
+            return await self._request_html(
+                self._external_client,
+                headers,
+            )
+
+        async with httpx.AsyncClient(
+            timeout=self.REQUEST_TIMEOUT,
+            follow_redirects=True,
+            headers=headers,
+        ) as client:
+            return await self._request_html(
+                client,
+                headers,
+            )
+
+    async def _request_html(
+        self,
+        client: httpx.AsyncClient,
+        headers: dict[str, str],
+    ) -> str:
+        try:
+            response = await client.get(
+                self.TIMELINE_URL,
+                headers=headers,
+            )
+            response.raise_for_status()
+
+        except httpx.HTTPError as error:
+            raise CheveningCollectorError(
+                "Impossible de récupérer la page "
+                "officielle Chevening."
+            ) from error
+
+        content_type = response.headers.get(
+            "content-type",
+            "",
+        )
+
+        if "text/html" not in content_type.lower():
+            raise CheveningCollectorError(
+                "Chevening n'a pas retourné une page HTML."
+            )
+
+        return response.text
+
+    @staticmethod
+    def _extract_page_text(html: str) -> str:
+        soup = BeautifulSoup(html, "html.parser")
+
+        for element in soup(
+            [
+                "script",
+                "style",
+                "noscript",
+                "svg",
+                "header",
+                "footer",
+                "nav",
+            ]
+        ):
+            element.decompose()
+
+        main_content = (
+            soup.find("main")
+            or soup.find("article")
+            or soup.body
+        )
+
+        if main_content is None:
+            raise CheveningCollectorError(
+                "Le contenu principal de la page "
+                "Chevening est introuvable."
+            )
+
+        text = main_content.get_text(
+            separator=" ",
+            strip=True,
+        )
+
+        normalized_text = re.sub(
+            r"\s+",
+            " ",
+            text,
+        ).strip()
+
+        if len(normalized_text) < 200:
+            raise CheveningCollectorError(
+                "Le contenu Chevening extrait est "
+                "anormalement court."
+            )
+
+        return normalized_text
+
+    @classmethod
+    def _extract_opening_date(
+        cls,
+        page_text: str,
+    ) -> date | None:
+        patterns = (
+            (
+                r"(\d{1,2}\s+[A-Za-z]+\s+\d{4})"
+                r".{0,100}?Applications open"
+            ),
+            (
+                r"Applications open.{0,100}?"
+                r"(\d{1,2}\s+[A-Za-z]+\s+\d{4})"
+            ),
+        )
+
+        return cls._find_date(
+            page_text,
+            patterns,
+        )
+
+    @classmethod
+    def _extract_deadline(
+        cls,
+        page_text: str,
+    ) -> date | None:
+        patterns = (
+            (
+                r"deadline for applications is\s+"
+                r"(\d{1,2}\s+[A-Za-z]+\s+\d{4})"
+            ),
+            (
+                r"(\d{1,2}\s+[A-Za-z]+\s+\d{4})"
+                r".{0,100}?Applications close"
+            ),
+            (
+                r"Applications close.{0,100}?"
+                r"(\d{1,2}\s+[A-Za-z]+\s+\d{4})"
+            ),
+        )
+
+        return cls._find_date(
+            page_text,
+            patterns,
+        )
+
+    @staticmethod
+    def _find_date(
+        page_text: str,
+        patterns: tuple[str, ...],
+    ) -> date | None:
+        for pattern in patterns:
+            match = re.search(
+                pattern,
+                page_text,
+                flags=re.IGNORECASE,
+            )
+
+            if match is None:
+                continue
+
+            try:
+                return datetime.strptime(
+                    match.group(1),
+                    "%d %B %Y",
+                ).date()
+
+            except ValueError:
+                continue
+
+        return None
+
+    @staticmethod
+    def _build_academic_year(
+        *,
+        opening_date: date | None,
+        deadline: date,
+    ) -> str:
+        starting_year = (
+            opening_date.year
+            if opening_date is not None
+            else deadline.year
+        )
+
+        return (
+            f"{starting_year + 1}/"
+            f"{str(starting_year + 2)[-2:]}"
+        )

requirements.txt

@@ -3,4 +3,6 @@ uvicorn[standard]>=0.35,<1
 pydantic-settings>=2.10,<3
 SQLAlchemy[asyncio]>=2.0,<3.0
 psycopg[binary]>=3.2,<4.0
-alembic>=1.16,<2.0
+alembic>=1.16,<2.0
+httpx>=0.28,<1.0
+beautifulsoup4>=4.13,<5.0

scripts/check_chevening_collector.py

@@ -0,0 +1,102 @@
+import asyncio
+import json
+import sys
+from dataclasses import asdict
+from datetime import date
+from typing import Any
+
+from app.collectors.providers.chevening import (
+    CheveningCollector,
+    CheveningCollectorError,
+)
+
+
+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, dict):
+        return {
+            key: serialize_value(item)
+            for key, item in value.items()
+        }
+
+    return value
+
+
+async def main() -> None:
+    collector = CheveningCollector()
+
+    print(
+        "[collector-check] Starting Chevening check...",
+        flush=True,
+    )
+
+    opportunities = await collector.collect()
+
+    if not opportunities:
+        raise RuntimeError(
+            "Chevening returned no opportunity."
+        )
+
+    for opportunity in opportunities:
+        payload = serialize_value(
+            asdict(opportunity)
+        )
+
+        # Le contenu complet est volontairement exclu des 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] Chevening check passed: "
+            f"{len(opportunities)} opportunity collected."
+        ),
+        flush=True,
+    )
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+
+    except CheveningCollectorError as error:
+        print(
+            f"[collector-check] 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