from sqlalchemy.ext.asyncio import AsyncSession import base64,io,logging,os,uuid from datetime import datetime,timezone from dotenv import load_dotenv from fastapi import HTTPException from pypdf import PdfReader from sqlalchemy import select from sqlalchemy.orm import selectinload from sqlmodel import true from job.job_post.models import JobPosts from job.job_post.serializers import serialize_job_post from job.candidate.serializers import serialize_candidate_profile,serialize_manual_upload_candidate from job.candidate.models import Notes,Manual_UPLOAD_CANDIDATE from job.notes.serializers import serialize_note from inbox.models import Inbox_Messages,Inbox from job.candidate.plugins import extract_candidate_email,normalize_spaced_text load_dotenv() logger=logging.getLogger("job.candidate.views") CV_QUEUE_NAME=os.getenv("TASKIQ_CV_QUEUE_NAME","cv_upload") MANUAL_UPLOAD_TO_ADDRESS=os.getenv( "MANUAL_UPLOAD_TO_ADDRESS","manual-cv-upload@hr-ats.local" ) class FileRead: def __init__(self,session:AsyncSession,filename=None,file=None): self.session=session self.filename=filename self.file=file async def read_file(self,file=None,filename=None): try: reader = PdfReader(io.BytesIO(self.file)) if reader.is_encrypted: raise HTTPException(400, "PDF is password protected") pages = [(page.extract_text() or "") for page in reader.pages] return { "filename": self.filename, "num_pages": len(reader.pages), "text": normalize_spaced_text("\n".join(pages)), } except HTTPException: raise except Exception as e: raise HTTPException(400, str(e)) async def injest_manual_upload(self): try: parsed=await self.read_file() text=(parsed.get("text") or "").strip() if not text: raise HTTPException(status_code=400,detail="No usable text could be extracted from the PDF") return parsed except HTTPException: raise except Exception as e: raise HTTPException(status_code=400,detail=str(e)) async def ingest_upload(self,candidate_email=None,candidate_name=None): """Persist a recruiter-uploaded CV with full email-ingestion parity.""" from inbox.file_decoder import AttachmentDecodeError,decode_attachment from inbox.cv_tasks import match_uploaded_cv from inbox.views import Email parsed=await self.read_file() text=parsed.get("text") or "" detected,emails_found=extract_candidate_email(text) supplied=(candidate_email or "").strip().lower() or None email=supplied or detected email_source="recruiter" if supplied else ("cv" if detected else None) if not email: raise HTTPException( status_code=422, detail={ "error_code":"CANDIDATE_EMAIL_REQUIRED", "filename":parsed.get("filename"), "num_pages":parsed.get("num_pages"), "emails_found":emails_found, "text":text, }, ) filename=self.filename or "resume.pdf" try: paths=await decode_attachment([{ "name":filename, "contentBytes":base64.b64encode(self.file).decode("ascii"), }]) except AttachmentDecodeError as e: raise HTTPException(status_code=400,detail=str(e)) if not paths: raise HTTPException(status_code=400,detail="attachment could not be saved") now=datetime.now(timezone.utc).isoformat() email_data={ "id":f"manual-cv:{uuid.uuid4()}", "subject":f"Manual CV upload — {filename}", "body":{"content":"","contentType":"text"}, "hasAttachments":True, "attachments":[{"name":filename}], "from":{"emailAddress":{"address":email,"name":(candidate_name or "").strip()}}, "toRecipients":[{"emailAddress":{"address":MANUAL_UPLOAD_TO_ADDRESS}}], "ccRecipients":[], "bccRecipients":[], "replyTo":[], "isRead":False, "sentDateTime":now, "receivedDateTime":now, } row,new_user_email=await Inbox_Messages.insert_email( self.session,email_data,file_path=paths, ) created_at=datetime.now(timezone.utc).isoformat() task=await match_uploaded_cv.kicker().with_labels( created_at=created_at, correlation_id=str(row.id), queue=CV_QUEUE_NAME, ).kiq(str(row.id),force=False) account_setup=None if new_user_email: try: account_setup=await Email(session=self.session).send_account_setup( [new_user_email] ) except Exception as e: logger.warning("account setup mail failed for %s: %s",new_user_email,e) account_setup=[{"email":new_user_email,"sent":False}] return { "queued":True, "inbox_message_id":str(row.id), "task_id":task.task_id, "filename":parsed.get("filename"), "num_pages":parsed.get("num_pages"), "candidate_email":email, "email_source":email_source, "account_setup":account_setup, "text":text, } async def match_inbox_cv(self,inbox_message_id): from inbox.plugins import resolve_attachment_path from inbox.tasks import match_inbox_message row=await Inbox_Messages.get_inbox_message_by_id(self.session,inbox_message_id) if not row: raise HTTPException(status_code=404,detail="Message not found") if not row.attachment or not row.file_path: raise HTTPException(status_code=400,detail="your file isnt in the system") found=None for path_str in (p.strip() for p in row.file_path.split(",") if p.strip()): path=resolve_attachment_path(path_str) if path.is_file(): found=path break if found is None: raise HTTPException(status_code=400,detail="your file isnt in the system") created_at=datetime.now(timezone.utc).isoformat() task=await match_inbox_message.kicker().with_labels( created_at=created_at, correlation_id=str(row.id), queue="inbox", ).kiq(str(row.id),force=True) file_name=(row.file_name or "").split(",")[0].strip() or found.name return { "queued":True, "inbox_message_id":str(row.id), "file_name":file_name, "task_id":task.task_id, } # async def get_intention(self,input): # 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 create_candidate(self,candidate_email=None,candidate_name=None,candidate_phone=None,job_post_id=None,current_company=None,platform=None,experience=None,status=None,full_text=None,current_user=None): try: email=(candidate_email or "").strip().lower() if not email: raise HTTPException(status_code=422,detail="candidate_email is required") if not current_user: raise HTTPException(status_code=400,detail="created_by is required") data={ "candidate_email":email, "candidate_name":(candidate_name or "").strip(), "candidate_phone":(candidate_phone or "").strip(), "job_post_id":job_post_id, "current_company":(current_company or "").strip(), "platform":(platform or "").strip(), "experience":(experience or "").strip(), "status":(status or "").strip(), "full_text":full_text or "", "created_by":current_user, } row=await Manual_UPLOAD_CANDIDATE.create_manual_upload_candidate(session=self.session,fields=data) return serialize_manual_upload_candidate(row) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500,detail=str(e)) async def get_candidate(self,user_id=None,limit=10,offset=0,search=None): try: detail=bool(user_id) # Detail mode must see every application for the candidate, not one page. fetch_limit=1000 if detail else limit rows=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=fetch_limit,offset=offset,search=search) if detail: return await self.attach_profile_detail(rows) return await self.attach_job_posts(rows) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500,detail=str(e)) async def count_candidates(self,user_id=None,search=None): try: return await Inbox.count_candidate_profiles(session=self.session,user_id=user_id,search=search) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500,detail=str(e)) async def update_candidate(self,user_id,payload): try: if not user_id: raise HTTPException(status_code=400,detail="user_id is required") fields={k:v for k,v in (payload or {}).items() if k in ("favorite","rating") and v is not None} if not fields: raise HTTPException(status_code=400,detail="favorite or rating is required") links=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=100,offset=0) records=links if isinstance(links,list) else ([links] if links else []) if not records: raise HTTPException(status_code=404,detail="Candidate not found") for link in records: await Inbox.update_inbox(self.session,link.id,fields) refreshed=await Inbox.get_candidate_profile(session=self.session,user_id=user_id,limit=100,offset=0) return await self.attach_profile_detail(refreshed) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500,detail=str(e)) 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 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.""" single=not isinstance(data,list) records=[data] if single else list(data or []) enriched=[] for record in records: payload=serialize_candidate_profile(record) payload["job_posts"]=[] payload["assigned_job_post"]=None 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) 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 async def attach_profile_detail(self,data): """Detail mode: flatten child collections across every Inbox row for the candidate.""" single=not isinstance(data,list) records=[data] if single else list(data or []) if not records: return {} if single else [] interviews=[] activity=[] feedback=[] documents=[] job_posts=[] assigned_job_post=None base=None user_id=None favorite=None rating=None for record in records: payload=serialize_candidate_profile(record,detail=True) if base is None: base=payload user_id=payload.get("user_id") favorite=payload.get("favorite") rating=payload.get("rating") interviews.extend(payload.get("interviews") or []) activity.extend(payload.get("activity") or []) feedback.extend(payload.get("feedback") or []) documents.extend(payload.get("documents") or []) if payload.get("assigned_job_post_id") and assigned_job_post is None: await self.get_job_post_by_id( record_id=payload.get("assigned_job_post_id"), data=payload, as_assigned=True, ) assigned_job_post=payload.get("assigned_job_post") if base.get("recruiter") is None and payload.get("recruiter"): base["recruiter"]=payload.get("recruiter") base["recruiter_id"]=payload.get("recruiter_id") if base.get("job_title") is None and payload.get("job_title"): base["job_title"]=payload.get("job_title") for job_id in payload.get("suggested_job_post_ids") or []: await self.get_job_post_by_id(record_id=job_id,data=payload) for jp in payload.get("job_posts") or []: if not any(x.get("id")==jp.get("id") for x in job_posts): job_posts.append(jp) if base.get("recruiter") is None and payload.get("recruiter"): base["recruiter"]=payload.get("recruiter") base["recruiter_id"]=payload.get("recruiter_id") if base.get("job_title") is None and payload.get("job_title"): base["job_title"]=payload.get("job_title") notes=[] uid=Notes._as_uuid(user_id) if user_id else None if uid is not None: result=await self.session.execute( select(Notes) .options(selectinload(Notes.author)) .where(Notes.user_id==uid) .order_by(Notes.created_at.desc()) ) notes=[serialize_note(r) for r in result.scalars().all()] activity.sort(key=lambda r:(r.get("activity_date") or ""),reverse=True) base["favorite"]=favorite base["rating"]=rating base["interviews"]=interviews base["activity"]=activity base["feedback"]=feedback base["documents"]=documents base["notes"]=notes base["job_posts"]=job_posts or base.get("job_posts") or [] base["assigned_job_post"]=assigned_job_post if assigned_job_post: base["assigned_job_post_id"]=assigned_job_post.get("id") if assigned_job_post.get("created_by_name"): base["recruiter"]=assigned_job_post.get("created_by_name") base["recruiter_id"]=assigned_job_post.get("created_by") if assigned_job_post.get("title"): base["job_title"]=assigned_job_post.get("title") return base