Files
gzth/yuthon_holiday/models/inherit_hr_leave.py
T
2026-03-02 10:51:37 +08:00

1149 lines
55 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
from datetime import datetime, timedelta, time
from email.policy import default
from dateutil.relativedelta import relativedelta
from odoo import models, fields, api, _
from odoo.exceptions import UserError, ValidationError
import re
from odoo.tools.misc import format_date
from collections import namedtuple, defaultdict
from datetime import date
import calendar
import logging
class HrLeave(models.Model):
_inherit = 'hr.leave'
number_of_days = fields.Float(string='天数', digits=(16, 3))
def _check_validity(self):
sorted_leaves = defaultdict(lambda: self.env['hr.leave'])
for leave in self:
sorted_leaves[(leave.holiday_status_id, leave.date_from.date())] |= leave
for (leave_type, date_from), leaves in sorted_leaves.items():
if leave_type.requires_allocation == 'no':
continue
employees = self.env['hr.employee']
for leave in leaves:
employees |= leave._get_employees_from_holiday_type()
leave_data = leave_type.get_allocation_data(employees, date_from)
if leave_type.allows_negative:
max_excess = leave_type.max_allowed_negative
for employee in employees:
if leave_data[employee] and leave_data[employee][0][1]['virtual_remaining_leaves'] < -max_excess:
raise ValidationError(_("There is no valid allocation to cover that request."))
continue
previous_leave_data = leave_type.with_context(ignored_leave_ids=leaves.ids).get_allocation_data(employees, date_from)
for employee in employees:
previous_emp_data = previous_leave_data[employee] and previous_leave_data[employee][0][1]['virtual_excess_data']
emp_data = leave_data[employee] and leave_data[employee][0][1]['virtual_excess_data']
if not previous_emp_data and not emp_data:
continue
# 这里加一个假期额度的判断
# if previous_emp_data != emp_data and len(emp_data) >= len(previous_emp_data):
# raise ValidationError("您的假期额度不足,无法申请该假期")
leave_hours = fields.Float(string='请假小时', default=0.0)
@api.onchange('request_hour_from', 'leave_hours')
def _onchange_request_hour_to(self):
if not (self.request_hour_from and self.leave_hours is not None):
self.request_hour_to = False
return
start_hour = float(self.request_hour_from)
leave_hours = float(self.leave_hours)
end_hour = (start_hour + leave_hours) % 24.0
end_hour_rounded = round(end_hour * 2) / 2
if end_hour_rounded.is_integer():
end_hour_key = str(int(end_hour_rounded))
else:
end_hour_key = str(end_hour_rounded)
selection_keys = [k for k, v in self._fields['request_hour_to'].selection]
if end_hour_key in selection_keys:
self.request_hour_to = end_hour_key
else:
self.request_hour_to = False
@api.ondelete(at_uninstall=False)
def _unlink_if_correct_states(self):
"""重写删除校验:先将请假单改为拒绝状态,再执行原生校验/删除"""
to_cancel_leaves = self.filtered(lambda r: r.state != 'draft')
if to_cancel_leaves:
to_cancel_leaves.write({
'state': 'draft',
'notes': _('%s:删除前自动标记为拒绝') % fields.Datetime.now()
})
error_message = _('You cannot delete a time off which is in %s state')
state_description_values = {elem[0]: elem[1] for elem in self._fields['state']._description_selection(self.env)}
if not self.user_has_groups('hr_holidays.group_hr_holidays_user'):
for hol in self:
if hol.state not in ['draft', 'confirm', 'validate1', 'cancel']:
raise UserError(error_message % state_description_values.get(hol.state))
if hol.sudo().employee_ids and not hol.employee_id:
raise UserError(_('You cannot delete a time off assigned to several employees'))
else:
for holiday in self.filtered(lambda holiday: holiday.state not in ['draft', 'cancel', 'confirm']):
raise UserError(error_message % (state_description_values.get(holiday.state),))
@api.model
def write(self, vals):
if not self.env.context.get('leave_skip_state_check'):
self = self.with_context(leave_skip_state_check=True)
if 'state' not in vals and self.filtered(lambda r: r.state != 'draft'):
vals['state'] = 'draft'
return super(HrLeave, self).write(vals)
def action_leave_wizard(self):
return {
"type": "ir.actions.act_window",
"name": "请假单",
"res_model": "yuthon.leave.wizard",
"view_mode": "form",
"target": "new",
}
@api.model
def default_get(self, fields_list):
default = super().default_get(fields_list)
# 删除default的holiday_status_id键
if 'holiday_status_id' in default and 'holiday_status_id_default_request_unit' in self.env.context:
if self.env.context.get('holiday_status_id_default_request_unit') == 'hour':
search_domain = [
('request_unit', '=', 'hour'),
'|',
('requires_allocation', '=', 'no'),
'&',
('has_valid_allocation', '=', True),
'|',
('allows_negative', '=', True),
'&',
('virtual_remaining_leaves', '>', 0),
('allows_negative', '=', False),
]
else:
search_domain = [
('request_unit', '=', 'day'),
'|',
('requires_allocation', '=', 'no'),
'&',
('has_valid_allocation', '=', True),
'|',
('allows_negative', '=', True),
'&',
('virtual_remaining_leaves', '>', 0),
('allows_negative', '=', False),
]
lt = self.env['hr.leave.type'].search(search_domain, limit=1, order='sequence')
if lt:
default['holiday_status_id'] = lt.id
default['request_unit_custom'] = False
return default
batch_id = fields.Many2one(string='批量', comodel_name='hr.leave.batch', ondelete='cascade')
hour_batch_id = fields.Many2one(string='批量', comodel_name='hr.leave.batch', ondelete='cascade')
leave_documents_ids = fields.Many2many(related='holiday_status_id.leave_documents_id', string="假期详情说明", readonly=True)
details = fields.Text(string="假期流程说明")
# custom_document_ids = fields.Many2many(
# string='文件',
# comodel_name='ir.attachment',
# relation='leave_custom_doc_rel',
# column1='leave_id',
# column2='attachment_id'
# )
request_hour_from = fields.Selection([
('0', '0:00'), ('0.5', '0:30'),
('1', '01:00'), ('1.5', '01:30'),
('2', '02:00'), ('2.5', '02:30'),
('3', '03:00'), ('3.5', '03:30'),
('4', '04:00'), ('4.5', '04:30'),
('5', '05:00'), ('5.5', '05:30'),
('6', '06:00'), ('6.5', '06:30'),
('7', '07:00'), ('7.5', '07:30'),
('8', '08:00'), ('8.5', '08:30'),
('9', '09:00'), ('9.5', '09:30'),
('10', '10:00'), ('10.5', '10:30'),
('11', '11:00'), ('11.5', '11:30'),
('12', '12:00'), ('12.5', '12:30'),
('13', '13:00'), ('13.5', '13:30'),
('14', '14:00'), ('14.5', '14:30'),
('15', '15:00'), ('15.5', '15:30'),
('16', '16:00'), ('16.5', '16:30'),
('17', '17:00'), ('17.5', '17:30'),
('18', '18:00'), ('18.5', '18:30'),
('19', '19:00'), ('19.5', '19:30'),
('20', '20:00'), ('20.5', '20:30'),
('21', '21:00'), ('21.5', '21:30'),
('22', '22:00'), ('22.5', '22:30'),
('23', '23:00'), ('23.5', '23:30')], string='开始时间')
request_hour_to = fields.Selection([
('0', '12:00'), ('0.5', '12:30'),
('1', '01:00'), ('1.5', '01:30'),
('2', '02:00'), ('2.5', '02:30'),
('3', '03:00'), ('3.5', '03:30'),
('4', '04:00'), ('4.5', '04:30'),
('5', '05:00'), ('5.5', '05:30'),
('6', '06:00'), ('6.5', '06:30'),
('7', '07:00'), ('7.5', '07:30'),
('8', '08:00'), ('8.5', '08:30'),
('9', '09:00'), ('9.5', '09:30'),
('10', '10:00'), ('10.5', '10:30'),
('11', '11:00'), ('11.5', '11:30'),
('12', '12:00'), ('12.5', '12:30'),
('13', '13:00'), ('13.5', '13:30'),
('14', '14:00'), ('14.5', '14:30'),
('15', '15:00'), ('15.5', '15:30'),
('16', '16:00'), ('16.5', '16:30'),
('17', '17:00'), ('17.5', '17:30'),
('18', '18:00'), ('18.5', '18:30'),
('19', '19:00'), ('19.5', '19:30'),
('20', '20:00'), ('20.5', '20:30'),
('21', '21:00'), ('21.5', '21:30'),
('22', '22:00'), ('22.5', '22:30'),
('23', '23:00'), ('23.5', '23:30')], string='结束时间')
def update_state(self):
leave_ids = self.env["hr.leave"].browse(
self._context.get('active_ids', self._context.get('active_id')))
for leave in leave_ids:
leave.state = 'draft'
@api.depends('date_from', 'date_to', 'resource_calendar_id', 'holiday_status_id.request_unit')
def _compute_duration(self):
super()._compute_duration()
for holiday in self:
# 如果是自然日:
if holiday.holiday_status_id.type_of_work == "1":
if holiday.batch_id:
days = (holiday.request_date_to - holiday.request_date_from).days + 1
hours = days * 8
else:
hours = (float(holiday.request_hour_to) - float(holiday.request_hour_from))
holiday.number_of_hours = hours
holiday.leave_hours = hours
if holiday.holiday_status_id.type_of_work == "0":
if holiday.request_hour_from and holiday.request_hour_to:
start_hour = float(holiday.request_hour_from)
end_hour = float(holiday.request_hour_to)
if start_hour == 9 and end_hour == 12:
holiday.number_of_hours = 4
holiday.leave_hours = 4
else:
hour = end_hour - start_hour
holiday.number_of_hours = hour
holiday.leave_hours = hour
else:
holiday.number_of_hours = 0
holiday.leave_hours = 0
# 按天请假
days = holiday.batch_id._compute_duration_day(holiday.holiday_status_id, holiday.request_date_from , holiday.request_date_to)
holiday.number_of_days = days
def _compute_request_unit_hours(self):
if not self.env.context.get('no_change_request_unit_hours'):
super()._compute_request_unit_hours()
@api.constrains('date_from', 'date_to', 'employee_id')
def _check_date(self):
if self.env.context.get('leave_skip_date_check', False):
return
all_employees = self.all_employee_ids
all_leaves = self.search([
('date_from', '<', max(self.mapped('date_to'))),
('date_to', '>', min(self.mapped('date_from'))),
('employee_id', 'in', all_employees.ids),
('id', 'not in', self.ids),
('state', 'not in', ['cancel', 'refuse']),
])
for holiday in self:
domain = [
('date_from', '<', holiday.date_to),
('date_to', '>', holiday.date_from),
('id', '!=', holiday.id),
('state', 'not in', ['cancel', 'refuse']),
]
employee_ids = (holiday.employee_id | holiday.employee_ids).ids
search_domain = domain + [('employee_id', 'in', employee_ids)]
conflicting_holidays = all_leaves.filtered_domain(search_domain)
if conflicting_holidays:
conflicting_holidays_list = []
# Do not display the name of the employee if the conflicting holidays have an employee_id.user_id equivalent to the user id
holidays_only_have_uid = bool(holiday.employee_id)
holiday_states = dict(conflicting_holidays.fields_get(allfields=['state'])['state']['selection'])
for conflicting_holiday in conflicting_holidays:
conflicting_holiday_data = {}
conflicting_holiday_data['employee_name'] = conflicting_holiday.employee_id.name
conflicting_holiday_data['date_from'] = format_date(self.env,
min(conflicting_holiday.mapped('date_from')))
conflicting_holiday_data['date_to'] = format_date(self.env,
min(conflicting_holiday.mapped('date_to')))
conflicting_holiday_data['state'] = holiday_states[conflicting_holiday.state]
if conflicting_holiday.employee_id.user_id.id != self.env.uid:
holidays_only_have_uid = False
if conflicting_holiday_data not in conflicting_holidays_list:
conflicting_holidays_list.append(conflicting_holiday_data)
if not conflicting_holidays_list:
return
conflicting_holidays_strings = []
if holidays_only_have_uid:
for conflicting_holiday_data in conflicting_holidays_list:
conflicting_holidays_string = _('%(date_from)s - %(date_to)s',
date_from=conflicting_holiday_data['date_from'],
date_to=conflicting_holiday_data['date_to'])
conflicting_holidays_strings.append(conflicting_holidays_string)
raise ValidationError(_("""\
您已有:「%s」 的假期申请待审批,请勿重复提交。如需调整,请先联系审批人或撤回原申请。
""",
"\n".join(conflicting_holidays_strings)))
super()._check_date()
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}日'
_logger = logging.getLogger(__name__)
class HolidaysRequestBatch(models.Model):
_name = 'hr.leave.batch'
_description = '请假审批'
_inherit = ['mail.thread.main.attachment', 'mail.activity.mixin', 'soong.workflow.mixin']
code = fields.Char(string="编号", default=lambda self: self.env['ir.sequence'].next_by_code('yuthon_holiday_code'))
name = fields.Char(string='标题')
create_date = fields.Date(string='创建日期', default=fields.Datetime.today())
employee_id = fields.Many2one(related='users_id.employee_id', string="员工", store=True)
company_id = fields.Many2one(related='users_id.employee_id.company_id', string="公司", store=True)
users_id = fields.Many2one('res.users', string="请假人", default=lambda self: self.env.user.id)
department_id = fields.Many2one(related='users_id.employee_id.department_id', string="部门")
user_role_id = fields.Many2one(related='users_id.employee_id.user_role_id', string="角色")
details = fields.Text(string="假期流程说明", compute='_compute_details')
days = fields.Float(string='天数')
hours = fields.Float(string='小时')
text = fields.Text(string='申请事由')
is_day = fields.Boolean(string='按天请假')
is_hour = fields.Boolean(string='按小时请假')
leave_ids = fields.One2many(string='按天请假详情', comodel_name='hr.leave', inverse_name='batch_id')
hour_leave_ids = fields.One2many(string='按小时请假详情', comodel_name='hr.leave', inverse_name='hour_batch_id')
is_issues = fields.Boolean(string='是否上会')
work_end = fields.Boolean(string="结束")
issues_result = fields.Html(string="会议结果")
entrusted_agent_id = fields.Many2one('res.users', string="委托代理人")
is_entrusted = fields.Boolean(string="是否生成工作流委托")
leave_by_hour = fields.Boolean(string='请假单位(小时)', default=False)
leave_by_day = fields.Boolean(string='请假单位(天)', default=True)
urgency_level = fields.Selection([('normal', '普件'), ('urgent', '急件'), ('emergency', '特急件')], string='紧急度',
default='normal')
start_date = fields.Date(string="开始时间")
end_date = fields.Date(string="结束时间")
document_ids = fields.Many2many(
string='附件',
comodel_name='ir.attachment',
relation='leave_custom_doc_rel',
column1='leave_id',
column2='attachment_id'
)
holiday_allowance = fields.Text(string="可用额度", compute='_compute_holiday_allowance')
active = fields.Boolean('Active', default=True, tracking=True)
def _get_default_name_title(self):
return self._description if self._description else self._name
name_title = fields.Char(default=_get_default_name_title)
self_users_ids = fields.Many2many('res.users', string="当前经办人", default=lambda self: [(6, 0, [self.env.user.id])])
@api.onchange('is_day')
def _onchange_is_day(self):
if not self.is_day:
self.leave_ids = [(5, 0, 0)]
@api.onchange('is_hour')
def _onchange_is_hour(self):
if not self.is_hour:
self.hour_leave_ids = [(5, 0, 0)]
@api.depends("users_id")
def _compute_details(self):
for record in self:
holiday_id = record.env['yuthon.holiday.setting'].search([('department_ids', 'in', record.users_id.employee_id.department_id.id)], limit=1)
if holiday_id:
record.details = holiday_id.name
else:
record.details = " "
# 提交
def action_submit(self):
if self.is_entrusted:
leave_batch_model = self.env['ir.model'].search(
[('model', '=', 'hr.leave.batch')],
limit=1
)
entrust_vals = {
'model_ids': [(6, 0, [leave_batch_model.id])],
'entrust_user_id': self.users_id.id,
'entrust_user_id2': self.entrusted_agent_id.id,
'is_expiration_date': True,
'start_date': fields.Date.today(),
'end_date': False,
'entrust_notice': True,
}
self.env['soong.workflow.entrust'].create(entrust_vals)
if self.leave_ids:
for record in self.leave_ids:
record.sudo().write({'state': 'validate'})
if self.hour_leave_ids:
for record in self.hour_leave_ids:
record.sudo().write({'state': 'validate'})
"""提交请假申请,区分年假和其他假期处理"""
self._process_specific_leaves('年假', is_annual_leave=True)
other_leave_types = [
'产假', '事假', '法定假', '调休', '补休', '值班调休',
'婚假', '陪产假', '计生假', '丧假', '探亲假', '哺乳假',
'家长会假', '病假(普通)', '病假(住院期)', '育儿假',
'病假(出院恢复期)', '考试假'
]
for leave_type in other_leave_types:
self._process_specific_leaves(leave_type, is_annual_leave=False)
return True
def _process_specific_leaves(self, leave_type_name, is_annual_leave):
"""
处理指定类型假期
:param leave_type_name: 假期类型名称
:param is_annual_leave: 是否为年假(True/False)
"""
# 收集天假和小时假记录
day_leaves = self._get_day_leaves(leave_type_name)
hourly_leaves = self._get_hourly_leaves(leave_type_name)
# 处理天假
for leave in day_leaves:
date_ranges = self._split_by_month(leave['start'], leave['end']) if not is_annual_leave else [
{'start': leave['start'], 'end': leave['end']}]
for period in date_ranges:
if leave_type_name == '补休':
# 计算小时数(天数 * 8)
hours = self._compute_duration_day(leave['type_id'], period['start'], period['end']) * 8
if hours > 0:
self._create_overtime_leave_line(period_start=period['start'], leave_data=leave, hours=hours, period=period, leave_type_name=leave_type_name)
else:
days = self._compute_duration_day(leave['type_id'], period['start'], period['end'])
if days > 0:
if is_annual_leave:
# 年假以小时为单位(天数 * 8)
self._create_annual_leave_line(period_start=period['start'], leave_data=leave, hours=days * 8, period=period)
else:
self._create_other_leave_line(period_start=period['start'], leave_data=leave, days=days, period=period, leave_type_name=leave_type_name)
# 更新考勤
self.update_attendance(self.users_id.employee_id, leave['start'], leave['end'], self.code)
# 处理小时假
for leave in hourly_leaves:
leave_date = leave['date']
hours = leave['days'] * 8
if hours > 0:
if is_annual_leave:
self._create_annual_leave_line(period_start=leave_date, leave_data=leave, hours=hours, period={'start': leave_date, 'end': leave_date})
elif leave_type_name == '补休':
self._create_overtime_leave_line(period_start=leave_date, leave_data=leave, hours=hours, period={'start': leave_date, 'end': leave_date}, leave_type_name=leave_type_name)
else:
# 其他类型小时假,使用天数
self._create_other_leave_line(period_start=leave_date, leave_data=leave, days=leave['days'], period={'start': leave_date, 'end': leave_date}, leave_type_name=leave_type_name)
# 更新考勤
self.update_attendance(self.users_id.employee_id, leave_date, leave_date, self.code)
def _get_day_leaves(self, leave_type_name):
workflow_id = self.env['soong.workflow.record'].search([('code', '=', self.code)], limit=1).id
return [{
'start': fields.Date.from_string(leave.request_date_from),
'end': fields.Date.from_string(leave.request_date_to),
'type_id': leave.holiday_status_id,
'workflow_id': workflow_id}
for leave in self.leave_ids
if leave.holiday_status_id.name == leave_type_name
and leave.request_date_from
and leave.request_date_to
]
def _get_hourly_leaves(self, leave_type_name):
workflow_id = self.env['soong.workflow.record'].search([('code', '=', self.code)], limit=1).id
hourly_records = []
for line in self.hour_leave_ids:
if line.holiday_status_id.name == leave_type_name and line.request_date_from:
hours_match = re.search(r'(\d+(?:\.\d+)?)', line.number_of_hours_text or '')
hours = float(hours_match.group(1)) if hours_match else 0
hourly_records.append({
'date': fields.Date.from_string(line.request_date_from),
'type_id': line.holiday_status_id,
'workflow_id': workflow_id,
'days': hours / 8
})
return hourly_records
def _split_by_month(self, start_date, end_date):
# 跨月拆分逻逻辑
date_ranges = []
if start_date.month == end_date.month and start_date.year == end_date.year:
date_ranges.append({'start': start_date, 'end': end_date})
return date_ranges
first_month_end = date(start_date.year, start_date.month,
calendar.monthrange(start_date.year, start_date.month)[1])
date_ranges.append({'start': start_date, 'end': first_month_end})
current_month = start_date.month + 1
current_year = start_date.year
while (current_year < end_date.year) or (current_year == end_date.year and current_month < end_date.month):
if current_month > 12:
current_month = 1
current_year += 1
month_end = date(current_year, current_month, calendar.monthrange(current_year, current_month)[1])
date_ranges.append({'start': date(current_year, current_month, 1), 'end': month_end})
current_month += 1
date_ranges.append({'start': date(end_date.year, end_date.month, 1), 'end': end_date})
return date_ranges
def _create_overtime_leave_line(self, period_start, leave_data, hours, period, leave_type_name):
"""
创建补休请假扣减记录(跨月自动拆分,分月创建)
:param period_start: 原始周期开始时间(datetime.date)
:param leave_data: 请假数据字典(含workflow_id等)
:param hours: 总请假小时数(float)
:param period: 请假时间段 {'start': date, 'end': date}
:param leave_type_name: 假期类型名称
"""
employee_id = self.users_id.employee_ids.id if self.users_id.employee_ids else False
if 'date' in leave_data:
leave_start = leave_data['date']
leave_end = leave_data['date']
else:
leave_start = leave_data['start'] if 'start' in leave_data else leave_data['end']
leave_end = leave_data['end']
month_intervals = []
current_start = leave_start
while current_start <= leave_end:
current_month = current_start.month
current_year = current_start.year
next_month = current_month + 1 if current_month < 12 else 1
next_year = current_year if current_month < 12 else current_year + 1
current_month_end = date(next_year, next_month, 1) - timedelta(days=1)
interval_end = min(current_month_end, leave_end)
month_intervals.append({
'start': current_start,
'end': interval_end,
'period_start': date(current_year, current_month, 1)
})
current_start = interval_end + timedelta(days=1)
holiday_status_id = self.env['hr.leave.type'].search([('name', '=', '补休')])
for interval in month_intervals:
interval_start = interval['start']
interval_end = interval['end']
if self.is_hour:
interval_hours = hours
else:
interval_hours = self._compute_duration_day(holiday_status_id, interval_start, interval_end) * 8
# 跳过小时数为0的无效区间
if interval_hours <= 1e-6:
continue
period_start = interval['period_start'] #
vacation_report1 = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', employee_id),
('vacation_date', '=', period_start),
], limit=1)
first_day = fields.Date.today().replace(day=1)
vacation_report = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', employee_id),
('vacation_date', '=', first_day),
], limit=1)
# 确定最终使用的报表
actual_vacation_report = vacation_report
if vacation_report1 and vacation_report1.vacation_state != 'confirmed':
actual_vacation_report = vacation_report1
if not actual_vacation_report:
continue
overtime_line_domain = [
('employee_id', '=', employee_id),
('workflow_record_id', '=', leave_data.get('workflow_id', False)),
('update_type', '=', 'reduce'),
('text', '=', f"请假审批"),
('is_void', '=', False),
('workday', '=', interval_hours),
]
existing_line_in_vacation_report = False
if vacation_report:
existing_line_in_vacation_report = self.env['vacation.over.line'].search(
[('vacation_id', '=', vacation_report.id)] + overtime_line_domain,
limit=1
)
existing_line_in_vacation_report1 = False
if vacation_report1:
existing_line_in_vacation_report1 = self.env['vacation.over.line'].search(
[('vacation_id', '=', vacation_report1.id)] + overtime_line_domain,
limit=1
)
# 4. 只有当两个报表中都没有该记录时,才创建新记录
if not existing_line_in_vacation_report and not existing_line_in_vacation_report1:
self.env['vacation.over.line'].create({
'vacation_id': actual_vacation_report.id,
'employee_id': employee_id,
'workflow_record_id': leave_data.get('workflow_id', False),
'update_type': 'reduce',
'workday': interval_hours,
'legal_day': 0,
'public_holidays': 0,
'meal_number': 0,
'update_date': fields.Date.today(),
'is_void': False,
'text': f"请假审批"
})
actual_vacation_report.update_line()
actual_vacation_report._compute_working_overtime()
if self.env.user.company_id.name == '集团本部':
domain = [
('employee_id', '=', employee_id),
"|",
('is_initial_number', '=', True),
'&', ('is_initial_number', '=', False),
'&', ('update_type', '=', 'add'),
'&', ('workflow_record_id', '!=', False),
'&', ('start_time', '>=', datetime(2025, 12, 31, 23, 59, 59)),
'&', ('vacation_id', '!=', False),
('employee_id.is_annual_salary', '!=', True)
]
else:
domain = [
('workflow_record_id', '!=', False),
('update_type', '=', 'add'),
('remaining_quantity', '!=', 0.0),
]
overtime_records = self.env['vacation.over.line'].search(domain, order='start_time ASC')
remaining_deduct = hours
for record in overtime_records:
if remaining_deduct <= 0:
break
current_remaining = record.remaining_quantity # 剩余数
deduct_amount = min(remaining_deduct, current_remaining)
if remaining_deduct < current_remaining:
carry_state = 'partial'
else:
carry_state = 'completed'
record.write({
'carry_forward_date': fields.Date.to_string(period_start),
'carry_forward_quantity': record.carry_forward_quantity + deduct_amount,
'carry_forward_state': carry_state,
})
remaining_deduct -= deduct_amount
def _create_annual_leave_line(self, period_start, leave_data, hours, period):
"""
创建年假扣减记录(跨月自动拆分,分月创建)
:param period_start: 原始周期开始时间(datetime.date)
:param leave_data: 请假数据字典(含workflow_id等)
:param hours: 总年假请假小时数(float)
:param period: 请假时间段 {'start': date, 'end': date}
:param leave_type_name: 假期类型名称(年假)
"""
employee_id = self.users_id.employee_ids.id if self.users_id.employee_ids else False
leave_start = period['start'] # 年假开始日期(date)
leave_end = period['end'] # 年假结束日期(date)
month_intervals = []
current_start = leave_start
while current_start <= leave_end:
current_month = current_start.month
current_year = current_start.year
next_month = current_month + 1 if current_month < 12 else 1
next_year = current_year if current_month < 12 else current_year + 1
current_month_end = date(next_year, next_month, 1) - timedelta(days=1)
interval_end = min(current_month_end, leave_end)
month_intervals.append({
'start': current_start,
'end': interval_end,
'period_start': date(current_year, current_month, 1)
})
current_start = interval_end + timedelta(days=1)
holiday_status_id = self.env['hr.leave.type'].search([('name', '=', '年假')], limit=1)
for interval in month_intervals:
interval_start = interval['start']
interval_end = interval['end']
if self.is_hour:
interval_hours = hours
else:
interval_hours = self._compute_duration_day(holiday_status_id, interval_start, interval_end) * 8
period_start = interval['period_start']
first_day = fields.Date.today().replace(day=1)
# 兜底报表(当月1号)
vacation_report = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', employee_id),
('vacation_date', '=', first_day),
], limit=1)
actual_vacation_report = vacation_report
# 优先匹配当月未确认的报表
vacation_report1 = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', employee_id),
('vacation_date', '=', period_start),
], limit=1)
if vacation_report1 and vacation_report1.vacation_state != 'confirmed':
actual_vacation_report = vacation_report1
if not actual_vacation_report:
continue
leave_line_domain = [
('employee_id', '=', employee_id),
('workflow_record_id', '=', leave_data.get('workflow_id', False)),
('update_type', '=', 'reduce'),
('allocation_type', '=', 'vacation'),
('text', '=', f"请假审批"),
('is_void', '=', False),
('update_lines', '=', interval_hours),
]
existing_line_in_vacation_report = False
if vacation_report:
existing_line_in_vacation_report = self.env['vacation.leave.line'].search(
[('vacation_id', '=', vacation_report.id)] + leave_line_domain,
limit=1
)
existing_line_in_vacation_report1 = False
if vacation_report1:
existing_line_in_vacation_report1 = self.env['vacation.leave.line'].search(
[('vacation_id', '=', vacation_report1.id)] + leave_line_domain,
limit=1
)
if not existing_line_in_vacation_report and not existing_line_in_vacation_report1:
self.env['vacation.leave.line'].create({
'vacation_id': actual_vacation_report.id,
'employee_id': employee_id,
'workflow_record_id': leave_data['workflow_id'],
'update_type': 'reduce',
'update_date': fields.Date.today(),
'update_lines': interval_hours,
'is_void': False,
'allocation_type': 'vacation',
'text': f"请假审批"
})
actual_vacation_report.update_line()
def _create_other_leave_line(self, period_start, leave_data, days, period, leave_type_name):
"""
创建其他假减记录(跨月自动拆分)
:param period_start: 原始周期开始时间(datetime.date)
:param leave_data: 请假数据字典(含workflow_id等)
:param days: 总请假天数(浮点型)
:param period: 请假时间段 {'start': date, 'end': date}
:param leave_type_name: 假期类型名称
"""
holiday_type = self.env['hr.leave.type'].search([('name', '=', leave_type_name)], limit=1)
employee_id = self.users_id.employee_ids.id if self.users_id.employee_ids else False
leave_start = period['start']
leave_end = period['end']
month_intervals = []
current_start = leave_start
while current_start <= leave_end:
current_month = current_start.month
current_year = current_start.year
next_month = current_month + 1 if current_month < 12 else 1
next_year = current_year if current_month < 12 else current_year + 1
current_month_end = date(next_year, next_month, 1) - timedelta(days=1)
interval_end = min(current_month_end, leave_end)
# 加入拆分后的区间列表
month_intervals.append({
'start': current_start,
'end': interval_end,
'period_start': date(current_year, current_month, 1)
})
current_start = interval_end + timedelta(days=1)
# 3. 遍历拆分后的每个月区间,分别创建记录
for interval in month_intervals:
interval_start = interval['start']
interval_end = interval['end']
interval_days = self._compute_duration_day(holiday_type, interval_start, interval_end)
period_start = interval['period_start']
vacation_report1 = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', employee_id),
('vacation_date', '=', period_start),
], limit=1)
first_day = fields.Date.today().replace(day=1)
vacation_report = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', employee_id),
('vacation_date', '=', first_day),
], limit=1)
actual_vacation_report = vacation_report
if vacation_report1 and vacation_report1.vacation_state != 'confirmed':
actual_vacation_report = vacation_report1
if not actual_vacation_report:
continue # 无报表,跳过创建
existing_leave_line = self.env['vacation.other.line'].search([
('vacation_id', '=', actual_vacation_report.id),
('employee_id', '=', employee_id),
('workflow_record_id', '=', leave_data.get('workflow_id', False)),
('holiday_type', '=', holiday_type.id),
('lines', '=', interval_days),
('update_type', '=', 'reduce'),
('text', '=', '请假审批'),
('is_void', '=', False),
], limit=1)
if not existing_leave_line:
self.env['vacation.other.line'].create({
'vacation_id': actual_vacation_report.id,
'employee_id': employee_id,
'workflow_record_id': leave_data.get('workflow_id', False),
'update_type': 'reduce',
'lines': interval_days,
'holiday_type': holiday_type.id,
'update_date': fields.Date.today(),
'is_void': False,
'text': f"请假审批"
})
def update_attendance(self, employee_id, start_date, end_date, code):
"""
更新考勤
:param employee_id: 员工id
:param start_date: 开始时间
:param end_date: 结束时间
:return:
"""
workflow_id = self.env['soong.workflow.record'].search([('code', '=', code)], limit=1)
attendance_records = self.env['hr.attendance'].search([
('employee_id', '=', employee_id.id),
('current_data', '>=', start_date),
('current_data', '<=', end_date),
])
if attendance_records:
attendance_records.write({
'workflow_id': workflow_id.id,
'is_workflow': True,
'is_no_card': False,
'is_early_card': False,
'is_late_card': False,
'is_abnormal': False,
'absenteeism_card': False
})
def _compute_duration_day(self, holiday_status_id, start_date, end_date):
"""
计算请假时长
:param holiday_status_id: 请假类型
:param start_date: 开始时间
:param end_date: 结束时间
:return:天
"""
# 如果是自然日:
if holiday_status_id.type_of_work == "1":
if start_date and end_date:
days = (end_date - start_date).days + 1
else:
days = 0
return days
if holiday_status_id.type_of_work == "0":
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)
])
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)
else:
delta = end_date - start_date
days = delta.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}日'
@api.constrains('is_day', 'is_hour', 'leave_ids', 'hour_leave_ids')
def _check_leave_type_and_details(self):
"""检查请假类型和明细行是否符合要求"""
current_date = fields.Date.today()
vacation_date = current_date
VALID_DAILY_HOURS = [2, 4, 6, 8]
for record in self:
if not record.is_day and not record.is_hour:
raise ValidationError(_("请选择按天请假或按小时请假!"))
if not record.leave_ids and not record.hour_leave_ids:
raise ValidationError(_("请至少添加一条按天或按小时的请假记录!"))
if record.leave_ids and not record.is_day:
raise ValidationError(_("添加了按天请假记录,必须选择'按天请假'选项!"))
if record.hour_leave_ids and not record.is_hour:
raise ValidationError(_("添加了按小时请假记录,必须选择'按小时请假'选项!"))
if record.hour_leave_ids and record.company_id.name == '天河物管':
for line in record.hour_leave_ids:
if not line.request_date_from or not line.request_hour_to:
continue
match = re.search(r'(\d+(?:\.\d+)?)', line.number_of_hours_text or '')
if not match:
raise ValidationError(
_(f"第{line.sequence or '一'}条小时请假记录:无法提取有效小时数!"))
daily_hours = float(match.group(1))
if daily_hours == 0.0:
raise ValidationError(
_(f"第{line.sequence or '一'}条小时请假记录:请假小时数不能为0!"))
if daily_hours not in VALID_DAILY_HOURS:
raise ValidationError(
_(f"第{line.sequence or '一'}条小时请假记录:休假时长必须为 2、4、6 或 8 小时才可以保存!")
)
leave_days = 0.0
overtime_days = 0.0
total_annual_leave_days = 0.0 # 年假总天数
total_compensatory_leave_days = 0.0 # 补休总天数
for leave_line in record.leave_ids:
if leave_line.holiday_status_id.name == '年假':
total_annual_leave_days += float(leave_line.number_of_days_display or 0)
elif leave_line.holiday_status_id.name == '补休':
total_compensatory_leave_days += float(leave_line.number_of_days_display or 0)
for hour_leave_line in record.hour_leave_ids:
leave_hours = float(hour_leave_line.leave_hours or 0)
leave_days = leave_hours / 8
if hour_leave_line.holiday_status_id == '年假':
total_annual_leave_days += leave_days
elif hour_leave_line.holiday_status_id == '补休':
total_compensatory_leave_days += leave_days
reports_id = self.env['yuthon.vacation.reports'].search(
[('employee_id', '=', record.users_id.employee_ids.id),
('vacation_date', '=', vacation_date.replace(day=1))
], limit=1)
if reports_id:
reports_id._compute_working_overtime()
leave_days = reports_id.leave
overtime_days = reports_id.working_overtime
if total_annual_leave_days > leave_days:
raise UserError('年假天数额度不足,无法请假')
if total_compensatory_leave_days > overtime_days:
raise UserError('补休天数额度不足,无法请假')
@api.onchange('leave_ids', 'hour_leave_ids', 'employee_id')
def _onchange_leave_dates(self):
all_dates = []
total_days = 0
first_hour_leave = self.hour_leave_ids[0] if self.hour_leave_ids else None
holiday_status_names = []
for leave in self.leave_ids:
# 收集假期类型名称(去重)
if leave.holiday_status_id and leave.holiday_status_id.name:
status_name = leave.holiday_status_id.name
if status_name not in holiday_status_names:
holiday_status_names.append(status_name)
# 收集日期
if leave.request_date_from:
all_dates.append(leave.request_date_from)
if leave.request_date_to:
all_dates.append(leave.request_date_to)
total_days += leave.number_of_days_display if hasattr(leave, 'number_of_days_display') else 0
for line in self.hour_leave_ids:
if line.request_date_from:
all_dates.append(line.request_date_from)
# 设置起止日期
self.start_date = min(all_dates) if all_dates else False
self.end_date = max(all_dates) if all_dates else False
# 计算小时请假折算天数(按8小时/天)
hours_days = 0
for line in self.hour_leave_ids:
hours_days += line.leave_hours
self.days = total_days + hours_days / 8
# 生成名称:员工名 + 假期类型 + 日期范围 + 天数/小时
if self.employee_id:
name_parts = [f"{self.employee_id.name}的"]
if holiday_status_names:
name_parts.append(','.join(holiday_status_names))
if self.start_date and self.end_date:
date_part = f"{self.start_date.strftime('%Y-%m-%d')}~{self.end_date.strftime('%Y-%m-%d')}"
name_parts.append(date_part)
if self.leave_ids:
name_parts.append(f"共计{int(total_days)}天")
# 拼接小时请假信息
if first_hour_leave and first_hour_leave.number_of_hours_text:
name_parts.append(first_hour_leave.number_of_hours_text)
self.name = " ".join(name_parts)
else:
self.name = ""
@api.depends('users_id')
def _compute_holiday_allowance(self):
current_date = fields.Date.today()
for record in self:
annual_leave_line = []
overtime_leave_line = []
parental_leave_line = []
dynamic_previous_year = 0.0
dynamic_remainder_year = 0.0
dynamic_previous_year1 = 0.0
this_year1 = 0.0
remaining_parental_leave = 0.0
if not record.users_id or not record.users_id.employee_ids:
record.holiday_allowance = ""
continue
employee = record.users_id.employee_ids[0]
vacation_date = current_date.replace(day=1)
reports_id = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', employee.id),
('vacation_date', '=', vacation_date)
], limit=1)
if reports_id:
dynamic_previous_year = reports_id.dynamic_previous_year or 0.0
dynamic_remainder_year = reports_id.dynamic_remainder_year or 0.0
dynamic_previous_year1 = reports_id.dynamic_previous_year1 or 0.0
this_year1 = reports_id.this_year1 or 0.0
parental_leave_type = self.env['hr.leave.type'].search([
('name', 'ilike', '育儿假'),
], limit=1)
if parental_leave_type:
# 筛选当前有效的育儿假分配记录
allocation_records = self.env['hr.leave.allocation'].search([
('employee_id', '=', employee.id),
('holiday_status_id', '=', parental_leave_type.id),
('state', 'in', ['validate', 'done']),
('date_from', '<=', current_date),
'|',
('date_to', '=', False),
('date_to', '>=', current_date)
])
# 累加有效分配总额
total_allocation = sum(
alloc.number_of_days_display for alloc in allocation_records) if allocation_records else 0.0
# 筛选当前有效的已使用育儿假记录
used_parental_leave_list = self.env['hr.leave'].search([
('employee_id', '=', employee.id),
('holiday_status_id', '=', parental_leave_type.id),
('state', 'in', ['validate', 'done']),
('date_from', '<=', current_date),
('date_to', '>=', current_date)
])
used_days = sum(leave.number_of_days_display for leave in
used_parental_leave_list) if used_parental_leave_list else 0.0
remaining_parental_leave = max(0.0, total_allocation - used_days)
if dynamic_previous_year > 0:
annual_leave_line.append(f"上年年假剩余{dynamic_previous_year:.3f}天")
if dynamic_remainder_year > 0:
annual_leave_line.append(f"本年年假剩余{dynamic_remainder_year:.3f}天")
if dynamic_previous_year1 > 0:
overtime_leave_line.append(f"上年补休剩余{dynamic_previous_year1:.3f}天")
if this_year1 > 0:
overtime_leave_line.append(f"本年补休剩余{this_year1:.3f}天")
if remaining_parental_leave > 0:
parental_leave_line.append(f"育儿假{remaining_parental_leave:.3f}天")
final_lines = []
if annual_leave_line:
final_lines.append(', '.join(annual_leave_line))
if overtime_leave_line:
final_lines.append(', '.join(overtime_leave_line))
if parental_leave_line:
final_lines.append(', '.join(parental_leave_line))
record.holiday_allowance = '\n'.join(final_lines) if final_lines else ""
def action_print_leave_batch(self):
"""按钮触发打印请假单"""
self.ensure_one()
report = self.env.ref('yuthon_holiday.action_hr_leave_batch_print')
return report.report_action(self, config=False)
class HrLeaveLine(models.Model):
_name = 'hr.leave.line'
#
# # 计算开始时间与结束时间之间的时长
# def _get_leave_duration(self, start_time: datetime, end_time: datetime):
# """
# 公共假期计算
# 查询所有公共假期(resource.calendar.leaves)
# 通过开始时间与结束时间,去除公共假期,再去除双休,计算时长,
# """
# self.ensure_one()
# all_resource_calendar_leaves = self.env['resource.calendar.leaves'].search([('resource_id', '=', False)])
# # 创建一个集合存储假期的每一天
# holidays = []
# for holiday in all_resource_calendar_leaves:
# # 根据公共假期的开始时间与结束时间计算时间段
# current = holiday.date_from
# while current <= holiday.date_to:
# # 将当天0点的时间戳放入假期集合
# holidays.append(int(int(current.timest上午p()) + 8 * 3600) // 86400 * 86400 - 8 * 3600)
# current += timedelta(days=1)
# effective_days = 0
# current = start_time
# while current <= end_time:
# current_timest上午p = int(int(current.timest上午p()) + 8 * 3600) // 86400 * 86400 - 8 * 3600
# if current.weekday() < 5 and current_timest上午p not in holidays: # 0-4为工作日
# effective_days += 1
# current += timedelta(days=1)
#
# return effective_days