From 4f9b03bd4f28f58636f1add38a2e7c9b08295ad3 Mon Sep 17 00:00:00 2001 From: "ahmed.mujtaba" Date: Thu, 13 Aug 2026 13:29:02 +0500 Subject: [PATCH] pipeline done --- .gitignore | 3 +- backend/job/app.py | 6 +- backend/job/candidate/views.py | 90 ++++++++++++++----- backend/job/job_post/models.py | 43 ++++++++- backend/job/job_post/plugins.py | 41 +++------ backend/job/job_post/views.py | 11 +-- .../manual/003_seed_social_platforms.sql | 22 +++++ 7 files changed, 153 insertions(+), 63 deletions(-) create mode 100644 backend/migrations/manual/003_seed_social_platforms.sql diff --git a/.gitignore b/.gitignore index 7d206f3..8d92c69 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ -# macOS + +migrations/** */ .DS_Store .AppleDouble .LSOverride diff --git a/backend/job/app.py b/backend/job/app.py index a880fe8..cfc8b0e 100644 --- a/backend/job/app.py +++ b/backend/job/app.py @@ -15,7 +15,7 @@ from users.permissions import PermissionTag, require_permission from job.job_post.views import JobPost,JobPostCreate import logging from users.views import User -from job.job_post.plugins import PlatformAlias +from job.job_post.models import SocialPlatform from fastapi import UploadFile, File, Form from dotenv import load_dotenv from datetime import datetime, time, timezone @@ -116,9 +116,9 @@ class HiringCostCreate(BaseModel): @router.get("/jobs/alias") -async def get_job_alias(): +async def get_job_alias(session: AsyncSession = Depends(get_session)): try: - alias_lst=[k.name for k in PlatformAlias] + alias_lst=await SocialPlatform.list_aliases(session) return JSONResponse(content={"data":alias_lst,"status_code":200}) except HTTPException: raise diff --git a/backend/job/candidate/views.py b/backend/job/candidate/views.py index b9f0cb1..e2bfb33 100644 --- a/backend/job/candidate/views.py +++ b/backend/job/candidate/views.py @@ -667,38 +667,72 @@ class CandidateView: except Exception as e: raise HTTPException(status_code=500,detail=str(e)) + @staticmethod + def _attach_job_post(data,payload,*,as_assigned=False): + """Merge one serialized job post onto a candidate payload. + + Pure, so the per-id path and the batched one below cannot drift. A missing + post attaches nothing, matching the old early return. + """ + if not isinstance(data,dict) or not payload: + return payload + if as_assigned: + data["assigned_job_post"]=payload + if payload.get("created_by_name"): + data["recruiter"]=payload.get("created_by_name") + data["recruiter_id"]=payload.get("created_by") + if payload.get("title"): + data["job_title"]=payload.get("title") + else: + data.setdefault("job_posts",[]).append(payload) + if data.get("recruiter") is None and payload.get("created_by_name"): + data["recruiter"]=payload.get("created_by_name") + data["recruiter_id"]=payload.get("created_by") + if data.get("job_title") is None and payload.get("title"): + data["job_title"]=payload.get("title") + return payload + + async def _job_posts_by_id(self,ids): + """Serialized job posts keyed by id — one query for a whole page of rows. + + active_only=False mirrors get_job_post_by_id, which filters neither flag: + an application assigned to a closed post must keep its title. + """ + wanted=[] + seen=set() + for raw in ids or []: + key=str(raw) if raw else None + if not key or key in seen: + continue + seen.add(key) + wanted.append(key) + if not wanted: + return {} + rows=await JobPosts.get_by_ids(self.session,wanted,active_only=False) + return {str(r.id):serialize_job_post(r) for r in rows} + async def get_job_post_by_id(self,record_id,data=None,*,as_assigned=False): """Load full job_posts row and optionally append it onto a candidate payload.""" try: job_post_data=await JobPosts.get_job_post_by_id(session=self.session,record_id=record_id) if not job_post_data: return None - payload=serialize_job_post(job_post_data) - if isinstance(data,dict): - if as_assigned: - data["assigned_job_post"]=payload - if payload.get("created_by_name"): - data["recruiter"]=payload.get("created_by_name") - data["recruiter_id"]=payload.get("created_by") - if payload.get("title"): - data["job_title"]=payload.get("title") - else: - data.setdefault("job_posts",[]).append(payload) - if data.get("recruiter") is None and payload.get("created_by_name"): - data["recruiter"]=payload.get("created_by_name") - data["recruiter_id"]=payload.get("created_by") - if data.get("job_title") is None and payload.get("title"): - data["job_title"]=payload.get("title") - return payload + return self._attach_job_post(data,serialize_job_post(job_post_data),as_assigned=as_assigned) except Exception as e: raise HTTPException(status_code=500,detail=str(e)) async def attach_job_posts(self,data): - """Normalize list/single, serialize each record, attach full job_posts rows.""" + """Normalize list/single, serialize each record, attach full job_posts rows. + + Two batched queries per page, not two per row — a 200-row pipeline board + issued 600+ sequential job_posts round trips before. + """ single=not isinstance(data,list) records=[data] if single else list(data or []) scores=await self._scores_by_message([getattr(r,"message_id",None) for r in records]) - enriched=[] + + payloads=[] + wanted=[] for record in records: payload=serialize_candidate_profile(record) payload["job_posts"]=[] @@ -707,13 +741,23 @@ class CandidateView: if scored: payload["ai_score"]=scored[0].match_score payload["recommendation"]=self._recommendation(scored[0].match_score) + payloads.append(payload) + if payload.get("assigned_job_post_id"): + wanted.append(payload["assigned_job_post_id"]) + wanted.extend(payload.get("suggested_job_post_ids") or []) + + posts=await self._job_posts_by_id(wanted) + # Copy per attach: two candidates on the same post held independent dicts + # back when every row re-serialized its own. + for payload in payloads: assigned_id=payload.get("assigned_job_post_id") if assigned_id: - await self.get_job_post_by_id(record_id=assigned_id,data=payload,as_assigned=True) + post=posts.get(str(assigned_id)) + self._attach_job_post(payload,dict(post) if post else None,as_assigned=True) for job_id in payload.get("suggested_job_post_ids") or []: - await self.get_job_post_by_id(record_id=job_id,data=payload) - enriched.append(payload) - return enriched[0] if single else enriched + post=posts.get(str(job_id)) + self._attach_job_post(payload,dict(post) if post else None) + return payloads[0] if single else payloads async def attach_profile_detail(self,data): """Detail mode: flatten child collections across every Inbox row for the candidate.""" diff --git a/backend/job/job_post/models.py b/backend/job/job_post/models.py index 0d954ab..54be9a6 100644 --- a/backend/job/job_post/models.py +++ b/backend/job/job_post/models.py @@ -215,4 +215,45 @@ class JobPosts(SQLModel, table=True): await session.refresh(row) return row -import users.models as _users_models \ No newline at end of file + +class SocialPlatform(SQLModel, table=True): + """Buffer publish aliases (fb → facebook). Spelling tolerance + UI list, not an allowlist.""" + + __tablename__ = "social_platforms" + + id: int | None = Field(default=None, primary_key=True) + alias: str = Field(max_length=40, unique=True, index=True) + buffer_service: str + label: str + is_active: bool = Field(default=True) + created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) + updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) + + @classmethod + async def get_by_alias(cls, session: AsyncSession, alias: str): + key = (alias or "").strip().lower() + if not key: + return None + result = await session.execute(select(cls).where(cls.alias == key)) + return result.scalars().first() + + @classmethod + async def list_active(cls, session: AsyncSession): + result = await session.execute( + select(cls).where(cls.is_active == True).order_by(cls.id.asc()) # noqa: E712 + ) + return list(result.scalars().all()) + + @classmethod + async def list_aliases(cls, session: AsyncSession): + rows = await cls.list_active(session) + return [r.alias for r in rows] + + @classmethod + async def alias_map(cls, session: AsyncSession) -> dict[str, str]: + """alias → Buffer service name for normalize_platform / resolve_channel.""" + rows = await cls.list_active(session) + return {r.alias: r.buffer_service for r in rows} + + +import users.models as _users_models \ No newline at end of file diff --git a/backend/job/job_post/plugins.py b/backend/job/job_post/plugins.py index b19fa26..bae385c 100644 --- a/backend/job/job_post/plugins.py +++ b/backend/job/job_post/plugins.py @@ -9,7 +9,6 @@ import json import os import re from datetime import datetime -from enum import Enum import httpx from dotenv import load_dotenv @@ -40,38 +39,20 @@ def local_status(buffer_status) -> str: return BUFFER_STATUS_TO_LOCAL.get(buffer_status or "", "scheduled") -class PlatformAlias(str, Enum): - """Shorthands people type, mapped to Buffer's own `Service` values. +def normalize_platform(value, aliases=None) -> str: + """Case/punctuation-insensitive key for comparing a requested platform. - Member name = what arrives in the request, value = what Buffer calls it. This is - spelling tolerance only, *not* the list of supported networks: anything absent here - still resolves, because `resolve_channel` matches against the services Buffer - actually reports. A newly connected network needs no entry. - - Names that share a value (ig/insta) become Enum aliases, which is exactly the - intent -- lookup is by member name via `__members__`. + `aliases` is an optional alias→Buffer-service map (from SocialPlatform). Missing + keys fall through to the normalized key so Buffer's live channel list remains the + real allowlist — DB rows are spelling tolerance only. """ - - fb = "facebook" - ig = "instagram" - insta = "instagram" - li = "linkedin" - x = "twitter" - tweet = "twitter" - yt = "youtube" - gbp = "googlebusiness" - google = "googlebusiness" - googlebusinessprofile = "googlebusiness" - - -def normalize_platform(value) -> str: - """Case/punctuation-insensitive key for comparing a requested platform.""" key = re.sub(r"[^a-z0-9]", "", str(value or "").lower()) - alias = PlatformAlias.__members__.get(key) - return alias.value if alias else key + if aliases: + return aliases.get(key, key) + return key -async def resolve_channel(platform, channels=None) -> dict: +async def resolve_channel(platform, channels=None, aliases=None) -> dict: """Return the connected channel for `platform`. Nothing is special-cased per network: the request is matched against the services @@ -79,14 +60,14 @@ async def resolve_channel(platform, channels=None) -> dict: resolve. Falls back to matching the channel's handle or display name so "ahmedmujtababaig" works as well as "linkedin". """ - wanted = normalize_platform(platform) + wanted = normalize_platform(platform, aliases) if not wanted: raise BufferError("No platform given") if channels is None: channels = await list_buffer_channels() for field in ("service", "name", "displayName"): for channel in channels: - if normalize_platform(channel.get(field)) == wanted: + if normalize_platform(channel.get(field), aliases) == wanted: return channel available = sorted({c.get("service") for c in channels if c.get("service")}) raise BufferError( diff --git a/backend/job/job_post/views.py b/backend/job/job_post/views.py index 7a3a271..a880367 100644 --- a/backend/job/job_post/views.py +++ b/backend/job/job_post/views.py @@ -6,7 +6,7 @@ from dotenv import load_dotenv from fastapi import HTTPException from sqlalchemy.ext.asyncio import AsyncSession from pydantic import BaseModel, model_validator -from job.job_post.models import JobPosts +from job.job_post.models import JobPosts,SocialPlatform from job.job_post.plugins import ( BufferError, create_buffer_post, @@ -56,7 +56,7 @@ class JobPost: self.buffer_api=os.getenv("BUFFER_API") self.channel_id=os.getenv("BUFFER_CHANNEL_ID") - async def _resolve_target(self,payload): + async def _resolve_target(self,payload,aliases=None): """Pick the Buffer channel to post to, and the service it belongs to. Precedence: an explicit channel_id, then the requested platform, then the @@ -66,7 +66,7 @@ class JobPost: if payload.get("channel_id"): return payload["channel_id"],None if payload.get("platform"): - channel=await resolve_channel(payload["platform"]) + channel=await resolve_channel(payload["platform"],aliases=aliases) return channel["id"],channel.get("service") if self.channel_id: return self.channel_id,None @@ -76,8 +76,9 @@ class JobPost: ) async def post_job(self,payload,current_user): + aliases=await SocialPlatform.alias_map(self.session) try: - channel_id,service=await self._resolve_target(payload) + channel_id,service=await self._resolve_target(payload,aliases) except (httpx.HTTPError,BufferError,RuntimeError) as e: raise HTTPException(status_code=502,detail=f"Failed to resolve Buffer channel: {e}") from e @@ -103,7 +104,7 @@ class JobPost: # Only set platform when it is actually known: passing None would override the # column default and break the NOT NULL constraint. Buffer's channelService # replaces this with the authoritative value once the post is created. - known_platform=service or normalize_platform(payload.get("platform")) + known_platform=service or normalize_platform(payload.get("platform"),aliases) if known_platform: fields["platform"]=known_platform row=await JobPosts.insert_job_post(self.session,fields) diff --git a/backend/migrations/manual/003_seed_social_platforms.sql b/backend/migrations/manual/003_seed_social_platforms.sql new file mode 100644 index 0000000..b37b4bb --- /dev/null +++ b/backend/migrations/manual/003_seed_social_platforms.sql @@ -0,0 +1,22 @@ +-- 003_seed_social_platforms.sql +-- Manual one-shot: seed Buffer publish aliases formerly hardcoded in +-- PlatformAlias (job/job_post/plugins.py). Spelling tolerance + /jobs/alias UI +-- list only — missing aliases still resolve via Buffer's live channel list. +-- +-- Order: (1) alembic upgrade / autogenerate so app.social_platforms exists, +-- (2) this file. Applied automatically at startup by alembic_setup.run_manual_sql() +-- once the schema is at head; recorded in manual_migrations. Safe to re-run by hand. + +INSERT INTO app.social_platforms (alias, buffer_service, label, is_active, created_at, updated_at) +VALUES + ('fb', 'facebook', 'Facebook', true, NOW(), NOW()), + ('ig', 'instagram', 'Instagram', true, NOW(), NOW()), + ('insta', 'instagram', 'Instagram', true, NOW(), NOW()), + ('li', 'linkedin', 'LinkedIn', true, NOW(), NOW()), + ('x', 'twitter', 'Twitter/X', true, NOW(), NOW()), + ('tweet', 'twitter', 'Twitter/X', true, NOW(), NOW()), + ('yt', 'youtube', 'YouTube', true, NOW(), NOW()), + ('gbp', 'googlebusiness', 'Google Business', true, NOW(), NOW()), + ('google', 'googlebusiness', 'Google Business', true, NOW(), NOW()), + ('googlebusinessprofile', 'googlebusiness', 'Google Business', true, NOW(), NOW()) +ON CONFLICT (alias) DO NOTHING;