235 lines
11 KiB
Python
235 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
from odoo import api, fields, models, tools, _
|
|
from datetime import datetime, time, timedelta
|
|
from odoo.exceptions import UserError
|
|
|
|
|
|
class MessageReminderTasks(models.Model):
|
|
_name = 'message.reminder.tasks'
|
|
_description = '消息提醒任务'
|
|
|
|
name = fields.Char(string="姓名")
|
|
department = fields.Char(string="部门")
|
|
company = fields.Char(string="公司")
|
|
shift_name = fields.Char(string="班次")
|
|
current_data = fields.Date(string='打卡日期')
|
|
state = fields.Selection([('todo', '待提醒'), ('done', '已提醒')], string='状态', default='todo')
|
|
type = fields.Selection([('work', '上班'), ('offWork', '下班')], string='类型')
|
|
reminder_time = fields.Datetime(string="提醒时间")
|
|
remake = fields.Char(string="备注")
|
|
holiday = fields.Boolean(string="假期")
|
|
|
|
def get_reminder_tasks(self):
|
|
current_date = (fields.Date.today() + relativedelta(hours=8)).date()
|
|
target_company_names = ['天河物管', '天河粮储', '集团本部']
|
|
target_companies = self.env['res.company'].search([('name', 'in', target_company_names)])
|
|
target_calendars = self.env['resource.calendar'].search([
|
|
('company_id', 'in', target_companies.ids)
|
|
])
|
|
# 获取日历关联的白名单员工
|
|
whitelist_employees = target_calendars.mapped('resource_user_ids').ids
|
|
condition1_employees = self.env['hr.employee'].search([
|
|
('id', 'not in', whitelist_employees),
|
|
('name', '!=', 'Administrator'),
|
|
('user_state', '=', 'in'),
|
|
('active', '=', True)
|
|
])
|
|
target_role_names = ['保安部', '工程部', '联安粮库']
|
|
target_roles = self.env['planning.role'].search([
|
|
('name', 'in', target_role_names)
|
|
])
|
|
default_reminder_types = ['work', 'offWork']
|
|
default_reminder_times = [
|
|
datetime.combine(current_date, datetime.strptime('00:55', '%H:%M').time()), # 时差
|
|
datetime.combine(current_date, datetime.strptime('10:00', '%H:%M').time())
|
|
]
|
|
condition2_employee_ids = []
|
|
if target_roles:
|
|
for role in target_roles:
|
|
resource_employees = role.resource_ids.mapped('employee_id')
|
|
condition2_employee_ids.extend(resource_employees)
|
|
condition2_employee_ids = list(set(condition2_employee_ids))
|
|
# 哺乳人员
|
|
breastfeeding_employees = target_calendars.mapped('breastfeeding_holiday_ids.employee_id')
|
|
breastfeeding_employee_ids = breastfeeding_employees.ids if breastfeeding_employees else []
|
|
# 获取不在白名单且是行政班次的人员
|
|
final_employees = condition1_employees.filtered(lambda emp: emp not in condition2_employee_ids)
|
|
task_vals_list = []
|
|
is_workday = self._process_shift_days(current_date)
|
|
if is_workday:
|
|
for employee in final_employees:
|
|
emp_name = employee.name
|
|
emp_department = employee.department_id.name
|
|
emp_company = employee.company_id.name
|
|
is_lactating = employee.id in breastfeeding_employee_ids
|
|
if is_lactating:
|
|
# 哺乳人员
|
|
reminder_times = self.get_reminder_lactating_personnel(employee, current_date)
|
|
reminder_types = ['work', 'offWork']
|
|
shift_name = '行政班次（哺乳）'
|
|
else:
|
|
reminder_times = default_reminder_times
|
|
reminder_types = default_reminder_types
|
|
shift_name = '行政班次'
|
|
|
|
for reminder_time, reminder_type in zip(reminder_times, reminder_types):
|
|
task_vals = {
|
|
'name': emp_name,
|
|
'current_data': reminder_time.date(),
|
|
'department': emp_department,
|
|
'company': emp_company,
|
|
'shift_name': shift_name,
|
|
'type': reminder_type,
|
|
'reminder_time': fields.Datetime.to_string(reminder_time),
|
|
}
|
|
task_vals_list.append(task_vals)
|
|
|
|
# 计划行政
|
|
slot_start = datetime.combine(current_date, datetime.min.time()) - timedelta(hours=8)
|
|
slot_end = datetime.combine(current_date + timedelta(days=1), datetime.min.time()) - timedelta(hours=8)
|
|
# 查找员工对应的班次
|
|
for employee in condition2_employee_ids:
|
|
reminder_plan_times = []
|
|
reminder_plan_types = []
|
|
planning_slot = self.env['planning.slot'].search([
|
|
('resource_id.name', '=', employee.name),
|
|
('start_datetime', '>=', slot_start),
|
|
('start_datetime', '<', slot_end),
|
|
], limit=1)
|
|
template = self.env['planning.slot.template'].search([('planning_name', '=', planning_slot.planning_name)], limit=1)
|
|
if template:
|
|
sch_check_in_time, sch_check_out_time = self.normal_clock_in_time(template, current_date)
|
|
|
|
is_lactating = employee.id in breastfeeding_employee_ids
|
|
if is_lactating:
|
|
reminder_plan_times = self.get_reminder_lactating_personnel(employee.id, current_date)
|
|
reminder_plan_types = ['work', 'offWork']
|
|
shift_name = f"{template.planning_name}（哺乳）"
|
|
else:
|
|
reminder_plan_times.append(sch_check_in_time - timedelta(minutes=5))
|
|
reminder_plan_types.append('work')
|
|
reminder_plan_times.append(sch_check_out_time)
|
|
reminder_plan_types.append('offWork')
|
|
shift_name = template.planning_name
|
|
|
|
for reminder_time, reminder_type in zip(reminder_plan_times, reminder_plan_types):
|
|
task_plan_vals = {
|
|
'name': employee.name,
|
|
'department': employee.department_id.name,
|
|
'company': employee.company_id.name,
|
|
'shift_name': shift_name,
|
|
'current_data': (reminder_time + relativedelta(hours=8)).date(),
|
|
'type': reminder_type,
|
|
'reminder_time': fields.Datetime.to_string(reminder_time),
|
|
}
|
|
task_vals_list.append(task_plan_vals)
|
|
|
|
self.env['message.reminder.tasks'].create(task_vals_list)
|
|
|
|
|
|
def get_reminder_lactating_personnel(self, employee, current_data):
|
|
# 查找当天的哺乳假记录（匹配员工+日期范围）
|
|
breastfeeding_record = self.env['breastfeeding.holiday'].search([
|
|
('employee_id', '=', employee.id),
|
|
('calendar_start_time', '<=', current_data),
|
|
('calendar_end_time', '>=', current_data),
|
|
], limit=1) # 取当天有效的第一条记录
|
|
def float_to_datetime(float_time, base_date):
|
|
"""将float时间（如8.5=8:30）转换为datetime"""
|
|
if not float_time or not base_date:
|
|
return None
|
|
hours = int(float_time)
|
|
minutes = int((float_time - hours) * 60)
|
|
return fields.Datetime.to_datetime(f"{base_date} {hours:02d}:{minutes:02d}:00")
|
|
|
|
sch_check_in_time = float_to_datetime(breastfeeding_record.check_in, current_data) - timedelta(hours=8)
|
|
sch_check_out_time = float_to_datetime(breastfeeding_record.check_out, current_data) - timedelta(hours=8)
|
|
return [
|
|
sch_check_in_time - timedelta(minutes=5),
|
|
sch_check_out_time
|
|
]
|
|
|
|
def normal_clock_in_time(self, template, current_data):
|
|
"""
|
|
:param template: 班次模板
|
|
:param current_data: 打卡日期
|
|
:return: 正常开始上班时间和下班时间
|
|
"""
|
|
if template.day2 == '次日':
|
|
check_out_end = template.end_time2
|
|
else:
|
|
check_out_end = template.end_time
|
|
check_in_start = template.start_time
|
|
date = current_data
|
|
start_hour = int(check_in_start)
|
|
start_minute = int((check_in_start - start_hour) * 60)
|
|
end_hour = int(check_out_end)
|
|
end_minute = int((check_out_end - end_hour) * 60)
|
|
start_datetime = datetime.combine(date, datetime.min.time()).replace(
|
|
hour=start_hour, minute=start_minute, second=0, microsecond=0
|
|
) - timedelta(hours=8)
|
|
if template.day2 == '次日':
|
|
end_datetime = datetime.combine(date, datetime.min.time()).replace(
|
|
hour=end_hour, minute=end_minute, second=0, microsecond=0
|
|
) - timedelta(hours=8) + timedelta(days=1)
|
|
else:
|
|
end_datetime = datetime.combine(date, datetime.min.time()).replace(
|
|
hour=end_hour, minute=end_minute, second=0, microsecond=0
|
|
) - timedelta(hours=8)
|
|
|
|
if template.planning_name == '中夜':
|
|
end_datetime = datetime.combine(date, datetime.min.time()).replace(
|
|
hour=22, minute=0, second=0, microsecond=0
|
|
) - timedelta(hours=8)
|
|
|
|
return start_datetime, end_datetime
|
|
|
|
def cron_delete_3days_ago_records(self):
|
|
three_days_ago = fields.Datetime.now() - timedelta(days=3)
|
|
old_records = self.search([
|
|
('create_date', '<=', three_days_ago)
|
|
])
|
|
if old_records:
|
|
old_records.unlink()
|
|
|
|
def _process_shift_days(self, current_data):
|
|
"""
|
|
判断是否是工作日 调休日 双休
|
|
"""
|
|
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]
|
|
result = f"{current_data.month}月{current_data.day}日"
|
|
if result in shift_days_list:
|
|
return True
|
|
elif (result in holidays_list) or (current_data.weekday() in [5, 6]):
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
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}日' |