312 lines
12 KiB
Python
312 lines
12 KiB
Python
import logging
|
||
from datetime import datetime, timedelta
|
||
from odoo import models, fields, api
|
||
|
||
|
||
class WorkLogSetting(models.Model):
|
||
_name = 'sf.work.log.setting'
|
||
_description = '工作日历设置'
|
||
|
||
def _get_code(self):
|
||
"""
|
||
自动生成编码
|
||
:return:
|
||
"""
|
||
fixture_material = self.env['sf.work.log.setting'].sudo().search(
|
||
[('code', '!=', '')],
|
||
limit=1,
|
||
order="id desc")
|
||
if not fixture_material:
|
||
num = "%03d" % 1
|
||
else:
|
||
m = int(fixture_material.code) + 1
|
||
num = "%03d" % m
|
||
return num
|
||
|
||
code = fields.Char(string='序号', default=_get_code)
|
||
name = fields.Char(string='工作日历名称', required=True, size=15)
|
||
|
||
working_shift_ids = fields.Many2many('sf.working.shift', string='班次', required=True)
|
||
start_time = fields.Datetime(string='日开始时间', readonly=True, compute='_compute_working_shift_ids')
|
||
end_time = fields.Datetime(string='日结束时间', readonly=True, compute='_compute_working_shift_ids')
|
||
duration = fields.Char(string='时长', readonly=True, compute='_compute_working_shift_ids')
|
||
|
||
day_off_ids = fields.Many2many('sf.day.off', string='休息日', required=True)
|
||
|
||
status = fields.Selection([('正常', '正常'), ('禁用', '禁用')], string='状态', default='正常')
|
||
update_person = fields.Char(string='更新人', default=lambda self: self.env.user.name)
|
||
update_time = fields.Datetime(string='更新时间', default=lambda self: fields.Datetime.now())
|
||
|
||
setting_to_calendar_ids = fields.One2many('sf.work.schedule.calendar', 'name_id', '工作日历')
|
||
check_status = fields.Boolean(string='启用状态', default=False, readonly=True)
|
||
|
||
def action_check(self):
|
||
"""
|
||
审核启用
|
||
"""
|
||
self.check_status = True
|
||
|
||
def action_uncheck(self):
|
||
"""
|
||
审核禁用
|
||
"""
|
||
self.check_status = False
|
||
|
||
@api.depends('working_shift_ids')
|
||
def _compute_working_shift_ids(self):
|
||
"""
|
||
根据所选班次自动生成开始时间和结束时间，同时计算出工作时长
|
||
:return:
|
||
"""
|
||
for record in self:
|
||
if record:
|
||
for working_shift_id in record.working_shift_ids:
|
||
if not record.start_time:
|
||
record.start_time = working_shift_id.start_time
|
||
record.end_time = working_shift_id.end_time
|
||
else:
|
||
if (working_shift_id.start_time - record.start_time).total_seconds() < 0:
|
||
record.start_time = working_shift_id.start_time
|
||
if (working_shift_id.end_time - record.end_time).total_seconds() > 0:
|
||
record.end_time = working_shift_id.end_time
|
||
|
||
record.duration = record.end_time - record.start_time
|
||
|
||
# @api.onchange('day_off_ids')
|
||
# def _onchange_day_off_ids(self):
|
||
# # 先删除之前创建的工作日历事件记录
|
||
# self.env['sf.work.schedule.calendar'].search([('calendar_code', '=', self.code)]).unlink()
|
||
#
|
||
# # 获取当年的一月一号的日期
|
||
# year = fields.Datetime.now().year # 2023
|
||
# first_day = datetime(year, 1, 1).date() # 2023-01-01
|
||
# day_of_week = first_day.strftime("%A") # 星期日
|
||
#
|
||
# # 根据day_of_week，将其设置为起始0，循环周一到周日，按循环顺序设置为0-6
|
||
# # 列：{'星期日': 0, '星期一': 1, '星期二': 2, '星期三': 3, '星期四': 4, '星期五': 5, '星期六': 6}
|
||
# desc = {}
|
||
# desc_weekdays = ['星期一', '星期二', '星期三', '星期四', '星期五', '星期六', '星期日']
|
||
# status = False
|
||
# ints = 0
|
||
# while len(desc) < 7:
|
||
# if len(desc) != 0:
|
||
# for week in desc_weekdays:
|
||
# if len(desc) < 7:
|
||
# desc.update({week: ints})
|
||
# ints += 1
|
||
# else:
|
||
# for week in desc_weekdays:
|
||
# if status:
|
||
# desc.update({week: ints})
|
||
# ints += 1
|
||
# if day_of_week == week:
|
||
# desc.update({week: ints})
|
||
# status = True
|
||
# ints += 1
|
||
#
|
||
# # 创建工作日历的休息日事件
|
||
# day_off_ids = self.day_off_id.mapped('name')
|
||
# record = self.env['sf.work.log.setting'].search([('code', '=', self.code)])
|
||
# # record.setting_to_calendar_ids.unlink()
|
||
# # self.env.cr.commit()
|
||
# for week in day_off_ids:
|
||
# num = desc[week]
|
||
# while num <= 30:
|
||
# target_date = first_day + timedelta(days=num)
|
||
# num += 7
|
||
# self.env['sf.work.schedule.calendar'].create({'name': '休息日',
|
||
# 'name_id': record.id,
|
||
# 'calendar_code': self.code,
|
||
# 'date_time': target_date})
|
||
|
||
@api.model_create_multi
|
||
def create(self, vals_list):
|
||
"""
|
||
创建记录时，生成工作日历
|
||
"""
|
||
record = super(WorkLogSetting, self).create(vals_list)
|
||
record.generate_work_calendar()
|
||
return record
|
||
|
||
def write(self, vals):
|
||
"""
|
||
更新记录时，生成工作日历
|
||
"""
|
||
# 先删除之前创建的工作日历事件记录
|
||
self.env['sf.work.schedule.calendar'].search([('calendar_code', '=', self.code)]).unlink()
|
||
result = super(WorkLogSetting, self).write(vals)
|
||
self.generate_work_calendar()
|
||
return result
|
||
|
||
def generate_work_calendar(self):
|
||
"""
|
||
生成工作日历
|
||
"""
|
||
self.ensure_one()
|
||
# start_date = date.today() # 开始日期
|
||
# end_date = start_date + timedelta(days=365) # 结束日期
|
||
# 获取本年第一天和最后一天
|
||
start_date = datetime.now().replace(month=1, day=1).date()
|
||
end_date = datetime.now().replace(month=12, day=31).date()
|
||
print(self.day_off_ids.mapped('name'))
|
||
# 休息日列表
|
||
rest_days = self.chinese_weekdays_to_english(self.day_off_ids.mapped('name'))
|
||
print(rest_days)
|
||
for single_date in self.daterange(start_date, end_date):
|
||
is_workday = self.chinese_weekday_to_english(single_date.strftime("%A"))
|
||
logging.info(f"每天的星期：{is_workday}")
|
||
if is_workday in rest_days:
|
||
print('is_workday in rest_days', is_workday)
|
||
self.env['sf.work.schedule.calendar'].sudo().create({
|
||
'name': '休息日',
|
||
'name_id': self.id,
|
||
'date_time': single_date})
|
||
|
||
@staticmethod
|
||
def chinese_weekdays_to_english(chinese_weekdays):
|
||
"""
|
||
将元组的中文星期转换成英文
|
||
:param chinese_weekdays:
|
||
:return:
|
||
"""
|
||
weekdays = {
|
||
'星期一': 'Monday', 'Monday': 'Monday',
|
||
'星期二': 'Tuesday', 'Tuesday': 'Tuesday',
|
||
'星期三': 'Wednesday', 'Wednesday': 'Wednesday',
|
||
'星期四': 'Thursday', 'Thursday': 'Thursday',
|
||
'星期五': 'Friday', 'Friday': 'Friday',
|
||
'星期六': 'Saturday', 'Saturday': 'Saturday',
|
||
'星期日': 'Sunday', 'Sunday': 'Sunday',
|
||
}
|
||
english_weekdays = []
|
||
for chinese_weekday in chinese_weekdays:
|
||
english_weekday = weekdays.get(chinese_weekday, 'Invalid weekday')
|
||
english_weekdays.append(english_weekday)
|
||
return english_weekdays
|
||
|
||
@staticmethod
|
||
def chinese_weekday_to_english(chinese_weekday):
|
||
"""
|
||
将单个的中文星期转换成英文
|
||
:param chinese_weekday:
|
||
:return:
|
||
"""
|
||
weekdays = {
|
||
'星期一': 'Monday', 'Monday': 'Monday',
|
||
'星期二': 'Tuesday', 'Tuesday': 'Tuesday',
|
||
'星期三': 'Wednesday', 'Wednesday': 'Wednesday',
|
||
'星期四': 'Thursday', 'Thursday': 'Thursday',
|
||
'星期五': 'Friday', 'Friday': 'Friday',
|
||
'星期六': 'Saturday', 'Saturday': 'Saturday',
|
||
'星期日': 'Sunday', 'Sunday': 'Sunday',
|
||
}
|
||
weekday = weekdays.get(chinese_weekday)
|
||
if weekday:
|
||
return weekday
|
||
return chinese_weekday
|
||
|
||
@staticmethod
|
||
def daterange(start_date, end_date):
|
||
"""
|
||
生成日期范围
|
||
"""
|
||
# 生成日期范围
|
||
for n in range(int((end_date - start_date).days)):
|
||
yield start_date + timedelta(n)
|
||
|
||
def open_work_schedule_calendar(self):
|
||
action = self.env.ref('sf_plan_management.sf_work_schedule_calendar_act')
|
||
result = action.read()[0]
|
||
result['domain'] = [('name_id', '=', self.id)]
|
||
return result
|
||
|
||
|
||
class WorkingShift(models.Model):
|
||
_name = 'sf.working.shift'
|
||
_description = '班次'
|
||
|
||
def _get_code(self):
|
||
"""
|
||
自动生成编码
|
||
:return:
|
||
"""
|
||
fixture_material = self.env['sf.working.shift'].sudo().search(
|
||
[('code', '!=', '')],
|
||
limit=1,
|
||
order="id desc")
|
||
if not fixture_material:
|
||
num = "%03d" % 1
|
||
else:
|
||
m = int(fixture_material.code) + 1
|
||
num = "%03d" % m
|
||
return num
|
||
|
||
code = fields.Char('编码', default=_get_code, readonly=True)
|
||
name = fields.Char('名称', required=True)
|
||
start_time = fields.Datetime('班次开始时间')
|
||
end_time = fields.Datetime('班次结束时间')
|
||
remark = fields.Char('备注')
|
||
check_status = fields.Boolean(string='启用状态', default=False, readonly=True)
|
||
|
||
def action_check(self):
|
||
"""
|
||
审核启用
|
||
"""
|
||
self.check_status = True
|
||
|
||
def action_uncheck(self):
|
||
"""
|
||
审核禁用
|
||
"""
|
||
self.check_status = False
|
||
|
||
|
||
class DayOff(models.Model):
|
||
_name = 'sf.day.off'
|
||
_description = '休息日'
|
||
|
||
name = fields.Selection([
|
||
('Monday ', '星期一'),
|
||
('Tuesday ', '星期二'),
|
||
('Wednesday ', '星期三'),
|
||
('Thursday ', '星期四'),
|
||
('Friday ', '星期五'),
|
||
('Saturday ', '星期六'),
|
||
('Sunday ', '星期日')], '休息日名称')
|
||
check_status = fields.Boolean(string='启用状态', default=False, readonly=True)
|
||
|
||
def action_check(self):
|
||
"""
|
||
审核启用
|
||
"""
|
||
self.check_status = True
|
||
|
||
def action_uncheck(self):
|
||
"""
|
||
审核禁用
|
||
"""
|
||
self.check_status = False
|
||
|
||
|
||
class WorkScheduleCalendar(models.Model):
|
||
_name = 'sf.work.schedule.calendar'
|
||
_description = '工作日历'
|
||
|
||
name = fields.Selection([('休息日', '休息日'), ('计划停机', '计划停机'), ('工作日', '工作日')], '日历事件名称')
|
||
date_time = fields.Date('休息时间')
|
||
name_id = fields.Many2one('sf.work.log.setting', '工作日历名称')
|
||
calendar_code = fields.Char('工作日历编码', readonly=True, compute='_compute_calendar_code')
|
||
|
||
day_off_id = fields.Many2many('sf.day.off', string='休息日')
|
||
scheduled_outage = fields.Char('计划停机')
|
||
monthly_rest_days = fields.Char('月休息天数', readonly=True)
|
||
annual_rest_days = fields.Char('年休息天数', readonly=True)
|
||
monthly_planned_downtime = fields.Char('月计划停机时长', readonly=True)
|
||
annual_planned_downtime = fields.Char('年计划停机时长', readonly=True)
|
||
|
||
@api.depends('name_id')
|
||
def _compute_calendar_code(self):
|
||
for record in self:
|
||
if record:
|
||
record.calendar_code = record.name_id.code
|