feat: add NER, NEL, and Classify APIs with change-stream worker

.dockerignore

@@ -0,0 +1,10 @@
+__pycache__
+*.pyc
+.git
+.env
+*.egg-info
+.venv
+venv
+benchmark_notebooks
+docs
+*.ipynb

.env.example

@@ -0,0 +1,29 @@
+# Classifier Service — Environment Variables
+
+
+# 1. Copy this file:  cp .env.example .env
+# 2. Fill in the values below.
+
+# REQUIRED — HuggingFace access token for the NER model.
+# Request access to tabiya/roberta-base-job-ner, then create a
+# read token at https://huggingface.co/settings/tokens
+HF_TOKEN=
+
+# API URLs (required for classify server)
+NER_API_URL=http://localhost:5002
+NEL_API_URL=http://localhost:5003
+CLASSIFY_API_URL=http://localhost:5001
+
+# Model Configuration
+NER_MODEL=tabiya/roberta-base-job-ner
+LINKER_MODEL=all-MiniLM-L6-v2
+
+# API Limits
+MAX_TEXT_LENGTH=50000
+MAX_BATCH_SIZE=500
+MAX_ENTITIES_PER_REQUEST=200
+MAX_TOP_K=50
+
+# MongoDB (only needed for change-stream worker)
+# APPLICATION_MONGODB_URI=mongodb+srv://<user>:<password>@<cluster>.mongodb.net/?retryWrites=true&w=majority
+#APPLICATION_DATABASE_NAME=db-name

Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+COPY pyproject.toml poetry.lock* ./
+RUN pip install --no-cache-dir poetry && \
+    poetry config virtualenvs.create false && \
+    poetry lock && \
+    poetry install --no-root --no-interaction --no-ansi && \
+    python -c "import nltk; nltk.download('punkt', quiet=True); nltk.download('punkt_tab', quiet=True)"
+
+COPY . .
+
+EXPOSE 5001 5002 5003
+
+CMD ["python", "app/server/classify_server.py"]

app/README.md

@@ -29,4 +29,56 @@ flask run --host=0.0.0.0 --port=5001
 
 3. **Click the "Analyze Job" button** to send the job description to the `/match` endpoint.
 
-4. **View the results** under "Predicted Occupations," "Predicted Skills," and "Predicted Qualifications."
+4. **View the results** under "Predicted Occupations," "Predicted Skills," and "Predicted Qualifications."
+
+---
+
+## Classification API (NER + NEL + Classify)
+
+A microservice-style API that splits the pipeline into three Flask servers:
+
+| Service | Port | Description |
+|---------|------|-------------|
+| NER API | 5002 | Extracts entity spans from job text |
+| NEL API | 5003 | Links entities to ESCO taxonomy via embedding similarity |
+| Classify API | 5001 | Orchestrator — calls NER then NEL and merges results |
+
+### Running the services
+
+```bash
+# Terminal 1
+python app/server/ner_server.py
+
+# Terminal 2
+python app/server/nel_server.py
+
+# Terminal 3
+python app/server/classify_server.py
+```
+
+Or with Docker:
+
+```bash
+docker compose up --build
+```
+
+### Quick test
+
+```bash
+curl -X POST http://localhost:5001/v1/classify \
+  -H "Content-Type: application/json" \
+  -d '{"text": "Head Chef with experience in menu planning."}'
+```
+
+### Change Stream Worker
+
+Watches MongoDB `raw-jobs` for new documents and classifies them automatically:
+
+```bash
+python app/worker/change_stream_worker.py
+python app/worker/change_stream_worker.py --backfill   # process existing unclassified jobs
+python app/worker/change_stream_worker.py --dry-run    # log only, no writes
+```
+
+Requires `APPLICATION_MONGODB_URI` in `.env` (see `.env.example`).
+

app/job_data_access.py

@@ -0,0 +1,200 @@
+from abc import ABC, abstractmethod
+from typing import Dict, List, Optional
+from datetime import datetime, timezone
+import logging
+import os
+
+from dotenv import load_dotenv
+
+from util.job_text import build_input_text, compute_hash
+
+load_dotenv()
+
+logger = logging.getLogger("repository")
+
+APPLICATION_MONGODB_URI = os.getenv("APPLICATION_MONGODB_URI") or os.getenv("MONGODB_URI", "mongodb://localhost:27017")
+APPLICATION_DATABASE_NAME = os.getenv("APPLICATION_DATABASE_NAME", "horizon-scraper-dev")
+
+
+#  Abstract Interface 
+
+class JobRepository(ABC):
+    """Interface that the classifier depends on.
+    Swap implementations to change databases without touching classifier code.
+    """
+
+    @abstractmethod
+    def get_job(self, fingerprint: str) -> Optional[Dict]:
+        ...
+
+    @abstractmethod
+    def get_unclassified_jobs(self, limit: int = 100, platform: Optional[str] = None) -> List[Dict]:
+        ...
+
+    @abstractmethod
+    def save_classification(self, result: Dict) -> None:
+        ...
+
+    @abstractmethod
+    def get_classification(self, fingerprint: str) -> Optional[Dict]:
+        ...
+
+    @abstractmethod
+    def is_already_classified(self, fingerprint: str, input_text_hash: str) -> bool:
+        ...
+
+    @abstractmethod
+    def get_all_classified_jobs(self, limit: int = 500, platform: Optional[str] = None) -> List[Dict]:
+        ...
+
+    @abstractmethod
+    def close(self) -> None:
+        ...
+
+
+# MongoDB Implementation (sync pymongo) 
+
+class MongoJobRepository(JobRepository):
+    """Synchronous MongoDB implementation using pymongo."""
+
+    def __init__(self, mongo_uri: Optional[str] = None, db_name: Optional[str] = None):
+        from pymongo import MongoClient
+
+        scraper_env = os.path.join(os.path.dirname(__file__), "..", "job_scraper", ".env")
+        if os.path.exists(scraper_env):
+            load_dotenv(scraper_env, override=False)
+
+        uri = mongo_uri or APPLICATION_MONGODB_URI
+        name = db_name or APPLICATION_DATABASE_NAME
+
+        self.client = MongoClient(uri)
+        self.db = self.client[name]
+        self.raw_jobs = self.db["raw-jobs"]
+        self.classified_jobs = self.db["classified-jobs"]
+
+    def get_job(self, fingerprint: str) -> Optional[Dict]:
+        return self.raw_jobs.find_one({"job_fingerprint": fingerprint})
+
+    def get_unclassified_jobs(self, limit: int = 100, platform: Optional[str] = None) -> List[Dict]:
+        match_stage: Dict = {"classification": {"$size": 0}}
+        if platform:
+            match_stage["sources.platform"] = platform
+
+        pipeline = [
+            {
+                "$lookup": {
+                    "from": "classified-jobs",
+                    "localField": "job_fingerprint",
+                    "foreignField": "job_fingerprint",
+                    "as": "classification",
+                }
+            },
+            {"$match": match_stage},
+            {"$project": {"classification": 0}},
+            {"$sort": {"created_at": -1}},
+            {"$limit": limit},
+        ]
+        return list(self.raw_jobs.aggregate(pipeline))
+
+    def save_classification(self, result: Dict) -> None:
+        now = datetime.now(timezone.utc)
+        result["updated_at"] = now
+        set_doc = {k: v for k, v in result.items() if k != "created_at"}
+
+        self.classified_jobs.update_one(
+            {"job_fingerprint": result["job_fingerprint"]},
+            {"$set": set_doc, "$setOnInsert": {"created_at": now}},
+            upsert=True,
+        )
+
+    def get_classification(self, fingerprint: str) -> Optional[Dict]:
+        return self.classified_jobs.find_one({"job_fingerprint": fingerprint})
+
+    def is_already_classified(self, fingerprint: str, input_text_hash: str) -> bool:
+        doc = self.classified_jobs.find_one({
+            "job_fingerprint": fingerprint,
+            "$or": [
+                {"input_text_hash": input_text_hash},
+                {"metadata.input_text_hash": input_text_hash},
+            ],
+        })
+        return doc is not None
+
+    def get_all_classified_jobs(self, limit: int = 500, platform: Optional[str] = None) -> List[Dict]:
+        query: Dict = {}
+        if platform:
+            query["source_fields.source_platform"] = platform
+        cursor = self.classified_jobs.find(query).sort("classified_at", -1).limit(limit)
+        return list(cursor)
+
+    def watch_raw_jobs(self):
+        """Yield full documents from the raw-jobs change stream."""
+        pipeline = [
+            {"$match": {"operationType": {"$in": ["insert", "replace", "update"]}}}
+        ]
+        with self.raw_jobs.watch(pipeline, full_document="updateLookup") as stream:
+            for change in stream:
+                doc = change.get("fullDocument")
+                if doc:
+                    yield doc
+
+    def close(self) -> None:
+        self.client.close()
+
+
+# In-Memory Implementation (for testing)
+
+class InMemoryJobRepository(JobRepository):
+    """Stores everything in Python dicts.  No database needed.
+    Useful for unit tests and local development without credentials.
+    """
+
+    def __init__(self):
+        self.raw_jobs: Dict[str, Dict] = {}
+        self.classifications: Dict[str, Dict] = {}
+
+    def get_job(self, fingerprint: str) -> Optional[Dict]:
+        return self.raw_jobs.get(fingerprint)
+
+    def get_unclassified_jobs(self, limit: int = 100, platform: Optional[str] = None) -> List[Dict]:
+        unclassified = [
+            job for fp, job in self.raw_jobs.items()
+            if fp not in self.classifications
+        ]
+        return unclassified[:limit]
+
+    def save_classification(self, result: Dict) -> None:
+        now = datetime.now(timezone.utc)
+        result.setdefault("created_at", now)
+        result["updated_at"] = now
+        self.classifications[result["job_fingerprint"]] = result
+
+    def get_classification(self, fingerprint: str) -> Optional[Dict]:
+        return self.classifications.get(fingerprint)
+
+    def is_already_classified(self, fingerprint: str, input_text_hash: str) -> bool:
+        doc = self.classifications.get(fingerprint)
+        if not doc:
+            return False
+        return doc.get("input_text_hash") == input_text_hash
+
+    def get_all_classified_jobs(self, limit: int = 500, platform: Optional[str] = None) -> List[Dict]:
+        results = list(self.classifications.values())
+        if platform:
+            results = [r for r in results if r.get("source_fields", {}).get("source_platform") == platform]
+        return results[:limit]
+
+    def add_raw_job(self, job: Dict) -> None:
+        """Helper to seed test data."""
+        self.raw_jobs[job["job_fingerprint"]] = job
+
+    def close(self) -> None:
+        pass
+
+
+# Helper
+
+def compute_input_text_hash(title: str, description: str) -> str:
+    """Compute SHA256 of the classifier input text for deduplication."""
+    text = build_input_text({"title": title, "description": description})
+    return compute_hash(text or "")