Merge pull request #20 from Devathmaj/refactor/deduplication-pipeline

feat: qwen-backed event matching with retroactive consolidation

Author: Devathmaj

CONTRIBUTING.md

@@ -96,7 +96,7 @@ Key obligations that apply directly to this project:
 git clone <repo-url>
 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:**
 

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

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
 

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
 

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 <onboarding@resend.dev>` | 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.