import base64 import json from datetime import timedelta from odoo import http, fields, Command from odoo.http import request from odoo.osv import expression import logging _logger = logging.getLogger(__name__) class YuthonWillTask(http.Controller): @http.route('/yuthon/yuthon/will/task', methods=['POST'], type='http', auth='none', csrf=False) def index2(self, **kwargs): """根据ID获取日程详情""" id = kwargs.get('id') data = request.env['yuthon.will.task'].sudo().search(domain=[("id", "=", int(id))]) request.env['yuthon.will.task'].create({ 'fields': kwargs.get('id'), }) file_list = [] pdf = ['application/pdf'] for file in data.document_ids: base_url = request.env['ir.config_parameter'].sudo().get_param('web.base.url') file_url = f"{base_url}/web/content/preview/{file.id}" if file.mimetype in pdf: file_url = f"{base_url}/web/content/preview/{file.id}" file_list.append({ 'file_url': file_url, 'file_name': file.name, }) employee_ids = [] for em in data.employee_ids: employee_ids.append({ 'id': em.id, 'name': em.name, 'image_1920': em.image_128.decode('utf-8') if em.image_128 else '', }) result = { 'id': data.id, 'name': data.name, 'code': data.code, 'start_date': fields.Date.to_string(data.start_date) if data.start_date else '', 'request_hour_from': data.request_hour_from, 'request_hour_to': data.request_hour_to, 'priority': data.priority, 'user_id': data.user_id.name, 'user_avatar': base64.b64decode(data.user_id.avatar_128).decode('utf-8') if data.user_id.avatar_128 else '', 'remarks': data.remarks, 'address': data.address, 'employee_ids': employee_ids, 'file_list': file_list, } # 转换 request_hour_from 和 request_hour_to 的显示值 request_hour_from_selection = request.env['yuthon.will.task'].sudo().fields_get(allfields=['request_hour_from'])['request_hour_from']['selection'] request_hour_to_selection = request.env['yuthon.will.task'].sudo().fields_get(allfields=['request_hour_to'])['request_hour_to']['selection'] for f in request_hour_from_selection: if f[0] == data.request_hour_from: result['request_hour_from'] = f[1] for f in request_hour_to_selection: if f[0] == data.request_hour_to: result['request_hour_to'] = f[1] # 计算 start_date 星期几 if data.start_date: result['start_date_week'] = data.start_date.strftime('%A') return json.dumps({ 'data': result, 'code': 200, 'message': 'success' }) @http.route('/will/task/list', methods=['POST'], type='http', auth='public', csrf=False) def will_task_list(self, **kwargs): """获取日程列表（手机端接口）""" start = int(kwargs.get('start')) users_id = int(kwargs.get('users_id')) _filter = kwargs.get('filter') state = kwargs.get('state') order = 'task_datetime desc' if state == 'done' else 'task_datetime asc' domain = [('users_ids', 'in', [users_id])] today = fields.Date.today() if _filter == 'today': domain = expression.AND([domain, [('start_date', '=', today)]]) elif _filter == 'week': start_date = today - timedelta(days=today.weekday()) end_date = start_date + timedelta(days=6) domain = expression.AND([domain, [('start_date', '>=', start_date), ('start_date', '<=', end_date)]]) domain = expression.AND([domain, [('task_state', '=', state)]]) WillTask = request.env['yuthon.will.task'].sudo() request_hour_selection = WillTask.fields_get(allfields=['request_hour_from'])['request_hour_from']['selection'] state_dict = dict(WillTask.fields_get(allfields=['task_state'])['task_state']['selection']) result_list = [] task_ids = request.env['yuthon.will.task'].sudo().search(domain, offset=start, limit=20, order=order) user_id = request.env['res.users'].sudo().search([('id', '=', users_id)]) company_id = user_id.company_id for task in task_ids: if company_id.id == task.company_id.id: weekday_str = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"][task.start_date.weekday()] result = { 'id': task.id, 'name': task.name, 'employee_id': task.employee_id.name or '', 'employee_ids': ', '.join(task.employee_ids.mapped('name')) if task.employee_ids else '', 'employee_num': ','.join(task.employee_ids.mapped('name')) if task.employee_ids else '', 'start_date': fields.Date.to_string(task.start_date) if task.start_date else '', 'weekday': weekday_str, 'address': task.address or '', 'request_hour_from': '', 'request_hour_to': '', 'state': task.task_state, 'display_state': state_dict.get(task.task_state), 'task_type': task.task_type } hour_dict = dict(request_hour_selection) result['request_hour_from'] = hour_dict.get(task.request_hour_from) result['request_hour_to'] = hour_dict.get(task.request_hour_to) result_list.append(result) return json.dumps({ 'data': result_list, 'code': 200, 'message': 'success' }) @http.route('/will/task/record', methods=['POST'], type='http', auth='public', csrf=False) def will_task_record(self, **kwargs): """获取单个日程记录详情""" task_id = kwargs.get('id') task = request.env['yuthon.will.task'].sudo().search([('id', '=', task_id)]) request_hour_selection = request.env['yuthon.will.task'].sudo().fields_get(allfields=['request_hour_from'])['request_hour_from']['selection'] file_list = [] pdf = ['application/pdf'] for file in task.document_ids: base_url = request.env['ir.config_parameter'].sudo().get_param('web.base.url') file_url = f"{base_url}/web/content/preview/{file.id}" if file.mimetype in pdf: file_url = f"{base_url}/web/content/preview/{file.id}" file_list.append({ 'file_url': file_url, 'file_name': file.name, }) weekday_str = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"][task.start_date.weekday()] result = { 'id': task.id, 'name': task.name, 'employee_id': task.employee_id.name, 'employee_ids': ','.join(task.employee_ids.mapped('name')), 'task_time_ids': ','.join(task.task_time_ids.mapped('name')), 'employee': task.employee_ids.read(['name']), 'task_time': task.task_time_ids.read(['name']), 'employee_num': len(task.employee_ids), 'start_date': fields.Date.to_string(task.start_date) if task.start_date else '', 'weekday': weekday_str, 'request_hour_from': task.request_hour_from, 'request_hour_to': task.request_hour_to, 'remarks': task.remarks, 'address': task.address, 'file_list': file_list, 'priority': task.priority, } hour_dict = dict(request_hour_selection) result['request_hour_from'] = hour_dict.get(task.request_hour_from) result['request_hour_to'] = hour_dict.get(task.request_hour_to) return json.dumps({ 'data': result, 'code': 200, 'message': 'success' }) @http.route('/will/task/create', methods=['POST'], type='http', auth='public', csrf=False) def will_task_create(self, **kwargs): """手机端创建日程接口""" request.env['yuthon.will.task'].sudo().create({ 'name': kwargs.get('name'), 'employee_id': kwargs.get('id'), 'employee_ids': kwargs.get('id'), 'start_date': kwargs.get('id'), 'request_hour_from': '', 'request_hour_to': '', 'remarks': kwargs.get('remarks'), 'address': kwargs.get('address'), }) return json.dumps({ 'code': 200, 'message': '数据创建成功' }) @http.route('/phone/employee/self', type='http', auth="none", csrf=False, cors='*') def phone_employee_self(self, **kw): """搜索员工（手机端接口）""" value = kw.get('value', False) domain = [] if value: domain.append(('name', 'ilike', value)) records = request.env['hr.employee'].sudo().search(domain, limit=15) data = records.read(['name']) return json.dumps({'data': data, 'code': 200, 'message': ''}) @http.route('/phone/task/time', type='http', auth="none", csrf=False, cors='*') def phone_task_time(self, **kw): """搜索提醒时间配置（手机端接口）""" value = kw.get('value', False) domain = [] if value: domain.append(('name', 'ilike', value)) records = request.env['yuthon.task.time'].sudo().search(domain, limit=15) data = records.read(['name']) return json.dumps({'data': data, 'code': 200, 'message': ''}) @http.route('/phone/task/create', type='http', auth="none", csrf=False, cors='*') def phone_task_create(self, **kw): """手机端创建/更新日程接口 record_id: 记录id（根据这个判断是否创建还是更新，创建的时候不传，更新的时候传） request_hour_from: 开始时间 request_hour_to: 结束时间 priority: 优先级 name: 任务内容 remarks: 备注 employee_ids: 关联员工 task_time_ids: 提醒 address: 地址 start_date: 任务日期 """ def str2list(value: str) -> list: if not value: return [] if not isinstance(value, str): raise ValueError('value must be str') str_list = value.split(',') return [int(s) for s in str_list] employee_ids = str2list(kw.get('employee_ids')) task_time_ids = str2list(kw.get('task_time_ids')) if not isinstance(task_time_ids, list): return json.dumps({'data': '', 'code': 201, 'message': '提醒参数错误'}) if not isinstance(employee_ids, list): return json.dumps({'data': '', 'code': 201, 'message': '关联员工参数错误'}) data = { 'request_hour_from': kw.get('request_hour_from', False), 'request_hour_to': kw.get('request_hour_to', False), 'priority': kw.get('priority', False), 'name': kw.get('name', False), 'remarks': kw.get('remarks', False), 'employee_ids': [(6, 0, employee_ids)], 'task_time_ids': [(6, 0, task_time_ids)], 'address': kw.get('address', False), 'start_date': kw.get('start_date', False) } Task = request.env['yuthon.will.task'].sudo() record_id = kw.get('record_id', False) if record_id: # 更新日程 record = Task.browse(int(record_id)) record.write(data) else: # 创建日程 users_id = int(kw.get('users_id')) user = request.env['res.users'].sudo().browse(int(users_id)) record = Task.with_user(user).sudo().create(data) return json.dumps({'data': record.id, 'code': 200, 'message': ''}) @http.route('/phone/task/unlink', type='http', auth="none", csrf=False, cors='*') def phone_task_unlink(self, **kw): """手机端删除日程接口""" record_id = kw.get('record_id', False) if record_id: record = request.env['yuthon.will.task'].sudo().browse(int(record_id)) record.sudo().unlink() else: return json.dumps({'data': '', 'code': 201, 'message': '记录不存在'}) return json.dumps({'data': 'success', 'code': 200, 'message': ''})