Files
gzth/yuthon_salary/wizard/salary_compute_wizard.py
2026-07-06 10:30:41 +08:00

554 lines
28 KiB
Python

# -*- coding: utf-8 -*-
from dateutil.relativedelta import relativedelta
from odoo import models, fields, api, _, SUPERUSER_ID
from odoo.exceptions import ValidationError, UserError
from odoo.exceptions import ValidationError
import calendar
from pypinyin import pinyin, Style
import math
import xlrd
import base64
import datetime
from datetime import datetime, timedelta
class SalaryComputeWizard(models.Model):
_name = 'salary.compute.wizard'
_description = '薪资计算弹窗'
employee_ids = fields.Many2many('hr.employee', string='人员', store=True)
company_id = fields.Many2one('res.company', string="公司", default=lambda self: self.env.company)
employee_number = fields.Integer(string="人数")
is_empty = fields.Boolean(string="清空人员")
select_year = fields.Selection(selection=[(str(i), str(i)) for i in range(2024, 2060)], string='年份',
default=lambda self: str(datetime.now().year))
select_time = fields.Selection([('1', '一月'), ('2', '二月'), ('3', '三月'), ('4', '四月'),
('5', '五月'), ('6', '六月'), ('7', '七月'), ('8', '八月'),
('9', '九月'), ('10', '十月'), ('11', '十一月'),
('12', '十二月')], string='月份',
default=lambda self: self._get_default_month())
def _get_default_month(self):
current_month = datetime.now().month
return str(current_month)
@api.onchange('is_empty')
def onchange_is_empty(self):
if self.is_empty:
self.employee_ids = [(5,)]
@api.onchange('select_year', 'select_time')
def onchange_select_year_select_time(self):
year = int(self.select_year)
month = int(self.select_time)
# 计算当月第一天和最后一天
first_day = datetime(year, month, 1).strftime('%Y-%m-%d')
last_day = datetime(year, month, 15).strftime('%Y-%m-%d')
if self.company_id:
employees = self.env['hr.employee'].search([
('company_id', '=', self.company_id.id),
('is_salary', '=', True),
('entry_date', '<=', last_day),
('contract_id', '!=', False),
'|',
('contract_id.termination_date', '>=', first_day),
'|',
'&',
('contract_id.termination_date', '=', False),
('contract_id.date_end', '>=', first_day),
'&',
('contract_id.termination_date', '=', False),
('contract_id.date_end', '=', False)
])
if month == 12:
next_year = year + 1
next_month = 1
else:
next_year = year
next_month = month + 1
last_day1 = (datetime(next_year, next_month, 1) - timedelta(days=1)).strftime('%Y-%m-%d')
existing_employees = self.env['hr.payslip'].search([
('date_from', '>=', first_day),
('date_to', '<=', last_day1),
('company_id', '=', self.company_id.id)
]).mapped('employee_id')
new_employees = employees - existing_employees
self.employee_ids = [(6, 0, new_employees.ids)]
self.employee_number = len(employees)
for employee in employees:
print(f"员工姓名: {employee.name}")
@api.onchange('employee_ids')
def onchange_employee_ids(self):
if self.employee_ids:
self.employee_number = len(self.employee_ids)
def confirm(self):
for employee in self.employee_ids:
# detail_id = self.env['hr.attendance.month.detail'].search([('employee_id', '=', employee.id),
# ('current_year', '=', self.select_year),
# ('current_month', '=', self.select_time)])
# if detail_id.state == 'unconfirmed':
# raise UserError(f'{employee.name}的{self.select_time}考勤还未确认,请先确认考勤再进行工资计算!')
is_bold = self.check_employee_dates(employee)
contract_id = self.env['hr.contract'].search([('employee_id', '=', employee.id)], limit=1)
year = int(self.select_year)
month = int(self.select_time)
payslip_run_name = f"{year}年-{self._get_chinese_month(month)}"
start_date = datetime(year, month, 1).strftime('%Y-%m-%d')
if month in [1, 3, 5, 7, 8, 10, 12]:
end_day = 31
elif month in [4, 6, 9, 11]:
end_day = 30
elif (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0):
end_day = 29
else:
end_day = 28
end_date = datetime(year, month, end_day).strftime('%Y-%m-%d')
payslip_run_id = self.env['hr.payslip.run'].search([('name', '=', payslip_run_name), ('company_id', '=', employee.company_id.id)], limit=1)
if not payslip_run_id:
payslip_run_id = self.env['hr.payslip.run'].create({
'company_id': employee.company_id.id,
'name': payslip_run_name,
'date_start': start_date,
'date_end': end_date,
'state': 'draft',
'payslip_count': 0
})
query_date = datetime(year, month, 1).date() - relativedelta(months=1)
vacation_report = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', employee.id),
('vacation_date', '=', query_date)
], limit=1)
if vacation_report:
update_vals = {
'done_leave': vacation_report.done_leave, # 年假
'done_working_overtime': vacation_report.done_working_overtime, # 加班时
'compensatory_leave': vacation_report.compensatory_leave, # 调休
'leave_of_absence': vacation_report.leave_of_absence, # 事假
'statutory_leave': vacation_report.statutory_leave, # 法定加班
'cop_leave': vacation_report.cop_leave, # 补休
'duty_off_leave': vacation_report.duty_off_leave, # 值班调休
'marriage_leave': vacation_report.marriage_leave, # 婚假
'maternity_leave': vacation_report.maternity_leave, # 产假
'paternity_leave': vacation_report.paternity_leave, # 陪产假
'family_planning_leave': vacation_report.family_planning_leave, # 计生假
'bereavement_leave': vacation_report.bereavement_leave, # 丧假
'home_visit_leave': vacation_report.home_visit_leave, # 探亲假
'parents_meeting_leave': vacation_report.parents_meeting_leave, # 家长会假
'general_sick_leave': vacation_report.general_sick_leave, # 普通病假
'hospital_sick_leave': vacation_report.hospital_sick_leave, # 住院病假
'recovery_sick_leave': vacation_report.recovery_sick_leave, # 出院恢复期病假
'exam_leave': vacation_report.exam_leave, # 考试假
}
# 更新员工花名册记录
employee.write(update_vals)
# 加班时长
employee.last_month_overtime = vacation_report.done_working_overtime
employee.last_attendance = vacation_report.should_attendance == vacation_report.actual_attendance
employee.meal_number = vacation_report.meal_fee
employee.salary_date = start_date
employee.salary_month = self.select_time
employee.salary_year_jin_ids = [(6, 0, self.employee_ids.ids)]
# 查找有效合同:date_start <= start_date <= date_end(或date_end为空)
contract_id = self.env['hr.contract'].search([
('employee_id', '=', employee.id), ('date_start', '<=', start_date), '|', ('date_end', '>=', start_date), ('date_end', '=', False),
], limit=1)
existing_payslip = self.env['hr.payslip'].search([('employee_id', '=', employee.id), ('payslip_run_id', '=', payslip_run_id.id),], limit=1)
if not existing_payslip:
existing_payslip = self.env['hr.payslip'].create({
'is_bold': is_bold,
'name': employee.name,
'employee_id': employee.id,
'contract_id': contract_id.id or employee.contract_id.id,
'date_from': start_date,
'date_to': end_date,
'salary_group_id': contract_id.salary_group_id.id if contract_id else employee.contract_id.salary_group_id.id,
'payslip_run_id': payslip_run_id.id,
'state': 'draft',
})
existing_payslip.compute_sheet()
internship_ids = self.env['hr.payslip'].search([
('is_push', '=', False),
('company_id', '=', self.company_id.id),
('date_from', '<=', end_date),
('date_to', '>=', start_date),
])
for internship_id in internship_ids:
employee = internship_id.employee_id
contract = employee.contract_id
year = int(self.select_year)
month = int(self.select_time)
# 获取离职日期
turnover_date = contract.date_end
start_date = fields.Date.from_string(start_date)
end_date = fields.Date.from_string(end_date)
is_turnover = self._is_employee_turnover(employee, year, month)
is_probation = self._is_employee_in_probation(employee, year, month)
original_start_date = internship_id.date_from
original_end_date = internship_id.date_to
if is_turnover or is_probation:
days = self._process_shift_days(original_start_date, original_end_date) # 总工作日
if is_turnover:
# 离职日期前的工作日
pre_retirement_days = self._process_shift_days(
start_date,
min(end_date, turnover_date)
)
work_days = pre_retirement_days # 离职前的工作日
elif is_probation:
contract_start_date = internship_id.employee_id.contract_id.start_date
contract_end_date = internship_id.employee_id.contract_id.end_date
start_date = original_start_date
end_date = original_end_date
internship_start = max(original_start_date, contract_start_date)
internship_end = min(original_end_date, contract_end_date)
last_days = self._process_shift_days(internship_start - relativedelta(months=1),internship_start - relativedelta(days=1))
# 判断该员工是否需补发上月工资
if contract_start_date.year == internship_start.year and contract_start_date.month == (internship_start - relativedelta(months=1)).month and contract_start_date.day > 15:
last_internship_days = self._process_shift_days(contract_start_date, internship_start - relativedelta(days=1))
else:
last_internship_days = 0
if internship_start <= internship_end:
if contract_end_date >= original_end_date:
# 整月都是实习期 # todo 21.75 待定
internship_days = self._process_shift_days(internship_start, internship_end)
work_days = internship_days
else:
# 部分时间是实习期,计算实际工作日
internship_days = self._process_shift_days(internship_start, internship_end)
work_days = 21 - internship_days
if internship_id.line_ids:
if is_probation and not internship_id.employee_id.is_annual_salary:
target_names = {
'基本工资': 'base_salary', # 0.8
'职级工资': 'rank_salary', # 0.8
'岗位工资': 'position_salary',
'全勤奖': 'full_attendance_award',
'月绩效奖金': 'performance_bonuses', # 0.8
'通讯补贴': 'commuting_allowance',
'工龄工资': 'seniority_salary',
'工会会费': 'union_dues',
'住房公积金': 'housing_provident_fund',
'企业(职业)年金': 'b_year', # 企业年金
}
# 定义比例常量,便于统一维护
RATIO_80 = 0.8
for line in internship_id.line_ids:
if line.name not in target_names:
continue
field_name = target_names[line.name]
result = 0.0 # 初始化浮点型结果
# 计算80%比例的基础值(提前处理ceil逻辑)向上取整
base_80 = math.ceil(line.total * RATIO_80) if field_name in ['rank_salary', 'base_salary', 'performance_bonuses'] else (line.total * RATIO_80)
if field_name == 'performance_bonuses':
if work_days != internship_days:
result = (base_80 * internship_days + line.total * work_days) / days
else:
result = base_80 * internship_days / days + base_80 * last_internship_days / last_days
result = math.ceil(result * 100) / 100
elif field_name == 'full_attendance_award':
result = line.total * last_internship_days / last_days
elif field_name in ['commuting_allowance', 'seniority_salary']:
if work_days != internship_days:
result = math.ceil(( line.total * internship_days + line.total * work_days) / days * 100) / 100
else:
result = math.ceil((line.total * internship_days / days + line.total * last_internship_days / last_days) * 100) / 100
elif field_name in ['rank_salary', 'base_salary']:
if work_days != internship_days:
result = ( base_80 * internship_days + line.total * work_days + line.total * last_internship_days * RATIO_80) / days
else:
result = base_80 * internship_days / days + base_80 * last_internship_days / last_days
result = math.ceil(result * 100) / 100
elif field_name == 'union_dues': # 单独处理工会会费
base_salary = getattr(internship_id, 'base_salary', 0)
result = abs(base_salary * 0.005) * -1
elif field_name == 'housing_provident_fund': # 单独处理住房公积金
# 获取合同入职日期的年份和月份
contract_year = str(contract_start_date.year)
contract_month = str(contract_start_date.month)
# 判断是否为入职当月
if contract_year == self.select_year and contract_month == self.select_time:
result = 0 # 入职当月住房公积金为0
# 新增:非入职当月时的默认值(根据业务需求调整)
else:
result = abs(line.total) * -1 # 或其他合理默认值
elif field_name == 'b_year':
if contract_end_date:
select_year = int(self.select_year)
select_month = int(self.select_time)
next_month = contract_end_date.month + 1
next_year = contract_end_date.year
if next_month > 12:
next_month = 1
next_year += 1
next_year = int(next_year)
next_month = int(next_month)
# 判断:选中年月 >= 转正后下个月
if (select_year > next_year) or (select_year == next_year and select_month >= next_month):
last_year_salary = internship_id.employee_id.last_year_salary or 0
result = last_year_salary * 0.002
else:
# 不满足条件(转正当月或之前)
result = 0
else:
result = 0
else:
if work_days != days:
result = line.total * 0.8 * internship_days / days + line.total * work_days / days
else:
result = line.total * 0.8 * internship_days / days
setattr(internship_id, field_name, result)
line.write({'total': result})
if is_turnover:
target_names = {
'基本工资': 'base_salary',
'岗位工资': 'position_salary',
'通讯补贴': 'commuting_allowance',
'综合补贴': 'comprehensive_subsidy',
'全勤奖': 'full_attendance_award',
}
for line in internship_id.line_ids:
if line.name in target_names:
field_name = target_names[line.name]
# 统一按公式计算:金额/21.75 * 离职前工作日数量,并保留两位小数
result = round((line.total / 21.75) * work_days, 2)
setattr(internship_id, field_name, result)
line.write({'total': result})
# 计算扣发绩效奖金
performance_deduction = self.salary_performance_deduction(vacation_report, internship_id.performance_bonuses)
employee.write({
'last_month_performance_deduction': performance_deduction,
})
# 计算扣发基本薪资
basic_salary_deduction = self.salary_basic_salary_deduction(vacation_report, internship_id.base_salary)
employee.write({
'last_month_base_salary_deduction': basic_salary_deduction,
})
# 计算扣发通讯
commuting_deduction = self.salary_commuting_deduction(vacation_report, internship_id.commuting_allowance)
employee.write({
'last_month_commuting_deduction': commuting_deduction,
})
year = int(self.select_year)
month = int(self.select_time)
if not self._is_employee_turnover(employee, year, month) and not self._is_employee_in_probation(employee, year, month):
existing_payslip.compute_sheet()
def _get_chinese_month(self, month):
chinese_months = ['一月', '二月', '三月', '四月', '五月', '六月', '七月', '八月', '九月', '十月', '十一月', '十二月']
return chinese_months[month - 1]
def _process_shift_days(self, start_date, end_date):
current_year = (fields.Date.today() + timedelta(hours=8)).year
query_start = datetime(current_year, 1, 1, 0, 0, 0) - timedelta(hours=8)
query_end = datetime(current_year, 12, 31, 23, 59, 59) - timedelta(hours=8)
calendar_leaves = self.env['resource.calendar.leaves'].sudo().search([
("calendar_id.name", "=", "行政考勤工作日"),
("date_from", ">=", query_start),
("date_from", "<=", query_end),
('resource_id', '=', False),
])
holidays_list = []
shift_days_list = []
# 提取节假日和调休日
for leave in calendar_leaves:
if leave.holidays:
holidays_list.extend(leave.holidays.split(','))
if leave.alternate_rest_day:
holidays_list.extend(leave.alternate_rest_day.split(','))
if leave.shift_replacement_day:
shift_days_list.extend(leave.shift_replacement_day.split(','))
holidays_list = [self.remove_leading_zero(date) for date in holidays_list]
shift_days_list = [self.remove_leading_zero(date) for date in shift_days_list]
days = 0 # 初始化工作日计数
current_date = start_date
while current_date <= end_date:
current_date_str = current_date.strftime('%m月%d日')
current_date_str = self.remove_leading_zero(current_date_str)
is_weekend = current_date.weekday() in [5, 6]
if current_date_str in holidays_list:
# 如果是法定节假日,不计为工作日
pass
elif current_date_str in shift_days_list:
# 如果是调休日,无论是否周末都算作工作日
days += 1
elif not is_weekend:
# 如果不是周末且不是节假日,算作工作日
days += 1
current_date += timedelta(days=1)
return days
def remove_leading_zero(self, date_str):
month, day = date_str.split('月')
month = month.lstrip('0')
day = day.rstrip('日')
day = day.lstrip('0')
return f'{month}月{day}日'
def check_employee_dates(self, employee):
"""检查员工为试用期入职日期和转正日期的月份 和退休(花名册)的是否等于 select_time 的月份"""
if not self.select_time:
return False # 未选择月份时返回 False
selected_month = int(self.select_time)
selected_year = int(self.select_year)
if employee.contract_id.is_probation:
entry_matches = False
if employee.entry_date:
entry_matches = (
employee.entry_date.month == selected_month and
employee.entry_date.year == selected_year
)
confirmation_matches = False
if employee.confirmation_date:
confirmation_matches = (
employee.confirmation_date.month == selected_month and
employee.confirmation_date.year == selected_year
)
# 检查退休日期
retire_matches = False
if employee.retire_date:
retire_matches = (
employee.retire_date.month == selected_month and
employee.retire_date.year == selected_year
)
# 满足任一条件即返回 True
return entry_matches or confirmation_matches or retire_matches
def _is_employee_turnover(self, employee, year, month):
"""判断员工是否在指定年月已离职"""
turnover_date = employee.contract_id.date_end
if not turnover_date:
return False
turnover_year = turnover_date.year
turnover_month = turnover_date.month
return (turnover_year < year) or (turnover_year == year and turnover_month <= month)
def _is_employee_in_probation(self, employee, year, month):
"""判断员工在指定年月是否处于试用期"""
contract = employee.contract_id
if not (contract and contract.start_date and contract.end_date and contract.is_probation):
return False
prob_start_year = contract.start_date.year
prob_start_month = contract.start_date.month
prob_end_year = contract.end_date.year
prob_end_month = contract.end_date.month
# 检查指定年月是否在试用期的年月范围内
if (prob_start_year < year < prob_end_year):
# 年份在试用期中间的所有月份都算试用期
return True
elif prob_start_year == year and prob_end_year == year:
# 同一年内,检查月份是否在范围内
return prob_start_month <= month <= prob_end_month
elif prob_start_year == year:
# 起始年份,检查月份是否大于等于起始月份
return month >= prob_start_month
elif prob_end_year == year:
# 结束年份,检查月份是否小于等于结束月份
return month <= prob_end_month
return False
def salary_performance_deduction(self, vacation_report, performance_bonuses):
if not (vacation_report and performance_bonuses):
return 0.00
deduction_leave_fields = [
'marriage_leave', # 婚假
'maternity_leave', # 产假
'paternity_leave', # 陪产假
'family_planning_leave', # 计生假
'bereavement_leave', # 丧假
'home_visit_leave', # 探亲假
'general_sick_leave', # 病假(普通)
'hospital_sick_leave', # 病假(住院期)
'recovery_sick_leave', # 病假(出院恢复期)
'exam_leave', # 考试假
'parents_meeting_leave', # 家长会假
'leave_of_absence' # 事假
]
total_leave_days = 0.0
for field in deduction_leave_fields:
leave_days = getattr(vacation_report, field, 0.0)
total_leave_days += leave_days if leave_days > 0 else 0.0
if total_leave_days <= 0:
return 0.00
return round((performance_bonuses / 21.75) * total_leave_days, 2)
def salary_basic_salary_deduction(self, vacation_report, base_salary):
if not (vacation_report and base_salary > 0):
return 0.00
deduction_rules = {
'general_sick_leave': 1.0, # 普通病假:1%
'recovery_sick_leave': 1.0, # 出院恢复期病假:1%
'leave_of_absence': 2.5 # 事假:2.5%
}
total_deduction = 0.0
for leave_field, ratio in deduction_rules.items():
leave_days = getattr(vacation_report, leave_field, 0.0)
if leave_days > 0:
single_deduction = base_salary * leave_days * (ratio / 100)
total_deduction += single_deduction
return round(total_deduction, 2)
def salary_commuting_deduction(self, vacation_report, communication_subsidy):
if not (vacation_report and communication_subsidy > 0):
return 0.00
maternity_days = vacation_report.maternity_leave # 产假天数
paternity_days = vacation_report.paternity_leave # 陪产假天数
family_planning_days = vacation_report.family_planning_leave # 计生假天数
total_days = sum([
maternity_days if maternity_days > 0 else 0,
paternity_days if paternity_days > 0 else 0,
family_planning_days if family_planning_days > 0 else 0
])
# 通讯补贴 / 21.75 * 总请假天数
if total_days <= 0:
return 0.00
deduction_amount = (communication_subsidy / 21.75) * total_days
return round(deduction_amount, 2)