feat(collectors): add University of Buea opportunity collector

Dockerfile

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

app/collectors/providers/university_buea.py

@@ -0,0 +1,895 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+import unicodedata
+from datetime import date, datetime, timedelta
+from io import BytesIO
+from pathlib import PurePosixPath
+from urllib.parse import urljoin, urlparse
+
+import httpx
+from bs4 import BeautifulSoup
+from pypdf import PdfReader
+
+from app.collectors.base import BaseOpportunityCollector
+from app.collectors.types import CollectedOpportunity
+
+
+logging.getLogger("pypdf").setLevel(logging.ERROR)
+
+
+class UniversityBueaCollectorError(RuntimeError):
+    """Controlled University of Buea collection error."""
+
+
+class UniversityBueaCollector(BaseOpportunityCollector):
+    """Collect open opportunities from official University of Buea pages."""
+
+    source_slug = "university-buea"
+
+    BASE_URL = "https://www.ubuea.cm/"
+    LISTING_URLS = (
+        "https://www.ubuea.cm/",
+        "https://www.ubuea.cm/index.php/announcements/",
+        "https://www.ubuea.cm/index.php/about-us/job-opportunities/",
+        "https://www.ubuea.cm/index.php/news-updates/",
+        "https://www.ubuea.cm/index.php/news-updates/page/2/",
+        "https://www.ubuea.cm/index.php/news-updates/page/3/",
+    )
+
+    ALLOWED_HOSTS = {
+        "www.ubuea.cm",
+        "ubuea.cm",
+        "www.ubstudent.online",
+        "ubstudent.online",
+    }
+
+    REQUEST_TIMEOUT = 45.0
+    MAX_CONCURRENT_REQUESTS = 5
+    MAX_CANDIDATES = 100
+    MAX_DOCUMENTS_PER_PAGE = 8
+    MAX_DOCUMENT_SIZE = 20 * 1024 * 1024
+    MAX_PDF_PAGES = 40
+    MAX_UNDATED_AGE_DAYS = 150
+
+    OPPORTUNITY_PATTERNS = (
+        r"\bcall\s+for\s+applications?\b",
+        r"\bapplications?\s+(?:are\s+)?open\b",
+        r"\bapply\s+(?:now|for)\b",
+        r"\badmissions?\b",
+        r"\bentrance\s+exam(?:ination)?s?\b",
+        r"\bcompetitive\s+entrance\b",
+        r"\bscholarships?\b",
+        r"\bfellowships?\b",
+        r"\binternships?\b",
+        r"\bjob\s+(?:advert|opportunit(?:y|ies)|position)\b",
+        r"\bopen\s+positions?\b",
+        r"\bresearch\s+assistants?\b",
+        r"\bproject\s+(?:manager|assistant)\b",
+        r"\bvacanc(?:y|ies)\b",
+        r"\brecruitment\s+notice\b",
+        r"\btraining\s+(?:programme|program|opportunity)\b",
+        r"\berasmus\s+programme\b",
+        r"\bgrants?\b",
+    )
+
+    RESULT_PATTERNS = (
+        r"\badmission\s+lists?\b",
+        r"\blist\s+of\s+(?:admitted|successful)\s+candidates\b",
+        r"\bresults?\s+(?:published|released)\b",
+        r"\bentrance\s+exam(?:ination)?\s+results?\b",
+        r"\bselected\s+candidates\b",
+        r"\bshortlisted\s+candidates\b",
+    )
+
+    CLOSED_PATTERNS = (
+        r"\bapplications?\s+(?:are\s+)?closed\b",
+        r"\bapplication\s+period\s+has\s+ended\b",
+        r"\bdeadline\s+has\s+passed\b",
+        r"\bposition\s+(?:has\s+been|is)\s+filled\b",
+        r"\bexpired\b",
+    )
+
+    CANDIDATE_LINK_PATTERNS = OPPORTUNITY_PATTERNS + (
+        r"\bapplication\s+forms?\b",
+        r"\bhttc\b",
+        r"\bhtttc\b",
+        r"\bjob\b",
+    )
+
+    DEADLINE_CONTEXT_PATTERNS = (
+        r"application\s+deadline",
+        r"deadline",
+        r"closing\s+date",
+        r"submit(?:ted)?\s+(?:on|by|before)",
+        r"submission\s+deadline",
+        r"not\s+later\s+than",
+        r"on\s+or\s+before",
+        r"latest\s+by",
+        r"before",
+    )
+
+    MONTHS = {
+        "january": 1,
+        "february": 2,
+        "march": 3,
+        "april": 4,
+        "may": 5,
+        "june": 6,
+        "july": 7,
+        "august": 8,
+        "september": 9,
+        "october": 10,
+        "november": 11,
+        "december": 12,
+    }
+
+    def __init__(self, client: httpx.AsyncClient | None = None) -> None:
+        self._external_client = client
+
+    async def collect(self) -> list[CollectedOpportunity]:
+        headers = self._build_headers()
+
+        if self._external_client is not None:
+            return await self._collect_with_client(
+                client=self._external_client,
+                headers=headers,
+            )
+
+        async with httpx.AsyncClient(
+            timeout=self.REQUEST_TIMEOUT,
+            follow_redirects=True,
+            headers=headers,
+        ) as client:
+            return await self._collect_with_client(
+                client=client,
+                headers=headers,
+            )
+
+    async def _collect_with_client(
+        self,
+        *,
+        client: httpx.AsyncClient,
+        headers: dict[str, str],
+    ) -> list[CollectedOpportunity]:
+        listing_results = await asyncio.gather(
+            *(
+                self._request_html(client=client, url=url, headers=headers)
+                for url in self.LISTING_URLS
+            ),
+            return_exceptions=True,
+        )
+
+        candidates: list[tuple[str, str]] = []
+
+        for listing_url, result in zip(
+            self.LISTING_URLS,
+            listing_results,
+            strict=True,
+        ):
+            if isinstance(result, Exception):
+                print(
+                    "[University of Buea] Listing ignored: "
+                    f"{listing_url} - {type(result).__name__}: {result}",
+                    flush=True,
+                )
+                continue
+
+            candidates.extend(
+                self._extract_candidate_links(
+                    html=result,
+                    page_url=listing_url,
+                )
+            )
+
+        candidates = self._deduplicate_candidates(candidates)[
+            : self.MAX_CANDIDATES
+        ]
+
+        if not candidates:
+            raise UniversityBueaCollectorError(
+                "No University of Buea opportunity candidate was discovered."
+            )
+
+        print(
+            f"[University of Buea] {len(candidates)} candidate(s) discovered.",
+            flush=True,
+        )
+
+        semaphore = asyncio.Semaphore(self.MAX_CONCURRENT_REQUESTS)
+
+        async def collect_one(
+            candidate_url: str,
+            discovered_title: str,
+        ) -> CollectedOpportunity | None:
+            async with semaphore:
+                return await self._collect_candidate(
+                    client=client,
+                    url=candidate_url,
+                    discovered_title=discovered_title,
+                    headers=headers,
+                )
+
+        results = await asyncio.gather(
+            *(
+                collect_one(candidate_url, discovered_title)
+                for candidate_url, discovered_title in candidates
+            ),
+            return_exceptions=True,
+        )
+
+        opportunities: list[CollectedOpportunity] = []
+
+        for (candidate_url, _), result in zip(
+            candidates,
+            results,
+            strict=True,
+        ):
+            if isinstance(result, CollectedOpportunity):
+                opportunities.append(result)
+            elif isinstance(result, Exception):
+                print(
+                    "[University of Buea] Candidate ignored: "
+                    f"{candidate_url} - {type(result).__name__}: {result}",
+                    flush=True,
+                )
+
+        opportunities = self._deduplicate_opportunities(opportunities)
+
+        print(
+            f"[University of Buea] {len(opportunities)} "
+            "open opportunity/opportunities retained.",
+            flush=True,
+        )
+
+        return opportunities
+
+    async def _collect_candidate(
+        self,
+        *,
+        client: httpx.AsyncClient,
+        url: str,
+        discovered_title: str,
+        headers: dict[str, str],
+    ) -> CollectedOpportunity | None:
+        response = await self._request_resource(
+            client=client,
+            url=url,
+            headers=headers,
+        )
+        content_type = response.headers.get("content-type", "").lower()
+        path = urlparse(str(response.url)).path.lower()
+
+        if "application/pdf" in content_type or path.endswith(".pdf"):
+            return self._collect_pdf_candidate(
+                response=response,
+                url=str(response.url),
+                discovered_title=discovered_title,
+            )
+
+        if "text/html" not in content_type:
+            return None
+
+        return await self._collect_html_candidate(
+            client=client,
+            html=response.text,
+            url=str(response.url),
+            discovered_title=discovered_title,
+            headers=headers,
+        )
+
+    def _collect_pdf_candidate(
+        self,
+        *,
+        response: httpx.Response,
+        url: str,
+        discovered_title: str,
+    ) -> CollectedOpportunity | None:
+        if len(response.content) > self.MAX_DOCUMENT_SIZE:
+            return None
+
+        pdf_text = self._extract_pdf_text(response.content)
+        title = self._normalize_text(discovered_title) or self._title_from_url(url)
+        full_text = self._normalize_text(f"{title} {pdf_text}")
+
+        if not self._is_open_candidate(full_text):
+            return None
+
+        publication_date = self._extract_upload_month(url)
+
+        return self._build_opportunity(
+            title=title,
+            official_url=url,
+            article_url=url,
+            full_text=full_text,
+            description=pdf_text,
+            publication_date=publication_date,
+            document_urls=[url],
+            application_url=url,
+            image_url=None,
+        )
+
+    async def _collect_html_candidate(
+        self,
+        *,
+        client: httpx.AsyncClient,
+        html: str,
+        url: str,
+        discovered_title: str,
+        headers: dict[str, str],
+    ) -> CollectedOpportunity | None:
+        soup = BeautifulSoup(html, "html.parser")
+        title = self._extract_title(soup) or discovered_title
+        article_text = self._extract_article_text(soup)
+
+        if not title or not article_text:
+            return None
+
+        document_urls = self._extract_document_urls(soup=soup, page_url=url)
+        document_texts = await self._collect_document_texts(
+            client=client,
+            document_urls=document_urls,
+            headers=headers,
+        )
+        full_text = self._normalize_text(
+            " ".join((title, article_text, *document_texts))
+        )
+
+        if not self._is_open_candidate(full_text):
+            return None
+
+        publication_date = self._extract_publication_date(soup=soup, url=url)
+        official_url = document_urls[0] if document_urls else url
+
+        return self._build_opportunity(
+            title=title,
+            official_url=official_url,
+            article_url=url,
+            full_text=full_text,
+            description=self._normalize_text(
+                " ".join((article_text, *document_texts))
+            )[:15000],
+            publication_date=publication_date,
+            document_urls=document_urls,
+            application_url=self._extract_application_url(
+                soup=soup,
+                page_url=url,
+                document_urls=document_urls,
+            ),
+            image_url=self._extract_image_url(soup=soup, page_url=url),
+        )
+
+    def _build_opportunity(
+        self,
+        *,
+        title: str,
+        official_url: str,
+        article_url: str,
+        full_text: str,
+        description: str,
+        publication_date: date | None,
+        document_urls: list[str],
+        application_url: str,
+        image_url: str | None,
+    ) -> CollectedOpportunity | None:
+        if self._contains_any(full_text, self.RESULT_PATTERNS):
+            print(
+                f"[University of Buea] Result ignored: {title}",
+                flush=True,
+            )
+            return None
+
+        if self._contains_any(full_text, self.CLOSED_PATTERNS):
+            print(
+                f"[University of Buea] Closed publication ignored: {title}",
+                flush=True,
+            )
+            return None
+
+        deadline = self._extract_deadline(full_text)
+        today = date.today()
+
+        if deadline is not None and deadline < today:
+            return None
+
+        requires_manual_review = deadline is None
+
+        if requires_manual_review:
+            if publication_date is None:
+                print(
+                    "[University of Buea] Missing publication date and "
+                    f"deadline: {title}",
+                    flush=True,
+                )
+                return None
+
+            if publication_date < today - timedelta(
+                days=self.MAX_UNDATED_AGE_DAYS
+            ):
+                return None
+
+            print(
+                f"[University of Buea] Retained for manual review: {title}",
+                flush=True,
+            )
+        else:
+            print(
+                f"[University of Buea] Open: {title} - "
+                f"deadline {deadline.isoformat()}",
+                flush=True,
+            )
+
+        category = self._classify_category(full_text)
+
+        return CollectedOpportunity(
+            source_slug=self.source_slug,
+            title=self._normalize_text(title)[:250],
+            official_url=official_url,
+            raw_content=full_text,
+            summary=self._build_summary(
+                title=title,
+                text=description,
+                category=category,
+                deadline=deadline,
+            )[:500],
+            description=description[:15000],
+            organization_name="University of Buea",
+            category=category,
+            country_code="CM",
+            location="Buea, Cameroon",
+            target_countries=["CM"],
+            study_levels=self._extract_study_levels(full_text),
+            eligibility={
+                "deadline_status": (
+                    "confirmed" if deadline is not None else "not_extracted"
+                ),
+                "requires_manual_review": requires_manual_review,
+                "article_url": article_url,
+                "official_documents": document_urls,
+            },
+            language="en",
+            publication_date=publication_date,
+            deadline=deadline,
+            application_url=application_url,
+            image_url=image_url,
+        )
+
+    async def _collect_document_texts(
+        self,
+        *,
+        client: httpx.AsyncClient,
+        document_urls: list[str],
+        headers: dict[str, str],
+    ) -> list[str]:
+        results = await asyncio.gather(
+            *(
+                self._request_document_text(
+                    client=client,
+                    url=url,
+                    headers=headers,
+                )
+                for url in document_urls[: self.MAX_DOCUMENTS_PER_PAGE]
+            ),
+            return_exceptions=True,
+        )
+        return [result for result in results if isinstance(result, str) and result]
+
+    async def _request_document_text(
+        self,
+        *,
+        client: httpx.AsyncClient,
+        url: str,
+        headers: dict[str, str],
+    ) -> str:
+        try:
+            response = await self._request_resource(
+                client=client,
+                url=url,
+                headers=headers,
+            )
+        except UniversityBueaCollectorError:
+            return ""
+
+        if len(response.content) > self.MAX_DOCUMENT_SIZE:
+            return ""
+
+        content_type = response.headers.get("content-type", "").lower()
+        if "application/pdf" in content_type or urlparse(url).path.lower().endswith(
+            ".pdf"
+        ):
+            return self._extract_pdf_text(response.content)
+        return ""
+
+    async def _request_resource(
+        self,
+        *,
+        client: httpx.AsyncClient,
+        url: str,
+        headers: dict[str, str],
+    ) -> httpx.Response:
+        try:
+            response = await client.get(url, headers=headers)
+            response.raise_for_status()
+            return response
+        except httpx.HTTPError as error:
+            raise UniversityBueaCollectorError(
+                f"Unable to fetch University of Buea resource: {url}"
+            ) from error
+
+    async def _request_html(
+        self,
+        *,
+        client: httpx.AsyncClient,
+        url: str,
+        headers: dict[str, str],
+    ) -> str:
+        response = await self._request_resource(
+            client=client,
+            url=url,
+            headers=headers,
+        )
+        if "text/html" not in response.headers.get("content-type", "").lower():
+            raise UniversityBueaCollectorError(
+                f"University of Buea resource is not HTML: {url}"
+            )
+        return response.text
+
+    @classmethod
+    def _extract_candidate_links(
+        cls,
+        *,
+        html: str,
+        page_url: str,
+    ) -> list[tuple[str, str]]:
+        soup = BeautifulSoup(html, "html.parser")
+        candidates: list[tuple[str, str]] = []
+
+        for anchor in soup.select("main a[href], article a[href], .gdlr-core-page-builder-body a[href]"):
+            href = anchor.get("href")
+            if not isinstance(href, str):
+                continue
+
+            label = cls._normalize_text(anchor.get_text(" ", strip=True))
+            absolute_url = urljoin(page_url, href.strip())
+            parsed = urlparse(absolute_url)
+
+            if parsed.netloc not in cls.ALLOWED_HOSTS:
+                continue
+            if not cls._looks_like_candidate(label=label, url=absolute_url):
+                continue
+
+            candidates.append((absolute_url, label or cls._title_from_url(absolute_url)))
+
+        return candidates
+
+    @classmethod
+    def _looks_like_candidate(cls, *, label: str, url: str) -> bool:
+        searchable = cls._fold_text(f"{label} {url}")
+        return cls._contains_any(searchable, cls.CANDIDATE_LINK_PATTERNS)
+
+    @classmethod
+    def _extract_document_urls(
+        cls,
+        *,
+        soup: BeautifulSoup,
+        page_url: str,
+    ) -> list[str]:
+        urls: list[str] = []
+        for anchor in soup.select("article a[href], .entry-content a[href], main a[href]"):
+            href = anchor.get("href")
+            if not isinstance(href, str):
+                continue
+            absolute_url = urljoin(page_url, href.strip())
+            path = urlparse(absolute_url).path.lower()
+            if path.endswith(".pdf") and urlparse(absolute_url).netloc in cls.ALLOWED_HOSTS:
+                urls.append(absolute_url)
+        return list(dict.fromkeys(urls))
+
+    @classmethod
+    def _extract_title(cls, soup: BeautifulSoup) -> str:
+        for selector in ("h1.entry-title", "article h1", "main h1", "h1"):
+            element = soup.select_one(selector)
+            if element is not None:
+                title = cls._normalize_text(element.get_text(" ", strip=True))
+                if title:
+                    return title
+        return ""
+
+    @classmethod
+    def _extract_article_text(cls, soup: BeautifulSoup) -> str:
+        for unwanted in soup.select(
+            "script, style, nav, footer, header, aside, form, noscript, "
+            ".comments-area, #comments"
+        ):
+            unwanted.decompose()
+
+        for selector in (
+            "article .entry-content",
+            ".entry-content",
+            ".gdlr-core-page-builder-body",
+            "article",
+            "main",
+        ):
+            element = soup.select_one(selector)
+            if element is None:
+                continue
+            text = cls._normalize_text(element.get_text(" ", strip=True))
+            if len(text) >= 50:
+                return text
+        return ""
+
+    @classmethod
+    def _extract_pdf_text(cls, content: bytes) -> str:
+        try:
+            reader = PdfReader(BytesIO(content))
+            pages = [
+                text
+                for page in reader.pages[: cls.MAX_PDF_PAGES]
+                if (text := page.extract_text() or "")
+            ]
+            return cls._normalize_text(" ".join(pages))
+        except Exception:
+            return ""
+
+    @classmethod
+    def _is_open_candidate(cls, text: str) -> bool:
+        return cls._contains_any(text, cls.OPPORTUNITY_PATTERNS)
+
+    @classmethod
+    def _extract_publication_date(
+        cls,
+        *,
+        soup: BeautifulSoup,
+        url: str,
+    ) -> date | None:
+        match = re.search(r"/index\.php/(20\d{2})/(\d{2})/(\d{2})/", url)
+        if match:
+            parsed = cls._safe_date(
+                year=int(match.group(1)),
+                month=int(match.group(2)),
+                day=int(match.group(3)),
+            )
+            if parsed is not None:
+                return parsed
+
+        for selector in (
+            'meta[property="article:published_time"]',
+            'meta[itemprop="datePublished"]',
+            "article time[datetime]",
+        ):
+            element = soup.select_one(selector)
+            if element is None:
+                continue
+            value = element.get("content") or element.get("datetime")
+            if isinstance(value, str):
+                try:
+                    return datetime.fromisoformat(
+                        value.strip().replace("Z", "+00:00")
+                    ).date()
+                except ValueError:
+                    continue
+
+        return cls._extract_upload_month(url)
+
+    @classmethod
+    def _extract_upload_month(cls, url: str) -> date | None:
+        match = re.search(r"/wp-content/uploads/(20\d{2})/(\d{2})/", url)
+        if not match:
+            return None
+        return cls._safe_date(
+            year=int(match.group(1)),
+            month=int(match.group(2)),
+            day=1,
+        )
+
+    @classmethod
+    def _extract_deadline(cls, text: str) -> date | None:
+        folded = cls._fold_text(cls._normalize_text(text))
+        context_pattern = "|".join(cls.DEADLINE_CONTEXT_PATTERNS)
+        windows = re.findall(
+            rf"(?:{context_pattern}).{{0,240}}",
+            folded,
+            flags=re.IGNORECASE,
+        )
+        candidates: list[date] = []
+        for window in windows:
+            candidates.extend(cls._extract_dates_from_text(window))
+        # Keep expired deadlines too: _build_opportunity will reject them.
+        # Returning None for a past deadline would incorrectly make an old
+        # publication look like a recent item requiring manual review.
+        return max(candidates) if candidates else None
+
+    @classmethod
+    def _extract_dates_from_text(cls, text: str) -> list[date]:
+        values: list[date] = []
+
+        for match in re.finditer(
+            r"\b([0-3]?\d)[/.\-]([01]?\d)[/.\-](20\d{2})\b",
+            text,
+        ):
+            parsed = cls._safe_date(
+                year=int(match.group(3)),
+                month=int(match.group(2)),
+                day=int(match.group(1)),
+            )
+            if parsed is not None:
+                values.append(parsed)
+
+        month_names = "|".join(cls.MONTHS)
+        pattern = re.compile(
+            rf"\b([0-3]?\d)(?:st|nd|rd|th)?\s+"
+            rf"({month_names})\s+(20\d{{2}})\b",
+            flags=re.IGNORECASE,
+        )
+        for match in pattern.finditer(text):
+            parsed = cls._safe_date(
+                year=int(match.group(3)),
+                month=cls.MONTHS[match.group(2).lower()],
+                day=int(match.group(1)),
+            )
+            if parsed is not None:
+                values.append(parsed)
+
+        return values
+
+    @staticmethod
+    def _safe_date(*, year: int, month: int, day: int) -> date | None:
+        try:
+            return date(year, month, day)
+        except ValueError:
+            return None
+
+    @classmethod
+    def _classify_category(cls, text: str) -> str:
+        folded = cls._fold_text(text)
+        if re.search(r"\binternships?\b", folded):
+            return "internship"
+        if re.search(r"\b(scholarships?|fellowships?|grants?)\b", folded):
+            return "scholarship"
+        if re.search(r"\b(entrance\s+exam|competitive\s+entrance)\b", folded):
+            return "competition"
+        if re.search(r"\b(job|position|vacancy|research\s+assistant)\b", folded):
+            return "job"
+        if re.search(r"\btraining\b", folded):
+            return "training"
+        return "program"
+
+    @classmethod
+    def _extract_study_levels(cls, text: str) -> list[str]:
+        folded = cls._fold_text(text)
+        patterns = {
+            "bachelor": (r"\bundergraduate\b", r"\bbachelor(?:'s)?\b", r"\bfirst cycle\b"),
+            "master": (r"\bmasters?\b", r"\bsecond cycle\b", r"\bgraduate\b"),
+            "phd": (r"\bph\.?d\.?\b", r"\bdoctoral\b", r"\bthird cycle\b"),
+            "hnd": (r"\bhnd\b", r"\bbts\b"),
+        }
+        return [
+            level
+            for level, level_patterns in patterns.items()
+            if any(re.search(pattern, folded) for pattern in level_patterns)
+        ]
+
+    @classmethod
+    def _extract_application_url(
+        cls,
+        *,
+        soup: BeautifulSoup,
+        page_url: str,
+        document_urls: list[str],
+    ) -> str:
+        for anchor in soup.select("article a[href], .entry-content a[href], main a[href]"):
+            href = anchor.get("href")
+            if not isinstance(href, str):
+                continue
+            label = cls._fold_text(anchor.get_text(" ", strip=True))
+            absolute_url = urljoin(page_url, href.strip())
+            if absolute_url in document_urls:
+                continue
+            if any(
+                word in label
+                for word in ("apply", "application", "register", "admission", "submit")
+            ):
+                return absolute_url
+        return page_url
+
+    @staticmethod
+    def _extract_image_url(
+        *,
+        soup: BeautifulSoup,
+        page_url: str,
+    ) -> str | None:
+        for selector in ('meta[property="og:image"]', 'meta[name="twitter:image"]'):
+            element = soup.select_one(selector)
+            if element is not None:
+                content = element.get("content")
+                if isinstance(content, str) and content.strip():
+                    return urljoin(page_url, content.strip())
+        return None
+
+    @classmethod
+    def _build_summary(
+        cls,
+        *,
+        title: str,
+        text: str,
+        category: str,
+        deadline: date | None,
+    ) -> str:
+        sentences = re.split(r"(?<=[.!?])\s+", cls._normalize_text(text))
+        selected: list[str] = []
+        for sentence in sentences:
+            if len(sentence) < 35:
+                continue
+            selected.append(sentence[:350])
+            if len(" ".join(selected)) >= 300:
+                break
+
+        summary = " ".join(selected) or (
+            f"Official {category} opportunity published by the "
+            f"University of Buea: {title}."
+        )
+        if deadline is not None:
+            summary += f" Application deadline: {deadline.isoformat()}."
+        else:
+            summary += " Consult the official document for the deadline."
+        return cls._normalize_text(summary)[:500]
+
+    @staticmethod
+    def _title_from_url(url: str) -> str:
+        stem = PurePosixPath(urlparse(url).path).stem
+        return re.sub(r"[-_]+", " ", stem).strip()
+
+    @staticmethod
+    def _normalize_text(value: str) -> str:
+        return re.sub(r"\s+", " ", value).strip()
+
+    @staticmethod
+    def _fold_text(value: str) -> str:
+        normalized = unicodedata.normalize("NFKD", value)
+        return "".join(
+            character
+            for character in normalized
+            if not unicodedata.combining(character)
+        ).lower()
+
+    @classmethod
+    def _contains_any(cls, text: str, patterns: tuple[str, ...]) -> bool:
+        folded = cls._fold_text(text)
+        return any(re.search(pattern, folded) for pattern in patterns)
+
+    @staticmethod
+    def _deduplicate_candidates(
+        candidates: list[tuple[str, str]],
+    ) -> list[tuple[str, str]]:
+        unique: dict[str, tuple[str, str]] = {}
+        for url, title in candidates:
+            key = url.rstrip("/").lower()
+            unique[key] = (url, title)
+        return list(unique.values())
+
+    @staticmethod
+    def _deduplicate_opportunities(
+        opportunities: list[CollectedOpportunity],
+    ) -> list[CollectedOpportunity]:
+        unique: dict[str, CollectedOpportunity] = {}
+        for opportunity in opportunities:
+            key = opportunity.official_url.rstrip("/").lower()
+            unique[key] = opportunity
+        return list(unique.values())
+
+    @staticmethod
+    def _build_headers() -> dict[str, str]:
+        return {
+            "User-Agent": (
+                "VerifPulseBot/1.0 (official opportunity verification; "
+                "contact@verifpulse.org)"
+            ),
+            "Accept": (
+                "text/html,application/xhtml+xml,"
+                "application/pdf;q=0.9,*/*;q=0.8"
+            ),
+            "Accept-Language": "en-GB,en;q=0.9,fr;q=0.7",
+            "Cache-Control": "no-cache",
+        }

app/services/opportunity_sync.py

@@ -59,6 +59,15 @@ SOURCE_CONFIGURATION = {
         ),
         "source_tier": "A",
     },
+    "university-buea": {
+        "name": "University of Buea",
+        "base_url": "https://www.ubuea.cm/",
+        "opportunities_url": (
+            "https://www.ubuea.cm/"
+            "index.php/announcements/"
+        ),
+        "source_tier": "A",
+    },
 }
 
 

scripts/check_university_buea_collector.py

@@ -0,0 +1,83 @@
+import asyncio
+import json
+import sys
+from dataclasses import asdict
+from datetime import date
+from typing import Any
+
+from app.collectors.providers.university_buea import (
+    UniversityBueaCollector,
+    UniversityBueaCollectorError,
+)
+
+
+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 = UniversityBueaCollector()
+
+    print(
+        "[collector-check] Starting University of Buea "
+        "opportunities check...",
+        flush=True,
+    )
+
+    opportunities = await collector.collect()
+
+    if not opportunities:
+        raise RuntimeError(
+            "University of Buea returned no current opportunity."
+        )
+
+    for opportunity in opportunities:
+        payload = serialize_value(asdict(opportunity))
+        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] University of Buea check passed: "
+        f"{len(opportunities)} opportunity collected.",
+        flush=True,
+    )
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except UniversityBueaCollectorError 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