utopia-ai
/
HR-ATS-Portal
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

LINK_X_USER_INBOX #6

Merged
ahmed.mujtaba merged 4 commits from LINK_X_USER_INBOX into main 2026-08-10 12:24:46 +00:00
Conversation 0 Commits 4 Files Changed 16
16 changed files with 976 additions and 190 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

676
backend/README.md Normal file
View File

@ -0,0 +1,676 @@
# HR-ATS-Portal — Backend
FastAPI service behind **TalentFlow**, an HR / Applicant-Tracking portal. It pulls candidate
applications out of a mailbox, decodes CV attachments, runs an LLM agent to match each CV
against live job posts, publishes job ads to social channels through Buffer, and serves the
whole thing to the frontend behind JWT auth and a database-driven RBAC model.
Everything in this document describes `backend/` only.
---
## Table of contents
- [Architecture](#architecture)
- [Tech stack](#tech-stack)
- [Directory layout](#directory-layout)
- [House style — what lives in which file](#house-style--what-lives-in-which-file)
- [Domains](#domains)
- [Data model](#data-model)
- [API reference](#api-reference)
- [Authentication and RBAC](#authentication-and-rbac)
- [Background jobs](#background-jobs)
- [The matching agent](#the-matching-agent)
- [External integrations](#external-integrations)
- [Configuration](#configuration)
- [Running locally](#running-locally)
- [Database migrations](#database-migrations)
- [Docker](#docker)
- [Response conventions](#response-conventions)
- [Known gaps and gotchas](#known-gaps-and-gotchas)
---
## Architecture
```mermaid
flowchart TB
FE["Frontend (Vite, :5173)"] -->|JWT Bearer| API
subgraph API["FastAPI — main.py :8000"]
R1["users / role / forget_password / notifications"]
R2["inbox"]
R3["job (job_post + candidate)"]
end
API --> PG[("PostgreSQL — schema app")]
API -->|enqueue| REDIS[("Redis Streams")]
REDIS --> W["Taskiq worker\ninbox.tasks + inbox.sync_tasks"]
SCHED["Taskiq scheduler\ncron"] --> REDIS
W --> PG
W --> AGENT["LangGraph agent\nagent/"]
AGENT --> OAI["OpenAI"]
API -->|GET /emails, /sync/read-status| MAILAPI["Email API (MS Graph proxy)"]
W --> MAILAPI
API -->|multipart send| TEAMS["Teams Mail API"]
API -->|GraphQL| BUF["Buffer"]
```
**The application flow, end to end:**
1. `GET /email/fetch` pulls messages from the external Email API, decodes PDF/DOC/DOCX
attachments to `inbox/decoded_attachments/`, and upserts them into `inbox_messages`.
2. When a message carries an attachment, the sender is linked to a `users` row — created with
the `candidate` role if new — through the `inbox` join table.
3. Any message with a stored attachment and no match result yet is enqueued onto Redis as an
`inbox.match_message` task.
4. The worker extracts the résumé text, hands it plus the active job posts to the LangGraph
agent, and writes `suggested_job_post_ids`, `match_summary`, `match_reasoning` and
`experience` back onto the row.
5. New candidate accounts land inactive and are mailed a confirmation link; the link is what
flips `is_active`.
6. A cron task sweeps Outlook read-status deltas back onto `inbox_messages.message_read`.
7. Recruiters read all of it through `/inbox/all-applications` and publish new roles with
`POST /job/post-job`, which renders the ad copy and pushes it to Buffer.
---
## Tech stack
| Concern | Choice |
|---|---|
| Web framework | FastAPI 0.136 + Uvicorn |
| ORM / models | SQLModel 0.0.38 on SQLAlchemy 2.0 (async, `asyncpg`) |
| Database | PostgreSQL, application objects in the `app` schema |
| Migrations | Alembic 1.18, driven by `alembic_setup.py` |
| Auth | PyJWT (HS256) access / refresh / reset tokens, `bcrypt` hashing |
| Task queue | Taskiq on Redis Streams, with a smart-retry + dead-letter middleware |
| LLM | OpenAI async client, orchestrated by LangGraph |
| PDF extraction | `pypdf` |
| HTTP client | `httpx` |
| Python | 3.12 (see `Dockerfile`) |
---
## Directory layout
```
backend/
├── main.py # FastAPI app, lifespan, CORS, router mounting
├── db_setup.py # Settings, async engine, sessions, init_db/lifespan
├── alembic_setup.py # Alembic scaffolding, autogenerate, migrate-on-boot
├── llm_setup.py # AsyncOpenAI client + llm_call helper
├── requirements.txt
├── Dockerfile # image for the Taskiq worker / scheduler
├── alembic.ini # generated by alembic_setup.py, not hand-written
├── migrations/ # generated env.py + versions/
├── LLM_CONTEXT_PROMPT.md # house-style prompt to paste into an LLM before editing
│
├── users/ # accounts, login, signup, RBAC enforcement
├── role/ # roles, permission bundles, permission tags
├── forget_password/ # reset-code request → verify → new password
├── notifications/ # email-confirmation tokens and mail
├── inbox/ # mailbox sync, attachments, applications
├── job/
│ ├── app.py # routes for both sub-domains
│ ├── job_post/ # job ads + Buffer publishing
│ └── candidate/ # CV reading, candidate profile
├── agent/ # LangGraph CV → job-post matching agent
└── taskiq_management/ # broker, scheduler, DLQ middleware, smoke task
```
There are **no `__init__.py` files**. The service is run from `backend/`, so imports are
top-level (`from users.app import router`, `from db_setup import get_session`).
---
## House style — what lives in which file
Every domain package follows the same six-file shape. This is enforced by convention, and
`LLM_CONTEXT_PROMPT.md` is the canonical statement of it.
| File | Owns | Must not do |
|---|---|---|
| `app.py` | Routes, inline request models, `JSONResponse`, dependency injection | Business rules, SQL |
| `views.py` | Business checks, calls models, raises `HTTPException` | Build login token envelopes |
| `models.py` | SQLModel table + `@classmethod async def` accessors | Import FastAPI, raise `HTTPException` |
| `serializers.py` | Hand-built `dict` builders (`serialize_*`) | Touch the DB or `Depends` |
| `plugins.py` | Pure helpers — hashing, JWT, HTTP calls to third parties | Import FastAPI |
| `permissions.py` | Bearer schemes and `Depends` aliases (auth domains only) | Hold route handlers |
Additional rules that matter when you edit this code:
- Request bodies are Pydantic models declared **inline in `app.py`**, never in `serializers.py`.
- There are **no Pydantic response models** — responses are hand-built dicts.
- Route paths are verb-in-path (`/users/create`, `/users/fetch`), not REST-resource-only, and
there is no `/api/v1` prefix.
- Non-DB config is module-level `load_dotenv()` + `os.getenv(...)`. Only database settings go
through `db_setup.Settings`.
---
## Domains
### `users/`
Signup, login, refresh, CRUD, role assignment, and the RBAC machinery every other domain
depends on. `users/permissions.py` defines the full `PermissionTag` vocabulary (13 modules ×
8 actions = 104 tags) and the `require_permission(...)` dependency. A startup assertion
(`_assert_vocabulary_complete`) fails loudly if the tag list ever drifts from
`PermissionModule × PermissionAction`.
Signup always assigns the `candidate` role, creates the account **inactive**, and sends a
confirmation email. Login rejects unconfirmed accounts.
### `role/`
Three-level permission model: `permission_tags` (atomic `module.action` rows) →
`permissions` (named bundles holding a JSONB array of tag ids) → `roles` (holding a JSONB
array of bundle ids). `Roles.resolve_tags()` walks that chain and returns a flat tuple of tag
names; dangling or inactive ids simply contribute nothing rather than erroring.
Eight system roles are seeded: `system_administrator`, `hr_administrator`, `recruiter`,
`hiring_manager`, `department_head`, `interviewer`, `ceo`, `candidate`.
### `inbox/`
The heart of the ingestion pipeline.
- `views.py::Email` talks to the external Email API, decodes attachments, upserts messages,
and enqueues matching work.
- `models.py` holds `Inbox_Messages` (the mail rows plus all agent output columns),
`Inbox_Alerts`, and `Inbox` — the join table linking a message to the candidate `Users` row
it came from. `_link_sender` creates the candidate account on first contact, skipping
`noreply@`-style senders.
- `file_decoder.py` turns Graph `contentBytes` into real PDF / DOCX / DOC files, validating
magic bytes for each format and stripping path traversal from filenames.
- `plugins.py` resolves attachment paths (handling Windows paths written by the host API but
read from a Linux worker), extracts résumé text, and calls the read-status sync endpoints.
- `tasks.py` / `sync_tasks.py` are the two Taskiq tasks.
### `job/`
Two sub-domains behind one router:
- **`job_post/`** — renders LinkedIn-shaped ad copy from a structured payload, resolves the
Buffer channel (by explicit id, by platform alias, or by the configured default), creates
the post over Buffer's GraphQL API, and records the mapped status. A queued post is recorded
as `scheduled`, not `published`; only Buffer reporting `sent` promotes it.
- **`candidate/`** — `FileRead` extracts text from an uploaded PDF (`pypdf`), and
`match_inbox_cv` force-requeues an existing inbox message for matching. `CandidateView`
reads the candidate profile through the `inbox` join.
### `notifications/` and `forget_password/`
Two parallel token flows, deliberately kept separate so each owns its own mail copy and env
reads:
- **Confirmation** — a 32-byte url-safe secret, bcrypt-hashed in
`email_confirmation_tokens`; the link carries `<row_id>.<secret>` because a bcrypt hash
cannot be looked up. Replays (mail scanners, back button) are handled idempotently.
- **Password reset** — a short code mailed to the user, bcrypt-hashed in
`password_reset_codes`, with a resend cooldown and a max-attempts cap. Verifying the code
mints a `type=reset` JWT carrying the code row id (`crid`), which is the only thing that
authorises the new-password call.
### `agent/`
LangGraph state machine — see [The matching agent](#the-matching-agent).
### `taskiq_management/`
Broker, scheduler, DLQ middleware, and a `ping` smoke task.
---
## Data model
All tables live in the `app` schema (`DB_DEFAULT_SCHEMA`), with a shared naming convention for
indexes, constraints and foreign keys. `SQLModel.metadata` is pointed at `Base.metadata` in
`db_setup.py` so SQLModel and DeclarativeBase share one registry and Alembic sees everything.
| Table | Key columns | Notes |
|---|---|---|
| `users` | `id` (uuid PK), `email` (unique), `role_id` → `roles.id`, `password`, `is_active`, `is_deleted` | Soft delete. `role` is `selectin`-loaded; lazy loads would raise `MissingGreenlet` under asyncio |
| `roles` | `id`, `role_name` (unique), `permissions` (JSONB int[]), `is_system` | |
| `permissions` | `id`, `name` (unique), `permission_tags` (JSONB int[]) | Named bundles |
| `permission_tags` | `id`, `tag_name` (unique), `module`, `action` | Unique on (`module`, `action`) |
| `inbox_messages` | `id` (uuid), `message_id` (upstream id, unique), `full_email_response` (JSONB), subject/body/from/to/cc/bcc, `message_read`, `attachment`, `file_name`, `file_path`, `application_status`, `resume_text`, `experience`, `suggested_job_post_ids` (JSONB), `match_summary`, `match_reasoning`, `match_status`, `match_error`, `matched_at` | One row per mail; agent output lands here |
| `inbox` | `id`, `user_id` → `users.id`, `message_id` → `inbox_messages.id`, `alert_id` | Join table linking a candidate to a message |
| `inbox_alerts` | `id`, `alert_sender_name`, `alert_sender_email`, `is_read` | |
| `job_posts` | `id` (uuid), `title`, `platform`, `channel_id`, `post_text`, `requirements`/`optional_skills` (JSON), `status`, `buffer_post_id`, `buffer_external_link`, `buffer_sent_at`, `buffer_error`, `created_by` → `users.id` | |
| `password_reset_codes` | `id`, `email`, `code_hash`, `expires_at`, `attempts`, `is_used`, `verified_at` | |
| `email_confirmation_tokens` | `id`, `user_id`, `email`, `token_hash`, `expires_at`, `is_used`, `confirmed_at` | |
`application_status` is a `str` enum: `PROCESS`, `PENDING`, `APPROVED`, `REJECTED`, `ONHOLD`,
`CLOSED`.
`match_status` is free-form text written by the worker: `processing`, `matched`, `skipped`,
`no_text`, `failed`, `dlq`.
---
## API reference
Base URL: `http://localhost:8000`. Interactive docs at `/docs`.
### Auth — `users/app.py`, `notifications/app.py`, `forget_password/app.py`
| Method | Path | Guard | Purpose |
|---|---|---|---|
| POST | `/users/signup` | public | Create a candidate account (inactive) and mail a confirmation link |
| POST | `/users/login` | public | Email/username + password → token envelope |
| POST | `/users/refresh` | public | Refresh token → new token pair |
| GET | `/users/me` | any authenticated user | Current user with resolved permissions |
| POST | `/users/confirm-email` | public | Consume a confirmation token, activate the account |
| POST | `/users/confirm-email/resend` | public | Re-issue a confirmation link (cooldown enforced) |
| POST | `/users/forget-password` | public | Mail a reset code |
| POST | `/users/forget-password/verify-code` | public | Verify the code → `type=reset` JWT |
| POST | `/users/forget-password/new-password` | reset JWT | Set the new password |
### Users — `users/app.py`
| Method | Path | Required tag |
|---|---|---|
| GET | `/users/fetch` | `rbac_users.view` |
| POST | `/users/create` | `rbac_users.create` |
| PUT | `/users/update?record_id=` | `rbac_users.edit` |
| PUT | `/users/assign-role?record_id=` | `rbac_users.edit` (+ `rbac_users.manage` inside the service) |
| PUT | `/users/remove-role?record_id=` | `rbac_users.edit` (+ `rbac_users.manage`) |
| DELETE | `/users/delete?record_id=` | `rbac_users.delete` |
### Roles and permissions — `role/app.py`
| Method | Path | Required tag |
|---|---|---|
| GET | `/roles/fetch` | `rbac_users.view` |
| POST | `/roles/create` | `rbac_users.create` |
| PUT | `/roles/update?record_id=` | `rbac_users.edit` |
| DELETE | `/roles/delete?record_id=` | `rbac_users.delete` |
| GET | `/permissions/fetch` | `rbac_users.view` |
| POST | `/permissions/create` | `rbac_users.manage` |
| PUT | `/permissions/update?record_id=` | `rbac_users.manage` |
| GET | `/permission-tags/fetch` | `rbac_users.view` |
### Inbox — `inbox/app.py`
| Method | Path | Guard | Purpose |
|---|---|---|---|
| GET | `/email/fetch` | upstream token only | Pull from the Email API, decode attachments, upsert, enqueue matching. `test_on=true` (default) returns raw payloads and skips the account-setup mails |
| GET | `/inbox/fetch` | none | Stored messages, with attachments inlined as base64 |
| GET | `/inbox/all-applications` | `inbox.view` | The Applications tab. Filters: `application_status`, `isread`, `search`, `record_id`, `top`, `skip` |
| POST | `/inbox/{record_id}/match` | `inbox.edit` | Force a re-match of one message |
| POST | `/inbox/{record_id}/read` | `inbox.edit` | Mark read locally |
| GET | `/inbox/{record_id}/read-status` | `inbox.edit` | Re-pull read status from upstream for one message |
### Jobs and candidates — `job/app.py`
| Method | Path | Required tag | Purpose |
|---|---|---|---|
| GET | `/jobs/alias` | public | Accepted platform shorthands (`fb`, `ig`, `li`, `x`, …) |
| POST | `/job/post-job` | `job_board.create` | Render the ad, create the Buffer post, persist the result |
| GET | `/job/buffer/channels` | `job_board.view` | Connected Buffer channels across all organizations |
| POST | `/candidate/cv_upload` | `candidates.create` | Upload a PDF, get extracted text back |
| POST | `/candidate/inbox-match?inbox_message_id=` | `candidates.edit` | Queue a forced re-match for a stored message |
| GET | `/candidate/fetch?user_id=` | `candidates.view` | Candidate profile via the `inbox` join |
`POST /job/post-job` takes `mode` ∈ `addToQueue` | `shareNow` | `customScheduled`; the
`customScheduled` mode requires `scheduler_date` (and optionally `scheduler_time`), which the
route combines into a UTC `due_at`.
---
## Authentication and RBAC
**Tokens.** PyJWT, HS256, with a `type` claim that `decode_token(..., expected_type=...)`
rejects on mismatch. Three types:
| Type | Lifetime (default) | Extra claims |
|---|---|---|
| `access` | 30 min | `email`, `role_id` |
| `refresh` | 7 days | — |
| `reset` | 10 min | `crid` (reset-code row id) |
`iat` / `exp` are always timezone-aware UTC. Every token carries a `jti`.
**Login response** puts the OAuth2 fields at the root so Swagger's Authorize button can read
them:
```json
{
"access_token": "...",
"refresh_token": "...",
"token_type": "bearer",
"expires_in": 1800,
"data": { "id": "...", "email": "...", "role_id": 3, "role_name": "recruiter", "...": "..." },
"status_code": 200
}
```
**Passwords** use `bcrypt` directly rather than `passlib` — passlib 1.7.4 reads
`bcrypt.__about__.__version__`, which bcrypt dropped in 4.1, and the failed probe makes it
reject every password as over 72 bytes. Input is truncated to 72 bytes on a character
boundary before hashing.
**Authorization.** `get_current_user` decodes the access token, loads the user by `sub`,
rejects missing / deleted / inactive accounts, resolves the role's tags, and returns a
serialized user dict with a `permissions` list. `require_permission(*tags, require_all=True)`
is the dependency that guards routes; a user with no role assigned gets a 403 before any tag
check runs.
Role assignment is additionally guarded in `users/views.py`: you cannot grant a role holding
permissions you do not yourself hold.
---
## Background jobs
Taskiq over **Redis Streams**, with a result backend and a Redis-backed schedule source.
**Middleware order matters** — `DeadLetterMiddleware` sits before `SmartRetryMiddleware` so a
`PermanentTaskError` can set `retry_on_error=False` before retry logic runs.
| Task | Trigger | What it does |
|---|---|---|
| `inbox.match_message` | enqueued by `/email/fetch`, `/inbox/{id}/match`, `/candidate/inbox-match` | Extract résumé text → run the agent → write match results |
| `inbox.sync_read_status` | cron, `EMAIL_SYNC_CRON` (default every minute) | Pull read-status deltas from the Email API and apply them |
| `ping` | manual | Framework smoke test |
**Retries.** `SmartRetryMiddleware` with jitter and exponential delay, `TASKIQ_MAX_RETRIES`
attempts, capped at `TASKIQ_MAX_DELAY`. Raising `PermanentTaskError` (missing record, no
attachment, blank `record_id`) skips retries entirely.
**Dead letter queue.** Exhausted or permanently-failed tasks are written to the
`taskiq:dlq` Redis stream as a JSON payload. For `inbox.match_message` specifically, the
middleware also stamps `match_status="dlq"` on the row so the failure is visible in the UI
rather than only in Redis.
**The read-status latch.** `apply_read_status` only ever applies `false → true`. Nothing
pushes local reads back to Outlook, so upstream keeps reporting `isRead=false`; without the
latch the every-minute sweep would un-read a mail the user just opened. The trade-off is that
un-reading a mail in Outlook no longer propagates here. `sync_read_status` holds a Redis lock
(`inbox:sync_read_status:lock`, 300s TTL) so overlapping cron ticks cannot double-run, and
pages at most 10 rounds per tick.
---
## The matching agent
A small LangGraph `StateGraph` over an `AgentState` TypedDict.
```
START → prepare → (conditional) → match_jobs → END
└→ END (when resume_text or job_posts is empty)
```
- **`prepare_context`** trims the subject and résumé text and normalizes the job posts down to
the fields the model needs. Empty résumé text or no active posts short-circuits to
`status="skipped"`.
- **`match_jobs`** calls `llm_call(prompt(), user_prompt(state), json_mode=True)` and parses
the response.
The parser (`agent/decorators.py::parse_match_response`) is deliberately strict: a suggested
id must be present in the posts that were actually sent, must be a valid UUID, and duplicates
are dropped. The model cannot invent a job post. `reasoning` returned as a list is joined into
a string; non-string fields fall back to `""`.
The graph is compiled once per process (`init_agent()` from the FastAPI lifespan, or lazily on
first use inside the worker) and the OpenAI client is created lazily and closed on shutdown.
Startup is **fault-tolerant**: if the broker or the LLM/agent fails to initialize, `main.py`
logs a warning and the API still serves. Only the database is a hard requirement.
---
## External integrations
| Service | Used by | Contract |
|---|---|---|
| **Email API** (a Microsoft Graph proxy) | `inbox/` | `GET {EMAIL_URL}/emails`, `GET {EMAIL_URL}/emails/{id}`, `GET {EMAIL_URL}/sync/read-status`, `GET {EMAIL_URL}/sync/read-status/message/{id}` — Bearer `EMAIL_API_TOKEN` |
| **Teams Mail API** | `notifications/`, `forget_password/` | multipart POST to `TEAMS_MAIL_API_URL`; success is HTTP **202**, anything else raises |
| **Buffer** | `job/job_post/` | GraphQL against `BUFFER_API_URL` — `createPost` mutation, `account { organizations }` and `channels` queries |
| **OpenAI** | `agent/`, `llm_setup.py` | Chat Completions with `response_format: json_object` |
Attachments are written to `backend/inbox/decoded_attachments/`. In Docker this directory is
bind-mounted into the worker so both processes see the same files.
---
## Configuration
Copy `.env.example` to `.env` and fill it in. `.env` is git-ignored; `.env.example` is not.
`db_setup.Settings` reads `backend/.env` or the repo-root `.env`; every other module reads its
own keys with `os.getenv`.
### Database
| Variable | Default | Notes |
|---|---|---|
| `DB_USERNAME`, `DB_PASSWORD`, `DB_HOST`, `DB_PORT`, `DB_NAME` | — | **Required.** `DB_PORT` must be an integer |
| `DATABASE_URL` | — | Full DSN; wins over the parts above |
| `DB_SSLMODE` | — | e.g. `require` on Azure; translated to asyncpg's `ssl` |
| `DB_SCHEMAS` | `app` | Comma-separated; created on startup |
| `DB_DEFAULT_SCHEMA` | `app` | Schema for models that declare none |
| `DB_ECHO` | `false` | SQL logging |
| `DB_POOL_SIZE` / `DB_MAX_OVERFLOW` / `DB_POOL_RECYCLE` | `5` / `10` / `1800` | |
| `DB_CONNECT_RETRIES` | `10` | Startup wait-for-Postgres |
| `DB_AUTO_MIGRATE` | `true` | Run `upgrade head` on startup |
| `DB_AUTOGENERATE` | `true` | Write a revision when models drift |
| `APP_NAME` | `hr-ats-portal` | Postgres `application_name` |
### Auth
| Variable | Default |
|---|---|
| `JWT_SECRET_KEY` | — (**required**) |
| `JWT_ALGORITHM` | `HS256` |
| `JWT_ACCESS_TOKEN_EXPIRE_MINUTES` | `30` |
| `JWT_REFRESH_TOKEN_EXPIRE_DAYS` | `7` |
| `JWT_RESET_TOKEN_EXPIRE_MINUTES` | `10` |
### Email ingestion
| Variable | Default |
|---|---|
| `EMAIL_URL`, `EMAIL_API_TOKEN` | — |
| `EMAIL_SYNC_FOLDER` | `inbox` |
| `EMAIL_SYNC_SINCE` | — |
| `EMAIL_SYNC_CRON` | `* * * * *` |
| `BACKEND_URL` | `http://localhost:8000` (used for the internal confirmation-resend call) |
| `DEFAULT_CANDIDATE_PASSWORD` | `Utopia!@#` — placeholder only; the account is inactive until confirmed |
### Mail out
| Variable | Default |
|---|---|
| `TEAMS_MAIL_API_URL`, `TEAMS_API_TOKEN` | — |
| `FRONTEND_URL` | `http://localhost:5173` |
| `CONFIRM_EMAIL_PATH` | `/auth/confirm-email` |
| `CONFIRM_TOKEN_TTL_SECONDS` | `86400` |
| `CONFIRM_TOKEN_RESEND_SECONDS` | `60` |
| `RESET_CODE_TTL_SECONDS` | `60` |
| `RESET_CODE_RESEND_SECONDS` | `30` |
| `RESET_CODE_MAX_ATTEMPTS` | `5` |
### Buffer
| Variable | Default |
|---|---|
| `BUFFER_API` | — (access token) |
| `BUFFER_API_URL` | `https://api.buffer.com` |
| `BUFFER_CHANNEL_ID` | — (fallback channel) |
### OpenAI
| Variable | Default |
|---|---|
| `OPENAI_API_KEY` | — |
| `OPENAI_MODEL` | `gpt-5.4-mini` |
| `OPENAI_TEMPERATURE` | `0` — leave blank to omit the parameter for models that reject it |
| `OPENAI_MAX_OUTPUT_TOKENS` | `32768` (`.env.example` ships `4096`) |
| `OPENAI_TIMEOUT` / `OPENAI_MAX_RETRIES` / `OPENAI_CONNECT_RETRIES` | `60` / `3` / `3` |
| `OPENAI_BASE_URL`, `OPENAI_ORGANIZATION`, `OPENAI_PROJECT` | — (set only for Azure or a gateway) |
### Taskiq / Redis
| Variable | Default |
|---|---|
| `REDIS_URL` | `redis://localhost:6379/0` |
| `TASKIQ_QUEUE_NAME` | `inbox` |
| `TASKIQ_CONSUMER_GROUP` | `taskiq` |
| `TASKIQ_MAX_RETRIES` | `3` |
| `TASKIQ_RETRY_DELAY` | `5` |
| `TASKIQ_MAX_DELAY` | `120` |
| `TASKIQ_IDLE_TIMEOUT_MS` | `600000` |
| `TASKIQ_DLQ_STREAM` | `taskiq:dlq` |
| `TASKIQ_WORKER_NAME` | falls back to `HOSTNAME` |
| `APP_VERSION` | `dev` |
---
## Running locally
**Prerequisites:** Python 3.12, PostgreSQL, Redis.
```bash
cd backend
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env # then fill it in
```
All commands must be run from `backend/` — the import paths depend on it.
**API:**
```bash
uvicorn main:app --reload --port 8000
```
Startup connects to Postgres (retrying with backoff), creates the configured schemas, runs
migrations to head, then starts the broker, the OpenAI client and the agent graph. Broker and
LLM failures are logged and skipped; the API still comes up.
**Worker** (needs Redis):
```bash
taskiq worker taskiq_management.broker_setup:broker \
inbox.tasks inbox.sync_tasks taskiq_management.tasks
```
**Scheduler** (cron ticks for `inbox.sync_read_status`):
```bash
taskiq scheduler taskiq_management.broker_setup:scheduler inbox.sync_tasks
```
Docs: <http://localhost:8000/docs>
---
## Database migrations
`alembic_setup.py` wraps Alembic so the plain `alembic` CLI and the app's own
migrate-on-startup share one configuration. It scaffolds `alembic.ini`, `migrations/env.py`
and `script.py.mako` on first use and never overwrites them. Model modules are discovered
automatically — every `<package>/models.py` under `backend/` is imported before the metadata is
diffed.
```bash
python alembic_setup.py migrate # upgrade to head, then autogenerate any drift
python alembic_setup.py revision -m "add x" # write a revision if the models have drifted
python alembic_setup.py upgrade -r head
python alembic_setup.py downgrade -r -1
python alembic_setup.py current
python alembic_setup.py head
```
Migrations run under a Postgres advisory lock, so several workers booting at once cannot
migrate concurrently. Empty revisions are suppressed. Alembic's own `alembic_version` table is
excluded from autogenerate, as is anything outside the configured schemas.
The module is named `alembic_setup` rather than `alembic` because `backend/` is on `sys.path`
and a module called `alembic.py` would shadow the installed package.
---
## Docker
The repo-root `docker-compose.yml` runs Redis plus the two Taskiq processes; the API itself is
expected to run on the host (the compose file points the containers at
`host.docker.internal` for the database).
```bash
docker compose up -d # from the repo root
docker compose logs -f taskiq-worker
```
`backend/Dockerfile` builds a `python:3.12-slim` image whose default command is the Taskiq
worker. `backend/inbox/decoded_attachments` is bind-mounted so the worker can read the
attachments the API wrote.
---
## Response conventions
Every handler wraps its body in the same try/except:
```python
try:
service=Email(session=session)
data=await service.some_method(...)
return JSONResponse(content={"data":data,"status_code":200})
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
```
| Shape | Response |
|---|---|
| List | `{"data": [...], "total": <n>, "status_code": 200}` |
| Single record | `{"data": {...}, "total": 1, "status_code": 200}` |
| Login / refresh | OAuth2 fields at the root, user under `data` |
| Error | FastAPI's `{"detail": "..."}` with the real status code |
Serializers always `str()` UUIDs, `.isoformat()` datetimes, and never emit `password`.
---
## Known gaps and gotchas
- **`DB_PORT` must be set.** `db_setup.Settings` evaluates `int(os.getenv("DB_PORT"))` at class
definition time, so a missing value raises `TypeError` on import rather than a friendly
config error.
- **CORS is fully open** (`allow_origins=["*"]` with credentials). Fine for development, needs
tightening before production.
- **`/email/fetch` and `/inbox/fetch` carry no permission guard.** `/email/fetch` authenticates
only against the upstream Email API token.
- **`.doc` / `.docx` résumés are decoded and stored but not parsed.** `extract_resume_text`
handles PDFs only and reports `no PDF attachment to extract` for the rest.
- **`serialize_application` returns `null` for `ats_score`, `phone`, `recruiter` and
`duplicate`** — `inbox_messages` has no columns for them yet, and `processing` is derived
from `message_read` alone, so it is only ever `"Read"` or `"Unread"`.
- **`Inbox.get_candidate_profile` filters on `cls.user.role_id`**, which is a relationship
attribute rather than a joined column; the candidate-profile query needs a join before it
behaves as intended.
- **Attachment paths may be Windows absolutes** written by the host API but read by a Linux
worker. `resolve_attachment_path` normalizes separators and falls back to the basename under
the mounted `decoded_attachments` directory.
- **Read status is a one-way latch** — see [Background jobs](#background-jobs).
- **There is no test suite** in `backend/` at present.
---
## Editing this codebase
Before changing anything here, read [`LLM_CONTEXT_PROMPT.md`](LLM_CONTEXT_PROMPT.md). It states
the house style in full and is the reference used to keep new code indistinguishable from
`users/` and `inbox/`. The short version: mirror the neighbouring file, keep the layer duties
intact, add no new layers, and do not reformat code you did not otherwise need to touch.
Adding an endpoint, in order:
1. Model accessor in `models.py` (if it touches the DB).
2. Service method in `views.py`.
3. `serialize_*` in `serializers.py` if the shape is new.
4. Route in `app.py` with the standard try/except and `JSONResponse`.
5. `CurrentUser` or `Depends(require_permission(...))` if the route is protected.

30
backend/agent/agent_setup.py
View File

@ -13,45 +13,39 @@ from __future__ import annotations
import logging
from langgraph.graph import END, START, StateGraph
from langgraph.graph import END,START,StateGraph
from agent.models import AgentState
from agent.views import finalize, match_jobs, prepare_context, route_after_prepare
from agent.views import match_jobs,prepare_context,route_after_prepare
logger = logging.getLogger("agent")
logger=logging.getLogger("agent")
_graph = None
_graph=None
def build_graph():
"""Construct and compile the HR-ATS candidate matching graph."""
graph = StateGraph(AgentState)
graph.add_node("prepare", prepare_context)
graph.add_node("match_jobs", match_jobs)
graph.add_node("finalize", finalize)
graph.add_edge(START, "prepare")
graph.add_conditional_edges("prepare", route_after_prepare)
graph.add_edge("match_jobs", "finalize")
graph.add_edge("finalize", END)
graph=StateGraph(AgentState)
graph.add_node("prepare",prepare_context)
graph.add_node("match_jobs",match_jobs)
graph.add_edge(START,"prepare")
graph.add_conditional_edges("prepare",route_after_prepare)
graph.add_edge("match_jobs",END)
return graph.compile()
def get_graph():
"""Return the cached compiled graph, building it on first use."""
global _graph
if _graph is None:
_graph = build_graph()
_graph=build_graph()
logger.info("langgraph compiled")
return _graph
async def init_agent():
"""Warm the compiled graph. LLM init stays on llm_setup.init_llm()."""
get_graph()
async def close_agent():
"""Drop the cached graph."""
global _graph
_graph = None
_graph=None
logger.info("agent graph closed")

71
backend/agent/decorators.py
View File

@ -11,45 +11,41 @@ import uuid
def normalize_job_posts(job_posts) -> list[dict]:
"""Keep only dict items with an id field; stringify ids for the LLM."""
if not job_posts:
return []
normalized: list[dict] = []
normalized=[]
for item in job_posts:
if not isinstance(item, dict):
if not isinstance(item,dict):
continue
job_id = item.get("id")
job_id=item.get("id")
if job_id is None:
continue
normalized.append(
{
"id": str(job_id),
"title": item.get("title") or "",
"description": item.get("description") or "",
"post_text": item.get("post_text") or "",
"requirements": item.get("requirements") or [],
"optional_skills": item.get("optional_skills") or [],
"location": item.get("location") or "",
"employment_type": item.get("employment_type") or "",
}
)
normalized.append({
"id":str(job_id),
"title":item.get("title") or "",
"description":item.get("description") or "",
"post_text":item.get("post_text") or "",
"requirements":item.get("requirements") or [],
"optional_skills":item.get("optional_skills") or [],
"location":item.get("location") or "",
"employment_type":item.get("employment_type") or "",
})
return normalized
def parse_match_response(data, allowed_ids) -> tuple[list[str], str, str]:
"""Filter model JSON ids to the allowed job-post set."""
if not isinstance(data, dict):
def parse_match_response(data,allowed_ids) -> tuple[list[str],str,str,str]:
if not isinstance(data,dict):
raise RuntimeError(f"model did not return a JSON object: {data!r}")
allowed = set(allowed_ids or [])
raw_ids = data.get("suggested_job_post_ids") or []
if not isinstance(raw_ids, list):
raw_ids = []
allowed=set(allowed_ids or [])
raw_ids=data.get("suggested_job_post_ids") or []
if not isinstance(raw_ids,list):
raw_ids=[]
suggested: list[str] = []
seen: set[str] = set()
suggested=[]
seen=set()
for raw_id in raw_ids:
job_id = str(raw_id).strip()
job_id=str(raw_id).strip()
if not job_id or job_id not in allowed or job_id in seen:
continue
try:
@ -59,13 +55,18 @@ def parse_match_response(data, allowed_ids) -> tuple[list[str], str, str]:
seen.add(job_id)
suggested.append(job_id)
summary = data.get("summary")
if not isinstance(summary, str):
summary = ""
summary=data.get("summary")
if not isinstance(summary,str):
summary=""
reasoning = data.get("reasoning")
if isinstance(reasoning, list):
reasoning = "\n".join(str(item) for item in reasoning)
if not isinstance(reasoning, str):
reasoning = ""
return suggested, summary.strip(), reasoning.strip()
reasoning=data.get("reasoning")
if isinstance(reasoning,list):
reasoning="\n".join(str(item) for item in reasoning)
if not isinstance(reasoning,str):
reasoning=""
experience=data.get("experience")
if not isinstance(experience,str):
experience=""
return suggested,summary.strip(),reasoning.strip(),experience.strip()

17
backend/agent/execute_agent.py
View File

@ -9,14 +9,11 @@ from agent.agent_setup import get_graph
from agent.serializers import serialize_agent_result
async def run_agent(*, subject="", resume_text="", job_posts=None) -> dict:
"""Run the default graph and return a serialized result dict."""
final_state = await get_graph().ainvoke(
{
"subject": subject or "",
"resume_text": resume_text or "",
"job_posts": job_posts or [],
"status": "pending",
}
)
async def run_agent(*,subject="",resume_text="",job_posts=None) -> dict:
final_state=await get_graph().ainvoke({
"subject":subject or "",
"resume_text":resume_text or "",
"job_posts":job_posts or [],
"status":"pending",
})
return serialize_agent_result(final_state)

23
backend/agent/models.py
View File

@ -5,17 +5,16 @@ Pure module: no FastAPI imports and no HTTPException.
from __future__ import annotations
from typing import Literal, TypedDict
from typing import Literal,TypedDict
class AgentState(TypedDict, total=False):
"""Shared state passed between graph nodes."""
subject: str
resume_text: str
job_posts: list[dict]
suggested_job_post_ids: list[str]
summary: str
reasoning: str
error: str
status: Literal["pending", "ready", "matched", "skipped", "failed"]
class AgentState(TypedDict,total=False):
subject:str
resume_text:str
experience:str
job_posts:list[dict]
suggested_job_post_ids:list[str]
summary:str
reasoning:str
error:str
status:Literal["pending","ready","matched","skipped","failed"]

3
backend/agent/prompt.py
View File

@ -26,7 +26,8 @@ Respond with JSON only:
{
"suggested_job_post_ids": ["uuid", "..."],
"summary": "one short sentence for the recruiter",
"reasoning": "brief bullet-style explanation per suggested match"
"reasoning": "brief bullet-style explanation per suggested match",
"experience": "the relevant experience of the candidate in years for the suggested match"
}
"""

14
backend/agent/serializers.py
View File

@ -6,12 +6,12 @@ Pure module: no FastAPI imports and no HTTPException.
from __future__ import annotations
def serialize_agent_result(state: dict) -> dict:
"""Plain dict for services/serializers — no ORM objects."""
def serialize_agent_result(state:dict) -> dict:
return {
"suggested_job_post_ids": state.get("suggested_job_post_ids") or [],
"summary": state.get("summary") or "",
"reasoning": state.get("reasoning") or "",
"status": state.get("status") or "failed",
"error": state.get("error") or "",
"suggested_job_post_ids":state.get("suggested_job_post_ids") or [],
"summary":state.get("summary") or "",
"reasoning":state.get("reasoning") or "",
"experience":state.get("experience") or "",
"status":state.get("status") or "failed",
"error":state.get("error") or "",
}

87
backend/agent/views.py
View File

@ -11,81 +11,58 @@ from typing import Literal
from langgraph.graph import END
from agent.decorators import normalize_job_posts, parse_match_response
from agent.decorators import normalize_job_posts,parse_match_response
from agent.models import AgentState
from agent.prompt import prompt, user_prompt
from agent.prompt import prompt,user_prompt
from llm_setup import llm_call
logger = logging.getLogger("agent")
logger=logging.getLogger("agent")
async def prepare_context(state: AgentState) -> dict:
"""Validate inputs and decide whether matching should run."""
subject = (state.get("subject") or "").strip()
resume_text = (state.get("resume_text") or "").strip()
job_posts = normalize_job_posts(state.get("job_posts"))
async def prepare_context(state:AgentState) -> dict:
subject=(state.get("subject") or "").strip()
resume_text=(state.get("resume_text") or "").strip()
job_posts=normalize_job_posts(state.get("job_posts"))
if not resume_text:
return {
"status": "skipped",
"error": "resume_text is empty",
"suggested_job_post_ids": [],
"summary": "",
"reasoning": "",
}
return {"status":"skipped","error":"resume_text is empty","suggested_job_post_ids":[]}
if not job_posts:
return {
"status": "skipped",
"error": "no active job posts to match against",
"suggested_job_post_ids": [],
"summary": "",
"reasoning": "",
}
return {"status":"skipped","error":"no active job posts to match against","suggested_job_post_ids":[]}
return {
"subject": subject,
"resume_text": resume_text,
"job_posts": job_posts,
"status": "ready",
"error": "",
"subject":subject,
"resume_text":resume_text,
"job_posts":job_posts,
"status":"ready",
"error":"",
}
def route_after_prepare(state: AgentState) -> Literal["match_jobs", "__end__"]:
if state.get("status") == "ready":
def route_after_prepare(state:AgentState) -> Literal["match_jobs","__end__"]:
if state.get("status")=="ready":
return "match_jobs"
return END
async def match_jobs(state: AgentState) -> dict:
"""Ask the LLM (via llm_setup.llm_call) to map the candidate to job posts."""
async def match_jobs(state:AgentState) -> dict:
try:
data = await llm_call(prompt(), user_prompt(state), json_mode=True)
allowed_ids = {item["id"] for item in state.get("job_posts") or []}
suggested, summary, reasoning = parse_match_response(data, allowed_ids)
data=await llm_call(prompt(),user_prompt(state),json_mode=True)
allowed_ids={item["id"] for item in state.get("job_posts") or []}
suggested,summary,reasoning,experience=parse_match_response(data,allowed_ids)
return {
"status": "matched",
"suggested_job_post_ids": suggested,
"summary": summary,
"reasoning": reasoning,
"status":"matched",
"suggested_job_post_ids":suggested,
"summary":summary,
"reasoning":reasoning,
"experience":experience,
}
except Exception as exc:
except Exception as e:
logger.exception("agent match_jobs failed")
return {
"status": "failed",
"error": str(exc),
"suggested_job_post_ids": [],
"summary": "",
"reasoning": "",
"status":"failed",
"error":str(e),
"suggested_job_post_ids":[],
"summary":"",
"reasoning":"",
"experience":"",
}
async def finalize(state: AgentState) -> dict:
"""Normalize terminal state for callers."""
return {
"suggested_job_post_ids": state.get("suggested_job_post_ids") or [],
"summary": state.get("summary") or "",
"reasoning": state.get("reasoning") or "",
"status": state.get("status") or "failed",
"error": state.get("error") or "",
}

9
backend/inbox/app.py
View File

@ -15,6 +15,7 @@ router = APIRouter()
async def fetch_email(
top:int=Query(100),
skip:int=Query(0,ge=0),
test_on: bool = Query(True),
token: str | None = Query(None),
session: AsyncSession = Depends(get_session),
):
@ -27,13 +28,17 @@ async def fetch_email(
items_lst=[]
for item in value:
message_id=item.get("id")
service_per_email=await service.get_email_by_id(message_id)
service_per_email=await service.get_email_by_id(message_id,test_on)
items_lst.append({"message_id":message_id,"email_contents":service_per_email})
if service.pending_match_ids:
await service.enqueue_matching(list(service.pending_match_ids),force=False)
return JSONResponse(content={"data":items_lst,"status_code":200})
account_setup=[]
if service.pending_confirmation_emails:
account_setup=await service.send_account_setup(list(service.pending_confirmation_emails))
return JSONResponse(content={"data":items_lst,"account_setup":account_setup,"status_code":200})
except HTTPException:
raise

109
backend/inbox/models.py
View File

@ -1,14 +1,31 @@
import logging
import os
import uuid
from datetime import datetime, timezone
from typing import Any, Optional
from dotenv import load_dotenv
from fastapi import HTTPException
from inbox.enums import Candidate_application_Status
from role.models import EnumRoles, Roles
from sqlalchemy import Column, DateTime, func, or_, update
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import Field, Relationship, SQLModel, select, true
from users.models import Users
from users.plugins import hash_password
load_dotenv()
logger = logging.getLogger("inbox.models")
# Placeholder only. The account lands inactive and the candidate is mailed a
# confirmation link; the real password comes from the reset flow afterwards.
DEFAULT_CANDIDATE_PASSWORD = os.getenv("DEFAULT_CANDIDATE_PASSWORD", "Utopia!@#")
CANDIDATE_ROLE_ID_FALLBACK = 8 # mirrors users/views.py:signup_user
SKIP_SENDER_PREFIXES = ("noreply", "no-reply", "donotreply", "do-not-reply",
"mailer-daemon", "postmaster", "bounce")
class Inbox(SQLModel, table=True):
@ -65,6 +82,7 @@ class Inbox_Messages(SQLModel, table=True):
file_name: str | None = Field(default=None)
file_path: str | None = Field(default=None)
resume_text: str | None = Field(default=None)
experience: str | None = Field(default=None)
suggested_job_post_ids: list[str] | None = Field(default=None, sa_column=Column(JSONB))
match_summary: str | None = Field(default=None)
match_reasoning: str | None = Field(default=None)
@ -83,6 +101,15 @@ class Inbox_Messages(SQLModel, table=True):
return body
return email_data.get("bodyPreview") or ""
# @classmethod
# async def get_candidate_profile(cls,session:AsyncSession,user_id:uuid.UUID|None=None):
# try:
# qryy=select(cls,Users).join(cls,cls.)
# if user_id
# except Exception as e:
# raise HTTPException(status_code=500,detail=str(e))
@classmethod
async def get_all_applications(cls,session:AsyncSession,message_id:uuid.UUID|int|None=None):
try:
@ -101,6 +128,7 @@ class Inbox_Messages(SQLModel, table=True):
record_id,
*,
resume_text=None,
experience=None,
suggested_job_post_ids=None,
summary="",
reasoning="",
@ -118,6 +146,7 @@ class Inbox_Messages(SQLModel, table=True):
row.match_reasoning = reasoning or None
row.match_status = status or None
row.match_error = error or None
row.experience = experience or None
row.matched_at = datetime.now(timezone.utc)
session.add(row)
await session.commit()
@ -156,6 +185,72 @@ class Inbox_Messages(SQLModel, table=True):
"full_email_response": email_data,
}
@classmethod
def _sender_address(cls, email_data: dict) -> str:
return (
email_data.get("from", {})
.get("emailAddress", {})
.get("address", "")
or ""
).strip().lower()
@classmethod
def _sender_display_name(cls, email_data: dict, address: str) -> str:
name = (
email_data.get("from", {})
.get("emailAddress", {})
.get("name")
or ""
).strip()
if name:
return name
return address.split("@", 1)[0] if address else "candidate"
@classmethod
def _is_linkable_sender(cls, address: str) -> bool:
if not address or "@" not in address:
return False
local = address.split("@", 1)[0]
return not local.startswith(SKIP_SENDER_PREFIXES)
@classmethod
async def _link_sender(cls,session:AsyncSession,email_data:dict,email):
address=cls._sender_address(email_data)
if not cls._is_linkable_sender(address):
return None
try:
user=(await session.execute(
select(Users).where(func.lower(Users.email)==address)
)).scalars().first()
if not user:
role=await Roles.get_role_by_name(session,EnumRoles.CANDIDATE.value)
user=Users(
name=cls._sender_display_name(email_data,address),
email=address,
role_id=role.id if role else CANDIDATE_ROLE_ID_FALLBACK,
password=hash_password(DEFAULT_CANDIDATE_PASSWORD),
)
session.add(user)
session.add(Inbox(user_id=user.id,message_id=email.id))
await session.commit()
return address
link=(await session.execute(
select(Inbox).where(Inbox.message_id==email.id,Inbox.user_id==user.id)
)).scalars().first()
if not link:
session.add(Inbox(user_id=user.id,message_id=email.id))
await session.commit()
return None
except IntegrityError:
await session.rollback()
return None
except Exception as e:
await session.rollback()
logger.warning("sender link failed for %s: %s",address,e)
return None
@classmethod
async def insert_email(
cls,
@ -163,9 +258,11 @@ class Inbox_Messages(SQLModel, table=True):
email_data: dict,
file_path: list[str] | None = None,
):
"""Returns (row, new_user_email). new_user_email is set only when this call
created the sender's Users row."""
fields = cls._fields_from_email(email_data, file_path)
external_id = fields.get("message_id")
link_user=None
if external_id:
existing = (
await session.execute(
@ -178,12 +275,18 @@ class Inbox_Messages(SQLModel, table=True):
session.add(existing)
await session.commit()
await session.refresh(existing)
return existing
if fields.get("attachment"):
link_user=await cls._link_sender(session, email_data, existing)
return existing, link_user
email = cls(**fields)
session.add(email)
await session.commit()
return email
if fields.get("attachment"):
link_user=await cls._link_sender(session, email_data, email)
return email, link_user
@classmethod
def _search_filter(cls, search: str):

15
backend/inbox/plugins.py
View File

@ -17,10 +17,25 @@ load_dotenv()
EMAIL_URL=os.getenv("EMAIL_URL")
EMAIL_API_TOKEN=os.getenv("EMAIL_API_TOKEN")
BACKEND_URL=os.getenv("BACKEND_URL","http://localhost:8000")
_ATTACHMENTS_DIR=Path(__file__).resolve().parent/"decoded_attachments"
async def request_email_confirmation(email):
"""POST /users/confirm-email/resend on this same service -> status code.
Goes through the endpoint rather than importing Confirmation so the token row,
resend cooldown and mail send stay on one code path.
"""
async with httpx.AsyncClient(timeout=20.0) as client:
response=await client.post(
f"{BACKEND_URL.rstrip('/')}/users/confirm-email/resend",
json={"email":email},
)
return response.status_code
async def fetch_read_status_delta(folder, since=None, limit=1000, max_pages=10, token=None):
"""GET /sync/read-status -> the raw round dict."""
if not EMAIL_URL:

2
backend/inbox/serializers.py
View File

@ -97,7 +97,7 @@ def serialize_application(message: Inbox_Messages) -> dict:
"resume_text": message.resume_text,
"ats_score": None,
"phone": None,
"experience": None,
"experience": message.experience or "",
"recruiter": None,
"duplicate": None,
}

53
backend/inbox/tasks.py
View File

@ -3,19 +3,19 @@
from __future__ import annotations
import logging
from datetime import datetime, timezone
from datetime import datetime,timezone
from agent.execute_agent import run_agent
from db_setup import session_scope
from inbox.models import Inbox_Messages
from inbox.plugins import extract_resume_text
from job.job_post.models import JobPosts
from job.job_post.serializers import serialize_job_post
from taskiq_management.broker_setup import MAX_RETRIES, RETRY_DELAY, broker
from taskiq_management.broker_setup import MAX_RETRIES,RETRY_DELAY,broker
from taskiq_management.middleware import PermanentTaskError
logger=logging.getLogger("inbox.tasks")
_DONE_STATUSES=frozenset({"matched","skipped","no_text","failed","dlq"})
_DONE=frozenset({"matched","skipped","no_text","failed","dlq"})
@broker.task(
@ -27,74 +27,51 @@ _DONE_STATUSES=frozenset({"matched","skipped","no_text","failed","dlq"})
async def match_inbox_message(record_id:str,force:bool=False) -> dict:
if not record_id or not str(record_id).strip():
raise PermanentTaskError("record_id is required")
record_id=str(record_id).strip()
async with session_scope() as session:
row=await Inbox_Messages.get_inbox_message_by_id(session,record_id)
if not row:
raise PermanentTaskError(f"inbox message {record_id} not found")
if not force and row.match_status in _DONE_STATUSES:
logger.info("skip %s — already %s",record_id,row.match_status)
if not force and row.match_status in _DONE:
return {"status":row.match_status,"skipped":True}
if not row.attachment or not row.file_path:
raise PermanentTaskError("message has no attachment to match")
paths=[p.strip() for p in row.file_path.split(",") if p.strip()]
subject=row.message_subject or ""
row.match_status="processing"
row.match_error=None
row.matched_at=datetime.now(timezone.utc)
session.add(row)
await session.commit()
await session.refresh(row)
paths=[p.strip() for p in (row.file_path or "").split(",") if p.strip()]
subject=row.message_subject or ""
posts=await JobPosts.get_active_job_posts(session)
job_posts=[serialize_job_post(p) for p in posts]
text,extract_err=await extract_resume_text(paths)
if not text:
async with session_scope() as session:
await Inbox_Messages.set_match_result(
session,
record_id,
status="no_text",
error=extract_err or "no text extracted",
session,record_id,status="no_text",error=extract_err or "no text extracted",
)
return {"status":"no_text","error":extract_err}
from agent.execute_agent import run_agent
async with session_scope() as session:
posts=await JobPosts.get_active_job_posts(session)
job_posts=[serialize_job_post(p) for p in posts]
try:
result=await run_agent(subject=subject,resume_text=text,job_posts=job_posts)
except Exception as exc:
logger.exception("agent failed for %s",record_id)
raise RuntimeError(f"agent matching failed: {exc}") from exc
result=await run_agent(subject=subject,resume_text=text,job_posts=job_posts)
status=result.get("status") or "failed"
error=result.get("error") or ""
if status=="failed":
raise RuntimeError(error or "agent returned failed status")
raise RuntimeError(result.get("error") or "agent returned failed status")
async with session_scope() as session:
await Inbox_Messages.set_match_result(
session,
record_id,
resume_text=text,
experience=result.get("experience") or "",
suggested_job_post_ids=result.get("suggested_job_post_ids") or [],
summary=result.get("summary") or "",
reasoning=result.get("reasoning") or "",
status=status,
error=error,
error=result.get("error") or "",
)
logger.info("matched inbox %s status=%s ids=%s",record_id,status,result.get("suggested_job_post_ids"))
return {
"status":status,
"suggested_job_post_ids":result.get("suggested_job_post_ids") or [],
}
return {"status":status,"suggested_job_post_ids":result.get("suggested_job_post_ids") or []}

27
backend/inbox/views.py
View File

@ -9,6 +9,7 @@ from inbox.plugins import (
EMAIL_API_TOKEN,
fetch_message_read_status,
load_message_files,
request_email_confirmation,
)
from dotenv import load_dotenv
load_dotenv()
@ -24,6 +25,7 @@ class Email:
self.get_url=os.getenv("EMAIL_URL")
self.token=token or EMAIL_API_TOKEN
self.pending_match_ids:list[str]=[]
self.pending_confirmation_emails:list[str]=[]
# async def get_all_applications(self,app_id=None):
# try:
@ -58,14 +60,12 @@ class Email:
if response.status_code==200:
data=response.json()
re_create_file=await decode_attachment(data.get("attachments"))
insert_func=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file)
if (
insert_func.attachment
and insert_func.file_path
and insert_func.match_status is None
):
self.pending_match_ids.append(str(insert_func.id))
return response.json()
row,new_user_email=await Inbox_Messages.insert_email(session=self.session,email_data=data,file_path=re_create_file)
if row.attachment and row.file_path and row.match_status is None:
self.pending_match_ids.append(str(row.id))
if new_user_email:
self.pending_confirmation_emails.append(new_user_email)
return data
else:
raise HTTPException(status_code=response.status_code,detail=response.text)
except Exception as e:
@ -128,6 +128,17 @@ class Email:
task_ids.append(task.task_id)
return task_ids
async def send_account_setup(self,emails):
results=[]
for email in emails or []:
try:
status=await request_email_confirmation(email)
results.append({"email":email,"sent":status==200})
except Exception as e:
logger.warning("confirmation request failed for %s: %s",email,e)
results.append({"email":email,"sent":False})
return results
async def count_inbox_messages(self,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED):
if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED:
return await Inbox_Messages.count_inbox_messages(self.session,search,application_status=application_status)

19
backend/job/app.py
View File

@ -101,3 +101,22 @@ async def buffer_channels(
raise
except Exception as e:
raise HTTPException(status_code=500,detail=str(e))
# @router.get("/candidate/fetch")
# async def fetch_candidate(
# user_id:str=Query(None),
# current_user: dict = Depends(require_permission(PermissionTag.CANDIDATES_VIEW)),
# session: AsyncSession = Depends(get_session),
# ):
# try:
# service=CandidateView(session=session)
# if user_id:
# data=await service.get_candidate(user_id=user_id)
# else:
# data=await service.get_candidate()
# return JSONResponse(content={"data":data,"status_code":200})
# except HTTPException:
# raise
# except Exception as e:
# raise HTTPException(status_code=500,detail=str(e))

11
backend/job/candidate/views.py
View File

@ -4,6 +4,7 @@ from datetime import datetime,timezone
from fastapi import HTTPException
from pypdf import PdfReader
from sqlalchemy import select
from sqlmodel import true
from inbox.models import Inbox_Messages
from job.candidate.plugins import normalize_spaced_text
@ -66,3 +67,13 @@ class FileRead:
# try:
# get_subject=Inbox_Messages.candidate_x_inbox(self.session,self.candidate_id)
# get_file=
# class CandidateView:
# def __init__(self,session:AsyncSession):
# self.session=session
# async def get_candidate(self,user_id=None):
# try:
# call_func=Inbox_Messages.get_candidate_profile(user_id=user_id)
# except Exception as e:
# raise HTTPException(status_code=500,detail=str(e))