python 調用接口

這個比較亂，抽口再修改一下。

 

工作需要調有贊API的接口數據,  返回數據. 進行數據處理

現在兩部分比較重要:1 自動獲取數據  ,  2處理excel的過程.

1. 明白接口的過程.傳入參數

　　  https://open.youzan.com/v3/apicenter/doc-api-main/1/2/salesman/youzan.salesman.trades.get 

      　這個地址里面,都有 各個很細的API 過程. 例如

 

　  　導入 外部API  。例如獲得分銷用戶的方法的API . 

　　

 

 

　　用戶的api 文檔

from app.external.yz.api_util import YzApiUtil


class YzUserApi(YzApiUtil):
    """
    有贊會員Api
    """

    def get_follower(self, fans_id=None, weixin_openid=None):
        """
        根據微信粉絲用戶的 weixin_openid 或 fans_id 獲取用戶信息
        :param fans_id:
        :param weixin_openid:
        :return:
        """

        params = {}

        if fans_id:
            params['fans_id'] = fans_id
        elif weixin_openid:
            params['weixin_openid'] = weixin_openid
        else:
            return None

        data = self._invoke('youzan.users.weixin.follower.get', '3.0.0', 'GET', params=params)

        if 'response' in data:
            return data['response']['user']
        else:
            return None

    def get_customer(self, fans_id=None, mobile=None, youzan_account=None):
        """
        獲取客戶詳情
        :param fans_id:
        :param mobile:
        :param youzan_account:
        :return:
        """

        params = {}

        if fans_id:
            params['account'] = '{"account_type":"FansID","account_id":"' + str(fans_id) + '"}'
        elif mobile:
            params['account'] = '{"account_type":"Mobile","account_id":"' + str(mobile) + '"}'
        elif youzan_account:
            params['account'] = '{"account_type":"YouZanAccount","account_id":"' + str(youzan_account) + '"}'
        else:
            return None

        data = self._invoke('youzan.scrm.customer.get', '3.1.0', 'GET', params=params)

        if 'response' in data:
            return data['response']
        else:
            return None

    def get_openid(self, mobile):
        """
        使用手機號獲取用戶openId（獲取的商家自有粉絲的openId。fans_type =1時）
        :param mobile: 會員手機號
        :return:
        {
          "open_id": "oTtVis-xiMQjlBME5Xi4Bc_twuqA",
          "union_id": "oqY0-wpXFmBsPI2IrTUYx3DigfjY"
        }
        """

        params = {
            'mobile': mobile,
        }

        data = self._invoke('youzan.user.weixin.openid.get', '3.0.0', 'GET', params=params)

        if 'response' in data:
            return data['response']
        else:
            return None

    def import_user(self):
        """
        導入三方帳號信息，用於App開店帳號一鍵授權
        :doc: https://open.youzan.com/v3/apicenter/doc-api-main/1/2/account_import/youzan.user.platform.import
        """

        params = {
            'country_code': '+86',
            'mobile': '17392940060',
            'open_user_id': '17392940060',
        }

        data = self._invoke('youzan.user.platform.import', '1.0.0', 'GET', params=params)

        if 'response' in data:
            return data['response']
        else:
            return None

 

 

　　2  項目中接口服務

　　測試接口通的數據代碼:

from app.external.yz.user_api import YzUserApi
from app.model.yz_user import YzUser
from app.utils.x_logger import logger


class YzUserTestService:
    """
    有贊用戶Test Service
    """


#  這是測試看是否通


    def __init__(self):
        self.yz_api_user = YzUserApi()

    def get_user(self, mobile):
        user = self.yz_api_user.get_customer(mobile=mobile)
        logger.info(user)
        return user




#主程序執行，理論上下面的這些不要寫在里面。 執行數據保存

if __name__ == '__main__':
    result_user = YzUserTestService().get_user("18874189337")
　　# 用戶傳入參數

    yz_user = YzUser()
    yz_user.name = result_user['name']
    yz_user.telephone = result_user['mobile']
    yz_user.save()

    a = 1

 

　　

 3  保存mysql 數據庫表里

 

 

 

 

在mabao51數據庫里建好同樣數據結構的表。

 

from app import db


class YzUser(db.Model):
    # 自定義表名:
    __tablename__ = 'yz_user'
    id = db.Column(db.BigInteger, primary_key=True)
    name = db.Column(db.String(50), nullable=False, default='')
    telephone = db.Column(db.String(20), default='')

    def save(self):
        db.session.add(self)
        db.session.commit()


if __name__ == '__main__':
    db.create_all()
    # test = YzUser.query.get(1)

    a = 1