"Fossies" - the Fresh Open Source Software Archive

Member "seahub-7.1.0-pro/thirdpart/weworkapi/AbstractApi.py" (30 Dec 2019, 4032 Bytes) of package /linux/www/seahub-7.1.0-pro.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file.

    1 #!/usr/bin/env python
    2 # -*- coding:utf-8 -*-
    3 ##
    4  # Copyright (C) 2018 All rights reserved.
    5  #   
    6  # @File AbstractApi.py
    7  # @Brief 
    8  # @Author abelzhu, abelzhu@tencent.com
    9  # @Version 1.0
   10  # @Date 2018-02-24
   11  #
   12  #
   13  
   14 import sys
   15 import os
   16 import re
   17 
   18 import json
   19 import requests
   20 
   21 DEBUG = False
   22 
   23 class ApiException(Exception) :
   24     def __init__(self, errCode, errMsg) :
   25         self.errCode = errCode
   26         self.errMsg = errMsg
   27 
   28 class AbstractApi(object) :
   29     def __init__(self) : 
   30         return
   31 
   32     def getAccessToken(self) :
   33         raise NotImplementedError
   34     def refreshAccessToken(self) :
   35         raise NotImplementedError
   36 
   37     def getSuiteAccessToken(self) :
   38         raise NotImplementedError
   39     def refreshSuiteAccessToken(self) :
   40         raise NotImplementedError
   41 
   42     def getProviderAccessToken(self) :
   43         raise NotImplementedError
   44     def refreshProviderAccessToken(self) :
   45         raise NotImplementedError
   46 
   47     def httpCall(self, urlType, args=None) : 
   48         shortUrl = urlType[0]
   49         method = urlType[1]
   50         response = {}
   51         for retryCnt in range(0, 3) :
   52             if 'POST' == method :
   53                 url = self.__makeUrl(shortUrl)
   54                 response = self.__httpPost(url, args)
   55             elif 'GET' == method :
   56                 url = self.__makeUrl(shortUrl)
   57                 url = self.__appendArgs(url, args)
   58                 response = self.__httpGet(url)
   59             else : 
   60                 raise ApiException(-1, "unknown method type")
   61 
   62             # check if token expired
   63             if self.__tokenExpired(response.get('errcode')) :
   64                 self.__refreshToken(shortUrl)
   65                 retryCnt += 1
   66                 continue
   67             else :
   68                 break
   69 
   70         return self.__checkResponse(response) 
   71 
   72     @staticmethod
   73     def __appendArgs(url, args) : 
   74         if args is None :
   75             return url
   76 
   77         for key, value in list(args.items()) : 
   78             if '?' in url : 
   79                 url += ('&' + key + '=' + value)
   80             else :
   81                 url += ('?' + key + '=' + value)
   82         return url
   83 
   84     @staticmethod
   85     def __makeUrl(shortUrl) :
   86         base = "https://qyapi.weixin.qq.com"
   87         if shortUrl[0] == '/' :
   88             return base + shortUrl
   89         else :
   90             return base + '/' + shortUrl 
   91 
   92     def __appendToken(self, url) : 
   93         if 'SUITE_ACCESS_TOKEN' in url :
   94             return url.replace('SUITE_ACCESS_TOKEN', self.getSuiteAccessToken())
   95         elif 'PROVIDER_ACCESS_TOKEN' in url :
   96             return url.replace('PROVIDER_ACCESS_TOKEN', self.getProviderAccessToken())
   97         elif 'ACCESS_TOKEN' in url :
   98             return url.replace('ACCESS_TOKEN', self.getAccessToken())
   99         else : 
  100             return url
  101 
  102     def __httpPost(self, url, args) :
  103         realUrl = self.__appendToken(url)
  104 
  105         if DEBUG is True : 
  106             print(realUrl, args) 
  107 
  108         return requests.post(realUrl, data = json.dumps(args, ensure_ascii = False).encode('utf-8')).json()
  109 
  110     def __httpGet(self, url) :
  111         realUrl = self.__appendToken(url)
  112 
  113         if DEBUG is True : 
  114             print(realUrl) 
  115 
  116         return requests.get(realUrl).json()
  117 
  118     def __post_file(self, url, media_file):
  119         return requests.post(url, file=media_file).json()
  120 
  121     @staticmethod
  122     def __checkResponse(response):
  123         errCode = response.get('errcode')
  124         errMsg = response.get('errmsg')
  125 
  126         if errCode is 0:
  127             return response 
  128         else:
  129             raise ApiException(errCode, errMsg)
  130 
  131     @staticmethod
  132     def __tokenExpired(errCode) :
  133         if errCode == 40014 or errCode == 42001 or errCode == 42007 or errCode == 42009 :
  134             return True
  135         else :
  136             return False
  137 
  138     def __refreshToken(self, url) :
  139         if 'SUITE_ACCESS_TOKEN' in url :
  140             self.refreshSuiteAccessToken()
  141         elif 'PROVIDER_ACCESS_TOKEN' in url :
  142             self.refreshProviderAccessToken()
  143         elif 'ACCESS_TOKEN' in url :
  144             self.refreshAccessToken()