# -*- coding: utf-8 -*- import os import base64 import pytz import calendar from datetime import datetime, timedelta, time import logging import traceback from odoo import api, fields, models, tools, _ from dateutil.relativedelta import relativedelta from datetime import date _logger = logging.getLogger(__name__) class HrAttendanceDetailImage(models.Model): _name = "hr.attendance.detail.image" _description = "打卡原始抓取数据图片" attendance_detail_id = fields.Many2one('hr.attendance.detail', string="原始打卡记录", ondelete="cascade", readonly=True) media_url = fields.Char(string='附件地址') media_file = fields.Binary(string='附件文件', compute="_compute_media_file") @api.depends('media_url') def _compute_media_file(self): for record in self: record.media_file = None wx_work_auth = self.env['auth.oauth.provider'].search([('name', '=', 'Wechat Card')], limit=1) if record.media_url: image_path = os.path.join(wx_work_auth.wx_work_image_path, record.media_url) record.media_file = base64.b64encode(open(image_path, 'rb').read()) if image_path else False class HrAttendanceDetail(models.Model): _name = "hr.attendance.detail" _description = "打卡原始抓取数据" _order = 'checkin_time desc' name = fields.Char(related='employee_id.name') employee_id = fields.Many2one('hr.employee', string="员工", readonly=True) company_id = fields.Many2one('res.company', string="公司", related="employee_id.company_id", store=True) department_id = fields.Many2one('hr.department', string="部门", related="employee_id.department_id", readonly=True, store=True) source = fields.Selection([('iot', '考勤机'), ('mini_programs', '小程序')], string="来源") attendance_iot_id = fields.Many2one('attendance.iot', string="设备号") attendance_address = fields.Char(related="attendance_iot_id.attendance_address", string="设备地址") userid = fields.Char(string='userid') groupname = fields.Char(string='打卡规则名称') checkin_type = fields.Char(string='打卡类型') exception_type = fields.Char(string='异常类型') checkin_time = fields.Datetime(string='打卡时间') location_title = fields.Char(string='打卡地点') location_detail = fields.Char(string='打卡地点详情') wifiname = fields.Char(string='打卡wifi名称') notes = fields.Char(string='打卡备注') wifimac = fields.Char(string='打卡的MAC地址') mediaids = fields.Char(string='打卡的附件media_id') lat = fields.Integer(string='位置打卡地点纬度') lng = fields.Integer(string='位置打卡地点经度') deviceid = fields.Char(string='打卡设备id') sch_checkin_time = fields.Datetime(string='标准打卡时间') groupid = fields.Integer(string='规则id') schedule_id = fields.Char(string='班次id') timeline_id = fields.Integer(string='时段id') remark = fields.Char(string="备注") attendance_detail_image_ids = fields.One2many(string="打卡记录图片", comodel_name='hr.attendance.detail.image', inverse_name='attendance_detail_id') hr_attendance_id = fields.Many2one('hr.attendance', string="出勤记录", ondelete="cascade", readonly=True) seconds = fields.Integer( string='秒', compute='_compute_seconds', ) @api.depends('checkin_time') def _compute_seconds(self): for record in self: if not record.checkin_time: record.seconds = 0 continue record.seconds = record.checkin_time.second def action_show_details(self): self.ensure_one() view = self.env.ref("yuthon_wechat.view_hr_attendance_detail_image_tree") return { "name": _("图片详情"), "type": "ir.actions.act_window", "view_mode": "form", "res_model": "hr.attendance.detail.image", "views": [(view.id, "tree")], "view_id": view.id, "target": "new", "res_id": self.id, "domain": [('attendance_detail_id', '=', self.id)], "context": {'default_attendance_detail_id': self.id} } @api.model def create(self, vals): record = super(HrAttendanceDetail, self).create(vals) # 批量导入时跳过企业微信通知 if self.env.context.get('skip_iot_notify'): return record if record.employee_id.name not in ['Administrator', '张丽芳']: resource_user_ids = record.employee_id.resource_calendar_id.resource_user_ids.ids is_resource_user = True if record.employee_id.id in resource_user_ids else False employee_id = record.employee_id checkin_datetime = record.checkin_time last_record = self.search([ ('employee_id', '=', employee_id.id), ('source', '=', 'iot'), ('checkin_time', '<', checkin_datetime) ], order='checkin_time desc', limit=1) # 如果上一条记录存在，并且时间差 < 60秒 → 不发提醒 skip_notify = False if last_record: time_diff = checkin_datetime - last_record.checkin_time if time_diff.total_seconds() < 60: skip_notify = True # 小于1分钟，跳过提醒 if record.employee_id and record.checkin_time and record.source == 'iot' and not is_resource_user and not skip_notify: current_data = record.checkin_time.date() + timedelta(hours=8) day_start = checkin_datetime.replace(hour=0, minute=0, second=0, microsecond=0) day_end = checkin_datetime.replace(hour=23, minute=59, second=59, microsecond=99) daily_iot_records = self.search([ ('employee_id', '=', employee_id.id), ('source', '=', 'iot'), ('checkin_time', '>=', day_start), ('checkin_time', '<=', day_end) ], order='checkin_time asc') if daily_iot_records and daily_iot_records[0].id == record.id: message = f"打卡成功！今日上班时间:{record.checkin_time + timedelta(hours=8)}: {employee_id.name}，祝您工作愉快！" self.sudo().env["wecom.apps"].sync_send_message( category='att', user_id=employee_id.wecom_userid, message=message, ) # if record.employee_id.name == '蒋彬': # self.sudo().env["wecom.apps"].sync_send_message( # category='att', user_id='18562027762', # message=message + f'管理员测试管理员测试{vals}', # ) self.write({'remark': message}) _logger.info(message) else: resource_user_ids = employee_id.resource_calendar_id.resource_user_ids.ids is_resource_user = True if employee_id.id in resource_user_ids else False # 哺乳人员 suckling_employees = employee_id.resource_calendar_id.breastfeeding_holiday_ids.mapped('employee_id') def get_all_upper_depts(dept): """递归获取部门及其所有上级部门的ID列表""" dept_ids = [dept.id] if dept.parent_id: dept_ids.extend(get_all_upper_depts(dept.parent_id)) return dept_ids all_dept_ids = get_all_upper_depts(employee_id.department_id) if not is_resource_user: slot_start = datetime.combine(current_data, datetime.min.time()) - timedelta(hours=8) slot_end = datetime.combine(current_data + timedelta(days=1), datetime.min.time()) - timedelta( hours=8) # 查找员工对应的班次 planning_slots = self.env['planning.slot'].search([ ('resource_id', '=', employee_id.resource_id.id), ('start_datetime', '>=', slot_start), ('start_datetime', '<', slot_end), ], limit=1) planning_name = planning_slots.planning_name if planning_slots else None template_domain = [] if planning_name: template_domain.append(('planning_name', '=', planning_name)) template_domain.append(('department_id', 'in', all_dept_ids)) template = self.env['planning.slot.template'].search(template_domain, limit=1) # 如果模板里面找不到这个人 则还是往下判断（模板为空） if employee_id.id not in template.role_id.resource_ids.employee_id.ids: template = None if not template: role_templates = [ {'role_name': '厨卫（集团）', 'template': None}, {'role_name': '项目运营岗', 'template': None} ] for item in role_templates: item['template'] = self.env['planning.slot.template'].search( [('role_id.name', '=', item['role_name'])], ) for item in role_templates: if item['template']: employee_ids = item['template'].role_id.resource_ids.employee_id.ids or [] employee_list = list(employee_ids) if employee_id.id in employee_list: template = item['template'] break if not template: template = self.env['planning.slot.template'].search( [('role_id.name', '=', '集团行政')], limit=1 ) sch_check_in_time, sch_check_out_time = self.normal_clock_in_time(template, current_data) # 判断是否是哺乳人员 if employee_id in suckling_employees: # 判断是否是哺乳人员并调整标准打卡时间 current_date = fields.Date.to_date(current_data) if current_data else None if current_date: # 查找当天的哺乳假记录（匹配员工+日期范围） breastfeeding_record = self.env['breastfeeding.holiday'].search([ ('employee_id', '=', employee_id.id), ('calendar_start_time', '<=', current_date), ('calendar_end_time', '>=', current_date), ], limit=1) if breastfeeding_record.time_slot_selection == 'afternoon': sch_check_out_time = sch_check_out_time - timedelta(hours=1) if checkin_datetime > sch_check_out_time: message = f"打卡成功！今日下班时间:{record.checkin_time + timedelta(hours=8)}: {employee_id.name}，祝您归途平安！" if record.employee_id.name not in ['林伟东']: self.sudo().env["wecom.apps"].sync_send_message( category='att', user_id=employee_id.wecom_userid, message=message, ) else: self.sudo().env["wecom.apps"].sync_send_message( category='att', user_id='18562027762', message=message + f'管理员测试管理员测试{vals}', ) self.write({'remark': message}) _logger.info(message) if record.employee_id and record.checkin_time: self._sync_to_hr_attendance(record) _logger.info(f"{record.employee_id.name}--------") return record def _sync_to_hr_attendance(self, record): """同步打卡记录到hr.attendance 更新上下班打卡时间 """ current_date = record.checkin_time.date() + timedelta(hours=8) self.env["hr.attendance.detail"].compute_attendance_detail(record.employee_id, current_date) def compute_attendance_detail(self, employee, current_data, is_manual= False): """重新计算逻辑 :param current_data: 打卡日期 :param employee_id: 员工 :param is_manual: 是否是手动计算 """ # 是否在白名单 resource_user_ids = employee.resource_calendar_id.resource_user_ids.ids is_resource_user = True if employee.id in resource_user_ids else False # 哺乳人员 suckling_employees = employee.resource_calendar_id.breastfeeding_holiday_ids.mapped('employee_id') def get_all_upper_depts(dept): """递归获取部门及其所有上级部门的ID列表、名称列表""" dept_ids = [dept.id] dept_names = [dept.name] if dept.parent_id: upper_ids, upper_names = get_all_upper_depts(dept.parent_id) dept_ids.extend(upper_ids) dept_names.extend(upper_names) return dept_ids, dept_names # 获取所有层级部门ID和部门名称 all_dept_ids, all_dept_names = get_all_upper_depts(employee.department_id) # # 打印所有层级的部门名称 # print("当前员工所属所有层级部门名称：", all_dept_names) if is_resource_user: attendance = self.env['hr.attendance'].search( [('employee_id', '=', employee.id), ('current_data', '=', current_data), ], ) if attendance: attendance.write({ 'is_early_card': False, 'is_late_card': False, 'is_abnormal': False, 'is_no_card': False, 'absenteeism_card': False, 'is_warning': False, }) if not is_resource_user: slot_start = datetime.combine(current_data, datetime.min.time()) - timedelta(hours=8) slot_end = datetime.combine(current_data + timedelta(days=1), datetime.min.time()) - timedelta(hours=8) # 查找员工对应的班次 planning_slots = self.env['planning.slot'].search([ ('resource_id', '=', employee.resource_id.id), ('start_datetime', '>=', slot_start), ('start_datetime', '<', slot_end), ], limit=1) planning_name = planning_slots.planning_name if planning_slots else None template_domain = [] if planning_name: template_domain.append(('planning_name', '=', planning_name)) template_domain.append(('department_id', 'in', all_dept_ids)) template = self.env['planning.slot.template'].search(template_domain, limit=1) if not planning_slots: template = None if not template: role_templates = [ {'role_name': '厨卫（集团）', 'template': None}, {'role_name': '项目运营岗', 'template': None} ] for item in role_templates: item['template'] = self.env['planning.slot.template'].search([('role_id.name', '=', item['role_name'])],) for item in role_templates: if item['template']: employee_ids = item['template'].role_id.resource_ids.employee_id.ids or [] employee_list = list(employee_ids) if employee.id in employee_list: template = item['template'] break if not template: template = self.env['planning.slot.template'].search( [('role_id.name', '=', '集团行政')], limit=1 ) next_day = current_data + timedelta(days=1) start_datetime = datetime.combine(current_data, datetime.min.time()) - timedelta(hours=8) # 判断是否需要包含次日记录 end_datetime = datetime.combine(current_data, datetime.max.time()) - timedelta(hours=8) if template.check_out_start == 0.0: template.check_out_start = 24.0 if template.check_out_end < template.check_out_start: end_datetime = datetime.combine(next_day, datetime.max.time()) - timedelta(hours=8) # 筛选当前员工当天的打卡记录 domain = [ ('employee_id', '=', employee.id), ('checkin_time', '>=', start_datetime), ('checkin_time', '<=', end_datetime) ] attendance_records = self.search(domain) checkin_times = [rec.checkin_time + timedelta(hours=8) for rec in attendance_records if rec.checkin_time] check_in_start = self.float_to_datetime(current_data, template.check_in_start) check_in_end = self.float_to_datetime(current_data, template.check_in_end) if template.check_in_end < template.check_in_start: check_in_end = self.float_to_datetime(next_day, template.check_in_end) valid_checkins = [ t for t in checkin_times if check_in_start <= t <= check_in_end ] # 3. 确定有效上下班时间 # 上班时间：在上班范围内的最小打卡时间 check_in_time = min(valid_checkins) - timedelta(hours=8) if valid_checkins else None check_out_start = self.float_to_datetime(current_data, template.check_out_start) check_out_end = self.float_to_datetime(current_data, template.check_out_end) if template.check_out_start == 0.0: template.check_out_start = 24.0 if template.check_out_end < template.check_out_start: check_out_end = self.float_to_datetime(next_day, template.check_out_end) # 筛选在下班时间范围内的打卡记录 valid_checkouts = [t for t in checkin_times if check_out_start <= t <= check_out_end] # 取最大的有效下班时间 check_out_time = max(valid_checkouts) - timedelta(hours=8) if valid_checkouts else None attendance = self.env['hr.attendance'].search( [('employee_id', '=', employee.id), ('current_data', '=', current_data), ], ) if check_in_time and check_out_time and check_in_time == check_out_time: check_out_time = None # 获取正常上下班时间 sch_check_in_time, sch_check_out_time = self.normal_clock_in_time(template, current_data) # 判断是否是哺乳人员 if employee in suckling_employees: # 判断是否是哺乳人员并调整标准打卡时间 current_date = fields.Date.to_date(current_data) if current_data else None if current_date: # 查找当天的哺乳假记录（匹配员工+日期范围） breastfeeding_record = self.env['breastfeeding.holiday'].search([ ('employee_id', '=', employee.id), ('calendar_start_time', '<=', current_date), ('calendar_end_time', '>=', current_date), ], limit=1) if breastfeeding_record: if breastfeeding_record.time_slot_selection == 'morning': sch_check_in_time = sch_check_in_time + timedelta(hours=1) else: sch_check_out_time = sch_check_out_time - timedelta(hours=1) is_early_card, is_late_card, is_abnormal, is_warning, absenteeism_card, is_no_card = self.state_judgment(employee, check_in_time, check_out_time, sch_check_in_time, sch_check_out_time, current_data, is_manual, attendance) # 如果是厨卫的 只用判断范围内是否打卡了 如果打卡就不算异常 if self._process_shift_days(current_data): if template.role_id.name == '厨卫（集团）': is_early_card = False is_late_card = False if check_in_time and check_out_time: is_abnormal = False is_no_card = False absenteeism_card = False else: is_abnormal = True is_no_card = True absenteeism_card = True is_expatriation = self.env['expatriation.approval'].search([('work_end', '=', True),('expatriates_ids', 'in', [employee.id]), ('is_effect', '=', False)]) # if not self._process_shift_days(current_data) and template.role_id.name == '集团行政': # start_datetime = datetime.combine(current_data, datetime.min.time()) # end_datetime = datetime.combine(current_data, datetime.max.time()) # domain = [ # ('employee_id', '=', employee.id), # ('checkin_time', '>=', start_datetime), # ('checkin_time', '<=', end_datetime) # ] # attendance_records = self.search(domain) # if attendance_records: # checkin_time_list = [rec.checkin_time for rec in attendance_records if rec.checkin_time] # if checkin_time_list: # min_time = min(checkin_time_list) # max_time = max(checkin_time_list) # if min_time == max_time: # check_in_time = min_time # check_out_time = None # else: # check_in_time = min_time # check_out_time = max_time if attendance: attendance.write({ 'check_in': check_in_time, 'check_out': check_out_time, 'shift_id': planning_slots.id, 'shift_name': planning_slots.display_name, 'is_early_card': is_early_card, 'is_late_card': is_late_card, 'is_abnormal': is_abnormal, 'is_warning': is_warning, 'is_check_in_red': False, 'is_check_out_red': False, 'is_approval': False, 'is_no_card': is_no_card, 'is_holiday': False, 'is_week': False, 'absenteeism_card': absenteeism_card, 'is_expatriation': is_expatriation, }) else: attendance = self.env['hr.attendance'].create({ 'employee_id': employee.id, 'shift_id': planning_slots.id, 'shift_name': planning_slots.display_name, 'current_data': current_data, 'check_in': check_in_time, 'check_out': check_out_time, 'sch_check_in_time': sch_check_in_time, 'sch_check_out_time': sch_check_out_time, 'is_early_card': is_early_card, 'is_late_card': is_late_card, 'is_abnormal': is_abnormal, 'is_no_card': is_no_card, 'is_warning': is_warning, 'is_check_in_red': False, 'is_check_out_red': False, 'is_approval': False, 'is_expatriation': is_expatriation, 'is_holiday': False, 'is_week': False, 'absenteeism_card': absenteeism_card, }) attendance.update_attendance(employee, current_data) else: att_id = self.env['hr.attendance'].search([('employee_id', '=', employee.id), ('current_data', '=', current_data)]) if att_id: att_id.write({ 'is_no_card': False, 'is_early_card': False, 'is_late_card': False, 'is_abnormal': False, 'is_warning': False, 'absenteeism_card': False, }) 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 if template.planning_name == '集团行政': 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 float_to_datetime(self, base_date, hour_float): """ 将小时浮点数（如9.5表示9:30）转换为具体datetime :param base_date: 基准日期（date对象） :param hour_float: 小时浮点数 :return: datetime对象 """ hours = int(hour_float) minutes = int((hour_float - hours) * 60) return datetime.combine(base_date, datetime.min.time()) + timedelta(hours=hours, minutes=minutes) def state_judgment(self, employee, check_in, check_out, start_datetime, end_datetime, current_data, is_manual, attendance): """ 判断打卡状态 :param check_in: 签到时间 :param check_out: 签出时间 :param start_datetime: 标准上班打卡时间 :param end_datetime: 标准下班打卡时间 :param current_data: 打卡日期 :param is_manual: 是否为手工补录 :return: is_early_card, is_late_card, is_abnormal, is_warning, absenteeism_card """ is_early_card = False is_late_card = False is_abnormal = False is_warning = False absenteeism_card = False is_no_card = False # 判断是否为外派中 expat_approval = self.env['expatriation.approval'].search([ ('is_effect', '=', False), ('work_end', '=', True), ('expatriates_ids', 'in', [employee.id]) ], limit=1) # 查找有效的返回审批单 back_approval = self.env['back.approval'].search([ ('is_void', '=', False), ('work_end', '=', True), ('expatriates_id', '=', employee.id) ], limit=1) # 3. 外派中：当前时间 >= 外派开始日期 → 所有状态为False if expat_approval and expat_approval.start_date: if current_data >= expat_approval.start_date: return is_early_card, is_late_card, is_abnormal, is_warning, absenteeism_card, is_no_card # 4. 返回期间：打卡日期在返回的开始-结束日期之间 → 所有状态为False if back_approval and back_approval.start_date and back_approval.end_date: if back_approval.start_date <= current_data <= back_approval.end_date: return is_early_card, is_late_card, is_abnormal, is_warning, absenteeism_card, is_no_card def remove_seconds(dt): """清除datetime对象的秒数和微秒""" if dt: return dt.replace(second=0, microsecond=0) return dt # 清除所有时间的秒数 check_in = remove_seconds(check_in) check_out = remove_seconds(check_out) start_datetime = remove_seconds(start_datetime) end_datetime = remove_seconds(end_datetime) if employee.department_id and employee.department_id.name == '厨卫': start_datetime = start_datetime.replace(hour=1, minute=0, second=0, microsecond=0) # 判断是否是行政班次 employee_list = [] role_ids = self.env['planning.role'].search([('name', 'not in', ['厨卫（集团）', '项目运营岗'])]) for role in role_ids: for resource_id in role.resource_ids: employee_list.append(resource_id.employee_id.id) if employee.id not in employee_list: # 判断是否工作日 if not self._process_shift_days(current_data): return is_early_card, is_late_card, is_abnormal, is_warning, absenteeism_card, is_no_card is_tianhe_property = employee.company_id.name == "天河物管" # 只要是旷工 就不算早退 或者迟到 # 迟到判断 if check_in and start_datetime and check_in > start_datetime: late_delta = (check_in - start_datetime).total_seconds() / 60 if is_tianhe_property: # 天河物管：2分钟包容时间；每月最多3次宽容迟到（不再依赖 employee.late_allowed 字段） if late_delta > 2: # 统计当前月份该员工已使用的宽容迟到次数（is_late_allowed=True） month_first = current_data.replace(day=1) if current_data.month == 12: month_last = current_data.replace(day=31) else: month_last = current_data.replace(month=current_data.month + 1, day=1) - timedelta(days=1) used_count = self.env['hr.attendance'].search_count([ ('employee_id', '=', employee.id), ('current_data', '>=', month_first), ('current_data', '<=', month_last), ('is_late_allowed', '=', True), ('id', '!=', attendance.id), ]) if used_count >= 3: # 本月已满3次宽容，不再宽容，直接算迟到（黄色也不显示） attendance.is_late_allowed = False is_late_card = True else: if attendance.is_late_allowed == False: attendance.is_late_allowed = True if late_delta > 10: is_late_card = True # 迟到超过60分钟标记为旷工 if late_delta > 60: absenteeism_card = True is_late_card = False else: if check_in > start_datetime: is_late_card = True # 迟到超过60分钟标记为旷工 if late_delta > 60: absenteeism_card = True is_late_card = False if not check_out: is_no_card = True absenteeism_card = True is_abnormal = True # 早退判断 if check_out and end_datetime: early_delta = (end_datetime - check_out).total_seconds() / 60 if is_tianhe_property: # 天河物管：超过2分钟才算早退 if early_delta > 2: is_early_card = True if early_delta > 60: absenteeism_card = True is_early_card = False else: if check_out < end_datetime: is_early_card = True if early_delta > 60: absenteeism_card = True is_early_card = False # 异常和警告判断 if is_early_card or is_late_card or absenteeism_card: is_abnormal = True is_warning = True if is_manual == True: if not check_out or not check_in: is_no_card = True absenteeism_card = True is_warning = True is_abnormal = True return is_early_card, is_late_card, is_abnormal, is_warning, absenteeism_card, is_no_card 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), ('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] 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}日' def _get_wx_attendance_user_list(self): """ 获取企业微信的关联用户 :return: 用户列表 """ user_obj = self.sudo().env['res.users'].search([('employee_id.id', '!=', False)]) total_users = len(user_obj) remaining_users = total_users user_list = [] for user in user_obj: if user.oauth_uid: _logger.info("{} id:{}，总:{}，剩:{}，已:{}，{}".format(user.name, user.id, total_users, remaining_users, total_users - remaining_users, user.company_id.name)) remaining_users -= 1 user_list.append(user.oauth_uid) return user_list @staticmethod def _get_wx_attendance_data(userid_limit, now_time, start_time): """ 获取企业微信的打卡信息 :param wx_work_checkin_api: 企业微信api :param userid_limit: 用户列表 :return: """ start_time_stamp = int(start_time.timestamp()) end_time_stamp = int(now_time.timestamp()) # 整理请求信息 data = { "opencheckindatatype": 3, "starttime": start_time_stamp, "endtime": end_time_stamp, "need_photo": 1, "useridlist": userid_limit } # 重写一个类处理result获取想要的结果 return data @staticmethod def _check_wx_attendance_data(response, userid_limit): """ 检查请求返回的信息是否正确 :param response: 请求返回信息 :param userid_limit: 用户列表 :return: 检查结果 """ # 对返回信息进行整理 if response.status_code != 200: _logger.error('抓取出勤数据请求失败\n异常:{0}\n用户列表{1}'.format(str(response.json()), str(userid_limit))) return [] result = response.json() if result.get('errcode') != 0: # 301021至少有一个不存在通讯录 # 301024获取打卡间隔受限 _logger.error('获取外出打卡数据失败\n异常:{0}\n数量:{1}\n用户列表:{2}'.format(str(result.get('errmsg')), len(userid_limit), str(userid_limit))) return [] checkin_data = result.get('checkindata', []) if not checkin_data: _logger.error('没有找到外出打卡记录\n异常:{0}\n用户列表{1}'.format(str(response.json()), str(userid_limit))) return [] return checkin_data def _check_wx_attendance_time(self, data, userid_limit): """ 检查保存之前的数据是否准确，并且分析出想要的打卡时间 :param data: 原始数据 :param userid_limit: 用户列表 :return: 分析后的打卡数据 """ userid = data.get("userid", "") checkin_time_stamp = data.get("checkin_time", "") sch_checkin_time_stamp = data.get("sch_checkin_time", None) if not userid or not checkin_time_stamp: _logger.error('没有抓取到用户出勤信息,异常:{0},用户列表{1}'.format(str(data), str(userid_limit))) return False checkin_time = datetime.fromtimestamp(checkin_time_stamp) if checkin_time_stamp else None sch_checkin_time = datetime.fromtimestamp(sch_checkin_time_stamp) if sch_checkin_time_stamp else None return checkin_time, sch_checkin_time def _check_wx_attendance_media(self, data): """ 分析返回来的数据获取图片原图 :param data: 原始数据 :param userid_limit: 用户列表 :return: 获取后的图片 """ media_list = [] userid = data.get("userid", "") mediaids_list = data.get("mediaids", []) if not mediaids_list: return media_list for media_id in mediaids_list: data = {"media_id": media_id} try: result, response = self.sudo().env["wecom.apps"].sync_media_get(data) if result: filename = response.headers['Content-Disposition'].split('; ')[2].replace('filename=', '').replace('"', '') wx_work_auth = self.env['auth.oauth.provider'].search([('name', '=', 'Wechat Card')], limit=1) wx_work_image = os.path.join(wx_work_auth.wx_work_image_path, str(filename)) with open(wx_work_image, 'wb') as f:f.write(response.content) media_list.append((0, 0, {'media_url': filename,})) else: _logger.error('检查保存之前的数据分析出想要的数据失败,异常:{0},数据信息{1}'.format(str(response), str(data))) except Exception as e: _logger.error('请求原始图片失败,异常:{0},数据信息{1},异常详情:{2}'.format(str(e), str(data),str(traceback.format_exc()))) return media_list def _save_wx_attendance_detail_data(self, data, checkin_time, sch_checkin_time): """ 保存相关的企业微信的原始打卡信息 :param data: 请求返回后的处理信息 :param checkin_time: 打卡时间 :param sch_checkin_time: 应打卡时间 :param media_list: 图片列表 :return: 保存结果 """ userid = data.get("userid", "") # 查询是否存在数据 if userid and checkin_time and self.env['hr.attendance.detail'].sudo().search_count( [("userid", "=", userid), ("checkin_time", "=", checkin_time)]) <= 0: res_user_obj = self.env['res.users'].sudo().search([("oauth_uid", "=", userid)], limit=1) hr_employee_obj = self.env['hr.employee'].sudo().search([("user_id", "=", res_user_obj.id)], limit=1) if not res_user_obj or not hr_employee_obj: _logger.error('检查保存之前的数据异常,异常:无关联的员工数据,数据信息{0}'.format(str(data))) return False attendance_detail_obj = self.env['hr.attendance.detail'].sudo().create({ 'employee_id': hr_employee_obj.id, "userid": userid, "groupname": data.get("groupname", ""), "checkin_type": data.get("checkin_type", ""), "exception_type": data.get("exception_type", ""), "checkin_time": checkin_time, "location_title": data.get("location_title", ""), "location_detail": data.get("location_detail", ""), "wifiname": data.get("wifiname", ""), "notes": data.get("notes", ""), "wifimac": data.get("wifimac", ""), "mediaids": str(data.get("mediaids", "")), "lat": data.get("lat", 0), "lng": data.get("lng", 0), "deviceid": data.get("deviceid", ""), "sch_checkin_time": sch_checkin_time, "groupid": data.get("groupid", 0), "schedule_id": data.get("schedule_id", ""), "timeline_id": data.get("timeline_id", 0), # "attendance_detail_image_ids": media_list }) _logger.info(f"{hr_employee_obj.name}----创建一条数据") # 分析返回来的数据获取图片原图 try: media_list = self._check_wx_attendance_media(data) attendance_detail_obj.attendance_detail_image_ids = media_list except Exception as e: _logger.error('检查保存之前的数据分析出想要的数据失败,异常:{0},数据信息{1},异常详情:{2}'.format(str(e), str(data), str(traceback.format_exc()))) def get_hr_attendance_time_wechat(self): """ 获取企业微信打卡数据的核心逻辑 :return: 处理结果 """ # 从用户列表里获取企业微信的用户 user_list = self._get_wx_attendance_user_list() # now_time = datetime(2025, 10, 31) # 替换为你需要的年份 # start_time = datetime(2025, 10, 1) # 替换为你需要的年份 now_time = datetime.now() + timedelta(hours=8) start_time = now_time - timedelta(days=25) # 企业微信获取打卡数据，要求用户分组100，分批获取打卡记录 # for userid_limit in self.utils_list_page(user_list, 1): for userid_limit in user_list: # 获取企业微信的打卡信息 try: attendance_data = self._get_wx_attendance_data(userid_limit, now_time, start_time) attendance_result, response = self.sudo().env["wecom.apps"].sync_checkin_data(attendance_data) except Exception as e: _logger.error('获取企业微信的打卡信息失败,异常:{0},用户列表{1},异常详情:{2}'.format(str(e), str(userid_limit), str(traceback.format_exc()))) continue # 检查请求返回的信息是否正确 try: checkin_data = self._check_wx_attendance_data(response, userid_limit) except Exception as e: _logger.error('检查请求返回的信息失败,异常:{0},用户列表{1},异常详情:{2}'.format(str(e), str(userid_limit), str(traceback.format_exc()))) continue # 将原始数据存储 for data in checkin_data: # 检查保存之前的数据是否准确，并且分析出想要的打卡时间 try: checkin_time, sch_checkin_time = self._check_wx_attendance_time(data, userid_limit) except Exception as e: _logger.error('检查保存之前的数据分析出想要的数据失败,异常:{0},数据信息{1},异常详情:{2}'.format(str(e), str(data), str(traceback.format_exc()))) continue # 保存相关的企业微信的原始打卡信息 try: self._save_wx_attendance_detail_data(data, checkin_time, sch_checkin_time) except Exception as e: _logger.error( '保存相关的企业微信的原始打卡信息失败,异常:{0},数据信息{1},异常详情:{2}'.format(str(e), str(data), str(traceback.format_exc()))) for user in user_list: for current_date in self.sudo().divide_time_range_fixed(start_time, now_time): self.sudo()._do_save_wx_attendance_data(user, current_date) self.sudo()._do_plan_save_wx_attendance_data(user, current_date) def _do_save_wx_attendance_data(self, user, current_date): """ 开始 整理考勤数据 :param user: :param current_date: :return: """ # 同步整理当天出勤数据 attendance_obj = self.env['hr.attendance'].search([("userid", "=", user), ("current_data", "=", current_date)], limit=1) hr_employee_obj = self.env['hr.employee'].sudo().search([("user_id.oauth_uid", "=", user)], limit=1) # if not attendance_obj: # 判断是否是 普通打卡 # if not hr_employee_obj.default_planning_role_id: # attendance_obj = self.env['hr.attendance'].sudo().create({ # 'employee_id': hr_employee_obj.id, # "userid": user.oauth_uid, # "current_data": current_date, # "attendance_type": "wx_work" # }) # _logger.info(f'创建一条考勤记录,员工:{hr_employee_obj.name}-日期{current_date}') if not hr_employee_obj.default_planning_role_id: if attendance_obj.is_approval: # 已经审批修复无需再次进行更新数据 return True self._do_not_planning_role(user, current_date, attendance_obj) return True def _do_plan_save_wx_attendance_data(self, user, current_date): """ 开始 整理 计划排版的 考勤数据 :param user: :param current_date: :return: """ # 同步整理当天出勤数据 hr_employee_obj = self.env['hr.employee'].sudo().search([("user_id", "=", user)], limit=1) current_datetime_start = datetime.combine(datetime.strptime(current_date, '%Y-%m-%d'), datetime.min.time()) current_datetime_end = datetime.combine(datetime.strptime(current_date, '%Y-%m-%d'), datetime.max.time()) # 判断是否是 排版打卡 if hr_employee_obj.default_planning_role_id: planning_slot_ids = self.env['planning.slot'].search( [('resource_id', '=', hr_employee_obj.resource_id.id), ('start_datetime', '>=', current_datetime_start - timedelta(hours=8)), ('start_datetime', '<=', current_datetime_end - timedelta(hours=8))]) for planning_slot_id in planning_slot_ids: # 查询其中的一个排版信息是否生成了打卡记录 attendance_obj = self.env['hr.attendance'].search( [("userid", "=", user), ("current_data", "=", current_date), ("shift_id", "=", planning_slot_id.id) ], limit=1) # if not attendance_obj: # self.env['hr.attendance'].sudo().create({ # 'employee_id': hr_employee_obj.id, # "userid": user, # "current_data": current_date, # "shift_id": planning_slot_id.id, # "attendance_type": "wx_work" # }) # _logger.info(f'创建一条考勤记录,员工:{hr_employee_obj.name}-日期{current_date}') if attendance_obj.is_approval: # 已经审批修复无需再次进行更新数据 return True self._do_planning_role(user, current_date, attendance_obj, planning_slot_id) return True def _do_not_planning_role(self, user, current_date, attendance_obj): """ 正常班次的 上班下班判断 :param user: 用户集合 :param current_date: 循环判断的时间 :param attendance_obj: 判断的时刻的考勤集合 :return: """ current_datetime_start = datetime.combine(datetime.strptime(current_date, '%Y-%m-%d'), datetime.min.time()) current_datetime_end = datetime.combine(datetime.strptime(current_date, '%Y-%m-%d'), datetime.max.time()) hr_employee_obj = self.env['hr.employee'].sudo().search([("user_id.oauth_uid", "=", user)], limit=1) # 白名单 resource_user_ids = hr_employee_obj.resource_calendar_id.resource_user_ids.ids is_resource_user = True if hr_employee_obj.id in resource_user_ids else False # 判断是否是星期 tz = hr_employee_obj.resource_calendar_id.tz current_year = current_date.split('-')[0] current_month = current_date.split('-')[1] current_day = current_date.split('-')[2] current_week = calendar.weekday(int(current_year), int(current_month), int(current_day)) work_time_result = self.get_employee_work_time(hr_employee_obj, current_week, current_date, tz) # # 说明今天休息 # if not work_time_result: # # 如果有人打开那么进行记录 # # attendance_obj.sudo().write({ # "is_late_card": False, # "is_early_card": False, # "is_no_card": False, # "is_abnormal": False, # }) # return True # 今天正常上班 # go_from, go_to, off_from, off_to, current_date = work_time_result # 获取考勤日历的上班下班时间 min_hour_from = self.env['resource.calendar'].search([])[0].attendance_ids[0].hour_from min_hour = int(min_hour_from) min_minute = int(min_hour_from * 10 % min_hour_from) max_hour_to = self.env['resource.calendar'].search([])[0].attendance_ids[-1].hour_to max_hour = int(max_hour_to) max_minute = int(max_hour_to * 10 % max_hour_to) will_datetime_start = current_datetime_start.replace(hour=min_hour, minute=min_minute) - timedelta(hours=8) will_datetime_end = current_datetime_end.replace(hour=max_hour, minute=max_minute) - timedelta(hours=8) # 判断打卡情况排除未打卡情况 # 6：00 - 17：00 // 10.00 - // 5.00。 # 改为区间打卡 check_in_start = datetime.strptime(str(current_date) + " " + "06:00", "%Y-%m-%d %H:%M") - timedelta(hours=8) check_in_end = datetime.strptime(str(current_date) + " " + "17:00", "%Y-%m-%d %H:%M") - timedelta(hours=8) check_out_start = datetime.strptime(str(current_date) + " " + "10:00", "%Y-%m-%d %H:%M") - timedelta(hours=8) check_out_end = datetime.strptime(str(current_date) + " " + "05:00", "%Y-%m-%d %H:%M") + relativedelta(days=1) - timedelta(hours=8) # 说明今天休息 if not work_time_result: # 如果有人打卡那么进行记录 相当于白名单 # 上班 attendance_detail_obj_in = self.env['hr.attendance.detail'].sudo().search( [("exception_type", "!=", "未打卡"), ("userid", "=", user), ("checkin_time", ">=", check_in_start), ("checkin_time", "<=", check_in_end)]) if len(attendance_detail_obj_in) > 0: # 关联 打卡详情 attendance_detail_obj_in.hr_attendance_id = attendance_obj.id # 多次打卡时间的集合 checkin_time_list = [obj.checkin_time for obj in attendance_detail_obj_in] if checkin_time_list: # 最早 时间 min_time = max(checkin_time_list) attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": False, "is_early_card": False, "is_no_card": False, "is_abnormal": False, }) # _logger.info(f'更新一条考勤记录,日期-{attendance_obj.current_date}-最小时间{min_time}') # 下班 attendance_detail_obj_out = self.env['hr.attendance.detail'].sudo().search( [("exception_type", "!=", "未打卡"), ("userid", "=", user), ("checkin_time", ">=", check_out_start), ("checkin_time", "<=", check_out_end), ]) if len(attendance_detail_obj_out) > 0: attendance_detail_obj_out.hr_attendance_id = attendance_obj.id checkout_time_list = [obj.checkin_time for obj in attendance_detail_obj_out] if checkout_time_list: # 最早 时间 max_time = max(checkout_time_list) # # 没有上班卡 # if not attendance_obj.check_in: # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": False, # "is_no_card": False, # "is_abnormal": False, # "absenteeism_card": False, # }) # elif attendance_obj.check_in and attendance_obj.check_in < max_time: attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_late_card": False, "is_early_card": False, "is_no_card": False, "is_abnormal": False, }) # _logger.info(f'更新一条考勤记录,日期-{attendance_obj.current_date}-最大时间{max_time}') attendance_obj.sudo().write({ "is_late_card": False, "is_early_card": False, "is_no_card": False, "is_abnormal": False, }) return True # 上班 attendance_detail_obj_in = self.env['hr.attendance.detail'].sudo().search( [("exception_type", "!=", "未打卡"), ("userid", "=", user), ("checkin_time", ">=", check_in_start), ("checkin_time", "<=", check_in_end)]) # 关联 打卡详情 attendance_detail_obj_in.hr_attendance_id = attendance_obj.id # 白名单用户直接正常 if is_resource_user: if len(attendance_detail_obj_in) > 0: # 多次打卡时间的集合 checkin_time_list = [obj.checkin_time for obj in attendance_detail_obj_in] # 最早 时间 min_time = min(checkin_time_list) attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": False, "is_no_card": False, "is_abnormal": False, "absenteeism_card": False, }) # _logger.info(f'更新一条考勤记录,日期-{attendance_obj.current_date}-最小时间{min_time}') else: # 没有打卡记录说明缺卡 if len(attendance_detail_obj_in) <= 0: attendance_obj.sudo().write({"is_no_card": True, "is_abnormal": False, "absenteeism_card": False}) else: # 多次打卡时间的集合 checkin_time_list = [obj.checkin_time for obj in attendance_detail_obj_in] if not checkin_time_list: attendance_obj.sudo().write({"is_no_card": True, "is_abnormal": False, "absenteeism_card": False}) else: # 最早 时间 min_time = min(checkin_time_list) # 查询最早时间上班时间 查询最晚下班时间 # 判断时间 在10分钟内说明是 异常 if will_datetime_start + timedelta(minutes=10) > min_time > will_datetime_start + timedelta(minutes=1): attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": False, "is_no_card": False, "is_abnormal": True, "absenteeism_card": False, }) elif will_datetime_start + timedelta(hours=1) > min_time > will_datetime_start + timedelta(minutes=1): attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": True, "is_no_card": False, "is_abnormal": False, "absenteeism_card": False, }) elif min_time > will_datetime_start + timedelta(hours=1) and min_time > will_datetime_start + timedelta(minutes=1): attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": False, "is_no_card": False, "is_abnormal": False, "absenteeism_card": True, }) else: attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": False, "is_no_card": False, "is_abnormal": False, "absenteeism_card": False, }) # 下班 attendance_detail_obj_out = self.env['hr.attendance.detail'].sudo().search( [("exception_type", "!=", "未打卡"), ("userid", "=", user), ("checkin_time", ">=", check_out_start), ("checkin_time", "<=", check_out_end), ]) # 关联 打卡详情 attendance_detail_obj_out.hr_attendance_id = attendance_obj.id # 白名单用户直接正常 if is_resource_user: # 多次打卡时间的集合 checkin_time_list = [obj.checkin_time for obj in attendance_detail_obj_out] if checkin_time_list: # 最早 时间 max_time = max(checkin_time_list) # # 如果上班没有打卡 # if not attendance_obj.check_in: # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": False, # "is_no_card": False, # "is_abnormal": False, # "absenteeism_card": False, # }) # # 如果上班时间早于下班时间说明 # elif attendance_obj.check_in and attendance_obj.check_in < max_time: attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_early_card": False, "is_no_card": False, "is_abnormal": False, "absenteeism_card": False, }) else: if len(attendance_detail_obj_out) <= 0: attendance_obj.sudo().write({ "is_no_card": True, "is_abnormal": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) # 打了一次卡记录说明缺卡 else: # 多次打卡时间的集合 checkin_time_list = [obj.checkin_time for obj in attendance_detail_obj_out] if not checkin_time_list: attendance_obj.sudo().write({ "is_no_card": True, "is_abnormal": True if attendance_obj.check_in and attendance_obj.is_abnormal else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) else: # 最早 时间 max_time = max(checkin_time_list) # # 如果上班没有打卡 # if not attendance_obj.check_in: # if will_datetime_start + timedelta(minutes=10) > max_time > will_datetime_start + timedelta(minutes=1): # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": False, # "is_no_card": True, # "is_abnormal": True, # "absenteeism_card": False, # }) # elif will_datetime_start + timedelta(hours=1) > max_time > will_datetime_start + timedelta(minutes=1): # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": True, # "is_no_card": True, # "is_abnormal": False, # "absenteeism_card": False, # }) # elif max_time > will_datetime_start + timedelta(hours=1) and max_time > will_datetime_start + timedelta(minutes=1): # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": False, # "is_no_card": True, # "is_abnormal": False, # "absenteeism_card": True, # }) # else: # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": False, # "is_no_card": True, # "is_abnormal": False, # "absenteeism_card": False, # }) # 如果上班时间早于下班时间说明下班的卡没打 if attendance_obj.check_in: if attendance_obj.check_in >= max_time: attendance_obj.sudo().write({ "is_no_card": True, "is_abnormal": True if attendance_obj.check_in and attendance_obj.is_abnormal else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) else: if will_datetime_end - timedelta(minutes=10) < max_time < will_datetime_end - timedelta(minutes=1): attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_early_card": False, "is_no_card": True if attendance_obj.check_in and attendance_obj.is_no_card else False, "is_abnormal": True, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) elif will_datetime_end - timedelta(hours=1) < max_time < will_datetime_end - timedelta(minutes=1): attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_early_card": True, "is_no_card": True if attendance_obj.check_in and attendance_obj.is_no_card else False, "is_abnormal": True if attendance_obj.check_in and attendance_obj.is_abnormal else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) elif max_time < will_datetime_end - timedelta(hours=1): attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_early_card": False, "is_no_card": True if attendance_obj.check_in and attendance_obj.is_no_card else False, "is_abnormal": True if attendance_obj.check_in and attendance_obj.is_abnormal else False, "absenteeism_card": True, }) else: attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_early_card": False, "is_no_card": True if attendance_obj.check_in and attendance_obj.is_no_card else False, "is_abnormal": True if attendance_obj.check_in and attendance_obj.is_abnormal else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) return True def _do_planning_role(self, user, current_date, attendance_obj, planning_id): """ 带有计划的班次判断 :param user: 用户集合 :param current_date: 循环判断的时间 :param attendance_obj: 判断的时刻的考勤集合 :param planning_id: :return: """ # 获取模板信息 template_id = planning_id.template_id # check_in_start = fields.Datetime(string="上班开始") check_in_start = template_id.check_in_start # check_in_end = fields.Datetime(string="上班结束") check_in_end = template_id.check_in_end # check_out_start = fields.Datetime(string="下班开始") check_out_start = template_id.check_out_start # check_out_end = fields.Datetime(string="下班结束") check_out_end = template_id.check_out_end # 这里需要将 浮点型时间转为 时间型 if check_in_start: check_in_start = self.str_to_time(current_date, check_in_start) if check_in_end: check_in_end = self.str_to_time(current_date, check_in_end) if check_out_start: check_out_start = self.str_to_time(current_date, check_out_start) if check_out_end: check_out_end = self.str_to_time(current_date, check_out_end) current_datetime_start = datetime.combine(datetime.strptime(current_date, '%Y-%m-%d'), datetime.min.time()) current_datetime_end = datetime.combine(datetime.strptime(current_date, '%Y-%m-%d'), datetime.max.time()) # 如果没有设定区间那么默认当天最早时间和最晚时间 到 应该上下班时间 if not check_in_start: check_in_start = current_datetime_start if not check_in_end: check_in_end = planning_id.start_datetime + timedelta(hours=8) if not check_out_start: check_out_start = planning_id.end_datetime + timedelta(hours=8) if not check_out_end: check_out_end = current_datetime_end # 如果结束时间小于开始时间说明 是第二天的时间 if check_in_end < check_in_start: check_in_end = check_in_end + relativedelta(days=1) if check_out_end < check_out_start: check_out_end = check_out_end + relativedelta(days=1) # 转换时区 check_in_start = check_in_start - timedelta(hours=8) check_in_end = check_in_end - timedelta(hours=8) check_out_start = check_out_start - timedelta(hours=8) check_out_end = check_out_end - timedelta(hours=8) # 判断打卡情况排除未打卡情况 # 上班 attendance_detail_obj_in = self.env['hr.attendance.detail'].sudo().search( [("exception_type", "!=", "未打卡"), ("userid", "=", user), ("checkin_time", ">=", check_in_start), ("checkin_time", "<=", check_in_end), ]) # 关联 打卡详情 attendance_detail_obj_in.hr_attendance_id = attendance_obj.id will_datetime_start = planning_id.start_datetime will_datetime_end = planning_id.end_datetime if len(attendance_detail_obj_in) <= 0: attendance_obj.sudo().write({"is_no_card": True, "is_abnormal": False, "absenteeism_card": False}) # 打了一次卡记录说明缺卡 else: # 多次打卡时间的集合 checkin_time_list = [obj.checkin_time for obj in attendance_detail_obj_in] if not checkin_time_list: attendance_obj.sudo().write({"is_no_card": True, "is_abnormal": False, "absenteeism_card": False}) else: # 最早 时间 min_time = min(checkin_time_list) # attendance_obj.sudo().write({ # "sch_check_in_time": planning_id.start_datetime, # "check_in": min_time, # "is_late_card": min_time > planning_id.start_datetime, # "is_no_card": False, # "is_abnormal": False, # }) if will_datetime_start + timedelta(minutes=10) > min_time > will_datetime_start: attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": False, "is_no_card": False, "is_abnormal": True, "absenteeism_card": False, }) elif will_datetime_start + timedelta(hours=1) > min_time > will_datetime_start: attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": True, "is_no_card": False, "is_abnormal": False, "absenteeism_card": False, }) elif min_time > will_datetime_start + timedelta(hours=1) and min_time > will_datetime_start: attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": False, "is_no_card": False, "is_abnormal": False, "absenteeism_card": True, }) else: attendance_obj.sudo().write({ "sch_check_in_time": will_datetime_start, "check_in": min_time, "is_late_card": False, "is_no_card": False, "is_abnormal": False, "absenteeism_card": False, }) # 下班 attendance_detail_obj_out = self.env['hr.attendance.detail'].sudo().search( [("exception_type", "!=", "未打卡"), ("userid", "=", user), ("checkin_time", ">=", check_out_start), ("checkin_time", "<=", check_out_end), ]) # 关联 打卡详情 attendance_detail_obj_out.hr_attendance_id = attendance_obj.id if len(attendance_detail_obj_out) <= 0: attendance_obj.sudo().write({ "is_no_card": True, "is_abnormal": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) # 打了一次卡记录说明缺卡 else: # 多次打卡时间的集合 checkin_time_list = [obj.checkin_time for obj in attendance_detail_obj_out] if not checkin_time_list: attendance_obj.sudo().write({ "is_no_card": True, "is_abnormal": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) else: # 最早 时间 max_time = max(checkin_time_list) # 如果上班没有打卡 # if not attendance_obj.check_in: # # attendance_obj.sudo().write({ # # "sch_check_in_time": planning_id.start_datetime, # # "check_in": max_time, # # "is_late_card": max_time > planning_id.start_datetime, # # "is_no_card": True, # # "is_abnormal": True, # # }) # # if will_datetime_start + timedelta(minutes=10) > max_time > will_datetime_start: # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": False, # "is_no_card": True, # "is_abnormal": True, # "absenteeism_card": False, # }) # elif will_datetime_start + timedelta(hours=1) > max_time > will_datetime_start: # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": True, # "is_no_card": True, # "is_abnormal": False, # "absenteeism_card": False, # }) # elif max_time > will_datetime_start + timedelta(hours=1) and max_time > will_datetime_start: # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": False, # "is_no_card": True, # "is_abnormal": False, # "absenteeism_card": True, # }) # else: # attendance_obj.sudo().write({ # "sch_check_in_time": will_datetime_start, # "check_in": max_time, # "is_late_card": False, # "is_no_card": True, # "is_abnormal": False, # "absenteeism_card": False, # }) # 如果上班时间早于下班时间说明下班的卡没打 if attendance_obj.check_in: if attendance_obj.check_in >= max_time: attendance_obj.sudo().write({ "is_no_card": True, "is_abnormal": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) else: # attendance_obj.sudo().write({ # "sch_check_out_time": planning_id.end_datetime, # "check_out": max_time, # "is_early_card": max_time < planning_id.end_datetime, # "is_no_card": False, # "is_abnormal": False, # }) if will_datetime_end - timedelta(minutes=10) < max_time < will_datetime_end: attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_early_card": False, "is_no_card": True if attendance_obj.check_in and attendance_obj.is_no_card else False, "is_abnormal": True, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) elif will_datetime_end - timedelta(hours=1) < max_time < will_datetime_end: attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_early_card": True, "is_no_card": True if attendance_obj.check_in and attendance_obj.is_no_card else False, "is_abnormal": True if attendance_obj.check_in and attendance_obj.is_abnormal else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) elif max_time < will_datetime_end - timedelta(hours=1): attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_early_card": False, "is_no_card": True if attendance_obj.check_in and attendance_obj.is_no_card else False, "is_abnormal": True if attendance_obj.check_in and attendance_obj.is_abnormal else False, "absenteeism_card": True, }) else: attendance_obj.sudo().write({ "sch_check_out_time": will_datetime_end, "check_out": max_time, "is_early_card": False, "is_no_card": True if attendance_obj.check_in and attendance_obj.is_no_card else False, "is_abnormal": True if attendance_obj.check_in and attendance_obj.is_abnormal else False, "absenteeism_card": True if attendance_obj.check_in and attendance_obj.absenteeism_card else False, }) return True @staticmethod def divide_time_range_fixed(start_time, end_time): time_ranges = [start_time.strftime("%Y-%m-%d")] while start_time + relativedelta(days=1) <= end_time: current_date = start_time + relativedelta(days=1) time_ranges.append(current_date.strftime("%Y-%m-%d")) start_time = current_date return time_ranges @staticmethod def str_to_time(current_data, data): """ %d:%d 的 str类型的时间转换成时间 :param current_data: 当前日期 :param data: str类型时间 :return: 转换后的时间 """ if data >= 24: data = 23.99 from_time = "%d:%d" % (int(data), round((data - int(data)) * 60)) from_times = str(current_data) + " " + from_time from_ti = datetime.strptime(from_times, "%Y-%m-%d %H:%M") return from_ti @staticmethod def utils_list_page(data, limit): """ 列表平分 :param data: list :param limit: 每组多少个 :return: 平分结果 """ for start in range(0, len(data), limit): if start + limit > len(data): end = len(data) else: end = start + limit yield data[start:end] def get_employee_work_time(self, employee, current_week, current_date, tz): """ 获取员工的当天是否应该出勤以及出勤时间 :param employee: 员工 :param current_week: 当天周 :param current_date: 当前天 :param tz: 时区 :return: 出勤时间 """ # 获取员工的工作信息 resource_calendar = employee.resource_calendar_id # 获取工作表 attendance_ids = resource_calendar.attendance_ids # 获取公休息日表 global_leave = resource_calendar.global_leave_ids # 有没有工作规则 if not attendance_ids: return False # 判断是否是调休,如果是直接返回当天上班 if self.get_leave_rest_day(global_leave, current_date): return True # 是不是工作时间 attendance_data = resource_calendar.attendance_ids.filtered( lambda s: s.dayofweek == str(current_week) and s.day_period in ["morning", "afternoon"]) if not attendance_data: return False # 获取工作时间 work_time = self.get_work_time(attendance_data, current_date, tz) if not work_time: return False go_from, go_to, off_from, off_to, current_date = work_time # 早上和下午上班 判断当天 是不是公休日 if self.get_leave_day(global_leave, go_from, tz) and self.get_leave_day(global_leave, off_from, tz): return False return go_from, go_to, off_from, off_to, current_date def get_work_time(self, attendance_data, current_date, tz): """ 当前时间的工作时间段 :param attendance_data: :param current_date: :param tz: :return: """ work_time_list = [] for work_time in attendance_data: work_time_list.append(self.time_to_utc_time(self.str_to_time(current_date, work_time.hour_from), tz)) work_time_list.append(self.time_to_utc_time(self.str_to_time(current_date, work_time.hour_to), tz)) work_time_sorted = sorted(work_time_list) go_from, go_to, off_from, off_to = work_time_sorted return go_from, go_to, off_from, off_to, current_date @staticmethod def time_to_utc_time(data, tz): """ 本地时间转换成tz时间 :param data: 时间 :param tz: 时区 :return: 转换时间 """ time_timezone = pytz.timezone(tz).localize(data, is_dst=False) time_to_utc = time_timezone.astimezone(pytz.timezone("utc")) utc_time_str = time_to_utc.strftime("%Y-%m-%d %H:%M:%S") utc_time = datetime.strptime(utc_time_str, "%Y-%m-%d %H:%M:%S") return utc_time def get_leave_rest_day(self, global_leave, current_time): """ 判断是否是公休日的调休日 :param global_leave: :param current_time: :param tz: :return: """ if not global_leave: return False for leave in global_leave: if leave.shift_replacement_day: for rest_day in str(leave.shift_replacement_day).split(","): # rest_day_month = str(rest_day).split("月")[0] rest_day_month = datetime.strptime(rest_day, "%m月%d日").strftime("%m") # rest_day_day = str(str(rest_day).split("月")[1].replace("月", "")).split("日")[0] rest_day_day = datetime.strptime(rest_day, "%m月%d日").strftime("%d") current_time_month = current_time.split('-')[1] current_time_day = current_time.split('-')[2] if int(rest_day_month) == int(current_time_month) and int(rest_day_day) == int(current_time_day): return True return False def get_leave_day(self, global_leave, current_time, tz): """ 判断是否是公休日 :param global_leave: :param current_time: :param tz: :return: """ if not global_leave: return False for leave in global_leave: if self.utc_time_to_tz(tz, leave.date_from) <= current_time.strftime("%Y-%m-%d %H:%M:%S") <= self.utc_time_to_tz(tz, leave.date_to): return True return False @staticmethod def utc_time_to_tz(tz, chin): """ utc时间的时区转换tz时间 :param tz: 时区 :param chin: 时间 :return: 转换时间 """ time_timezone = pytz.timezone('utc').localize(chin, is_dst=False) time_to_utc = time_timezone.astimezone(pytz.timezone(tz)) utc_time_str = time_to_utc.strftime("%Y-%m-%d %H:%M:%S") return utc_time_str