203 lines
7.2 KiB
Python
203 lines
7.2 KiB
Python
import asyncio
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from inbox.models import Inbox
|
|
from job.candidate.models import CandidateHistory, Interviews, Manual_UPLOAD_CANDIDATE
|
|
from job.history.enums import HistoryEvent
|
|
from job.history.serializers import serialize_history
|
|
from users.models import Users
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
INTERVIEW_HISTORY_EVENTS = {
|
|
HistoryEvent.INTERVIEW_CREATED.value,
|
|
HistoryEvent.INTERVIEW_UPDATED.value,
|
|
HistoryEvent.CALENDAR_CREATED.value,
|
|
HistoryEvent.CALENDAR_RESCHEDULED.value,
|
|
HistoryEvent.CALENDAR_CANCELLED.value,
|
|
}
|
|
|
|
|
|
def _text(value):
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
if isinstance(value, datetime):
|
|
return value.isoformat()
|
|
return str(value)
|
|
|
|
|
|
class HistoryRecorder:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def resolve_user_id(
|
|
self,
|
|
*,
|
|
user_id=None,
|
|
inbox_id=None,
|
|
manual_upload_candidate_id=None,
|
|
message_id=None,
|
|
):
|
|
uid = CandidateHistory._as_uuid(user_id)
|
|
if uid is not None:
|
|
return uid
|
|
if inbox_id is not None:
|
|
row = await Inbox.get_inbox_by_id(self.session, inbox_id)
|
|
if row and row.user_id:
|
|
return row.user_id
|
|
if manual_upload_candidate_id is not None:
|
|
row = await Manual_UPLOAD_CANDIDATE.get_by_id(self.session, manual_upload_candidate_id)
|
|
if row and row.user_id:
|
|
return row.user_id
|
|
if message_id is not None:
|
|
row = await Inbox.get_inbox_by_message_id(self.session, message_id)
|
|
if row and row.user_id:
|
|
return row.user_id
|
|
return None
|
|
|
|
def _actor_id(self, current_user=None, actor_id=None):
|
|
if actor_id is not None:
|
|
return CandidateHistory._as_uuid(actor_id)
|
|
if isinstance(current_user, dict) and current_user.get("id"):
|
|
return CandidateHistory._as_uuid(current_user.get("id"))
|
|
if current_user and not isinstance(current_user, dict):
|
|
return CandidateHistory._as_uuid(current_user)
|
|
return None
|
|
|
|
async def record(
|
|
self,
|
|
event_type,
|
|
*,
|
|
current_user=None,
|
|
actor_id=None,
|
|
user_id=None,
|
|
inbox_id=None,
|
|
manual_upload_candidate_id=None,
|
|
message_id=None,
|
|
entity_type=None,
|
|
entity_id=None,
|
|
from_value=None,
|
|
to_value=None,
|
|
description=None,
|
|
meta=None,
|
|
actor_kind=None,
|
|
commit=False,
|
|
):
|
|
try:
|
|
resolved = await self.resolve_user_id(
|
|
user_id=user_id,
|
|
inbox_id=inbox_id,
|
|
manual_upload_candidate_id=manual_upload_candidate_id,
|
|
message_id=message_id,
|
|
)
|
|
if resolved is None:
|
|
return None
|
|
fields = {
|
|
"user_id": resolved,
|
|
"inbox_id": int(inbox_id) if inbox_id is not None else None,
|
|
"manual_upload_candidate_id": CandidateHistory._as_uuid(manual_upload_candidate_id),
|
|
"event_type": event_type,
|
|
"entity_type": entity_type,
|
|
"entity_id": str(entity_id) if entity_id is not None else None,
|
|
"from_value": _text(from_value),
|
|
"to_value": _text(to_value),
|
|
"description": description,
|
|
"actor_id": self._actor_id(current_user=current_user, actor_id=actor_id),
|
|
"actor_kind": actor_kind or "user",
|
|
"meta": meta,
|
|
}
|
|
row = await CandidateHistory.insert_event(self.session, fields, commit=commit)
|
|
try:
|
|
from notifications.views import notify_candidate_history
|
|
await notify_candidate_history(
|
|
self.session,
|
|
row,
|
|
inbox_id=inbox_id,
|
|
manual_upload_candidate_id=manual_upload_candidate_id,
|
|
commit=commit,
|
|
)
|
|
except Exception:
|
|
logger.exception("candidate history notification failed for %s", event_type)
|
|
return row
|
|
except Exception:
|
|
logger.exception("candidate history record failed for %s", event_type)
|
|
if commit:
|
|
try:
|
|
await self.session.rollback()
|
|
except Exception:
|
|
logger.exception("candidate history rollback failed")
|
|
return None
|
|
|
|
async def _event_emails(self, event_id):
|
|
from interview.plugins import get_event
|
|
from interview.serializers import participants_from_event
|
|
|
|
try:
|
|
raw = await get_event(event_id)
|
|
except Exception:
|
|
logger.exception("calendar event fetch failed for history")
|
|
return None, []
|
|
if not raw:
|
|
return None, []
|
|
organizer, attendees = participants_from_event(raw)
|
|
org_email = (organizer or {}).get("email")
|
|
attendee_emails = [a.get("email") for a in (attendees or []) if a.get("email")]
|
|
return org_email, attendee_emails
|
|
|
|
async def _attach_outlook_emails(self, rows, items):
|
|
pairs = [
|
|
(row, item)
|
|
for row, item in zip(rows, items)
|
|
if row.event_type in INTERVIEW_HISTORY_EVENTS and row.entity_id
|
|
]
|
|
if not pairs:
|
|
return
|
|
interviews = await Interviews.get_interviews_by_ids(
|
|
self.session, {row.entity_id for row, _item in pairs}
|
|
)
|
|
event_by_interview = {
|
|
str(r.id): r.graph_event_id for r in interviews if r.graph_event_id
|
|
}
|
|
pending = []
|
|
event_ids = []
|
|
for row, item in pairs:
|
|
eid = event_by_interview.get(str(row.entity_id))
|
|
if not eid:
|
|
continue
|
|
event_ids.append(eid)
|
|
pending.append((item, eid))
|
|
ids = list(dict.fromkeys(event_ids))
|
|
if not ids:
|
|
return
|
|
sem = asyncio.Semaphore(5)
|
|
|
|
async def one(eid):
|
|
async with sem:
|
|
return eid, *(await self._event_emails(eid))
|
|
|
|
fetched = {eid: (org, atts) for eid, org, atts in await asyncio.gather(*[one(eid) for eid in ids])}
|
|
for item, eid in pending:
|
|
org_email, attendee_emails = fetched.get(eid, (None, []))
|
|
if org_email:
|
|
item["organizer_email"] = org_email
|
|
if attendee_emails:
|
|
item["attendee_emails"] = attendee_emails
|
|
|
|
async def list_for_user(self, user_id, *, limit=200, offset=0):
|
|
rows, total = await CandidateHistory.fetch_by_user(
|
|
self.session, user_id, limit=limit, offset=offset
|
|
)
|
|
actor_ids = {r.actor_id for r in rows if r.actor_id}
|
|
names = await Users.names_by_ids(self.session, actor_ids)
|
|
items = [serialize_history(r, actor_name=names.get(str(r.actor_id))) for r in rows]
|
|
try:
|
|
await self._attach_outlook_emails(rows, items)
|
|
except Exception:
|
|
logger.exception("calendar participant hydrate failed for history")
|
|
return items, total
|