odoo18/addons_extensions/hr_recruitment_extended/models/hr_applicant.py

516 lines
24 KiB
Python

from odoo import models, fields, api, _
from dateutil.relativedelta import relativedelta
from datetime import datetime
from odoo.exceptions import ValidationError
from odoo.tools.mimetypes import guess_mimetype, fix_filename_extension
class HRApplicant(models.Model):
_inherit = 'hr.applicant'
_track_duration_field = 'recruitment_stage_id'
hide_chatter_suggestion = fields.Boolean(string="Hide Chatter Suggestions", default=False, tracking=True)
primary_skill_match_percentage = fields.Float(
string="Primary Skill Match (%)",
compute='_compute_skill_match_percentages',
store=True,
digits=(16, 2),
)
secondary_skill_match_percentage = fields.Float(
string="Secondary Skill Match (%)",
compute='_compute_skill_match_percentages',
store=True,
digits=(16, 2),
)
overall_skill_match_percentage = fields.Float(
string="Overall Skill Match (%)",
compute='_compute_skill_match_percentages',
store=True,
digits=(16, 2),
)
candidate_image = fields.Image(related='candidate_id.candidate_image', readonly=False, compute_sudo=True)
submitted_to_client = fields.Boolean(string="Submitted to Client", default=False, tracking=True)
client_submission_date = fields.Datetime(string="Submission Date", tracking=True)
submitted_stage = fields.Many2one('hr.recruitment.stage', string="Submitted Stage", tracking=True)
refused_stage = fields.Many2one('hr.recruitment.stage', string="Reject Stage")
refused_comments = fields.Text(string='Reject Comments')
is_on_hold = fields.Boolean(string="Is On Hold", default=False)
hold_state = fields.Selection(
[('hold', 'On Hold')],
string="Hold Status",
compute='_compute_hold_state',
)
stage_comment_ids = fields.One2many(
'hr.applicant.stage.comment',
'applicant_id',
string='Stage Comments',
)
stage_comment_count = fields.Integer(compute='_compute_stage_comment_count')
stage_comment_tooltips = fields.Json(compute='_compute_stage_comment_tooltips')
@api.depends('is_on_hold')
def _compute_hold_state(self):
for applicant in self:
applicant.hold_state = 'hold' if applicant.is_on_hold else False
def hold_unhold_button(self):
for rec in self:
if rec.is_on_hold:
rec.is_on_hold = False
else:
rec.is_on_hold = True
return {'type': 'ir.actions.client', 'tag': 'reload'}
def action_toggle_chatter_visibility(self):
for record in self:
record.hide_chatter_suggestion = not record.hide_chatter_suggestion
return {'type': 'ir.actions.client', 'tag': 'reload'}
@api.depends('stage_comment_ids')
def _compute_stage_comment_count(self):
for applicant in self:
applicant.stage_comment_count = len(applicant.stage_comment_ids)
@api.depends('stage_comment_ids.comment', 'stage_comment_ids.stage_id', 'stage_comment_ids.user_id', 'stage_comment_ids.comment_date')
def _compute_stage_comment_tooltips(self):
for applicant in self:
grouped_comments = {}
for comment in applicant.stage_comment_ids.sorted(lambda item: item.comment_date or fields.Datetime.now()):
if not comment.stage_id:
continue
line = '%s: %s' % (comment.user_id.name, comment.comment)
grouped_comments.setdefault(comment.stage_id.id, []).append(line)
applicant.stage_comment_tooltips = {
stage_id: '\n'.join(comments[-4:])
for stage_id, comments in grouped_comments.items()
}
def action_open_stage_comment_wizard(self):
self.ensure_one()
return {
'type': 'ir.actions.act_window',
'name': _('Stage Comments'),
'res_model': 'applicant.stage.comment.wizard',
'view_mode': 'form',
'view_id': self.env.ref('hr_recruitment_extended.view_applicant_stage_comment_wizard_form').id,
'target': 'new',
'context': {
'active_id': self.id,
'default_applicant_id': self.id,
'default_stage_id': self.recruitment_stage_id.id,
},
}
@api.onchange('submitted_to_client')
def _onchange_submitted_to_client(self):
for applicant in self:
if applicant.submitted_to_client:
applicant.client_submission_date = applicant.client_submission_date or fields.Datetime.now()
applicant.submitted_stage = applicant.submitted_stage or applicant.recruitment_stage_id
else:
applicant.client_submission_date = False
applicant.submitted_stage = False
@api.model_create_multi
def create(self, vals_list):
for vals in vals_list:
if vals.get('submitted_to_client'):
vals.setdefault('client_submission_date', fields.Datetime.now())
vals.setdefault('submitted_stage', vals.get('recruitment_stage_id'))
return super().create(vals_list)
@api.depends('hr_job_recruitment.skill_ids', 'hr_job_recruitment.secondary_skill_ids', 'candidate_id.skill_ids')
def _compute_skill_match_percentages(self):
for applicant in self:
percentages = {
'primary_skill_match_percentage': 0.0,
'secondary_skill_match_percentage': 0.0,
'overall_skill_match_percentage': 0.0,
}
if applicant.hr_job_recruitment and applicant.candidate_id:
percentages = applicant.hr_job_recruitment._get_skill_match_percentages(applicant.candidate_id.skill_ids)
percentages = {
key: value for key, value in percentages.items()
if key in {'primary_skill_match_percentage', 'secondary_skill_match_percentage', 'overall_skill_match_percentage'}
}
applicant.update(percentages)
@api.constrains('candidate_id','hr_job_recruitment')
def hr_applicant_constrains(self):
for rec in self:
if rec.candidate_id and rec.hr_job_recruitment:
self.sudo().search([('candidate_id','=',rec.candidate_id.id),('hr_job_recruitment','=',rec.hr_job_recruitment.id),('id','!=',rec.id)])
@api.model
def _read_group_recruitment_stage_ids(self, stages, domain):
# retrieve job_id from the context and write the domain: ids + contextual columns (job or default)
job_recruitment_id = self._context.get('default_hr_job_recruitment')
search_domain = []
if job_recruitment_id:
search_domain = [('job_recruitment_ids', '=', job_recruitment_id)] + search_domain
stage_ids = stages.sudo()._search(search_domain, order=stages._order)
return stages.browse(stage_ids)
def write(self, vals):
if vals.get('submitted_to_client') and 'submitted_stage' not in vals and len(self) > 1:
for applicant in self:
applicant_vals = dict(vals)
applicant_vals.setdefault('client_submission_date', fields.Datetime.now())
applicant_vals['submitted_stage'] = vals.get('recruitment_stage_id') or applicant.recruitment_stage_id.id or False
applicant.write(applicant_vals)
return True
if vals.get('submitted_to_client'):
vals.setdefault('client_submission_date', fields.Datetime.now())
if 'submitted_stage' not in vals:
submitted_stage = vals.get('recruitment_stage_id') or self[:1].recruitment_stage_id.id
if submitted_stage:
vals['submitted_stage'] = submitted_stage
elif vals.get('submitted_to_client') is False:
vals.setdefault('client_submission_date', False)
vals.setdefault('submitted_stage', False)
if 'recruitment_stage_id' in vals:
blocked_records = self.filtered(
lambda applicant: applicant.is_on_hold and applicant.recruitment_stage_id.id != vals.get('recruitment_stage_id')
)
if blocked_records:
raise ValidationError(_("You cannot change the stage of an applicant while it is on hold. Please unhold it first."))
if 'stage_id' in vals:
blocked_records = self.filtered(
lambda applicant: applicant.is_on_hold and applicant.stage_id.id != vals.get('stage_id')
)
if blocked_records:
raise ValidationError(_("You cannot change the stage of an applicant while it is on hold. Please unhold it first."))
# user_id change: update date_open
res = super().write(vals)
if vals.get('user_id'):
vals['date_open'] = fields.Datetime.now()
old_interviewers = self.interviewer_ids
# stage_id: track last stage before update
if 'recruitment_stage_id' in vals:
vals['date_last_stage_update'] = fields.Datetime.now()
if 'kanban_state' not in vals:
vals['kanban_state'] = 'normal'
for applicant in self:
vals['last_stage_id'] = applicant.recruitment_stage_id.id
return res
@api.depends('hr_job_recruitment')
def _compute_department(self):
for applicant in self:
applicant.department_id = applicant.hr_job_recruitment.department_id.id
@api.depends('hr_job_recruitment')
def _compute_stage(self):
for applicant in self:
if applicant.hr_job_recruitment:
if not applicant.recruitment_stage_id:
stage_ids = self.env['hr.recruitment.stage'].search([
'|',
('job_recruitment_ids', '=', False),
('job_recruitment_ids', '=', applicant.hr_job_recruitment.id),
('fold', '=', False)
], order='sequence asc', limit=1).ids
applicant.recruitment_stage_id = stage_ids[0] if stage_ids else False
else:
applicant.recruitment_stage_id = False
@api.depends('job_id')
def _compute_user(self):
for applicant in self:
applicant.user_id = applicant.hr_job_recruitment.user_id.id
def init(self):
super().init()
self.env.cr.execute("""
CREATE INDEX IF NOT EXISTS hr_applicant_job_id_recruitment_stage_id_idx
ON hr_applicant(job_id, recruitment_stage_id)
WHERE active IS TRUE
""")
refused_state = fields.Many2one('hr.recruitment.stage', readonly=True, force_save=True, string="Reject state")
hr_job_recruitment = fields.Many2one('hr.job.recruitment')
job_id = fields.Many2one('hr.job', related='hr_job_recruitment.job_id', store=True)
recruitment_stage_id = fields.Many2one('hr.recruitment.stage', 'Stage', ondelete='restrict', tracking=True,
compute='_compute_recruitment_stage', store=True, readonly=False,
domain="[('job_recruitment_ids', '=', hr_job_recruitment)]",
copy=False, index=True,
group_expand='_read_group_recruitment_stage_ids')
stage_color = fields.Char(related="recruitment_stage_id.stage_color")
request_form_ids = fields.One2many(
'applicant.request.forms',
'applicant_id',
string='Request Forms'
)
post_onboarding_form_status = fields.Selection([('draft','Draft'),('email_sent_to_candidate','Email Sent to Candidate'),('done','Done')], default='draft')
legend_blocked = fields.Char(related='recruitment_stage_id.legend_blocked', string='Kanban Blocked')
legend_done = fields.Char(related='recruitment_stage_id.legend_done', string='Kanban Valid')
legend_normal = fields.Char(related='recruitment_stage_id.legend_normal', string='Kanban Ongoing')
employee_code = fields.Char(related="employee_id.employee_id")
recruitment_attachments = fields.Many2many(
'recruitment.attachments',
string='Attachments Request')
joining_attachment_ids = fields.One2many('employee.recruitment.attachments','applicant_id',string="Attachments")
attachments_validation_status = fields.Selection([('pending', 'Pending'),
('validated', 'Validated')], default='pending')
approval_required = fields.Boolean(related='recruitment_stage_id.require_approval')
application_submitted = fields.Boolean(string="Application Submitted")
resume = fields.Binary(related='candidate_id.resume', readonly=False, compute_sudo=True)
resume_type = fields.Char(related='candidate_id.resume_type', readonly=False, compute_sudo=True)
resume_name = fields.Char(related='candidate_id.resume_name', readonly=False, compute_sudo=True)
@api.onchange('resume')
def onchange_resume(self):
for rec in self:
if rec.resume:
attachment = self.env.ref("hr_recruitment_extended.employee_recruitment_attachments_preview")
file = attachment.sudo().write({
'datas': rec.resume,
})
if file:
rec.resume_type = attachment.mimetype
else:
rec.resume_type = ''
rec.resume_name = ''
def preview_resume(self):
pass
def replace_joining_attachments(self, attachments_data):
self.ensure_one()
latest_attachments = {}
for attachment in attachments_data or []:
attachment_id = attachment.get('attachment_rec_id')
file_content = attachment.get('file_content')
if not attachment_id or not file_content:
continue
latest_attachments[int(attachment_id)] = {
'name': attachment.get('file_name', ''),
'recruitment_attachment_id': int(attachment_id),
'file': file_content,
}
if not latest_attachments:
return
existing_lines = self.joining_attachment_ids.filtered(
lambda line: line.recruitment_attachment_id.id in latest_attachments
)
if existing_lines:
existing_lines.unlink()
self.write({
'joining_attachment_ids': [(0, 0, values) for values in latest_attachments.values()]
})
def action_share_applicant(self):
self.ensure_one()
return {
'type': 'ir.actions.act_window',
'name': _('Share Applicant'),
'res_model': 'client.submission.mails.template.wizard',
'view_mode': 'form',
'view_id': self.env.ref('hr_recruitment_extended.view_client_submission_mails_template_wizard_form').id,
'target': 'new',
'context': {
'default_template_id': self.env.ref(
'hr_recruitment_extended.application_client_submission_email_template'
).id,
},
}
def submit_for_approval(self):
for rec in self:
manager_id = self.env['ir.config_parameter'].sudo().get_param('requisitions.requisition_manager')
if not manager_id:
raise ValidationError(_("Recruitment Manager is not selected please go into the Configuration->Settings and add the Manager"))
mail_template = self.env.ref('hr_recruitment_extended.email_template_candidate_approval')
manager_id = self.env['res.users'].sudo().browse(int(manager_id))
render_ctx = dict(recruitment_manager=manager_id)
mail_template.with_context(render_ctx).send_mail(
self.id,
force_send=True,
email_layout_xmlid='mail.mail_notification_light')
rec.application_submitted = True
def approve_applicant(self):
for rec in self:
manager_id = self.env['ir.config_parameter'].sudo().get_param('requisitions.requisition_manager')
if not manager_id:
raise ValidationError(
_("Recruitment Manager is not selected please go into the Configuration->Settings and add the Manager"))
mail_template = self.env.ref('hr_recruitment_extended.email_template_stage_approved')
manager_id = self.env['res.users'].sudo().browse(int(manager_id))
render_ctx = dict(recruitment_manager=manager_id)
mail_template.with_context(render_ctx).send_mail(
self.id,
force_send=True,
email_layout_xmlid='mail.mail_notification_light')
rec.application_submitted = False
recruitment_stage_ids = rec.hr_job_recruitment.recruitment_stage_ids.ids
current_stage = self.env['hr.recruitment.stage'].browse(rec.recruitment_stage_id.id)
next_stage = self.env['hr.recruitment.stage'].search([
('id', 'in', recruitment_stage_ids),
('sequence', '>', current_stage.sequence)
], order='sequence asc', limit=1)
if next_stage:
rec.recruitment_stage_id = next_stage.id
def action_validate_attachments(self):
for rec in self:
if rec.employee_id and rec.joining_attachment_ids:
rec.joining_attachment_ids.write({'employee_id': rec.employee_id.id})
rec.attachments_validation_status = 'validated'
else:
raise ValidationError(_("No Data to Validate"))
def send_second_application_form_to_candidate(self):
"""Send the salary expectation and experience form to the candidate."""
template = self.env.ref('hr_recruitment_extended.email_template_second_application_form', raise_if_not_found=False)
for applicant in self:
if template and applicant.email_from:
template.send_mail(applicant.id, force_send=True)
applicant.second_application_form_status = 'email_sent_to_candidate'
def send_jod_form_to_employee(self):
for rec in self:
if not rec.employee_id:
raise ValidationError(_('You must first create the employee before before Sending the Post Onboarding Form'))
elif not rec.employee_id.employee_id:
raise ValidationError(_('Employee Code for the Employee (%s) is missing')%(rec.employee_id.name))
return {
'type': 'ir.actions.act_window',
'name': 'Select Attachments',
'res_model': 'post.onboarding.attachment.wizard',
'view_mode': 'form',
'view_type': 'form',
'target': 'new',
'context': {'default_req_attachment_ids': []}
}
def send_pre_onboarding_doc_request_form_to_candidate(self):
return {
'type': 'ir.actions.act_window',
'name': 'Select Attachments',
'res_model': 'post.onboarding.attachment.wizard',
'view_mode': 'form',
'view_type': 'form',
'target': 'new',
'context': {'default_req_attachment_ids': [],'default_is_pre_onboarding_attachment_request': True}
}
def _track_template(self, changes):
res = super(HRApplicant, self)._track_template(changes)
applicant = self[0]
# When applcant is unarchived, they are put back to the default stage automatically. In this case,
# don't post automated message related to the stage change.
if 'recruitment_stage_id' in changes and applicant.exists()\
and applicant.recruitment_stage_id.template_id\
and not applicant._context.get('just_moved')\
and not applicant._context.get('just_unarchived'):
res['recruitment_stage_id'] = (applicant.recruitment_stage_id.template_id, {
'auto_delete_keep_log': False,
'subtype_id': self.env['ir.model.data']._xmlid_to_res_id('mail.mt_note'),
'email_layout_xmlid': 'hr_recruitment.mail_notification_light_without_background'
})
return res
def _track_subtype(self, init_values):
record = self[0]
if 'recruitment_stage_id' in init_values and record.recruitment_stage_id:
return self.env.ref('hr_recruitment.mt_applicant_stage_changed')
return super(HRApplicant, self)._track_subtype(init_values)
def message_new(self, msg, custom_values=None):
stage = False
defaults = {}
if custom_values and 'hr_job_recruitment' in custom_values:
recruitment_stage_id = self.env['hr.job.recruitment'].browse(custom_values['hr_job_recruitment'])._get_first_stage()
if stage and stage.id:
defaults['recruitment_stage_id'] = recruitment_stage_id.id
res = super(HRApplicant, self).message_new(msg, custom_values=defaults)
return res
def reset_applicant(self):
""" Reinsert the applicant into the recruitment pipe in the first stage"""
default_stage = dict()
for hr_job_recruitment in self.mapped('hr_job_recruitment'):
default_stage[hr_job_recruitment.id] = self.env['hr.recruitment.stage'].search(
[
('job_recruitment_ids', '=', hr_job_recruitment.id),
('fold', '=', False)
], order='sequence asc', limit=1).id
for applicant in self:
applicant.write(
{'recruitment_stage_id': applicant.hr_job_recruitment.id and default_stage[applicant.hr_job_recruitment.id],
'refuse_reason_id': False})
@api.depends('recruitment_stage_id.hired_stage')
def _compute_date_closed(self):
for applicant in self:
if applicant.recruitment_stage_id and applicant.recruitment_stage_id.hired_stage and not applicant.date_closed:
applicant.date_closed = fields.datetime.now()
if not applicant.recruitment_stage_id.hired_stage:
applicant.date_closed = False
@api.depends('hr_job_recruitment')
def _compute_recruitment_stage(self):
for applicant in self:
if applicant.hr_job_recruitment:
if not applicant.recruitment_stage_id:
stage_ids = self.env['hr.recruitment.stage'].search([
'|',
('job_recruitment_ids', '=', False),
('job_recruitment_ids', '=', applicant.hr_job_recruitment.id),
('fold', '=', False)
], order='sequence asc', limit=1).ids
applicant.recruitment_stage_id = stage_ids[0] if stage_ids else False
else:
applicant.recruitment_stage_id = False
def _get_duration_from_tracking(self, trackings):
json = super()._get_duration_from_tracking(trackings)
now = datetime.now()
for applicant in self:
if applicant.refuse_reason_id and applicant.refuse_date:
json[applicant.recruitment_stage_id.id] -= (now - applicant.refuse_date).total_seconds()
return json
def create_employee_from_applicant(self):
self.ensure_one()
action = self.candidate_id.create_employee_from_candidate()
employee = self.env['hr.employee'].browse(action['res_id'])
employee.write({
'image_1920': self.candidate_image,
'job_id': self.job_id.id,
'job_title': self.job_id.name,
'department_id': self.department_id.id,
'work_email': self.department_id.company_id.email or self.email_from, # To have a valid email address by default
'work_phone': self.department_id.company_id.phone,
})
return action
def print_joining_form(self):
return self.env.ref('hr_recruitment_extended.action_joining_form_report').report_action(self)
class ApplicantGetRefuseReason(models.TransientModel):
_inherit = 'applicant.get.refuse.reason'
def action_refuse_reason_apply(self):
res = super(ApplicantGetRefuseReason, self).action_refuse_reason_apply()
refused_applications = self.applicant_ids
refused_applications.write({'refused_state': refused_applications.stage_id.id})
return res