"""Extend hr.employee with SAP B1 linkage + full bidirectional field mapping. Direction of truth: - B1 held the real employee roster (imported into Odoo once). - Odoo is the source of truth going forward and pushes changes back to B1, matched by the stored B1 EmployeeID (no fragile name matching, no dupes). Mapping covers the HR-meaningful subset of B1's EmployeeInfo (identity, contact, work/org, nationality/passport for foreign staff, marital/family, employment date, salary, social insurance). Country-localization tax fields (DE/BR) are intentionally ignored. """ import logging from odoo import _, fields, models from odoo.exceptions import UserError from .sapb1_client import SAPB1Client, check_production_write_allowed _logger = logging.getLogger(__name__) # -- enum maps between B1 and Odoo ----------------------------------------- _SEX_B1_TO_ODOO = {"gt_Male": "male", "gt_Female": "female"} _SEX_ODOO_TO_B1 = {"male": "gt_Male", "female": "gt_Female", "other": "gt_Undefined"} _MARITAL_B1_TO_ODOO = { "mts_Single": "single", "mts_Married": "married", "mts_Divorced": "divorced", "mts_Widowed": "widower", } _MARITAL_ODOO_TO_B1 = { "single": "mts_Single", "married": "mts_Married", "divorced": "mts_Divorced", "widower": "mts_Widowed", "cohabitant": "mts_Single", } def _b1_date(value): """B1 DateTime -> 'YYYY-MM-DD' (Odoo date) or False.""" if not value: return False return str(value)[:10] def _odoo_date(value): """Odoo date -> B1 DateTime string or None.""" if not value: return None return f"{value}T00:00:00Z" class HrEmployee(models.Model): _inherit = "hr.employee" # groups= on all B1 fields: (1) salary/bank data must not be readable by # every internal user; (2) custom stored hr.employee fields without a # groups restriction break *any* employee read by a non-HR user (e.g. # the Time Off type dropdown) -- Odoo prefetches all stored fields and # hr.employee.public's private-fields check rejects unknown ones. b1_employee_id = fields.Integer( string="SAP B1 Employee ID", index=True, copy=False, readonly=True, groups="hr.group_hr_user", help="EmployeeID of the linked record in SAP Business One.", ) b1_synced = fields.Datetime(string="Last B1 Sync", readonly=True, copy=False, groups="hr.group_hr_user") # Fields B1 has but Odoo Community lacks natively (kept here for a complete sync) b1_start_date = fields.Date(string="Employment Start (B1)", groups="hr.group_hr_user") b1_salary = fields.Float(string="Salary (B1)", groups="hr.group_hr_user") b1_bank_account = fields.Char(string="Bank Account (B1)", groups="hr.group_hr_user") # -- mapping: B1 -> Odoo ------------------------------------------------ def _b1_to_odoo_vals(self, row): Country = self.env["res.country"] Dept = self.env["hr.department"] first = (row.get("FirstName") or "").strip() last = (row.get("LastName") or "").strip() name = " ".join(p for p in (first, last) if p) or f"B1 #{row.get('EmployeeID')}" def country_of(code): if not code: return False c = Country.search([("code", "=", code)], limit=1) return c.id or False vals = { "name": name, "job_title": row.get("JobTitle") or False, "work_email": row.get("eMail") or False, "work_phone": row.get("OfficePhone") or False, "mobile_phone": row.get("MobilePhone") or False, "birthday": _b1_date(row.get("DateOfBirth")), "place_of_birth": row.get("BirthPlace") or False, "identification_id": row.get("IdNumber") or False, "passport_id": row.get("PassportNumber") or False, "passport_expiration_date": _b1_date(row.get("PassportExpirationDate")), "children": row.get("NumOfChildren") or 0, "ssnid": row.get("SocialInsuranceNumber") or False, "active": (row.get("Active") != "tNO"), "b1_start_date": _b1_date(row.get("StartDate")), "b1_salary": row.get("Salary") or 0.0, "b1_bank_account": row.get("BankAccount") or False, } if row.get("Gender") in _SEX_B1_TO_ODOO: vals["sex"] = _SEX_B1_TO_ODOO[row["Gender"]] if row.get("MartialStatus") in _MARITAL_B1_TO_ODOO: vals["marital"] = _MARITAL_B1_TO_ODOO[row["MartialStatus"]] nat = country_of(row.get("CitizenshipCountryCode")) if nat: vals["country_id"] = nat cob = country_of(row.get("CountryOfBirth")) if cob: vals["country_of_birth"] = cob if row.get("Department"): dept = Dept.with_context(active_test=False).search( [("b1_dept_code", "=", row["Department"])], limit=1 ) if dept: vals["department_id"] = dept.id return vals # -- mapping: Odoo -> B1 ------------------------------------------------ def _odoo_to_b1_vals(self): self.ensure_one() parts = (self.name or "").split(" ", 1) body = { "FirstName": parts[0], "LastName": parts[1] if len(parts) > 1 else parts[0], # SAP B1's JobTitle property rejects values beyond a short limit # (observed failure at 24 chars, success confirmed up to 21) — # truncate only for the outbound push; Odoo's own job_title is # left intact. "JobTitle": (self.job_title or "")[:20], "eMail": self.work_email or "", "OfficePhone": self.work_phone or "", "MobilePhone": self.mobile_phone or "", "DateOfBirth": _odoo_date(self.birthday), "BirthPlace": self.place_of_birth or "", "IdNumber": self.identification_id or "", "PassportNumber": self.passport_id or "", "PassportExpirationDate": _odoo_date(self.passport_expiration_date), "NumOfChildren": self.children or 0, "SocialInsuranceNumber": self.ssnid or "", "StartDate": _odoo_date(self.b1_start_date), "Salary": self.b1_salary or 0.0, "BankAccount": self.b1_bank_account or "", } if self.sex in _SEX_ODOO_TO_B1: body["Gender"] = _SEX_ODOO_TO_B1[self.sex] if self.marital in _MARITAL_ODOO_TO_B1: body["MartialStatus"] = _MARITAL_ODOO_TO_B1[self.marital] if self.country_id and self.country_id.code: body["CitizenshipCountryCode"] = self.country_id.code if self.country_of_birth and self.country_of_birth.code: body["CountryOfBirth"] = self.country_of_birth.code if self.department_id and self.department_id.b1_dept_code: body["Department"] = self.department_id.b1_dept_code if self.parent_id and self.parent_id.b1_employee_id: body["Manager"] = self.parent_id.b1_employee_id return body # -- import: B1 -> Odoo (read-only on B1) ------------------------------- def action_import_from_b1(self): Employee = self.env["hr.employee"].sudo() Dept = self.env["hr.department"].sudo() select = ( "EmployeeID,FirstName,LastName,JobTitle,eMail,OfficePhone,MobilePhone," "DateOfBirth,BirthPlace,IdNumber,PassportNumber,PassportExpirationDate," "NumOfChildren,SocialInsuranceNumber,Gender,MartialStatus," "CitizenshipCountryCode,CountryOfBirth,Department,Manager,StartDate," "Salary,BankAccount,Active" ) with SAPB1Client.from_env(self.env) as b1: # department master first, so employee dept mapping resolves for d in b1.get_all("Departments", params={"$select": "Code,Name"}): Dept._b1_get_or_create(d["Code"], d.get("Name")) rows = b1.get_all("EmployeesInfo", params={"$select": select}) now = fields.Datetime.now() by_b1id = {} created = updated = 0 for row in rows: b1_id = row.get("EmployeeID") vals = self._b1_to_odoo_vals(row) vals["b1_synced"] = now vals["b1_employee_id"] = b1_id emp = Employee.with_context(active_test=False).search( [("b1_employee_id", "=", b1_id)], limit=1 ) if not emp: emp = Employee.with_context(active_test=False).search( [("name", "=", vals["name"]), ("b1_employee_id", "=", False)], limit=1 ) if emp: emp.write(vals) updated += 1 else: emp = Employee.create(vals) created += 1 by_b1id[b1_id] = emp row["_odoo_emp"] = emp # second pass: resolve managers now that all employees exist for row in rows: mgr_b1 = row.get("Manager") if mgr_b1 and mgr_b1 in by_b1id: row["_odoo_emp"].parent_id = by_b1id[mgr_b1].id _logger.info("B1 import: %s created, %s updated", created, updated) return {"created": created, "updated": updated, "total": len(rows)} # -- push: Odoo -> B1 --------------------------------------------------- def action_push_to_b1(self): if not self: # Previously silently fell back to "all employees" when called # on an empty recordset -- a scripting mistake or misconfigured # binding would then push the entire company without anyone # intending it. Fail loudly instead; use action_push_all_to_b1() # if pushing everyone is genuinely what's wanted. raise UserError(_( "No employee selected to push. Use 'Push all employees to " "SAP B1' from the list view if that's what you meant." )) check_production_write_allowed(self.env) pushed = 0 with SAPB1Client.from_env(self.env) as b1: for emp in self: body = emp._odoo_to_b1_vals() if emp.b1_employee_id: b1.patch(f"EmployeesInfo({emp.b1_employee_id})", json=body) else: body["ExternalEmployeeNumber"] = f"ODOO-{emp.id}" created = b1.post("EmployeesInfo", json=body) if created and created.get("EmployeeID"): emp.b1_employee_id = created["EmployeeID"] emp.b1_synced = fields.Datetime.now() pushed += 1 _logger.info("B1 push: %s employees", pushed) return pushed def action_push_all_to_b1(self): """Explicit bulk push -- the only place 'push everyone' is allowed, so it can never happen by accident (see action_push_to_b1).""" return self.env["hr.employee"].search([]).action_push_to_b1() def _cron_push_to_b1(self): candidates = self.search([]) stale = candidates.filtered( lambda e: not e.b1_synced or (e.write_date and e.write_date > e.b1_synced) ) if stale: if not check_production_write_allowed(self.env, soft=True): return stale.action_push_to_b1()