pengyuthon/gzth
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

!4374 添加功能

Browse Source 
Merge pull request !4374 from 李鹏宇/dev_zhou
This commit is contained in:
李鹏宇
2025-08-07 07:10:12 +00:00
committed by Gitee
parent ffa9448b5d ac6609730f
commit 93efaafbc0
1 changed files with 96 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yuthon_holiday/models/inherit_hr_leave.py
+96 -177
View File
@@ -520,8 +520,6 @@ class HolidaysRequestBatch(models.Model):
]
for leave_type in other_leave_types:
self._process_specific_leaves(leave_type, is_annual_leave=False)
_logger.info(f"请假申请：{self.code} 处理完成")
return True
def _process_specific_leaves(self, leave_type_name, is_annual_leave):
@@ -530,209 +528,130 @@ class HolidaysRequestBatch(models.Model):
:param leave_type_name: 假期类型名称
:param is_annual_leave: 是否为年假（True/False）
"""
_logger.info(f"开始处理假期类型：{leave_type_name}，是否为年假：{is_annual_leave}，申请编号：{self.code}")
# 收集天假和小时假记录
day_leaves = self._get_day_leaves(leave_type_name)
hourly_leaves = self._get_hourly_leaves(leave_type_name)
_logger.debug(f"假期类型：{leave_type_name}，天假记录数：{len(day_leaves)}，小时假记录数：{len(hourly_leaves)}")
# 处理天假
for idx, leave in enumerate(day_leaves):
try:
# 跨月拆分逻辑（仅非年假）
date_ranges = self._split_by_month(leave['start'], leave['end']) if not is_annual_leave else [
{'start': leave['start'], 'end': leave['end']}
]
_logger.debug(
f"天假记录 {idx + 1} 拆分后日期段数：{len(date_ranges)}，原始起止：{leave['start']}至{leave['end']}")
for period in date_ranges:
days = self._compute_duration_day(leave['type_id'], period['start'], period['end'])
_logger.debug(f"日期段 {period['start']}至{period['end']} 计算天数：{days}")
if days > 0:
if is_annual_leave:
self._create_annual_leave_line(period['start'], leave, days, period)
else:
self._create_other_leave_line(period['start'], leave, days, period, leave_type_name)
except Exception as e:
_logger.error(f"处理天假记录 {idx + 1} 失败（类型：{leave_type_name}）：{str(e)}", exc_info=True)
# 处理小时假
for idx, leave in enumerate(hourly_leaves):
try:
leave_date = leave['date']
days = leave['days']
_logger.debug(f"小时假记录 {idx + 1} 日期：{leave_date}，转换天数：{days}")
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:
days = self._compute_duration_day(leave['type_id'], period['start'], period['end'])
if days > 0:
if is_annual_leave:
self._create_annual_leave_line(leave_date, leave, days,
{'start': leave_date, 'end': leave_date})
self._create_annual_leave_line(period['start'], leave, days, period)
else:
self._create_other_leave_line(leave_date, leave, days, {'start': leave_date, 'end': leave_date},
leave_type_name)
except Exception as e:
_logger.error(f"处理小时假记录 {idx + 1} 失败（类型：{leave_type_name}）：{str(e)}", exc_info=True)
_logger.info(f"假期类型：{leave_type_name} 处理完成")
self._create_other_leave_line(period['start'], leave, days, period, leave_type_name)
# 处理小时假
for leave in hourly_leaves:
leave_date = leave['date']
days = leave['days']
if days > 0:
if is_annual_leave:
self._create_annual_leave_line(leave_date, leave, days, {'start': leave_date, 'end': leave_date})
else:
self._create_other_leave_line(leave_date, leave, days,{'start': leave_date, 'end': leave_date}, leave_type_name)
def _create_other_leave_line(self, period_start, leave_data, days, period, leave_type_name):
"""为非年假类型创建vacation.other.line记录"""
try:
report_month = date(period_start.year, period_start.month, 1)
vacation_report = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', self.users_id.employee_id.id),
('vacation_date', '>=', fields.Date.to_string(report_month)),
('vacation_date', '<=', fields.Date.to_string(
date(period_start.year, period_start.month,
calendar.monthrange(period_start.year, period_start.month)[1]))
)
], limit=1)
# 查找关联的假期额度记录
report_month = date(period_start.year, period_start.month, 1)
vacation_report = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', self.users_id.employee_id.id),
('vacation_date', '>=', fields.Date.to_string(report_month)),
('vacation_date', '<=', fields.Date.to_string(
date(period_start.year, period_start.month,
calendar.monthrange(period_start.year, period_start.month)[1]))
)
], limit=1)
_logger.debug(
f"创建非年假明细（类型：{leave_type_name}），关联假期记录：{vacation_report.id if vacation_report else '无'}，天数：{days}")
record = self.env['vacation.other.line'].sudo().create({
'vacation_id': vacation_report.id if vacation_report else False,
'holiday_type': leave_data['type_id'].id,
'employee_id': self.users_id.employee_id.id,
'workflow_record_id': leave_data['workflow_id'],
'update_date': fields.Date.today(),
'update_type': 'reduce',
'lines': days,
'text': f"请假审批"
})
_logger.info(f"成功创建非年假明细记录：{record.id}（类型：{leave_type_name}）")
except Exception as e:
_logger.error(f"创建非年假明细失败（类型：{leave_type_name}）：{str(e)}", exc_info=True)
self.env['vacation.other.line'].create({
'vacation_id': vacation_report.id if vacation_report else False,
'holiday_type': leave_data['type_id'].id,
'employee_id': self.users_id.employee_id.id,
'workflow_record_id': leave_data['workflow_id'],
'update_date': fields.Date.today(),
'update_type': 'reduce',
'lines': days,
'text': f"请假审批"
})
_logger.info(f"成功创建非年假明细记录（类型：{leave_type_name}），天数为{days}天，员工ID为{self.users_id.employee_id.id}")
def _get_day_leaves(self, leave_type_name):
try:
workflow_id = self.env['soong.workflow.record'].search([('code', '=', self.code)], limit=1).id
records = [{
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
]
_logger.debug(f"收集天假记录（类型：{leave_type_name}）：{len(records)}条")
return records
except Exception as e:
_logger.error(f"收集天假记录失败（类型：{leave_type_name}）：{str(e)}", exc_info=True)
return []
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):
try:
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
days = hours / 8
hourly_records.append({
'date': fields.Date.from_string(line.request_date_from),
'type_id': line.holiday_status_id,
'workflow_id': workflow_id,
'days': days
})
_logger.debug(f"收集小时假记录（类型：{leave_type_name}）：{len(hourly_records)}条")
return hourly_records
except Exception as e:
_logger.error(f"收集小时假记录失败（类型：{leave_type_name}）：{str(e)}", exc_info=True)
return []
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):
try:
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})
_logger.debug(f"无需跨月拆分：{start_date}至{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})
_logger.debug(f"跨月拆分完成：原始{start_date}至{end_date}，拆分为{len(date_ranges)}段")
# 跨月拆分逻逻辑
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
except Exception as e:
_logger.error(f"跨月拆分失败（{start_date}至{end_date}）：{str(e)}", exc_info=True)
return []
def _compute_duration_day(self, holiday_type_id, start_date, end_date):
try:
if not start_date or not end_date or start_date > end_date:
_logger.warning(f"无效日期范围：{start_date}至{end_date}，返回0天")
return 0.0
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
holiday = self.env['hr.leave.type'].browse(holiday_type_id.id)
total_days = (end_date - start_date).days + 1
if holiday.work_entry_type_id:
work_days = 0
current_date = start_date
while current_date <= end_date:
if current_date.isoweekday() not in [6, 7]:
work_days += 1
current_date += timedelta(days=1)
_logger.debug(f"工作日计算：{start_date}至{end_date}，共{work_days}天（排除周末）")
return work_days
_logger.debug(f"自然日计算：{start_date}至{end_date}，共{total_days}天")
return total_days
except Exception as e:
_logger.error(f"计算天数失败（{start_date}至{end_date}）：{str(e)}", exc_info=True)
return 0.0
date_ranges.append({'start': date(end_date.year, end_date.month, 1), 'end': end_date})
return date_ranges
def _create_annual_leave_line(self, period_start, leave_data, days, period):
"""更新年假记录"""
try:
report_month = date(period_start.year, period_start.month, 1)
vacation_report = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', self.users_id.employee_id.id),
('vacation_date', '>=', fields.Date.to_string(report_month)),
('vacation_date', '<=', fields.Date.to_string(
date(period_start.year, period_start.month,
calendar.monthrange(period_start.year, period_start.month)[1]))
)
], limit=1)
_logger.debug(f"创建年假明细，关联假期记录：{vacation_report.id if vacation_report else '无'}，天数：{days}")
record = self.env['vacation.leave.line'].create({
'vacation_id': vacation_report.id if vacation_report else False,
'employee_id': self.users_id.employee_id.id,
'workflow_record_id': leave_data['workflow_id'],
'update_type': 'reduce',
'update_date': fields.Date.today(),
'update_lines': days,
'is_void': False,
'allocation_type': 'vacation',
'text': f"请假审批"
})
_logger.info(f"成功创建年假明细记录：{record.id}")
except Exception as e:
_logger.error(f"创建年假明细失败：{str(e)}", exc_info=True)
report_month = date(period_start.year, period_start.month, 1)
vacation_report = self.env['yuthon.vacation.reports'].search([
('employee_id', '=', self.users_id.employee_id.id),
('vacation_date', '>=', fields.Date.to_string(report_month)),
('vacation_date', '<=', fields.Date.to_string(
date(period_start.year, period_start.month,
calendar.monthrange(period_start.year, period_start.month)[1]))
)
], limit=1)
# 创建年假明细记录
self.env['vacation.leave.line'].create({
'vacation_id': vacation_report.id if vacation_report else False,
'employee_id': self.users_id.employee_id.id,
'workflow_record_id': leave_data['workflow_id'],
'update_type': 'reduce',
'update_date': fields.Date.today(),
'update_lines': days,
'is_void': False,
'allocation_type': 'vacation',
'text': f"请假审批"
})
def _compute_duration_day(self, holiday_status_id, start_date, end_date):