odoo18/addons_extensions/employee_bridge/models/recruitment_employee_bridge.py

560 lines
25 KiB
Python

import json
from datetime import datetime
from secrets import token_urlsafe
from odoo import _, api, fields, models
from odoo.exceptions import ValidationError
class RecruitmentEmployeeBridge(models.Model):
_name = "recruitment.employee.bridge"
_description = "Onboarding"
_inherit = ["mail.thread", "mail.activity.mixin"]
_rec_name = "name"
_order = "id desc"
@api.depends('name', 'employee_id', 'employee_name')
def _compute_display_name(self):
for record in self:
employee_name = record.employee_name or record.employee_id.name or ''
if employee_name:
record.display_name = f"{record.name} - {employee_name}"
else:
record.display_name = record.name
name = fields.Char(default="/", copy=False, readonly=True, tracking=True)
active = fields.Boolean(default=True)
employee_id = fields.Many2one(
"hr.employee",
string="Employee",
ondelete="restrict",
tracking=True,
index=True,
)
company_id = fields.Many2one(
"res.company",
string="Company",
default=lambda self: self.env.company,
required=True,
index=True,
)
source = fields.Selection(
[
("jod", "JOD"),
("manual", "Manual"),
("recruitment", "Recruitment"),
],
default="jod",
required=True,
tracking=True,
)
state = fields.Selection(
[
("draft", "Draft"),
("jod_sent", "JOD Sent"),
("jod_received", "JOD Received"),
("validated", "Validated"),
],
default="draft",
required=True,
tracking=True,
)
employee_name = fields.Char(string="Employee Name", tracking=True)
employee_code = fields.Char(string="Employee Code", tracking=True)
candidate_image = fields.Image(string="Photo")
work_email = fields.Char(string="Work Email", tracking=True)
private_email = fields.Char(string="Private Email")
alternate_phone = fields.Char(string="Alternate Mobile")
work_phone = fields.Char(string="Work Phone")
mobile_phone = fields.Char(string="Mobile Phone")
department_id = fields.Many2one("hr.department", string="Department")
job_id = fields.Many2one("hr.job", string="Job Position")
emp_type = fields.Many2one("hr.contract.type", string="Employment Type")
doj = fields.Date(string="Date of Joining")
gender = fields.Selection([
("male", "Male"),
("female", "Female"),
("other", "Other"),
])
birthday = fields.Date()
blood_group = fields.Selection([
("A+", "A+"),
("A-", "A-"),
("B+", "B+"),
("B-", "B-"),
("O+", "O+"),
("O-", "O-"),
("AB+", "AB+"),
("AB-", "AB-"),
], string="Blood Group")
marital = fields.Selection(
selection=[
("single", "Single"),
("married", "Married"),
("cohabitant", "Legal Cohabitant"),
("widower", "Widower"),
("divorced", "Divorced"),
],
default="single",
string="Marital Status",
)
marriage_anniversary_date = fields.Date(string="Anniversary Date")
private_street = fields.Char(string="Private Street")
private_street2 = fields.Char(string="Private Street2")
private_city = fields.Char(string="Private City")
private_state_id = fields.Many2one("res.country.state", string="Private State")
private_zip = fields.Char(string="Private Zip")
private_country_id = fields.Many2one("res.country", string="Private Country")
permanent_street = fields.Char(string="Permanent Street")
permanent_street2 = fields.Char(string="Permanent Street2")
permanent_city = fields.Char(string="Permanent City")
permanent_state_id = fields.Many2one("res.country.state", string="Permanent State")
permanent_zip = fields.Char(string="Permanent Zip")
permanent_country_id = fields.Many2one("res.country", string="Permanent Country")
pan_no = fields.Char(string="PAN No")
identification_id = fields.Char(string="Aadhar No")
previous_company_pf_no = fields.Char(string="Previous Company PF No")
previous_company_uan_no = fields.Char(string="Previous Company UAN No")
passport_no = fields.Char(string="Passport No")
passport_start_date = fields.Date(string="Passport Issued Date")
passport_end_date = fields.Date(string="Passport End Date")
passport_issued_location = fields.Char(string="Passport Issued Location")
full_name_as_in_bank = fields.Char(string="Name as per Bank")
bank_name = fields.Char()
bank_branch = fields.Char()
bank_account_no = fields.Char()
bank_ifsc_code = fields.Char()
requested_attachment_ids = fields.Many2many(
"employee.bridge.requested.attachment",
"employee_bridge_requested_attachment_rel",
"bridge_id",
"attachment_id",
string="Attachments to Request",
)
joining_attachment_ids = fields.One2many(
"employee.bridge.attachment",
"bridge_id",
string="Attachments",
)
education_history = fields.One2many("education.history", "bridge_id", string="Education Details")
employer_history = fields.One2many("employer.history", "bridge_id", string="Employer History")
family_details = fields.One2many("family.details", "bridge_id", string="Family Details")
joining_form_link = fields.Char(string="Joining Form Link", copy=False, readonly=True)
access_token = fields.Char(copy=False, readonly=True)
total_exp = fields.Float(string="Total Exp")
_sql_constraints = [
(
"unique_bridge_employee",
"unique(employee_id)",
"This employee is already connected to an employee bridge.",
),
]
def _get_base_url(self):
return self.env["ir.config_parameter"].sudo().get_param("web.base.url")
def _ensure_access_token(self):
for bridge in self:
if not bridge.access_token:
bridge.access_token = token_urlsafe(32)
def _get_joining_form_link(self):
self.ensure_one()
self._ensure_access_token()
return "%s/employee_bridge/jod/%s" % (self._get_base_url(), self.access_token)
def _group_requested_attachments(self, attachments):
return {
"personal_docs": attachments.filtered(lambda a: a.attachment_type == "personal").mapped("name"),
"education_docs": attachments.filtered(lambda a: a.attachment_type == "education").mapped("name"),
"previous_employer_docs": attachments.filtered(lambda a: a.attachment_type == "previous_employer").mapped("name"),
"other_docs": attachments.filtered(lambda a: a.attachment_type == "others").mapped("name"),
}
def _send_jod_mail(self, req_attachment_ids=False, email_values=None, email_body=False):
for bridge in self:
email_to = bridge.work_email or bridge.private_email or bridge.employee_id.work_email
if not email_to:
raise ValidationError(_("Add an email on the bridge before sending the JOD form."))
link = bridge._get_joining_form_link()
bridge.write({
"joining_form_link": link,
"state": "jod_sent",
})
attachments = req_attachment_ids or self.env["employee.bridge.requested.attachment"].browse()
template = self.env.ref("employee_bridge.email_template_jod_form", raise_if_not_found=False)
context = {
"joining_form_link": link,
**bridge._group_requested_attachments(attachments),
}
values = dict(email_values or {})
values.setdefault("email_from", self.env.company.email or self.env.user.email)
values.setdefault("email_to", email_to)
if email_body:
values["body_html"] = email_body
if template:
rendered_subject = template.with_context(**context)._render_field("subject", [bridge.id])[bridge.id]
rendered_body = template.with_context(**context)._render_field(
"body_html", [bridge.id], compute_lang=True
)[bridge.id]
values.setdefault("subject", rendered_subject)
values.setdefault("body_html", rendered_body)
template.sudo().with_context(**context).send_mail(bridge.id, email_values=values, force_send=True)
else:
values.setdefault("subject", _("Joining Onboarding Details"))
values.setdefault("body_html", _(
"<p>Hello %(name)s,</p>"
"<p>Please submit your joining onboarding details using this link:</p>"
"<p><a href='%(link)s'>Open Joining Form</a></p>"
) % {
"name": bridge.employee_name or bridge.employee_id.name or _("Employee"),
"link": link,
})
self.env["mail.mail"].sudo().create(values).send()
return True
def action_send_jod_form(self):
self.ensure_one()
return {
"type": "ir.actions.act_window",
"name": _("Select Attachments"),
"res_model": "employee.bridge.attachment.wizard",
"view_mode": "form",
"target": "new",
"context": {
"default_bridge_id": self.id,
},
}
def action_open_employee(self):
self.ensure_one()
if not self.employee_id:
raise ValidationError(_("No employee is linked to this bridge."))
return {
"name": _("Employee"),
"type": "ir.actions.act_window",
"res_model": "hr.employee",
"view_mode": "form",
"res_id": self.employee_id.id,
}
def action_validate_employee_details(self):
for bridge in self:
if not bridge.employee_id:
raise ValidationError(_("Link or create an employee before validating the JOD details."))
bridge.employee_id.write(bridge._get_employee_update_vals())
bridge.education_history.write({"employee_id": bridge.employee_id.id})
bridge.employer_history.write({"employee_id": bridge.employee_id.id})
bridge.family_details.write({"employee_id": bridge.employee_id.id})
bridge.joining_attachment_ids.write({"employee_id": bridge.employee_id.id})
bridge.state = "validated"
def _get_employee_create_vals(self):
self.ensure_one()
vals = {
"name": self.employee_name,
"employee_id": self.employee_code,
"work_email": self.work_email,
"private_email": self.private_email,
"work_phone": self.work_phone,
"mobile_phone": self.mobile_phone,
"image_1920": self.candidate_image,
"department_id": self.department_id.id,
"job_id": self.job_id.id,
"job_title": self.job_id.name,
"emp_type": self.emp_type.id,
"company_id": self.company_id.id,
}
vals = {key: value for key, value in vals.items() if value not in (False, "", 0)}
if not vals.get("name"):
raise ValidationError(_("Employee Name is required to create an employee."))
return vals
def _get_employee_update_vals(self):
self.ensure_one()
vals = {
"name": self.employee_name,
"employee_id": self.employee_code,
"work_email": self.work_email,
"private_email": self.private_email,
"work_phone": self.work_phone,
"mobile_phone": self.mobile_phone,
"image_1920": self.candidate_image,
"department_id": self.department_id.id,
"job_id": self.job_id.id,
"job_title": self.job_id.name,
"emp_type": self.emp_type.id,
"doj": self.doj,
"gender": self.gender,
"birthday": self.birthday,
"blood_group": self.blood_group,
"marital": self.marital,
"marriage_anniversary_date": self.marriage_anniversary_date,
"private_street": self.private_street,
"private_street2": self.private_street2,
"private_city": self.private_city,
"private_state_id": self.private_state_id.id,
"private_zip": self.private_zip,
"private_country_id": self.private_country_id.id,
"permanent_street": self.permanent_street,
"permanent_street2": self.permanent_street2,
"permanent_city": self.permanent_city,
"permanent_state_id": self.permanent_state_id.id,
"permanent_zip": self.permanent_zip,
"permanent_country_id": self.permanent_country_id.id,
"pan_no": self.pan_no,
"identification_id": self.identification_id,
"previous_company_pf_no": self.previous_company_pf_no,
"previous_company_uan_no": self.previous_company_uan_no,
"passport_id": self.passport_no,
"passport_start_date": self.passport_start_date,
"passport_end_date": self.passport_end_date,
"passport_issued_location": self.passport_issued_location,
}
return {
key: value
for key, value in vals.items()
if key in self.employee_id._fields and value not in (False, "", 0)
}
def action_create_employee(self):
for bridge in self:
if bridge.employee_id:
continue
bridge.employee_id = self.env["hr.employee"].sudo().create(bridge._get_employee_create_vals())
return self.action_open_employee()
def _sync_from_employee(self):
for bridge in self.filtered("employee_id"):
employee = bridge.employee_id
bridge.write({
"employee_name": bridge.employee_name or employee.name,
"employee_code": bridge.employee_code or employee.employee_id,
"work_email": bridge.work_email or employee.work_email,
"private_email": bridge.private_email or employee.private_email,
"work_phone": bridge.work_phone or employee.work_phone,
"mobile_phone": bridge.mobile_phone or employee.mobile_phone,
"department_id": bridge.department_id.id or employee.department_id.id,
"job_id": bridge.job_id.id or employee.job_id.id,
"emp_type": bridge.emp_type.id or employee.emp_type.id,
"company_id": employee.company_id.id or bridge.company_id.id,
"doj": bridge.doj or employee.doj,
"gender": bridge.gender or employee.gender,
"birthday": bridge.birthday or employee.birthday,
"blood_group": bridge.blood_group or employee.blood_group,
"marital": bridge.marital or employee.marital,
"marriage_anniversary_date": bridge.marriage_anniversary_date or employee.marriage_anniversary_date,
"private_street": bridge.private_street or employee.private_street,
"private_street2": bridge.private_street2 or employee.private_street2,
"private_city": bridge.private_city or employee.private_city,
"private_state_id": bridge.private_state_id.id or employee.private_state_id.id,
"private_zip": bridge.private_zip or employee.private_zip,
"private_country_id": bridge.private_country_id.id or employee.private_country_id.id,
"permanent_street": bridge.permanent_street or employee.permanent_street,
"permanent_street2": bridge.permanent_street2 or employee.permanent_street2,
"permanent_city": bridge.permanent_city or employee.permanent_city,
"permanent_state_id": bridge.permanent_state_id.id or employee.permanent_state_id.id,
"permanent_zip": bridge.permanent_zip or employee.permanent_zip,
"permanent_country_id": bridge.permanent_country_id.id or employee.permanent_country_id.id,
"pan_no": bridge.pan_no or employee.pan_no,
"identification_id": bridge.identification_id or employee.identification_id,
"previous_company_pf_no": bridge.previous_company_pf_no or employee.previous_company_pf_no,
"previous_company_uan_no": bridge.previous_company_uan_no or employee.previous_company_uan_no,
"passport_no": bridge.passport_no or employee.passport_id,
"passport_start_date": bridge.passport_start_date or employee.passport_start_date,
"passport_end_date": bridge.passport_end_date or employee.passport_end_date,
"passport_issued_location": bridge.passport_issued_location or employee.passport_issued_location,
})
@api.onchange("employee_id")
def _onchange_employee_id(self):
for bridge in self:
employee = bridge.employee_id
if not employee:
continue
bridge.employee_name = bridge.employee_name or employee.name
bridge.employee_code = bridge.employee_code or employee.employee_id
bridge.work_email = bridge.work_email or employee.work_email
bridge.private_email = bridge.private_email or employee.private_email
bridge.work_phone = bridge.work_phone or employee.work_phone
bridge.mobile_phone = bridge.mobile_phone or employee.mobile_phone
bridge.department_id = bridge.department_id or employee.department_id
bridge.job_id = bridge.job_id or employee.job_id
bridge.emp_type = bridge.emp_type or employee.emp_type
def write_from_public_form(self, values):
self.ensure_one()
private_state = self.env["res.country.state"].sudo().browse(int(values.get("present_state") or 0))
permanent_state = self.env["res.country.state"].sudo().browse(int(values.get("permanent_state") or 0))
mapped_values = dict(values)
mapped_values.update({
"candidate_image": values.get("candidate_image_base64"),
"private_email": values.get("email_from"),
"mobile_phone": values.get("partner_phone"),
"alternate_phone": values.get("alternate_phone"),
"private_street": values.get("present_street"),
"private_street2": values.get("present_street2"),
"private_city": values.get("present_city"),
"private_zip": values.get("present_zip"),
"private_state_id": private_state.id if private_state else False,
"private_country_id": private_state.country_id.id if private_state else False,
"permanent_state_id": permanent_state.id if permanent_state else False,
"permanent_country_id": permanent_state.country_id.id if permanent_state else False,
})
allowed_fields = {
"employee_name",
"candidate_image",
"private_email",
"mobile_phone",
"alternate_phone",
"doj",
"gender",
"birthday",
"blood_group",
"marital",
"marriage_anniversary_date",
"private_street",
"private_street2",
"private_city",
"private_state_id",
"private_country_id",
"private_zip",
"permanent_street",
"permanent_street2",
"permanent_city",
"permanent_state_id",
"permanent_country_id",
"permanent_zip",
"full_name_as_in_bank",
"bank_name",
"bank_branch",
"bank_account_no",
"bank_ifsc_code",
"pan_no",
"identification_id",
"previous_company_pf_no",
"previous_company_uan_no",
"passport_no",
"passport_start_date",
"passport_end_date",
"passport_issued_location",
}
bridge_vals = {
field_name: value or False
for field_name, value in mapped_values.items()
if field_name in allowed_fields and field_name in self._fields
}
if bridge_vals:
self.sudo().write(bridge_vals)
self._write_history_from_public_form(values)
self._replace_joining_attachments_from_public_form(values)
self.state = "jod_received"
def _safe_date(self, value):
try:
return datetime.strptime(value, "%Y-%m-%d").date() if value else False
except (TypeError, ValueError):
return False
def _safe_int(self, value):
try:
return int(value) if value else False
except (TypeError, ValueError):
return False
def _json_list(self, values, key):
raw = values.get(key, "[]")
try:
data = json.loads(raw) if raw else []
except (TypeError, ValueError):
data = []
return data if isinstance(data, list) else []
def _write_history_from_public_form(self, values):
self.ensure_one()
family_values = []
for member in self._json_list(values, "family_data_json"):
if member.get("name") and member.get("relation"):
family_values.append((0, 0, {
"relation_type": member.get("relation"),
"name": member.get("name"),
"contact_no": member.get("contact"),
"dob": self._safe_date(member.get("dob")),
"location": member.get("location"),
}))
education_values = []
for education in self._json_list(values, "education_data_json"):
if education.get("specialization"):
education_values.append((0, 0, {
"education_type": education.get("education_type") or "additional",
"name": education.get("specialization"),
"university": education.get("university"),
"start_year": self._safe_int(education.get("start_year")),
"end_year": self._safe_int(education.get("end_year")),
"marks_or_grade": education.get("marks_or_grade"),
}))
employer_values = []
for employer in self._json_list(values, "employer_history_data_json"):
if employer.get("company_name"):
employer_values.append((0, 0, {
"company_name": employer.get("company_name"),
"designation": employer.get("designation"),
"date_of_joining": self._safe_date(employer.get("date_of_joining")),
"last_working_day": self._safe_date(employer.get("last_working_day")),
"ctc": employer.get("ctc"),
}))
write_vals = {}
if family_values:
write_vals["family_details"] = [(5, 0, 0)] + family_values
if education_values:
write_vals["education_history"] = [(5, 0, 0)] + education_values
if employer_values:
write_vals["employer_history"] = [(5, 0, 0)] + employer_values
if write_vals:
self.sudo().write(write_vals)
def _replace_joining_attachments_from_public_form(self, values):
self.ensure_one()
attachment_values = []
seen_files = set()
for item in self._json_list(values, "attachments_data_json"):
attachment = self.env["employee.bridge.requested.attachment"].sudo().browse(
int(item.get("attachment_rec_id") or 0)
)
file_content = item.get("file_content")
file_name = item.get("file_name")
file_key = (attachment.id, file_name, file_content)
if attachment.exists() and file_content and file_key not in seen_files:
seen_files.add(file_key)
attachment_values.append((0, 0, {
"name": attachment.name,
"requested_attachment_id": attachment.id,
"file": file_content,
"file_name": file_name,
}))
self.sudo().write({"joining_attachment_ids": [(5, 0, 0)] + attachment_values})
@api.model_create_multi
def create(self, vals_list):
for vals in vals_list:
if vals.get("name", "/") == "/":
vals["name"] = self.env["ir.sequence"].next_by_code("recruitment.employee.bridge") or "/"
if vals.get("employee_id") and not vals.get("company_id"):
employee = self.env["hr.employee"].browse(vals["employee_id"])
vals["company_id"] = employee.company_id.id or self.env.company.id
bridges = super().create(vals_list)
bridges._sync_from_employee()
return bridges
def write(self, vals):
res = super().write(vals)
if "employee_id" in vals:
self._sync_from_employee()
return res