qc-dgp/dgp/tests/get_order_dairly_yangguang.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# 20201028

## 20201027新添加 阳光 平台接口
########################## 阳光接口返回数据格式
'''
{
"error_code": 0,             ## 错误码      0 成功
"error_msg": "",             ## 错误信息
"data":{                     ## 数据内容
    "last_id": 1,                 ## 最后一条数据的 ID 请求下一页数据时进行传递
    "count": 500,                 ## 此次查询数据量
    "push_time": 1570701536,      ## 返回时间戳
    "list": [                     ## 数据列表
        {
        "merchant_id": "20180627151621_1_UdyF",                   ## 商户订单号
        "transaction_id": "4200000122201806277014647642",         ## 交易单号
        "type": "1",                        ##  订单类型,'1'=>书币充值 '2'=>VIP 充值
        "money": "50",                      ##  总额
        "state": "1",                       ##  订单状态,'0'=>未完成 '1' => 完成
        "from": "0",                        ##  
        "create_time": "1530083789",        ##  下单时间
        "finish_time": "1530083789",        ##  完成时间
        "book_name": "一品邪少",             ##  书名
        "book_tags": "现代都市",             ## 
        "referral_url": "/t/392109",        ##  推广链接
        "user_id": "112333",                ##  用户 id
        "channel_id": "1231"                ##  渠道 id
        user_createtime                     ##  用户注册时间
        openid                              ##  用户 openid
        },
    ]
'''
########################## 

import random
import hashlib
import time
from concurrent.futures import ProcessPoolExecutor
import requests
from util import date_util
from util import platform_config_util      ## 账号配置
from util import robust_util
from util.MySQLConnection import MySQLConnection


def md5(s):
    md5 = hashlib.md5()
    md5.update(s.encode("utf-8"))
    return md5.hexdigest()


def sha1(s):
    sha1 = hashlib.sha1()
    sha1.update(s.encode("utf-8"))
    return sha1.hexdigest()


def get_random_str(num=5):
    H = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789'
    salt = ''
    for i in range(num):
        salt += random.choice(H)
    return salt


def getSelfDateStr(times=time.time(),date_format='%Y%m%d'):
    """
    ## 20201028添加，阳光接口，日期参数请求格式20201028，一日一拉api数据
    description:  获取指定时间戳
    time:         秒 默认当前时间
    return:       返回指定时间戳的前一日日期 。 比如 ：接收20190512号的时间戳，返回 20190513 -> str
    tips:         一天86400秒
    """
    timestamps = str(time.strftime(date_format,time.localtime(times)))
    return timestamps


def get_yangguang_account_list():
    """
    des cription:  阳光账号列表
    return:       [['host_name', 'channel_id', 'secert_key', 'channel', 'stage']] ->list
    """
    return platform_config_util.get_account_list('阳光', 'yangguang_account_config.csv')


@robust_util.catch_exception
def get_yangguang_order(st,et,account_list):
    total_order_list = ()
    start_exec_seconds = date_util.getCurrentSecondTime()
    #account_list = get_yangguang_account_list()

    executor = ProcessPoolExecutor(max_workers=5)

    futures = []
    for account in account_list:
        future = executor.submit(get_yangguang_order_task, st, et, account)
        futures.append(future)

    executor.shutdown(True)

    for future in futures:
        order_list = future.result()
        if len(order_list) > 0:
            total_order_list = order_list + total_order_list

    print('阳光订单数量：', len(total_order_list), '执行时长（秒）：', date_util.getCurrentSecondTime() - start_exec_seconds)
    return total_order_list


def get_yangguang_order_task(st,et,account):
    order_list = ()
    url = 'https://api.yifengaf.cn:443/api/channeldataapi/orders'
    url_frequency = 0 
    ##接口鉴权参数
    client_id =  account[0]     
    token =  account[1]
    ##订单接口参数必填
    channel_id = account[3] 
    ##需要保存的参数
    stage = account[2]
    channel = account[4]

    order_st_date = date_util.getSelfDateStr(int(st-86400),date_format='%Y-%m-%d')
    for i in range((et-st)//86400 + 1):
        last_id = 111
        statis_unix_time = st + (i-1)*86400
        search_date = getSelfDateStr(statis_unix_time,"%Y%m%d")

        while True:
            if st == et:
                break
            
            if url_frequency % 3 == 0 :
                time.sleep(61)
            url_frequency += 1    
            
            nonce = get_random_str()
            timestamp = int(time.time())
            signaure = str(token)+str(timestamp)+str(client_id)+str(nonce)
            signaure = sha1(signaure)
            params = {
                ## 授权url参数
                'client_id': client_id,
                'token': token,
                'nonce': nonce,
                'timestamp': timestamp,
                'signaure': signaure,
                ## 订单url参数，那个渠道，那天的数据
                'channel_id': channel_id,
                'search_date': search_date,     
            }
            response_result_json = requests.get(url=url, params=params).json()
            code = response_result_json['error_code']
            if code != 0:
                # error_msg = response_result_json['error_msg']
                # print("阳光查询充值接口错误信息：",error_msg)
                print("阳光异常vip公众号：",channel,"所在期数：",stage)
                print('阳光查询充值接口异常：',response_result_json,'传入参数', params)
                break

            result_data = response_result_json['data'] 
            '''    
            # json['data']返回数据案例  {'last_id': 108847045, 'count': 4, 'push_time': 1604303106, 
            # 'list': [{'user_id': '531307203', 'merchant_id': '20201101111026_531307203_lYzh', 'transaction_id': '', 'type': '1', 'money': '18.00', 
            #            'state': '0', 'from': '0', 'create_time': '1604200226', 'finish_time': '', 'referral_id': '3515889', 'referral_id_permanent': '3515894', 
            #            'channel_id': '26885', 'book_name': '猛虎出山', 'book_tags': '都市', 'referral_url': '/t/3515889', 'subscribe_time': '1603687515', 
            #            'user_createtime': '1603687515', 'openid': 'om90f6EKBtwcNo9gcgISKsGNyk5o', 'register_ip': '', 'nickname': '爱拼才会赢(星辉电镀化工)', 'avatar': url连接 } ]}
            '''
            page_count = result_data['count']
            if page_count == 0:      
                break
            
            if  last_id  == result_data['last_id']:
                break
            last_id = result_data['last_id']
            params['last_id'] = result_data['last_id']

            order_item_list = result_data['list']
            for order_item in order_item_list:
                if int(order_item['state']) != 1:
                    continue

                order = {}
                order['amount'] = order_item['money']
                order['channel_id'] = order_item['channel_id']
                order['order_id'] = str(order_item['transaction_id'])
                order['order_time'] = date_util.getSecondsToDatetime(int(order_item['create_time']),"%Y-%m-%d %H:%M:%S")   ## 原数据时间戳，转日期默认=%Y-%m-%d %H:%M:%S 
                order['user_id'] = order_item['user_id']
                order['platform'] = '阳光'
                order['channel'] = channel 
                order['reg_time'] = date_util.getSecondsToDatetime(int(order_item['user_createtime']),"%Y-%m-%d %H:%M:%S")  ## 原数据时间戳，转日期默认=%Y-%m-%d %H:%M:%S
                order['from_novel'] = order_item['book_name']                                       ## str
                order['stage'] = stage                                                              ## str
                order['date'] = (int(order_item['create_time']) + 8 * 3600) // 86400 * 86400 - 8 * 3600

                x = sorted(order.items(), key=lambda item: item[0])
                x = dict(x)
                x = tuple(x.values())
                order_list = order_list + ((x),)

    print(f"数据日期-{order_st_date}到{search_date}-公众号-{channel}-获取数据-{len(order_list)}条，例如: {order_list[0:1]}")
    return order_list


def batch_save_order(data):
    if data is None or len(data) == 0:
        print('数据为空，不执行数据库操作！')
    else:
        sql = 'INSERT IGNORE INTO quchen_text.order(amount,channel,channel_id,date,from_novel,order_id,order_time,platform,reg_time,stage,user_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);'
        connect = MySQLConnection()
        try:
            num = connect.batch(sql, data)
            # 提交
            connect.commit()
            print('订单数据最终入库【{num}】条'.format(num=num))
        except Exception as e:
            print('订单数据入库失败：', e)
        finally:
            connect.close()

def batch_save_order_new(data):
    if data is None or len(data) == 0:
        print('数据为空，不执行数据库操作！')
    else:
        sql = 'INSERT IGNORE INTO quchen_text.ods_order(amount,channel,channel_id,date,from_novel,order_id,order_time,platform,reg_time,stage,user_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);'
        connect = MySQLConnection()
        try:
            num = connect.batch(sql, data)
            # 提交
            connect.commit()
            print('订单数据最终入库【{num}】条'.format(num=num))
        except Exception as e:
            print('订单数据入库失败：', e)
        finally:
            connect.close()


def start_order_job():
    start_exec_seconds = date_util.getCurrentSecondTime()
    st_unix, et_unix = date_util.getPreviousHourAndCurrentHourSecondTime(start_exec_seconds)
    account_list = get_yangguang_account_list()
    # st_unix = 1602313200  # 2020/10/10 15:0:0
    # et_unix = 1602316800  # 2020/10/10 16:0:0
    #print('查询开始时间：', st_unix, date_util.getSecondsToDatetime(st_unix))
    #print('查询结束时间：', et_unix, date_util.getSecondsToDatetime(et_unix))
    ## 20201028添加阳光平台
    batch_save_order(get_yangguang_order(st_unix, et_unix, account_list))
    batch_save_order_new(get_yangguang_order(st_unix, et_unix, account_list))
    print('订单同步执行时间（秒）：', date_util.getCurrentSecondTime() - start_exec_seconds)


if __name__ == '__main__':
    print(date_util.getCurrentFormatTimeStr())
    start_order_job()