import logging import os import uuid from datetime import datetime, timezone from typing import Any, List, 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 sqlalchemy.orm import selectinload from sqlmodel import Field, Relationship, SQLModel, select, true from job.candidate.models import Activity, Feedback, Interviews 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): __tablename__ = "inbox" id: int | None = Field(default=None, primary_key=True) user_id: uuid.UUID | None = Field(default=None, foreign_key="users.id") alert_id: uuid.UUID | None = Field(default=None, foreign_key="inbox_alerts.id") alerts: Optional["Inbox_Alerts"] = Relationship(back_populates="inbox") message_id: uuid.UUID | None = Field(default=None, foreign_key="inbox_messages.id") messages: Optional["Inbox_Messages"] = Relationship(back_populates="inbox") created_at: datetime = Field(default_factory=datetime.now) updated_at: datetime = Field(default_factory=datetime.now) favorite: Optional[bool] = Field(default=False) rating: Optional[float] = Field(default=0.0) # selectin on one-to-many: joined would repeat the inbox row per child interviews: List["Interviews"] = Relationship( back_populates="inbox", sa_relationship_kwargs={"lazy": "selectin"}, ) activity: List["Activity"] = Relationship( back_populates="inbox", sa_relationship_kwargs={"lazy": "selectin"}, ) feedback: List["Feedback"] = Relationship( back_populates="inbox", sa_relationship_kwargs={"lazy": "selectin"}, ) user: Optional["Users"] = Relationship( back_populates="inbox", sa_relationship_kwargs={"lazy": "joined"}, ) @classmethod def _candidate_search_filter(cls, search: str): pattern = f"%{search}%" return or_(Users.name.ilike(pattern), Users.email.ilike(pattern)) @classmethod async def get_candidate_profile(cls,session:AsyncSession,user_id:uuid.UUID|None=None,limit:int=10,offset:int=0,search:str|None=None): try: options=[selectinload(cls.messages)] if user_id: options.extend([ selectinload(cls.interviews), selectinload(cls.activity), selectinload(cls.feedback), ]) qry = ( select(cls) .options(*options) .join(Users, cls.user_id == Users.id) .join(Roles, Users.role_id == Roles.id) .where(Roles.role_name == EnumRoles.CANDIDATE.value) ) if user_id: qry = qry.where(cls.user_id == user_id) if search: qry = qry.where(cls._candidate_search_filter(search)) qry = qry.limit(limit).offset(offset) result = await session.execute(qry) rows = result.scalars().all() if user_id and len(rows) == 1: return rows[0] return rows except Exception as e: raise HTTPException(status_code=500,detail=str(e)) @classmethod async def count_candidate_profiles(cls,session:AsyncSession,user_id:uuid.UUID|None=None,search:str|None=None): """Result-set size for the same predicate get_candidate_profile pages over.""" try: qry = ( select(func.count()) .select_from(cls) .join(Users, cls.user_id == Users.id) .join(Roles, Users.role_id == Roles.id) .where(Roles.role_name == EnumRoles.CANDIDATE.value) ) if user_id: qry = qry.where(cls.user_id == user_id) if search: qry = qry.where(cls._candidate_search_filter(search)) result = await session.execute(qry) return result.scalar_one() except Exception as e: raise HTTPException(status_code=500,detail=str(e)) @classmethod async def get_inbox_by_id(cls,session:AsyncSession,record_id:int|str|None): if record_id is None: return None try: iid=int(record_id) except (TypeError,ValueError): return None result=await session.execute(select(cls).where(cls.id==iid)) return result.scalars().first() @classmethod async def get_inbox_by_message_id(cls,session:AsyncSession,message_id): try: mid=uuid.UUID(str(message_id)) except ValueError: return None result=await session.execute( select(cls).where(cls.message_id==mid).order_by(cls.created_at.desc()) ) return result.scalars().first() @classmethod async def get_inbox_by_user_id(cls,session:AsyncSession,user_id): try: uid=uuid.UUID(str(user_id)) except ValueError: return None result=await session.execute( select(cls).where(cls.user_id==uid).order_by(cls.created_at.desc()) ) return result.scalars().first() @classmethod async def update_inbox(cls,session:AsyncSession,record_id,fields:dict): row=await cls.get_inbox_by_id(session,record_id) if not row: return None for key,value in fields.items(): setattr(row,key,value) row.updated_at=datetime.now() session.add(row) await session.commit() await session.refresh(row) return row class Inbox_Alerts(SQLModel, table=True): __tablename__ = "inbox_alerts" id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) alert_sender_name: str alert_sender_email: str is_read: bool = Field(default=False) recieve_time: datetime = Field(default_factory=datetime.now) inbox: list[Inbox] = Relationship(back_populates="alerts") class Inbox_Messages(SQLModel, table=True): __tablename__ = "inbox_messages" id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) message_id: str | None = Field(default=None, index=True, unique=True) full_email_response: dict[str, Any] | None = Field( default=None, sa_column=Column(JSONB) ) application_status: Candidate_application_Status = Field(default=Candidate_application_Status.CLOSED) message_subject: str message_body: str message_sent_time: str message_received_time: str message_from: str message_to: str message_cc: str | None = Field(default=None) message_bcc: str | None = Field(default=None) message_read: bool = Field(default=False) attachment: bool = Field(default=False) message_reply: str | None = Field(default=None) 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)) assigned_job_post_id: uuid.UUID | None = Field(default=None, foreign_key="job_posts.id", index=True) match_summary: str | None = Field(default=None) match_reasoning: str | None = Field(default=None) match_status: str | None = Field(default=None) match_error: str | None = Field(default=None) matched_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) candidate_phone_number: str | None = Field(default="xxx-xxx-xxxx") candidate_education: str | None = Field(default=None) current_employment: str | None = Field(default=None) inbox: list[Inbox] = Relationship(back_populates="messages") @staticmethod def _body_text(email_data: dict) -> str: body = email_data.get("body") if isinstance(body, dict): return body.get("content") or "" if isinstance(body, str): return body return email_data.get("bodyPreview") or "" @classmethod async def get_all_applications(cls,session:AsyncSession,message_id:uuid.UUID|int|None=None): try: qry=select(cls.message_id,cls.full_email_response,cls.message_subject,cls.message_from,cls.message_to,cls.message_sent_time,cls.message_read,cls.attachment) if message_id: qry=qry.where(cls.message_id==message_id) result=await session.execute(qry) return result.scalars().all() except Exception as e: raise HTTPException(status_code=500,detail=str(e)) @classmethod async def set_match_result( cls, session: AsyncSession, record_id, *, resume_text=None, experience=None, candidate_education=None, candidate_phone_number=None, current_employment=None, suggested_job_post_ids=None, summary="", reasoning="", status="", error="", ): """Persist agent output onto one inbox row; returns the row or None.""" row = await cls.get_inbox_message_by_id(session, record_id) if not row: return None if resume_text is not None: row.resume_text = resume_text if candidate_phone_number is not None: row.candidate_phone_number = candidate_phone_number if candidate_education is not None: row.candidate_education = candidate_education if current_employment is not None: row.current_employment = current_employment row.suggested_job_post_ids = suggested_job_post_ids row.match_summary = summary or None 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() await session.refresh(row) return row @classmethod def _fields_from_email(cls, email_data: dict, file_path: list[str] | None = None) -> dict: return { "message_subject": email_data.get("subject") or "", "message_body": cls._body_text(email_data), "message_sent_time": email_data.get("sentDateTime") or "", "message_read": bool(email_data.get("isRead")), "message_received_time": email_data.get("receivedDateTime") or "", "message_from": email_data.get("from", {}) .get("emailAddress", {}) .get("address", ""), "message_to": ",".join( [r["emailAddress"]["address"] for r in email_data.get("toRecipients", [])] ), "message_cc": ",".join( [r["emailAddress"]["address"] for r in email_data.get("ccRecipients", [])] ), "message_bcc": ",".join( [r["emailAddress"]["address"] for r in email_data.get("bccRecipients", [])] ), "attachment": bool(email_data.get("hasAttachments")), "message_reply": ",".join( [r["emailAddress"]["address"] for r in email_data.get("replyTo", [])] ), "message_id": email_data.get("id"), "file_name": ",".join( [r.get("name") for r in email_data.get("attachments", [])] ), "file_path": ",".join(file_path) if file_path else None, "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: # id-only: avoid Users.job_posts selectin / role lazy loads under asyncio user_id=(await session.execute( select(Users.id).where(func.lower(Users.email)==address) )).scalar_one_or_none() if user_id is None: 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) # autoflush=False: flush so users.id exists before inbox FK insert # (Relationship helps ordering, but flush keeps this path explicit). await session.flush() session.add(Inbox(user_id=user.id,message_id=email.id)) await session.commit() return address link=(await session.execute( select(Inbox.id).where(Inbox.message_id==email.id,Inbox.user_id==user_id) )).scalar_one_or_none() if link is None: 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, session: AsyncSession, 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( select(cls).where(cls.message_id == external_id) ) ).scalars().first() if existing: for key, value in fields.items(): setattr(existing, key, value) session.add(existing) await session.commit() await session.refresh(existing) if fields.get("attachment"): link_user=await cls._link_sender(session, email_data, existing) # _link_sender may rollback (IntegrityError); that expires this row await session.refresh(existing) return existing, link_user email = cls(**fields) session.add(email) await session.commit() await session.refresh(email) if fields.get("attachment"): link_user=await cls._link_sender(session, email_data, email) await session.refresh(email) return email, link_user @classmethod def _search_filter(cls, search: str): pattern = f"%{search}%" return or_( cls.message_subject.ilike(pattern), cls.message_from.ilike(pattern), cls.message_body.ilike(pattern), ) @classmethod async def get_inbox_messages( cls, session: AsyncSession, top: int | None, skip: int, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None ): statement = select(cls).order_by(cls.message_received_time.desc()) if search: statement = statement.where(cls._search_filter(search)) if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED: statement = statement.where(cls.application_status==application_status) if assigned is True: statement = statement.where(cls.assigned_job_post_id.is_not(None)) elif assigned is False: statement = statement.where(cls.assigned_job_post_id.is_(None)) if skip: statement = statement.offset(skip) if top is not None: statement = statement.limit(top) if isread==False: statement = statement.where(cls.message_read==False) result = await session.execute(statement) return result.scalars().all() @classmethod async def get_inbox_message_by_id(cls, session: AsyncSession, record_id: str): try: uid = uuid.UUID(str(record_id)) except ValueError: return None result = await session.execute(select(cls).where(cls.id == uid)) return result.scalars().first() @classmethod async def set_assigned_job_post(cls, session: AsyncSession, record_id, job_post_id): """Set or clear assigned_job_post_id; returns the row or None if missing.""" row = await cls.get_inbox_message_by_id(session, record_id) if not row: return None if job_post_id is None: row.assigned_job_post_id = None else: try: row.assigned_job_post_id = uuid.UUID(str(job_post_id)) except ValueError: return None session.add(row) await session.commit() await session.refresh(row) return row @classmethod async def count_inbox_messages(cls, session: AsyncSession, search: str | None, isread: bool=True, application_status: Candidate_application_Status=Candidate_application_Status.CLOSED, assigned: bool | None=None): statement = select(func.count()).select_from(cls) if search: statement = statement.where(cls._search_filter(search)) if application_status == Candidate_application_Status.PROCESS or application_status==Candidate_application_Status.REJECTED: statement = statement.where(cls.application_status==application_status) if assigned is True: statement = statement.where(cls.assigned_job_post_id.is_not(None)) elif assigned is False: statement = statement.where(cls.assigned_job_post_id.is_(None)) if isread==False: statement = statement.where(cls.message_read==False) result = await session.execute(statement) return result.scalar_one() @classmethod async def apply_read_status(cls, session: AsyncSession, changes) -> int: """[{id, isRead, ...}] -> bulk UPDATE message_read. Returns rows touched. read is a ONE-WAY LATCH: only false -> true is applied, never the reverse. mark_message_read writes the local column only — nothing pushes the state back to Outlook — so upstream keeps reporting isRead=false and the every-minute sync_read_status sweep would otherwise revert a mail the user just opened. Cost of the latch: un-reading a mail in Outlook no longer propagates here. """ if not changes: return 0 read_ids=[c.get("id") for c in changes if c.get("id") and c.get("isRead")] if not read_ids: return 0 result=await session.execute( update(cls).where(cls.message_id.in_(read_ids)).values(message_read=True) ) await session.commit() return result.rowcount or 0 @classmethod async def mark_message_read(cls, session: AsyncSession, record_id): row=await cls.get_inbox_message_by_id(session,record_id) if not row: return None row.message_read=True session.add(row) await session.commit() await session.refresh(row) return row