450 lines
20 KiB
Python
450 lines
20 KiB
Python
import logging
|
||
import json
|
||
from datetime import datetime, timedelta
|
||
from urllib.parse import quote
|
||
from odoo import models, fields, api
|
||
import requests
|
||
|
||
_logger = logging.getLogger(__name__)
|
||
|
||
|
||
class WeComApps(models.Model):
|
||
_name = "wecom.apps"
|
||
|
||
name = fields.Char(string="应用名称")
|
||
corpid = fields.Char(string="corpid")
|
||
agent_id = fields.Char(string="agent_id")
|
||
corpsecret = fields.Char(string="corpsecret")
|
||
company_id = fields.Many2one("res.company", string="公司", default=lambda self: self.env.company)
|
||
access_token = fields.Char(string="access_token", default="1")
|
||
token_expiration_time = fields.Datetime(string="Token失效时间")
|
||
|
||
wx_work_image_path = fields.Char(string="企业微信图片保存地址")
|
||
|
||
def sync_access_token(self):
|
||
"""
|
||
获取access_token
|
||
:return: access_token
|
||
"""
|
||
data = {
|
||
"corpid": self.corpid,
|
||
"corpsecret": self.corpsecret
|
||
}
|
||
access_token = self.sudo().env['wecom.api'].function_get_access_token(data)
|
||
self.access_token = access_token
|
||
return access_token
|
||
|
||
def sync_function_api(self, apps_type, api_function, data, category="default"):
|
||
response = self.sync_function_api_base(apps_type, api_function, data, category)
|
||
result_data = json.loads(response.content)
|
||
return result_data
|
||
|
||
def sync_function_api_base(self, apps_type, api_function, data, category="default"):
|
||
wecom_app = self.sudo().env["wecom.apps.config"].search([("wecom_apps_type", "=", apps_type), ("wecom_apps_category", "=", category)], limit=1).wecom_apps_id
|
||
response = self.sudo().env['wecom.api'].function_http_request(api_function, data, wecom_app.access_token)
|
||
result_data = json.loads(response.content)
|
||
if result_data.get("errcode", 0) in [40014, 42001]:
|
||
wecom_app.sync_access_token()
|
||
wecom_app = self.sudo().env["wecom.apps.config"].search([("wecom_apps_type", "=", apps_type), ("wecom_apps_category", "=", category)],
|
||
limit=1).wecom_apps_id
|
||
response = self.sudo().env['wecom.api'].function_http_request(api_function, data, wecom_app.access_token)
|
||
return response
|
||
|
||
def sync_department_list(self):
|
||
wecom_departments = []
|
||
result_data = self.sync_function_api("contacts", "DEPARTMENT_LIST", {})
|
||
if result_data["errcode"] == 0:
|
||
wecom_departments = result_data.get("department", [])
|
||
return True, wecom_departments
|
||
return False, result_data
|
||
|
||
def sync_checkin_data(self, data):
|
||
response = self.sync_function_api_base("attendance", "GET_CHECKIN_DATA", data)
|
||
result_data = json.loads(response.content)
|
||
if result_data["errcode"] == 0:
|
||
return True, response
|
||
return False, response
|
||
|
||
def sync_media_get(self, data):
|
||
response = self.sync_function_api_base("attendance", "MEDIA_GET", data)
|
||
result_data = json.loads(response.content)
|
||
if result_data["errcode"] == 0:
|
||
return True, response
|
||
return False, response
|
||
|
||
def sync_user_list(self, department_id):
|
||
user_list = []
|
||
data = {
|
||
"department_id": department_id
|
||
}
|
||
result_data = self.sync_function_api("contacts", "USER_LIST", data)
|
||
if result_data["errcode"] == 0:
|
||
user_list = result_data.get("userlist", [])
|
||
return True, user_list
|
||
return False, result_data
|
||
|
||
def sync_send_message(self, category='default', user_id=None, message=None):
|
||
wecom_app = self.sudo().env["wecom.apps.config"].search([("wecom_apps_type", "=", "message_send"), ("wecom_apps_category", "=", category)], limit=1).wecom_apps_id
|
||
data = {
|
||
'touser': user_id,
|
||
'msgtype': 'text',
|
||
'agentid': wecom_app.agent_id,
|
||
'text': {
|
||
'content': message
|
||
}
|
||
}
|
||
result_data = self.sync_function_api("message_send", "MESSAGE_SEND", data, category)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data
|
||
|
||
def back_message(self, msg_id='', category="default"):
|
||
token = ''
|
||
if category == 'task':
|
||
token = self.env['wecom.apps'].search([('name', '=', '日程消息通知')])['access_token']
|
||
if category == 'notice':
|
||
token = self.env['wecom.apps'].search([('name', '=', '通知公告消息通知')])['access_token']
|
||
if category == 'approval':
|
||
token = self.env['wecom.apps'].search([('name', '=', '审批消息通知')])['access_token']
|
||
data = {'msgid': msg_id}
|
||
url = f'https://qyapi.weixin.qq.com/cgi-bin/message/recall?access_token={token}'
|
||
_logger.info(f"------------------------获取msg_id：{msg_id}tokenID:{token}当前url{url}-------------------------------")
|
||
requests.post(url=url, data=json.dumps(data))
|
||
|
||
def sync_send_message_textcard(self, content_item='', touser='', category='default', user_id=None, title="", description="", url="", btntxt="详情", exception_alerts = False):
|
||
"""
|
||
touser 否 成员ID列表（消息接收者，多个接收者用‘|’分隔，最多支持1000个）。特殊情况：指定为@all，则向关注该企业应用的全部成员发送
|
||
toparty 否 部门ID列表，多个接收者用‘|’分隔，最多支持100个。当touser为@all时忽略本参数
|
||
totag 否 标签ID列表，多个接收者用‘|’分隔，最多支持100个。当touser为@all时忽略本参数
|
||
msgtype 是 消息类型，此时固定为：textcard
|
||
agentid 是 企业应用的id，整型。企业内部开发，可在应用的设置页面查看；第三方服务商，可通过接口 获取企业授权信息 获取该参数值
|
||
title 是 标题，不超过128个字符，超过会自动截断（支持id转译）
|
||
description 是 描述，不超过512个字符，超过会自动截断（支持id转译）
|
||
url 是 点击后跳转的链接。最长2048字节，请确保包含了协议头(http/https)
|
||
btntxt 否 按钮文字。 默认为“详情”， 不超过4个文字，超过自动截断。
|
||
enable_id_trans 否 表示是否开启id转译，0表示否，1表示是，默认0
|
||
enable_duplicate_check 否 表示是否开启重复消息检查，0表示否，1表示是，默认0
|
||
duplicate_check_interval 否 表示是否重复消息检查的时间间隔，默认1800s，最大不超过4小时
|
||
{
|
||
"touser" : "UserID1|UserID2|UserID3",
|
||
"toparty" : "PartyID1 | PartyID2",
|
||
"totag" : "TagID1 | TagID2",
|
||
"msgtype" : "textcard",
|
||
"agentid" : 1,
|
||
"textcard" : {
|
||
"title" : "领奖通知",
|
||
"description" : "<div class=\"gray\">2016年9月26日</div> <div class=\"normal\">恭喜你抽中iPhone 7一台，领奖码：xxxx</div><div class=\"highlight\">请于2016年10月10日前联系行政同事领取</div>",
|
||
"url" : "URL",
|
||
"btntxt":"更多"
|
||
},
|
||
"enable_id_trans": 0,
|
||
"enable_duplicate_check": 0,
|
||
"duplicate_check_interval": 1800
|
||
}
|
||
:param category:
|
||
:param user_id:
|
||
:param title:
|
||
:param description:
|
||
:param url:
|
||
:param btntxt:
|
||
:return:
|
||
"""
|
||
wecom_app = self.sudo().env["wecom.apps.config"].search([("wecom_apps_type", "=", "message_send"), ("wecom_apps_category", "=", category)], limit=1).wecom_apps_id
|
||
if category == 'oa':
|
||
data = {
|
||
'touser': user_id,
|
||
'msgtype': 'textcard',
|
||
'agentid': wecom_app.agent_id,
|
||
"textcard": {
|
||
"title": title,
|
||
"description": description,
|
||
"url": url,
|
||
"btntxt": btntxt
|
||
},
|
||
}
|
||
elif category in ['task', 'att_ok', 'salary', 'can']:
|
||
data = {
|
||
'touser': user_id,
|
||
'msgtype': 'textcard',
|
||
'agentid': wecom_app.agent_id,
|
||
"textcard": {
|
||
"title": title,
|
||
"description": description,
|
||
"url": url,
|
||
"btntxt": btntxt
|
||
},
|
||
}
|
||
elif category == 'att':
|
||
if exception_alerts == True:
|
||
data = {
|
||
'touser': user_id,
|
||
'msgtype': 'textcard',
|
||
'agentid': wecom_app.agent_id,
|
||
"textcard": {
|
||
"title": title,
|
||
"description": description,
|
||
"url": url,
|
||
"btntxt": btntxt
|
||
},
|
||
}
|
||
else:
|
||
data = {
|
||
"touser": user_id,
|
||
"msgtype": "miniprogram_notice",
|
||
"miniprogram_notice": {
|
||
"appid": "wxda530cb253dec90d",
|
||
"page": "pages/clock_in/clock_in",
|
||
"title": "打卡提醒",
|
||
"description": description,
|
||
"content_item": content_item
|
||
},
|
||
}
|
||
else:
|
||
data = {
|
||
'touser': touser,
|
||
'msgtype': 'textcard',
|
||
'agentid': wecom_app.agent_id,
|
||
"textcard": {
|
||
"title": title,
|
||
"description": description,
|
||
"url": url,
|
||
"btntxt": btntxt
|
||
},
|
||
}
|
||
service_address = self.env['ir.config_parameter'].sudo().get_param('web.base.url')
|
||
_logger.info(f"系统当前地址{service_address}-------------------------------")
|
||
if 'oa' in str(service_address):
|
||
_logger.info(f"----------地址识别成功，已经发送消息通知---------------------")
|
||
_logger.info(f"----------地址识别成功，已经发送消息通知---------------------")
|
||
result_data = self.sync_function_api("message_send", "MESSAGE_SEND", data, category)
|
||
employee_id = self.sudo().env["res.users"].search([("login", "=", user_id)], limit=1).employee_ids
|
||
notice_type = 'attendance' if '记得打卡' in description else 'process'
|
||
self.env["attendance.message.log"].create({
|
||
"employee_id": employee_id.id,
|
||
"content": description,
|
||
"notice_type": notice_type,
|
||
"service_address":service_address,
|
||
'update_date': fields.Datetime.now(),
|
||
"message_url": url,
|
||
})
|
||
_logger.info(f"------------------------接口收到响应返回值{result_data}-------------------------------")
|
||
_logger.info(f"------------------------接口收到响应返回值{service_address}-------------------------------")
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data
|
||
|
||
def sync_user_update(self, user_id=None, name=None, mobile=None, biz_mail=None, department=None):
|
||
data = {}
|
||
if user_id:
|
||
data.update({'userid': user_id})
|
||
if name:
|
||
data.update({'name': name})
|
||
if mobile:
|
||
data.update({'mobile': mobile})
|
||
if biz_mail:
|
||
data.update({'biz_mail': biz_mail})
|
||
if department:
|
||
data.update({'department': [department]})
|
||
result_data = self.sync_function_api("user_create", "USER_UPDATE", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data.get("errmsg", "")
|
||
if result_data.get("errcode", None) in [60111]:
|
||
"""
|
||
{'errcode': 60111, 'errmsg': 'invalid string value `test111`. userid not found, hint: [1720622364419201522972678], from ip: 82.156.218.102, more info at https://open.work.weixin.qq.com/devtool/query?e=60111'}
|
||
"""
|
||
result_data = self.sync_function_api("user_create", "USER_CREATE", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data.get("errmsg", "")
|
||
return False, result_data.get("errmsg", "")
|
||
|
||
def sync_department_update(self, wecom_department_id=None, name=None, name_en=None, parentid=None):
|
||
data = {}
|
||
if wecom_department_id:
|
||
data.update({'id': wecom_department_id})
|
||
if name:
|
||
data.update({'name': name})
|
||
if name_en:
|
||
data.update({'name_en': name_en})
|
||
if parentid:
|
||
data.update({'parentid': parentid})
|
||
if wecom_department_id:
|
||
result_data = self.sync_function_api("user_create", "DEPARTMENT_UPDATE", data)
|
||
else:
|
||
result_data = self.sync_function_api("user_create", "DEPARTMENT_CREATE", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data.get("errmsg", "")
|
||
|
||
def sync_meeting_create(self, admin_userid, title, meeting_start, meeting_duration, *args, **kwargs):
|
||
data = {
|
||
"admin_userid": admin_userid,
|
||
"title": title,
|
||
"meeting_start": meeting_start,
|
||
"meeting_duration": meeting_duration,
|
||
}
|
||
data.update(kwargs)
|
||
result_data = self.sync_function_api("meeting", "MEETING_CREATE", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data.get("errmsg", "")
|
||
|
||
def sync_meeting_update(self, meetingid, *args, **kwargs):
|
||
data = {
|
||
"admin_userid": meetingid,
|
||
}
|
||
data.update(kwargs)
|
||
result_data = self.sync_function_api("meeting", "MEETING_UPDATE", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data.get("errmsg", "")
|
||
|
||
def sync_meeting_cancel(self, meetingid, sub_meetingid=None):
|
||
data = {
|
||
"meetingid": meetingid,
|
||
}
|
||
if sub_meetingid:
|
||
data.update({'sub_meetingid': sub_meetingid})
|
||
result_data = self.sync_function_api("meeting", "MEETING_CANCEL", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data.get("errmsg", "")
|
||
|
||
def sync_meeting_get_info(self, meetingid=None, meeting_code=None):
|
||
data = {}
|
||
if meetingid:
|
||
data.update({'meetingid': meetingid})
|
||
if meeting_code:
|
||
data.update({'meeting_code': meeting_code})
|
||
result_data = self.sync_function_api("meeting", "MEETING_GET_INFO", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data.get("errmsg", "")
|
||
|
||
def sync_get_realtime_attendee_list(self, meetingid, sub_meetingid=None, cursor=None, limit=None):
|
||
data = {
|
||
"meetingid": meetingid
|
||
}
|
||
if sub_meetingid:
|
||
data.update({'sub_meetingid': sub_meetingid})
|
||
if cursor:
|
||
data.update({'cursor': cursor})
|
||
if limit:
|
||
data.update({'limit': limit})
|
||
result_data = self.sync_function_api("meeting", "MEETING_GET_REALTIME_ATTENDEE_LIST", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data.get("errmsg", "")
|
||
|
||
def sync_get_user_meetingid(self, userid, cursor=None, limit=None, begin_time=None, end_time=None):
|
||
data = {
|
||
"userid": userid
|
||
}
|
||
if cursor:
|
||
data.update({'cursor': cursor})
|
||
if limit:
|
||
data.update({'limit': limit})
|
||
if begin_time:
|
||
data.update({'begin_time': begin_time})
|
||
if end_time:
|
||
data.update({'end_time': end_time})
|
||
result_data = self.sync_function_api("meeting", "MEETING_GET_USER_MEETINGID", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data.get("errmsg", "")
|
||
|
||
def sync_get_attendee_list(self, meetingid, *args, **kwargs):
|
||
data = {
|
||
"meetingid": meetingid
|
||
}
|
||
data.update(kwargs)
|
||
result_data = self.sync_function_api("meeting", "GET_ATTENDEE_LIST", data)
|
||
if result_data.get("errcode", None) == 0:
|
||
return True, result_data
|
||
return False, result_data.get("errmsg", "")
|
||
|
||
# @api.model
|
||
# def generate_oauth_login_url(self, target_url, category='default'):
|
||
# """
|
||
# 生成企业微信OAuth登录链接，用于在企业微信消息中发送
|
||
#
|
||
# 参数:
|
||
# target_url: 登录成功后要跳转的目标URL（完整URL或相对路径）
|
||
#
|
||
# 返回:
|
||
# OAuth登录链接URL
|
||
# """
|
||
# base_url = self.env['ir.config_parameter'].sudo().get_param('web.base.url')
|
||
# # 如果target_url不是完整URL，补全为完整URL
|
||
# if not target_url.startswith('http'):
|
||
# if target_url.startswith('/'):
|
||
# target_url = f"{base_url}{target_url}"
|
||
# else:
|
||
# target_url = f"{base_url}/{target_url}"
|
||
# # 编码目标URL作为state参数
|
||
# encoded_target_url = quote(target_url, safe='')
|
||
# # 编码回调URI
|
||
# redirect_uri = quote(f'{base_url}/wechat/login_by_code', safe='')
|
||
#
|
||
# # 企业微信OAuth参数
|
||
# if category == 'oa':
|
||
# agent_id = self.env['wecom.apps'].search([('name', '=', '审批消息通知')])['agent_id']
|
||
# appid = self.env['wecom.apps'].search([('name', '=', '审批消息通知')])['corpid']
|
||
# else:
|
||
# appid = 'wwc955c117b738b052'
|
||
# agent_id = '1000120'
|
||
#
|
||
# # 生成OAuth登录链接
|
||
# oauth_url = (
|
||
# f'https://open.weixin.qq.com/connect/oauth2/authorize?'
|
||
# f'appid={appid}&'
|
||
# f'redirect_uri={redirect_uri}&'
|
||
# f'response_type=code&'
|
||
# f'scope=snsapi_base&'
|
||
# f'state={encoded_target_url}&'
|
||
# f'agentid={agent_id}#wechat_redirect'
|
||
# )
|
||
#
|
||
# return oauth_url
|
||
|
||
@api.model
|
||
def generate_oauth_login_url(self, target_url, user_id=None):
|
||
"""
|
||
|
||
user_id = record.employee_id.user_id.id if record.employee_id.user_id else None
|
||
url = self.sudo().env["wecom.apps"].generate_oauth_login_url(target_url, user_id=user_id)
|
||
|
||
生成企业微信免登录链接，用于在企业微信消息中发送
|
||
与 /wechat/login_by_user_id 接口联动
|
||
|
||
参数:
|
||
target_url: 登录成功后要跳转的目标URL（完整URL或相对路径）
|
||
user_id: 用户ID (res.users的id)，必需参数
|
||
|
||
返回:
|
||
免登录链接URL，格式: /wechat/login_by_user_id?user_id=xxx&redirect_url=xxx
|
||
"""
|
||
if not user_id:
|
||
_logger.warning("generate_oauth_login_url: user_id 参数未提供，无法生成免登录链接")
|
||
return target_url # 如果未提供user_id，返回原始URL
|
||
|
||
base_url = self.env['ir.config_parameter'].sudo().get_param('web.base.url')
|
||
|
||
# 如果target_url不是完整URL，补全为完整URL
|
||
if not target_url.startswith('http'):
|
||
if target_url.startswith('/'):
|
||
target_url = f"{base_url}{target_url}"
|
||
else:
|
||
target_url = f"{base_url}/{target_url}"
|
||
|
||
# 编码目标URL作为redirect_url参数
|
||
encoded_target_url = quote(target_url, safe='')
|
||
|
||
# 生成免登录链接，使用 /wechat/login_by_user_id 接口
|
||
login_url = (
|
||
f'{base_url}/wechat/login_by_user_id?'
|
||
f'user_id={user_id}&'
|
||
f'redirect_url={encoded_target_url}'
|
||
)
|
||
|
||
return login_url
|