企业微信支持第三方服务商开发 (#672)

* 企业微信支持第三方服务商开发

* FIX: Mismatched parameters in overridden method

* 第三方库和库本身的 import 用空行分割

Co-authored-by: messense <messense@icloud.com>

wechatpy/work/services/__init__.py

@@ -0,0 +1,113 @@
+# -*- coding: utf-8 -*-
+import json
+import time
+import logging
+import requests
+
+from wechatpy.client.base import BaseWeChatClient
+from wechatpy.exceptions import WeChatClientException
+from wechatpy.work.services import api
+
+logger = logging.getLogger(__name__)
+
+
+class WeChatServiceClient(BaseWeChatClient):
+    """
+    注意：access_token在第三方应用变更为suite_access_token参数
+    """
+
+    API_BASE_URL = "https://qyapi.weixin.qq.com/cgi-bin/"
+
+    auth = api.WeChatAuth()
+    miniprogram = api.WeChatMiniProgram()
+
+    def __init__(
+        self,
+        corp_id,
+        suite_id,
+        suite_secret,
+        suite_ticket,
+        access_token=None,
+        session=None,
+        timeout=None,
+        auto_retry=True,
+    ):
+        self.corp_id = corp_id
+        self.suite_id = suite_id
+        self.suite_secret = suite_secret
+        self.suite_ticket = suite_ticket
+        super().__init__(corp_id, access_token, session, timeout, auto_retry)
+
+    @property
+    def access_token_key(self):
+        return f"services_{self.corp_id}_{self.suite_id}_access_token"
+
+    def _fetch_access_token(self, url, params):
+        """The real fetch access token"""
+        logger.info("Fetching access token")
+        res = self._http.post(url=url, json=params)
+        try:
+            res.raise_for_status()
+        except requests.RequestException as reqe:
+            raise WeChatClientException(
+                errcode=None,
+                errmsg=None,
+                client=self,
+                request=reqe.request,
+                response=reqe.response,
+            )
+        result = res.json()
+        if "errcode" in result and result["errcode"] != 0:
+            raise WeChatClientException(
+                result["errcode"],
+                result["errmsg"],
+                client=self,
+                request=res.request,
+                response=res,
+            )
+
+        expires_in = 7200
+        if "expires_in" in result:
+            expires_in = result["expires_in"]
+        self.session.set(self.access_token_key, result["suite_access_token"], expires_in)
+        self.expires_at = int(time.time()) + expires_in
+        return result
+
+    def _request(self, method, url_or_endpoint, **kwargs):
+        if not url_or_endpoint.startswith(("http://", "https://")):
+            api_base_url = kwargs.pop("api_base_url", self.API_BASE_URL)
+            url = f"{api_base_url}{url_or_endpoint}"
+        else:
+            url = url_or_endpoint
+
+        if "params" not in kwargs:
+            kwargs["params"] = {}
+        if isinstance(kwargs["params"], dict) and "suite_access_token" not in kwargs["params"]:
+            kwargs["params"]["suite_access_token"] = self.access_token
+        if isinstance(kwargs.get("data", ""), dict):
+            body = json.dumps(kwargs["data"], ensure_ascii=False)
+            body = body.encode("utf-8")
+            kwargs["data"] = body
+
+        kwargs["timeout"] = kwargs.get("timeout", self.timeout)
+        result_processor = kwargs.pop("result_processor", None)
+        res = self._http.request(method=method, url=url, **kwargs)
+        try:
+            res.raise_for_status()
+        except requests.RequestException as reqe:
+            raise WeChatClientException(
+                errcode=None,
+                errmsg=None,
+                client=self,
+                request=reqe.request,
+                response=reqe.response,
+            )
+
+        return self._handle_result(res, method, url, result_processor, **kwargs)
+
+    def fetch_access_token(self):
+        """Fetch access token"""
+        return self._fetch_access_token(
+            url="https://qyapi.weixin.qq.com/cgi-bin/service/get_suite_token",
+            params={"suite_id": self.suite_id, "suite_secret": self.suite_secret, "suite_ticket": self.suite_ticket},
+        )

wechatpy/work/services/api/__init__.py

@@ -0,0 +1,2 @@
+from wechatpy.work.services.api.auth import WeChatAuth  # NOQA
+from wechatpy.work.services.api.miniprogram import WeChatMiniProgram  # NOQA

wechatpy/work/services/api/auth.py

@@ -0,0 +1,80 @@
+# -*- coding: utf-8 -*-
+
+
+from wechatpy.client.api.base import BaseWeChatAPI
+
+
+class WeChatAuth(BaseWeChatAPI):
+    """
+    应用授权（服务商、第三方应用开发相关）
+
+    https://work.weixin.qq.com/api/doc/90001/90143/90597
+
+    新的授权体系有部分接口未实现，欢迎提交 PR。
+    """
+
+    def get_pre_auth_code(self):
+        """
+        获取预授权码
+
+        详情请参考
+        https://open.work.weixin.qq.com/api/doc/90001/90143/90601
+        :return: 返回的 JSON 数据包
+        """
+        return self._get(
+            "service/get_pre_auth_code",
+        )
+
+    def set_session_info(self, pre_auth_code, app_id=None, auth_type=0):
+        """
+        设置授权配置
+
+        详情请参考
+        https://open.work.weixin.qq.com/api/doc/90001/90143/90602
+        :param pre_auth_code: 预授权码
+        :param app_id: 允许进行授权的应用id，如1、2、3， 不填或者填空数组都表示允许授权套件内所有应用
+                       （仅旧的多应用套件可传此参数，新开发者可忽略）
+        :param auth_type: 授权类型：0 正式授权， 1 测试授权。 默认值为0。注意，请确保应用在正式发布后的授权类型为“正式授权”
+        :return: 返回的 JSON 数据包
+        """
+        session_info = {"auth_type": auth_type}
+        if app_id:
+            session_info["appid"] = app_id
+        return self._post(
+            "service/set_session_info",
+            data={"pre_auth_code": pre_auth_code, "session_info": session_info},
+        )
+
+    def get_permanent_code(self, auth_code):
+        """
+        获取企业永久授权码
+
+        详情请参考
+        https://work.weixin.qq.com/api/doc/90001/90143/90603
+        :param auth_code: 临时授权码，会在授权成功时附加在redirect_uri中跳转回第三方服务商网站，或通过回调推送给服务商。长度为64至512个字节
+        :return: 返回的 JSON 数据包
+        """
+        return self._post(
+            "service/get_permanent_code",
+            data={
+                "auth_code": auth_code,
+            },
+        )
+
+    def get_auth_info(self, auth_code, permanent_code):
+        """
+        获取企业授权信息
+
+        详情请参考
+        https://work.weixin.qq.com/api/doc/10975
+        :param auth_code: 临时授权码，会在授权成功时附加在redirect_uri中跳转回第三方服务商网站，或通过回调推送给服务商。长度为64至512个字节
+        :param permanent_code: 	永久授权码，通过get_permanent_code获取
+        :return: 返回的 JSON 数据包
+        """
+        return self._post(
+            "service/get_auth_info",
+            data={
+                "auth_code": auth_code,
+                "permanent_code": permanent_code,
+            },
+        )

wechatpy/work/services/api/miniprogram.py

@@ -0,0 +1,28 @@
+# -*- coding: utf-8 -*-
+
+
+from wechatpy.client.api.base import BaseWeChatAPI
+
+
+class WeChatMiniProgram(BaseWeChatAPI):
+    """
+    小程序接口（服务商、第三方应用开发相关）
+
+    https://work.weixin.qq.com/api/doc/90001/90144/92423
+
+    新的授权体系有部分接口未实现，欢迎提交 PR。
+    """
+
+    def jscode2session(self, js_code):
+        """
+        临时登录凭证校验接口
+
+        详情请参考
+        https://work.weixin.qq.com/api/doc/90001/90143/90603
+        :param js_code: 登录时获取的 code
+        :return: 返回的 JSON 数据包
+        """
+        return self._get(
+            "service/miniprogram/jscode2session",
+            params={"js_code": js_code, "grant_type": "authorization_code"},
+        )