diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 50e0508..7732eef 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -96,7 +96,7 @@ Key obligations that apply directly to this project: git clone cd voucherbot python -m venv .venv && source .venv/bin/activate -pip install -r requirements.txt +pip install -e ".[dev]" # 2. Configure environment cp .env.example .env @@ -120,7 +120,7 @@ voucherbot/ ├── main.py # FastAPI app and lifespan ├── config/settings.py # All configuration via pydantic-settings ├── core/ # Exceptions and logging -├── database/ # Engine, init, and bootstrap +├── database/ # Engine, migrations, and bootstrap ├── models/ # SQLAlchemy ORM models ├── providers/ │ ├── base.py # BaseCollector contract @@ -129,14 +129,21 @@ voucherbot/ │ ├── reddit/ │ │ ├── client.py # Reddit API client ⚠️ │ │ └── collector.py -│ └── website/collector.py +│ ├── website/collector.py +│ ├── pearsonvue/collector.py # Pearson VUE vendor page scraper +│ └── training_provider/collector.py # Training partner page scraper ├── services/ │ ├── scheduler.py # Asyncio scheduler loop │ ├── dispatcher.py # Lease + source lifecycle +│ ├── event_consolidation.py # Periodic merge of duplicate events +│ ├── retention.py # Null out stale post content │ ├── ingestion/ # Pipeline, dedup, event matching -│ ├── ai/ # Groq + Gemini provider chain -│ └── email/ # Resend notifications -└── api/routers/ # Read-only REST endpoints +│ ├── ai/ # Groq + Gemini provider chain, AI event matcher +│ ├── email/ # Resend notifications (transactional outbox) +│ └── bot_notification/ # Voucher webhook to a bot server +└── api/ + ├── rate_limit.py # Health-endpoint rate limiter + └── routers/health.py # Read-only health endpoint ``` Files marked ⚠️ contain policy-sensitive logic. Changes to these files require extra care and a detailed explanation in the PR. @@ -161,10 +168,10 @@ Files marked ⚠️ contain policy-sensitive logic. Changes to these files requi New sources are defined in `voucherbot/database/bootstrap.py`. A source entry requires at minimum: - `name` — unique, descriptive -- `type` — one of `REDDIT`, `RSS`, `BLOG`, `EVENT`, `FORUM`, `WEBSITE`, `API` +- `type` — one of `REDDIT`, `RSS`, `BLOG`, `EVENT`, `FORUM`, `WEBSITE`, `API`, `PEARSONVUE`, `TRAINING_PROVIDER` - `base_url` - `priority_tier` — A, B, C, or D (see the scheduler table above) -- `config` — a JSONB object with `feed_url` (RSS), `article_selector` + `content_selector` (Website), or `subreddit` (Reddit) +- `config` — a JSONB object with `feed_url` (RSS), `article_selector` + `content_selector` (Website), `subreddit` (Reddit), or the vendor page type for Pearson VUE / training provider sources **Before adding a new web source:** diff --git a/README.md b/README.md index f05fa06..3d15d25 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ *Continuously monitors community and official sources for certification discounts, free exam opportunities, beta exams, and promotional campaigns.* -![Python](https://img.shields.io/badge/Python-3.10%2B-3776AB?logo=python&logoColor=white) +![Python](https://img.shields.io/badge/Python-3.11%2B-3776AB?logo=python&logoColor=white) ![License](https://img.shields.io/badge/License-see%20LICENSE-lightgrey) ![Deploy](https://img.shields.io/badge/Deploy-Render-46E3B7?logo=render&logoColor=white) ![Status](https://img.shields.io/badge/status-active-success) @@ -23,6 +23,8 @@ Instead of going through the entire setup and hosting it yourself, you can now * Just head over to **[voucherbot-preview.pages.dev/#notifications](https://voucherbot-preview.pages.dev/#notifications)** to learn all about it and get it set up in minutes. +The code for the Discord and Telegram bots lives in the separate [Notification-Bot](https://github.com/Devathmaj/Notification-Bot) repository, feel free to check it out. + --- ## Table of Contents diff --git a/Sources/source.md b/Sources/source.md index 1c46ac9..721bd8d 100644 --- a/Sources/source.md +++ b/Sources/source.md @@ -2,7 +2,7 @@ Human-readable reference for all official ingestion sources. The **authoritative runtime catalog** is [`voucherbot/database/bootstrap.py`](../voucherbot/database/bootstrap.py), which seeds the database on app startup. -Policy reference: [`deep-research-report (1).md`](../deep-research-report%20(1).md). Collectors prefer RSS/APIs, identify as `VoucherBot`, obey `robots.txt` / Crawl-delay, and skip sources marked `unsupported` (ToS bans HTML scraping). +Collectors prefer RSS/APIs, identify as `VoucherBot`, obey `robots.txt` / Crawl-delay, and skip sources marked `unsupported` (ToS bans HTML scraping). ## Files @@ -27,7 +27,8 @@ Policy reference: [`deep-research-report (1).md`](../deep-research-report%20(1). | Collector | Items requested | |-----------|-----------------| | Reddit | 25 (`REDDIT_FETCH_LIMIT`) | -| RSS / Website | 25 | +| RSS / Website / Pearson VUE / Training Provider | 10 | +| Curated voucher pages (`note_selector`) | 50 | Reddit is collected from public RSS feeds by default. The `REDDIT_INGESTION_ENABLED` flag in `.env` (default `false`) gates only the OAuth API: when `false`, the OAuth API is never called and posts come from the RSS feeds. @@ -40,6 +41,7 @@ These remain in the catalog but are `enabled=false` / `unsupported=true` (RSS al - ISC2 Insights - Red Hat Training Specials - AWS Events / re:Invent pages +- The Register (feed blocked by a proof-of-work challenge) ## Verify sources diff --git a/docs/details/architecture.md b/docs/details/architecture.md index 7211385..05ff5f5 100644 --- a/docs/details/architecture.md +++ b/docs/details/architecture.md @@ -11,7 +11,7 @@ VoucherBot is an async Python service that monitors certification-related source The application entry point is [voucherbot/main.py](../../voucherbot/main.py). During startup it: 1. configures logging, -2. creates tables and seeds source/keyword data when `IS_PROD=false`, +2. applies Alembic migrations and seeds source/keyword data when `IS_PROD=false`, 3. resets all sources to be due again, and 4. starts the scheduler task. @@ -45,8 +45,10 @@ The main implementation areas are: - [voucherbot/services/dispatcher.py](../../voucherbot/services/dispatcher.py) — lease handling, due-source selection, success/failure state updates - [voucherbot/services/ingestion/pipeline.py](../../voucherbot/services/ingestion/pipeline.py) — end-to-end per-source pipeline - [voucherbot/services/ingestion/event_matcher.py](../../voucherbot/services/ingestion/event_matcher.py) — canonical event matching and field merging +- [voucherbot/services/event_consolidation.py](../../voucherbot/services/event_consolidation.py) — periodic merge of duplicate canonical events - [voucherbot/services/ai/analyzer.py](../../voucherbot/services/ai/analyzer.py) — AI extraction provider chain and batching -- [voucherbot/api/routers](../../voucherbot/api/routers) — read-only HTTP endpoints for sources, posts, alerts, and health +- [voucherbot/services/ai/event_matcher_ai.py](../../voucherbot/services/ai/event_matcher_ai.py) — qwen-based same-promotion judge +- [voucherbot/api/routers/health.py](../../voucherbot/api/routers/health.py) — read-only health endpoint with rate limiting ## Scheduler and dispatcher @@ -98,20 +100,32 @@ New or updated posts are sent to [voucherbot/services/ai/analyzer.py](../../vouc ### 5. Event matching -The matcher in [voucherbot/services/ingestion/event_matcher.py](../../voucherbot/services/ingestion/event_matcher.py) compares extracted fields against existing active events. It uses a weighted score with thresholds for: +The matcher in [voucherbot/services/ingestion/event_matcher.py](../../voucherbot/services/ingestion/event_matcher.py) decides whether an extracted promotion is the same real-world promotion as an existing active event. -- registration URL -- voucher code -- promotion name similarity -- vendor -- certification overlap -- date overlap +By default it runs the incoming promotion through the qwen reasoning model ([voucherbot/services/ai/event_matcher_ai.py](../../voucherbot/services/ai/event_matcher_ai.py)), comparing it against the candidate events that the deterministic weighted score flags as possible matches (score >= `possible_match_threshold`, capped by `ai_candidate_limit`) and letting the model decide whether each is the same promotion: + +- `is_same_promotion` and `confidence >= ai_auto_merge_confidence` → `AUTO_MERGED` +- `is_same_promotion` and `confidence >= ai_possible_match_confidence` → `POSSIBLE_MATCH` +- otherwise → `NEW` + +When the model is unavailable, no `GROQ_API_KEY` is configured, or no candidates exist, the matcher falls back to the legacy weighted score over registration URL, voucher code, promotion-name similarity, vendor, discount, promotion type, certification overlap, and date overlap. The model's `reason` is recorded in `merge_log` for auditability. The result is one of `AUTO_MERGED`, `POSSIBLE_MATCH`, or `NEW`, and the matcher may merge fields into the canonical event while appending to `merge_log`. ### 6. Email notification -If the AI extraction yields a voucher candidate and the event decision is not `AUTO_MERGED`, the notification service sends an email through Resend. The post is marked `is_notified` only after the send succeeds. +If the AI extraction yields a voucher candidate and the event decision is not `AUTO_MERGED`, delivery intent is staged into the transactional notification outbox in the same commit as the pipeline. Delivery is attempted immediately through Resend with a stable idempotency key; failures stay `PENDING` and are retried by the scheduler. The post is marked `is_notified` only after a send succeeds. The same payload is POSTed to the optional bot server webhook alongside the email (best-effort — a webhook failure never fails the pipeline). + +## Event consolidation + +Two posts describing the same promotion can become separate events when their sources were processed at different times — the ingestion-time matcher only sees candidates that already exist at that moment. The consolidation sweep in [voucherbot/services/event_consolidation.py](../../voucherbot/services/event_consolidation.py) fixes this retroactively. It runs after every scheduler sweep (throttled by `settings.consolidation.interval_minutes`) and is cross-instance serialised with a Postgres advisory transaction lock. + +1. **Discover** — active events are grouped into candidate pairs sharing a cheap identity signal: normalised registration URL, voucher code (case-normalised), or vendor. Pairs are deduplicated by the canonical `(min_id, max_id)` key and capped by `max_pairs_per_sweep`; buckets are sampled to bound quadratic work. +2. **Gate** — each pair is scored with the same deterministic weighted score used at ingestion; only pairs at or above `possible_match_threshold` proceed. +3. **Confirm** — when a Groq key is configured, qwen is asked whether the pair is the same real-world promotion via `compare_events` (the same judge used by the matcher). A `same` decision at `confidence >= ai_possible_match_confidence` merges; otherwise the pair is kept separate. A model outage falls back to the deterministic score at or above `deterministic_auto_merge_threshold`. +4. **Merge** — the pair's survivor is the event with more posts (ties keep the older event). The absorbed event's fields are folded in through the same `_merge_fields` source-priority machinery, its posts are re-pointed to the survivor, both `merge_log` entries are appended, and the absorbed event is set to `ARCHIVED`. + +An absorbed event is never folded into a second target within one sweep, and the whole job never raises — failures are logged so the scheduler loop stays healthy. ## Data model summary @@ -121,6 +135,8 @@ The core SQLAlchemy models are: - [voucherbot/models/post.py](../../voucherbot/models/post.py) — `Post`, `PostStatus`, `VoucherPost` - [voucherbot/models/event.py](../../voucherbot/models/event.py) — `Event`, `EventStatus`, `MatchConfidence` - [voucherbot/models/keyword.py](../../voucherbot/models/keyword.py) — keyword scoring rows used by the pipeline +- [voucherbot/models/vendor_mapping.py](../../voucherbot/models/vendor_mapping.py) — URL/source-name pattern → vendor lookup +- [voucherbot/models/notification.py](../../voucherbot/models/notification.py) — notification outbox for voucher alert emails - [voucherbot/models/pipeline_lock.py](../../voucherbot/models/pipeline_lock.py) — pipeline lease row used by the dispatcher The important relationships are: @@ -131,13 +147,9 @@ The important relationships are: ## API surface -The FastAPI routes are intentionally read-only and do not implement authentication: +The FastAPI app exposes a single read-only endpoint and does not implement authentication: -- `GET /health` — simple liveness endpoint -- `GET /ready` — DB reachability probe -- `GET /sources` — list sources with optional filters by type or enabled state -- `GET /posts` — list posts with optional filters by status, source type, and minimum score -- `GET /alerts` — list AI-confirmed voucher candidates from the `voucher_posts` view +- `GET /health` — liveness + DB reachability probe (rate-limited per IP) ## Configuration and deployment diff --git a/docs/details/configuration.md b/docs/details/configuration.md index 2115ecd..28029b4 100644 --- a/docs/details/configuration.md +++ b/docs/details/configuration.md @@ -12,6 +12,7 @@ These values are loaded from `.env` through Pydantic settings. |---|---:|---| | `DATABASE_URL` | required | Async SQLAlchemy connection string for PostgreSQL | | `IS_PROD` | `false` | When `true`, startup skips schema/bootstrap work and assumes the database is already prepared | +| `IS_TEST` | `false` | When `true`, seeds a `website:local_test` source pointing at `http://localhost:35926/` for end-to-end pipeline testing | | `LOG_LEVEL` | `INFO` | Logging level used by the application | ### Email @@ -21,8 +22,16 @@ These values are loaded from `.env` through Pydantic settings. | `RESEND_API_KEY` | `None` | API key for Resend-based email delivery | | `EMAIL_FROM` | `VoucherBot ` | Sender address used for alerts | | `EMAIL_ID` | `None` | Recipient address for voucher notifications | +| `EMAIL_REPLY_TO` | `None` | Optional per-email Reply-To; when unset Resend falls back to the From address | | `EMAIL_MIN_INTERVAL_SECONDS` | `5.0` | Minimum delay between email sends | +### API rate limiting + +| Variable | Default | Purpose | +|---|---:|---| +| `HEALTH_RATE_LIMIT_PER_MINUTE` | `60` | Max `/health` requests per IP per minute; `0` disables the limit | +| `RATE_LIMIT_TRUSTED_PROXIES` | `[]` | Comma-separated proxy IPs whose `X-Forwarded-For` values are trusted for rate limiting | + ### Bot webhook notification | Variable | Default | Purpose | @@ -59,6 +68,7 @@ These values are loaded from `.env` through Pydantic settings. | `TICK_JOB_TIMEOUT_SECONDS` | `None` | Optional timeout for scheduler jobs | | `SOURCE_BACKOFF_BASE_MINUTES` | `5` | Base delay used for recoverable source failures | | `SOURCE_BACKOFF_MAX_MINUTES` | `360` | Maximum backoff delay for a source | +| `CONTENT_RETENTION_DAYS` | `7` | Posts older than this are content-purged each scheduler sweep | ### AI providers @@ -75,33 +85,58 @@ These values are loaded from `.env` through Pydantic settings. Some settings are not loaded from `.env` directly. They are defined in code and can be overridden in tests or custom runtime wiring. -### Event matching weights +### Event matching These are defined in the `EventMatcherConfig` model: | Setting | Default | Purpose | |---|---:|---| -| `weight_registration_url` | `50` | Score weight for exact registration URL matches | -| `weight_voucher_code` | `40` | Score weight for exact voucher-code matches | -| `weight_promotion_name` | `20` | Score weight for promotion-name similarity | -| `weight_vendor` | `15` | Score weight for vendor matches | -| `weight_certifications` | `15` | Score weight for certification overlap | -| `weight_date_overlap` | `10` | Score weight for date-range overlap | -| `auto_merge_threshold` | `75` | Threshold above which an event is auto-merged | -| `possible_match_threshold` | `60` | Threshold above which a possible match is flagged | -| `name_similarity_threshold` | `0.60` | Similarity cutoff for promotion-name credit | +| `use_ai_matcher` | `True` | When enabled, the qwen reasoning model decides whether an incoming promotion matches an existing event | +| `ai_candidate_limit` | `5` | Maximum deterministic-matched candidates submitted to the model per post | +| `ai_auto_merge_confidence` | `0.8` | Model confidence above which a same-promotion decision is an AUTO_MERGED | +| `ai_possible_match_confidence` | `0.5` | Model confidence below which a same-promotion decision is treated as a new event | +| `weight_registration_url` | `50` | Deterministic-fallback score weight for exact registration URL matches | +| `weight_voucher_code` | `40` | Deterministic-fallback score weight for exact voucher-code matches | +| `weight_promotion_name` | `25` | Deterministic-fallback score weight for promotion-name similarity | +| `weight_vendor` | `20` | Deterministic-fallback score weight for vendor matches | +| `weight_discount` | `20` | Deterministic-fallback score weight for discount matches | +| `weight_promotion_type` | `10` | Deterministic-fallback score weight for promotion-type matches | +| `weight_certifications` | `15` | Deterministic-fallback score weight for certification overlap | +| `weight_date_overlap` | `10` | Deterministic-fallback score weight for date-range overlap | +| `auto_merge_threshold` | `70` | Deterministic-fallback threshold above which an event is auto-merged | +| `possible_match_threshold` | `45` | Deterministic-fallback threshold above which a possible match is flagged | +| `name_similarity_threshold` | `0.60` | Deterministic-fallback similarity cutoff for promotion-name credit | +| `candidate_limit` | `100` | Maximum candidate events retrieved for matching | + +The deterministic weighted score is only used as a fallback when the qwen model is unavailable, no `GROQ_API_KEY` is configured, or no candidates exist. + +### Event consolidation + +These are defined in the `EventConsolidationConfig` model and tune the periodic sweep that merges duplicate canonical events ([voucherbot/services/event_consolidation.py](../../voucherbot/services/event_consolidation.py)): + +| Setting | Default | Purpose | +|---|---:|---| +| `enabled` | `True` | Master switch for the consolidation sweep | +| `interval_minutes` | `60` | Minimum wall-clock time between sweeps (rate-limits the qwen spend) | +| `max_pairs_per_sweep` | `1000` | Hard cap on candidate pairs examined per sweep | +| `max_ai_calls_per_sweep` | `25` | How many qwen confirmations to allow per sweep | +| `deterministic_auto_merge_threshold` | `70` | Deterministic-score floor for merging when the model is unavailable | + +The sweep runs after each scheduler sweep, groups active events by normalised registration URL, voucher code, or vendor, gates pairs with the deterministic weighted score (`possible_match_threshold`), and lets qwen confirm whether each pair is the same real-world promotion before merging and archiving the loser. ### Source priority ordering The `SOURCE_PRIORITY` list defines how source types are ranked when merging event fields: 1. `WEBSITE` -2. `EVENT` -3. `BLOG` -4. `RSS` -5. `FORUM` -6. `REDDIT` -7. `API` +2. `PEARSONVUE` +3. `TRAINING_PROVIDER` +4. `EVENT` +5. `BLOG` +6. `RSS` +7. `FORUM` +8. `REDDIT` +9. `API` Higher-priority sources overwrite lower-priority values when a new post updates an existing event. diff --git a/docs/details/detailed-summary.md b/docs/details/detailed-summary.md index 04f68be..bff2d6c 100644 --- a/docs/details/detailed-summary.md +++ b/docs/details/detailed-summary.md @@ -37,11 +37,7 @@ This flow is shared by every source type. The pipeline is deliberately provider- ```text FastAPI process (uvicorn) ├─ REST API -│ ├─ /health -│ ├─ /ready -│ ├─ /sources -│ ├─ /posts -│ └─ /alerts +│ └─ /health (rate-limited liveness + DB probe) └─ Background scheduler └─ sweep → dispatch_tick → pipeline → sleep @@ -70,6 +66,8 @@ voucherbot/ │ ├── post.py │ ├── event.py │ ├── keyword.py +│ ├── vendor_mapping.py +│ ├── notification.py │ └── pipeline_lock.py ├── providers/ │ ├── base.py @@ -78,25 +76,31 @@ voucherbot/ │ ├── reddit/ │ │ ├── client.py │ │ └── collector.py -│ └── website/collector.py +│ ├── website/collector.py +│ ├── pearsonvue/collector.py +│ └── training_provider/collector.py ├── services/ │ ├── scheduler.py │ ├── dispatcher.py +│ ├── event_consolidation.py +│ ├── retention.py │ ├── ingestion/ │ │ ├── pipeline.py │ │ ├── dedup.py │ │ └── event_matcher.py │ ├── ai/ │ │ ├── analyzer.py +│ │ ├── event_matcher_ai.py │ │ └── schema.py -│ └── email/ -│ ├── sender.py -│ └── notifications.py -└── api/routers/ - ├── health.py - ├── sources.py - ├── posts.py - └── alerts.py +│ ├── email/ +│ │ ├── sender.py +│ │ └── notifications.py +│ └── bot_notification/ +│ └── notifier.py +└── api/ + ├── rate_limit.py + └── routers/ + └── health.py ``` --- @@ -125,6 +129,7 @@ Each sweep calls `dispatch_tick` repeatedly until it returns `idle`. The loop is A source is eligible when: - `enabled = true` +- config is not marked `unsupported` - `next_due_at IS NULL OR next_due_at <= now()` - `backoff_until IS NULL OR backoff_until <= now()` - Reddit sources are excluded when `reddit_ingestion_enabled = false` @@ -230,20 +235,19 @@ The structured fields include: ### Stage 4 — Event matching -The matcher in [voucherbot/services/ingestion/event_matcher.py](../../voucherbot/services/ingestion/event_matcher.py) compares extracted fields against existing active events. It uses a weighted score with thresholds for: +The matcher in [voucherbot/services/ingestion/event_matcher.py](../../voucherbot/services/ingestion/event_matcher.py) decides whether an extracted promotion is the same real-world promotion as an existing active event. By default it asks the qwen reasoning model ([voucherbot/services/ai/event_matcher_ai.py](../../voucherbot/services/ai/event_matcher_ai.py)) to compare the incoming promotion against the candidate events that the deterministic weighted score flags as possible matches (score >= `possible_match_threshold`, capped by `ai_candidate_limit`): -- registration URL -- voucher code -- promotion name similarity -- vendor -- certification overlap -- date overlap +- `is_same_promotion` and `confidence >= ai_auto_merge_confidence` → `AUTO_MERGED` +- `is_same_promotion` and `confidence >= ai_possible_match_confidence` → `POSSIBLE_MATCH` +- otherwise → `NEW` + +When the model is unavailable, no `GROQ_API_KEY` is configured, or no candidates exist, it falls back to the legacy weighted score over registration URL, voucher code, promotion-name similarity, vendor, discount, promotion type, certification overlap, and date overlap. The result is one of `AUTO_MERGED`, `POSSIBLE_MATCH`, or `NEW`. ### Stage 5 — Email notification -If the AI extraction yields a voucher candidate and the event decision is not `AUTO_MERGED`, the notification service sends an email through Resend. The post is marked `is_notified` only after the send succeeds. +If the AI extraction yields a voucher candidate and the event decision is not `AUTO_MERGED`, the pipeline stages a delivery intent into the transactional outbox (`notification_outbox`) before the final commit. The scheduler then delivers PENDING rows through Resend with a stable idempotency key; the post is marked `is_notified` only after the send succeeds. The same payload is POSTed to the optional bot server webhook alongside the email. --- @@ -271,6 +275,16 @@ All collectors implement a common contract around normalized posts. - can extract structured notes from curated voucher pages, - skips sources that are blocked by policy. +### Pearson VUE collector + +- scrapes official vendor pages (AWS, Microsoft, Cisco, CompTIA, etc.) for exam promotions, +- extracts slide/promo/card content with per-vendor heuristics. + +### Training provider collector + +- scrapes training partner promotion pages (Global Knowledge, Ascendient, etc.), +- supports per-vendor extractors and generic card/heading fallbacks. + ### HTTP policy All HTTP traffic goes through a polite request layer that checks `robots.txt`, enforces crawl delays, and uses an identifying user-agent. This keeps the service aligned with site policies while still allowing broad ingestion. @@ -283,7 +297,8 @@ The AI analyzer uses a provider chain anchored around Groq and Gemini. ### Provider chain -- Groq models are tried first, +- Groq models are tried first — each post is routed 50/50 across `openai/gpt-oss-20b` and `openai/gpt-oss-120b`, +- low-confidence primary results are re-analyzed by the qwen reasoning model (`qwen/qwen3.6-27b`), - the first successful response wins, - Gemini is used as the final fallback, - retries are applied for rate-limit errors, @@ -305,7 +320,7 @@ Per-model rate limiting tracks requests and token budgets. The system also uses |---|---|---| | id | integer PK | | | name | string UNIQUE | | -| type | enum | REDDIT, RSS, BLOG, EVENT, FORUM, WEBSITE, API | +| type | enum | REDDIT, RSS, BLOG, EVENT, FORUM, WEBSITE, API, PEARSONVUE, TRAINING_PROVIDER | | base_url | string | | | enabled | boolean | false skips the scheduler | | priority | integer | higher processed first within tier | @@ -331,7 +346,7 @@ Per-model rate limiting tracks requests and token budgets. The system also uses | summary | text | optional short description | | author | string | | | published_at | timestamptz | | -| status | enum | QUEUED / FILTERED / PROCESSED | +| status | enum | NEW / FILTERED / QUEUED / PROCESSING / PROCESSED / NOTIFIED / FAILED | | score | integer | keyword score | | raw_data | JSONB | original collection payload | | vendor | string (nullable) | resolved from vendor_mappings table | @@ -382,10 +397,25 @@ URL patterns are checked first (startswith match against post URL), then source | Column | Type | Notes | |---|---|---| | id | integer PK | | -| url_pattern | string (nullable) | base URL prefix for startswith matching | +| url_pattern | string (nullable, unique) | base URL prefix for startswith matching | | source_name_pattern | string (nullable, unique) | lowercase substring pattern for source name | | vendor | string (not null) | canonical vendor name (e.g. "aws", "microsoft") | +### notification_outbox + +Transactional outbox for voucher alert emails. Delivery intent is persisted in the same transaction as the pipeline; a background sweep retries PENDING rows until SENT using a stable `idempotency_key` so replays cannot duplicate emails. + +| Column | Type | Notes | +|---|---|---| +| id | integer PK | | +| post_id | integer FK | linked post | +| idempotency_key | string UNIQUE | stable per (post, content) | +| status | enum | PENDING / SENT / FAILED | +| attempts | integer | delivery attempt counter | +| last_error | string | last failure reason | +| last_attempt_at / sent_at | timestamptz | delivery timing | +| subject / html_body / text_body | string/text | rendered email snapshot | + ### voucher_posts view This read-only view exposes AI-confirmed vouchers for the alerts API. It flattens the AI JSON payload into columns suitable for simple list endpoints. The `vendor` column now comes directly from `posts.vendor` (resolved from `vendor_mappings`) rather than the AI guess in `ai_result->>'vendor'`. @@ -394,21 +424,17 @@ This read-only view exposes AI-confirmed vouchers for the alerts API. It flatten ## REST API -All routes are read-only. No authentication is implemented. +The API is intentionally minimal and read-only, with a single rate-limited health endpoint: | Method | Path | Description | |---|---|---| -| GET | `/health` | Returns service status | -| GET | `/ready` | Executes `SELECT 1` and reports DB reachability | -| GET | `/sources` | Lists sources, optionally filtered by type or enabled state | -| GET | `/posts` | Lists posts, optionally filtered by status, source type, and minimum score | -| GET | `/alerts` | Lists AI-confirmed voucher candidates from the `voucher_posts` view | +| GET | `/health` | Returns service status and checks DB reachability via `SELECT count(*) FROM sources` | --- ## Email notifications -The notification layer uses Resend and sends both HTML and plain-text emails containing voucher details such as vendor, promotion name, certification list, discount, voucher code, registration URL, and dates. A post is marked as notified only after the provider confirms acceptance. +The notification layer uses Resend and sends both HTML and plain-text emails containing voucher details such as vendor, promotion name, certification list, discount, voucher code, registration URL, and dates. Delivery is staged through a transactional outbox (`notification_outbox`) in the same transaction as the pipeline run, then delivered by the scheduler with a stable idempotency key so retries can never duplicate an email. The post is marked `is_notified` only after the provider confirms acceptance. The same voucher payload is also POSTed to an optional bot server webhook (`NOTIFICATION_BOT_SERVER_URL`). --- @@ -432,11 +458,15 @@ All settings are loaded from `.env` through `pydantic-settings`. |---|---|---| | `DATABASE_URL` | required | Asyncpg connection string | | `IS_PROD` | `false` | Skip DB init/bootstrap on startup | +| `IS_TEST` | `false` | Seed a localhost test source for end-to-end pipeline testing | | `LOG_LEVEL` | `INFO` | Logging level | | `RESEND_API_KEY` | — | Email sending | | `EMAIL_FROM` | `VoucherBot ` | Sender address | | `EMAIL_ID` | — | Recipient address for alerts | +| `EMAIL_REPLY_TO` | — | Optional per-email Reply-To | | `EMAIL_MIN_INTERVAL_SECONDS` | `5.0` | Throttle between sends | +| `NOTIFICATION_BOT_SERVER_URL` | — | Webhook endpoint that receives the same voucher alert data | +| `WEBHOOK_SECRET` | — | Secret sent in the `Authorization` header of the webhook POST | | `REDDIT_CLIENT_ID` | — | Reddit API credentials | | `REDDIT_CLIENT_SECRET` | — | Reddit API credentials | | `REDDIT_USER_AGENT` | — | Reddit API credentials | @@ -450,9 +480,11 @@ All settings are loaded from `.env` through `pydantic-settings`. | `SCRAPER_RESPECT_ROBOTS` | `true` | Obey robots.txt | | `SCRAPER_MIN_DELAY_SECONDS` | `2.0` | Minimum per-host crawl delay | | `SCRAPER_USER_AGENT` | — | Override default UA string | +| `HEALTH_RATE_LIMIT_PER_MINUTE` | `60` | /health requests per IP per minute (0 disables) | | `TICK_LEASE_TTL_SECONDS` | `21600` | Pipeline lease TTL | | `SOURCE_BACKOFF_BASE_MINUTES` | `5` | Backoff base for failures | | `SOURCE_BACKOFF_MAX_MINUTES` | `360` | Backoff ceiling | +| `CONTENT_RETENTION_DAYS` | `7` | Posts older than this are content-purged on each sweep | --- diff --git a/docs/details/project-info.md b/docs/details/project-info.md index f418477..4009b12 100644 --- a/docs/details/project-info.md +++ b/docs/details/project-info.md @@ -160,32 +160,30 @@ The core database model is split across several tables and views: - `posts` stores ingested content and the AI analysis result alongside deduplication fields. - `events` stores canonical promotional events and merge log history. - `keywords` stores the keyword scoring catalog. +- `vendor_mappings` resolves a source URL/name to its canonical vendor. +- `notification_outbox` is the transactional outbox for voucher alert emails. - `pipeline_lock` is used for the dispatcher lease. -- `voucher_posts` is a view used by the alerts endpoint. +- `voucher_posts` is a read-only view of AI-confirmed vouchers. The important relationship is that many posts can reference the same canonical event, while the posts themselves remain distinct and are never merged. ### REST API surface -The API is intentionally read-only and does not implement authentication. The main routes are: +The API is intentionally minimal and read-only, and does not implement authentication. The single route is: - `GET /health` -- `GET /ready` -- `GET /sources` -- `GET /posts` -- `GET /alerts` -These routes expose source state, ingested posts, and AI-confirmed voucher candidates. +This endpoint reports service liveness and database reachability, and is rate-limited per client IP. ### Email and notifications -The notification flow is implemented through the email service and Resend integration. Alerts contain voucher details such as vendor, promotion name, certification list, discount, voucher code, registration URL, and date information. +The notification flow is implemented through the email service and Resend integration using a transactional outbox. Alerts contain voucher details such as vendor, promotion name, certification list, discount, voucher code, registration URL, and date information. The same voucher payload can also be POSTed to a remote bot server webhook. ### Database and deployment details The project uses SQLAlchemy async with `asyncpg`, Alembic migrations, and Docker-based local deployment. The startup path differs by environment: -- in non-production mode, the app creates tables and seeds the source catalog, +- in non-production mode, the app applies Alembic migrations and seeds the source catalog, - in production mode, the app assumes the schema and seed data already exist and uses a DML-only role. The repository also includes Render deployment configuration and a source-catalog verification script for smoke testing ingestion sources. diff --git a/docs/details/schema.md b/docs/details/schema.md index a5cf8df..7602bc9 100644 --- a/docs/details/schema.md +++ b/docs/details/schema.md @@ -1,6 +1,6 @@ # Database schema -**Current revision:** `j0k1l2m3n4o5` +**Current revision:** `o9p8q7r6s5t4` Apply with: `alembic upgrade head` ## Objects @@ -13,12 +13,16 @@ Apply with: `alembic upgrade head` | `keywords` | table | Keyword scoring catalog | | `vendor_mappings` | table | URL/source-name pattern → vendor lookup | | `pipeline_lock` | table | Dispatcher lease | +| `notification_outbox` | table | Transactional outbox for voucher alert emails | | `alembic_version` | table | Migration pointer | | `voucher_posts` | **view** | AI-confirmed vouchers only (`is_voucher` + `PROCESSED`) | ## Enums -- `sourcetype`, `poststatus`, `eventstatus` +- `sourcetype` (includes `PEARSONVUE`, `TRAINING_PROVIDER`) +- `poststatus` +- `eventstatus` +- `notificationstatus` ## Prod rule diff --git a/docs/details/testing.md b/docs/details/testing.md index 6012950..41fd8fb 100644 --- a/docs/details/testing.md +++ b/docs/details/testing.md @@ -120,7 +120,7 @@ pytest -v A typical test run will produce output similar to: ```text -341 passed, 15 skipped, 1 warning in 6.29s +418 passed, 15 skipped, 1 warning in 6.84s ``` ### ✅ Passed @@ -160,23 +160,28 @@ The suite is organised by module — each file targets one service, provider, or | Test file (tests) | Module under test | Highlights | |-------------------|-------------------|------------| -| `test_analyzer.py` (35) | `voucherbot/services/ai/analyzer.py` | JSON extraction parsing (plain/fenced/invalid → safe default), token estimation, Groq/Gemini rate budgets and daily exhaustion, 429 retry handling, model fallback order, `analyze_post_batch` order preservation | -| `test_bootstrap.py` (24) | `voucherbot/database/bootstrap.py` | Reddit tier/cadence rules, invalid-selector warnings, transient-error detection, retry-with-backoff (and no retry on `IntegrityError`), keyword seeding, bootstrap ordering + advisory-lock skip | +| `test_analyzer.py` (43) | `voucherbot/services/ai/analyzer.py` | JSON extraction parsing (plain/fenced/invalid → safe default), token estimation, Groq/Gemini rate budgets and daily exhaustion, 429 retry handling, model fallback order, qwen low-confidence escalation, `analyze_post_batch` order preservation | +| `test_bootstrap.py` (25) | `voucherbot/database/bootstrap.py` | Reddit tier/cadence rules, invalid-selector warnings, transient-error detection, retry-with-backoff (and no retry on `IntegrityError`), keyword seeding, bootstrap ordering + advisory-lock skip | | `test_pipeline.py` (29) | `voucherbot/services/ingestion/pipeline.py` | URL normalisation, vendor/collector resolution, fetch-limit resolution, `_process_one_source` state machine (keyword filter → dedup → AI → match → outbox → delivery) | | `test_pearsonvue_collector.py` (20) | `voucherbot/providers/pearsonvue/collector.py` | Slide/promo/card extraction, card dedup, URL resolution, robots/401/403/429/timeout paths, fetch limits | | `test_training_provider_collector.py` (18) | `voucherbot/providers/training_provider/collector.py` | GK/Ascendient/generic extractors, nav-link exclusion, description fallbacks, relative-URL resolution, error/limit paths | | `test_settings.py` (10) | `voucherbot/config/settings.py` | Hermetic pydantic-settings construction: defaults, empty-string→`None` validators, trusted-proxy lists, `EventMatcherConfig`, stable `SOURCE_PRIORITY` order | | `test_email_sender.py` (8) | `voucherbot/services/email/sender.py` | `send_email` params/reply-to/idempotency key, init state, `send_test_email` skip/send, skip-when-uninitialised | | `test_init_db.py` (5) | `voucherbot/database/init_db.py` | Source-type enum migration (add-only), `create_all` excluding view models | -| `test_collectors.py` (64) | `voucherbot/providers/{rss,website}/collector.py` | Feed-URL normalisation, HTML/content-type rejection for RSS, mocked `polite_get` collection, UA identification, per-type skips | +| `test_collectors.py` (67) | `voucherbot/providers/{rss,website}.collector` | Feed-URL normalisation, HTML/content-type rejection for RSS, mocked `polite_get` collection, UA identification, per-type skips | | `test_dedup.py` (21) | `voucherbot/services/ingestion/dedup.py` | URL canonicalisation (tracking params, fragments, scheme), content/identity hashing, batch deduplication | -| `test_dispatcher.py` (21) | `voucherbot/services/dispatcher.py` | Backoff growth/cap, poll-interval resolution, tick lifecycle (busy/idle/ran/failed), due-source selection | -| `test_event_matcher.py` (74) | `voucherbot/services/ingestion/event_matcher.py` | Shared-event scenarios (two posts → one event), possible matches, event updates | +| `test_dispatcher.py` (22) | `voucherbot/services/dispatcher.py` | Backoff growth/cap, poll-interval resolution, tick lifecycle (busy/idle/ran/failed), due-source selection, unrecoverable-error disables | +| `test_event_matcher.py` (82) | `voucherbot/services/ingestion/event_matcher.py` | Shared-event scenarios (two posts → one event), possible matches, event updates, source-priority field merging | +| `test_event_matcher_ai.py` (14) | `voucherbot/services/ai/event_matcher_ai.py` | `EventMatchDecision` parsing, prompt/serialisation helpers, `compare_candidate`/`compare_events` fallback semantics | +| `test_bot_notification.py` (7) | `voucherbot/services/bot_notification/notifier.py` | Webhook payload builder, `Authorization` header, skip-when-unconfigured, HTTP error handling | +| `test_event_consolidation.py` (24) | `voucherbot/services/event_consolidation.py` | Candidate-pair discovery, deterministic gating, AI/deterministic merge decisions, survivor selection, throttling | | `test_email_notifications.py` (4) | `voucherbot/services/email/notifications.py` | Safe-URL allow/deny list, voucher email builder | | `test_http_policy.py` (2) | `voucherbot/providers/http_policy.py` | Robots.txt parsing/caching, politeness delays, per-domain policy state | | `test_logging.py` (7) | `voucherbot/core/logging.py` | Structlog processor chain, log-level setup | -| `test_main.py` (6) | `voucherbot/main.py` + `api/routers/health.py` | `/health` 200 via dependency-overridden session; rate limiting (boundary, disable-at-0, proxy handling) | +| `test_main.py` (9) | `voucherbot/main.py` + `api/routers/health.py` | `/health` 200 via dependency-overridden session; rate limiting (boundary, disable-at-0, proxy handling) | +| `test_migrations.py` (4) | `migrations/` | Chain reachable from base to head, numeric `sourcetype` enum values replayable | | `test_notification_outbox.py` (8) | `voucherbot/services/email/notifications.py` | Idempotency keys, outbox staging, delivery + `is_notified` update, retry/`FAILED` at max attempts, skip-when-unconfigured | +| `test_retention.py` (4) | `voucherbot/services/retention.py` | Content-purge cutoff, untouched columns, config-driven behaviour | **Conventions:** HTTP providers patch `polite_get` with an `AsyncMock` returning a hand-built `httpx.Response`; AI providers patch the client factories and rate-budget internals; email patches `resend.Emails.send`. Modules reading a global `settings` object get it patched with a `SimpleNamespace(...)` helper, while `test_settings.py` builds fresh `Settings` instances with `_env_file=None` and an autouse fixture clearing the environment. DB-bound functions use fake async sessions that route on `str(statement)` so unexpected SQL fails loudly. @@ -195,7 +200,7 @@ IS_TEST=true IS_PROD=false ``` -- `IS_TEST=true` — seeds a `website:local_test` source pointing at `http://localhost:35926/` (see `voucherbot/database/bootstrap.py:967-985`) +- `IS_TEST=true` — seeds a `website:local_test` source pointing at `http://localhost:35926/` (see `voucherbot/database/bootstrap.py:983-1001`) - `IS_PROD=false` — the app creates tables and runs bootstrap on startup ### 2. Start the Local Test Server @@ -219,7 +224,7 @@ The server listens on `http://localhost:35926/`. ### 3. How the Scraper Works -The test source is defined at **`voucherbot/database/bootstrap.py:967-985`** (`_test_source`): +The test source is defined at **`voucherbot/database/bootstrap.py:983-1001`** (`_test_source`): ```python "config": { @@ -233,7 +238,7 @@ The test source is defined at **`voucherbot/database/bootstrap.py:967-985`** (`_ } ``` -The `WebsiteCollector` (`voucherbot/providers/website/collector.py:28-41`) reads these selectors and scrapes the page using BeautifulSoup. +The `WebsiteCollector` (`voucherbot/providers/website/collector.py:38-46`) reads these selectors and scrapes the page using BeautifulSoup. The `index.html` at `D:\components\index.html` contains `.item` divs with `

` titles — this structure matches the default selectors. **To test different content, edit the HTML or the selectors.** @@ -287,9 +292,9 @@ To change how the test page is parsed, edit: | File | Lines | What to change | |------|-------|----------------| -| `voucherbot/database/bootstrap.py` | 967-985 | Source config (`article_selector`, `title_selector`, `link_selector`, `query_terms`) | -| `voucherbot/providers/website/collector.py` | 28-41 | Default selector fallbacks | -| `voucherbot/config/settings.py` | 68 | `is_test` setting | +| `voucherbot/database/bootstrap.py` | 983-1001 | Source config (`article_selector`, `title_selector`, `link_selector`, `query_terms`) | +| `voucherbot/providers/website/collector.py` | 38-46 | Default selector fallbacks | +| `voucherbot/config/settings.py` | 103-176 | `is_test` and related settings | After changing source config, restart the app so bootstrap re-upserts the source. diff --git a/tests/test_event_consolidation.py b/tests/test_event_consolidation.py new file mode 100644 index 0000000..1a5d632 --- /dev/null +++ b/tests/test_event_consolidation.py @@ -0,0 +1,380 @@ +"""Unit tests for the periodic event-consolidation sweep. + +Covers candidate-pair discovery, merge decisioning (AI-confirmed, deterministic +fallback, and budget cap), field folding + post repointing, and the entry-point +guards. No database is used; sessions and provider calls are mocked. +""" + +from __future__ import annotations + +import time +from types import SimpleNamespace +from typing import Any, cast +from unittest.mock import AsyncMock, patch + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from voucherbot.config.settings import EventConsolidationConfig +from voucherbot.models.event import Event, EventStatus, MatchConfidence +from voucherbot.models.source import SourceType +from voucherbot.services import event_consolidation +from voucherbot.services.ai.event_matcher_ai import EventMatchDecision +from voucherbot.services.event_consolidation import ( + _apply_merge, + _candidate_pairs, + _choose_survivor, + _discover_merges, + _event_to_extracted, + _origin_source, + _pair_score, + consolidate_events, +) + + +def _event(**kwargs: object) -> Event: + defaults: dict[str, object] = dict( + id=1, + vendor=None, + promotion_name=None, + promotion_type=None, + certifications=None, + voucher_code=None, + discount=None, + registration_url=None, + start_date=None, + end_date=None, + regions=None, + status=EventStatus.ACTIVE, + merge_log=[], + ) + defaults.update(kwargs) + return Event(**defaults) + + +def _code_pair(a_id: int, b_id: int) -> tuple[Event, Event]: + """Two Events that deterministically score high (code + vendor + discount).""" + return ( + _event( + id=a_id, + vendor="microsoft", + voucher_code="AZ900-50", + discount="50%", + ), + _event( + id=b_id, + vendor="microsoft", + voucher_code="AZ900-50", + discount="50%", + ), + ) + + +def _same(conf: float = 0.9, reason: str = "same vendor") -> EventMatchDecision: + return EventMatchDecision(is_same_promotion=True, confidence=conf, reason=reason) + + +# --------------------------------------------------------------------------- +# _candidate_pairs +# --------------------------------------------------------------------------- + + +class TestCandidatePairs: + def test_groups_by_normalised_url(self) -> None: + a = _event(id=1, registration_url="https://microsoft.com/voucher?utm_source=x") + b = _event(id=2, registration_url="http://MICROSOFT.com/voucher") + assert (a, b) in _candidate_pairs([a, b]) + + def test_groups_by_code_case_insensitive(self) -> None: + a = _event(id=1, voucher_code="AZ900-50") + b = _event(id=2, voucher_code="az900-50") + assert (a, b) in _candidate_pairs([a, b]) + + def test_groups_by_vendor(self) -> None: + a = _event(id=1, vendor="Microsoft") + b = _event(id=2, vendor="microsoft") + assert (a, b) in _candidate_pairs([a, b]) + + def test_dedupes_cross_signal_pairs(self) -> None: + a = _event( + id=1, + vendor="microsoft", + voucher_code="AZ900-50", + registration_url="https://microsoft.com/voucher", + ) + b = _event( + id=2, + vendor="microsoft", + voucher_code="AZ900-50", + registration_url="https://microsoft.com/voucher", + ) + assert _candidate_pairs([a, b]) == [(a, b)] + + def test_ignores_na_voucher_codes(self) -> None: + a = _event(id=1, voucher_code="N/A") + b = _event(id=2, voucher_code="N/A") + assert _candidate_pairs([a, b]) == [] + + def test_no_pairs_without_shared_signal(self) -> None: + a = _event(id=1, vendor="microsoft") + b = _event(id=2, vendor="amazon") + assert _candidate_pairs([a, b]) == [] + + def test_bucket_sample_cap(self) -> None: + events = [_event(id=i, vendor="microsoft") for i in range(1, 6)] + with patch.object(event_consolidation, "_MAX_BUCKET_SAMPLE", 3): + pairs = _candidate_pairs(events) + assert len(pairs) == 3 # C(3, 2) from the first three events + + +# --------------------------------------------------------------------------- +# _event_to_extracted / _origin_source +# --------------------------------------------------------------------------- + + +class TestEventProjection: + def test_projects_iso_dates(self) -> None: + from datetime import datetime, timezone as tz + + event = _event( + id=1, + vendor="microsoft", + start_date=datetime(2026, 8, 1, tzinfo=tz.utc), + regions=["US"], + ) + ex = _event_to_extracted(event) + assert ex.is_voucher is True + assert ex.confidence == 1.0 + assert ex.vendor == "microsoft" + assert ex.start_date == "2026-08-01T00:00:00+00:00" + assert ex.end_date is None + assert ex.regions == ["US"] + + +class TestOriginSource: + def test_uses_creation_source(self) -> None: + event = _event( + merge_log=[ + {"source_type": "BLOG", "fields_updated": ["vendor"]}, + {"source_type": "WEBSITE", "fields_updated": ["end_date"]}, + ] + ) + assert _origin_source(event) == SourceType.BLOG + + def test_skips_unreadable_entries(self) -> None: + event = _event( + merge_log=[ + {"source_type": "UNKNOWN", "fields_updated": []}, + {"source_type": "RSS", "fields_updated": []}, + ] + ) + assert _origin_source(event) == SourceType.RSS + + def test_falls_back_to_rss(self) -> None: + assert _origin_source(_event()) == SourceType.RSS + + +# --------------------------------------------------------------------------- +# _choose_survivor +# --------------------------------------------------------------------------- + + +class TestChooseSurvivor: + def test_more_posts_wins(self) -> None: + a = _event(id=1) + b = _event(id=2) + survivor, loser = _choose_survivor(a, b, {1: 1, 2: 5}) + assert survivor is b + assert loser is a + + def test_tie_keeps_older_event(self) -> None: + a = _event(id=1) + b = _event(id=2) + survivor, loser = _choose_survivor(a, b, {1: 2, 2: 2}) + assert survivor is a + assert loser is b + + +# --------------------------------------------------------------------------- +# _discover_merges +# --------------------------------------------------------------------------- + + +class TestDiscoverMerges: + @pytest.mark.asyncio + async def test_ai_confirms_merges_preferring_high_post_count(self) -> None: + a, b = _code_pair(1, 2) + compare = AsyncMock(return_value=_same()) + merges, ai_calls, pairs, gated = await _discover_merges( + [a, b], {1: 2, 2: 1}, compare + ) + assert ai_calls == 1 + assert pairs == 1 + assert gated == 1 + assert merges == [(a, b, 90, "same vendor")] + + @pytest.mark.asyncio + async def test_ai_says_different_skips(self) -> None: + a, b = _code_pair(1, 2) + compare = AsyncMock( + return_value=EventMatchDecision( + is_same_promotion=False, confidence=0.9, reason="different certs" + ) + ) + merges, ai_calls, *_ = await _discover_merges([a, b], {}, compare) + assert merges == [] + assert ai_calls == 1 + + @pytest.mark.asyncio + async def test_ai_same_but_low_confidence_skips(self) -> None: + a, b = _code_pair(1, 2) + compare = AsyncMock(return_value=_same(conf=0.4)) + merges, ai_calls, *_ = await _discover_merges([a, b], {}, compare) + assert merges == [] + assert ai_calls == 1 + + @pytest.mark.asyncio + async def test_model_unavailable_falls_back_deterministically(self) -> None: + a, b = _code_pair(1, 2) + compare = AsyncMock(return_value=None) + merges, ai_calls, *_ = await _discover_merges([a, b], {1: 1}, compare) + assert ai_calls == 1 + # code 40 + vendor 20 + discount 20 + absent-dates overlap 10 = 90. + assert merges == [(a, b, 90, None)] + + @pytest.mark.asyncio + async def test_model_unavailable_below_deterministic_floor_skips(self) -> None: + a = _event(id=1, vendor="microsoft", promotion_name="AI Skills Fest") + b = _event(id=2, vendor="microsoft", promotion_name="AI Skills Fest") + compare = AsyncMock(return_value=None) + merges, ai_calls, *_ = await _discover_merges([a, b], {}, compare) + assert merges == [] + assert ai_calls == 1 + + @pytest.mark.asyncio + async def test_ai_budget_cap_limits_calls(self) -> None: + events = [_code_pair(i, i + 1)[0] for i in range(1, 4)] + compare = AsyncMock(return_value=_same()) + cfg = EventConsolidationConfig(max_ai_calls_per_sweep=1) + with patch.object(event_consolidation.settings, "consolidation", cfg): + merges, ai_calls, *_ = await _discover_merges(events, {}, compare) + assert ai_calls == 1 + assert len(merges) == 1 + + @pytest.mark.asyncio + async def test_absorbed_events_not_double_merged(self) -> None: + events = [ + _event(id=i, vendor="microsoft", voucher_code="AZ900-50", discount="50%") + for i in range(1, 5) + ] + compare = AsyncMock(return_value=_same()) + merges, *_ = await _discover_merges(events, {}, compare) + # With equal scores the sweep merges (1,2) and (3,4); absorbed events + # are never folded into a second target. + assert len(merges) == 2 + involved: set[int] = set() + for survivor, loser, *_ in merges: + assert loser.id not in involved + involved.add(survivor.id) + involved.add(loser.id) + assert {loser.id for _, loser, *_ in merges} == {2, 4} + + +# --------------------------------------------------------------------------- +# _apply_merge +# --------------------------------------------------------------------------- + + +class _FakeSession: + def __init__(self, rowcount: int = 3) -> None: + self.rowcount = rowcount + self.execute_calls: list[tuple[Any, Any]] = [] + + async def execute(self, stmt: Any, params: Any = None) -> Any: + self.execute_calls.append((stmt, params)) + return SimpleNamespace(rowcount=self.rowcount) + + +class TestApplyMerge: + @pytest.mark.asyncio + async def test_merges_fields_repoints_posts_and_archives(self) -> None: + survivor = _event( + id=1, + vendor="microsoft", + merge_log=[{"source_type": "RSS", "fields_updated": ["vendor"]}], + ) + loser = _event( + id=2, + vendor="microsoft", + voucher_code="AZ900-50", + discount="50%", + merge_log=[{"source_type": "BLOG", "fields_updated": ["discount"]}], + ) + session = _FakeSession(rowcount=3) + + rowcount = await _apply_merge( + cast(AsyncSession, session), survivor, loser, 90, "same vendor" + ) + + assert rowcount == 3 + assert len(session.execute_calls) == 1 + stmt, params = session.execute_calls[0] + assert ( + str(stmt) + == "UPDATE posts SET event_id = :survivor_id WHERE event_id = :loser_id" + ) + assert params == {"survivor_id": 1, "loser_id": 2} + + assert loser.status == EventStatus.ARCHIVED + survivor_log = survivor.merge_log + assert survivor_log is not None + survivor_entry = survivor_log[-1] + assert survivor_entry["source_type"] == "BLOG" + assert survivor_entry["post_id"] == 2 + assert survivor_entry["match_score"] == 90 + assert survivor_entry["match_confidence"] == MatchConfidence.AUTO_MERGED.value + assert survivor_entry["reason"] == "same vendor" + assert survivor.voucher_code == "AZ900-50" + + loser_log = loser.merge_log + assert loser_log is not None + assert loser_log[-1]["reason"] == "consolidated into event 1" + + +# --------------------------------------------------------------------------- +# consolidate_events entry-point guards +# --------------------------------------------------------------------------- + + +class TestConsolidateEvents: + @pytest.mark.asyncio + async def test_disabled_returns_zeros(self) -> None: + cfg = EventConsolidationConfig(enabled=False) + with patch.object(event_consolidation.settings, "consolidation", cfg): + stats = await consolidate_events() + assert stats == { + "candidate_pairs": 0, + "gated_pairs": 0, + "ai_calls": 0, + "merged": 0, + "posts_repointed": 0, + } + + @pytest.mark.asyncio + async def test_throttled_returns_zeros(self) -> None: + with patch.object(event_consolidation, "_last_merge_ts", time.monotonic()): + stats = await consolidate_events() + assert stats["merged"] == 0 + + +# --------------------------------------------------------------------------- +# _pair_score sanity +# --------------------------------------------------------------------------- + + +class TestPairScore: + def test_matches_within_deterministic_bands(self) -> None: + a, b = _code_pair(1, 2) + # code 40 + vendor 20 + discount 20 + absent-dates overlap 10 = 90. + assert _pair_score(a, b) == 90 + assert _pair_score(b, a) == 90 diff --git a/tests/test_event_matcher.py b/tests/test_event_matcher.py index 77d2f68..4e18d5e 100644 --- a/tests/test_event_matcher.py +++ b/tests/test_event_matcher.py @@ -15,7 +15,7 @@ from __future__ import annotations from datetime import datetime, timezone -from typing import Any +from typing import Any, cast from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -23,6 +23,7 @@ from voucherbot.config.settings import EventMatcherConfig, settings from voucherbot.models.event import Event, EventStatus, MatchConfidence from voucherbot.models.source import SourceType +from voucherbot.services.ai.event_matcher_ai import EventMatchDecision from voucherbot.services.ai.schema import ExtractedEvent from voucherbot.services.ingestion.event_matcher import ( _dates_overlap, @@ -68,6 +69,26 @@ def _extracted(**kwargs: Any) -> ExtractedEvent: return ExtractedEvent(**defaults) +def _db_mock() -> AsyncMock: + """AsyncMock DB assigning incremental ids to newly added Events on flush.""" + created: list[Event] = [] + db = AsyncMock() + + def _add_side_effect(obj: object) -> None: + if isinstance(obj, Event): + created.append(obj) + + db.add = MagicMock(side_effect=_add_side_effect) + + async def _flush_side_effect() -> None: + for idx, evt in enumerate(created, start=1): + if evt.id is None: + evt.id = 9000 + idx + + db.flush.side_effect = _flush_side_effect + return db + + # --------------------------------------------------------------------------- # _score_candidate # --------------------------------------------------------------------------- @@ -533,10 +554,16 @@ async def _flush_side_effect() -> None: created_id = event_a.id # --- Post B: same promotion → should match existing Event --- - with patch.object( - matcher, - "_find_candidates", - return_value=[event_a], # candidate found + with ( + patch.object(matcher, "_find_candidates", return_value=[event_a]), + patch( + "voucherbot.services.ingestion.event_matcher.compare_candidate", + new=AsyncMock( + return_value=EventMatchDecision( + is_same_promotion=True, confidence=0.95 + ) + ), + ), ): post_b = MagicMock() post_b.id = 2 @@ -725,3 +752,170 @@ def test_no_shared_fields_stays_below_threshold(self) -> None: score = _score_candidate(e, x) # Only dates_absent (10) contributes. assert score == cfg.weight_date_overlap + + +# --------------------------------------------------------------------------- +# AI-backed matching (qwen decides whether two promotions are the same) +# --------------------------------------------------------------------------- + + +class TestAIMatching: + """Stage 3 AI path: the qwen decision replaces the weighted score.""" + + async def _match( + self, + matcher: EventMatcherCls, + candidates: list[Event], + extracted: ExtractedEvent, + decision: EventMatchDecision | None, + ) -> tuple[Event, MatchConfidence, AsyncMock, AsyncMock]: + db = _db_mock() + post = MagicMock() + post.id = 7 + post.event_id = None + with ( + patch.object(matcher, "_find_candidates", return_value=candidates), + patch( + "voucherbot.services.ingestion.event_matcher.compare_candidate", + new=AsyncMock(return_value=decision), + ) as compare, + patch.object(settings, "groq_api_key", "gsk_test"), + ): + event, confidence = await matcher.match_or_create( + db, extracted, post, SourceType.BLOG + ) + return event, confidence, db, compare + + @pytest.mark.asyncio + async def test_ai_high_confidence_match_auto_merges( + self, matcher: EventMatcherCls + ) -> None: + candidate = _event(id=1, vendor="microsoft", discount="50%") + extracted = _extracted(vendor="microsoft", discount="50%") + decision = EventMatchDecision( + is_same_promotion=True, confidence=0.95, reason="same promo" + ) + + event, confidence, db, compare = await self._match( + matcher, [candidate], extracted, decision + ) + + assert confidence == MatchConfidence.AUTO_MERGED + assert event.id == candidate.id + merge_log = cast(list[Any], event.merge_log or []) + assert merge_log and merge_log[-1]["reason"] == "same promo" + compare.assert_awaited_once() + + @pytest.mark.asyncio + async def test_ai_low_confidence_match_flags_possible( + self, matcher: EventMatcherCls + ) -> None: + candidate = _event(id=1, vendor="microsoft", discount="50%") + extracted = _extracted(vendor="microsoft", discount="50%") + decision = EventMatchDecision(is_same_promotion=True, confidence=0.6) + + event, confidence, _, compare = await self._match( + matcher, [candidate], extracted, decision + ) + + assert confidence == MatchConfidence.POSSIBLE_MATCH + assert event.id == candidate.id + + @pytest.mark.asyncio + async def test_ai_uncertain_match_creates_new_event( + self, matcher: EventMatcherCls + ) -> None: + candidate = _event(id=1, vendor="microsoft", discount="50%") + extracted = _extracted(vendor="microsoft", discount="50%") + decision = EventMatchDecision(is_same_promotion=True, confidence=0.3) + + event, confidence, _, compare = await self._match( + matcher, [candidate], extracted, decision + ) + + assert confidence == MatchConfidence.NEW + assert event.id != candidate.id + + @pytest.mark.asyncio + async def test_ai_confident_no_match_creates_new_event( + self, matcher: EventMatcherCls + ) -> None: + candidate = _event(id=1, vendor="microsoft", discount="50%") + extracted = _extracted(vendor="microsoft", discount="50%") + decision = EventMatchDecision(is_same_promotion=False, confidence=0.9) + + event, confidence, _, compare = await self._match( + matcher, [candidate], extracted, decision + ) + + assert confidence == MatchConfidence.NEW + assert event.id != candidate.id + + @pytest.mark.asyncio + async def test_ai_unavailable_falls_back_to_scoring( + self, matcher: EventMatcherCls + ) -> None: + candidate = _event( + id=1, + registration_url="https://ms.com/promo", + voucher_code="AZURE50", + ) + extracted = _extracted( + registration_url="https://ms.com/promo", + voucher_code="AZURE50", + ) + + event, confidence, _, compare = await self._match( + matcher, [candidate], extracted, None + ) + + assert compare.await_count == 1 # attempted, but the model failed + assert confidence == MatchConfidence.AUTO_MERGED + assert event.id == candidate.id + merge_log = cast(list[Any], event.merge_log or []) + assert merge_log and "reason" not in (merge_log[-1] or {}) + + @pytest.mark.asyncio + async def test_ai_unavailable_fallback_does_not_merge_sparse( + self, matcher: EventMatcherCls + ) -> None: + candidate = _event(id=1, vendor="microsoft") + extracted = _extracted(vendor="amazon") + + event, confidence, _, _ = await self._match( + matcher, [candidate], extracted, None + ) + + assert confidence == MatchConfidence.NEW + assert event.id != candidate.id + + @pytest.mark.asyncio + async def test_ai_skips_candidates_below_deterministic_threshold( + self, matcher: EventMatcherCls + ) -> None: + candidate = _event(id=1, vendor="microsoft") + extracted = _extracted(vendor="amazon") # deterministic score = 10 < 45 + + event, confidence, _, compare = await self._match( + matcher, + [candidate], + extracted, + EventMatchDecision(is_same_promotion=True, confidence=0.95), + ) + + assert confidence == MatchConfidence.NEW + compare.assert_not_awaited() # gate keeps the model call bounded + + @pytest.mark.asyncio + async def test_no_candidates_skips_ai(self, matcher: EventMatcherCls) -> None: + extracted = _extracted(vendor="microsoft") + + event, confidence, _, compare = await self._match( + matcher, + [], + extracted, + EventMatchDecision(is_same_promotion=True, confidence=0.95), + ) + + assert confidence == MatchConfidence.NEW + compare.assert_not_awaited() diff --git a/tests/test_event_matcher_ai.py b/tests/test_event_matcher_ai.py new file mode 100644 index 0000000..2cbfe98 --- /dev/null +++ b/tests/test_event_matcher_ai.py @@ -0,0 +1,257 @@ +"""Unit tests for the qwen-backed event matcher (event_matcher_ai). + +Covers serialization, decision parsing, and the compare_candidate Groq call. +All provider calls are mocked; no live API traffic. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, patch + +import pytest + +from voucherbot.models.event import Event, EventStatus +from voucherbot.services.ai import event_matcher_ai +from voucherbot.services.ai.analyzer import _GROQ_REASONER_MODEL +from voucherbot.services.ai.event_matcher_ai import ( + compare_candidate, + compare_events, + _event_to_dict, + _extracted_to_dict, + _parse_decision, +) +from voucherbot.services.ai.schema import ExtractedEvent + + +VALID_DECISION_JSON = ( + '{"is_same_promotion": true, "confidence": 0.9, "reason": "same vendor"}' +) + + +def _settings(**overrides: object) -> SimpleNamespace: + base = SimpleNamespace(groq_api_key="gsk_test") + base.__dict__.update(overrides) + return base + + +def _event(**kwargs: object) -> Event: + defaults: dict[str, object] = dict( + vendor="microsoft", + promotion_name=None, + promotion_type=None, + certifications=None, + voucher_code=None, + discount="50%", + registration_url=None, + start_date=datetime(2026, 8, 1, tzinfo=timezone.utc), + end_date=None, + regions=None, + status=EventStatus.ACTIVE, + merge_log=[], + ) + defaults.update(kwargs) + return Event(**defaults) + + +def _extracted(**kwargs: Any) -> ExtractedEvent: + defaults: dict[str, Any] = dict( + is_voucher=True, + confidence=0.9, + vendor="microsoft", + discount="50%", + ) + defaults.update(kwargs) + return ExtractedEvent(**defaults) + + +# --------------------------------------------------------------------------- +# _event_to_dict / _extracted_to_dict +# --------------------------------------------------------------------------- + + +class TestSerialisation: + def test_event_dates_become_iso(self) -> None: + d = _event_to_dict(_event()) + assert d["vendor"] == "microsoft" + assert d["start_date"] == "2026-08-01T00:00:00+00:00" + assert d["end_date"] is None + + def test_extracted_passes_fields_through(self) -> None: + ex = _extracted(vendor="microsoft", discount="50%") + assert _extracted_to_dict(ex)["vendor"] == "microsoft" + assert _extracted_to_dict(ex)["registration_url"] is None + + +# --------------------------------------------------------------------------- +# _parse_decision +# --------------------------------------------------------------------------- + + +class TestParseDecision: + def test_parses_plain_json(self) -> None: + decision = _parse_decision(VALID_DECISION_JSON) + assert decision is not None + assert decision.is_same_promotion is True + assert decision.confidence == 0.9 + assert decision.reason == "same vendor" + + def test_parses_markdown_fenced_json(self) -> None: + decision = _parse_decision("```json\n" + VALID_DECISION_JSON + "\n```") + assert decision is not None + assert decision.is_same_promotion is True + + def test_returns_none_on_invalid_json(self) -> None: + assert _parse_decision("not json at all") is None + + def test_returns_none_on_wrong_schema(self) -> None: + assert _parse_decision('{"nope": 1}') is None + + def test_clamps_confidence_on_parse(self) -> None: + decision = _parse_decision('{"is_same_promotion": false, "confidence": 1.5}') + assert decision is not None + assert decision.confidence == 1.0 + decision = _parse_decision('{"is_same_promotion": false, "confidence": -0.2}') + assert decision is not None + assert decision.confidence == 0.0 + + +# --------------------------------------------------------------------------- +# compare_candidate +# --------------------------------------------------------------------------- + + +class TestCompareCandidate: + @pytest.mark.asyncio + async def test_returns_none_without_groq_key(self) -> None: + with ( + patch( + "voucherbot.services.ai.event_matcher_ai.settings", + _settings(groq_api_key=None), + ), + patch( + "voucherbot.services.ai.event_matcher_ai._call_groq_raw", + new=AsyncMock(), + ) as call_raw, + ): + result = await compare_candidate(_event(), _extracted()) + + assert result is None + call_raw.assert_not_awaited() + + @pytest.mark.asyncio + async def test_calls_qwen_with_both_records(self) -> None: + candidate = _event(vendor="microsoft", discount="50%") + extracted = _extracted(vendor="microsoft", discount="50%") + raw = AsyncMock(return_value=VALID_DECISION_JSON) + with ( + patch("voucherbot.services.ai.event_matcher_ai.settings", _settings()), + patch( + "voucherbot.services.ai.event_matcher_ai._call_groq_raw", + new=raw, + ) as call_raw, + ): + result = await compare_candidate(candidate, extracted) + + assert result is not None + assert result.is_same_promotion is True + call_raw.assert_awaited_once() + call = call_raw.await_args + assert call is not None + messages = call.args[0] + assert call.args[1] == _GROQ_REASONER_MODEL + assert messages[0]["role"] == "system" + assert messages[0]["content"] == event_matcher_ai._MATCH_SYSTEM_PROMPT + assert "INCOMING (newly detected) promotion" in messages[1]["content"] + assert '"microsoft"' in messages[1]["content"] + assert '"50%"' in messages[1]["content"] + + @pytest.mark.asyncio + async def test_returns_none_when_model_fails(self) -> None: + with ( + patch("voucherbot.services.ai.event_matcher_ai.settings", _settings()), + patch( + "voucherbot.services.ai.event_matcher_ai._call_groq_raw", + new=AsyncMock(return_value=None), + ), + ): + result = await compare_candidate(_event(), _extracted()) + + assert result is None + + @pytest.mark.asyncio + async def test_returns_none_on_unparseable_response(self) -> None: + with ( + patch("voucherbot.services.ai.event_matcher_ai.settings", _settings()), + patch( + "voucherbot.services.ai.event_matcher_ai._call_groq_raw", + new=AsyncMock(return_value="garbage"), + ), + ): + result = await compare_candidate(_event(), _extracted()) + + assert result is None + + +# --------------------------------------------------------------------------- +# compare_events +# --------------------------------------------------------------------------- + + +class TestCompareEvents: + @pytest.mark.asyncio + async def test_returns_none_without_groq_key(self) -> None: + with ( + patch( + "voucherbot.services.ai.event_matcher_ai.settings", + _settings(groq_api_key=None), + ), + patch( + "voucherbot.services.ai.event_matcher_ai._call_groq_raw", + new=AsyncMock(), + ) as call_raw, + ): + result = await compare_events(_event(), _event()) + + assert result is None + call_raw.assert_not_awaited() + + @pytest.mark.asyncio + async def test_calls_qwen_with_both_events(self) -> None: + existing = _event(vendor="microsoft", discount="50%") + incoming = _event(vendor="microsoft", discount="50%") + raw = AsyncMock(return_value=VALID_DECISION_JSON) + with ( + patch("voucherbot.services.ai.event_matcher_ai.settings", _settings()), + patch( + "voucherbot.services.ai.event_matcher_ai._call_groq_raw", + new=raw, + ) as call_raw, + ): + result = await compare_events(existing, incoming) + + assert result is not None + assert result.is_same_promotion is True + call_raw.assert_awaited_once() + call = call_raw.await_args + assert call is not None + messages = call.args[0] + assert call.args[1] == _GROQ_REASONER_MODEL + assert messages[0]["role"] == "system" + assert "INCOMING (newly detected) promotion" in messages[1]["content"] + assert '"microsoft"' in messages[1]["content"] + + @pytest.mark.asyncio + async def test_returns_none_when_model_fails(self) -> None: + with ( + patch("voucherbot.services.ai.event_matcher_ai.settings", _settings()), + patch( + "voucherbot.services.ai.event_matcher_ai._call_groq_raw", + new=AsyncMock(return_value=None), + ), + ): + result = await compare_events(_event(), _event()) + + assert result is None diff --git a/voucherbot/config/settings.py b/voucherbot/config/settings.py index 447860b..d90867e 100644 --- a/voucherbot/config/settings.py +++ b/voucherbot/config/settings.py @@ -43,6 +43,45 @@ class EventMatcherConfig(BaseModel): # "Microsoft Fabric Data Days" vs "Fabric Data Days" (~0.76) name_similarity_threshold: float = 0.60 + # --- AI-backed matching --- + # When enabled, the qwen reasoning model decides whether an incoming + # promotion is the same as an existing candidate instead of the weighted + # score. Deterministic scoring is kept as a fallback when the model is + # unavailable or no candidates exist. + use_ai_matcher: bool = True + # How many relevance-ranked candidates to submit to the model per post. + ai_candidate_limit: int = 5 + # Decision-confidence bands (0–1) for the AI matcher output. A candidate + # flagged as the same promotion merges when confidence >= possible band; + # it is an auto-merge only when confidence >= auto-merge band. + ai_auto_merge_confidence: float = 0.8 + ai_possible_match_confidence: float = 0.5 + + +class EventConsolidationConfig(BaseModel): + """Periodic sweep that merges duplicate canonical Events. + + Two Events that the ingestion-time matcher could not see at once (e.g. + created from different sources on different sweeps) are detected by a + periodic job: Events sharing a cheap identity signal (normalised + registration URL, voucher code, or vendor) are gated by the deterministic + weighted score, and qwen then confirms whether each pair is the same + real-world promotion. The survivor keeps its Events posts, the absorbed + Event is archived, and its posts are re-pointed. + """ + + # Master switch for the consolidation sweep. + enabled: bool = True + # Minimum wall-clock time between sweeps (rate-limits the qwen spend). + interval_minutes: int = 60 + # Hard cap on candidate pairs examined per sweep (bounds quadratic work). + max_pairs_per_sweep: int = 1000 + # How many qwen calls to allow per sweep (each is billed). + max_ai_calls_per_sweep: int = 25 + # Deterministic-score floor for a merge when the model is unavailable; + # mirrors auto_merge_threshold so behaviour matches ingestion-time matching. + deterministic_auto_merge_threshold: int = 70 + # Ordered from most to least authoritative. Lower index = higher priority. # Used by the EventMatcher when merging fields from a new post into an existing @@ -129,6 +168,9 @@ class Settings(BaseSettings): # constructing Settings(event_matcher=EventMatcherConfig(...))) event_matcher: EventMatcherConfig = EventMatcherConfig() + # Periodic deduplication of already-created canonical Events. + consolidation: EventConsolidationConfig = EventConsolidationConfig() + model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", extra="ignore" ) diff --git a/voucherbot/services/ai/analyzer.py b/voucherbot/services/ai/analyzer.py index 2792f77..23bea3e 100644 --- a/voucherbot/services/ai/analyzer.py +++ b/voucherbot/services/ai/analyzer.py @@ -360,24 +360,21 @@ def _parse_to_extracted_event(raw_text: str) -> ExtractedEvent | None: # --------------------------------------------------------------------------- # Provider adapters # --------------------------------------------------------------------------- -async def _call_groq_model( - title: str, content: str | None, model: str, source_name: str | None = None -) -> ExtractedEvent | None: - """Call a specific Groq model. Returns None on daily limit or non-retryable failure.""" +async def _call_groq_raw(messages: list[dict[str, str]], model: str) -> str | None: + """Low-level Groq chat completion call sharing rate-limit and budget logic. + + Used by both the structured extractor (``_call_groq_model``) and the qwen + event matcher (``voucherbot.services.ai.event_matcher_ai``). Returns the + raw response text, or None on model unavailability, daily-limit exhaustion, + or non-retryable failure. + """ if not is_model_available(model): logger.info("ai.analyzer: model daily limit exhausted, skipping", model=model) return None client = AsyncGroq(api_key=settings.groq_api_key) - content_for_prompt = content or "(no content)" - - source_hint = f"Source: {source_name}\n" if source_name else "" - user_prompt = f"{source_hint}Title: {title}\n\nContent: {content_for_prompt}" - messages = [ - {"role": "system", "content": _SYSTEM_PROMPT}, - {"role": "user", "content": user_prompt}, - ] - estimated_tokens = _estimate_tokens(_SYSTEM_PROMPT + user_prompt) + prompt_text = "\n".join(m.get("content", "") for m in messages) + estimated_tokens = _estimate_tokens(prompt_text) for attempt in range(1, _MAX_RETRIES + 1): try: @@ -401,7 +398,7 @@ async def _call_groq_model( actual = getattr(resp.usage, "total_tokens", None) or estimated_tokens await _settle_groq_budget(rid, actual, model) raw_text: str = resp.choices[0].message.content.strip() - return _parse_to_extracted_event(raw_text) + return raw_text or None except Exception as exc: error_str = str(exc) if "429" in error_str: @@ -436,6 +433,24 @@ async def _call_groq_model( return None +async def _call_groq_model( + title: str, content: str | None, model: str, source_name: str | None = None +) -> ExtractedEvent | None: + """Call a specific Groq model. Returns None on daily limit or non-retryable failure.""" + content_for_prompt = content or "(no content)" + + source_hint = f"Source: {source_name}\n" if source_name else "" + user_prompt = f"{source_hint}Title: {title}\n\nContent: {content_for_prompt}" + messages = [ + {"role": "system", "content": _SYSTEM_PROMPT}, + {"role": "user", "content": user_prompt}, + ] + raw_text = await _call_groq_raw(messages, model) + if raw_text is None: + return None + return _parse_to_extracted_event(raw_text) + + async def _call_groq( title: str, content: str | None, source_name: str | None = None ) -> ExtractedEvent | None: diff --git a/voucherbot/services/ai/event_matcher_ai.py b/voucherbot/services/ai/event_matcher_ai.py new file mode 100644 index 0000000..4cdc969 --- /dev/null +++ b/voucherbot/services/ai/event_matcher_ai.py @@ -0,0 +1,182 @@ +""" +AI-backed canonical event matching. + +The qwen reasoning model decides whether an incoming ``ExtractedEvent`` +describes the same real-world promotion as an existing canonical ``Event``. +This replaces the deterministic weighted scoring used for the merge decision in +the EventMatcher; deterministic scoring is retained only as a fallback when the +model is unavailable. + +The decision output is an ``EventMatchDecision``: ``is_same_promotion`` plus a +0–1 ``confidence`` and a human-readable ``reason`` for the audit log. +""" + +from __future__ import annotations + +import json +import re +from datetime import datetime +from typing import Any, Optional + +import structlog +from pydantic import BaseModel, field_validator + +from voucherbot.config.settings import settings +from voucherbot.models.event import Event +from voucherbot.services.ai.analyzer import _GROQ_REASONER_MODEL, _call_groq_raw +from voucherbot.services.ai.schema import ExtractedEvent + +logger = structlog.get_logger(__name__) + +# Fields compared when the model decides whether two promotions are the same. +_FIELD_NAMES: tuple[str, ...] = ( + "vendor", + "promotion_name", + "promotion_type", + "certifications", + "voucher_code", + "discount", + "registration_url", + "start_date", + "end_date", + "regions", +) + +_MATCH_SYSTEM_PROMPT = ( + "You are a deduplication judge for a certification voucher aggregator. " + "You are given two promotion records — EXISTING (an already-known " + "promotion) and INCOMING (a newly detected promotion). Decide whether they " + "describe the SAME real-world promotion.\n\n" + "They are the SAME promotion when the same vendor offers the same deal for " + "the same certification exams, even if wording, casing, URLs, tracking " + "parameters, or publication dates differ. Strong indicators of sameness:\n" + "- identical or near-identical voucher code or registration URL\n" + "- identical vendor with the same or overlapping certification list\n" + "- the same promotion name with minor wording differences\n" + "- the same discount value (e.g. both '50% off AZ-900')\n\n" + "They are DIFFERENT promotions when:\n" + "- vendors differ\n" + "- the certification exams differ\n" + "- discounts differ materially (e.g. 50% off vs 80% off)\n" + "- the date ranges do not overlap and describe distinct campaigns\n\n" + "When uncertain, prefer is_same_promotion=false to avoid merging distinct " + "promotions — a false merge destroys provenance.\n\n" + "Respond with ONLY a valid JSON object matching this exact schema:\n" + "{\n" + ' "is_same_promotion": true | false,\n' + ' "confidence": 0.0-1.0,\n' + ' "reason": "string"\n' + "}\n" +) + + +class EventMatchDecision(BaseModel): + """Model output deciding whether two promotions are the same real-world event.""" + + is_same_promotion: bool # required: absence of the field is a parse failure + confidence: float = 0.0 # 0.0 – 1.0, belief in the decision + reason: Optional[str] = None + + @field_validator("confidence") + @classmethod + def clamp_confidence(cls, v: float) -> float: + return max(0.0, min(1.0, v)) + + +def _event_to_dict(event: Event) -> dict[str, Any]: + """Serialize the comparable Event fields for the prompt (dates as ISO).""" + + def _serialise(value: Any) -> Any: + if isinstance(value, datetime): + return value.isoformat() + return value + + return {field: _serialise(getattr(event, field, None)) for field in _FIELD_NAMES} + + +def _extracted_to_dict(extracted: ExtractedEvent) -> dict[str, Any]: + """Serialize the comparable ExtractedEvent fields for the prompt.""" + return {field: getattr(extracted, field, None) for field in _FIELD_NAMES} + + +def _parse_decision(raw_text: str) -> Optional[EventMatchDecision]: + """Parse a raw provider response into an ``EventMatchDecision``. + + Handles accidental markdown fences. Returns ``None`` on parse failure so + callers can fall back to deterministic scoring. + """ + try: + text = raw_text.strip() + if text.startswith("```"): + text = re.sub(r"^```[a-zA-Z]*\n?", "", text) + text = re.sub(r"\n?```$", "", text) + + data: dict[str, Any] = json.loads(text) + return EventMatchDecision.model_validate(data) + except Exception as exc: + logger.warning( + "ai.event_matcher: failed to parse match decision", + error=str(exc), + raw=raw_text[:300], + ) + return None + + +def _build_match_messages( + existing: dict[str, Any], incoming: dict[str, Any] +) -> list[dict[str, str]]: + """Build the qwen chat messages framing two records for a same-promotion judge.""" + prompt = ( + "EXISTING (already-known) promotion:\n" + + json.dumps(existing, ensure_ascii=False, indent=2) + + "\n\nINCOMING (newly detected) promotion:\n" + + json.dumps(incoming, ensure_ascii=False, indent=2) + ) + return [ + {"role": "system", "content": _MATCH_SYSTEM_PROMPT}, + {"role": "user", "content": prompt}, + ] + + +async def _ask_match_decision( + messages: list[dict[str, str]], +) -> Optional[EventMatchDecision]: + """Send match-judge messages to qwen and parse the decision.""" + raw_text = await _call_groq_raw(messages, _GROQ_REASONER_MODEL) + if raw_text is None: + return None + return _parse_decision(raw_text) + + +async def compare_candidate( + candidate: Event, + extracted: ExtractedEvent, +) -> Optional[EventMatchDecision]: + """Ask qwen whether ``extracted`` is the same promotion as ``candidate``. + + Returns ``None`` when no Groq key is configured, the model is unavailable, + or the response cannot be parsed — callers then fall back to deterministic + scoring. + """ + if not settings.groq_api_key: + return None + messages = _build_match_messages( + _event_to_dict(candidate), _extracted_to_dict(extracted) + ) + return await _ask_match_decision(messages) + + +async def compare_events( + existing: Event, + incoming: Event, +) -> Optional[EventMatchDecision]: + """Ask qwen whether two canonical Events are the same real-world promotion. + + Used by the periodic consolidation sweep to decide whether two already- + created Events should be merged into one. Same fallback semantics as + ``compare_candidate``. + """ + if not settings.groq_api_key: + return None + messages = _build_match_messages(_event_to_dict(existing), _event_to_dict(incoming)) + return await _ask_match_decision(messages) diff --git a/voucherbot/services/event_consolidation.py b/voucherbot/services/event_consolidation.py new file mode 100644 index 0000000..9177902 --- /dev/null +++ b/voucherbot/services/event_consolidation.py @@ -0,0 +1,344 @@ +"""Event consolidation: periodically merge duplicate canonical Events. + +Two Posts describing the same real-world promotion can become separate Events +when their sources were processed at different times (the ingestion-time +matcher only sees candidates that already exist at that moment). This module +runs a periodic sweep that finds those duplicates retroactively: + +1. Events sharing a cheap identity signal (normalised registration URL, + voucher code, or vendor) are grouped into candidate pairs. +2. Each pair is gated by the same deterministic weighted score used at + ingestion; only pairs at or above ``possible_match_threshold`` are kept. +3. When Groq is configured, qwen is asked whether the pair is the same + real-world promotion (see ``compare_events``). A ``same`` decision with + ``confidence >= ai_possible_match_confidence`` merges the pair; a model + outage falls back to the deterministic score >= auto-merge threshold. +4. The pair's merge picks a survivor (the Event with more Posts; ties keep the + older Event), folds the absorber's fields into it via the shared + ``_merge_fields`` machinery, re-points the absorber's Posts, and archives + it. Provenance is preserved: Posts are never merged. + +The sweep is cross-instance serialised with a Postgres advisory transaction +lock, throttled so qwen spend stays bounded, and isolated from the rest of the +scheduler loop — it never raises. +""" + +from __future__ import annotations + +import time +from collections.abc import Awaitable, Callable +from datetime import datetime, timezone +from typing import Any, Optional, cast + +import structlog +from sqlalchemy import CursorResult, func, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from voucherbot.config.settings import settings as settings +from voucherbot.database.connection import session_scope +from voucherbot.models.event import Event, EventStatus, MatchConfidence +from voucherbot.models.post import Post +from voucherbot.models.source import SourceType +from voucherbot.services.ai.event_matcher_ai import ( + EventMatchDecision, + compare_events, +) +from voucherbot.services.ai.schema import ExtractedEvent +from voucherbot.services.ingestion.dedup import normalise_url +from voucherbot.services.ingestion.event_matcher import ( + _merge_fields, + _score_candidate, +) + +logger = structlog.get_logger(__name__) + +# pg_advisory_xact_lock key scoping this job to a single scheduler instance. +_LOCK_KEY = 734_001 +# Per-bucket cap while building the candidate-pair graph (bounds quadratic work). +_MAX_BUCKET_SAMPLE = 200 + +# Decision function shared by discovery and the sweep entry point. +CompareFn = Callable[[Event, Event], Awaitable[Optional[EventMatchDecision]]] + +_last_merge_ts: float = 0.0 + + +# --------------------------------------------------------------------------- +# Candidate discovery +# --------------------------------------------------------------------------- + + +def _candidate_pairs(events: list[Event]) -> list[tuple[Event, Event]]: + """Return distinct Event pairs sharing a cheap identity signal. + + Pair order (a, b) is deterministic in the input list; duplicates are + collapsed by the canonical ``(min_id, max_id)`` key so a pair found via + both a shared URL and voucher code is only reported once. + """ + buckets: dict[tuple[str, str], list[Event]] = {} + for event in events: + if event.registration_url: + key = ("url", normalise_url(event.registration_url)) + buckets.setdefault(key, []).append(event) + if event.voucher_code and event.voucher_code.upper() not in ("N/A", "NA"): + key = ("code", event.voucher_code.upper()) + buckets.setdefault(key, []).append(event) + if event.vendor: + key = ("vendor", event.vendor.lower()) + buckets.setdefault(key, []).append(event) + + seen: set[tuple[int, int]] = set() + pairs: list[tuple[Event, Event]] = [] + for bucket in buckets.values(): + sampled = bucket[:_MAX_BUCKET_SAMPLE] + for i in range(len(sampled)): + a = sampled[i] + for j in range(i + 1, len(sampled)): + b = sampled[j] + pair_key = (min(a.id, b.id), max(a.id, b.id)) + if pair_key in seen: + continue + seen.add(pair_key) + pairs.append((a, b)) + return pairs + + +def _event_to_extracted(event: Event) -> ExtractedEvent: + """Project an Event onto the ExtractedEvent shape the scorer understands.""" + return ExtractedEvent( + is_voucher=True, + confidence=1.0, + vendor=event.vendor, + promotion_name=event.promotion_name, + promotion_type=event.promotion_type, + certifications=event.certifications, + voucher_code=event.voucher_code, + discount=event.discount, + registration_url=event.registration_url, + start_date=event.start_date.isoformat() if event.start_date else None, + end_date=event.end_date.isoformat() if event.end_date else None, + regions=event.regions, + ) + + +def _pair_score(a: Event, b: Event) -> int: + """Reuse the ingestion-time deterministic weighted score between two Events.""" + return _score_candidate(a, _event_to_extracted(b)) + + +def _choose_survivor( + a: Event, b: Event, post_counts: dict[int, int] +) -> tuple[Event, Event]: + """Return (survivor, absorbed): the Event with more Posts wins. + + Ties keep the older Event (lower id) as the canonical record. + """ + a_count = post_counts.get(a.id, 0) + b_count = post_counts.get(b.id, 0) + if a_count > b_count: + return a, b + if b_count > a_count: + return b, a + if a.id < b.id: + return a, b + return b, a + + +def _origin_source(event: Event) -> SourceType: + """Best-guess originating source of an Event for the merge audit log. + + The earliest readable ``source_type`` in the Event's merge_log (the source + that created it) is used; falls back to RSS (lowest authority) when + unrestorable. + """ + for entry in event.merge_log or []: + raw = entry.get("source_type") + if not isinstance(raw, str): + continue + try: + return SourceType(raw) + except ValueError: + continue + return SourceType.RSS + + +# --------------------------------------------------------------------------- +# Merge discovery + application +# --------------------------------------------------------------------------- + + +async def _discover_merges( + events: list[Event], + post_counts: dict[int, int], + compare: CompareFn, +) -> tuple[list[tuple[Event, Event, int, Optional[str]]], int, int, int]: + """Decide which Event pairs to merge this sweep. + + Returns ``(merges, ai_calls, candidate_pairs, gated_pairs)`` where each + merge is ``(survivor, absorbed, match_score, match_reason)``. Deterministic + gating bounds qwen volume, and the absorbed-set prevents one Event from + being folded into two targets in a single sweep. + """ + cfg = settings.event_matcher + cons = settings.consolidation + + merges: list[tuple[Event, Event, int, Optional[str]]] = [] + absorbed: set[int] = set() + ai_calls = 0 + + pairs = _candidate_pairs(events)[: cons.max_pairs_per_sweep] + gated: list[tuple[int, Event, Event]] = [] + for a, b in pairs: + score = _pair_score(a, b) + if score >= cfg.possible_match_threshold: + gated.append((score, a, b)) + gated.sort(key=lambda item: item[0], reverse=True) + + for score, a, b in gated: + if a.id in absorbed or b.id in absorbed: + continue + survivor, loser = _choose_survivor(a, b, post_counts) + + if ai_calls < cons.max_ai_calls_per_sweep: + ai_calls += 1 + decision = await compare(survivor, loser) + if decision is None: + # Model unavailable — fall back to the deterministic floor. + if score < cons.deterministic_auto_merge_threshold: + continue + merges.append((survivor, loser, score, None)) + elif ( + not decision.is_same_promotion + or decision.confidence < cfg.ai_possible_match_confidence + ): + # Model (or its weak confidence) says these are different. + continue + else: + ai_score = int(round(decision.confidence * 100)) + merges.append((survivor, loser, ai_score, decision.reason)) + else: + # AI budget spent — deterministic floor only. + if score < cons.deterministic_auto_merge_threshold: + continue + merges.append((survivor, loser, score, None)) + + absorbed.update({a.id, b.id}) + + return merges, ai_calls, len(pairs), len(gated) + + +async def _post_counts(session: AsyncSession) -> dict[int, int]: + """Count Posts per Event for survivor selection.""" + result = await session.execute( + select(Post.event_id, func.count()) + .where(Post.event_id.is_not(None)) + .group_by(Post.event_id) + ) + return { + int(event_id): int(count) + for event_id, count in result.all() + if event_id is not None + } + + +async def _apply_merge( + session: AsyncSession, + survivor: Event, + loser: Event, + score: int, + reason: Optional[str], +) -> int: + """Fold ``loser`` into ``survivor`` and archive it. + + Repoints the absorbed Event's Posts to the survivor, merges its fields via + source priority, and records the consolidation on both audit logs. Returns + the number of Posts re-pointed. + """ + loser_source = _origin_source(loser) + _merge_fields( + survivor, + loser, + loser_source, + loser.id, + score, + MatchConfidence.AUTO_MERGED, + match_reason=reason, + ) + result = await session.execute( + text("UPDATE posts SET event_id = :survivor_id WHERE event_id = :loser_id"), + {"survivor_id": survivor.id, "loser_id": loser.id}, + ) + rowcount = cast(CursorResult[Any], result).rowcount or 0 + + loser.status = EventStatus.ARCHIVED + loser.merge_log = (loser.merge_log or []) + [ + { + "timestamp": datetime.now(timezone.utc).isoformat(), + "source_type": loser_source.value, + "post_id": loser.id, + "match_score": score, + "match_confidence": EventStatus.ARCHIVED.value, + "fields_updated": [], + "reason": f"consolidated into event {survivor.id}", + } + ] + return int(rowcount) + + +# --------------------------------------------------------------------------- +# Sweep entry point +# --------------------------------------------------------------------------- + + +async def consolidate_events() -> dict[str, int]: + """Run one throttled consolidation sweep, returning stats. Never raises.""" + stats = { + "candidate_pairs": 0, + "gated_pairs": 0, + "ai_calls": 0, + "merged": 0, + "posts_repointed": 0, + } + global _last_merge_ts + cfg = settings.consolidation + if not cfg.enabled: + return stats + if time.monotonic() - _last_merge_ts < cfg.interval_minutes * 60: + return stats + _last_merge_ts = time.monotonic() + + try: + async with session_scope() as session: + # Cross-instance serialisation: hold a transaction-level advisory + # lock for the whole sweep so only one scheduler runs it at a time. + await session.execute( + text("SELECT pg_advisory_xact_lock(:key)"), {"key": _LOCK_KEY} + ) + events = list( + ( + await session.execute( + select(Event).where(Event.status == EventStatus.ACTIVE) + ) + ) + .scalars() + .all() + ) + post_counts = await _post_counts(session) + merges, ai_calls, candidate_pairs, gated_pairs = await _discover_merges( + events, post_counts, compare_events + ) + repointed = 0 + for survivor, loser, score, reason in merges: + repointed += await _apply_merge(session, survivor, loser, score, reason) + await session.commit() + + stats["candidate_pairs"] = candidate_pairs + stats["gated_pairs"] = gated_pairs + stats["ai_calls"] = ai_calls + stats["merged"] = len(merges) + stats["posts_repointed"] = repointed + if stats["merged"]: + logger.info("event_consolidation: merged duplicate events", **stats) + except Exception as exc: + logger.warning("event_consolidation: sweep failed", error=str(exc)[:200]) + return stats diff --git a/voucherbot/services/ingestion/event_matcher.py b/voucherbot/services/ingestion/event_matcher.py index 9b6586d..463b584 100644 --- a/voucherbot/services/ingestion/event_matcher.py +++ b/voucherbot/services/ingestion/event_matcher.py @@ -5,8 +5,9 @@ determines whether the data describes an existing canonical ``Event`` (and attaches the Post to it), or whether a brand-new Event should be created. -Matching is purely deterministic, operating on structured fields — never on -raw article text. +Matching operates on structured fields (never raw article text). By default +the merge decision is handed to the qwen reasoning model (see below); the +deterministic weighted score below is only used as a fallback. Scoring ------- @@ -28,6 +29,24 @@ >= possible_match_threshold (45) → flag as POSSIBLE_MATCH (future review) < possible_match_threshold → create a new Event +AI-backed matching +------------------ +When ``settings.event_matcher.use_ai_matcher`` is enabled and candidates exist, +the qwen reasoning model is asked whether the incoming extracted promotion is +the same as each candidate that the deterministic weighted score flags as a +possible match (``score >= possible_match_threshold``, sorted best-first and +capped by ``ai_candidate_limit``; see ``voucherbot.services.ai.event_matcher_ai``). +The model's ``is_same_promotion`` and ``confidence`` drive the decision: + + is_same_promotion and confidence >= ai_auto_merge_confidence + → AUTO_MERGED + is_same_promotion and confidence >= ai_possible_match_confidence + → POSSIBLE_MATCH + otherwise → new Event + +The deterministic weighted scoring above remains as a fallback when the model +is unavailable, no Groq key is configured, or no candidates exist. + Source Priority & Field Merging -------------------------------- When an Event is updated by a new Post, fields are merged field-by-field @@ -58,6 +77,10 @@ from voucherbot.models.event import Event, EventStatus, MatchConfidence from voucherbot.models.post import Post from voucherbot.models.source import SourceType +from voucherbot.services.ai.event_matcher_ai import ( + EventMatchDecision, + compare_candidate, +) from voucherbot.services.ai.schema import ExtractedEvent from voucherbot.services.ingestion.dedup import normalise_url @@ -293,14 +316,19 @@ def _candidate_relevance(event: Event, extracted: ExtractedEvent) -> int: def _merge_fields( event: Event, - extracted: ExtractedEvent, + extracted: ExtractedEvent | Event, source_type: SourceType, post_id: int, match_score: int, match_confidence: MatchConfidence, + match_reason: Optional[str] = None, ) -> list[str]: """Merge non-null fields from ``extracted`` into ``event`` using source priority. + ``extracted`` may be an ``ExtractedEvent`` from the AI pipeline or another + ``Event`` (used by the consolidation sweep to fold one canonical Event into + another); both expose the same ``_MERGEABLE_FIELDS`` attributes. + Merge rules (in order): 1. Null incoming values are **never** written (preserves existing data). 2. Null existing values are **backfilled** from the incoming data regardless @@ -353,6 +381,8 @@ def _merge_fields( "match_confidence": match_confidence.value, "fields_updated": updated_fields, } + if match_reason: + log_entry["reason"] = match_reason current_log: list[Any] = event.merge_log or [] event.merge_log = current_log + [log_entry] @@ -541,6 +571,44 @@ async def update_existing( ) return event, MatchConfidence.UPDATED + async def _pick_ai_match( + self, candidates: list[Event], extracted: ExtractedEvent + ) -> tuple[Optional[Event], Optional[EventMatchDecision]]: + """Ask qwen whether any candidate is the same promotion as ``extracted``. + + The deterministic weighted score is used as a recall gate: only + candidates scoring at or above ``possible_match_threshold`` are + submitted to the model (sorted best-first, capped by + ``ai_candidate_limit``) so qwen calls stay bounded. Returns the first + such candidate the model flags as the same promotion together with its + decision. When the model is available but judges nothing a match, + returns ``(None, last_decision)`` so the caller creates a new Event + instead of merging. When no candidate passes the gate or the model is + unavailable (a ``None`` decision), returns ``(None, None)`` so the + caller can fall back to deterministic scoring. + """ + cfg = settings.event_matcher + gated: list[tuple[int, Event]] = [] + for candidate in candidates: + score = _score_candidate(candidate, extracted) + if score >= cfg.possible_match_threshold: + gated.append((score, candidate)) + gated.sort(key=lambda item: item[0], reverse=True) + ai_candidates = [event for _, event in gated[: cfg.ai_candidate_limit]] + + last_decision: Optional[EventMatchDecision] = None + for candidate in ai_candidates: + decision = await compare_candidate(candidate, extracted) + if decision is None: + return None, None + last_decision = decision + if ( + decision.is_same_promotion + and decision.confidence >= cfg.ai_possible_match_confidence + ): + return candidate, decision + return None, last_decision + async def match_or_create( self, db: AsyncSession, @@ -556,33 +624,61 @@ async def match_or_create( candidates = await self._find_candidates(db, extracted) best_event: Optional[Event] = None - best_score = 0 - - for candidate in candidates: - score = _score_candidate(candidate, extracted) - if score > best_score: - best_score = score - best_event = candidate - - # --- Determine confidence band --- - if best_score >= cfg.auto_merge_threshold and best_event is not None: - confidence = MatchConfidence.AUTO_MERGED - elif best_score >= cfg.possible_match_threshold and best_event is not None: - confidence = MatchConfidence.POSSIBLE_MATCH - else: - confidence = MatchConfidence.NEW - best_event = None # ignore low-confidence candidates + match_score = 0 + best_reason: Optional[str] = None + confidence: MatchConfidence + + # --- AI path: qwen decides whether a candidate is the same promotion. --- + ai_match = False + if cfg.use_ai_matcher and candidates and settings.groq_api_key: + ai_event, ai_decision = await self._pick_ai_match(candidates, extracted) + if ai_event is not None and ai_decision is not None: + best_event = ai_event + best_reason = ai_decision.reason + match_score = int(round(ai_decision.confidence * 100)) + if ai_decision.confidence >= cfg.ai_auto_merge_confidence: + confidence = MatchConfidence.AUTO_MERGED + else: + confidence = MatchConfidence.POSSIBLE_MATCH + ai_match = True + elif ai_decision is not None: + # The model was available and judged no candidate a match. + confidence = MatchConfidence.NEW + ai_match = True + + # --- Deterministic fallback when AI was not used or was unavailable. --- + if not ai_match: + best_score = 0 + for candidate in candidates: + score = _score_candidate(candidate, extracted) + if score > best_score: + best_score = score + best_event = candidate + match_score = best_score + if best_score >= cfg.auto_merge_threshold and best_event is not None: + confidence = MatchConfidence.AUTO_MERGED + elif best_score >= cfg.possible_match_threshold and best_event is not None: + confidence = MatchConfidence.POSSIBLE_MATCH + else: + confidence = MatchConfidence.NEW + best_event = None # ignore low-confidence candidates if best_event is not None: # --- Attach to existing Event --- updated = _merge_fields( - best_event, extracted, source_type, post.id, best_score, confidence + best_event, + extracted, + source_type, + post.id, + match_score, + confidence, + match_reason=best_reason, ) logger.info( "event_matcher: attached post to existing event", event_id=best_event.id, post_id=post.id, - score=best_score, + score=match_score, confidence=confidence.value, fields_updated=updated, ) diff --git a/voucherbot/services/scheduler.py b/voucherbot/services/scheduler.py index 65b51cc..86f4b25 100644 --- a/voucherbot/services/scheduler.py +++ b/voucherbot/services/scheduler.py @@ -27,6 +27,7 @@ from voucherbot.providers.training_provider.collector import TrainingProviderCollector from voucherbot.services.dispatcher import dispatch_tick from voucherbot.services.email.notifications import retry_pending_notifications +from voucherbot.services.event_consolidation import consolidate_events from voucherbot.services.retention import purge_expired_post_content logger = structlog.get_logger(__name__) @@ -149,6 +150,9 @@ async def _run_loop() -> None: await retry_pending_notifications() # Null out content of posts older than the retention window. await purge_expired_post_content() + # Merge duplicate canonical Events that never saw each other at + # ingestion time (throttled internally). + await consolidate_events() sleep_seconds = await _seconds_until_next_due() logger.info( "scheduler: sleeping until next sweep",