diff --git a/.env.example b/.env.example index 8929170..eac9a77 100644 --- a/.env.example +++ b/.env.example @@ -24,6 +24,12 @@ RESEND_API_KEY="your_resend_api_key" EMAIL_ID="your_email_id" EMAIL_MIN_INTERVAL_SECONDS="5.0" # Rate limiting between emails +# Bot Webhook Notification +# VoucherBot POSTs the same voucher data it sends by email to this endpoint, +# authenticated with WEBHOOK_SECRET sent as an Authorization header. +NOTIFICATION_BOT_SERVER_URL="https://your-bot-server.example.com/webhook" +WEBHOOK_SECRET="your_webhook_secret" + # AI Module # See docs/setup.md for how to get your Gemini and Groq API keys. GEMINI_API_KEY="your_gemini_api_key" diff --git a/docs/details/configuration.md b/docs/details/configuration.md index 6f72761..2115ecd 100644 --- a/docs/details/configuration.md +++ b/docs/details/configuration.md @@ -23,6 +23,13 @@ These values are loaded from `.env` through Pydantic settings. | `EMAIL_ID` | `None` | Recipient address for voucher notifications | | `EMAIL_MIN_INTERVAL_SECONDS` | `5.0` | Minimum delay between email sends | +### Bot webhook notification + +| Variable | Default | Purpose | +|---|---:|---| +| `NOTIFICATION_BOT_SERVER_URL` | `None` | Endpoint that receives a POST with the same voucher alert data as the email, for a remote bot server | +| `WEBHOOK_SECRET` | `None` | Secret sent in the `Authorization` header of the webhook POST | + ### Reddit ingestion | Variable | Default | Purpose | diff --git a/tests/test_analyzer.py b/tests/test_analyzer.py index 716937a..9930476 100644 --- a/tests/test_analyzer.py +++ b/tests/test_analyzer.py @@ -166,22 +166,14 @@ def test_picks_with_weighted_probabilities(self) -> None: population = list(choices.call_args.args[0]) weights = list(choices.call_args.kwargs["weights"]) assert population == analyzer._GROQ_BATCH_MODELS - assert weights == [40, 40, 20] + assert set(weights) == {0.5} + assert "qwen/qwen3.6-27b" not in population def test_skips_exhausted_models_when_picking(self) -> None: - _exhaust_daily("llama-3.3-70b-versatile") - with patch( - "voucherbot.services.ai.analyzer.random.choices", - return_value=["openai/gpt-oss-20b"], - ) as choices: - picked = analyzer._pick_groq_model() + _exhaust_daily("openai/gpt-oss-20b") + picked = analyzer._pick_groq_model() - assert picked is not None - population = list(choices.call_args.args[0]) - assert "llama-3.3-70b-versatile" not in population - assert "llama-3.3-70b-versatile" not in list( - choices.call_args.kwargs["weights"] - ) + assert picked == "openai/gpt-oss-120b" def _exhaust_daily(model: str) -> None: @@ -267,6 +259,32 @@ async def test_call_groq_model_returns_parsed_event() -> None: assert settle_call.args[0] == 1 +@pytest.mark.asyncio +async def test_call_groq_model_sets_qwen_params() -> None: + client = _fake_groq_client([_groq_response()]) + with ( + patch("voucherbot.services.ai.analyzer.settings", _settings()), + patch("voucherbot.services.ai.analyzer.AsyncGroq", return_value=client), + patch( + "voucherbot.services.ai.analyzer._wait_for_groq_budget", + new=AsyncMock(return_value=1), + ), + patch("voucherbot.services.ai.analyzer._settle_groq_budget", new=AsyncMock()), + ): + result = await analyzer._call_groq_model("Title", "Content", "qwen/qwen3.6-27b") + + assert result is not None + call = client.chat.completions.create.await_args + assert call is not None + params = call.kwargs + assert params["reasoning_effort"] == "default" + assert params["reasoning_format"] == "hidden" + assert params["response_format"] == {"type": "json_object"} + assert params["temperature"] == 0.6 + assert params["top_p"] == 0.95 + assert params["max_completion_tokens"] == 2048 + + @pytest.mark.asyncio async def test_call_groq_model_skips_when_daily_exhausted() -> None: _exhaust_daily("openai/gpt-oss-20b") @@ -429,6 +447,83 @@ async def test_call_groq_returns_none_when_all_models_fail() -> None: assert result is None +# --------------------------------------------------------------------------- +# _maybe_escalate_to_qwen +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_escalate_routes_low_confidence_to_qwen() -> None: + primary = ExtractedEvent(is_voucher=True, confidence=0.4) + refined = ExtractedEvent(is_voucher=True, confidence=0.9, reason="refined") + with ( + patch( + "voucherbot.services.ai.analyzer.is_model_available", + side_effect=lambda model: True, + ), + patch( + "voucherbot.services.ai.analyzer._call_groq_model", + new=AsyncMock(return_value=refined), + ) as call_model, + ): + result = await analyzer._maybe_escalate_to_qwen("Title", "Content", primary) + + assert result is refined + call = call_model.await_args + assert call is not None + assert call.args[2] == analyzer._GROQ_REASONER_MODEL + + +@pytest.mark.asyncio +async def test_escalate_keeps_high_confidence_result() -> None: + primary = ExtractedEvent(is_voucher=True, confidence=0.9) + with patch( + "voucherbot.services.ai.analyzer._call_groq_model", + new=AsyncMock(), + ) as call_model: + result = await analyzer._maybe_escalate_to_qwen("Title", "Content", primary) + + assert result is primary + call_model.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_escalate_keeps_original_when_qwen_unavailable() -> None: + primary = ExtractedEvent(is_voucher=True, confidence=0.4) + with ( + patch( + "voucherbot.services.ai.analyzer.is_model_available", + side_effect=lambda model: False, + ), + patch( + "voucherbot.services.ai.analyzer._call_groq_model", + new=AsyncMock(), + ) as call_model, + ): + result = await analyzer._maybe_escalate_to_qwen("Title", "Content", primary) + + assert result is primary + call_model.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_escalate_keeps_original_when_qwen_fails() -> None: + primary = ExtractedEvent(is_voucher=True, confidence=0.4) + with ( + patch( + "voucherbot.services.ai.analyzer.is_model_available", + side_effect=lambda model: True, + ), + patch( + "voucherbot.services.ai.analyzer._call_groq_model", + new=AsyncMock(return_value=None), + ), + ): + result = await analyzer._maybe_escalate_to_qwen("Title", "Content", primary) + + assert result is primary + + # --------------------------------------------------------------------------- # _call_gemini # --------------------------------------------------------------------------- diff --git a/tests/test_bot_notification.py b/tests/test_bot_notification.py new file mode 100644 index 0000000..3b57783 --- /dev/null +++ b/tests/test_bot_notification.py @@ -0,0 +1,195 @@ +"""Tests for the bot webhook notification service.""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any +from unittest.mock import patch + +import httpx +import pytest + +from voucherbot.services.ai.schema import ExtractedEvent +from voucherbot.services.bot_notification import notifier + + +def _post(**kwargs: object) -> Any: + base: dict[str, object] = dict( + id=7, + title="Free AZ-900 this week", + url="https://example.com/posts/1", + summary=None, + content_hash="hash123", + ) + base.update(kwargs) + return SimpleNamespace(**base) + + +def _extracted() -> ExtractedEvent: + return ExtractedEvent( + is_voucher=True, + confidence=0.55, + vendor="microsoft", + promotion_name="AI Skills Fest", + voucher_code="MS-LEARN-50", + discount="50%", + promotion_type="voucher", + certifications=["AZ-900"], + regions=["US"], + start_date="2026-09-01", + end_date="2026-09-30", + reason="Free Microsoft exam voucher.", + ) + + +def _settings(**overrides: object) -> SimpleNamespace: + base = SimpleNamespace( + notification_bot_server_url="https://bot.example.com/webhook", + webhook_secret="super-secret", + ) + base.__dict__.update(overrides) + return base + + +def test_build_voucher_payload_shape() -> None: + payload = notifier.build_voucher_payload(_post(), _extracted()) + + assert payload["event"] == "voucher_alert" + assert payload["title"] == "Voucher: Microsoft — AI Skills Fest" + assert payload["post"] == "https://example.com/posts/1" + assert payload["claim_url"] == "https://example.com/posts/1" + assert payload["confidence"] == 0.55 + assert payload["sent_at"] + assert payload["vendor"] == "microsoft" + assert payload["promotion_name"] == "AI Skills Fest" + assert payload["promotion_type"] == "voucher" + assert payload["certifications"] == ["AZ-900"] + assert payload["voucher_code"] == "MS-LEARN-50" + assert payload["discount"] == "50%" + assert payload["regions"] == ["US"] + assert payload["start_date"] == "2026-09-01" + assert payload["end_date"] == "2026-09-30" + + +def test_build_voucher_payload_omits_nulls() -> None: + payload = notifier.build_voucher_payload(_post(), ExtractedEvent(is_voucher=True)) + + assert payload["event"] == "voucher_alert" + assert "vendor" not in payload + assert "promotion_name" not in payload + assert "voucher_code" not in payload + assert "discount" not in payload + assert "certifications" not in payload + assert "discount" not in payload + + +def test_build_voucher_payload_claim_url_falls_back_to_post() -> None: + extracted = _extracted() + extracted.registration_url = "https://aws.example.com/claim" + payload = notifier.build_voucher_payload(_post(), extracted) + + assert payload["claim_url"] == "https://aws.example.com/claim" + + +@pytest.mark.asyncio +async def test_send_skips_when_not_configured() -> None: + with patch( + "voucherbot.services.bot_notification.notifier.settings", + _settings(notification_bot_server_url=None), + ): + result = await notifier.send_bot_notification(_post(), _extracted()) + assert result is False + + +@pytest.mark.asyncio +async def test_send_skips_when_secret_missing() -> None: + with patch( + "voucherbot.services.bot_notification.notifier.settings", + _settings(webhook_secret=None), + ): + result = await notifier.send_bot_notification(_post(), _extracted()) + assert result is False + + +@pytest.mark.asyncio +async def test_send_posts_with_auth_header() -> None: + class FakeResponse: + status_code = 200 + raise_for_status = lambda self: None # noqa: E731 + + captured: dict[str, Any] = {} + + class FakeClient: + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + async def __aenter__(self) -> "FakeClient": + return self + + async def __aexit__(self, *args: Any) -> bool: + return False + + async def post( + self, url: str, *, json: Any, headers: dict[str, str] + ) -> FakeResponse: + captured["url"] = url + captured["json"] = json + captured["headers"] = headers + return FakeResponse() + + with ( + patch( + "voucherbot.services.bot_notification.notifier.settings", + _settings(), + ), + patch( + "voucherbot.services.bot_notification.notifier.httpx.AsyncClient", + FakeClient, + ), + ): + result = await notifier.send_bot_notification(_post(), _extracted()) + + assert result is True + assert captured["url"] == "https://bot.example.com/webhook" + assert captured["headers"] == { + "Authorization": "Bearer super-secret", + "Content-Type": "application/json", + } + assert captured["json"]["event"] == "voucher_alert" + assert captured["json"]["promotion_name"] == "AI Skills Fest" + + +@pytest.mark.asyncio +async def test_send_returns_false_on_http_error() -> None: + async def _fail_post(*args: Any, **kwargs: Any) -> None: + raise httpx.HTTPStatusError( + "500", + request=httpx.Request("POST", "https://bot.example.com"), + response=httpx.Response(500), + ) + + class FakeClient: + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + async def __aenter__(self) -> "FakeClient": + return self + + async def __aexit__(self, *args: Any) -> bool: + return False + + post = _fail_post + + with ( + patch( + "voucherbot.services.bot_notification.notifier.settings", + _settings(), + ), + patch( + "voucherbot.services.bot_notification.notifier.httpx.AsyncClient", + FakeClient, + ), + ): + result = await notifier.send_bot_notification(_post(), _extracted()) + + assert result is False diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 0e62d44..236268a 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -323,6 +323,10 @@ async def test_process_one_source_new_voucher_full_pipeline() -> None: "voucherbot.services.ingestion.pipeline.deliver_pending_notifications", new=AsyncMock(return_value=1), ) as deliver, + patch( + "voucherbot.services.ingestion.pipeline.send_bot_notification", + new=AsyncMock(return_value=True), + ) as bot_notify, ): stats = await pipeline._process_one_source(db, source, collector, keywords, 10) @@ -332,11 +336,13 @@ async def test_process_one_source_new_voucher_full_pipeline() -> None: assert stats["ai_analyzed"] == 1 assert stats["events_created"] == 1 assert stats["notified"] == 1 + assert stats["bot_notified"] == 1 assert db_post.ai_result == extracted.model_dump() assert db_post.status == PostStatus.PROCESSED assert source.error_count == 0 db.commit.assert_awaited() deliver.assert_awaited_once() + bot_notify.assert_awaited_once() @pytest.mark.asyncio diff --git a/voucherbot/config/settings.py b/voucherbot/config/settings.py index d9964ea..447860b 100644 --- a/voucherbot/config/settings.py +++ b/voucherbot/config/settings.py @@ -84,6 +84,12 @@ class Settings(BaseSettings): # Optional per-email Reply-To; when unset Resend falls back to the From. email_reply_to: Optional[str] = None + # Bot webhook notification (Discord-style bot server) + # Endpoint that receives a POST with the same voucher data as the email + # alert; protected by WEBHOOK_SECRET in the Authorization header. + notification_bot_server_url: Optional[str] = None + webhook_secret: Optional[str] = None + # Reddit reddit_client_id: Optional[str] = None reddit_client_secret: Optional[str] = None diff --git a/voucherbot/services/ai/analyzer.py b/voucherbot/services/ai/analyzer.py index 5271f1e..2792f77 100644 --- a/voucherbot/services/ai/analyzer.py +++ b/voucherbot/services/ai/analyzer.py @@ -7,9 +7,9 @@ canonical ``ExtractedEvent`` (defined in ``voucherbot.services.ai.schema``). Internally, providers are tried in priority order: - 1. Groq — each post routed to a model with weighted probability: - openai/gpt-oss-20b (40%), openai/gpt-oss-120b (40%), - llama-3.3-70b-versatile (20%) + 1. Groq — each post routed 50/50 across openai/gpt-oss-20b and + openai/gpt-oss-120b; low-confidence results are re-analyzed by + qwen/qwen3.6-27b (a reasoning model) 2. Gemini (final fallback on non-429 failure) Each adapter is responsible for converting its raw provider response into an @@ -103,31 +103,53 @@ # --------------------------------------------------------------------------- _MAX_RETRIES = 3 _FALLBACK_WAIT_S = 65 -# Weighted Groq routing. Each post is routed to one of these models with the -# given probability (40% / 40% / 20%). openai/gpt-oss-20b and -# openai/gpt-oss-120b both carry a 200K token/day quota; llama-3.3-70b-versatile -# carries a 100K token/day quota. +# Primary Groq routing. Each post is routed to one of these models with equal +# weight (50/50) across the two gpt-oss models. qwen/qwen3.6-27b is NOT a +# primary router: it is a reasoning model reserved for re-analyzing +# low-confidence results from the gpt-oss pair (see ``_maybe_escalate_to_qwen``). _GROQ_MODEL_WEIGHTS: dict[str, float] = { - "openai/gpt-oss-20b": 40, - "openai/gpt-oss-120b": 40, - "llama-3.3-70b-versatile": 20, + "openai/gpt-oss-20b": 0.5, + "openai/gpt-oss-120b": 0.5, } _GROQ_BATCH_MODELS: list[str] = list(_GROQ_MODEL_WEIGHTS.keys()) +# Reasoning model used to re-analyze low-confidence primary results. Kept out of +# ``_GROQ_BATCH_MODELS`` so daily-exhaustion and budget accounting treat it as +# the escalation tier rather than a primary router. +_GROQ_REASONER_MODEL = "qwen/qwen3.6-27b" + +# Primary results below this confidence trigger a qwen re-analysis. +_GROQ_LOW_CONFIDENCE_THRESHOLD = 0.6 + _GROQ_MODEL_TPM = { "openai/gpt-oss-120b": 8000, "openai/gpt-oss-20b": 8000, - "llama-3.3-70b-versatile": 12000, + "qwen/qwen3.6-27b": 8000, } _GROQ_MODEL_TPD = { "openai/gpt-oss-120b": 200_000, "openai/gpt-oss-20b": 200_000, - "llama-3.3-70b-versatile": 100_000, + "qwen/qwen3.6-27b": 200_000, } _GROQ_MODEL_RPD = { "openai/gpt-oss-120b": 1_000, "openai/gpt-oss-20b": 1_000, - "llama-3.3-70b-versatile": 1_000, + "qwen/qwen3.6-27b": 1_000, +} + +# Per-model tuning overrides. Reasoning models burn completion tokens on +# "thinking", so the default 1024 budget easily truncates the answer and the +# server-side JSON validator then rejects it (400 `failed_generation`). +# Qwen3.6 27B also performs best around temperature 0.6 (Groq docs). +_GROQ_MODEL_PARAMS = { + "qwen/qwen3.6-27b": { + "temperature": 0.6, + "top_p": 0.95, + "max_completion_tokens": 2048, + "reasoning_effort": "default", + "reasoning_format": "hidden", + "response_format": {"type": "json_object"}, + }, } # Global cap: max concurrent AI calls across all models combined. @@ -371,10 +393,9 @@ async def _call_groq_model( "max_completion_tokens": settings.groq_max_completion_tokens, "top_p": 1, } + params.update(_GROQ_MODEL_PARAMS.get(model, {})) if model.startswith("openai/gpt-oss-"): params["reasoning_effort"] = "medium" - if "llama" in model.lower(): - params["response_format"] = {"type": "json_object"} resp = await client.chat.completions.create(**params) # type: ignore[call-overload] actual = getattr(resp.usage, "total_tokens", None) or estimated_tokens @@ -418,10 +439,11 @@ async def _call_groq_model( async def _call_groq( title: str, content: str | None, source_name: str | None = None ) -> ExtractedEvent | None: - """Try Groq models weighted 40/40/20, skipping daily-exhausted ones. + """Try Groq models weighted 50/50, skipping daily-exhausted ones. The first pick follows ``_GROQ_MODEL_WEIGHTS``; on failure the remaining - available models are tried as fallback. + available models are tried as fallback. A low-confidence result from the + gpt-oss pair is then re-analyzed by the qwen reasoning model. """ tried: set[str] = set() while True: @@ -431,7 +453,44 @@ async def _call_groq( tried.add(model) result = await _call_groq_model(title, content, model, source_name) if result is not None: - return result + return await _maybe_escalate_to_qwen(title, content, result, source_name) + + +async def _maybe_escalate_to_qwen( + title: str, + content: str | None, + result: ExtractedEvent, + source_name: str | None = None, +) -> ExtractedEvent: + """Re-analyze a low-confidence result with the qwen reasoning model. + + Only primary gpt-oss results with confidence below + ``_GROQ_LOW_CONFIDENCE_THRESHOLD`` are escalated. The call is + best-effort: if qwen is unavailable or fails, the original result is + returned unchanged so a lower-quality answer never replaces a valid one. + """ + if ( + _GROQ_LOW_CONFIDENCE_THRESHOLD is None + or result.confidence >= _GROQ_LOW_CONFIDENCE_THRESHOLD + ): + return result + if not is_model_available(_GROQ_REASONER_MODEL): + logger.info( + "ai.analyzer: qwen reasoning model unavailable, keeping primary result", + model=_GROQ_REASONER_MODEL, + confidence=result.confidence, + ) + return result + + logger.info( + "ai.analyzer: escalating low-confidence result to qwen", + model=_GROQ_REASONER_MODEL, + confidence=result.confidence, + ) + refined = await _call_groq_model(title, content, _GROQ_REASONER_MODEL, source_name) + if refined is None: + return result + return refined async def _call_gemini( @@ -485,7 +544,7 @@ async def analyze_post( ) -> ExtractedEvent | None: """Extract structured promotion data from a single post. - Provider priority: Groq (primary, weighted 40/40/20) → Gemini (final fallback). + Provider priority: Groq (primary, weighted 50/50) → Gemini (final fallback). Fallback is triggered only on non-429 failures; 429s are retried within each provider. """ if settings.groq_api_key: @@ -512,9 +571,10 @@ async def analyze_post_batch( ) -> list[ExtractedEvent | None]: """Analyze multiple posts concurrently, distributing across available Groq models. - Each post is assigned a Groq model weighted 40/40/20 per - ``_GROQ_MODEL_WEIGHTS``; on per-model failure the remaining available - models are tried, then Gemini as final fallback. + Each post is assigned a Groq model weighted 50/50 across the gpt-oss pair + per ``_GROQ_MODEL_WEIGHTS``; on per-model failure the remaining available + models are tried. Low-confidence primary results are re-analyzed by the + qwen reasoning model, then Gemini is used as the final fallback. Returns results in the same order as the input list. """ if not settings.groq_api_key or not posts: @@ -538,7 +598,9 @@ async def _call_one( tried.add(model) result = await _call_groq_model(title, content, model, source_name) if result is not None: - return idx, result + return idx, await _maybe_escalate_to_qwen( + title, content, result, source_name + ) if settings.gemini_api_key: return idx, await _call_gemini(title, content, source_name) return idx, None diff --git a/voucherbot/services/bot_notification/__init__.py b/voucherbot/services/bot_notification/__init__.py new file mode 100644 index 0000000..5161f0b --- /dev/null +++ b/voucherbot/services/bot_notification/__init__.py @@ -0,0 +1,13 @@ +""" +Bot webhook notification service. +""" + +from voucherbot.services.bot_notification.notifier import ( + build_voucher_payload, + send_bot_notification, +) + +__all__ = [ + "build_voucher_payload", + "send_bot_notification", +] diff --git a/voucherbot/services/bot_notification/notifier.py b/voucherbot/services/bot_notification/notifier.py new file mode 100644 index 0000000..0f13be5 --- /dev/null +++ b/voucherbot/services/bot_notification/notifier.py @@ -0,0 +1,128 @@ +""" +Bot webhook notification service. + +POSTs the same voucher data that the email notification module sends to a +remote bot server (e.g. a Discord bot) so the recipient gets an alert there +too. The request is authenticated with ``WEBHOOK_SECRET`` in the +``Authorization`` header. + +Payload shape mirrors ``build_voucher_email`` so both channels carry the same +information: vendor, promotion, type, certifications, voucher code, discount, +regions, dates, URLs, and a confidence signal. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Any + +import httpx +import structlog + +from voucherbot.config.settings import settings + +if TYPE_CHECKING: + from voucherbot.models.post import Post + from voucherbot.services.ai.schema import ExtractedEvent + +logger = structlog.get_logger(__name__) + +_DEFAULT_TIMEOUT = 10.0 + + +def build_voucher_payload(post: Post, extracted: ExtractedEvent) -> dict[str, Any]: + """Build the JSON payload POSTed to the bot server. + + Mirrors the email notification content: + - ``title`` — human-readable alert heading like the email subject + - ``post`` — source post URL (like the email's "View source post") + - ``claim_url`` — registration URL when present, else the post URL + - one key per voucher field the AI extracted (null fields are omitted) + - ``confidence`` — AI confidence in ``is_voucher`` + - ``sent_at`` — ISO timestamp of this notification + """ + vendor = (extracted.vendor or "").strip() + promo = (extracted.promotion_name or post.title or "Voucher").strip() + subject_bits = [b for b in (vendor.title() if vendor else "", promo) if b] + title = "Voucher: " + " — ".join(subject_bits[:2]) + + claim_url = extracted.registration_url or post.url + + payload: dict[str, Any] = { + "event": "voucher_alert", + "title": title, + "post": post.url, + "claim_url": claim_url, + "confidence": extracted.confidence, + "sent_at": datetime.now(timezone.utc).isoformat(), + } + + # Non-null voucher fields, mirroring build_voucher_email's table rows. + field_map: dict[str, Any] = { + "vendor": extracted.vendor, + "promotion_name": extracted.promotion_name, + "promotion_type": extracted.promotion_type, + "certifications": extracted.certifications, + "voucher_code": extracted.voucher_code, + "discount": extracted.discount, + "regions": extracted.regions, + "start_date": extracted.start_date, + "end_date": extracted.end_date, + "reason": extracted.reason, + } + for key, value in field_map.items(): + if value not in (None, "", []): + payload[key] = value + + return payload + + +async def send_bot_notification(post: Post, extracted: ExtractedEvent) -> bool: + """POST a voucher alert to the configured bot server. + + Skips (False) when ``NOTIFICATION_BOT_SERVER_URL`` or ``WEBHOOK_SECRET`` + are not set — never raises. Returns True when the server accepts the + request (2xx). + """ + if not settings.notification_bot_server_url: + logger.warning( + "bot_notification.send: NOTIFICATION_BOT_SERVER_URL not set — skipping", + post_id=post.id, + ) + return False + if not settings.webhook_secret: + logger.warning( + "bot_notification.send: WEBHOOK_SECRET not set — skipping", + post_id=post.id, + ) + return False + + payload = build_voucher_payload(post, extracted) + headers = { + "Authorization": f"Bearer {settings.webhook_secret}", + "Content-Type": "application/json", + } + + try: + async with httpx.AsyncClient(timeout=_DEFAULT_TIMEOUT) as client: + response = await client.post( + settings.notification_bot_server_url, + json=payload, + headers=headers, + ) + response.raise_for_status() + except httpx.HTTPError as exc: + logger.warning( + "bot_notification.send: webhook POST failed", + post_id=post.id, + url=settings.notification_bot_server_url, + error=str(exc)[:160], + ) + return False + + logger.info( + "bot_notification.send: voucher alert sent to bot server", + post_id=post.id, + status_code=response.status_code, + ) + return True diff --git a/voucherbot/services/ingestion/pipeline.py b/voucherbot/services/ingestion/pipeline.py index da575bf..a704f9c 100644 --- a/voucherbot/services/ingestion/pipeline.py +++ b/voucherbot/services/ingestion/pipeline.py @@ -45,6 +45,7 @@ deliver_pending_notifications, stage_voucher_notification, ) +from voucherbot.services.bot_notification import send_bot_notification from voucherbot.services.ingestion.dedup import identity_hash, content_hash from voucherbot.services.ingestion.event_matcher import EventMatcher from voucherbot.models.event import MatchConfidence @@ -204,6 +205,7 @@ async def _process_one_source( "events_attached": 0, "possible_matches": 0, "notified": 0, + "bot_notified": 0, } # ── Stage 0: Collect ────────────────────────────────────────────────────── @@ -411,4 +413,11 @@ async def _process_one_source( if staged: stats["notified"] = await deliver_pending_notifications(db) + # Send the same voucher alert to the bot webhook alongside the email. + # Best-effort: a webhook failure never fails the pipeline or the email. + stats["bot_notified"] = 0 + for db_post, extracted in pending_notifications: + if await send_bot_notification(db_post, extracted): + stats["bot_notified"] += 1 + return stats