1108 lines
52 KiB
Python
1108 lines
52 KiB
Python
import asyncio
|
|
import logging
|
|
import uuid
|
|
import httpx,os
|
|
from fastapi import HTTPException
|
|
from inbox.enums import Candidate_application_Status
|
|
from inbox.models import Inbox_Messages,Inbox_Message_Triage,MailboxSyncRun,InboxRescanRun,Inbox,SourceChannels,AtsResults
|
|
from inbox.file_decoder import extract_pdf_attachments
|
|
from inbox.serializers import serialize_application, serialize_message, serialize_triage, serialize_mailbox_sync_run, serialize_inbox_rescan_run, serialize_ats_result
|
|
from inbox.plugins import (
|
|
EMAIL_API_TOKEN,
|
|
attach_email_pdfs_to_s3,
|
|
fetch_message_read_status,
|
|
load_message_files,
|
|
request_email_confirmation,
|
|
send_mail,
|
|
)
|
|
from inbox.semaphore import GraphSemaphore
|
|
|
|
from inbox_classifier.decorators import is_manual_upload,triage_fields
|
|
from inbox_classifier.execute_agent import classify_email
|
|
from inbox_classifier.plugins import (
|
|
TRIAGE_CONCURRENCY,
|
|
TRIAGE_ENABLED,
|
|
TRIAGE_STATUSES,
|
|
sender_domain,
|
|
should_ingest,
|
|
triage_model_name,
|
|
)
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from datetime import datetime,timezone
|
|
|
|
logger=logging.getLogger("inbox.match")
|
|
triage_logger=logging.getLogger("inbox.triage")
|
|
|
|
# One statement, one round trip — but an unbounded id list is still a client-supplied
|
|
# IN () of arbitrary size, so the batch is capped and the route answers 413.
|
|
MAX_BULK_READ_IDS=100
|
|
|
|
|
|
class Email:
|
|
def __init__(self,session:AsyncSession,token=None):
|
|
self.session=session
|
|
self.get_url=os.getenv("EMAIL_URL")
|
|
self.token=token or EMAIL_API_TOKEN
|
|
self.graph_slots=GraphSemaphore(concurrency=4,max_retries=5)
|
|
self.pending_match_ids:list[str]=[]
|
|
self.pending_confirmation_emails:list[str]=[]
|
|
# Upstream ids the intake gate judged not to be job applications. They get a
|
|
# verdict row and no inbox_messages row.
|
|
self.skipped_message_ids:list[str]=[]
|
|
self.triage_errors:list[str]=[]
|
|
|
|
# async def get_all_applications(self,app_id=None):
|
|
# try:
|
|
# if app_id:
|
|
# application_lst=await Inbox_Messages.get_all_applications(self.session,message_id=app_id)
|
|
# else:
|
|
# application_lst=await Inbox_Messages.get_all_applications(self.session)
|
|
# return application_lst
|
|
# except Exception as e:
|
|
# raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
async def service_email(self,top,skip):
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
response=await self.graph_slots.get(
|
|
client,f"{self.get_url}/emails",
|
|
params={"skip":skip,"top":top},
|
|
headers={"Authorization":f"Bearer {self.token}"},
|
|
)
|
|
if response.status_code==200:
|
|
return response.json()
|
|
else:
|
|
raise HTTPException(status_code=response.status_code,detail=response.text)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
async def fetch_message(self,message_id):
|
|
"""GET /emails/{id} on the upstream Email API -> the Graph payload."""
|
|
async with httpx.AsyncClient() as client:
|
|
response=await self.graph_slots.get(
|
|
client,f"{self.get_url}/emails/{message_id}",
|
|
headers={"Authorization":f"Bearer {self.token}"},
|
|
)
|
|
if response.status_code!=200:
|
|
raise HTTPException(status_code=response.status_code,detail=response.text)
|
|
return response.json()
|
|
|
|
|
|
async def triage_round(self,message_ids):
|
|
"""Fetch and classify a whole /email/fetch page, bounded by a semaphore."""
|
|
|
|
ids=[str(m) for m in message_ids or [] if m]
|
|
decisions={}
|
|
if not ids:
|
|
return decisions
|
|
known=await Inbox_Messages.existing_message_ids(self.session,ids)
|
|
recorded=await Inbox_Message_Triage.verdicts_for_message_ids(self.session,ids)
|
|
pending=[]
|
|
for message_id in ids:
|
|
if message_id in known:
|
|
decisions[message_id]={"ingest":True,"status":"known","fresh":False}
|
|
elif message_id in recorded:
|
|
decisions[message_id]={"ingest":recorded[message_id],"status":"recorded","fresh":False}
|
|
else:
|
|
pending.append(message_id)
|
|
if not pending:
|
|
triage_logger.info("triage round: page=%s known=%s classified=0",len(ids),len(decisions))
|
|
return decisions
|
|
|
|
semaphore=asyncio.Semaphore(TRIAGE_CONCURRENCY)
|
|
|
|
async def run(message_id):
|
|
async with semaphore:
|
|
data=await self.fetch_message(message_id)
|
|
if not TRIAGE_ENABLED or is_manual_upload(data):
|
|
return message_id,{"data":data,"ingest":True,"status":"disabled","fresh":False}
|
|
verdict,error=await classify_email(data)
|
|
ingest,status,reason=should_ingest(verdict,error)
|
|
return message_id,{"data":data,"verdict":verdict,"error":error,"ingest":ingest,
|
|
"status":status,"reason":reason,"fresh":True}
|
|
|
|
results=await asyncio.gather(*(run(m) for m in pending),return_exceptions=True)
|
|
accepted=rejected=errors=0
|
|
for result in results:
|
|
if isinstance(result,BaseException):
|
|
# First failure wins, preserving today's all-or-nothing behaviour for a
|
|
# failing upstream message. Never catch BaseException itself: a
|
|
# CancelledError must keep propagating.
|
|
raise result
|
|
message_id,decision=result
|
|
decisions[message_id]=decision
|
|
if decision.get("status")=="error":
|
|
errors+=1
|
|
if decision.get("ingest"):
|
|
accepted+=1
|
|
else:
|
|
rejected+=1
|
|
triage_logger.info(
|
|
"triage round: page=%s known=%s classified=%s accepted=%s rejected=%s errors=%s",
|
|
len(ids),len(known)+len(recorded),len(pending),accepted,rejected,errors,
|
|
)
|
|
return decisions
|
|
|
|
async def record_triage(self,data,decision,ingested):
|
|
"""Persist one verdict. Never raises into the ingestion path.
|
|
|
|
A failed audit write must not cost us a candidate: the worst case is that the
|
|
next fetch re-classifies this message.
|
|
"""
|
|
try:
|
|
fields=triage_fields(
|
|
data,
|
|
decision.get("verdict"),
|
|
decision.get("status") or "classified",
|
|
decision.get("reason") or "",
|
|
error=decision.get("error") or "",
|
|
model_name=triage_model_name(),
|
|
ingested=ingested,
|
|
)
|
|
await Inbox_Message_Triage.record_verdict(self.session,fields)
|
|
# Allowlisted keys only: sender DOMAIN not address, attachment COUNT not
|
|
# names, no subject or body text, no evidence text.
|
|
verdict=decision.get("verdict")
|
|
triage_logger.info(
|
|
"triage %s: application=%s reason=%s confidence=%s domain=%s attachments=%s",
|
|
fields["message_id"],
|
|
fields["is_application"],
|
|
fields["reason_code"],
|
|
getattr(verdict,"confidence",None),
|
|
sender_domain(fields["message_from"]),
|
|
len(data.get("attachments") or []),
|
|
)
|
|
except Exception as e:
|
|
triage_logger.warning("triage record failed: %s",type(e).__name__)
|
|
|
|
async def get_email_by_id(self,message_id,test_on=True,decision=None):
|
|
"""Persist one upstream message, gated by the application classifier.
|
|
|
|
`decision` is the pre-computed verdict from triage_round; without one this
|
|
classifies inline, so a single-message call still works.
|
|
|
|
Ordering is deliberate. The verdict comes BEFORE PDF extract / S3 upload: a
|
|
rejected mail must not create a candidate Users row or queue confirmation mail.
|
|
Flow: extract PDF bytes in memory → insert inbox_messages → link sender →
|
|
upload Email/{id}/{user_id}/file.pdf → store permanent S3 URL on file_path.
|
|
If S3 fails on a brand-new row, the table entry is deleted (atomicity).
|
|
"""
|
|
try:
|
|
if decision is None:
|
|
decision=(await self.triage_round([message_id])).get(str(message_id)) or {}
|
|
data=decision.get("data") or await self.fetch_message(message_id)
|
|
|
|
if not decision.get("ingest"):
|
|
if decision.get("fresh"):
|
|
await self.record_triage(data,decision,ingested=False)
|
|
self.skipped_message_ids.append(str(message_id))
|
|
if decision.get("status")=="error":
|
|
self.triage_errors.append(str(message_id))
|
|
return {"message_id":str(message_id),"skipped":"not_application",
|
|
"reason":decision.get("reason") or "","status":decision.get("status") or ""}
|
|
|
|
upstream_id=data.get("id")
|
|
already=await Inbox_Messages.get_by_upstream_id(self.session,upstream_id) if upstream_id else None
|
|
pdfs=extract_pdf_attachments(data.get("attachments"))
|
|
# Insert first (no file_path yet) so S3 keys can use the table PK.
|
|
row,new_user_email=await Inbox_Messages.insert_email(
|
|
session=self.session,email_data=data,file_path=None,
|
|
)
|
|
await Reapplied(session=self.session).sync_for_email(row.message_from)
|
|
if pdfs:
|
|
row=await attach_email_pdfs_to_s3(
|
|
self.session,row,pdfs,created_new=(already is None),
|
|
)
|
|
if decision.get("fresh"):
|
|
await self.record_triage(data,decision,ingested=True)
|
|
if row.attachment and row.file_path and row.match_status is None:
|
|
self.pending_match_ids.append(str(row.id))
|
|
if test_on:
|
|
return data
|
|
if new_user_email:
|
|
self.pending_confirmation_emails.append(new_user_email)
|
|
return data
|
|
except HTTPException:
|
|
# Was missing: the bare `except Exception` below caught the upstream-status
|
|
# HTTPException and re-raised every one of them as a 500.
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
|
|
async def get_inbox_messages(self,top,skip,search=None):
|
|
messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search)
|
|
urls=await Inbox.linkedin_urls_by_message_ids(self.session,[m.id for m in messages])
|
|
items=[]
|
|
for m in messages:
|
|
item=serialize_message(m,linkedin_url=urls.get(m.id))
|
|
files=load_message_files(m)
|
|
if files:
|
|
item["files"]=files
|
|
items.append(item)
|
|
return items
|
|
|
|
async def get_inbox_message_by_id(self,record_id):
|
|
message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
|
|
if not message:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
urls=await Inbox.linkedin_urls_by_message_ids(self.session,[message.id])
|
|
item=serialize_message(message,linkedin_url=urls.get(message.id))
|
|
files=load_message_files(message)
|
|
if files:
|
|
item["files"]=files
|
|
from job.candidate.views import CandidateView
|
|
cv=CandidateView(session=self.session)
|
|
items=await self._attach_job_posts([item])
|
|
return await cv.attach_application_history(items[0])
|
|
|
|
async def get_all_applications(self,top,skip,search=None,isread:bool=True,application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,assigned=None,is_duplicate=None,no_suggestions=None,processing_state=None,city=None,source=None):
|
|
if application_status in (Candidate_application_Status.PROCESS, Candidate_application_Status.REJECTED, Candidate_application_Status.SCREENING, Candidate_application_Status.ASSESSMENT, Candidate_application_Status.INTERVIEW, Candidate_application_Status.OFFER, Candidate_application_Status.HIRED) or processing_state:
|
|
messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source,light=True)
|
|
elif isread==False:
|
|
messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,isread,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source,light=True)
|
|
else:
|
|
messages=await Inbox_Messages.get_inbox_messages(self.session,top,skip,search,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source,light=True)
|
|
urls=await Inbox.linkedin_urls_by_message_ids(self.session,[m.id for m in messages])
|
|
items=[serialize_application(m,linkedin_url=urls.get(m.id),light=True) for m in messages]
|
|
items=await self._attach_job_posts(items)
|
|
from job.candidate.views import CandidateView
|
|
return await CandidateView(session=self.session).attach_application_history(items)
|
|
|
|
async def _attach_job_posts(self,items):
|
|
"""List payload needs assigned + suggested titles — export reads names.
|
|
|
|
Detail hydrates one row. Full JD (location, requirements) loads when
|
|
the recruiter expands a card. Assigned job and Suggested jobs in the
|
|
Inbox .xlsx stay as titles.
|
|
"""
|
|
ids=[]
|
|
for item in items:
|
|
aid=item.get("assigned_job_post_id")
|
|
if aid:
|
|
ids.append(aid)
|
|
for sid in item.get("suggested_job_post_ids") or []:
|
|
if sid:
|
|
ids.append(sid)
|
|
by_id={}
|
|
if ids:
|
|
from job.job_post.models import JobPosts
|
|
from job.job_post.serializers import serialize_job_post_title
|
|
for post in await JobPosts.titles_by_ids(self.session,ids,active_only=False):
|
|
payload=serialize_job_post_title(post)
|
|
if post.is_deleted or not post.is_active:
|
|
payload={**payload,"unavailable":True}
|
|
by_id[str(post.id)]=payload
|
|
for item in items:
|
|
aid=item.get("assigned_job_post_id")
|
|
item["assigned_job_post"]=by_id.get(str(aid)) if aid else None
|
|
suggested=[]
|
|
for sid in item.get("suggested_job_post_ids") or []:
|
|
if not sid:
|
|
continue
|
|
payload=by_id.get(str(sid))
|
|
if payload is None:
|
|
suggested.append({"id":str(sid),"unavailable":True})
|
|
else:
|
|
suggested.append(dict(payload))
|
|
item["suggested_job_posts"]=suggested
|
|
items=await self._paint_inbox_ats(items)
|
|
return items
|
|
|
|
async def _paint_inbox_ats(self,items):
|
|
"""Attach per-job ATS scores onto suggested/assigned posts and stamp max.
|
|
|
|
Unassigned rows show the highest suggestion score; assigned rows show
|
|
the score against assigned_job_post_id — same rule as Sheet Forms.
|
|
"""
|
|
if not items:
|
|
return items
|
|
by_msg=await AtsResults.get_latest_by_job_for_messages(
|
|
self.session,[item.get("id") for item in items],
|
|
)
|
|
for item in items:
|
|
rows=by_msg.get(str(item.get("id") or "")) or []
|
|
scores=[serialize_ats_result(r) for r in rows]
|
|
item["ats_results"]=scores
|
|
score_by_job={
|
|
str(s["job_post_id"]):s for s in scores if s.get("job_post_id")
|
|
}
|
|
for post in item.get("suggested_job_posts") or []:
|
|
hit=score_by_job.get(str(post.get("id")))
|
|
if hit:
|
|
post["overall_score"]=hit.get("overall_score")
|
|
post["band"]=hit.get("band")
|
|
assigned=item.get("assigned_job_post")
|
|
if assigned:
|
|
hit=score_by_job.get(str(assigned.get("id")))
|
|
if hit:
|
|
assigned["overall_score"]=hit.get("overall_score")
|
|
assigned["band"]=hit.get("band")
|
|
aid=item.get("assigned_job_post_id")
|
|
assigned_score=score_by_job.get(str(aid)) if aid else None
|
|
if assigned_score and assigned_score.get("overall_score") is not None:
|
|
item["ats_score"]=round(float(assigned_score.get("overall_score")))
|
|
else:
|
|
nums=[s.get("overall_score") for s in scores if s.get("overall_score") is not None]
|
|
if nums:
|
|
item["ats_score"]=round(float(max(nums)))
|
|
return items
|
|
|
|
async def get_application_by_id(self,record_id):
|
|
message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
|
|
if not message:
|
|
raise HTTPException(status_code=404,detail="Application not found")
|
|
urls=await Inbox.linkedin_urls_by_message_ids(self.session,[message.id])
|
|
item=serialize_application(message,linkedin_url=urls.get(message.id))
|
|
from job.candidate.views import CandidateView
|
|
return await CandidateView(session=self.session).attach_application_history(item)
|
|
|
|
async def queue_rematch(self,record_id):
|
|
message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
|
|
if not message:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
if not message.attachment or not message.file_path:
|
|
raise HTTPException(status_code=400,detail="Message has no attachment to match")
|
|
return str(message.id)
|
|
|
|
async def enqueue_matching(self,inbox_ids,force=False):
|
|
from inbox.tasks import match_inbox_message
|
|
task_ids=[]
|
|
for record_id in inbox_ids or []:
|
|
created_at=datetime.now(timezone.utc).isoformat()
|
|
task=await match_inbox_message.kicker().with_labels(
|
|
created_at=created_at,
|
|
correlation_id=str(record_id),
|
|
queue="inbox",
|
|
).kiq(str(record_id),force=force)
|
|
task_ids.append(task.task_id)
|
|
return task_ids
|
|
|
|
async def start_mailbox_sync(self,current_user=None,top=100,skip=0,test_on=True):
|
|
"""Enqueue Outlook pull on the mailbox_sync queue; return the run row.
|
|
|
|
If a queued/running sync already exists, return it instead of stacking another.
|
|
"""
|
|
active=await MailboxSyncRun.get_active(self.session)
|
|
if active:
|
|
return serialize_mailbox_sync_run(active)
|
|
|
|
created_by=None
|
|
if isinstance(current_user,dict) and current_user.get("id"):
|
|
created_by=MailboxSyncRun._as_uuid(current_user.get("id"))
|
|
|
|
row=await MailboxSyncRun.insert_run(self.session,{
|
|
"status":"queued",
|
|
"created_by":created_by,
|
|
"top":int(top or 100),
|
|
"skip":int(skip or 0),
|
|
"test_on":bool(test_on) if test_on is not None else True,
|
|
})
|
|
|
|
from inbox.mailbox_sync_tasks import sync_mailbox
|
|
task=await sync_mailbox.kicker().with_labels(
|
|
created_at=datetime.now(timezone.utc).isoformat(),
|
|
correlation_id=str(row.id),
|
|
queue="mailbox_sync",
|
|
).kiq(str(row.id))
|
|
row=await MailboxSyncRun.update_run(self.session,row.id,{"task_id":task.task_id})
|
|
return serialize_mailbox_sync_run(row)
|
|
|
|
async def get_mailbox_sync(self,run_id=None):
|
|
if run_id:
|
|
row=await MailboxSyncRun.get_by_id(self.session,run_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="Sync run not found")
|
|
return serialize_mailbox_sync_run(row)
|
|
row=await MailboxSyncRun.get_active(self.session)
|
|
if row:
|
|
return serialize_mailbox_sync_run(row)
|
|
# Latest finished run so the UI can still show the last result after refresh.
|
|
row=await MailboxSyncRun.get_latest(self.session)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="No sync runs yet")
|
|
return serialize_mailbox_sync_run(row)
|
|
|
|
async def start_on_hold_rescan(self,channel="all",sheet=None,current_user=None):
|
|
"""Queue an On-Hold catalogue ATS scan against every job_posts row.
|
|
|
|
Returns an existing queued/running/scoring run instead of stacking another.
|
|
"""
|
|
channel=(channel or "all").strip().lower()
|
|
if channel not in ("all","email","forms"):
|
|
raise HTTPException(status_code=422,detail="channel must be all, email, or forms")
|
|
sheet=(sheet or "").strip() or None
|
|
if channel!="forms":
|
|
sheet=None
|
|
|
|
active=await InboxRescanRun.get_active(self.session)
|
|
if active:
|
|
return serialize_inbox_rescan_run(active)
|
|
|
|
created_by=None
|
|
if isinstance(current_user,dict) and current_user.get("id"):
|
|
created_by=InboxRescanRun._as_uuid(current_user.get("id"))
|
|
row=await InboxRescanRun.insert_run(self.session,{
|
|
"status":"queued",
|
|
"channel":channel,
|
|
"sheet":sheet,
|
|
"created_by":created_by,
|
|
})
|
|
from inbox.tasks import rescan_on_hold_run
|
|
try:
|
|
task=await rescan_on_hold_run.kicker().with_labels(
|
|
created_at=datetime.now(timezone.utc).isoformat(),
|
|
correlation_id=str(row.id),
|
|
queue="inbox",
|
|
).kiq(str(row.id),0)
|
|
except Exception as exc:
|
|
await InboxRescanRun.update_run(self.session,row.id,{
|
|
"status":"failed",
|
|
"error":str(exc),
|
|
"finished_at":datetime.now(timezone.utc),
|
|
})
|
|
raise HTTPException(status_code=503,detail="Could not queue On-Hold rescan") from exc
|
|
row=await InboxRescanRun.update_run(self.session,row.id,{"task_id":task.task_id})
|
|
return serialize_inbox_rescan_run(row)
|
|
|
|
async def get_on_hold_rescan(self,run_id=None):
|
|
if run_id:
|
|
row=await InboxRescanRun.get_by_id(self.session,run_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="Rescan run not found")
|
|
return serialize_inbox_rescan_run(row)
|
|
row=await InboxRescanRun.get_active(self.session)
|
|
if row:
|
|
return serialize_inbox_rescan_run(row)
|
|
row=await InboxRescanRun.get_latest(self.session)
|
|
if not row:
|
|
return None
|
|
return serialize_inbox_rescan_run(row)
|
|
|
|
async def plan_on_hold_pairs(self,channel,sheet=None):
|
|
"""Build (candidate, job) pairs that have never been ATS-scored.
|
|
|
|
Skip a candidate who already has a score against an active job.
|
|
Skip a pair that already exists on ats_results for that person/row.
|
|
"""
|
|
from g_sheet.models import FormData
|
|
from job.job_post.models import JobPosts
|
|
|
|
job_ids=[str(jid) for jid in await JobPosts.list_ids(self.session)]
|
|
active_ids={str(jid) for jid in await JobPosts.list_ids(self.session,active_only=True)}
|
|
inbox_rows=[]
|
|
form_rows=[]
|
|
if channel in ("all","email"):
|
|
inbox_rows=await Inbox_Messages.list_on_hold_scan_rows(self.session)
|
|
if channel in ("all","forms"):
|
|
form_rows=await FormData.list_on_hold_scan_rows(self.session,sheet=sheet)
|
|
|
|
emails=[]
|
|
for row in inbox_rows:
|
|
if row.get("email"):
|
|
emails.append(row["email"])
|
|
for row in form_rows:
|
|
if row.get("email"):
|
|
emails.append(row["email"])
|
|
scored_by_email=await AtsResults.job_ids_by_emails(self.session,emails)
|
|
scored_by_message=await AtsResults.job_ids_for_messages(
|
|
self.session,[row["id"] for row in inbox_rows],
|
|
)
|
|
scored_by_form=await AtsResults.job_ids_for_forms(
|
|
self.session,[row["id"] for row in form_rows],
|
|
)
|
|
skipped_active=set()
|
|
for email,jobs in scored_by_email.items():
|
|
if jobs & active_ids:
|
|
skipped_active.add(email)
|
|
known_by_email={key:set(jobs) for key,jobs in scored_by_email.items()}
|
|
|
|
pairs=[]
|
|
skipped_candidates=0
|
|
skipped_pairs=0
|
|
|
|
def _already(email,row_jobs):
|
|
jobs=set(row_jobs or ())
|
|
if email:
|
|
jobs |= known_by_email.get(email) or set()
|
|
return jobs
|
|
|
|
def _mark(email,job_id):
|
|
if email:
|
|
known_by_email.setdefault(email,set()).add(job_id)
|
|
|
|
def _skip_for_active(email,row_jobs):
|
|
if email and email in skipped_active:
|
|
return True
|
|
if (row_jobs or set()) & active_ids:
|
|
return True
|
|
return False
|
|
|
|
for row in inbox_rows:
|
|
email=row.get("email")
|
|
mid=str(row["id"])
|
|
row_jobs=scored_by_message.get(mid) or set()
|
|
if not row.get("file_path"):
|
|
skipped_candidates += 1
|
|
continue
|
|
if _skip_for_active(email,row_jobs):
|
|
skipped_candidates += 1
|
|
continue
|
|
known=_already(email,row_jobs)
|
|
for job_id in job_ids:
|
|
if job_id in known:
|
|
skipped_pairs += 1
|
|
continue
|
|
pairs.append({"kind":"inbox","record_id":mid,"job_id":job_id})
|
|
known.add(job_id)
|
|
_mark(email,job_id)
|
|
|
|
for row in form_rows:
|
|
email=row.get("email")
|
|
fid=str(row["id"])
|
|
row_jobs=scored_by_form.get(fid) or set()
|
|
if _skip_for_active(email,row_jobs):
|
|
skipped_candidates += 1
|
|
continue
|
|
known=_already(email,row_jobs)
|
|
for job_id in job_ids:
|
|
if job_id in known:
|
|
skipped_pairs += 1
|
|
continue
|
|
pairs.append({"kind":"form","record_id":fid,"job_id":job_id})
|
|
known.add(job_id)
|
|
_mark(email,job_id)
|
|
|
|
return {
|
|
"job_count":len(job_ids),
|
|
"candidate_count":len(inbox_rows)+len(form_rows),
|
|
"skipped_candidates":skipped_candidates,
|
|
"skipped_pairs":skipped_pairs,
|
|
"pairs":pairs,
|
|
}
|
|
|
|
async def prepare_on_hold_rescan_chunk(self,run_id,cursor=0):
|
|
"""Plan pairs if needed and return the next batch. Scoring happens outside."""
|
|
row=await InboxRescanRun.get_by_id(self.session,run_id)
|
|
if row is None:
|
|
return {"status":"missing"}
|
|
if row.status in ("failed","completed"):
|
|
return {"status":row.status}
|
|
|
|
now=datetime.now(timezone.utc)
|
|
entries=list(row.entries or [])
|
|
if not entries and int(cursor or 0)==0:
|
|
await InboxRescanRun.update_run(self.session,run_id,{
|
|
"status":"running",
|
|
"started_at":row.started_at or now,
|
|
})
|
|
plan=await self.plan_on_hold_pairs(row.channel,row.sheet)
|
|
entries=plan["pairs"]
|
|
await InboxRescanRun.update_run(self.session,run_id,{
|
|
"status":"scoring" if entries else "completed",
|
|
"job_count":plan["job_count"],
|
|
"candidate_count":plan["candidate_count"],
|
|
"skipped_candidates":plan["skipped_candidates"],
|
|
"skipped_pairs":plan["skipped_pairs"],
|
|
"pair_count":len(entries),
|
|
"done_count":0,
|
|
"entries":entries,
|
|
"finished_at":None if entries else now,
|
|
})
|
|
if not entries:
|
|
return {"status":"completed","pair_count":0,"batch":[]}
|
|
|
|
if not entries:
|
|
await InboxRescanRun.update_run(self.session,run_id,{
|
|
"status":"failed",
|
|
"error":"Rescan has no stored pairs",
|
|
"finished_at":datetime.now(timezone.utc),
|
|
})
|
|
return {"status":"failed","batch":[]}
|
|
|
|
chunk=4
|
|
start=int(cursor or 0)
|
|
batch=entries[start:start+chunk]
|
|
return {
|
|
"status":"scoring",
|
|
"batch":batch,
|
|
"pair_count":len(entries),
|
|
"next_cursor":min(start+len(batch),len(entries)),
|
|
"more":(start+len(batch))<len(entries),
|
|
}
|
|
|
|
async def finish_on_hold_rescan_chunk(self,run_id,next_cursor,more):
|
|
"""Stamp progress and enqueue the next chunk after scores land."""
|
|
from inbox.tasks import rescan_on_hold_run
|
|
|
|
fields={
|
|
"status":"scoring" if more else "completed",
|
|
"done_count":int(next_cursor or 0),
|
|
"finished_at":None if more else datetime.now(timezone.utc),
|
|
}
|
|
await InboxRescanRun.update_run(self.session,run_id,fields)
|
|
if more:
|
|
try:
|
|
await rescan_on_hold_run.kicker().with_labels(
|
|
created_at=datetime.now(timezone.utc).isoformat(),
|
|
correlation_id=str(run_id),
|
|
queue="inbox",
|
|
).kiq(str(run_id),int(next_cursor or 0))
|
|
except Exception as exc:
|
|
await InboxRescanRun.update_run(self.session,run_id,{
|
|
"status":"failed",
|
|
"error":str(exc),
|
|
"finished_at":datetime.now(timezone.utc),
|
|
})
|
|
raise
|
|
return {"status":fields["status"],"done_count":fields["done_count"]}
|
|
|
|
async def run_mailbox_sync_page(self,top=100,skip=0,test_on=True,on_progress=None):
|
|
"""Pull one Outlook page, triage, ingest, enqueue matching. Returns summary.
|
|
|
|
Shared by the legacy synchronous /email/fetch and the background Taskiq worker.
|
|
`on_progress(processed, expected, entries)` is optional; the mailbox_sync
|
|
worker uses it so the Sync button can poll a live percentage.
|
|
"""
|
|
data=await self.service_email(top,skip)
|
|
value=data.get("value") or []
|
|
expected=len(value)
|
|
if on_progress:
|
|
await on_progress(0,expected,[])
|
|
decisions=await self.triage_round([item.get("id") for item in value])
|
|
entries=[]
|
|
for item in value:
|
|
message_id=item.get("id")
|
|
decision=decisions.get(str(message_id)) or {}
|
|
try:
|
|
result=await self.get_email_by_id(message_id,test_on,decision=decision)
|
|
if isinstance(result,dict) and result.get("skipped"):
|
|
entries.append({
|
|
"message_id":str(message_id),
|
|
"status":"skipped",
|
|
"reason":result.get("reason") or "",
|
|
"triage_status":result.get("status") or "",
|
|
})
|
|
else:
|
|
# Prefer the decision status (known/recorded/…) when present.
|
|
status=decision.get("status") or "ingested"
|
|
if status in ("known","recorded","disabled"):
|
|
entry_status="known" if status=="known" else "ingested"
|
|
else:
|
|
entry_status="ingested"
|
|
entries.append({
|
|
"message_id":str(message_id),
|
|
"status":entry_status,
|
|
"reason":decision.get("reason") or "",
|
|
"triage_status":status,
|
|
})
|
|
except Exception as e:
|
|
entries.append({
|
|
"message_id":str(message_id),
|
|
"status":"error",
|
|
"reason":str(e),
|
|
"triage_status":"error",
|
|
})
|
|
self.triage_errors.append(str(message_id))
|
|
if on_progress:
|
|
await on_progress(len(entries),expected,entries)
|
|
|
|
if self.pending_match_ids:
|
|
await self.enqueue_matching(list(self.pending_match_ids),force=False)
|
|
|
|
account_setup=[]
|
|
if not test_on and self.pending_confirmation_emails:
|
|
account_setup=await self.send_account_setup(list(self.pending_confirmation_emails))
|
|
|
|
skipped=len(self.skipped_message_ids)
|
|
ingested=sum(1 for e in entries if e.get("status") in ("ingested","known"))
|
|
triage={
|
|
"ingested":ingested,
|
|
"skipped":skipped,
|
|
"errors":len(self.triage_errors),
|
|
"total":len(entries),
|
|
}
|
|
return {"entries":entries,"triage":triage,"account_setup":account_setup}
|
|
|
|
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,assigned=None,is_duplicate=None,no_suggestions=None,processing_state=None,city=None,source=None):
|
|
if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED or processing_state:
|
|
return await Inbox_Messages.count_inbox_messages(self.session,search,application_status=application_status,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source)
|
|
elif isread==False:
|
|
return await Inbox_Messages.count_inbox_messages(self.session,search,isread=False,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source)
|
|
else:
|
|
return await Inbox_Messages.count_inbox_messages(self.session,search,assigned=assigned,is_duplicate=is_duplicate,no_suggestions=no_suggestions,processing_state=processing_state,city=city,source=source)
|
|
|
|
async def list_cities(self):
|
|
"""Proper city names for the Inbox filter — same OpenAI mapping as CV parse."""
|
|
from employment_agent.execute_agent import normalize_cities
|
|
from g_sheet.models import FormData
|
|
inbox=await Inbox_Messages.distinct_cities(self.session)
|
|
forms=await FormData.distinct_cities(self.session)
|
|
merged=Reapplied(session=self.session).merge_cities(inbox,forms)
|
|
return await normalize_cities(merged)
|
|
|
|
async def list_sources(self):
|
|
"""Source / platform labels: seeded channels, Google Sheet, form sources."""
|
|
from g_sheet.models import FormData
|
|
channels=await SourceChannels.list_active(self.session)
|
|
forms=await FormData.distinct_sources(self.session)
|
|
return Reapplied(session=self.session).merge_cities(
|
|
[c.label for c in channels],["Google Sheet"],forms,
|
|
)
|
|
|
|
async def assign_job_post(self,record_id,job_post_id):
|
|
message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
|
|
if not message:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
if job_post_id is not None:
|
|
from job.job_post.models import JobPosts
|
|
post=await JobPosts.get_job_post_by_id(self.session,job_post_id)
|
|
if not post or post.is_deleted or not post.is_active:
|
|
raise HTTPException(status_code=422,detail="Job post is missing, deleted, or inactive")
|
|
updated=await Inbox_Messages.set_assigned_job_post(self.session,record_id,job_post_id)
|
|
if not updated:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
await Reapplied(session=self.session).sync_for_email(updated.message_from)
|
|
if job_post_id is not None:
|
|
# Assignment pairs this CV with a JD we already have — queue the ATS
|
|
# score in the background so the recruiter is not held on an OpenAI
|
|
# call. Idempotent server-side; broker-down just logs (the profile's
|
|
# Score-with-ATS button remains the manual fallback).
|
|
from inbox.tasks import score_inbox_message
|
|
try:
|
|
await score_inbox_message.kicker().with_labels(
|
|
created_at=datetime.now(timezone.utc).isoformat(),
|
|
correlation_id=str(record_id),
|
|
queue="inbox",
|
|
).kiq(str(record_id),str(job_post_id))
|
|
except Exception as exc:
|
|
logger.warning("could not queue ats score for %s: %s",record_id,exc)
|
|
return await self.get_inbox_message_by_id(record_id)
|
|
|
|
async def mark_read(self,record_id,read=True):
|
|
message=await Inbox_Messages.mark_message_read(self.session,record_id,read)
|
|
if not message:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
return serialize_message(message)
|
|
|
|
async def set_read_bulk(self,record_ids,read):
|
|
"""Flip read state for a hand-picked selection.
|
|
|
|
Returns counts, never rows: a 500-id selection would otherwise serialize 500
|
|
full messages back at a client that only needs to know it worked.
|
|
|
|
`updated` < `requested` means some ids no longer exist — a stale selection
|
|
against a list that moved. That is reported, not raised, because the rows that
|
|
DID exist were already committed.
|
|
"""
|
|
ids=[str(r).strip() for r in (record_ids or []) if str(r or "").strip()]
|
|
if not ids:
|
|
raise HTTPException(status_code=422,detail="record_ids must contain at least one id")
|
|
if len(ids)>MAX_BULK_READ_IDS:
|
|
raise HTTPException(status_code=413,
|
|
detail=f"At most {MAX_BULK_READ_IDS} ids per request")
|
|
updated=await Inbox_Messages.set_read_bulk(self.session,ids,read)
|
|
logger.info("bulk read: requested=%s updated=%s read=%s",len(ids),updated,bool(read))
|
|
return {"requested":len(ids),"updated":updated,"read":bool(read)}
|
|
|
|
async def set_read_all(self,read,search=None,isread:bool=True,
|
|
application_status:Candidate_application_Status=Candidate_application_Status.CLOSED,
|
|
assigned=None,is_duplicate=None,no_suggestions=None,processing_state=None,
|
|
city=None,source=None):
|
|
"""Mark every row the SAME filter set would have listed.
|
|
|
|
The filter arguments are the caller's current view, not a free-form query: the
|
|
button says "mark all read in this view" and the WHERE chain is literally the
|
|
list's own (Inbox_Messages._apply_filters), so the two cannot drift.
|
|
"""
|
|
updated=await Inbox_Messages.set_read_scope(
|
|
self.session,read,search=search,isread=isread,
|
|
application_status=application_status,assigned=assigned,is_duplicate=is_duplicate,
|
|
no_suggestions=no_suggestions,processing_state=processing_state,
|
|
city=city,source=source,
|
|
)
|
|
logger.info("scope read: updated=%s read=%s isread=%s assigned=%s status=%s dup=%s",
|
|
updated,bool(read),isread,assigned,getattr(application_status,"value",application_status),is_duplicate)
|
|
return {"updated":updated,"read":bool(read)}
|
|
|
|
async def refresh_read_status(self,record_id):
|
|
message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
|
|
if not message:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
if not message.message_id:
|
|
raise HTTPException(status_code=400,detail="Message has no upstream id")
|
|
try:
|
|
status=await fetch_message_read_status(message.message_id,token=self.token)
|
|
except httpx.HTTPStatusError as e:
|
|
raise HTTPException(status_code=e.response.status_code,detail=e.response.text)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500,detail=str(e))
|
|
if status is None:
|
|
raise HTTPException(status_code=404,detail="Message not found upstream")
|
|
await Inbox_Messages.apply_read_status(self.session,[status])
|
|
refreshed=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
|
|
return serialize_message(refreshed)
|
|
|
|
async def get_counts(self):
|
|
return await Inbox_Messages.count_processing(self.session)
|
|
|
|
async def set_processing_state(self,record_id,processing_state,current_user=None):
|
|
allowed=("unread","imported","processed","rejected")
|
|
if processing_state not in allowed:
|
|
raise HTTPException(status_code=422,detail=f"processing_state must be one of {', '.join(allowed)}")
|
|
if processing_state=="processed":
|
|
existing=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
|
|
if not existing:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
if not existing.assigned_job_post_id:
|
|
raise HTTPException(status_code=422,detail="Assign a job post before moving to shortlist")
|
|
# Record CLOSED/PROCESS → PENDING so the board history matches the card.
|
|
current=existing.application_status
|
|
current_val=current.value if isinstance(current,Candidate_application_Status) else str(current or "")
|
|
if current_val in ("","CLOSED","PROCESS"):
|
|
link=await Inbox.get_inbox_by_message_id(self.session,existing.id)
|
|
if link is not None:
|
|
from job.pipeline.views import Pipeline
|
|
try:
|
|
await Pipeline(self.session).change_stage(
|
|
Candidate_application_Status.PENDING.value,
|
|
current_user,
|
|
inbox_id=link.id,
|
|
change_reason="Moved to shortlist from inbox",
|
|
)
|
|
except HTTPException as exc:
|
|
if exc.status_code!=400:
|
|
raise
|
|
message=await Inbox_Messages.set_processing_state(self.session,record_id,processing_state)
|
|
if not message:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
return serialize_application(message)
|
|
|
|
async def set_duplicate(self,record_id,is_duplicate):
|
|
if not isinstance(is_duplicate,bool):
|
|
raise HTTPException(status_code=422,detail="is_duplicate must be a boolean")
|
|
message=await Inbox_Messages.set_duplicate(self.session,record_id,is_duplicate)
|
|
if not message:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
return serialize_application(message)
|
|
|
|
async def get_triage_messages(self,top,skip,search=None,is_application=None,status=None):
|
|
"""The intake gate's verdict log — mostly the mail that never became a row.
|
|
|
|
A hard gate's only real risk is the silent false negative, so the rejections
|
|
have to be reviewable.
|
|
"""
|
|
if status is not None and status not in TRIAGE_STATUSES:
|
|
raise HTTPException(status_code=422,detail=f"status must be one of {', '.join(TRIAGE_STATUSES)}")
|
|
rows=await Inbox_Message_Triage.list_triage(self.session,top,skip,is_application,status,search)
|
|
return [serialize_triage(row) for row in rows]
|
|
|
|
async def count_triage(self,search=None,is_application=None,status=None):
|
|
return await Inbox_Message_Triage.count_triage(self.session,is_application,status,search)
|
|
|
|
async def override_triage(self,record_id,is_application,current_user=None):
|
|
"""Overturn a verdict a recruiter disagrees with.
|
|
|
|
false -> true re-fetches the mail from upstream and runs the normal ingestion
|
|
path, which is why the body was never stored.
|
|
|
|
true -> false does NOT delete the inbox_messages row: inbox, ats_results,
|
|
assessments, notifications and application_stage_transitions all reference it,
|
|
so a purge would take candidate accounts and scores with it. It moves the row to
|
|
processing_state 'rejected' instead, an already-allowlisted value.
|
|
"""
|
|
if not isinstance(is_application,bool):
|
|
raise HTTPException(status_code=422,detail="is_application must be a boolean")
|
|
row=await Inbox_Message_Triage.get_triage_by_id(self.session,record_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404,detail="Triage record not found")
|
|
|
|
user_id=(current_user or {}).get("id")
|
|
if is_application and not row.ingested:
|
|
data=await self.fetch_message(row.message_id)
|
|
already=await Inbox_Messages.get_by_upstream_id(self.session,row.message_id)
|
|
pdfs=extract_pdf_attachments(data.get("attachments"))
|
|
message,new_user_email=await Inbox_Messages.insert_email(
|
|
session=self.session,email_data=data,file_path=None,
|
|
)
|
|
await Reapplied(session=self.session).sync_for_email(message.message_from)
|
|
if pdfs:
|
|
message=await attach_email_pdfs_to_s3(
|
|
self.session,message,pdfs,created_new=(already is None),
|
|
)
|
|
await Inbox_Message_Triage.mark_ingested(self.session,row.message_id,True)
|
|
if message.attachment and message.file_path and message.match_status is None:
|
|
await self.enqueue_matching([str(message.id)],force=False)
|
|
if new_user_email:
|
|
await self.send_account_setup([new_user_email])
|
|
elif not is_application and row.ingested:
|
|
message=await Inbox_Messages.get_by_upstream_id(self.session,row.message_id)
|
|
if message:
|
|
await Inbox_Messages.set_processing_state(self.session,message.id,"rejected")
|
|
|
|
updated=await Inbox_Message_Triage.set_override(self.session,record_id,is_application,user_id)
|
|
return serialize_triage(updated)
|
|
|
|
async def send_email(self,payload,current_user):
|
|
to_email=(payload.get("to") or "").strip()
|
|
subject=(payload.get("subject") or "").strip()
|
|
body=payload.get("body") or ""
|
|
content_type=(payload.get("content_type") or "html").strip() or "html"
|
|
if not to_email:
|
|
raise HTTPException(status_code=422,detail="to is required")
|
|
if not subject:
|
|
raise HTTPException(status_code=422,detail="subject is required")
|
|
if not body:
|
|
raise HTTPException(status_code=422,detail="body is required")
|
|
try:
|
|
await send_mail(to_email,subject,body,content_type=content_type)
|
|
except (httpx.HTTPError,RuntimeError) as e:
|
|
raise HTTPException(status_code=502,detail="Failed to send email") from e
|
|
inbox_id=payload.get("inbox_id")
|
|
job_post_id=None
|
|
try:
|
|
from notifications.models import Notifications
|
|
uid=None
|
|
raw=current_user.get("id") if current_user else None
|
|
if raw:
|
|
uid=uuid.UUID(str(raw))
|
|
if uid:
|
|
await Notifications.insert_notification(self.session,{
|
|
"user_id":uid,
|
|
"kind":"message",
|
|
"title":"Email sent",
|
|
"body":f"Sent “{subject}” to {to_email}",
|
|
"link_path":"/inbox",
|
|
"inbox_id":int(inbox_id) if inbox_id is not None else None,
|
|
"job_post_id":job_post_id,
|
|
})
|
|
except Exception as exc:
|
|
logger.warning("notification insert skipped: %s",exc)
|
|
return {"accepted":True,"to":to_email,"subject":subject}
|
|
|
|
async def reply_email(self,payload,current_user):
|
|
record_id=payload.get("record_id")
|
|
body=payload.get("body") or ""
|
|
if not record_id:
|
|
raise HTTPException(status_code=422,detail="record_id is required")
|
|
if not str(body).strip():
|
|
raise HTTPException(status_code=422,detail="body is required")
|
|
message=await Inbox_Messages.get_inbox_message_by_id(self.session,record_id)
|
|
if not message:
|
|
raise HTTPException(status_code=404,detail="Message not found")
|
|
to_email=(message.message_from or "").strip()
|
|
if not to_email:
|
|
raise HTTPException(status_code=422,detail="Message has no sender address")
|
|
original=(message.message_subject or "").strip()
|
|
subject=original if original.lower().startswith("re:") else f"Re: {original}" if original else "Re:"
|
|
try:
|
|
await send_mail(to_email,subject,body,content_type="html")
|
|
except (httpx.HTTPError,RuntimeError) as e:
|
|
raise HTTPException(status_code=502,detail="Failed to send email") from e
|
|
try:
|
|
from notifications.models import Notifications
|
|
uid=None
|
|
raw=current_user.get("id") if current_user else None
|
|
if raw:
|
|
uid=uuid.UUID(str(raw))
|
|
if uid:
|
|
await Notifications.insert_notification(self.session,{
|
|
"user_id":uid,
|
|
"kind":"message",
|
|
"title":"Email sent",
|
|
"body":f"Replied to {to_email}",
|
|
"link_path":"/inbox",
|
|
})
|
|
except Exception as exc:
|
|
logger.warning("notification insert skipped: %s",exc)
|
|
return {"accepted":True,"to":to_email,"subject":subject}
|
|
|
|
|
|
class Reapplied:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
self.collected={}
|
|
self.seen={}
|
|
|
|
def _norm_email(self,email):
|
|
return (email or "").strip().lower()
|
|
|
|
def _as_job_id(self,value):
|
|
if value in (None,""):
|
|
return None
|
|
text=str(value).strip()
|
|
return text or None
|
|
|
|
def _add(self,email,job_id):
|
|
uid=self._as_job_id(job_id)
|
|
if not uid or email not in self.seen or uid in self.seen[email]:
|
|
return
|
|
self.seen[email].add(uid)
|
|
self.collected[email].append(uid)
|
|
|
|
def merge_cities(self,*groups):
|
|
"""Case-insensitive unique cities, first spelling wins, sorted."""
|
|
seen=set()
|
|
out=[]
|
|
for group in groups:
|
|
for raw in group or []:
|
|
text=(raw or "").strip()
|
|
if not text:
|
|
continue
|
|
key=text.lower()
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(text)
|
|
out.sort(key=str.lower)
|
|
return out
|
|
|
|
async def sync_for_email(self,email):
|
|
stamped=await self.sync_for_emails([email])
|
|
return stamped.get(self._norm_email(email),[])
|
|
|
|
async def sync_for_emails(self,emails):
|
|
"""Collect linked job_post_ids for these emails and stamp reapplied on all 3 tables.
|
|
|
|
Records with no job_post_id are ignored while collecting. Empty collections
|
|
leave reapplied untouched.
|
|
"""
|
|
from g_sheet.models import FormData
|
|
from job.candidate.models import Manual_UPLOAD_CANDIDATE
|
|
from users.models import Users
|
|
|
|
lowers=sorted({self._norm_email(e) for e in (emails or []) if self._norm_email(e)})
|
|
if not lowers:
|
|
return {}
|
|
self.collected={email:[] for email in lowers}
|
|
self.seen={email:set() for email in lowers}
|
|
|
|
for email,job_id in await Manual_UPLOAD_CANDIDATE.job_post_ids_by_emails(self.session,lowers):
|
|
self._add(email,job_id)
|
|
for email,job_id in await FormData.job_post_ids_by_emails(self.session,lowers):
|
|
self._add(email,job_id)
|
|
for email,job_id in await Inbox_Messages.assigned_job_post_ids_by_emails(self.session,lowers):
|
|
self._add(email,job_id)
|
|
|
|
stamped={email:ids for email,ids in self.collected.items() if ids}
|
|
if not stamped:
|
|
return {}
|
|
await Users.set_reapplied_by_emails(self.session,stamped)
|
|
await Manual_UPLOAD_CANDIDATE.set_reapplied_by_emails(self.session,stamped)
|
|
await FormData.set_reapplied_by_emails(self.session,stamped)
|
|
return stamped
|