pipeline done

pull/14/head
ahmed.mujtaba 2026-08-13 13:29:02 +05:00
parent 3b7e83a4a7
commit 4f9b03bd4f
7 changed files with 153 additions and 63 deletions

3
.gitignore vendored
View File

@ -1,4 +1,5 @@
# macOS
migrations/** */
.DS_Store
.AppleDouble
.LSOverride

View File

@ -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

View File

@ -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."""

View File

@ -215,4 +215,45 @@ class JobPosts(SQLModel, table=True):
await session.refresh(row)
return row
import users.models as _users_models
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

View File

@ -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 aliasBuffer-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(

View File

@ -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)

View File

@ -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;