GithubStar/weworkapi_python
1
0
Fork 0
mirror of https://github.com/sbzhu/weworkapi_python.git synced 2026-10-29 14:19:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
abelzhu
2018-02-24 22:20:03 +08:00
committed by GitHub
parent d95943e99a
commit 987cc4a5aa
7 changed files with 477 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/examples/ServiceCorpTest.py
+49
View File
@@ -0,0 +1,49 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
##
# Copyright (C) 2018 All rights reserved.
#
# @File ServiceCorpTest.py
# @Brief
# @Author abelzhu, abelzhu@tencent.com
# @Version 1.0
# @Date 2018-02-24
#
#
import sys
sys.path.append("../src/")
from ServiceCorpApi import *
from TestConf import *
api = ServiceCorpApi(
"SUITE_ID",
"SUITE_SECRET",
"SUITE_TICKET"
);
try :
pre_auth_code = api.httpCall(SERVICE_CORP_API_TYPE['GET_PRE_AUTH_CODE']).get('pre_auth_code')
print pre_auth_code
except ApiException as e :
print e.errCode, e.errMsg
## 第三方服务商使用永久授权码调用企业接口的方法
api = ServiceCorpApi(
"SUITE_ID",
"SUITE_SECRET",
"SUITE_TICKET",
'AUTH_CORPID',
'PERMANENT_CODE'
);
try :
response = api.httpCall(
CORP_API_TYPE['USER_GET'],
{
'userid' : 'zhangsan',
})
print response
except ApiException as e :
print e.errCode, e.errMsg
api/examples/TestConf.py
+37
View File
@@ -0,0 +1,37 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
##
# Copyright (C) 2018 All rights reserved.
#
# @File conf.py
# @Brief
# @Author abelzhu, abelzhu@tencent.com
# @Version 1.0
# @Date 2018-02-24
#
#
# 请将下面参数改为自己的企业相关参数再进行测试
TestConf = {
# 企业的id，在管理端->"我的企业" 可以看到
"CORP_ID" : "ww55ca070cb9b7eb22",
# "通讯录同步"应用的secret, 开启api接口同步后，可以在管理端->"通讯录同步"看到
"CONTACT_SYNC_SECRET" : "ktmzrVIlUH0UW63zi7-JyzsgTL9NfwUhHde6or6zwQY",
# 某个自建应用的id及secret, 在管理端 -> 企业应用 -> 自建应用, 点进相应应用可以看到
"APP_ID" : 1000002,
"APP_SECRET" : "v1Z2KSw2WqPFECAwn2R0a1dFsanVF5sE4IE6X5ogveQ",
# 打卡应用的 id 及secrete， 在管理端 -> 企业应用 -> 基础应用 -> 打卡，
# 点进去，有个"api"按钮，点开后，会看到
"CHECKIN_APP_ID" : 3010011,
"CHECKIN_APP_SECRET" : "3Qz2OGPvE1Eb6WKpEDfczvyQjL5Lr1CjrDTKn0RHdLE",
# 审批应用的 id 及secrete， 在管理端 -> 企业应用 -> 基础应用 -> 审批，
# 点进去，有个"api"按钮，点开后，会看到
"APPROVAL_APP_ID" : 3010040,
"APPROVAL_APP_SECRET" : "1vrlwItWpz_5Qkud55aImQPCvpzi51H3F2j-1OQzhYE",
}
api/examples/UserTest.py
+63
View File
@@ -0,0 +1,63 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
##
# Copyright (C) 2018 All rights reserved.
#
# @File UserTest.py
# @Brief
# @Author abelzhu, abelzhu@tencent.com
# @Version 1.0
# @Date 2018-02-24
#
#
import sys
sys.path.append("../src/")
from CorpApi import *
from TestConf import *
## test
api = CorpApi(TestConf['CORP_ID'], TestConf['CONTACT_SYNC_SECRET'])
try :
##
response = api.httpCall(
CORP_API_TYPE['USER_CREATE'],
{
'userid' : 'zhangsan',
'name' : 'zhangsanfeng',
'mobile' : '131488888888',
'email' : 'zhangsan@ipp.cas.cn',
'department' : 1,
})
print response
##
response = api.httpCall(
CORP_API_TYPE['USER_GET'],
{
'userid' : 'zhangsan',
})
print response
##
response = api.httpCall(
CORP_API_TYPE['USER_DELETE'],
{
'userid' : 'zhangsan',
})
print response
except ApiException as e :
print e.errCode, e.errMsg
##
response = api.httpCall(
CORP_API_TYPE['USER_DELETE'],
{
'userid' : 'zhangsan',
})
print response
api/src/AbstractApi.py
+146
View File
@@ -0,0 +1,146 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
##
# Copyright (C) 2018 All rights reserved.
#
# @File AbstractApi.py
# @Brief
# @Author abelzhu, abelzhu@tencent.com
# @Version 1.0
# @Date 2018-02-24
#
#
import sys
import os
import re
import json
import requests
sys.path.append("../../")
from conf import DEBUG
class ApiException(Exception) :
def __init__(self, errCode, errMsg) :
self.errCode = errCode
self.errMsg = errMsg
class AbstractApi(object) :
def __init__(self) :
return
def getAccessToken(self) :
raise NotImplementedError
def refreshAccessToken(self) :
raise NotImplementedError
def getSuiteAccessToken(self) :
raise NotImplementedError
def refreshSuiteAccessToken(self) :
raise NotImplementedError
def getProviderAccessToken(self) :
raise NotImplementedError
def refreshProviderAccessToken(self) :
raise NotImplementedError
def httpCall(self, urlType, args=None) :
shortUrl = urlType[0]
method = urlType[1]
response = {}
for retryCnt in range(0, 3) :
if 'POST' == method :
url = self.__makeUrl(shortUrl)
response = self.__httpPost(url, args)
elif 'GET' == method :
url = self.__makeUrl(shortUrl)
url = self.__appendArgs(url, args)
response = self.__httpGet(url)
else :
raise ApiException(-1, "unknown method type")
# check if token expired
if self.__tokenExpired(response.get('errcode')) :
self.__refreshToken(shortUrl)
++ retryCnt
continue
else :
break
return self.__checkResponse(response)
@staticmethod
def __appendArgs(url, args) :
if args is None :
return url
for key, value in args.items() :
if '?' in url :
url += ('&' + key + '=' + value)
else :
url += ('?' + key + '=' + value)
return url
@staticmethod
def __makeUrl(shortUrl) :
base = "https://qyapi.weixin.qq.com"
if shortUrl[0] == '/' :
return base + shortUrl
else :
return base + '/' + shortUrl
def __appendToken(self, url) :
if 'SUITE_ACCESS_TOKEN' in url :
return url.replace('SUITE_ACCESS_TOKEN', self.getSuiteAccessToken())
elif 'PROVIDER_ACCESS_TOKEN' in url :
return url.replace('PROVIDER_ACCESS_TOKEN', self.getProviderAccessToken())
elif 'ACCESS_TOKEN' in url :
return url.replace('ACCESS_TOKEN', self.getAccessToken())
else :
return url
def __httpPost(self, url, args) :
realUrl = self.__appendToken(url)
if DEBUG is True :
print realUrl, args
return requests.post(realUrl, data=json.dumps(args).decode('unicode-escape').encode("utf-8")).json()
def __httpGet(self, url) :
realUrl = self.__appendToken(url)
if DEBUG is True :
print realUrl
return requests.get(realUrl).json()
def __post_file(self, url, media_file):
return requests.post(url, file=media_file).json()
@staticmethod
def __checkResponse(response):
errCode = response.get('errcode')
errMsg = response.get('errmsg')
if errCode is 0:
return response
else:
raise ApiException(errCode, errMsg)
@staticmethod
def __tokenExpired(errCode) :
if errCode == 40014 or errCode == 42001 or errCode == 42007 or errCode == 42009 :
return True
else :
return False
def __refreshToken(self, url) :
if 'SUITE_ACCESS_TOKEN' in url :
self.refreshSuiteAccessToken()
elif 'PROVIDER_ACCESS_TOKEN' in url :
self.refreshProviderAccessToken()
elif 'ACCESS_TOKEN' in url :
self.refreshAccessToken()
api/src/CorpApi.py
+96
View File
@@ -0,0 +1,96 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
##
# Copyright (C) 2018 All rights reserved.
#
# @File CorpApi.py
# @Brief
# @Author abelzhu, abelzhu@tencent.com
# @Version 1.0
# @Date 2018-02-24
#
#
from AbstractApi import *
CORP_API_TYPE = {
'GET_ACCESS_TOKEN' : ['/cgi-bin/gettoken', 'GET'],
'USER_CREATE' : ['/cgi-bin/user/create?access_token=ACCESS_TOKEN', 'POST'],
'USER_GET' : ['/cgi-bin/user/get?access_token=ACCESS_TOKEN', 'GET'],
'USER_UPDATE' : ['/cgi-bin/user/update?access_token=ACCESS_TOKEN', 'POST'],
'USER_DELETE' : ['/cgi-bin/user/delete?access_token=ACCESS_TOKEN', 'GET'],
'USER_BATCH_DELETE': ['/cgi-bin/user/batchdelete?access_token=ACCESS_TOKEN', 'POST'],
'USER_SIMPLE_LIST ': ['/cgi-bin/user/simplelist?access_token=ACCESS_TOKEN', 'GET'],
'USER_LIST' : ['/cgi-bin/user/list?access_token=ACCESS_TOKEN', 'GET'],
'USERID_TO_OPENID' : ['/cgi-bin/user/convert_to_openid?access_token=ACCESS_TOKEN', 'POST'],
'OPENID_TO_USERID' : ['/cgi-bin/user/convert_to_userid?access_token=ACCESS_TOKEN', 'POST'],
'USER_AUTH_SUCCESS': ['/cgi-bin/user/authsucc?access_token=ACCESS_TOKEN', 'GET'],
'DEPARTMENT_CREATE': ['/cgi-bin/department/create?access_token=ACCESS_TOKEN', 'POST'],
'DEPARTMENT_UPDATE': ['/cgi-bin/department/update?access_token=ACCESS_TOKEN', 'POST'],
'DEPARTMENT_DELETE': ['/cgi-bin/department/delete?access_token=ACCESS_TOKEN', 'GET'],
'DEPARTMENT_LIST' : ['/cgi-bin/department/list?access_token=ACCESS_TOKEN', 'GET'],
'TAG_CREATE' : ['/cgi-bin/tag/create?access_token=ACCESS_TOKEN', 'POST'],
'TAG_UPDATE' : ['/cgi-bin/tag/update?access_token=ACCESS_TOKEN', 'POST'],
'TAG_DELETE' : ['/cgi-bin/tag/delete?access_token=ACCESS_TOKEN', 'GET'],
'TAG_GET_USER' : ['/cgi-bin/tag/get?access_token=ACCESS_TOKEN', 'GET'],
'TAG_ADD_USER' : ['/cgi-bin/tag/addtagusers?access_token=ACCESS_TOKEN', 'POST'],
'TAG_DELETE_USER' : ['/cgi-bin/tag/deltagusers?access_token=ACCESS_TOKEN', 'POST'],
'TAG_GET_LIST' : ['/cgi-bin/tag/list?access_token=ACCESS_TOKEN', 'GET'],
'BATCH_JOB_GET_RESULT' : ['/cgi-bin/batch/getresult?access_token=ACCESS_TOKEN', 'GET'],
'BATCH_INVITE' : ['/cgi-bin/batch/invite?access_token=ACCESS_TOKEN', 'POST'],
'AGENT_GET' : ['/cgi-bin/agent/get?access_token=ACCESS_TOKEN', 'GET'],
'AGENT_SET' : ['/cgi-bin/agent/set?access_token=ACCESS_TOKEN', 'POST'],
'AGENT_GET_LIST' : ['/cgi-bin/agent/list?access_token=ACCESS_TOKEN', 'GET'],
'MENU_CREATE' : ['/cgi-bin/menu/create?access_token=ACCESS_TOKEN', 'POST'],
'MENU_GET' : ['/cgi-bin/menu/get?access_token=ACCESS_TOKEN', 'GET'],
'MENU_DELETE' : ['/cgi-bin/menu/delete?access_token=ACCESS_TOKEN', 'GET'],
'MESSAGE_SEND' : ['/cgi-bin/message/send?access_token=ACCESS_TOKEN', 'POST'],
'MEDIA_GET' : ['/cgi-bin/media/get?access_token=ACCESS_TOKEN', 'GET'],
'GET_USER_INFO_BY_CODE' : ['/cgi-bin/user/getuserinfo?access_token=ACCESS_TOKEN', 'GET'],
'GET_USER_DETAIL' : ['/cgi-bin/user/getuserdetail?access_token=ACCESS_TOKEN', 'POST'],
'GET_TICKET' : ['/cgi-bin/ticket/get?access_token=ACCESS_TOKEN', 'GET'],
'GET_JSAPI_TICKET' : ['/cgi-bin/get_jsapi_ticket?access_token=ACCESS_TOKEN', 'GET'],
'GET_CHECKIN_OPTION' : ['/cgi-bin/checkin/getcheckinoption?access_token=ACCESS_TOKEN', 'POST'],
'GET_CHECKIN_DATA' : ['/cgi-bin/checkin/getcheckindata?access_token=ACCESS_TOKEN', 'POST'],
'GET_APPROVAL_DATA': ['/cgi-bin/corp/getapprovaldata?access_token=ACCESS_TOKEN', 'POST'],
'GET_INVOICE_INFO' : ['/cgi-bin/card/invoice/reimburse/getinvoiceinfo?access_token=ACCESS_TOKEN', 'POST'],
'UPDATE_INVOICE_STATUS' :
['/cgi-bin/card/invoice/reimburse/updateinvoicestatus?access_token=ACCESS_TOKEN', 'POST'],
'BATCH_UPDATE_INVOICE_STATUS' :
['/cgi-bin/card/invoice/reimburse/updatestatusbatch?access_token=ACCESS_TOKEN', 'POST'],
'BATCH_GET_INVOICE_INFO' :
['/cgi-bin/card/invoice/reimburse/getinvoiceinfobatch?access_token=ACCESS_TOKEN', 'POST'],
}
class CorpApi(AbstractApi) :
def __init__(self, corpid, secret) :
self.corpid = corpid
self.secret = secret
self.access_token = None
def getAccessToken(self) :
if self.access_token is None :
self.refreshAccessToken()
return self.access_token
def refreshAccessToken(self) :
response = self.httpCall(
CORP_API_TYPE['GET_ACCESS_TOKEN'],
{
'corpid' : self.corpid,
'corpsecret': self.secret,
})
self.access_token = response.get('access_token')
api/src/ServiceCorpApi.py
+70
View File
@@ -0,0 +1,70 @@
#!/usr/bin/env python
# -*- coding:utf-8 -*-
##
# Copyright (C) 2018 All rights reserved.
#
# @File ServiceCorp.py
# @Brief
# @Author abelzhu, abelzhu@tencent.com
# @Version 1.0
# @Date 2018-02-24
#
#
from CorpApi import *
SERVICE_CORP_API_TYPE = {
'GET_CORP_TOKEN' : ['/cgi-bin/service/get_corp_token?suite_access_token=SUITE_ACCESS_TOKEN', 'POST'],
'GET_SUITE_TOKEN' : ['/cgi-bin/service/get_suite_token', 'POST'],
'GET_PRE_AUTH_CODE' : ['/cgi-bin/service/get_pre_auth_code?suite_access_token=SUITE_ACCESS_TOKEN', 'GET'],
'SET_SESSION_INFO ' : ['/cgi-bin/service/set_session_info?suite_access_token=SUITE_ACCESS_TOKEN', 'POST'],
'GET_PERMANENT_CODE' : ['/cgi-bin/service/get_permanent_code?suite_access_token=SUITE_ACCESS_TOKEN', 'POST'],
'GET_AUTH_INFO ' : ['/cgi-bin/service/get_auth_info?suite_access_token=SUITE_ACCESS_TOKEN', 'POST'],
'GET_ADMIN_LIST ' : ['/cgi-bin/service/get_admin_list?suite_access_token=SUITE_ACCESS_TOKEN', 'POST'],
'GET_USER_INFO_BY_3RD' : ['/cgi-bin/service/getuserinfo3rd?suite_access_token=SUITE_ACCESS_TOKEN', 'GET'],
'GET_USER_DETAIL_BY_3RD' : ['/cgi-bin/service/getuserdetail3rd?suite_access_token=SUITE_ACCESS_TOKEN', 'POST'],
}
class ServiceCorpApi(CorpApi) :
def __init__(self, suite_id, suite_secret, suite_ticket, auth_corpid=None, permanent_code=None) :
self.suite_id = suite_id
self.suite_secret = suite_secret
self.suite_ticket = suite_ticket
# 调用 CorpAPI 的function， 需要设置这两个参数
self.auth_corpid = auth_corpid
self.permanent_code = permanent_code
self.access_token = None
self.suite_access_token = None
## override CorpApi 的 refreshAccessToken， 使用第三方服务商的方法
def getAccessToken(self) :
if self.access_token is None :
self.refreshAccessToken()
return self.access_token
def refreshAccessToken(self) :
response = self.httpCall(
SERVICE_CORP_API_TYPE['GET_CORP_TOKEN'],
{
"auth_corpid" : self.auth_corpid,
"permanent_code": self.permanent_code,
})
self.access_token = response.get('access_token')
##
def getSuiteAccessToken(self) :
if self.suite_access_token is None :
self.refreshSuiteAccessToken()
return self.suite_access_token
def refreshSuiteAccessToken(self) :
response = self.httpCall(
SERVICE_CORP_API_TYPE['GET_SUITE_TOKEN'],
{
"suite_id" : self.suite_id,
"suite_secret" : self.suite_secret,
"suite_ticket" : self.suite_ticket,
})
self.suite_access_token= response.get('suite_access_token')