From 52f6c88ed0b669f208f43d146967f12800252033 Mon Sep 17 00:00:00 2001 From: ArpitKumar8649 Date: Sun, 10 May 2026 16:23:44 +0000 Subject: [PATCH 1/9] docs: add README for email-generator-service --- email-generator-service/README.md | 214 +++++++++--------------------- 1 file changed, 61 insertions(+), 153 deletions(-) diff --git a/email-generator-service/README.md b/email-generator-service/README.md index 05225a3..1811cdb 100644 --- a/email-generator-service/README.md +++ b/email-generator-service/README.md @@ -1,66 +1,31 @@ -# Cold Email Generator Service +# Email Generator Service -A FastAPI microservice that drafts personalized cold emails for job applications using **Jinja2 templates** + an optional **local Ollama LLM** (mistral/llama3) for a one-line company product observation. Falls back to curated static observations from `fallbacks.yaml` when Ollama is unavailable. +## Purpose of the service ---- +This service drafts personalized cold emails for job outreach, stores those drafts in PostgreSQL, and can send them through Gmail SMTP. It combines Jinja2 templates with an optional local Ollama LLM observation and falls back to `fallbacks.yaml` when Ollama is unavailable. -## Project layout +## Request/Data Flow -``` -email-generator-service/ -├── main.py # FastAPI app — all endpoints -├── generator.py # Email generation pipeline (Ollama → fallback → Jinja2) -├── ollama_client.py # Async Ollama REST client -├── mailer.py # Gmail SMTP_SSL sender -├── storage.py # asyncpg pool, emails table DDL, CRUD -├── fallbacks.yaml # Static observations by domain (fintech, devtools…) -├── templates/ -│ ├── cold_outreach.j2 # To hiring manager / engineer (≤150 words) -│ ├── recruiter_outreach.j2 # To recruiter (≤120 words) -│ └── followup.j2 # Follow-up after no reply (≤80 words) -├── requirements.txt -├── Dockerfile -└── README.md -``` - ---- +1. `POST /generate` receives a request with `template`, optional `job_id` / `contact_id`, candidate details, and context. +2. `main.py` opens the database pool and resolves job/contact records from PostgreSQL via `storage.py`. +3. `generator.py` prepares the email context, attempts to generate a product observation through `ollama_client.py`, and renders one of the Jinja2 templates. +4. The rendered subject/body are persisted to the `emails` table with `storage.insert_email()`. +5. Clients can query drafts with `GET /emails` and `GET /emails/{id}`. +6. `POST /emails/{id}/send` looks up the linked contact email, sends the message with `mailer.py`, and updates `sent_at` / status. -## Configuration (environment variables) +## Important Files/Modules -| Variable | Default | Description | -|---|---|---| -| `DATABASE_URL` | — | asyncpg DSN (shared with aggregator / contact discovery) | -| `OLLAMA_BASE_URL` | `http://localhost:11434` | Ollama REST API base URL | -| `GMAIL_ADDRESS` | — | Your Gmail address for sending | -| `GMAIL_APP_PASSWORD` | — | 16-char Gmail App Password (not login password) | -| `YOUR_NAME` | `Applicant` | Shown in email sign-off | -| `YOUR_GITHUB_URL` | — | GitHub profile URL embedded in emails | +- `main.py` — FastAPI app, endpoints, lifecycle hooks, request/response models. +- `generator.py` — email generation pipeline, Ollama fallback logic, Jinja2 rendering. +- `ollama_client.py` — async Ollama REST client and remote model detection. +- `mailer.py` — Gmail SMTP sender using `GMAIL_ADDRESS` and `GMAIL_APP_PASSWORD`. +- `storage.py` — asyncpg pool management, schema initialization, CRUD for `jobs`, `contacts`, and `emails`. +- `fallbacks.yaml` — domain-based static product observations. +- `templates/` — Jinja2 templates for `cold_outreach`, `recruiter_outreach`, and `followup`. ---- +## Local Execution -## Ollama setup (optional but recommended) - -Install Ollama, then pull the preferred model: - -```bash -# Install (Linux) -curl -fsSL https://ollama.com/install.sh | sh - -# Pull models (mistral is preferred; llama3 is the fallback) -ollama pull mistral -ollama pull llama3 # optional - -# Start the server (runs on http://localhost:11434 by default) -ollama serve -``` - -The service auto-detects which model is available. If Ollama is not running, it falls back to `fallbacks.yaml` silently — **no configuration change needed**. - ---- - -## Quick start - -### Local dev +### Run locally with Python ```bash cd email-generator-service @@ -76,7 +41,7 @@ export YOUR_GITHUB_URL="https://github.com/yourhandle" uvicorn main:app --reload --port 8003 ``` -### Docker +### Run with Docker ```bash docker build -t email-generator . @@ -91,133 +56,76 @@ docker run \ email-generator ``` ---- +### Verify service start -## Gmail App Password setup +Open for FastAPI interactive docs. -1. Go to -2. Select **Mail** + **Linux** (or any device) -3. Copy the generated 16-character password → `GMAIL_APP_PASSWORD` +## Environment Variables -> You must have **2-Step Verification** enabled on your Google account. +- `DATABASE_URL` — PostgreSQL DSN used by asyncpg. +- `OLLAMA_BASE_URL` — Ollama API base URL; defaults to `http://localhost:11434`. +- `GMAIL_ADDRESS` — Gmail address used as the sender. +- `GMAIL_APP_PASSWORD` — Gmail App Password used for SMTP login. +- `YOUR_NAME` — sender name used in rendered emails and SMTP `From`. +- `YOUR_GITHUB_URL` — GitHub profile URL included in emails. +- `GRADUATION_YEAR` — optional fallback graduation year used by templates when not provided in payload. ---- +## Service Interactions -## API reference +- Uses PostgreSQL to read `jobs` and `contacts` records and to store `emails` drafts. +- Reads `jobs` / `contacts` tables from the same database, so it can work with `aggregator-service` and `contact-discovery-service` if they share the same PostgreSQL instance. +- Calls a local Ollama server to generate a single product observation sentence, with static fallback observations from `fallbacks.yaml` if Ollama is unavailable. +- Sends outbound mail through Gmail SMTP on port `465`. -Interactive docs: +## Debugging/Setup Notes -### `GET /health` +- `DATABASE_URL` is required before the service can start; `storage.py` creates `jobs`, `contacts`, and `emails` tables automatically. +- `OLLAMA_BASE_URL` defaults to `http://localhost:11434`; missing Ollama does not break generation because `fallbacks.yaml` is used. +- Gmail sending requires a valid `GMAIL_APP_PASSWORD` and must use an App Password, not the normal Gmail login password. +- The service uses `pgcrypto` for `gen_random_uuid()`, so PostgreSQL must allow extension creation. +- `mailer.py` performs SMTP over SSL; connection failures usually indicate network/blocking or invalid credentials. +- If `YOUR_NAME` or `YOUR_GITHUB_URL` are not supplied in the request, the service falls back to environment values. -```bash -curl http://localhost:8003/health -``` +## Example Requests/Workflows ---- - -### `POST /generate` — generate and store an email - -```bash -# Cold outreach to a hiring manager -curl -X POST http://localhost:8003/generate \ - -H "Content-Type: application/json" \ - -d '{ - "job_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "contact_id": "a1b2c3d4-0000-0000-0000-000000000001", - "template": "cold_outreach", - "your_name": "Vaibhav Sharma", - "your_stack": ["Python", "FastAPI", "PostgreSQL"], - "github_url": "https://github.com/sharmavaibhav31", - "graduation_year": 2025 - }' -``` -```json -{ - "email_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479", - "subject": "Backend Engineer opportunity — Vaibhav Sharma", - "body": "Hi Alice,\n\nI came across Razorpay's Backend Engineer opening..." -} -``` +### Generate a cold outreach email ```bash -# Recruiter outreach curl -X POST http://localhost:8003/generate \ - -H "Content-Type: application/json" \ - -d '{"job_id": "...", "contact_id": "...", "template": "recruiter_outreach", - "your_name": "Vaibhav Sharma", "your_stack": ["Go", "Kubernetes"], - "github_url": "https://github.com/sharmavaibhav31", "graduation_year": 2025}' - -# Follow-up (7 days later) -curl -X POST http://localhost:8003/generate \ - -H "Content-Type: application/json" \ - -d '{"job_id": "...", "contact_id": "...", "template": "followup", - "your_name": "Vaibhav Sharma", "your_stack": [], - "github_url": "https://github.com/sharmavaibhav31", - "availability": "Monday and Wednesday between 2–5 PM IST"}' + -H "Content-Type: application/json" \ + -d '{ + "job_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6", + "contact_id": "a1b2c3d4-0000-0000-0000-000000000001", + "template": "cold_outreach", + "your_name": "Vaibhav Sharma", + "your_stack": ["Python", "FastAPI", "PostgreSQL"], + "github_url": "https://github.com/sharmavaibhav31", + "graduation_year": 2025 + }' ``` ---- - -### `GET /emails?job_id={uuid}` — list emails for a job +### Fetch emails for a job ```bash curl "http://localhost:8003/emails?job_id=3fa85f64-5717-4562-b3fc-2c963f66afa6" ``` ---- - -### `GET /emails/{id}` — fetch a single email +### Fetch a single email draft ```bash curl http://localhost:8003/emails/f47ac10b-58cc-4372-a567-0e02b2c3d479 ``` ---- - -### `PATCH /emails/{id}/status` — update status - -Valid values: `draft`, `sent`, `replied`. +### Update status ```bash curl -X PATCH http://localhost:8003/emails/f47ac10b-.../status \ - -H "Content-Type: application/json" \ - -d '{"status": "replied"}' + -H "Content-Type: application/json" \ + -d '{"status": "replied"}' ``` ---- - -### `POST /emails/{id}/send` — send via Gmail - -Fetches the recipient address from the linked contact record. Marks `sent_at` and `status = sent` on success. +### Send a generated email ```bash curl -X POST http://localhost:8003/emails/f47ac10b-.../send ``` - ---- - -## Template customization - -All templates live in `templates/`. Edit them freely — they use standard [Jinja2](https://jinja.palletsprojects.com/) syntax. - -Available context variables: - -| Variable | Source | -|---|---| -| `company` | jobs table | -| `role` | jobs table | -| `your_name` | request body / `YOUR_NAME` env | -| `your_stack` | request body | -| `github_url` | request body / `YOUR_GITHUB_URL` env | -| `product_observation` | Ollama or fallbacks.yaml | -| `contact_name` | contacts table | -| `graduation_year` | request body | -| `availability` | request body (followup only) | - ---- - -## How the fallback selection works - -`fallbacks.yaml` contains observations in 5 categories: `fintech`, `devtools`, `saas`, `ecommerce`, `infra`. - -The generator counts keyword matches between the job's `product` + `stack` fields and each category's `keywords` list. The category with the highest score wins, and a random observation from that category's list is chosen. A `default` category is used when no keywords match. From 197c2053f7d74cbb6c15dbcd782864a86f1a7366 Mon Sep 17 00:00:00 2001 From: ArpitKumar8649 Date: Mon, 11 May 2026 17:42:20 +0000 Subject: [PATCH 2/9] docs: update README with internal architecture focus --- email-generator-service/README.md | 150 +++++++----------------------- 1 file changed, 32 insertions(+), 118 deletions(-) diff --git a/email-generator-service/README.md b/email-generator-service/README.md index 1811cdb..9999256 100644 --- a/email-generator-service/README.md +++ b/email-generator-service/README.md @@ -1,131 +1,45 @@ # Email Generator Service -## Purpose of the service - -This service drafts personalized cold emails for job outreach, stores those drafts in PostgreSQL, and can send them through Gmail SMTP. It combines Jinja2 templates with an optional local Ollama LLM observation and falls back to `fallbacks.yaml` when Ollama is unavailable. - ## Request/Data Flow -1. `POST /generate` receives a request with `template`, optional `job_id` / `contact_id`, candidate details, and context. -2. `main.py` opens the database pool and resolves job/contact records from PostgreSQL via `storage.py`. -3. `generator.py` prepares the email context, attempts to generate a product observation through `ollama_client.py`, and renders one of the Jinja2 templates. -4. The rendered subject/body are persisted to the `emails` table with `storage.insert_email()`. -5. Clients can query drafts with `GET /emails` and `GET /emails/{id}`. -6. `POST /emails/{id}/send` looks up the linked contact email, sends the message with `mailer.py`, and updates `sent_at` / status. - -## Important Files/Modules - -- `main.py` — FastAPI app, endpoints, lifecycle hooks, request/response models. -- `generator.py` — email generation pipeline, Ollama fallback logic, Jinja2 rendering. -- `ollama_client.py` — async Ollama REST client and remote model detection. -- `mailer.py` — Gmail SMTP sender using `GMAIL_ADDRESS` and `GMAIL_APP_PASSWORD`. -- `storage.py` — asyncpg pool management, schema initialization, CRUD for `jobs`, `contacts`, and `emails`. -- `fallbacks.yaml` — domain-based static product observations. -- `templates/` — Jinja2 templates for `cold_outreach`, `recruiter_outreach`, and `followup`. - -## Local Execution - -### Run locally with Python - -```bash -cd email-generator-service -python -m venv .venv && source .venv/bin/activate -pip install -r requirements.txt - -export DATABASE_URL="postgresql://jobuser:jobpass@localhost:5432/jobsdb" -export GMAIL_ADDRESS="you@gmail.com" -export GMAIL_APP_PASSWORD="abcd efgh ijkl mnop" -export YOUR_NAME="Your Name" -export YOUR_GITHUB_URL="https://github.com/yourhandle" - -uvicorn main:app --reload --port 8003 -``` - -### Run with Docker - -```bash -docker build -t email-generator . -docker run \ - -e DATABASE_URL="postgresql://jobuser:jobpass@host.docker.internal:5432/jobsdb" \ - -e OLLAMA_BASE_URL="http://host.docker.internal:11434" \ - -e GMAIL_ADDRESS="you@gmail.com" \ - -e GMAIL_APP_PASSWORD="abcd efgh ijkl mnop" \ - -e YOUR_NAME="Your Name" \ - -e YOUR_GITHUB_URL="https://github.com/yourhandle" \ - -p 8003:8000 \ - email-generator -``` +1. `POST /generate` accepts a JSON payload with `template`, optional `job_id`/`contact_id`, candidate details, and context fields. +2. `main.py` initializes the asyncpg pool and fetches job/contact records from PostgreSQL using `storage.py` if IDs are provided. +3. `generator.py` constructs the email context, invokes `ollama_client.py` for product observation generation, and renders Jinja2 templates. +4. Rendered subject/body are inserted into the `emails` table via `storage.insert_email()`, returning the email UUID. +5. `GET /emails` and `GET /emails/{id}` retrieve drafts from PostgreSQL using `storage.py` queries. +6. `POST /emails/{id}/send` resolves the recipient email from the linked contact record, sends via `mailer.py`, and updates `sent_at`/`status` in the database. -### Verify service start +## Internal Execution Pipeline -Open for FastAPI interactive docs. +- **Generation Pipeline**: `generator.generate_email()` loads job/contact data, attempts Ollama observation via `ollama_client.generate_observation()`, falls back to YAML-based selection from `fallbacks.yaml`, and renders templates using Jinja2 with context variables. +- **Ollama Integration**: `ollama_client.py` detects available models (mistral/llama3), sends a prompt to `/api/generate`, and returns a single observation sentence or None on failure. +- **Fallback Logic**: `generator._select_fallback()` matches keywords from product/stack against `fallbacks.yaml` categories, selecting a random observation from the best-matched bucket. +- **Template Rendering**: Jinja2 environment in `generator.py` processes templates with variables like `company`, `role`, `your_name`, `product_observation`, and `contact_name`. +- **Storage Operations**: `storage.py` manages asyncpg pool, executes DDL for `jobs`/`contacts`/`emails` tables, and performs CRUD with parameterized queries. +- **Email Sending**: `mailer.send_email()` uses `smtplib.SMTP_SSL` with Gmail credentials, sending plain-text emails and handling authentication errors. -## Environment Variables +## Important Modules/Files -- `DATABASE_URL` — PostgreSQL DSN used by asyncpg. -- `OLLAMA_BASE_URL` — Ollama API base URL; defaults to `http://localhost:11434`. -- `GMAIL_ADDRESS` — Gmail address used as the sender. -- `GMAIL_APP_PASSWORD` — Gmail App Password used for SMTP login. -- `YOUR_NAME` — sender name used in rendered emails and SMTP `From`. -- `YOUR_GITHUB_URL` — GitHub profile URL included in emails. -- `GRADUATION_YEAR` — optional fallback graduation year used by templates when not provided in payload. +- `main.py`: FastAPI application with endpoints (`/generate`, `/emails`, `/emails/{id}`, `/emails/{id}/send`), Pydantic models for requests/responses, and lifespan hooks for pool management. +- `generator.py`: Core logic for email creation, including Ollama client calls, fallback selection, and Jinja2 rendering with context assembly. +- `ollama_client.py`: Async HTTP client for Ollama API, model detection via `/api/tags`, and observation generation with timeout handling. +- `mailer.py`: Synchronous Gmail SMTP sender wrapped in `asyncio.run_in_executor`, using SSL on port 465 with App Password authentication. +- `storage.py`: asyncpg pool lifecycle, schema initialization (including `pgcrypto` extension), and CRUD functions for jobs, contacts, and emails tables. +- `fallbacks.yaml`: YAML structure with keyword lists per category (e.g., fintech, devtools) and observation arrays for static fallbacks. +- `templates/`: Directory containing Jinja2 templates (`cold_outreach.j2`, `recruiter_outreach.j2`, `followup.j2`) with conditional rendering and variable substitution. ## Service Interactions -- Uses PostgreSQL to read `jobs` and `contacts` records and to store `emails` drafts. -- Reads `jobs` / `contacts` tables from the same database, so it can work with `aggregator-service` and `contact-discovery-service` if they share the same PostgreSQL instance. -- Calls a local Ollama server to generate a single product observation sentence, with static fallback observations from `fallbacks.yaml` if Ollama is unavailable. -- Sends outbound mail through Gmail SMTP on port `465`. - -## Debugging/Setup Notes - -- `DATABASE_URL` is required before the service can start; `storage.py` creates `jobs`, `contacts`, and `emails` tables automatically. -- `OLLAMA_BASE_URL` defaults to `http://localhost:11434`; missing Ollama does not break generation because `fallbacks.yaml` is used. -- Gmail sending requires a valid `GMAIL_APP_PASSWORD` and must use an App Password, not the normal Gmail login password. -- The service uses `pgcrypto` for `gen_random_uuid()`, so PostgreSQL must allow extension creation. -- `mailer.py` performs SMTP over SSL; connection failures usually indicate network/blocking or invalid credentials. -- If `YOUR_NAME` or `YOUR_GITHUB_URL` are not supplied in the request, the service falls back to environment values. - -## Example Requests/Workflows - -### Generate a cold outreach email - -```bash -curl -X POST http://localhost:8003/generate \ - -H "Content-Type: application/json" \ - -d '{ - "job_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "contact_id": "a1b2c3d4-0000-0000-0000-000000000001", - "template": "cold_outreach", - "your_name": "Vaibhav Sharma", - "your_stack": ["Python", "FastAPI", "PostgreSQL"], - "github_url": "https://github.com/sharmavaibhav31", - "graduation_year": 2025 - }' -``` - -### Fetch emails for a job - -```bash -curl "http://localhost:8003/emails?job_id=3fa85f64-5717-4562-b3fc-2c963f66afa6" -``` - -### Fetch a single email draft - -```bash -curl http://localhost:8003/emails/f47ac10b-58cc-4372-a567-0e02b2c3d479 -``` - -### Update status - -```bash -curl -X PATCH http://localhost:8003/emails/f47ac10b-.../status \ - -H "Content-Type: application/json" \ - -d '{"status": "replied"}' -``` +- Reads `jobs` and `contacts` tables from PostgreSQL to populate email templates and resolve recipient addresses. +- Writes to the `emails` table for draft storage and status updates, sharing the database with `aggregator-service` and `contact-discovery-service`. +- Makes HTTP requests to a local Ollama server at `OLLAMA_BASE_URL` for product observations, with automatic fallback to static data. +- Sends outbound emails via Gmail SMTP, requiring external Gmail account configuration for delivery. -### Send a generated email +## Debugging Notes -```bash -curl -X POST http://localhost:8003/emails/f47ac10b-.../send -``` +- Ollama failures log warnings in `ollama_client.py` for timeouts or HTTP errors, but do not halt generation due to fallback mechanism. +- Database connection issues in `storage.py` raise `RuntimeError` if pool is not initialized, logged during lifespan startup. +- SMTP authentication errors in `mailer.py` propagate as `smtplib.SMTPAuthenticationError`, indicating invalid `GMAIL_APP_PASSWORD`. +- Template rendering errors in `generator.py` raise `ValueError` for unknown templates, logged with template names. +- Asyncpg query timeouts default to 30 seconds in `storage.py`, potentially causing delays in high-load scenarios. +- Fallback selection in `generator.py` logs category scores and selected observations for keyword matching verification. From c74ca96670853092b043f7a1b18d3a8c52c26218 Mon Sep 17 00:00:00 2001 From: ArpitKumar8649 Date: Mon, 11 May 2026 17:43:35 +0000 Subject: [PATCH 3/9] docs: add README for aggregator-service with internal architecture --- aggregator-service/README.md | 256 ++++------------------------------- 1 file changed, 28 insertions(+), 228 deletions(-) diff --git a/aggregator-service/README.md b/aggregator-service/README.md index d53dae0..e85b192 100644 --- a/aggregator-service/README.md +++ b/aggregator-service/README.md @@ -1,238 +1,38 @@ -# Job Aggregator Service +# Aggregator Service -A FastAPI microservice that consumes normalized job events from the `jobs:raw` Redis Stream (produced by the Scrapy crawler service), deduplicates them, persists to PostgreSQL, and exposes a REST API for querying and managing jobs. +## Request/Data Flow -``` -┌─────────────────┐ Redis Stream ┌───────────────────┐ asyncpg ┌──────────────┐ -│ Crawler Service│ ──► jobs:raw ──────► │ Aggregator Service│ ────────────► │ PostgreSQL │ -│ (Scrapy) │ │ (FastAPI + asyncio)│ │ jobs table │ -└─────────────────┘ └───────────────────┘ └──────────────┘ -``` +1. `GET /jobs` accepts query parameters for role, stack, status, sort, and limit, querying PostgreSQL with filters and returning job records. +2. `GET /jobs/{id}` fetches a single job by UUID from PostgreSQL. +3. `PATCH /jobs/{id}/status` updates the job status in PostgreSQL and returns the updated record. +4. `GET /stats` aggregates counts by source and status from PostgreSQL. +5. Background consumer reads messages from Redis 'jobs:raw' stream, parses fields, checks deduplication key, and conditionally inserts into PostgreSQL. -## Tech stack +## Internal Execution Pipeline -| Layer | Library | -|-------|---------| -| API framework | FastAPI + Uvicorn | -| DB driver | asyncpg (raw SQL, no ORM) | -| Redis client | redis-py (asyncio) | -| Schemas | Pydantic v2 | -| Runtime | Python 3.11 | +- **Consumer Loop**: `consumer.run_consumer()` connects to Redis, ensures consumer group exists, claims pending messages, then reads in batches using `XPENDING` and `XREADGROUP`, processing each message asynchronously. +- **Message Processing**: `_process_message()` decodes stream fields, normalizes company/role for deduplication, checks Redis key existence, parses stack as JSON array and posted_at as datetime, then calls `db.insert_job()` with conflict resolution. +- **Deduplication**: `_dedup_key()` generates MD5 hash from normalized company+role, sets Redis key with 7-day TTL on successful insert to prevent reprocessing. +- **Database Operations**: `db.insert_job()` uses `INSERT ... ON CONFLICT DO NOTHING` for URL uniqueness, `get_jobs()` applies ILIKE for role substring and GIN containment for stack arrays, with sorting by posted_at. +- **Stats Aggregation**: `db.get_stats()` executes GROUP BY queries on source and status columns, returning dictionaries of counts. ---- +## Important Modules/Files -## Project layout +- `main.py`: FastAPI application with endpoints (`/jobs`, `/jobs/{id}`, `/jobs/{id}/status`, `/stats`), lifespan hooks for pool and consumer task management, and Pydantic response models. +- `consumer.py`: Async Redis stream consumer with consumer group management, message parsing, deduplication logic, and batch processing using `XAUTOCLAIM` for pending messages. +- `db.py`: asyncpg pool lifecycle, schema DDL for jobs table with GIN/BTREE indexes, and CRUD functions with parameterized queries and conflict handling. +- `models.py`: Pydantic models for `JobOut`, `StatusUpdate`, and `StatsOut` with validation for status values and attribute-based configuration. -``` -aggregator-service/ -├── main.py # FastAPI app, lifespan, endpoints -├── db.py # asyncpg pool, schema init, query helpers -├── consumer.py # Redis Stream consumer loop -├── models.py # Pydantic response schemas -├── requirements.txt -├── Dockerfile -├── docker-compose.yml -└── README.md -``` +## Service Interactions ---- +- Consumes messages from Redis 'jobs:raw' stream using consumer groups for reliable delivery. +- Writes job records to PostgreSQL jobs table, shared with other services like contact-discovery-service and email-generator-service. +- Provides REST API for job queries and status updates, consumed by gateway or other components. -## Configuration (environment variables) +## Debugging Notes -| Variable | Default | Description | -|---|---|---| -| `REDIS_HOST` | `localhost` | Redis hostname | -| `REDIS_PORT` | `6379` | Redis port | -| `DATABASE_URL` | — | asyncpg DSN, e.g. `postgresql://user:pass@host:5432/db` | -| `PORT` | `8000` | Port the API listens on | -| `POSTGRES_USER` | `jobuser` | (docker-compose only) | -| `POSTGRES_PASSWORD` | `jobpass` | (docker-compose only) | -| `POSTGRES_DB` | `jobsdb` | (docker-compose only) | - ---- - -## Quick start - -### With Docker Compose (recommended) - -> Run from the `aggregator-service/` directory. - -```bash -# 1. Copy and tweak env (optional — defaults work out of the box) -cp .env.example .env # edit JOBSEEKER_ROLE, JOBSEEKER_STACK, etc. - -# 2. Start everything (Redis + Postgres + Crawler + Aggregator) -docker compose up --build - -# 3. Once the aggregator is healthy, fire the crawler once -docker compose run --rm crawler scrapy crawl remotive -``` - -### Local development (no Docker) - -```bash -cd aggregator-service -python -m venv .venv && source .venv/bin/activate -pip install -r requirements.txt - -export REDIS_HOST=localhost -export REDIS_PORT=6379 -export DATABASE_URL="postgresql://jobuser:jobpass@localhost:5432/jobsdb" - -uvicorn main:app --reload --port 8000 -``` - ---- - -## API Reference - -Interactive docs: - -### `GET /health` — liveness probe - -```bash -curl http://localhost:8000/health -``` -```json -{"status": "ok"} -``` - ---- - -### `GET /jobs` — list jobs - -| Query param | Type | Default | Description | -|---|---|---|---| -| `role` | string | — | Substring match on role title (case-insensitive) | -| `stack` | string | — | Comma-separated tags; returns jobs whose stack contains **all** of them | -| `status` | string | — | `new` \| `applied` \| `ignored` | -| `sort` | string | `latest` | `latest` or `oldest` (by `posted_at`) | -| `limit` | int | `50` | Max results (1–500) | - -```bash -# All new jobs -curl "http://localhost:8000/jobs?status=new" - -# Backend roles using Python, sorted oldest first -curl "http://localhost:8000/jobs?role=backend&stack=Python&sort=oldest&limit=20" - -# Jobs that require both FastAPI and PostgreSQL -curl "http://localhost:8000/jobs?stack=FastAPI,PostgreSQL" -``` - -
-Example response - -```json -[ - { - "id": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "company": "Acme Corp", - "role": "Senior Backend Engineer", - "source": "remotive", - "url": "https://remotive.com/jobs/123", - "stack": ["Python", "FastAPI", "PostgreSQL"], - "product": "Platform", - "location": "Remote", - "posted_at": "2026-03-18T10:00:00Z", - "status": "new", - "created_at": "2026-03-19T04:00:00Z" - } -] -``` -
- ---- - -### `GET /jobs/{id}` — get single job - -```bash -curl http://localhost:8000/jobs/3fa85f64-5717-4562-b3fc-2c963f66afa6 -``` - -Returns `404` if the job is not found. - ---- - -### `PATCH /jobs/{id}/status` — update status - -Valid statuses: `new`, `applied`, `ignored`. - -```bash -# Mark a job as applied -curl -X PATCH http://localhost:8000/jobs/3fa85f64-5717-4562-b3fc-2c963f66afa6/status \ - -H "Content-Type: application/json" \ - -d '{"status": "applied"}' - -# Ignore a job -curl -X PATCH http://localhost:8000/jobs/3fa85f64-5717-4562-b3fc-2c963f66afa6/status \ - -H "Content-Type: application/json" \ - -d '{"status": "ignored"}' -``` - -Returns the updated job record, or `404` if not found. - ---- - -### `GET /stats` — aggregate counts - -```bash -curl http://localhost:8000/stats -``` -```json -{ - "by_source": { - "remotive": 142, - "greenhouse": 37 - }, - "by_status": { - "new": 155, - "applied": 18, - "ignored": 6 - } -} -``` - ---- - -## How deduplication works - -Each incoming event is hashed using **MD5(normalised_company + "|" + normalised_role)**. - -- Before inserting, the service checks whether the Redis key `dedup:agg:{hash}` exists. -- If it exists → event is ACK-ed and skipped. -- If not → job is inserted into Postgres and the key is set with a **7-day TTL**. - -This prevents the same role from being re-inserted if the crawler runs multiple times per day. - ---- - -## Consumer group & crash recovery - -The service subscribes as consumer group `aggregator-group` on the `jobs:raw` stream. -On startup it runs `XAUTOCLAIM` to reclaim any messages that were delivered but never ACK-ed (e.g. after a crash), ensuring **at-least-once delivery**. - ---- - -## Database schema - -```sql -CREATE TABLE jobs ( - id UUID PRIMARY KEY DEFAULT gen_random_uuid(), - company TEXT NOT NULL, - role TEXT NOT NULL, - source TEXT, - url TEXT, - stack TEXT[], - product TEXT, - location TEXT, - posted_at TIMESTAMPTZ, - status TEXT NOT NULL DEFAULT 'new', - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() -); - --- Indexes -CREATE INDEX idx_jobs_stack ON jobs USING GIN (stack); -CREATE INDEX idx_jobs_posted_at ON jobs (posted_at DESC NULLS LAST); -CREATE INDEX idx_jobs_status ON jobs (status); -``` +- Consumer logs warnings for missing company/role in messages, debug for duplicates skipped, and info for successful inserts with job IDs. +- Database connection errors raise `RuntimeError` if pool is uninitialized, logged during lifespan startup. +- Redis connection issues in consumer loop may cause retries, with group creation errors logged as info if already exists. +- Query timeouts in `db.py` default to 30 seconds, potentially delaying API responses under high load. +- Deduplication key collisions logged as debug, with MD5 hash generation ensuring consistent keying. From 75c9be6c09663c19d5a03438b1f8378f4aaea6c2 Mon Sep 17 00:00:00 2001 From: ArpitKumar8649 Date: Mon, 11 May 2026 17:44:19 +0000 Subject: [PATCH 4/9] docs: add README for contact-discovery-service with internal architecture --- contact-discovery-service/README.md | 272 ++++------------------------ 1 file changed, 40 insertions(+), 232 deletions(-) diff --git a/contact-discovery-service/README.md b/contact-discovery-service/README.md index 4519263..30e3ff1 100644 --- a/contact-discovery-service/README.md +++ b/contact-discovery-service/README.md @@ -1,234 +1,42 @@ # Contact Discovery Service -> ⚖️ **Ethical use policy** — This service queries only publicly available data sources (Clearbit Autocomplete, GitHub public APIs, LinkedIn public directory, and SMTP handshakes). It is designed exclusively for **professional outreach in an engineering placement context** — i.e., to reach out to recruiters or engineering managers at companies where a job has been found. It must not be used for spam, bulk unsolicited email, harvesting contact data for resale, or any purpose that violates the terms of service of the queried platforms. SMTP verification uses a non-sending probe (RCPT TO without DATA) and is capped at **5 attempts per domain per hour** to avoid IP blacklisting. - ---- - -## Architecture - -``` -POST /discover - │ - ▼ -[1] Domain inference company name → domain (Clearbit + direct probe) - │ - ▼ -[2] Email pattern detection domain → pattern (GitHub commit emails) - │ - ▼ (concurrent) -[3] Name discovery GitHub org members + LinkedIn public dir - │ - ▼ -[4] Email construction name × pattern → email string - │ - ▼ -[5] SMTP verification email → verified | unverified | invalid - │ - ▼ - PostgreSQL contacts table -``` - ---- - -## Project layout - -``` -contact-discovery-service/ -├── main.py # FastAPI app — endpoints -├── discovery.py # Full pipeline (stages 1–5) -├── verifier.py # SMTP email verifier + rate limiter -├── storage.py # asyncpg pool, schema, CRUD -├── requirements.txt -├── Dockerfile -└── README.md -``` - ---- - -## Configuration (environment variables) - -| Variable | Default | Description | -|---|---|---| -| `DATABASE_URL` | — | asyncpg DSN (shared with aggregator service) | -| `GITHUB_TOKEN` | — | Optional PAT — raises GitHub rate limit 60→5000 req/hr | -| `REDIS_HOST` | `localhost` | Not used directly, available for future use | -| `REDIS_PORT` | `6379` | — | - ---- - -## Quick start - -### Docker - -```bash -cd contact-discovery-service -docker build -t contact-discovery . -docker run \ - -e DATABASE_URL="postgresql://jobuser:jobpass@host.docker.internal:5432/jobsdb" \ - -e GITHUB_TOKEN="ghp_..." \ - -p 8002:8000 \ - contact-discovery -``` - -### Local dev - -```bash -cd contact-discovery-service -python -m venv .venv && source .venv/bin/activate -pip install -r requirements.txt -playwright install chromium - -export DATABASE_URL="postgresql://jobuser:jobpass@localhost:5432/jobsdb" -export GITHUB_TOKEN="ghp_..." # optional - -uvicorn main:app --reload --port 8002 -``` - ---- - -## API reference - -Interactive docs: - -### `GET /health` - -```bash -curl http://localhost:8002/health -``` -```json -{"status": "ok"} -``` - ---- - -### `POST /discover` — trigger contact discovery - -Returns immediately; all pipeline stages run in the background. - -```bash -# Basic usage — discover contacts at Razorpay linked to a specific job -curl -X POST http://localhost:8002/discover \ - -H "Content-Type: application/json" \ - -d '{ - "company": "Razorpay", - "job_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "roles": ["Engineering Manager", "Recruiter"] - }' -``` -```json -{ - "triggered": true, - "company": "Razorpay", - "roles": ["Engineering Manager", "Recruiter"], - "message": "Discovery running in background. Poll GET /contacts?company=... for results." -} -``` - -If you already know the domain (faster): -```bash -curl -X POST http://localhost:8002/discover \ - -H "Content-Type: application/json" \ - -d '{"company": "Razorpay", "domain": "razorpay.com", "roles": ["Recruiter"]}' -``` - ---- - -### `GET /contacts` — list by company - -```bash -# Poll after triggering discovery -curl "http://localhost:8002/contacts?company=Razorpay" -``` -```json -[ - { - "id": "...", - "job_id": "3fa85f64-...", - "company": "Razorpay", - "domain": "razorpay.com", - "name": "Alice Smith", - "email": "alice.smith@razorpay.com", - "role": "Engineering Manager", - "source": "github", - "verified": "verified", - "created_at": "2026-03-19T06:00:00+00:00" - } -] -``` - ---- - -### `GET /contacts/{job_id}` — list by job UUID - -```bash -curl http://localhost:8002/contacts/3fa85f64-5717-4562-b3fc-2c963f66afa6 -``` - ---- - -### `DELETE /contacts/{id}` — remove a contact - -```bash -curl -X DELETE http://localhost:8002/contacts/a1b2c3d4-... -# Returns 204 No Content on success -``` - ---- - -## Pipeline details - -### Email pattern detection - -Mines GitHub commit emails for the inferred domain. Voting on the most common local-part structure determines the pattern: - -| Pattern | Template string | -|---|---| -| `alice.smith@acme.com` | `{first}.{last}@{domain}` | -| `asmith@acme.com` | `{fi}{last}@{domain}` | -| `alices@acme.com` | `{first}{li}@{domain}` | -| `alice@acme.com` | `{first}@{domain}` | - -Falls back to `{first}.{last}@{domain}` if no GitHub commits are found. - -### SMTP verification - -| Result | Meaning | -|---|---| -| `verified` | MX server returned SMTP 250 (mailbox likely exists) | -| `unverified` | MX unreachable, 4xx response, or rate-limit hit | -| `invalid` | MX server returned 550/551 (mailbox does not exist) | - -> ⚠️ Large providers (Gmail, Outlook) always return 250 regardless of validity. Treat `verified` as a best-effort signal. - -**Rate limit:** 5 verifications per domain per hour (in-memory; restarts reset the counter). - -### LinkedIn selector notes (last verified March 2026) - -| Selector | Element | -|---|---| -| `li.result-card, div.base-search-card` | Job/person card | -| `span.actor-name, span.name` | Name text | -| `h3.base-search-card__title` | Name fallback | -| `p.subline-level-1` | Role / subtitle | - -Update these in `discovery.py → _linkedin_public_names()` when LinkedIn changes its markup. The scraper detects `/authwall` redirects and exits gracefully. - ---- - -## Database schema - -```sql -CREATE TABLE contacts ( - id UUID PRIMARY KEY DEFAULT gen_random_uuid(), - job_id UUID REFERENCES jobs(id) ON DELETE SET NULL, - company TEXT NOT NULL, - domain TEXT, - name TEXT, - email TEXT, - role TEXT, - source TEXT, - verified TEXT NOT NULL DEFAULT 'unverified', - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - UNIQUE (company, email) -); -``` +## Request/Data Flow + +1. `POST /discover` accepts company name, optional job_id, roles list, and domain, triggering background discovery pipeline. +2. Background task calls `discovery.run_pipeline()` to execute stages: domain inference, pattern detection, name discovery, email construction, and SMTP verification. +3. Verified contacts are upserted into PostgreSQL via `storage.upsert_contact()` with conflict resolution on (company, email). +4. `GET /contacts?company={company}` queries PostgreSQL for contacts by company substring match. +5. `GET /contacts/{job_id}` retrieves contacts linked to a specific job UUID. +6. `DELETE /contacts/{id}` removes a contact record from PostgreSQL. + +## Internal Execution Pipeline + +- **Domain Inference**: `domain_inference()` queries Clearbit autocomplete API, falls back to direct HTTPS probes of common suffixes (com, io, co.in) to determine company domain. +- **Email Pattern Detection**: `email_pattern_detection()` searches GitHub commits and org members for domain emails, infers patterns (e.g., {first}.{last}@{domain}) using regex matching, defaults to {first}.{last} if undetected. +- **Name Discovery**: `_github_org_names()` fetches GitHub org members' public profiles for names/roles, supplemented by LinkedIn search parsing for additional contacts. +- **Email Construction**: Combines discovered names with inferred pattern to generate email addresses. +- **SMTP Verification**: `verifier.verify_email()` resolves MX records, performs SMTP RCPT TO checks with per-domain rate limiting (5/hour), returning 'verified', 'unverified', or 'invalid'. +- **Storage Upsert**: `storage.upsert_contact()` uses `INSERT ... ON CONFLICT (company, email) DO UPDATE` to merge new data with existing records. + +## Important Modules/Files + +- `main.py`: FastAPI application with endpoints (`/discover`, `/contacts`, `/contacts/{job_id}`, `/contacts/{id}`), background task management, and Pydantic models for requests/responses. +- `discovery.py`: Multi-stage pipeline using httpx for HTTP requests, GitHub API for name/pattern mining, and regex for email pattern inference. +- `storage.py`: asyncpg pool lifecycle, schema DDL for contacts table with indexes, and CRUD functions with unique constraint on (company, email). +- `verifier.py`: SMTP verification with dnspython MX resolution, rate limiting via in-memory dict, and synchronous SMTP checks wrapped in asyncio executor. + +## Service Interactions + +- Reads jobs table from PostgreSQL to link contacts to job postings. +- Writes contacts records to PostgreSQL, shared with email-generator-service for recipient lookup. +- Makes external HTTP requests to Clearbit, GitHub API, and LinkedIn for public data mining. +- Performs SMTP connections to domain MX servers for email verification. + +## Debugging Notes + +- Domain inference logs Clearbit successes/failures as info/debug, with warnings for unresolved domains. +- GitHub API errors logged as debug, with rate limit handling via optional GITHUB_TOKEN. +- SMTP verification logs results per email, with rate limit warnings and connection errors as debug. +- Database upsert conflicts logged implicitly via ON CONFLICT, with unique constraint added dynamically. +- Pipeline stage failures in `discovery.py` may skip to next stage, logged with exceptions. +- Asyncpg timeouts default to 30 seconds, potentially affecting contact listing under load. From 239e67d20aced9797b091d0928912a7612637cef Mon Sep 17 00:00:00 2001 From: ArpitKumar8649 Date: Mon, 11 May 2026 17:45:05 +0000 Subject: [PATCH 5/9] docs: add README for crawler-service with internal architecture --- crawler-service/README.md | 241 +++++--------------------------------- 1 file changed, 27 insertions(+), 214 deletions(-) diff --git a/crawler-service/README.md b/crawler-service/README.md index 461f38f..0ccca6b 100644 --- a/crawler-service/README.md +++ b/crawler-service/README.md @@ -1,225 +1,38 @@ # Crawler Service -Part of the Arachnode(JobHunter) microservice system. This service crawls startup job -directories and career pages, filters results by your role and stack, deduplicates -them via Redis, and emits normalized `JobPosting` events onto a Redis Stream for -downstream services to consume. +## Request/Data Flow ---- +1. Scrapy spiders (e.g., remotive_spider, wellfound_spider) crawl job boards, yielding job items with company, role, stack, etc. +2. Items pass through pipelines: `DeduplicationPipeline` checks for duplicates, `StackFilterPipeline` filters by target stack/role, `RedisStreamPipeline` emits to 'jobs:raw' Redis stream. +3. `read_stream.py` monitors the Redis stream, displaying new job events in real-time or dumping historical entries. -## Project structure +## Internal Execution Pipeline -``` -crawler-service/ -├── crawler/ -│ ├── spiders/ -│ │ ├── base_spider.py # shared rate limiting and stack matching -│ │ ├── yc_spider.py # YC jobs page (plain HTML, fastest) -│ │ ├── wellfound_spider.py # Wellfound / AngelList (Playwright) -│ │ └── remotive_spider.py # Remotive JSON API (most reliable) -│ ├── parsers/ -│ │ ├── ats_detector.py # Lever / Greenhouse / Ashby API clients -│ │ └── generic_parser.py # Playwright fallback for custom career pages -│ ├── pipelines/ -│ │ ├── dedup_pipeline.py # Redis-based dedup (7-day TTL) -│ │ ├── filter_pipeline.py # Stack and role matching -│ │ └── emit_pipeline.py # Push to Redis Stream jobs:raw -│ ├── models.py # JobItem and JobPosting dataclass -│ ├── settings.py # Scrapy config -│ └── middlewares.py # User-agent rotation -├── tests/ -│ └── test_ats_detector.py -├── read_stream.py # Monitor Redis Stream output -├── run_local.sh # Run without Docker -├── Dockerfile -├── docker-compose.yml -└── requirements.txt -``` +- **Spider Parsing**: Base spider classes use `stack_matches()` and `role_matches()` to filter jobs based on JOBSEEKER_STACK and JOBSEEKER_ROLE settings. +- **Deduplication Pipeline**: `dedup_pipeline.py` uses Redis keys for URL-based deduplication, skipping already-seen jobs. +- **Stack Filter Pipeline**: `filter_pipeline.py` applies stack intersection checks, dropping items that don't match target technologies. +- **Redis Emit Pipeline**: `emit_pipeline.py` serializes items to JSON strings, adds to 'jobs:raw' stream with maxlen 10,000, logging emissions. +- **Stream Monitoring**: `read_stream.py` uses Redis XREAD/XREVRANGE for tailing or dumping stream contents, formatting job details. ---- +## Important Modules/Files -## Quickstart (local, no Docker) +- `read_stream.py`: CLI script for Redis stream monitoring with options for tailing, dumping, and count limits. +- `scrapy.cfg`: Scrapy project configuration pointing to crawler.settings. +- `crawler/settings.py`: Scrapy settings with politeness delays, Playwright middleware, pipeline order, Redis connection, and jobseeker profile. +- `crawler/spiders/base_spider.py`: Abstract base spider with stack/role matching logic and crawler instantiation. +- `crawler/pipelines/emit_pipeline.py`: Final pipeline stage emitting job items to Redis stream with JSON serialization and maxlen capping. -### 1. Prerequisites +## Service Interactions -- Python 3.11+ -- Redis running on localhost:6379 - ```bash - # macOS - brew install redis && brew services start redis - # Ubuntu - sudo apt install redis-server && sudo systemctl start redis - ``` +- Crawls external job board websites (Remotive, Wellfound, YC, Naukri, LinkedIn) using Scrapy with Playwright for JavaScript rendering. +- Writes job events to Redis 'jobs:raw' stream, consumed by aggregator-service for database insertion. +- Uses Redis for deduplication keys and stream storage. -### 2. Set up the Python environment +## Debugging Notes -```bash -cd crawler-service -python -m venv venv -source venv/bin/activate # Windows: venv\Scripts\activate -pip install -r requirements.txt -playwright install chromium # one-time browser install (~150MB) -``` - -### 3. Run your first spider - -Start with Remotive — it uses a public JSON API, never breaks, and is a -perfect way to verify the pipeline end-to-end before touching scrapers. - -```bash -# Edit your profile first, then: -export JOBSEEKER_ROLE="Backend Engineer" -export JOBSEEKER_STACK="Python,Go,FastAPI,PostgreSQL,Kubernetes" - -scrapy crawl remotive -``` - -Or use the convenience script: -```bash -chmod +x run_local.sh -./run_local.sh remotive -``` - -### 4. Check what was emitted - -```bash -python read_stream.py --count 20 # see last 20 jobs -python read_stream.py --all # dump everything -python read_stream.py # tail new events live -``` - ---- - -## Quickstart (Docker) - -```bash -docker-compose up --build -``` - -This starts Redis + the crawler (Remotive spider by default). -To run a different spider: - -```bash -docker-compose run crawler scrapy crawl wellfound -docker-compose run crawler scrapy crawl yc_jobs -``` - ---- - -## Available spiders - -| Spider | Source | JS needed | Notes | -|--------|--------|-----------|-------| -| `remotive` | remotive.com API | No | Best first spider to test with | -| `yc_jobs` | ycombinator.com/jobs | No | YC-backed companies only | -| `wellfound` | wellfound.com | Yes (Playwright) | Best breadth of funded startups | - ---- - -## Configuring your profile - -Edit in `settings.py` or pass as environment variables / `-s` flags: - -```bash -scrapy crawl remotive \ - -s JOBSEEKER_ROLE="Full Stack Engineer" \ - -s JOBSEEKER_STACK="React,Node.js,PostgreSQL,AWS" -``` - -Stack matching is OR-based — a job passes if *any* of your stack tags appear -in the posting's tags or role title. - ---- - -## Using the ATS parsers directly - -The Lever and Greenhouse parsers can be used standalone to pull jobs from any -company that uses these systems: - -```python -from crawler.parsers.ats_detector import fetch_lever_jobs, fetch_greenhouse_jobs - -# Get all open roles at Razorpay (Lever) -jobs = fetch_lever_jobs("razorpay") - -# Get all open roles at Notion (Greenhouse) -jobs = fetch_greenhouse_jobs("notion") - -for job in jobs: - print(job["role"], "—", job["url"]) -``` - ---- - -## Extending with a new spider - -1. Create `crawler/spiders/mysite_spider.py` -2. Subclass `BaseStartupSpider` -3. Implement `parse(self, response)` and yield `JobItem` objects -4. That's it — the pipelines (dedup, filter, emit) run automatically - -```python -from crawler.spiders.base_spider import BaseStartupSpider -from crawler.models import JobItem - -class MySiteSpider(BaseStartupSpider): - name = "mysite" - start_urls = ["https://example.com/jobs"] - - def parse(self, response): - for job in response.css("div.job"): - role = job.css("h3::text").get("") - if self.role_matches(role): - yield JobItem( - company="Example Corp", - role=role, - source=self.name, - url=response.url, - stack=[], - ) -``` - ---- - -## Running tests - -```bash -pytest tests/ -v -``` - ---- - -## Scheduling (cron) - -To run automatically every 8 hours: - -```bash -crontab -e -``` - -Add: -``` -0 */8 * * * cd /path/to/crawler-service && source venv/bin/activate && scrapy crawl remotive >> logs/remotive.log 2>&1 -0 */8 * * * cd /path/to/crawler-service && source venv/bin/activate && scrapy crawl yc_jobs >> logs/yc.log 2>&1 -``` - ---- - -## Troubleshooting - -**Redis connection refused** -Make sure Redis is running: `redis-cli ping` should return `PONG`. - -**Playwright: browser not found** -Run `playwright install chromium` inside your virtualenv. - -**Wellfound returns nothing** -Wellfound has aggressive bot detection. Add a longer delay: -`scrapy crawl wellfound -s DOWNLOAD_DELAY=5` - -**Stream is empty after a run** -Check that items aren't all being dropped by the filter pipeline. -Run with `-s LOG_LEVEL=DEBUG` to see drop reasons. - -**Items showing as duplicates immediately** -The dedup TTL is 7 days. To reset: `redis-cli DEL $(redis-cli KEYS "dedup:*")` +- Spider logs include parse successes/failures, with debug level for filtered items. +- Pipeline logs emission counts and skipped duplicates, with warnings for Redis connection issues. +- Playwright timeouts logged as errors, with AUTOTHROTTLE adjusting delays for rate limiting. +- Stream monitoring logs connection errors if Redis is unreachable. +- Item serialization errors in emit_pipeline logged as exceptions, potentially dropping malformed items. +- Concurrent requests limited to 4 globally, 1 per domain, to avoid bans. From e51b83942f4536fcee08097e7bd39066b8fe6da8 Mon Sep 17 00:00:00 2001 From: ArpitKumar8649 Date: Mon, 11 May 2026 17:45:49 +0000 Subject: [PATCH 6/9] docs: add README for scraper-service with internal architecture --- scraper-service/README.md | 198 ++++++-------------------------------- 1 file changed, 27 insertions(+), 171 deletions(-) diff --git a/scraper-service/README.md b/scraper-service/README.md index ef75e29..3ed85e1 100644 --- a/scraper-service/README.md +++ b/scraper-service/README.md @@ -1,181 +1,37 @@ -# Platform Scraper Service +# Scraper Service -A FastAPI microservice that scrapes **Naukri.com**, **LinkedIn** (public pages only), and **Internshala** for engineering job listings and emits them onto the same `jobs:raw` Redis Stream consumed by the Job Aggregator Service. +## Request/Data Flow -``` -POST /scrape → asyncio.gather([NaukriScraper, LinkedInScraper, IntershalaScraper]) - ↓ each scraper - jobs:raw Redis Stream → Aggregator Service → PostgreSQL -``` +1. `POST /scrape` accepts role and stack parameters, triggering background concurrent scraping of Naukri, LinkedIn, and Internshala. +2. Each scraper (NaukriScraper, LinkedInScraper, IntershalaScraper) runs asynchronously, returning job dicts with company, role, url, etc. +3. `emit.emit_jobs()` pushes each job dict to the 'jobs:raw' Redis stream, serializing stack to JSON and capping stream length at 50,000. +4. Background task completes, logging emission counts per platform. ---- +## Internal Execution Pipeline -## Project layout +- **Scraper Execution**: `main._run_all_scrapers()` uses `asyncio.gather()` to run all platform scrapers in parallel, capturing exceptions per scraper. +- **LinkedIn Scraping**: `linkedin.LinkedInScraper.scrape()` launches Playwright browser, navigates to search URL, scrolls to load cards, parses selectors for title, company, location, URL, and posted date. +- **Emission Process**: `emit.emit_job()` normalizes fields, checks for required company/role, adds to Redis stream with maxlen and approximate trimming. +- **Error Handling**: Scraper exceptions logged as errors, emission failures logged with exceptions, skipping malformed jobs. -``` -scraper-service/ -├── main.py # FastAPI app — POST /scrape, GET /health -├── emit.py # Shared Redis Stream emitter -├── scrapers/ -│ ├── __init__.py -│ ├── base.py # PlatformScraper ABC -│ ├── naukri.py # httpx + BeautifulSoup (3 pages) -│ ├── linkedin.py # Playwright headless (public-only) -│ └── internshala.py # httpx + BeautifulSoup (2 pages) -├── requirements.txt -├── Dockerfile -└── README.md -``` +## Important Modules/Files ---- +- `main.py`: FastAPI application with /scrape endpoint, background task management, and concurrent scraper execution. +- `emit.py`: Async Redis emitter with connection pooling, job normalization, and stream addition with maxlen capping. +- `scrapers/base.py`: Abstract base class defining scraper contract with source_name and scrape() method. +- `scrapers/linkedin.py`: Playwright-based LinkedIn scraper with selector parsing, scroll delays, and auth-wall detection. -## Configuration (environment variables) +## Service Interactions -| Variable | Default | Description | -|---|---|---| -| `REDIS_HOST` | `localhost` | Redis hostname | -| `REDIS_PORT` | `6379` | Redis port | -| `SCRAPER_DELAY_SECONDS` | `3` | Base delay between requests per scraper | +- Scrapes job listings from Naukri, LinkedIn public search, and Internshala using HTTP requests and Playwright for JavaScript sites. +- Emits job events to Redis 'jobs:raw' stream, consumed by aggregator-service for database storage. +- Uses Redis for stream storage and emission. ---- +## Debugging Notes -## Quick start - -### Docker - -```bash -# From the project root (requires Redis already running, e.g. from aggregator-service) -cd scraper-service -docker build -t scraper-service . -docker run -e REDIS_HOST=host.docker.internal -p 8001:8000 scraper-service -``` - -### Local dev - -```bash -cd scraper-service -python -m venv .venv && source .venv/bin/activate -pip install -r requirements.txt -playwright install chromium # download Chromium binary once - -export REDIS_HOST=localhost -export REDIS_PORT=6379 - -uvicorn main:app --reload --port 8001 -``` - ---- - -## API endpoints - -### `GET /health` - -```bash -curl http://localhost:8001/health -``` -```json -{"status": "ok"} -``` - ---- - -### `POST /scrape` — trigger all scrapers - -The endpoint returns **immediately**; scraping happens in the background. - -```bash -curl -X POST http://localhost:8001/scrape \ - -H "Content-Type: application/json" \ - -d '{"role": "Backend Engineer", "stack": ["Python", "Go"]}' -``` -```json -{ - "triggered": true, - "platforms": ["naukri", "linkedin", "internshala"] -} -``` - -Jobs flow into the Redis Stream and are picked up by the Aggregator Service automatically. Query them via: - -```bash -curl "http://localhost:8000/jobs?sort=latest&limit=20" -``` - ---- - -## Selector reference & maintenance guide - -All CSS selectors are defined as **named module-level constants** at the top of each scraper file. When a platform changes its markup, update only those constants. - -### Naukri.com (`scrapers/naukri.py`) - -> Naukri delivers most listing data as server-side HTML on the first 3 pages. -> If they shift to a full JS API, switch to `https://www.naukri.com/jobapi/v3/search?keyword={role}&pageNo={n}`. - -| Selector constant | Target element | -|---|---| -| `card_sel` | `article.jobTuple` / `div.srp-jobtuple-root` — one job card | -| `title_sel` | `a.title` / `a.jobTitle` — job title anchor | -| `company_sel` | `a.comp-name` / `span.comp-name` — company name | -| `loc_sel` | `span.loc span a` / `li.location span` — city / location | -| `skills_sel` | `ul.tags-gt li` / `ul.tags li` — skill badge list items | -| `date_sel` | `span.job-post-day` — "3 days ago" text | - -URL pattern: `https://www.naukri.com/{role-slug}-jobs?pageNo={1,2,3}` - ---- - -### LinkedIn (`scrapers/linkedin.py`) - -> ⚠️ LinkedIn's DOM changes frequently. Selectors were last verified March 2026. -> If you see 0 results, inspect the page source and update the constants below. -> If you are redirected to `/authwall`, your IP has been rate-limited — wait 30–60 min. - -| Constant | CSS selector | Notes | -|---|---|---| -| `_CARD_SEL` | `li > div.base-search-card` | Outermost card wrapper | -| `_TITLE_SEL` | `h3.base-search-card__title` | Job title | -| `_COMPANY_SEL` | `h4.base-search-card__subtitle` | Company name | -| `_LOCATION_SEL` | `span.job-search-card__location` | City/country | -| `_LINK_SEL` | `a.base-card__full-link` | Card permalink | -| `_TIME_SEL` | `time` | ISO datetime attribute | - -URL pattern: `https://www.linkedin.com/jobs/search/?keywords={role}&location=India&geoId=102713980&f_TPR=r604800` - -**Known limitations:** -- `stack` field is always empty (not exposed on listing pages). -- Only ~25 cards are loaded per scroll session (no multi-page support without auth). -- Playwright Chromium adds ~400 MB to the Docker image and 3–5 s cold-start latency. - ---- - -### Internshala (`scrapers/internshala.py`) - -> Mostly server-rendered; httpx works well. CAPTCHA appears after excessive requests. -> Keep `SCRAPER_DELAY_SECONDS ≥ 2`. - -| Selector constant | Target element | -|---|---| -| `card_sel` | `div.individual_internship` — one posting | -| `title_sel` | `h3.job-internship-name a` / `div.profile h3 a` — title anchor | -| `company_sel` | `h4.company-name a` / `div.company_name a` — company | -| `loc_sel` | `a.location_link` — city | -| `skills_sel` | `div.round_tabs_container span.round_tabs` — skill badges | -| `stipend_sel` | `div.stipend_container span.stipend` — stipend/salary (mapped to `product` field) | - -URL pattern: `https://internshala.com/jobs/keywords-{role-slug}/` (page 2: `?page=2`) - ---- - -## Rate limiting & operational notes - -| Platform | Strategy | Min delay | -|---|---|---| -| Naukri | Randomised delay between pages (`SCRAPER_DELAY_SECONDS ± 1 s`) | 1.5 s | -| LinkedIn | Fixed scroll delay (enforced ≥ 4 s regardless of env var) | 4 s | -| Internshala | Randomised delay between pages | 2 s | - -- All scrapers respect a User-Agent that mimics a real Chrome browser. -- The `POST /scrape` endpoint runs all three scrapers **concurrently** via `asyncio.gather`. - Total latency is bounded by the slowest scraper (typically LinkedIn Playwright ~20–30 s). -- The service does **not** manage dedup itself — that is handled by the Aggregator Service's Redis dedup key (`dedup:agg:{md5}`). +- Scraper logs include job counts per platform, with errors for exceptions during scraping. +- Playwright timeouts and selector failures logged as warnings/debug, with auth redirects indicating rate limits. +- Emission logs debug-level for each job, with exceptions for Redis connection issues. +- Concurrent scraper runs may have varying completion times, logged per platform. +- Stream maxlen may drop old events if emission rate exceeds consumption. +- Selector updates required when LinkedIn changes markup, logged as parsing errors. From 1429c3f3eee2d12386bf3db849ff7a5c8f7720fd Mon Sep 17 00:00:00 2001 From: ArpitKumar8649 Date: Mon, 11 May 2026 17:46:33 +0000 Subject: [PATCH 7/9] docs: add README for gateway with internal architecture --- gateway/README.md | 183 +++++++--------------------------------------- 1 file changed, 27 insertions(+), 156 deletions(-) diff --git a/gateway/README.md b/gateway/README.md index 614cfa5..39d5d83 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -1,166 +1,37 @@ -# API Gateway & Dashboard +# Gateway Service -The API Gateway proxies all four backend services behind a single port and serves a vanilla-JS dashboard at `GET /`. +## Request/Data Flow -``` -Browser / curl - │ - ▼ - Gateway :8080 - ├── /api/jobs/* → aggregator:8000 - ├── /api/scrape → scraper:8000 - ├── /api/contacts/* → contact:8000 - ├── /api/emails/* → email-gen:8000 - ├── /api/workflow/apply ← composite (orchestrated here) - ├── /api/health ← fans out to all services - └── / ← dashboard.html -``` +1. Proxy routes (/api/jobs/*, /api/scrape, /api/contacts/*, /api/emails/*) forward requests to aggregator, scraper, contact-discovery, and email-generator services using httpx. +2. `GET /api/health` fans out health checks to all four services concurrently, returning aggregated status. +3. `GET /api/summary` reads scheduler run summary from shared volume at /data/run_summary.json. +4. `POST /api/workflow/apply` orchestrates composite workflow: fetches job, triggers contact discovery, retrieves contacts, generates email draft. +5. `GET /` serves dashboard.html as a single-page application. ---- +## Internal Execution Pipeline -## Project layout +- **Proxying**: `proxy.proxy_request()` forwards method, headers (excluding hop-by-hop), body, and query params to upstream, returning Response with preserved status/content-type. +- **Health Fan-out**: `main.gateway_health()` uses `asyncio.gather()` for parallel health checks, aggregating results into JSON response. +- **Composite Workflow**: `main.workflow_apply()` calls `proxy.get_job()`, `proxy.trigger_discovery()`, `proxy.get_contacts_for_company()`, `proxy.generate_email()` sequentially, handling exceptions. +- **Dashboard Serving**: `main.dashboard()` returns FileResponse for dashboard.html, which uses JavaScript to call /api/* endpoints. -``` -gateway/ -├── main.py # FastAPI gateway — routes + composite endpoint -├── proxy.py # httpx helpers: generic forwarder + typed workflow helpers -├── dashboard.html # Single-file vanilla JS dashboard -├── requirements.txt -├── Dockerfile -└── README.md -``` +## Important Modules/Files ---- +- `main.py`: FastAPI application with proxy routes, composite workflow endpoint, health fan-out, and dashboard serving. +- `proxy.py`: httpx client management, generic proxy_request() helper, typed service helpers for workflow orchestration. +- `dashboard.html`: Single-page dashboard SPA with tabs for jobs, contacts, emails, stats, using Chart.js and fetch API for /api/* calls. -## Starting the full stack +## Service Interactions -### Prerequisites +- Proxies requests to aggregator-service (jobs/stats), scraper-service (scrape), contact-discovery-service (contacts/discover), email-generator-service (emails/generate). +- Reads scheduler run summary from shared Docker volume. +- Serves dashboard that consumes all /api/* endpoints. -```bash -# Copy env template and fill in your values -cp .env.example .env # see variables listed in root README -``` +## Debugging Notes -### One command to launch everything - -```bash -# From the repository root -docker compose up --build - -# Then open the dashboard -open http://localhost:8080 -``` - -### Service startup order (automatic via healthchecks) - -``` -redis + postgres → aggregator + scraper + contact + email-gen → gateway -``` - -The gateway will not start until all four backend services report healthy. - -### Run the crawler (one-shot, separate command) - -```bash -docker compose run --rm crawler scrapy crawl remotive -``` - ---- - -## Using the dashboard - -### Typical workflow - -1. **Open** `http://localhost:8080` -2. **Jobs tab** — click **▶ Run Scraper** to fetch fresh listings from Naukri, LinkedIn & Internshala. Jobs appear within ~30 s. -3. On any job card, click **👤 Discover** to find engineering contacts at that company (runs in background). -4. Click **✉ Draft** — the gateway calls `/api/workflow/apply`, which orchestrates job lookup → contact discovery → email generation in one request and opens a preview modal. -5. Click **Send** inside the modal to send the email via Gmail. -6. Switch to **Contacts**, **Emails**, and **Stats** tabs to view results. - ---- - -## API reference - -Interactive docs: `http://localhost:8080/api/docs` - -### `GET /api/health` — fan-out health check - -```bash -curl http://localhost:8080/api/health -``` -```json -{ - "gateway": "ok", - "services": [ - {"service": "aggregator", "status": "ok"}, - {"service": "scraper", "status": "ok"}, - {"service": "contact", "status": "ok"}, - {"service": "email-gen", "status": "ok"} - ] -} -``` - -### `POST /api/workflow/apply` — full apply workflow - -```bash -curl -X POST http://localhost:8080/api/workflow/apply \ - -H "Content-Type: application/json" \ - -d '{"job_id": "", "template": "cold_outreach"}' -``` -```json -{ - "job": { "id": "...", "company": "Razorpay", "role": "Backend Engineer", ... }, - "contacts": [{ "name": "Alice Smith", "email": "alice.smith@razorpay.com", ... }], - "draft_email": { "email_id": "...", "subject": "...", "body": "..." } -} -``` - -### Proxy routes - -| Gateway path | Upstream | -|---|---| -| `GET/POST /api/jobs` | aggregator `/jobs` | -| `GET/PATCH /api/jobs/{id}/status` | aggregator `/jobs/{id}/status` | -| `GET /api/stats` | aggregator `/stats` | -| `POST /api/scrape` | scraper `/scrape` | -| `POST /api/discover` | contact `/discover` | -| `GET /api/contacts` | contact `/contacts` | -| `DELETE /api/contacts/{id}` | contact `/contacts/{id}` | -| `POST /api/generate` | email-gen `/generate` | -| `GET /api/emails` | email-gen `/emails` | -| `PATCH /api/emails/{id}/status` | email-gen `/emails/{id}/status` | -| `POST /api/emails/{id}/send` | email-gen `/emails/{id}/send` | - ---- - -## Adding a new service behind the gateway - -1. Add the service to `docker-compose.yml` on the `jobnet` network. -2. Add its base URL as an env var in `proxy.py` (e.g. `MY_SVC_URL`). -3. Add a proxy route in `main.py`: - ```python - @app.api_route("/api/my-service/{path:path}", methods=["GET","POST"]) - async def proxy_my_service(path: str, request: Request): - return await proxy.proxy_request(request, f"{proxy.MY_SVC_URL}/{path}") - ``` -4. Rebuild the gateway container: `docker compose up --build gateway` - ---- - -## Environment variables (root `.env`) - -| Variable | Default | Description | -|---|---|---| -| `POSTGRES_USER` | `jobuser` | | -| `POSTGRES_PASSWORD` | `jobpass` | | -| `POSTGRES_DB` | `jobsdb` | | -| `JOBSEEKER_ROLE` | `Backend Engineer` | Crawler search role | -| `JOBSEEKER_STACK` | `Python,...` | Crawler stack filter | -| `GITHUB_TOKEN` | — | Contact discovery (optional, raises GH rate limit) | -| `GMAIL_ADDRESS` | — | Gmail sender address | -| `GMAIL_APP_PASSWORD` | — | 16-char Gmail App Password | -| `YOUR_NAME` | `Applicant` | Used in email templates | -| `YOUR_GITHUB_URL` | — | Embedded in email templates | -| `OLLAMA_BASE_URL` | `http://host.docker.internal:11434` | Ollama server for email gen | -| `GATEWAY_PORT` | `8080` | Host port for the gateway | +- Proxy timeouts logged as 503/504 HTTPExceptions, with upstream URL in detail. +- Health check failures logged in aggregated response, with unreachable services marked. +- Workflow orchestration logs exceptions per step, raising HTTPException on failures. +- Dashboard JavaScript errors not logged server-side, debug via browser console. +- Shared volume read errors for summary logged as 500 responses. +- Concurrent health checks may have varying response times, aggregated in single JSON. From 74accafb8676d6c9de4344f35bac0eb7837b3208 Mon Sep 17 00:00:00 2001 From: ArpitKumar8649 Date: Mon, 11 May 2026 17:47:23 +0000 Subject: [PATCH 8/9] docs: add README for scheduler with internal architecture --- scheduler/README.md | 192 ++++++-------------------------------------- 1 file changed, 26 insertions(+), 166 deletions(-) diff --git a/scheduler/README.md b/scheduler/README.md index 8f923f3..1d2a88c 100644 --- a/scheduler/README.md +++ b/scheduler/README.md @@ -1,177 +1,37 @@ # Scheduler Service -A standalone Python process (not FastAPI) that automates the full job discovery pipeline on a cron schedule using **APScheduler 3.x**. +## Request/Data Flow ---- +1. APScheduler runs `run_scrape_cycle()` every CRAWL_INTERVAL_HOURS: POST /api/scrape to gateway, runs Scrapy subprocess for remotive/yc_jobs spiders, waits for pipelines, counts job delta. +2. APScheduler runs `run_discover_cycle()` every DISCOVER_INTERVAL_HOURS: fetches new jobs via GET /api/jobs, POST /api/discover for each job with delay between calls. +3. APScheduler runs `run_draft_cycle()` every DISCOVER_INTERVAL_HOURS: fetches new jobs, checks for contacts via GET /api/contacts, POST /api/generate for jobs with contacts. +4. After each task, summary is written to /data/run_summary.json with jobs_discovered, contacts_found, emails_drafted, errors. -## Architecture +## Internal Execution Pipeline -``` -APScheduler (in-process, background scheduler) - │ - ├── Every 8h : run_scrape_cycle() - │ POST /api/scrape ← platform scrapers (Naukri/LI/Internshala) - │ scrapy crawl remotive ← via subprocess - │ scrapy crawl yc_jobs ← via subprocess - │ Wait 60s → count delta jobs - │ - ├── Every 24h (+4h offset): run_discover_cycle() - │ For each new job → POST /api/discover - │ 30s delay between calls - │ - └── Every 24h (+8h offset): run_draft_cycle() - For each new job with contacts → POST /api/generate - Pre-generates cold_outreach drafts in Postgres +- **Scrape Cycle**: `tasks.run_scrape_cycle()` counts jobs before, triggers platform scrapers via httpx POST, runs Scrapy subprocess with timeout, waits SCRAPER_WAIT_SECS, counts delta and updates summary. +- **Discover Cycle**: `tasks.run_discover_cycle()` fetches new jobs via httpx GET, loops over jobs, POST /api/discover with DISCOVER_DELAY_SECS pause, increments contacts_found on success. +- **Draft Cycle**: `tasks.run_draft_cycle()` fetches new jobs, checks contacts via GET /api/contacts, generates drafts via POST /api/generate for jobs with contacts, updates emails_drafted. +- **Summary Flushing**: `main._write_summary()` serializes summary dict to JSON file, logged on success/failure. +- **Manual Runs**: `main._maybe_manual_run()` dispatches single tasks or all via MANUAL_TASK env var, exits after completion. -After each cycle → /data/run_summary.json -Gateway reads it via GET /api/summary -``` +## Important Modules/Files ---- +- `main.py`: APScheduler setup with interval jobs for scrape/discover/draft, graceful shutdown handling, summary writing, manual task dispatch. +- `tasks.py`: Synchronous task functions with httpx calls to gateway, subprocess for Scrapy, summary state management, error recording. +- `logger.py`: JSONFormatter for structured logging to stdout, StructLogger adapter for extra fields. -## Project layout +## Service Interactions -``` -scheduler/ -├── main.py # APScheduler setup, signal handlers, manual run -├── tasks.py # One function per scheduled task -├── logger.py # Structured JSON logging to stdout -├── requirements.txt -├── Dockerfile -└── README.md -``` +- Calls gateway /api/scrape, /api/jobs, /api/discover, /api/contacts, /api/generate via httpx. +- Runs Scrapy subprocess for crawler-service spiders. +- Writes run summary to shared Docker volume /data/run_summary.json, read by gateway. ---- +## Debugging Notes -## Configuration (environment variables) - -| Variable | Default | Description | -|---|---|---| -| `GATEWAY_URL` | `http://gateway:8080` | Base URL for all API calls | -| `JOBSEEKER_ROLE` | `Backend Engineer` | Role sent to scraper | -| `JOBSEEKER_STACK` | `Python,FastAPI,...` | Stack filter (comma-separated) | -| `CRAWL_INTERVAL_HOURS` | `8` | How often to run the scrape cycle | -| `DISCOVER_INTERVAL_HOURS` | `24` | How often to run discover + draft cycles | -| `SCRAPER_WAIT_SECS` | `60` | Wait time after triggering scrapers | -| `DISCOVER_DELAY_SECS` | `30` | Delay between each per-job discover call | -| `SCRAPY_PROJECT_DIR` | `/crawler` | CWD for `scrapy crawl` subprocess | -| `SUMMARY_PATH` | `/data/run_summary.json` | Output path for run summary | - ---- - -## Quick start - -### Docker (recommended — part of full stack) - -```bash -cd /path/to/jobCrawler -docker compose up --build -# Scheduler auto-starts after gateway is healthy -``` - -### Local dev - -```bash -cd scheduler -python -m venv .venv && source .venv/bin/activate -pip install -r requirements.txt - -export GATEWAY_URL="http://localhost:8080" -export JOBSEEKER_ROLE="Backend Engineer" -export JOBSEEKER_STACK="Python,Go,FastAPI" - -python main.py -``` - ---- - -## Triggering a manual run without waiting for the cron - -Set the `MANUAL_TASK` environment variable before starting the process. -The scheduler will run that task immediately and exit (no APScheduler loop). - -```bash -# Run only the scrape cycle now -MANUAL_TASK=scrape python main.py - -# Run only the discover cycle now -MANUAL_TASK=discover python main.py - -# Run only the draft cycle now -MANUAL_TASK=draft python main.py - -# Run all three cycles in sequence -MANUAL_TASK=all python main.py -``` - -**Via Docker (without restarting the container):** - -```bash -docker compose run --rm \ - -e MANUAL_TASK=scrape \ - -e GATEWAY_URL=http://gateway:8080 \ - scheduler -``` - ---- - -## Changing the schedule - -Edit `main.py → build_scheduler()`. - -The three jobs use `trigger="interval"` with an `hours=` parameter. -To switch to a cron-style trigger (e.g., run at 6 AM every day): - -```python -from apscheduler.triggers.cron import CronTrigger - -scheduler.add_job( - func=lambda: _run(tasks.run_scrape_cycle, "scrape"), - trigger=CronTrigger(hour=6, minute=0, timezone="Asia/Kolkata"), - id="scrape", -) -``` - -Alternatively, adjust the `CRAWL_INTERVAL_HOURS` and `DISCOVER_INTERVAL_HOURS` -environment variables without rebuilding the image. - ---- - -## Run summary - -The scheduler writes `/data/run_summary.json` after every cycle: - -```json -{ - "run_at": "2026-03-19T12:00:00Z", - "jobs_discovered": 42, - "contacts_found": 18, - "emails_drafted": 12, - "errors": [] -} -``` - -Read it via the gateway: - -```bash -curl http://localhost:8080/api/summary -``` - -The `/data` directory is a shared Docker volume (`scheduler_data`) mounted by -both the **scheduler** (writer) and the **gateway** (reader). - ---- - -## Graceful shutdown - -The process handles `SIGTERM` and `SIGINT`: - -``` -SIGTERM received - → APScheduler stops accepting new jobs - → Waits for the currently executing task to complete - → Exits cleanly -``` - -Docker sends `SIGTERM` on `docker compose stop` / `docker compose down`, -so no data will be lost mid-cycle. +- Task exceptions logged as JSON with context/detail, appended to summary errors. +- HTTP timeouts/failures logged in tasks, with _record_error() updating summary. +- Subprocess timeouts/errors logged, with stderr captured in summary. +- Scheduler misfires logged via APScheduler, with grace time 300s. +- Manual runs exit after task, useful for testing without schedule. +- Concurrent tasks prevented by _task_lock, with warnings for skipped runs. From 5825a4c5514f2eb615c9c2b11914798460f63494 Mon Sep 17 00:00:00 2001 From: Arpit Kumar Date: Wed, 27 May 2026 17:51:03 +0000 Subject: [PATCH 9/9] docs: add architecture docs and narrow READMEs to contributor-focused content --- aggregator-service/ARCHITECTURE.md | 47 +++++++++++++++++ aggregator-service/README.md | 37 ++----------- contact-discovery-service/ARCHITECTURE.md | 50 ++++++++++++++++++ contact-discovery-service/README.md | 41 ++------------- crawler-service/ARCHITECTURE.md | 48 +++++++++++++++++ crawler-service/README.md | 37 ++----------- email-generator-service/ARCHITECTURE.md | 63 +++++++++++++++++++++++ email-generator-service/README.md | 44 ++-------------- gateway/ARCHITECTURE.md | 42 +++++++++++++++ gateway/README.md | 36 ++----------- scheduler/ARCHITECTURE.md | 47 +++++++++++++++++ scheduler/README.md | 36 ++----------- scraper-service/ARCHITECTURE.md | 44 ++++++++++++++++ scraper-service/README.md | 36 ++----------- 14 files changed, 369 insertions(+), 239 deletions(-) create mode 100644 aggregator-service/ARCHITECTURE.md create mode 100644 contact-discovery-service/ARCHITECTURE.md create mode 100644 crawler-service/ARCHITECTURE.md create mode 100644 email-generator-service/ARCHITECTURE.md create mode 100644 gateway/ARCHITECTURE.md create mode 100644 scheduler/ARCHITECTURE.md create mode 100644 scraper-service/ARCHITECTURE.md diff --git a/aggregator-service/ARCHITECTURE.md b/aggregator-service/ARCHITECTURE.md new file mode 100644 index 0000000..0adf83e --- /dev/null +++ b/aggregator-service/ARCHITECTURE.md @@ -0,0 +1,47 @@ +# Aggregator — Architecture & Contributor Notes + +This file focuses on contributor-oriented/internal architecture information: request/data flow, internal execution pipeline, important modules/files, debugging notes, and service interactions. + +## Overview + +The Aggregator service consumes raw job events from a Redis stream, deduplicates and normalizes them, and persists canonical job records to PostgreSQL for downstream services. + +## Request / Data Flow + +1. Consumer reads from Redis `jobs:raw` stream (consumer groups) and claims pending messages when necessary. +2. Each message is parsed and normalized (company, role, stack, posted_at) and passed to `db.insert_job()`. +3. `db.insert_job()` attempts `INSERT ... ON CONFLICT DO NOTHING` to avoid duplicates; successful inserts set a deduplication key in Redis with a TTL. +4. The service exposes `GET /jobs`, `GET /jobs/{id}`, `PATCH /jobs/{id}/status`, and `GET /stats` for querying aggregated data. + +## Internal Execution Pipeline + +- `consumer.py`: connection and consumer group management, `run_consumer()` loop, pending message claiming, and batch processing using `XREADGROUP`/`XAUTOCLAIM`. +- `_process_message()`: normalizes payloads, computes dedup key (MD5 of normalized company+role), and invokes `db.insert_job()`. +- `db.py`: asyncpg pool lifecycle, schema setup with GIN/BTREE indexes, and query helpers supporting ILIKE and array GIN containment for stack queries. + +## Important Modules / Files + +- `main.py` — FastAPI entry, endpoint handlers, lifespan hooks for pool and consumer task management. +- `consumer.py` — Redis stream consumer and message processing logic. +- `db.py` — database schema, pool, and CRUD helpers. +- `models.py` — Pydantic response/request models. + +## Service Interactions + +- Consumes `jobs:raw` Redis stream emitted by scraper-service. +- Persists canonical job records to PostgreSQL, read by contact-discovery and email-generator services. + +## Debugging Notes + +- Missing fields in messages are logged and skipped; add extra logging in `_process_message()` for edge cases. +- Database or Redis connectivity issues surface during lifespan startup; verify pool/connection settings and credentials. +- Query timeouts default to 30s; increase when running long analytical queries. + +## Contributor Tips + +- To test deduplication, craft messages with similar normalized company+role and observe Redis dedup key TTL behavior. +- Add unit tests for `db.insert_job()` when changing schema or uniqueness constraints. + +## Next (Operational) Docs + +Operational/API usage (example requests, deployment env vars) are deferred to a follow-up docs issue. diff --git a/aggregator-service/README.md b/aggregator-service/README.md index e85b192..54ed61e 100644 --- a/aggregator-service/README.md +++ b/aggregator-service/README.md @@ -1,38 +1,9 @@ # Aggregator Service -## Request/Data Flow +This repository contains the Aggregator service. Contributor-oriented architecture and developer notes are moved to a dedicated file to reduce duplication with other services. -1. `GET /jobs` accepts query parameters for role, stack, status, sort, and limit, querying PostgreSQL with filters and returning job records. -2. `GET /jobs/{id}` fetches a single job by UUID from PostgreSQL. -3. `PATCH /jobs/{id}/status` updates the job status in PostgreSQL and returns the updated record. -4. `GET /stats` aggregates counts by source and status from PostgreSQL. -5. Background consumer reads messages from Redis 'jobs:raw' stream, parses fields, checks deduplication key, and conditionally inserts into PostgreSQL. +See the contributor-oriented material in: [aggregator-service/ARCHITECTURE.md](aggregator-service/ARCHITECTURE.md) -## Internal Execution Pipeline +Operational/API usage and setup details are deferred to a follow-up docs issue to avoid duplication across services. -- **Consumer Loop**: `consumer.run_consumer()` connects to Redis, ensures consumer group exists, claims pending messages, then reads in batches using `XPENDING` and `XREADGROUP`, processing each message asynchronously. -- **Message Processing**: `_process_message()` decodes stream fields, normalizes company/role for deduplication, checks Redis key existence, parses stack as JSON array and posted_at as datetime, then calls `db.insert_job()` with conflict resolution. -- **Deduplication**: `_dedup_key()` generates MD5 hash from normalized company+role, sets Redis key with 7-day TTL on successful insert to prevent reprocessing. -- **Database Operations**: `db.insert_job()` uses `INSERT ... ON CONFLICT DO NOTHING` for URL uniqueness, `get_jobs()` applies ILIKE for role substring and GIN containment for stack arrays, with sorting by posted_at. -- **Stats Aggregation**: `db.get_stats()` executes GROUP BY queries on source and status columns, returning dictionaries of counts. - -## Important Modules/Files - -- `main.py`: FastAPI application with endpoints (`/jobs`, `/jobs/{id}`, `/jobs/{id}/status`, `/stats`), lifespan hooks for pool and consumer task management, and Pydantic response models. -- `consumer.py`: Async Redis stream consumer with consumer group management, message parsing, deduplication logic, and batch processing using `XAUTOCLAIM` for pending messages. -- `db.py`: asyncpg pool lifecycle, schema DDL for jobs table with GIN/BTREE indexes, and CRUD functions with parameterized queries and conflict handling. -- `models.py`: Pydantic models for `JobOut`, `StatusUpdate`, and `StatsOut` with validation for status values and attribute-based configuration. - -## Service Interactions - -- Consumes messages from Redis 'jobs:raw' stream using consumer groups for reliable delivery. -- Writes job records to PostgreSQL jobs table, shared with other services like contact-discovery-service and email-generator-service. -- Provides REST API for job queries and status updates, consumed by gateway or other components. - -## Debugging Notes - -- Consumer logs warnings for missing company/role in messages, debug for duplicates skipped, and info for successful inserts with job IDs. -- Database connection errors raise `RuntimeError` if pool is uninitialized, logged during lifespan startup. -- Redis connection issues in consumer loop may cause retries, with group creation errors logged as info if already exists. -- Query timeouts in `db.py` default to 30 seconds, potentially delaying API responses under high load. -- Deduplication key collisions logged as debug, with MD5 hash generation ensuring consistent keying. +If you need quick operational references, inspect `main.py`, `consumer.py`, and `db.py` directly. diff --git a/contact-discovery-service/ARCHITECTURE.md b/contact-discovery-service/ARCHITECTURE.md new file mode 100644 index 0000000..077cd9f --- /dev/null +++ b/contact-discovery-service/ARCHITECTURE.md @@ -0,0 +1,50 @@ +# Contact Discovery — Architecture & Contributor Notes + +This file focuses on contributor-oriented/internal architecture information: request/data flow, internal execution pipeline, important modules/files, debugging notes, and service interactions. + +## Overview + +The Contact Discovery service attempts to discover likely contact emails for a company using public data sources, pattern inference, and SMTP verification, then persists verified contacts to PostgreSQL for downstream use. + +## Request / Data Flow + +1. `POST /discover` starts a multi-stage pipeline for a given company (optionally with `job_id` and roles). +2. Pipeline stages: domain inference, email pattern detection, name discovery (GitHub/LinkedIn), email construction, SMTP verification. +3. Verified contacts are upserted into PostgreSQL using `storage.upsert_contact()`. +4. API supports querying contacts via `GET /contacts` and deletion via `DELETE /contacts/{id}`. + +## Internal Execution Pipeline + +- `discovery.py`: orchestrates domain inference, pattern mining, name discovery, and verification. +- `domain_inference()`: Clearbit first, fallback to suffix probes and heuristics. +- `email_pattern_detection()`: mines GitHub/org data and infers common patterns with regex. +- `verifier.py`: MX lookup and SMTP RCPT checks with rate limiting. +- `storage.py`: asyncpg pool and upsert logic with unique constraint on (company, email). + +## Important Modules / Files + +- `main.py` — FastAPI app and background task wiring. +- `discovery.py` — multi-stage discovery pipeline. +- `verifier.py` — SMTP verification utilities. +- `storage.py` — DB pool and CRUD/upsert helpers. + +## Service Interactions + +- Reads job records from aggregator-service to link contacts to jobs. +- Persists contacts to PostgreSQL for consumption by email-generator-service. +- Calls external services: Clearbit, GitHub API, LinkedIn scraping, and MX hosts for verification. + +## Debugging Notes + +- Enable a `GITHUB_TOKEN` to avoid rate limits during name mining; failures are logged but pipeline continues where possible. +- SMTP verification can be noisy; use logging in `verifier.py` to record MX/RCPT outcomes and rate-limiting events. +- Database upsert conflicts are handled by `ON CONFLICT`; add targeted logging when debugging merges. + +## Contributor Tips + +- Write unit tests for pattern inference when modifying regex or GitHub parsing logic. +- Mock SMTP and DNS responses for deterministic verification tests. + +## Next (Operational) Docs + +Operational/runbook content (third-party API keys, env vars, deployment) is deferred to a follow-up docs issue. diff --git a/contact-discovery-service/README.md b/contact-discovery-service/README.md index 30e3ff1..f905c02 100644 --- a/contact-discovery-service/README.md +++ b/contact-discovery-service/README.md @@ -1,42 +1,9 @@ # Contact Discovery Service -## Request/Data Flow +This repository contains the Contact Discovery service. Contributor-oriented architecture and developer notes are moved to a dedicated file to reduce duplication with other services. -1. `POST /discover` accepts company name, optional job_id, roles list, and domain, triggering background discovery pipeline. -2. Background task calls `discovery.run_pipeline()` to execute stages: domain inference, pattern detection, name discovery, email construction, and SMTP verification. -3. Verified contacts are upserted into PostgreSQL via `storage.upsert_contact()` with conflict resolution on (company, email). -4. `GET /contacts?company={company}` queries PostgreSQL for contacts by company substring match. -5. `GET /contacts/{job_id}` retrieves contacts linked to a specific job UUID. -6. `DELETE /contacts/{id}` removes a contact record from PostgreSQL. +See the contributor-oriented material in: [contact-discovery-service/ARCHITECTURE.md](contact-discovery-service/ARCHITECTURE.md) -## Internal Execution Pipeline +Operational/API usage and setup details are deferred to a follow-up docs issue to avoid duplication across services. -- **Domain Inference**: `domain_inference()` queries Clearbit autocomplete API, falls back to direct HTTPS probes of common suffixes (com, io, co.in) to determine company domain. -- **Email Pattern Detection**: `email_pattern_detection()` searches GitHub commits and org members for domain emails, infers patterns (e.g., {first}.{last}@{domain}) using regex matching, defaults to {first}.{last} if undetected. -- **Name Discovery**: `_github_org_names()` fetches GitHub org members' public profiles for names/roles, supplemented by LinkedIn search parsing for additional contacts. -- **Email Construction**: Combines discovered names with inferred pattern to generate email addresses. -- **SMTP Verification**: `verifier.verify_email()` resolves MX records, performs SMTP RCPT TO checks with per-domain rate limiting (5/hour), returning 'verified', 'unverified', or 'invalid'. -- **Storage Upsert**: `storage.upsert_contact()` uses `INSERT ... ON CONFLICT (company, email) DO UPDATE` to merge new data with existing records. - -## Important Modules/Files - -- `main.py`: FastAPI application with endpoints (`/discover`, `/contacts`, `/contacts/{job_id}`, `/contacts/{id}`), background task management, and Pydantic models for requests/responses. -- `discovery.py`: Multi-stage pipeline using httpx for HTTP requests, GitHub API for name/pattern mining, and regex for email pattern inference. -- `storage.py`: asyncpg pool lifecycle, schema DDL for contacts table with indexes, and CRUD functions with unique constraint on (company, email). -- `verifier.py`: SMTP verification with dnspython MX resolution, rate limiting via in-memory dict, and synchronous SMTP checks wrapped in asyncio executor. - -## Service Interactions - -- Reads jobs table from PostgreSQL to link contacts to job postings. -- Writes contacts records to PostgreSQL, shared with email-generator-service for recipient lookup. -- Makes external HTTP requests to Clearbit, GitHub API, and LinkedIn for public data mining. -- Performs SMTP connections to domain MX servers for email verification. - -## Debugging Notes - -- Domain inference logs Clearbit successes/failures as info/debug, with warnings for unresolved domains. -- GitHub API errors logged as debug, with rate limit handling via optional GITHUB_TOKEN. -- SMTP verification logs results per email, with rate limit warnings and connection errors as debug. -- Database upsert conflicts logged implicitly via ON CONFLICT, with unique constraint added dynamically. -- Pipeline stage failures in `discovery.py` may skip to next stage, logged with exceptions. -- Asyncpg timeouts default to 30 seconds, potentially affecting contact listing under load. +If you need quick operational references, inspect `discovery.py`, `verifier.py`, and `storage.py`. diff --git a/crawler-service/ARCHITECTURE.md b/crawler-service/ARCHITECTURE.md new file mode 100644 index 0000000..999205b --- /dev/null +++ b/crawler-service/ARCHITECTURE.md @@ -0,0 +1,48 @@ +# Crawler — Architecture & Contributor Notes + +This file focuses on contributor-oriented/internal architecture information: request/data flow, internal execution pipeline, important modules/files, debugging notes, and service interactions. + +## Overview + +The Crawler service runs Scrapy spiders to fetch job listings from public job boards, filters and normalizes items, and emits events to the Redis `jobs:raw` stream for downstream consumption. + +## Request / Data Flow + +1. Spiders (remotive, wellfound, yc_spider) crawl external job boards and yield job items. +2. Items go through pipelines: deduplication, stack/role filtering, normalization, and emission. +3. `emit_pipeline.py` serializes items and adds them to the Redis stream `jobs:raw` with maxlen capping. + +## Internal Execution Pipeline + +- `crawler/spiders/*`: spider implementations and parsing logic with `stack_matches()` and `role_matches()` helpers. +- `crawler/pipelines/dedup_pipeline.py`: URL-based deduplication using Redis keys. +- `crawler/pipelines/filter_pipeline.py`: stack/role filtering logic to drop irrelevant items. +- `crawler/pipelines/emit_pipeline.py`: JSON serialization and Redis stream emission. +- `read_stream.py`: utility for tailing or dumping the Redis stream for debugging. + +## Important Modules / Files + +- `crawler/spiders/*` — spider implementations (remotive, wellfound, yc_spider). +- `crawler/pipelines/*` — deduplication, filtering, and emission pipelines. +- `read_stream.py` — CLI stream monitor. +- `scrapy.cfg` and `crawler/settings.py` — Scrapy project configuration and pipeline ordering. + +## Service Interactions + +- Emits events to Redis `jobs:raw`, consumed by aggregator-service. +- Uses Playwright for JavaScript-heavy sites and respects politeness settings in `crawler/settings.py`. + +## Debugging Notes + +- Update selectors when site markup changes; add unit tests for parsing when possible. +- Use `read_stream.py` to inspect emitted events and verify pipeline behavior. +- Playwright and Scrapy timeouts are common failure points; tune delays and concurrency in `crawler/settings.py`. + +## Contributor Tips + +- Run spiders locally with `scrapy crawl ` for quick parsing checks. +- Add integration tests for critical parsing logic in `crawler/parsers/`. + +## Next (Operational) Docs + +Operational/runbook content (docker-compose, Playwright setup, quotas) is deferred to a follow-up docs issue. diff --git a/crawler-service/README.md b/crawler-service/README.md index 0ccca6b..afb1fbe 100644 --- a/crawler-service/README.md +++ b/crawler-service/README.md @@ -1,38 +1,9 @@ # Crawler Service -## Request/Data Flow +This repository contains the crawler (Scrapy) service. Contributor-oriented architecture and developer notes are moved to a dedicated file to reduce duplication with other services. -1. Scrapy spiders (e.g., remotive_spider, wellfound_spider) crawl job boards, yielding job items with company, role, stack, etc. -2. Items pass through pipelines: `DeduplicationPipeline` checks for duplicates, `StackFilterPipeline` filters by target stack/role, `RedisStreamPipeline` emits to 'jobs:raw' Redis stream. -3. `read_stream.py` monitors the Redis stream, displaying new job events in real-time or dumping historical entries. +See the contributor-oriented material in: [crawler-service/ARCHITECTURE.md](crawler-service/ARCHITECTURE.md) -## Internal Execution Pipeline +Operational/API usage and setup details are deferred to a follow-up docs issue to avoid duplication across services. -- **Spider Parsing**: Base spider classes use `stack_matches()` and `role_matches()` to filter jobs based on JOBSEEKER_STACK and JOBSEEKER_ROLE settings. -- **Deduplication Pipeline**: `dedup_pipeline.py` uses Redis keys for URL-based deduplication, skipping already-seen jobs. -- **Stack Filter Pipeline**: `filter_pipeline.py` applies stack intersection checks, dropping items that don't match target technologies. -- **Redis Emit Pipeline**: `emit_pipeline.py` serializes items to JSON strings, adds to 'jobs:raw' stream with maxlen 10,000, logging emissions. -- **Stream Monitoring**: `read_stream.py` uses Redis XREAD/XREVRANGE for tailing or dumping stream contents, formatting job details. - -## Important Modules/Files - -- `read_stream.py`: CLI script for Redis stream monitoring with options for tailing, dumping, and count limits. -- `scrapy.cfg`: Scrapy project configuration pointing to crawler.settings. -- `crawler/settings.py`: Scrapy settings with politeness delays, Playwright middleware, pipeline order, Redis connection, and jobseeker profile. -- `crawler/spiders/base_spider.py`: Abstract base spider with stack/role matching logic and crawler instantiation. -- `crawler/pipelines/emit_pipeline.py`: Final pipeline stage emitting job items to Redis stream with JSON serialization and maxlen capping. - -## Service Interactions - -- Crawls external job board websites (Remotive, Wellfound, YC, Naukri, LinkedIn) using Scrapy with Playwright for JavaScript rendering. -- Writes job events to Redis 'jobs:raw' stream, consumed by aggregator-service for database insertion. -- Uses Redis for deduplication keys and stream storage. - -## Debugging Notes - -- Spider logs include parse successes/failures, with debug level for filtered items. -- Pipeline logs emission counts and skipped duplicates, with warnings for Redis connection issues. -- Playwright timeouts logged as errors, with AUTOTHROTTLE adjusting delays for rate limiting. -- Stream monitoring logs connection errors if Redis is unreachable. -- Item serialization errors in emit_pipeline logged as exceptions, potentially dropping malformed items. -- Concurrent requests limited to 4 globally, 1 per domain, to avoid bans. +If you need quick operational references, inspect `crawler/` package, `scrapy.cfg`, and `read_stream.py`. diff --git a/email-generator-service/ARCHITECTURE.md b/email-generator-service/ARCHITECTURE.md new file mode 100644 index 0000000..742efb1 --- /dev/null +++ b/email-generator-service/ARCHITECTURE.md @@ -0,0 +1,63 @@ +# Email Generator — Architecture & Contributor Notes + +This file is focused on contributor-oriented/internal architecture information: request/data flow, service interactions, important modules/files, debugging notes, and the internal execution pipeline. + +## Overview + +The Email Generator service creates draft outreach emails from job/contact records and template prompts, optionally enriching content via a local Ollama model. It stores drafts in PostgreSQL and can send messages via SMTP. + +## Request / Data Flow + +1. External caller (gateway or CLI) triggers generation by calling the service endpoint or invoking the internal API surface. +2. The service loads context data from PostgreSQL (`jobs`, `contacts`) when `job_id` or `contact_id` are provided. +3. `generator.generate_email()` builds a rendering context (company, role, stack, candidate details, your_name, etc.). +4. The service requests a short `product_observation` from Ollama via `ollama_client.py`. On failure it falls back to static observations from `fallbacks.yaml`. +5. Jinja2 templates in `templates/` render subject and body using the assembled context. +6. The rendered draft is persisted to the `emails` table via `storage.insert_email()` and an email UUID is returned. +7. Sending (`/emails/{id}/send`) resolves recipient from the contact record and hands the message to `mailer.send_email()` to perform SMTP delivery and update status fields. + +## Internal Execution Pipeline + +- `main.py`: FastAPI app; registers endpoints and manages lifespan events (DB pool lifecycle). +- `generator.py`: Central orchestration — loads DB rows, calls Ollama client, applies fallback selection, renders templates, and calls `storage` to persist drafts. +- `ollama_client.py`: Async client that probes available models and posts prompts to the local Ollama API with timeouts and retries. +- `storage.py`: Manages `asyncpg` pool, schema initialization, and CRUD helpers for `jobs`, `contacts`, and `emails`. +- `mailer.py`: Synchronous SMTP sender wrapped via `asyncio.run_in_executor` for non-blocking behavior. + +Execution details: +- Generation attempts Ollama first; fallback is deterministic based on keyword scoring against entries in `fallbacks.yaml`. +- Template rendering uses a small, trusted set of variables — avoid adding unvalidated user input into template context. + +## Important Modules / Files + +- `main.py` — app entry; endpoints: `/generate`, `/emails`, `/emails/{id}`, `/emails/{id}/send`. +- `generator.py` — build context, call Ollama, select fallback, render templates. +- `ollama_client.py` — model detection and single-observation generation. +- `storage.py` — asyncpg pool and CRUD functions. +- `mailer.py` — SMTP send logic. +- `fallbacks.yaml` — curated static observations keyed by category. +- `templates/` — Jinja2 templates used for rendering. + +## Service Interactions + +- Reads `jobs` and `contacts` from the shared PostgreSQL instance. +- Writes drafts and status updates to the `emails` table. +- Calls a local Ollama server (`OLLAMA_BASE_URL`) for short product observations. +- Sends outbound mail via external SMTP (configured via environment variables). + +## Debugging Notes + +- Ollama timeouts and HTTP errors are logged in `ollama_client.py`; generation continues using fallbacks. +- If the DB pool is not initialized, `storage` raises `RuntimeError` during lifespan — check logs at startup. +- SMTP authentication failures raise `smtplib.SMTPAuthenticationError` — verify credentials and App Password. +- Template rendering errors surface as `jinja2` exceptions; test templates locally with a small context for quicker iteration. + +## Contributor Tips + +- Local testing: run unit tests in `tests/unit` and integration tests in `tests/integration` as applicable. +- To debug generation flow: add temporary logging in `generator.generate_email()` around Ollama calls and fallback selection. +- Keep `fallbacks.yaml` categories small and focused; add a unit test when adding new categories to ensure predictable selection. + +## Next (Operational) Docs + +Operational/API usage (endpoint examples, env var setup, deployment notes) are intentionally deferred to a follow-up docs issue to avoid duplication across services. If you need an operational snapshot now, inspect `main.py` and the Pydantic models used for requests/responses. diff --git a/email-generator-service/README.md b/email-generator-service/README.md index 9999256..155b29a 100644 --- a/email-generator-service/README.md +++ b/email-generator-service/README.md @@ -1,45 +1,9 @@ # Email Generator Service -## Request/Data Flow +This repository contains the Email Generator service. Contributor-focused architecture and developer notes are intentionally moved to a dedicated file to reduce duplication with other services. -1. `POST /generate` accepts a JSON payload with `template`, optional `job_id`/`contact_id`, candidate details, and context fields. -2. `main.py` initializes the asyncpg pool and fetches job/contact records from PostgreSQL using `storage.py` if IDs are provided. -3. `generator.py` constructs the email context, invokes `ollama_client.py` for product observation generation, and renders Jinja2 templates. -4. Rendered subject/body are inserted into the `emails` table via `storage.insert_email()`, returning the email UUID. -5. `GET /emails` and `GET /emails/{id}` retrieve drafts from PostgreSQL using `storage.py` queries. -6. `POST /emails/{id}/send` resolves the recipient email from the linked contact record, sends via `mailer.py`, and updates `sent_at`/`status` in the database. +See the contributor-oriented material in: [email-generator-service/ARCHITECTURE.md](email-generator-service/ARCHITECTURE.md) -## Internal Execution Pipeline +Operational/API usage and setup details are intentionally deferred to a follow-up docs issue (to avoid duplication across services). -- **Generation Pipeline**: `generator.generate_email()` loads job/contact data, attempts Ollama observation via `ollama_client.generate_observation()`, falls back to YAML-based selection from `fallbacks.yaml`, and renders templates using Jinja2 with context variables. -- **Ollama Integration**: `ollama_client.py` detects available models (mistral/llama3), sends a prompt to `/api/generate`, and returns a single observation sentence or None on failure. -- **Fallback Logic**: `generator._select_fallback()` matches keywords from product/stack against `fallbacks.yaml` categories, selecting a random observation from the best-matched bucket. -- **Template Rendering**: Jinja2 environment in `generator.py` processes templates with variables like `company`, `role`, `your_name`, `product_observation`, and `contact_name`. -- **Storage Operations**: `storage.py` manages asyncpg pool, executes DDL for `jobs`/`contacts`/`emails` tables, and performs CRUD with parameterized queries. -- **Email Sending**: `mailer.send_email()` uses `smtplib.SMTP_SSL` with Gmail credentials, sending plain-text emails and handling authentication errors. - -## Important Modules/Files - -- `main.py`: FastAPI application with endpoints (`/generate`, `/emails`, `/emails/{id}`, `/emails/{id}/send`), Pydantic models for requests/responses, and lifespan hooks for pool management. -- `generator.py`: Core logic for email creation, including Ollama client calls, fallback selection, and Jinja2 rendering with context assembly. -- `ollama_client.py`: Async HTTP client for Ollama API, model detection via `/api/tags`, and observation generation with timeout handling. -- `mailer.py`: Synchronous Gmail SMTP sender wrapped in `asyncio.run_in_executor`, using SSL on port 465 with App Password authentication. -- `storage.py`: asyncpg pool lifecycle, schema initialization (including `pgcrypto` extension), and CRUD functions for jobs, contacts, and emails tables. -- `fallbacks.yaml`: YAML structure with keyword lists per category (e.g., fintech, devtools) and observation arrays for static fallbacks. -- `templates/`: Directory containing Jinja2 templates (`cold_outreach.j2`, `recruiter_outreach.j2`, `followup.j2`) with conditional rendering and variable substitution. - -## Service Interactions - -- Reads `jobs` and `contacts` tables from PostgreSQL to populate email templates and resolve recipient addresses. -- Writes to the `emails` table for draft storage and status updates, sharing the database with `aggregator-service` and `contact-discovery-service`. -- Makes HTTP requests to a local Ollama server at `OLLAMA_BASE_URL` for product observations, with automatic fallback to static data. -- Sends outbound emails via Gmail SMTP, requiring external Gmail account configuration for delivery. - -## Debugging Notes - -- Ollama failures log warnings in `ollama_client.py` for timeouts or HTTP errors, but do not halt generation due to fallback mechanism. -- Database connection issues in `storage.py` raise `RuntimeError` if pool is not initialized, logged during lifespan startup. -- SMTP authentication errors in `mailer.py` propagate as `smtplib.SMTPAuthenticationError`, indicating invalid `GMAIL_APP_PASSWORD`. -- Template rendering errors in `generator.py` raise `ValueError` for unknown templates, logged with template names. -- Asyncpg query timeouts default to 30 seconds in `storage.py`, potentially causing delays in high-load scenarios. -- Fallback selection in `generator.py` logs category scores and selected observations for keyword matching verification. +If you need quick operational references while the follow-up doc is created, the implementation files are in the repository and can be inspected directly. diff --git a/gateway/ARCHITECTURE.md b/gateway/ARCHITECTURE.md new file mode 100644 index 0000000..ed46129 --- /dev/null +++ b/gateway/ARCHITECTURE.md @@ -0,0 +1,42 @@ +# Gateway — Architecture & Contributor Notes + +This file focuses on contributor-oriented/internal architecture information: request/data flow, internal execution pipeline, important modules/files, debugging notes, and service interactions. + +## Overview + +The Gateway service proxies API requests to backend services, aggregates health/status, and orchestrates composite workflows that involve multiple services. + +## Request / Data Flow + +1. Client requests hit proxy endpoints (e.g., `/api/jobs`, `/api/scrape`, `/api/contacts`, `/api/emails`) which forward requests to the appropriate backend service. +2. Health checks fan out to multiple services concurrently and return aggregated status. +3. Composite workflow (`/api/workflow/apply`) sequences downstream calls: fetch job, trigger discovery, fetch contacts, generate email draft. + +## Internal Execution Pipeline + +- `proxy.py`: httpx-based forwarding helper that preserves headers and body while stripping hop-by-hop headers. +- `main.py`: endpoints for proxying, health fan-out (asyncio.gather), workflow orchestration, and dashboard serving. + +## Important Modules / Files + +- `main.py` — FastAPI entry and workflow orchestration. +- `proxy.py` — generic `proxy_request()` and typed helpers for downstream services. +- `dashboard.html` — dashboard SPA that consumes `/api/*` endpoints. + +## Service Interactions + +- Proxies to aggregator-service, scraper-service, contact-discovery-service, and email-generator-service. +- Reads scheduler run summary from shared volume for `/api/summary`. + +## Debugging Notes + +- Proxy timeouts surface as 503/504 with upstream URL in logs. +- Workflow orchestration logs per-step exceptions; increase logging for problematic steps when reproducing bugs. + +## Contributor Tips + +- Use the gateway to run end-to-end workflows locally by wiring services to local ports and invoking `/api/workflow/apply`. + +## Next (Operational) Docs + +Operational/runbook content (CORS, TLS, deployment) is deferred to a follow-up docs issue. diff --git a/gateway/README.md b/gateway/README.md index 39d5d83..ada531d 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -1,37 +1,9 @@ # Gateway Service -## Request/Data Flow +This repository contains the Gateway service. Contributor-oriented architecture and developer notes are moved to a dedicated file to reduce duplication with other services. -1. Proxy routes (/api/jobs/*, /api/scrape, /api/contacts/*, /api/emails/*) forward requests to aggregator, scraper, contact-discovery, and email-generator services using httpx. -2. `GET /api/health` fans out health checks to all four services concurrently, returning aggregated status. -3. `GET /api/summary` reads scheduler run summary from shared volume at /data/run_summary.json. -4. `POST /api/workflow/apply` orchestrates composite workflow: fetches job, triggers contact discovery, retrieves contacts, generates email draft. -5. `GET /` serves dashboard.html as a single-page application. +See the contributor-oriented material in: [gateway/ARCHITECTURE.md](gateway/ARCHITECTURE.md) -## Internal Execution Pipeline +Operational/API usage and setup details are deferred to a follow-up docs issue to avoid duplication across services. -- **Proxying**: `proxy.proxy_request()` forwards method, headers (excluding hop-by-hop), body, and query params to upstream, returning Response with preserved status/content-type. -- **Health Fan-out**: `main.gateway_health()` uses `asyncio.gather()` for parallel health checks, aggregating results into JSON response. -- **Composite Workflow**: `main.workflow_apply()` calls `proxy.get_job()`, `proxy.trigger_discovery()`, `proxy.get_contacts_for_company()`, `proxy.generate_email()` sequentially, handling exceptions. -- **Dashboard Serving**: `main.dashboard()` returns FileResponse for dashboard.html, which uses JavaScript to call /api/* endpoints. - -## Important Modules/Files - -- `main.py`: FastAPI application with proxy routes, composite workflow endpoint, health fan-out, and dashboard serving. -- `proxy.py`: httpx client management, generic proxy_request() helper, typed service helpers for workflow orchestration. -- `dashboard.html`: Single-page dashboard SPA with tabs for jobs, contacts, emails, stats, using Chart.js and fetch API for /api/* calls. - -## Service Interactions - -- Proxies requests to aggregator-service (jobs/stats), scraper-service (scrape), contact-discovery-service (contacts/discover), email-generator-service (emails/generate). -- Reads scheduler run summary from shared Docker volume. -- Serves dashboard that consumes all /api/* endpoints. - -## Debugging Notes - -- Proxy timeouts logged as 503/504 HTTPExceptions, with upstream URL in detail. -- Health check failures logged in aggregated response, with unreachable services marked. -- Workflow orchestration logs exceptions per step, raising HTTPException on failures. -- Dashboard JavaScript errors not logged server-side, debug via browser console. -- Shared volume read errors for summary logged as 500 responses. -- Concurrent health checks may have varying response times, aggregated in single JSON. +If you need quick operational references, inspect `main.py` and `proxy.py`. diff --git a/scheduler/ARCHITECTURE.md b/scheduler/ARCHITECTURE.md new file mode 100644 index 0000000..d6ca3f9 --- /dev/null +++ b/scheduler/ARCHITECTURE.md @@ -0,0 +1,47 @@ +# Scheduler — Architecture & Contributor Notes + +This file focuses on contributor-oriented/internal architecture information: request/data flow, internal execution pipeline, important modules/files, debugging notes, and service interactions. + +## Overview + +The Scheduler service runs periodic cycles (scrape, discover, draft) using APScheduler to orchestrate scraping, contact discovery, and draft generation pipelines. + +## Request / Data Flow + +1. `run_scrape_cycle()` triggers scrapers via the Gateway and runs Scrapy subprocesses, then measures job deltas. +2. `run_discover_cycle()` fetches new jobs and calls contact-discovery for each job with appropriate delays. +3. `run_draft_cycle()` fetches jobs, verifies contacts, and calls email-generator to draft messages. +4. After each run, a summary JSON is written to `/data/run_summary.json` for observability. + +## Internal Execution Pipeline + +- `main.py`: APScheduler setup, interval jobs, manual run dispatch, and summary writing. +- `tasks.py`: synchronous task implementations that call gateway endpoints and manage subprocesses for scraping. +- `logger.py`: structured JSON logging for tasks and error context. + +## Important Modules / Files + +- `main.py` — scheduler lifecycle and job registration. +- `tasks.py` — scrape/discover/draft task implementations. +- `logger.py` — JSONFormatter and structured logging utilities. + +## Service Interactions + +- Calls gateway endpoints to trigger scraping, discovery, and generation. +- Runs Scrapy subprocesses for crawler-service when applicable. +- Writes run summaries to shared volume consumed by gateway. + +## Debugging Notes + +- Inspect `/data/run_summary.json` to see per-run stats and recorded errors. +- Subprocess stderr and HTTP errors are captured in the summary for post-mortem analysis. +- Use manual runs for isolated testing of individual tasks. + +## Contributor Tips + +- Adjust intervals and delays in env vars when testing locally to avoid rate limits. +- Use manual dispatch to reproduce issues without waiting for scheduled runs. + +## Next (Operational) Docs + +Operational/runbook content (cron tuning, deployment) is deferred to a follow-up docs issue. diff --git a/scheduler/README.md b/scheduler/README.md index 1d2a88c..28e8db0 100644 --- a/scheduler/README.md +++ b/scheduler/README.md @@ -1,37 +1,9 @@ # Scheduler Service -## Request/Data Flow +This repository contains the Scheduler service. Contributor-oriented architecture and developer notes are moved to a dedicated file to reduce duplication with other services. -1. APScheduler runs `run_scrape_cycle()` every CRAWL_INTERVAL_HOURS: POST /api/scrape to gateway, runs Scrapy subprocess for remotive/yc_jobs spiders, waits for pipelines, counts job delta. -2. APScheduler runs `run_discover_cycle()` every DISCOVER_INTERVAL_HOURS: fetches new jobs via GET /api/jobs, POST /api/discover for each job with delay between calls. -3. APScheduler runs `run_draft_cycle()` every DISCOVER_INTERVAL_HOURS: fetches new jobs, checks for contacts via GET /api/contacts, POST /api/generate for jobs with contacts. -4. After each task, summary is written to /data/run_summary.json with jobs_discovered, contacts_found, emails_drafted, errors. +See the contributor-oriented material in: [scheduler/ARCHITECTURE.md](scheduler/ARCHITECTURE.md) -## Internal Execution Pipeline +Operational/API usage and setup details are deferred to a follow-up docs issue to avoid duplication across services. -- **Scrape Cycle**: `tasks.run_scrape_cycle()` counts jobs before, triggers platform scrapers via httpx POST, runs Scrapy subprocess with timeout, waits SCRAPER_WAIT_SECS, counts delta and updates summary. -- **Discover Cycle**: `tasks.run_discover_cycle()` fetches new jobs via httpx GET, loops over jobs, POST /api/discover with DISCOVER_DELAY_SECS pause, increments contacts_found on success. -- **Draft Cycle**: `tasks.run_draft_cycle()` fetches new jobs, checks contacts via GET /api/contacts, generates drafts via POST /api/generate for jobs with contacts, updates emails_drafted. -- **Summary Flushing**: `main._write_summary()` serializes summary dict to JSON file, logged on success/failure. -- **Manual Runs**: `main._maybe_manual_run()` dispatches single tasks or all via MANUAL_TASK env var, exits after completion. - -## Important Modules/Files - -- `main.py`: APScheduler setup with interval jobs for scrape/discover/draft, graceful shutdown handling, summary writing, manual task dispatch. -- `tasks.py`: Synchronous task functions with httpx calls to gateway, subprocess for Scrapy, summary state management, error recording. -- `logger.py`: JSONFormatter for structured logging to stdout, StructLogger adapter for extra fields. - -## Service Interactions - -- Calls gateway /api/scrape, /api/jobs, /api/discover, /api/contacts, /api/generate via httpx. -- Runs Scrapy subprocess for crawler-service spiders. -- Writes run summary to shared Docker volume /data/run_summary.json, read by gateway. - -## Debugging Notes - -- Task exceptions logged as JSON with context/detail, appended to summary errors. -- HTTP timeouts/failures logged in tasks, with _record_error() updating summary. -- Subprocess timeouts/errors logged, with stderr captured in summary. -- Scheduler misfires logged via APScheduler, with grace time 300s. -- Manual runs exit after task, useful for testing without schedule. -- Concurrent tasks prevented by _task_lock, with warnings for skipped runs. +If you need quick operational references, inspect `tasks.py`, `main.py`, and `logger.py`. diff --git a/scraper-service/ARCHITECTURE.md b/scraper-service/ARCHITECTURE.md new file mode 100644 index 0000000..acc96a6 --- /dev/null +++ b/scraper-service/ARCHITECTURE.md @@ -0,0 +1,44 @@ +# Scraper — Architecture & Contributor Notes + +This file focuses on contributor-oriented/internal architecture information: request/data flow, internal execution pipeline, important modules/files, debugging notes, and service interactions. + +## Overview + +The Scraper service runs platform-specific scrapers (Naukri, LinkedIn, Internshala), normalizes job listings, and emits events to the Redis `jobs:raw` stream for downstream processing. + +## Request / Data Flow + +1. `POST /scrape` triggers background scraping for the requested role/stack. +2. Each platform scraper yields job dictionaries that are normalized and validated. +3. `emit.emit_job()` serializes and pushes jobs to Redis `jobs:raw` with maxlen capping. + +## Internal Execution Pipeline + +- `main._run_all_scrapers()` runs scrapers concurrently with `asyncio.gather()` and collects results. +- `scrapers/*` implement the platform-specific scraping and parsing logic. +- `emit.py` handles Redis emission and normalization. + +## Important Modules / Files + +- `main.py` — FastAPI endpoint and background task orchestration. +- `emit.py` — Redis emitter and normalization. +- `scrapers/` — platform-specific scrapers and `base.py` contract. + +## Service Interactions + +- Emits job events to Redis `jobs:raw` for aggregator-service consumption. +- Uses Playwright for JS-heavy pages and honors politeness and concurrency settings. + +## Debugging Notes + +- Monitor Playwright selector changes and update parsing logic accordingly. +- Use logs to inspect per-platform job counts and emission errors. + +## Contributor Tips + +- Run individual scrapers locally for focused debugging and selector iteration. +- Add unit tests for parsing when updating selectors. + +## Next (Operational) Docs + +Operational/runbook content (Playwright setup, credentials, Docker compose) is deferred to a follow-up docs issue. diff --git a/scraper-service/README.md b/scraper-service/README.md index 3ed85e1..01c8b19 100644 --- a/scraper-service/README.md +++ b/scraper-service/README.md @@ -1,37 +1,9 @@ # Scraper Service -## Request/Data Flow +This repository contains the Scraper service. Contributor-oriented architecture and developer notes are moved to a dedicated file to reduce duplication with other services. -1. `POST /scrape` accepts role and stack parameters, triggering background concurrent scraping of Naukri, LinkedIn, and Internshala. -2. Each scraper (NaukriScraper, LinkedInScraper, IntershalaScraper) runs asynchronously, returning job dicts with company, role, url, etc. -3. `emit.emit_jobs()` pushes each job dict to the 'jobs:raw' Redis stream, serializing stack to JSON and capping stream length at 50,000. -4. Background task completes, logging emission counts per platform. +See the contributor-oriented material in: [scraper-service/ARCHITECTURE.md](scraper-service/ARCHITECTURE.md) -## Internal Execution Pipeline +Operational/API usage and setup details are deferred to a follow-up docs issue to avoid duplication across services. -- **Scraper Execution**: `main._run_all_scrapers()` uses `asyncio.gather()` to run all platform scrapers in parallel, capturing exceptions per scraper. -- **LinkedIn Scraping**: `linkedin.LinkedInScraper.scrape()` launches Playwright browser, navigates to search URL, scrolls to load cards, parses selectors for title, company, location, URL, and posted date. -- **Emission Process**: `emit.emit_job()` normalizes fields, checks for required company/role, adds to Redis stream with maxlen and approximate trimming. -- **Error Handling**: Scraper exceptions logged as errors, emission failures logged with exceptions, skipping malformed jobs. - -## Important Modules/Files - -- `main.py`: FastAPI application with /scrape endpoint, background task management, and concurrent scraper execution. -- `emit.py`: Async Redis emitter with connection pooling, job normalization, and stream addition with maxlen capping. -- `scrapers/base.py`: Abstract base class defining scraper contract with source_name and scrape() method. -- `scrapers/linkedin.py`: Playwright-based LinkedIn scraper with selector parsing, scroll delays, and auth-wall detection. - -## Service Interactions - -- Scrapes job listings from Naukri, LinkedIn public search, and Internshala using HTTP requests and Playwright for JavaScript sites. -- Emits job events to Redis 'jobs:raw' stream, consumed by aggregator-service for database storage. -- Uses Redis for stream storage and emission. - -## Debugging Notes - -- Scraper logs include job counts per platform, with errors for exceptions during scraping. -- Playwright timeouts and selector failures logged as warnings/debug, with auth redirects indicating rate limits. -- Emission logs debug-level for each job, with exceptions for Redis connection issues. -- Concurrent scraper runs may have varying completion times, logged per platform. -- Stream maxlen may drop old events if emission rate exceeds consumption. -- Selector updates required when LinkedIn changes markup, logged as parsing errors. +If you need quick operational references, inspect `main.py`, `emit.py`, and the `scrapers/` package.