odoo18/addons_extensions/bench_management_system/models/bench_management.py

175 lines
5.8 KiB
Python

from odoo import models, fields, tools, api
class BenchManagementLine(models.Model):
_name = "bench.management.line"
_description = "Employee Availability (Bench)"
_auto = False
_rec_name = 'employee_id'
employee_id = fields.Many2one("hr.employee", readonly=True)
job_id = fields.Many2one("hr.job", readonly=True)
company_id = fields.Many2one("res.company", related="employee_id.company_id")
project_line_ids = fields.Many2many(
'project.team.line',
compute='_compute_bench_details',
string='Project Assignments',
readonly=True,
)
limited_project_line_ids = fields.Many2many(
compute='_compute_bench_details',
comodel_name='project.team.line',
string='Kanban Projects',
readonly=True,
)
project_names_tooltip = fields.Text(
string="Project Names",
compute='_compute_bench_details',
readonly=True,
)
project_count = fields.Integer(
string="Project Count",
compute='_compute_bench_details',
readonly=True,
)
active_project_count = fields.Integer(
string="Active Projects",
compute='_compute_bench_details',
readonly=True,
)
future_project_count = fields.Integer(
string="Upcoming Projects",
compute='_compute_bench_details',
readonly=True,
)
completed_project_count = fields.Integer(
string="Completed Projects",
compute='_compute_bench_details',
readonly=True,
)
status = fields.Selection([
("bench", "Bench"),
("partial", "Partial"),
("full", "Full"),
], readonly=True)
def _get_line_availability_status(self, line, today):
return line.status or 'not_started'
def _compute_bench_details(self):
project_team_line = self.env['project.team.line'].sudo()
today = fields.Date.context_today(self)
for rec in self:
project_lines = project_team_line.search(
[('employee_id', '=', rec.employee_id.id)],
order='start_date desc, id desc'
)
active_lines = project_lines.filtered(
lambda line: rec._get_line_availability_status(line, today) == 'in_progress'
)
future_lines = project_lines.filtered(
lambda line: rec._get_line_availability_status(line, today) == 'not_started'
)
completed_lines = project_lines.filtered(
lambda line: rec._get_line_availability_status(line, today) == 'done'
)
project_records = project_lines.mapped('project_id')
if active_lines:
bench_status = 'full'
elif future_lines:
bench_status = 'partial'
else:
bench_status = 'bench'
rec.project_line_ids = project_lines
rec.limited_project_line_ids = project_lines[:3]
rec.project_count = len(project_records)
rec.active_project_count = len(active_lines)
rec.future_project_count = len(future_lines)
rec.completed_project_count = len(completed_lines)
rec.project_names_tooltip = '\n'.join(
f"{line.project_id.display_name or 'No Project'} - {dict(line._fields['status'].selection).get(rec._get_line_availability_status(line, today), 'N/A')}"
for line in project_lines
) or ''
def init(self):
tools.drop_view_if_exists(self.env.cr, self._table)
self.env.cr.execute("""
CREATE OR REPLACE VIEW bench_management_line AS (
SELECT
he.id AS id,
he.id AS employee_id,
he.job_id AS job_id,
CASE
WHEN EXISTS (
SELECT 1
FROM project_team_line tpl
WHERE tpl.employee_id = he.id
AND tpl.status = 'in_progress'
) THEN 'full'
WHEN EXISTS (
SELECT 1
FROM project_team_line tpl
WHERE tpl.employee_id = he.id
AND tpl.status = 'not_started'
) THEN 'partial'
ELSE 'bench'
END AS status
FROM hr_employee he
)
""")
class ProjectTeamLine(models.Model):
_inherit = 'project.team.line'
line_status_color = fields.Integer(
compute='_compute_line_status_color',
string='Status Color',
readonly=True,
)
@api.depends('status')
def _compute_line_status_color(self):
color_map = {
'not_started': 8,
'in_progress': 2,
'done': 10,
}
for line in self:
line.line_status_color = color_map.get(line.status, 0)
def name_get(self):
result = []
for rec in self:
name = rec.project_id.display_name or 'No Project'
result.append((rec.id, name))
return result
def _sync_project_members(self):
if self.env.context.get('skip_project_team_member_sync'):
return True
self.mapped('project_id')._sync_members_from_team_lines()
return True
@api.model_create_multi
def create(self, vals_list):
records = super().create(vals_list)
records._sync_project_members()
return records
def write(self, vals):
projects = self.mapped('project_id')
res = super().write(vals)
(projects | self.mapped('project_id'))._sync_members_from_team_lines()
return res
def unlink(self):
projects = self.mapped('project_id')
res = super().unlink()
projects._sync_members_from_team_lines()
return res