From a399478223dd9a56d0c7be4d2521c0efb13e62ee Mon Sep 17 00:00:00 2001 From: Hamz Date: Wed, 4 Mar 2026 12:00:07 -0500 Subject: [PATCH] feat: add NER, NEL, and Classify APIs with change-stream worker - Add standalone NER and NEL inference modules reusing EntityLinker helpers - Add Flask-based NER (port 5002), NEL (port 5003), and Classify (port 5001) API servers - Add change-stream worker to auto-classify new MongoDB raw-jobs - Add job data access layer (pymongo) with in-memory variant for testing - Add Docker and docker-compose setup for all services --- .dockerignore | 10 + .env.example | 29 +++ Dockerfile | 16 ++ app/README.md | 54 +++++- app/job_data_access.py | 200 ++++++++++++++++++++ app/server/classify_server.py | 289 +++++++++++++++++++++++++++++ app/server/common.py | 17 ++ app/server/nel_server.py | 99 ++++++++++ app/server/ner_server.py | 96 ++++++++++ app/worker/__init__.py | 0 app/worker/change_stream_worker.py | 196 +++++++++++++++++++ docker-compose.yaml | 87 +++++++++ inference/nel.py | 166 +++++++++++++++++ inference/ner.py | 101 ++++++++++ pyproject.toml | 4 + tests/__init__.py | 0 tests/test_job_data_access.py | 129 +++++++++++++ tests/test_ner_nel_split.py | 80 ++++++++ tests/test_worker_simulate.py | 185 ++++++++++++++++++ util/job_text.py | 31 ++++ 20 files changed, 1788 insertions(+), 1 deletion(-) create mode 100644 .dockerignore create mode 100644 .env.example create mode 100644 Dockerfile create mode 100644 app/job_data_access.py create mode 100644 app/server/classify_server.py create mode 100644 app/server/common.py create mode 100644 app/server/nel_server.py create mode 100644 app/server/ner_server.py create mode 100644 app/worker/__init__.py create mode 100644 app/worker/change_stream_worker.py create mode 100644 docker-compose.yaml create mode 100644 inference/nel.py create mode 100644 inference/ner.py create mode 100644 tests/__init__.py create mode 100644 tests/test_job_data_access.py create mode 100644 tests/test_ner_nel_split.py create mode 100644 tests/test_worker_simulate.py create mode 100644 util/job_text.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..4b71bec --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +__pycache__ +*.pyc +.git +.env +*.egg-info +.venv +venv +benchmark_notebooks +docs +*.ipynb diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..18406ae --- /dev/null +++ b/.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://:@.mongodb.net/?retryWrites=true&w=majority +#APPLICATION_DATABASE_NAME=db-name diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..ef505fe --- /dev/null +++ b/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"] diff --git a/app/README.md b/app/README.md index 88c72b4..11846a0 100644 --- a/app/README.md +++ b/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." \ No newline at end of file +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`). + diff --git a/app/job_data_access.py b/app/job_data_access.py new file mode 100644 index 0000000..b586c82 --- /dev/null +++ b/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 "") diff --git a/app/server/classify_server.py b/app/server/classify_server.py new file mode 100644 index 0000000..f35bb43 --- /dev/null +++ b/app/server/classify_server.py @@ -0,0 +1,289 @@ +import sys +import os +import time +import uuid +import logging +import threading + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +from flask import request, jsonify +from dotenv import load_dotenv +import requests as http_requests + +from app.server.common import setup_logging, create_app +from util.job_text import build_input_text, compute_hash + +load_dotenv() +setup_logging() +log = logging.getLogger("classify-api") + +NER_API_URL = os.getenv("NER_API_URL") +NEL_API_URL = os.getenv("NEL_API_URL") +CLASSIFIER_VERSION = os.getenv("CLASSIFIER_VERSION", "1.0.0") +MAX_TEXT_LENGTH = int(os.getenv("MAX_TEXT_LENGTH", "50000")) +MAX_BATCH_SIZE = int(os.getenv("MAX_BATCH_SIZE", "500")) + +app = create_app("classify-api", __name__) + +_batches = {} + + + +# Core classification logic + +def _classify_text(input_text, options=None): + """Call NER then NEL and merge results (synchronous, uses requests).""" + options = options or {} + entity_types = options.get("extract_entities") + top_k = options.get("top_k", 5) + min_similarity = options.get("min_similarity", 0.0) + + start = time.time() + + ner_payload = {"text": input_text} + if entity_types: + ner_payload["entity_types"] = entity_types + + ner_resp = http_requests.post(f"{NER_API_URL}/v1/ner", json=ner_payload, timeout=60) + ner_resp.raise_for_status() + ner_data = ner_resp.json() + ner_entities = ner_data.get("entities", []) + + linkable_types = {"occupation", "skill", "qualification"} + nel_input = [ + {"text": e["surface_form"], "entity_type": e["entity_type"]} + for e in ner_entities + if e["entity_type"] in linkable_types + ] + + linked_map = {} + nel_metadata = {} + if nel_input: + nel_resp = http_requests.post( + f"{NEL_API_URL}/v1/nel", + json={"entities": nel_input, "options": {"top_k": top_k, "min_similarity": min_similarity}}, + timeout=60, + ) + nel_resp.raise_for_status() + nel_data = nel_resp.json() + nel_metadata = nel_data.get("metadata", {}) + + for item in nel_data.get("linked_entities", []): + key = (item["input_text"], item["entity_type"]) + linked_map[key] = item["matches"] + + merged_entities = [] + entity_counts = {} + + for entity in ner_entities: + etype = entity["entity_type"] + entity_counts[etype] = entity_counts.get(etype, 0) + 1 + + merged = { + "entity_type": etype, + "surface_form": entity["surface_form"], + "span": entity["span"], + } + + key = (entity["surface_form"], etype) + if key in linked_map: + merged["linked_entities"] = linked_map[key] + + merged_entities.append(merged) + + processing_time = round((time.time() - start) * 1000, 1) + input_text_hash = compute_hash(input_text) + + return { + "classification": { + "entities": merged_entities, + "entity_counts": entity_counts, + }, + "metadata": { + "classifier_version": CLASSIFIER_VERSION, + "model_name": ner_data.get("metadata", {}).get("model_name", "unknown"), + "linker_model": nel_metadata.get("linker_model", "unknown"), + "processing_time_ms": processing_time, + "input_text_hash": input_text_hash, + }, + } + + + +# Single classify endpoint + + +@app.route("/v1/classify", methods=["POST"]) +def classify(): + data = request.get_json(silent=True) or {} + + input_text = build_input_text(data, allow_text_field=True) + if not input_text: + return jsonify({"error": "Provide 'text' or 'title'+'description'"}), 400 + + if len(input_text) > MAX_TEXT_LENGTH: + return jsonify({ + "error": f"Text exceeds maximum length ({MAX_TEXT_LENGTH} chars)" + }), 413 + + log.info("Classify request: %d chars", len(input_text)) + + try: + result = _classify_text(input_text, data.get("options")) + except http_requests.HTTPError as e: + log.error("Downstream error: %s", e) + return jsonify({"error": f"Downstream API error: {e}"}), 502 + + entity_count = sum(result["classification"]["entity_counts"].values()) + log.info("Classify done: %d entities in %.1fms", entity_count, result["metadata"]["processing_time_ms"]) + return jsonify(result) + + +# Batch endpoints + + +def _process_batch(batch_id, jobs, options): + """Background thread: classify each job and update batch state.""" + batch = _batches[batch_id] + + for i, job in enumerate(jobs): + input_text = build_input_text(job, allow_text_field=True) + job_id = job.get("job_id", f"job_{i}") + + if not input_text: + batch["results"].append({ + "job_id": job_id, + "status": "error", + "error": "No classifiable text found", + }) + elif len(input_text) > MAX_TEXT_LENGTH: + batch["results"].append({ + "job_id": job_id, + "status": "error", + "error": f"Text exceeds {MAX_TEXT_LENGTH} char limit", + }) + else: + try: + result = _classify_text(input_text, options) + batch["results"].append({ + "job_id": job_id, + "status": "completed", + **result, + }) + except Exception as e: + log.error("[batch-%s] Job %s failed: %s", batch_id, job_id, e) + batch["results"].append({ + "job_id": job_id, + "status": "error", + "error": str(e), + }) + + batch["processed"] = i + 1 + + batch["status"] = "completed" + batch["completed_at"] = time.time() + + +@app.route("/v1/classify/batch", methods=["POST"]) +def submit_batch(): + data = request.get_json(silent=True) or {} + jobs = data.get("jobs", []) + options = data.get("options") + + if not jobs: + return jsonify({"error": "Field 'jobs' is required and must be non-empty"}), 400 + + if len(jobs) > MAX_BATCH_SIZE: + return jsonify({ + "error": f"Batch too large ({len(jobs)} jobs). Maximum is {MAX_BATCH_SIZE}." + }), 413 + + batch_id = str(uuid.uuid4())[:8] + _batches[batch_id] = { + "status": "processing", + "total": len(jobs), + "processed": 0, + "results": [], + "created_at": time.time(), + "completed_at": None, + } + + thread = threading.Thread( + target=_process_batch, args=(batch_id, jobs, options), daemon=True + ) + thread.start() + + log.info("Batch %s submitted: %d jobs", batch_id, len(jobs)) + return jsonify({"batch_id": batch_id, "total": len(jobs), "status": "processing"}), 202 + + +@app.route("/v1/batch//status", methods=["GET"]) +def batch_status(batch_id): + batch = _batches.get(batch_id) + if not batch: + return jsonify({"error": "Batch not found"}), 404 + + return jsonify({ + "batch_id": batch_id, + "status": batch["status"], + "total": batch["total"], + "processed": batch["processed"], + }) + + +@app.route("/v1/batch//results", methods=["GET"]) +def batch_results(batch_id): + batch = _batches.get(batch_id) + if not batch: + return jsonify({"error": "Batch not found"}), 404 + + return jsonify({ + "batch_id": batch_id, + "status": batch["status"], + "total": batch["total"], + "processed": batch["processed"], + "results": batch["results"], + }) + + +# Health / version + +@app.route("/v1/health", methods=["GET"]) +def health(): + ner_ok = False + nel_ok = False + + try: + r = http_requests.get(f"{NER_API_URL}/v1/health", timeout=5) + ner_ok = r.status_code == 200 + except Exception: + pass + + try: + r = http_requests.get(f"{NEL_API_URL}/v1/health", timeout=5) + nel_ok = r.status_code == 200 + except Exception: + pass + + overall = "healthy" if (ner_ok and nel_ok) else "degraded" + return jsonify({ + "status": overall, + "service": "classify-api", + "dependencies": { + "ner_api": "healthy" if ner_ok else "unavailable", + "nel_api": "healthy" if nel_ok else "unavailable", + }, + }) + + +@app.route("/v1/version", methods=["GET"]) +def version(): + return jsonify({ + "service": "classify-api", + "version": CLASSIFIER_VERSION, + }) + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=5001, debug=True) diff --git a/app/server/common.py b/app/server/common.py new file mode 100644 index 0000000..9f555a6 --- /dev/null +++ b/app/server/common.py @@ -0,0 +1,17 @@ +import logging + +from flask import Flask +from flask_cors import CORS + +LOG_FORMAT = "%(asctime)s [%(levelname)s] [%(name)s] %(message)s" + + +def setup_logging() -> None: + logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) + + +def create_app(name: str, import_name: str) -> Flask: + """Create a Flask app with CORS enabled (same pattern as matching.py).""" + app = Flask(import_name) + CORS(app, resources={r"/*": {"origins": "*"}}) + return app diff --git a/app/server/nel_server.py b/app/server/nel_server.py new file mode 100644 index 0000000..7e68270 --- /dev/null +++ b/app/server/nel_server.py @@ -0,0 +1,99 @@ +import sys +import os +import time +import logging + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +from flask import request, jsonify +from dotenv import load_dotenv + +from app.server.common import setup_logging, create_app + +load_dotenv() +setup_logging() +log = logging.getLogger("nel-api") + +LINKER_MODEL = os.getenv("LINKER_MODEL", "all-MiniLM-L6-v2") +MAX_ENTITIES_PER_REQUEST = int(os.getenv("MAX_ENTITIES_PER_REQUEST", "200")) +MAX_TOP_K = int(os.getenv("MAX_TOP_K", "50")) + +app = create_app("nel-api", __name__) + +nel_linker = None +_linker_load_error = None + + +def _load_linker(): + global nel_linker, _linker_load_error + if nel_linker is not None or _linker_load_error is not None: + return + try: + from inference.nel import NELLinker + nel_linker = NELLinker(similarity_model=LINKER_MODEL) + log.info("NEL linker loaded: %s", LINKER_MODEL) + except Exception as e: + _linker_load_error = str(e) + log.error("Failed to load NEL linker: %s", e) + + +@app.route("/v1/nel", methods=["POST"]) +def link_entities(): + data = request.get_json(silent=True) or {} + entities = data.get("entities", []) + options = data.get("options", {}) + + if not entities: + return jsonify({"error": "Field 'entities' is required and must be non-empty"}), 400 + + if len(entities) > MAX_ENTITIES_PER_REQUEST: + return jsonify({ + "error": f"Too many entities ({len(entities)}). Maximum is {MAX_ENTITIES_PER_REQUEST}." + }), 413 + + top_k = min(options.get("top_k", 5), MAX_TOP_K) + min_similarity = options.get("min_similarity", 0.0) + + if nel_linker is None: + return jsonify({"error": _linker_load_error or "NEL linker not loaded"}), 503 + + log.info("NEL request: %d entities, top_k=%d", len(entities), top_k) + start = time.time() + + try: + results = nel_linker.link(entities, top_k=top_k, min_similarity=min_similarity) + except Exception as e: + log.error("NEL linking failed: %s", e) + return jsonify({"error": f"Entity linking failed: {e}"}), 500 + + processing_time = round((time.time() - start) * 1000, 1) + log.info("NEL done: %d linked in %.1fms", len(results), processing_time) + + return jsonify({ + "linked_entities": results, + "metadata": { + "linker_model": nel_linker.similarity_model_name, + "taxonomy": "esco", + "processing_time_ms": processing_time, + }, + }) + + +@app.route("/v1/health", methods=["GET"]) +def health(): + linker_ok = nel_linker is not None + resp = { + "status": "healthy" if linker_ok else "unavailable", + "service": "nel-api", + "model_loaded": linker_ok, + } + if linker_ok: + resp["linker_model"] = nel_linker.similarity_model_name + if _linker_load_error: + resp["error"] = _linker_load_error + return jsonify(resp), 200 if linker_ok else 503 + + +if __name__ == "__main__": + _load_linker() + app.run(host="0.0.0.0", port=5003, debug=True) diff --git a/app/server/ner_server.py b/app/server/ner_server.py new file mode 100644 index 0000000..577b51a --- /dev/null +++ b/app/server/ner_server.py @@ -0,0 +1,96 @@ +import sys +import os +import time +import logging + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +from flask import request, jsonify +from dotenv import load_dotenv + +from app.server.common import setup_logging, create_app + +load_dotenv() +setup_logging() +log = logging.getLogger("ner-api") + +NER_MODEL = os.getenv("NER_MODEL", "tabiya/roberta-base-job-ner") +MAX_TEXT_LENGTH = int(os.getenv("MAX_TEXT_LENGTH", "50000")) + +app = create_app("ner-api", __name__) + +ner_model = None +_model_load_error = None + + +def _load_model(): + global ner_model, _model_load_error + if ner_model is not None or _model_load_error is not None: + return + try: + from inference.ner import NERModel + ner_model = NERModel(model_name=NER_MODEL) + log.info("NER model loaded: %s", NER_MODEL) + except Exception as e: + _model_load_error = str(e) + log.error("Failed to load NER model: %s", e) + + +@app.route("/v1/ner", methods=["POST"]) +def extract_entities(): + data = request.get_json(silent=True) or {} + text = data.get("text", "") + + if not text: + return jsonify({"error": "Field 'text' is required"}), 400 + + if len(text) > MAX_TEXT_LENGTH: + return jsonify({"error": f"Text exceeds maximum length ({MAX_TEXT_LENGTH} chars)"}), 413 + + if ner_model is None: + return jsonify({"error": _model_load_error or "NER model not loaded"}), 503 + + start = time.time() + try: + entities = ner_model.extract(text) + except Exception as e: + log.error("NER inference failed: %s", e) + return jsonify({"error": f"Model inference failed: {e}"}), 500 + + processing_time = round((time.time() - start) * 1000, 1) + + entity_types = data.get("entity_types") + if entity_types: + allowed = {t.lower() for t in entity_types} + entities = [e for e in entities if e["entity_type"] in allowed] + + log.info("NER done: %d entities in %.1fms", len(entities), processing_time) + + return jsonify({ + "entities": entities, + "metadata": { + "model_name": ner_model.model_name, + "entity_count": len(entities), + "processing_time_ms": processing_time, + }, + }) + + +@app.route("/v1/health", methods=["GET"]) +def health(): + model_ok = ner_model is not None + resp = { + "status": "healthy" if model_ok else "unavailable", + "service": "ner-api", + "model_loaded": model_ok, + } + if model_ok: + resp["model_name"] = ner_model.model_name + if _model_load_error: + resp["error"] = _model_load_error + return jsonify(resp), 200 if model_ok else 503 + + +if __name__ == "__main__": + _load_model() + app.run(host="0.0.0.0", port=5002, debug=True) diff --git a/app/worker/__init__.py b/app/worker/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/worker/change_stream_worker.py b/app/worker/change_stream_worker.py new file mode 100644 index 0000000..0a4d780 --- /dev/null +++ b/app/worker/change_stream_worker.py @@ -0,0 +1,196 @@ +import sys +import os +import time +import logging +import argparse +from typing import Optional + +import requests + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +from dotenv import load_dotenv + +from app.server.common import setup_logging +from util.job_text import build_input_text, compute_hash + +load_dotenv() +scraper_env = os.path.join(os.path.dirname(__file__), "..", "..", "..", "job_scraper", ".env") +if os.path.exists(scraper_env): + load_dotenv(scraper_env, override=False) + +setup_logging() +log = logging.getLogger("change-stream-worker") + +CLASSIFY_API_URL = os.getenv("CLASSIFY_API_URL", "http://localhost:5001") +MAX_RETRIES = int(os.getenv("WORKER_MAX_RETRIES", "0")) +RETRY_DELAY = int(os.getenv("WORKER_RETRY_DELAY_SECONDS", "10")) + +ALL_SOURCE_FIELDS = [ + "title", "employer", "location", "description", + "employment_type", "education", "experience", + "salary_text", "posted_date", "closing_date", + "application_url", "source_platform", +] + + +def extract_source_fields(doc: dict) -> dict: + """Pull all known source fields from a raw-jobs document.""" + fields = {} + for key in ALL_SOURCE_FIELDS: + val = doc.get(key) + if val is not None: + fields[key] = val + if "salary_text" in fields: + fields["salary"] = fields.pop("salary_text") + if "closing_date" in fields: + fields["expiry_date"] = fields.pop("closing_date") + return fields + + +def classify_job(input_text: str, job_fingerprint: str, dry_run: bool = False) -> Optional[dict]: + """Send a job to the Classify API. Returns the response dict or None on failure.""" + if dry_run: + log.info(" [DRY-RUN] Would classify job %s: %s...", job_fingerprint, input_text[:80]) + return {"dry_run": True} + + try: + resp = requests.post( + f"{CLASSIFY_API_URL}/v1/classify", + json={"text": input_text}, + timeout=60, + ) + resp.raise_for_status() + return resp.json() + except requests.RequestException as e: + log.error(" Classify API error for %s: %s", job_fingerprint, e) + return None + + +def watch_change_stream(dry_run: bool = False): + """Connect to MongoDB via JobRepository and watch raw-jobs for changes.""" + from app.job_data_access import MongoJobRepository + + repo = MongoJobRepository() + log.info("Watching raw-jobs change stream... (Ctrl+C to stop)") + + try: + for doc in repo.watch_raw_jobs(): + fingerprint = doc.get("job_fingerprint", "unknown") + input_text = build_input_text(doc) + + if not input_text: + log.warning(" Skipping %s — no title or description", fingerprint) + continue + + input_hash = compute_hash(input_text) + + if repo.is_already_classified(fingerprint, input_hash): + log.info(" Skipping %s — already classified (hash match)", fingerprint) + continue + + log.info(" New job detected: %s", fingerprint) + result = classify_job(input_text, fingerprint, dry_run=dry_run) + + if result and not dry_run: + classification_doc = { + "job_fingerprint": fingerprint, + "input_text_hash": input_hash, + "classification": result.get("classification", {}), + "metadata": result.get("metadata", {}), + "source_fields": extract_source_fields(doc), + "classified_at": time.time(), + "status": "completed", + } + repo.save_classification(classification_doc) + log.info(" Saved classification for %s", fingerprint) + elif result: + log.info(" [DRY-RUN] Classification complete for %s", fingerprint) + finally: + repo.close() + + +def backfill_unclassified(dry_run: bool = False, platform: Optional[str] = None): + """Batch-process any raw-jobs that haven't been classified yet.""" + from app.job_data_access import MongoJobRepository + + repo = MongoJobRepository() + log.info("Backfilling unclassified jobs (platform=%s)...", platform or "all") + + try: + jobs = repo.get_unclassified_jobs(limit=500, platform=platform) + log.info("Found %d unclassified jobs", len(jobs)) + + for i, doc in enumerate(jobs, 1): + fingerprint = doc.get("job_fingerprint", "unknown") + input_text = build_input_text(doc) + + if not input_text: + log.warning(" [%d/%d] Skipping %s — no text", i, len(jobs), fingerprint) + continue + + input_hash = compute_hash(input_text) + + if repo.is_already_classified(fingerprint, input_hash): + log.info(" [%d/%d] Skipping %s — already classified", i, len(jobs), fingerprint) + continue + + log.info(" [%d/%d] Classifying: %s", i, len(jobs), fingerprint) + result = classify_job(input_text, fingerprint, dry_run=dry_run) + + if result and not dry_run: + classification_doc = { + "job_fingerprint": fingerprint, + "input_text_hash": input_hash, + "classification": result.get("classification", {}), + "metadata": result.get("metadata", {}), + "source_fields": extract_source_fields(doc), + "classified_at": time.time(), + "status": "completed", + } + repo.save_classification(classification_doc) + log.info(" Saved classification for %s", fingerprint) + elif result: + log.info(" [%d/%d] [DRY-RUN] done", i, len(jobs)) + + time.sleep(0.2) + + log.info("Backfill complete") + finally: + repo.close() + + +def main(): + parser = argparse.ArgumentParser(description="Change Stream Worker") + parser.add_argument("--dry-run", action="store_true", help="Log only, don't write to DB") + parser.add_argument("--backfill", action="store_true", help="Batch-classify unclassified jobs then exit") + parser.add_argument("--platform", type=str, help="Filter backfill to a specific platform") + args = parser.parse_args() + + if args.dry_run: + log.info("Running in DRY-RUN mode — no writes to MongoDB") + + if args.backfill: + backfill_unclassified(dry_run=args.dry_run, platform=args.platform) + return + + retries = 0 + while True: + try: + watch_change_stream(dry_run=args.dry_run) + break + except KeyboardInterrupt: + log.info("Worker stopped by user.") + break + except Exception as e: + retries += 1 + if MAX_RETRIES > 0 and retries > MAX_RETRIES: + log.error("Max retries (%d) exceeded. Exiting.", MAX_RETRIES) + break + log.error("Worker error (attempt %d): %s", retries, e) + log.info("Reconnecting in %ds...", RETRY_DELAY) + time.sleep(RETRY_DELAY) + + +if __name__ == "__main__": + main() diff --git a/docker-compose.yaml b/docker-compose.yaml new file mode 100644 index 0000000..069202e --- /dev/null +++ b/docker-compose.yaml @@ -0,0 +1,87 @@ +services: + + # NER API — entity extraction (port 5002) + + ner-api: + build: . + command: python app/server/ner_server.py + ports: + - "5002:5002" + environment: + - HF_TOKEN=${HF_TOKEN:-} + volumes: + - model-cache:/root/.cache + restart: unless-stopped + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:5002/v1/health')"] + interval: 30s + timeout: 10s + retries: 10 + start_period: 300s + + + # NEL API — entity linking to ESCO (port 5003) + + nel-api: + build: . + command: python app/server/nel_server.py + ports: + - "5003:5003" + environment: + - HF_TOKEN=${HF_TOKEN:-} + volumes: + - model-cache:/root/.cache + restart: unless-stopped + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:5003/v1/health')"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 60s + + + # Classify API — orchestrator (port 5001) + # Calls ner-api then nel-api, serves /v1/classify + batch endpoints + + classify-api: + build: . + command: python app/server/classify_server.py + ports: + - "5001:5001" + environment: + - NER_API_URL=http://ner-api:5002 + - NEL_API_URL=http://nel-api:5003 + depends_on: + ner-api: + condition: service_healthy + nel-api: + condition: service_healthy + restart: unless-stopped + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:5001/v1/health')"] + interval: 30s + timeout: 10s + retries: 3 + start_period: 10s + + + # Change Stream Worker — watches MongoDB, feeds classify-api + # Only runs if MONGODB_URI is set. Activate with: + # docker compose --profile with-mongodb up + + change-stream-worker: + build: . + command: python app/worker/change_stream_worker.py + environment: + - CLASSIFY_API_URL=http://classify-api:5001 + - APPLICATION_MONGODB_URI=${APPLICATION_MONGODB_URI:-} + - APPLICATION_DATABASE_NAME=${APPLICATION_DATABASE_NAME:-horizon-scraper-dev} + depends_on: + classify-api: + condition: service_healthy + restart: unless-stopped + profiles: + - with-mongodb + +volumes: + model-cache: diff --git a/inference/nel.py b/inference/nel.py new file mode 100644 index 0000000..5f806b8 --- /dev/null +++ b/inference/nel.py @@ -0,0 +1,166 @@ +from typing import List, Optional, Tuple +import pickle +import torch +from sentence_transformers import SentenceTransformer, util +import pandas as pd +import os +from dotenv import load_dotenv + +from inference.linker import EntityLinker + +load_dotenv() + + +class NELLinker: + """Links entity text to ESCO taxonomy entries using embedding similarity.""" + + def __init__( + self, + similarity_model: str = "all-MiniLM-L6-v2", + k: int = 32, + from_cache: bool = True, + ): + self.similarity_model_name = similarity_model + self.similarity_model = SentenceTransformer(similarity_model) + self.k = k + self.from_cache = from_cache + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + self.path_to_files = os.path.abspath( + os.path.join(os.path.dirname(__file__), "files") + ) + + self.df_occ = pd.read_csv(os.path.join(self.path_to_files, "occupations_augmented.csv")) + self.df_skill = pd.read_csv(os.path.join(self.path_to_files, "skills.csv")) + self.df_qual = pd.read_csv(os.path.join(self.path_to_files, "qualifications.csv")) + + self.occupation_emb, self.skill_emb, self.qualification_emb = self._load_tensors() + + def link( + self, + entities: List[dict], + top_k: Optional[int] = None, + min_similarity: float = 0.0, + ) -> List[dict]: + """Link entities to ESCO taxonomy entries. + + Each entity dict needs ``text`` and ``entity_type`` keys. + """ + k = top_k or self.k + results = [] + + for entity in entities: + text = entity["text"] + entity_type = entity["entity_type"].lower() + + if entity_type not in ("occupation", "skill", "qualification"): + results.append({ + "input_text": text, + "entity_type": entity_type, + "matches": [], + }) + continue + + emb = self.similarity_model.encode(text) + emb = torch.from_numpy(emb).to(self.device) + + matches = self._top_k(emb, entity_type, k, min_similarity) + + results.append({ + "input_text": text, + "entity_type": entity_type, + "matches": matches, + }) + + return results + + def _top_k( + self, + embedding: torch.Tensor, + entity_type: str, + k: int, + min_similarity: float, + ) -> List[dict]: + """Retrieve top-k ESCO matches for a single entity embedding.""" + if entity_type == "occupation": + local_df = self.df_occ + local_emb = self.occupation_emb + elif entity_type == "qualification": + local_df = self.df_qual + local_emb = self.qualification_emb + else: + local_df = self.df_skill + local_emb = self.skill_emb + + cos_scores = util.cos_sim(embedding, local_emb)[0] + top_k_results = torch.topk(cos_scores, k=min(k, len(cos_scores))) + + matches = [] + for idx, score in zip( + top_k_results.indices.tolist(), top_k_results.values.tolist() + ): + if score < min_similarity: + continue + + row = local_df.iloc[idx] + match = { + "similarity_score": round(score, 4), + "taxonomy": "esco", + } + + if entity_type == "occupation": + match["label"] = row.get("occupation", row.get("preffered_label", "")) + if "esco_code" in row: + match["code"] = str(row["esco_code"]) + if "uuid" in row: + match["uri"] = f"http://data.europa.eu/esco/occupation/{row['uuid']}" + elif entity_type == "skill": + match["label"] = row.get("skills", "") + if "uuid" in row: + match["uri"] = f"http://data.europa.eu/esco/skill/{row['uuid']}" + elif entity_type == "qualification": + match["label"] = row.get("qualification", "") + if "eqf_level" in row: + match["eqf_level"] = str(row["eqf_level"]) + + matches.append(match) + + return matches + + def _load_tensors(self) -> Tuple: + """Load precomputed or compute fresh embeddings for all reference sets. + + uses EntityLinker.create_tensors() for the cached-load path. + """ + path = os.path.join(self.path_to_files, self.similarity_model_name) + + if self.from_cache: + occupation_emb = EntityLinker.create_tensors( + os.path.join(path, "occupations.pkl"), self.device + ) + skill_emb = EntityLinker.create_tensors( + os.path.join(path, "skills.pkl"), self.device + ) + qualification_emb = EntityLinker.create_tensors( + os.path.join(path, "qualifications.pkl"), self.device + ) + else: + os.makedirs(path, exist_ok=True) + occupation_emb = self._compute_and_cache( + list(self.df_occ["occupation"]), "occupations", path + ) + skill_emb = self._compute_and_cache( + list(self.df_skill["skills"]), "skills", path + ) + qualification_emb = self._compute_and_cache( + list(self.df_qual["qualification"]), "qualifications", path + ) + + return occupation_emb, skill_emb, qualification_emb + + def _compute_and_cache(self, corpus: List[str], name: str, path: str) -> torch.Tensor: + """Encode corpus, cache to disk, and return the tensor.""" + embeddings = self.similarity_model.encode(corpus, convert_to_tensor=True) + filepath = os.path.join(path, f"{name}.pkl") + with open(filepath, "wb") as f: + pickle.dump(embeddings, f) + return EntityLinker.create_tensors(filepath, self.device) diff --git a/inference/ner.py b/inference/ner.py new file mode 100644 index 0000000..25ae40e --- /dev/null +++ b/inference/ner.py @@ -0,0 +1,101 @@ +from typing import List +import os + +import torch +from transformers import AutoModelForTokenClassification, AutoTokenizer +from nltk.tokenize import sent_tokenize +from dotenv import load_dotenv + +from util.transformersCRF import AutoModelCrfForNer +from inference.linker import EntityLinker + +load_dotenv() + +HF_TOKEN = os.getenv("HF_TOKEN") + + +class NERModel: + """Extracts entity spans from job-related text using a fine-tuned transformer.""" + + def __init__( + self, + model_name: str = "tabiya/roberta-base-job-ner", + crf: bool = False, + ): + self.model_name = model_name + self.crf = crf + self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + + if self.crf: + self.model = AutoModelCrfForNer.from_pretrained(model_name) + else: + self.model = AutoModelForTokenClassification.from_pretrained( + model_name, token=HF_TOKEN + ) + + self.model.to(self.device) + self.tokenizer = AutoTokenizer.from_pretrained( + model_name, token=HF_TOKEN + ) + + def extract(self, text: str) -> List[dict]: + """Extract entities from text, returning entity_type, surface_form, and span for each.""" + text = text.replace("\n", " ") + sentences = sent_tokenize(text) + all_entities: List[dict] = [] + char_offset = 0 + + for sentence in sentences: + sent_start = text.find(sentence, char_offset) + raw_entities = self._ner_pipeline(sentence) + + for entity in raw_entities: + surface = entity["tokens"] + entity_start = text.find(surface, sent_start) + entity_end = entity_start + len(surface) if entity_start != -1 else sent_start + + all_entities.append({ + "entity_type": entity["type"].lower(), + "surface_form": surface, + "span": { + "start": max(entity_start, 0), + "end": max(entity_end, 0), + }, + }) + + char_offset = sent_start + len(sentence) + + return all_entities + + def _ner_pipeline(self, text: str) -> List[dict]: + """Run NER on a single sentence, reusing EntityLinker's static helpers.""" + inputs = self.tokenizer(text, return_tensors="pt", truncation=True).to(self.device) + + if self.crf: + with torch.no_grad(): + logits = self.model(**inputs) + predictions = logits[1][0] + else: + with torch.no_grad(): + logits = self.model(**inputs).logits + predictions = torch.argmax(logits, dim=2) + + predicted_tags = [ + self.model.config.id2label[t.item()] for t in predictions[0] + ] + + predicted_tags = EntityLinker.fix_bio_tags(predicted_tags) + + input_ids, predicted_tags = EntityLinker.remove_special_tokens_and_tags( + inputs["input_ids"][0], predicted_tags, self.tokenizer + ) + + result = EntityLinker.extract_entities(input_ids, predicted_tags) + + for entry in result: + sentence = self.tokenizer.decode(entry["tokens"]) + if sentence.startswith(" "): + sentence = sentence[1:] + entry["tokens"] = sentence + + return result diff --git a/pyproject.toml b/pyproject.toml index bfd658b..0fa30fb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,10 @@ tqdm = "^4.65.0" flask = "^3.0.3" flask-cors = "^4.0.1" python-dotenv = "^1.0.1" +transformers = "^4.32.1" +requests = "^2.31.0" +pymongo = "^4.6" +dnspython = "^2.6.0" [tool.poetry.group.train.dependencies] transformers= "^4.32.1" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_job_data_access.py b/tests/test_job_data_access.py new file mode 100644 index 0000000..c1d4383 --- /dev/null +++ b/tests/test_job_data_access.py @@ -0,0 +1,129 @@ +""" +Local test: Verify Repository layer works. + - Part A: InMemoryJobRepository (no database needed) + - Part B: MongoJobRepository (reads from your Atlas raw-jobs) + +All synchronous — matches the Flask-based codebase. +Run from the classifier root: python -m tests.test_repository +""" + + +def test_in_memory(): + """Test the InMemory repository — no credentials, no database.""" + from app.job_data_access import InMemoryJobRepository, compute_input_text_hash + + print("\n[Part A] Testing InMemoryJobRepository...") + repo = InMemoryJobRepository() + + fake_job = { + "job_fingerprint": "test123", + "title": "Head Chef", + "description": "We need someone who can plan menus and manage kitchen staff.", + "employer": "Sarova Hotels", + "location": "Nairobi", + } + repo.add_raw_job(fake_job) + + job = repo.get_job("test123") + assert job is not None, "get_job failed" + assert job["title"] == "Head Chef" + print(" ok get_job — found job by fingerprint") + + missing = repo.get_job("nonexistent") + assert missing is None + print(" ok get_job — returns None for missing fingerprint") + + unclassified = repo.get_unclassified_jobs() + assert len(unclassified) == 1 + print(f" ok get_unclassified_jobs — found {len(unclassified)} unclassified job(s)") + + text_hash = compute_input_text_hash(fake_job["title"], fake_job["description"]) + classification = { + "job_fingerprint": "test123", + "source_job": { + "title": fake_job["title"], + "description": fake_job["description"], + }, + "classification": { + "entities": [{"entity_type": "occupation", "surface_form": "Head Chef"}], + "entity_counts": {"occupation": 1, "skill": 2}, + }, + "metadata": { + "classifier_version": "1.0.0", + "input_text_hash": text_hash, + }, + "status": "completed", + } + repo.save_classification(classification) + print(" ok save_classification — saved without error") + + result = repo.get_classification("test123") + assert result is not None + assert result["status"] == "completed" + print(" ok get_classification — retrieved saved classification") + + is_done = repo.is_already_classified("test123", text_hash) + assert is_done is True + print(" ok is_already_classified — returns True for same text hash") + + is_done_diff = repo.is_already_classified("test123", "different_hash") + assert is_done_diff is False + print(" ok is_already_classified — returns False for different text hash") + + unclassified = repo.get_unclassified_jobs() + assert len(unclassified) == 0 + print(f" ok get_unclassified_jobs — {len(unclassified)} unclassified after classification") + + repo.close() + print("\n Part A PASSED — all InMemory tests passed") + + +def test_mongo(): + """Test the Mongo repository — reads from Atlas.""" + from app.job_data_access import MongoJobRepository + + print("\n[Part B] Testing MongoJobRepository (Atlas)...") + repo = MongoJobRepository() + + print(" Connecting to Atlas and reading from raw-jobs...") + sample = repo.raw_jobs.find_one() + + if sample is None: + print(" WARNING: raw-jobs collection is empty — skipping Mongo tests") + repo.close() + return + + fp = sample.get("job_fingerprint", "unknown") + title = sample.get("title", "no title") + print(f" ok Connected — found job: \"{title}\" (fingerprint: {fp[:12]}...)") + + job = repo.get_job(fp) + assert job is not None + print(" ok get_job — retrieved by fingerprint") + + unclassified = repo.get_unclassified_jobs(limit=5) + print(f" ok get_unclassified_jobs — found {len(unclassified)} unclassified job(s)") + + raw_count = repo.raw_jobs.count_documents({}) + classified_count = repo.classified_jobs.count_documents({}) + print(f" info raw-jobs: {raw_count} documents, classified-jobs: {classified_count} documents") + + repo.close() + print("\n Part B PASSED — Mongo connection and reads working") + + +def main(): + print("=" * 60) + print("TEST: Repository Layer") + print("=" * 60) + + test_in_memory() + test_mongo() + + print("\n" + "=" * 60) + print("REPOSITORY TESTS COMPLETE") + print("=" * 60) + + +if __name__ == "__main__": + main() diff --git a/tests/test_ner_nel_split.py b/tests/test_ner_nel_split.py new file mode 100644 index 0000000..59c9f48 --- /dev/null +++ b/tests/test_ner_nel_split.py @@ -0,0 +1,80 @@ +""" +Local test: Verify NER and NEL modules work independently. +Run from the classifier root: python -m tests.test_ner_nel_split +""" + +import time + +SAMPLE_TEXT = ( + "Head Chef. We are looking for an experienced Head Chef who can plan menus, " + "manage kitchen staff, and ensure food safety standards. " + "A diploma in Culinary Arts is required." +) + + +def main(): + print("=" * 60) + print("TEST: NER and NEL modules (standalone)") + print("=" * 60) + + # --- Test NER --- + print("\n[1/3] Loading NER model...") + t0 = time.time() + from inference.ner import NERModel + ner = NERModel() + print(f" Loaded in {time.time() - t0:.1f}s") + + print("\n[2/3] Running NER on sample text...") + print(f' Input: "{SAMPLE_TEXT[:80]}..."') + t0 = time.time() + entities = ner.extract(SAMPLE_TEXT) + ner_time = time.time() - t0 + + print(f" Found {len(entities)} entities in {ner_time:.3f}s:\n") + for e in entities: + print(f" {e['entity_type']:15s} \"{e['surface_form']}\" " + f"(chars {e['span']['start']}-{e['span']['end']})") + + # --- Test NEL --- + print("\n[3/3] Loading NEL linker and linking entities...") + t0 = time.time() + from inference.nel import NELLinker + nel = NELLinker() + print(f" Loaded in {time.time() - t0:.1f}s") + + nel_input = [ + {"text": e["surface_form"], "entity_type": e["entity_type"]} + for e in entities + if e["entity_type"] in ("occupation", "skill", "qualification") + ] + + print(f" Linking {len(nel_input)} entities to ESCO...") + t0 = time.time() + linked = nel.link(nel_input, top_k=3, min_similarity=0.3) + nel_time = time.time() - t0 + + print(f" Linked in {nel_time:.3f}s:\n") + for item in linked: + print(f" \"{item['input_text']}\" ({item['entity_type']}):") + for m in item["matches"][:3]: + label = m.get("label", "?") + score = m.get("similarity_score", 0) + code = m.get("code", "") + suffix = f" code={code}" if code else "" + print(f" -> {label} (score={score}){suffix}") + print() + + + print("=" * 60) + print("RESULTS SUMMARY") + print("=" * 60) + print(f" NER: {len(entities)} entities extracted in {ner_time:.3f}s") + print(f" NEL: {len(linked)} entities linked in {nel_time:.3f}s") + print(f" NER module: inference/ner.py (reuses EntityLinker statics)") + print(f" NEL module: inference/nel.py (reuses EntityLinker.create_tensors)") + print(f" Original linker.py: untouched") + print("=" * 60) + + +if __name__ == "__main__": + main() diff --git a/tests/test_worker_simulate.py b/tests/test_worker_simulate.py new file mode 100644 index 0000000..fb3e5c3 --- /dev/null +++ b/tests/test_worker_simulate.py @@ -0,0 +1,185 @@ +""" +Simulate the Change Stream Worker with dummy data. + +Mimics what the real worker does: + 1. A "new job" appears (dummy data) + 2. Worker detects it + 3. Sends it to Classify API (http://localhost:5001) + 4. Stores the result in-memory + 5. Prints the full pipeline trace + +Requires: ner_server (5002), nel_server (5003), classify_server (5001) running. +No MongoDB needed. + +Usage: + python -m tests.test_worker_simulate +""" + +import sys +import os +import time + +import requests + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from util.job_text import build_input_text, compute_hash + +CLASSIFY_API_URL = "http://localhost:5001" + +DUMMY_JOBS = [ + { + "job_fingerprint": "sim_001_nurse", + "title": "Registered Nurse", + "description": ( + "We are hiring a Registered Nurse for our Nairobi clinic. " + "The ideal candidate should have a Diploma in Nursing and at least " + "2 years of experience in patient care. Skills required include " + "wound management, vital signs monitoring, and medication administration." + ), + "employer": "HealthPlus Kenya", + "location": "Nairobi, Kenya", + }, + { + "job_fingerprint": "sim_002_driver", + "title": "Heavy Truck Driver", + "description": ( + "Looking for an experienced heavy truck driver for long-haul routes " + "between Dar es Salaam and Mombasa. Must hold a valid Class C driving " + "license. Minimum 3 years experience in commercial driving. Knowledge " + "of vehicle maintenance and road safety regulations is essential." + ), + "employer": "TransEast Logistics", + "location": "Dar es Salaam, Tanzania", + }, + { + "job_fingerprint": "sim_003_accountant", + "title": "Junior Accountant", + "description": ( + "Entry-level accounting position available. Responsibilities include " + "bookkeeping, preparing financial statements, tax filing, and payroll " + "processing. Bachelor degree in Accounting or Finance required. " + "Proficiency in QuickBooks and Microsoft Excel is a must." + ), + "employer": "Finserve Ltd", + "location": "Kampala, Uganda", + }, +] + + +def main(): + print("=" * 70) + print("CHANGE STREAM WORKER — SIMULATION MODE (no MongoDB)") + print("=" * 70) + + try: + r = requests.get(f"{CLASSIFY_API_URL}/v1/health", timeout=5) + health = r.json() + deps = health.get("dependencies", {}) + print(f"\nClassify API health: {health['status']}") + print(f" NER API: {deps.get('ner_api', '?')}") + print(f" NEL API: {deps.get('nel_api', '?')}") + if health["status"] != "healthy": + print("\nERROR: Classify API is not fully healthy. Start all 3 servers first.") + return + except requests.RequestException as e: + print(f"\nERROR: Cannot reach Classify API at {CLASSIFY_API_URL}: {e}") + print("Make sure all 3 servers are running (ports 5001, 5002, 5003).") + return + + results_store = {} + + print(f"\nSimulating {len(DUMMY_JOBS)} incoming jobs...\n") + + for i, job in enumerate(DUMMY_JOBS, 1): + fp = job["job_fingerprint"] + input_text = build_input_text(job) + input_hash = compute_hash(input_text) + + print("-" * 70) + print(f"[{i}/{len(DUMMY_JOBS)}] CHANGE EVENT: new job detected") + print(f" Fingerprint : {fp}") + print(f" Title : {job['title']}") + print(f" Employer : {job['employer']}") + print(f" Location : {job['location']}") + print(f" Input hash : {input_hash[:16]}...") + + if fp in results_store and results_store[fp]["input_text_hash"] == input_hash: + print(" SKIP — already classified with same input hash") + continue + + print(" Calling Classify API...") + start = time.time() + try: + resp = requests.post( + f"{CLASSIFY_API_URL}/v1/classify", + json={"text": input_text}, + timeout=60, + ) + resp.raise_for_status() + result = resp.json() + except requests.RequestException as e: + print(f" ERROR: {e}") + continue + + elapsed = round((time.time() - start) * 1000) + classification = result.get("classification", {}) + metadata = result.get("metadata", {}) + entities = classification.get("entities", []) + counts = classification.get("entity_counts", {}) + + results_store[fp] = { + "job_fingerprint": fp, + "input_text_hash": input_hash, + "classification": classification, + "metadata": metadata, + "source_fields": { + "title": job["title"], + "employer": job["employer"], + "location": job["location"], + }, + } + + print(f" Classification complete in {elapsed}ms") + print(f" Entities found: {sum(counts.values())} total — {dict(counts)}") + print(f" Top entities:") + for ent in entities[:5]: + linked = ent.get("linked_entities", []) + top_match = ( + f" -> {linked[0]['label']} ({linked[0]['similarity_score']})" + if linked else "" + ) + print(f" [{ent['entity_type']}] \"{ent['surface_form']}\"{top_match}") + if len(entities) > 5: + print(f" ... and {len(entities) - 5} more") + + print(" SAVED to classified-jobs (in-memory)") + + # Dedup test + print("\n" + "=" * 70) + print("DEDUP TEST: Replaying first job (should be skipped)") + print("=" * 70) + first = DUMMY_JOBS[0] + fp = first["job_fingerprint"] + input_text = build_input_text(first) + input_hash = compute_hash(input_text) + if fp in results_store and results_store[fp]["input_text_hash"] == input_hash: + print(f" SKIP — {fp} already classified with same input hash (ok)") + else: + print(" ERROR — dedup failed!") + + # Summary + print("\n" + "=" * 70) + print("PIPELINE SUMMARY") + print("=" * 70) + print(f" Jobs processed : {len(results_store)}") + print(f" Jobs skipped : 1 (dedup test)") + print(f" Storage backend : in-memory (no MongoDB)") + first_result = list(results_store.values())[0] + print(f" NER model : {first_result['metadata'].get('model_name', '?')}") + print(f" NEL model : {first_result['metadata'].get('linker_model', '?')}") + print(f"\nFull pipeline: Scraper -> raw-jobs -> [Worker] -> Classify API -> classified-jobs") + + +if __name__ == "__main__": + main() diff --git a/util/job_text.py b/util/job_text.py new file mode 100644 index 0000000..974e731 --- /dev/null +++ b/util/job_text.py @@ -0,0 +1,31 @@ +import hashlib +from typing import Optional + + +def build_input_text( + doc: dict, + *, + allow_text_field: bool = False, +) -> Optional[str]: + """Build classifier input from a job dict. + + When *allow_text_field* is True, a non-empty ``text`` key is used as-is + (the classify API accepts this). Workers that only have title+description + should leave it False. Returns None when no usable text can be assembled. + """ + if allow_text_field: + text = doc.get("text") + if text and isinstance(text, str) and text.strip(): + return text.strip() + + title = (doc.get("title") or "").strip() + description = (doc.get("description") or "").strip() + + if title and description: + return f"{title}.\n{description}" + return description or title or None + + +def compute_hash(text: str) -> str: + """SHA-256 hex digest of *text*, used for classification dedup.""" + return hashlib.sha256(text.encode("utf-8")).hexdigest()