manyi/venv/Lib/site-packages/weixin/mp.py

# -*- coding: utf-8 -*-


from __future__ import unicode_literals

import os
import time
import json
import hashlib
import string
import random
import requests

from .base import Map, WeixinError


__all__ = ("WeixinMPError", "WeixinMP")


DEFAULT_DIR = os.getenv("HOME", os.getcwd())


class WeixinMPError(WeixinError):

    def __init__(self, msg):
        super(WeixinMPError, self).__init__(msg)


class WeixinMP(object):
    api_uri = "https://api.weixin.qq.com/cgi-bin"

    def __init__(self, app_id, app_secret, ac_path=None, jt_path=None):
        self.app_id = app_id
        self.app_secret = app_secret
        self.session = requests.Session()
        if ac_path is None:
            ac_path = os.path.join(DEFAULT_DIR, ".access_token")
        if jt_path is None:
            jt_path = os.path.join(DEFAULT_DIR, ".jsapi_ticket")
        self.ac_path = ac_path
        self.jt_path = jt_path

    def fetch(self, method, url, params=None, data=None, headers=None):
        req = requests.Request(method, url, params=params,
                               data=data, headers=headers)
        prepped = req.prepare()
        resp = self.session.send(prepped, timeout=20)
        data = Map(resp.json())
        if data.errcode:
            msg = "%(errcode)d %(errmsg)s" % data
            raise WeixinMPError(msg)
        return data

    def get(self, path, params=None, token=True):
        url = "{0}{1}".format(self.api_uri, path)
        params = {} if not params else params
        token and params.setdefault("access_token", self.access_token)
        return self.fetch("GET", url, params)

    def post(self, path, data, json_encode=True, token=True):
        url = "{0}{1}".format(self.api_uri, path)
        params = {}
        token and params.setdefault("access_token", self.access_token)
        headers = {}
        if json_encode:
            data = json.dumps(data, ensure_ascii=False)
            headers["Content-Type"] = "application/json"
        return self.fetch("POST", url, params=params, data=data, headers=headers)

    @property
    def access_token(self):
        """
        获取服务端凭证
        """
        timestamp = time.time()
        if not os.path.exists(self.ac_path) or \
                int(os.path.getmtime(self.ac_path)) < timestamp:
            params = dict()
            params.setdefault("grant_type", "client_credential")
            params.setdefault("appid", self.app_id)
            params.setdefault("secret", self.app_secret)
            data = self.get("/token", params, False)
            with open(self.ac_path, 'wb') as fp:
                fp.write(data.access_token.encode("utf-8"))
            os.utime(self.ac_path, (timestamp, timestamp + data.expires_in - 600))
        return open(self.ac_path).read()

    @property
    def jsapi_ticket(self):
        """
        获取jsapi ticket
        """
        timestamp = time.time()
        if not os.path.exists(self.jt_path) or \
                int(os.path.getmtime(self.jt_path)) < timestamp:
            params = dict()
            params.setdefault("type", "jsapi")
            data = self.get("/ticket/getticket", params, True)
            with open(self.jt_path, 'wb') as fp:
                fp.write(data.ticket.encode("utf-8"))
            os.utime(self.jt_path, (timestamp, timestamp + data.expires_in - 600))
        return open(self.jt_path).read()

    @property
    def nonce_str(self):
        char = string.ascii_letters + string.digits
        return "".join(random.choice(char) for _ in range(32))

    def jsapi_sign(self, **kwargs):
        """
        生成签名给js使用
        """
        timestamp = str(int(time.time()))
        nonce_str = self.nonce_str
        kwargs.setdefault("jsapi_ticket", self.jsapi_ticket)
        kwargs.setdefault("timestamp", timestamp)
        kwargs.setdefault("noncestr", nonce_str)
        raw = [(k, kwargs[k]) for k in sorted(kwargs.keys())]
        s = "&".join("=".join(kv) for kv in raw if kv[1])
        sign = hashlib.sha1(s.encode("utf-8")).hexdigest().lower()
        return Map(sign=sign, timestamp=timestamp, noncestr=nonce_str)

    def groups_create(self, name):
        """
        创建分组

        :param name: 分组名
        """
        data = dict(group=dict(name=name))
        return self.post("/groups/create", data)

    def groups_get(self):
        """
        获取所有分组
        """
        return self.get("/groups/get")

    def groups_getid(self, openid):
        """
        查询用户所在分组

        :param openid: 用户id
        """
        data = dict(openid=openid)
        return self.post("/groups/getid", data)

    def groups_update(self, id, name):
        """
        修改分组名

        :param id: 分组id
        :param name: 分组名
        """
        data = dict(group=dict(id=id, name=name))
        return self.post("/groups/update", data)

    def groups_members_update(self, to_groupid, openid):
        """
        移动用户分组

        :param to_groupid: 分组id
        :param openid: 用户唯一标识符
        """
        data = dict(openid=openid, to_groupid=to_groupid)
        return self.post("/groups/members/update", data)

    def groups_members_batchupdate(self, to_groupid, *openid):
        """
        批量移动用户分组

        :param to_groupid: 分组id
        :param openid: 用户唯一标示列表
        """
        data = dict(openid_list=openid, to_groupid=to_groupid)
        return self.post("/groups/members/batchupdate", data)

    def groups_delete(self, id):
        """
        删除组

        :param id: 分组的id
        """
        data = dict(group=dict(id=id))
        return self.post("/groups/delete", data)

    def user_info_updateremark(self, openid, remark):
        """
        设置备注名

        :param openid: 用户唯一标识符
        :param remark: 备注
        """
        data = dict(openid=openid, remark=remark)
        return self.post("/user/info/updateremark", data)

    def user_info(self, openid):
        """
        获取用户信息
        包含subscribe字段，可以用来判断用户是否关注公众号

        :param openid: 用户id
        """
        args = dict(openid=openid, lang="zh_CN")
        return self.get("/user/info", args)

    def user_info_batchget(self, *openid):
        """
        批量获取用户信息
        """
        user_list = []
        for id in openid:
            user_list.append(dict(openid=openid, lang="zh_CN"))
        data = dict(user_list=user_list)
        return self.post("/user/info/batchget", data)

    def user_get(self, next_openid=None):
        """
        获取公众号关注列表
        一次最多返回1000个

        :param next_openid: 第一个拉取的openid,不填默认从头开始
        """
        args = dict()
        if next_openid:
            args.setdefault("next_openid", next_openid)
        return self.get("/user/get", args)

    def menu_create(self, data):
        data = dict(button=data)
        return self.post("/menu/create", data)

    def menu_get(self):
        return self.get("/menu/get")

    def menu_delete(self):
        return self.get("/menu/delete")

    def get_current_selfmenu_info(self):
        return self.get("/get_current_selfmenu_info")

    def shorturl(self, long_url):
        """
        长链接转为短链接

        :param long_url: 长链接
        """
        data = dict(action="long2short", long_url=long_url)
        return self.post("/shorturl", data)

    def qrcode_create(self, scene_id, expires=30):
        """
        创建qrcode
        """
        data = dict(
            action_name="QR_SCENE", expire_seconds=expires,
            action_info=dict(scene=dict(scene_id=scene_id)),
        )
        return self.post("/qrcode/create", data)

    def qrcode_create_limit(self, input):
        """
        创建qrcode限制方式
        """
        data = dict()
        if isinstance(input, int):
            data["action_name"] = "QR_LIMIT_SCENE"
            data["action_info"] = dict(scene=dict(
                scene_id=input,
            ))
        elif isinstance(input, str):
            data["action_name"] = "QR_LIMIT_STR_SCENE"
            data["action_info"] = dict(scene=dict(
                scene_str=input,
            ))
        else:
            raise ValueError("invalid type")
        return self.post("/qrcode/create", data)

    def qrcode_show(self, ticket):
        """
        显示qrcode
        """
        url = "https://mp.weixin.qq.com/cgi-bin/showqrcode"
        return self.add_query(url, dict(ticket=ticket))