1

dgp/tests/get_order_hourly_new.py

@@ -11,7 +11,8 @@ from util import db_order_util_new
 from util import platform_order_api_util_new
 import time
 import sys
-
+from get_order_wending_new import start_order_job_wending
+from get_order_qiyue_new import start_order_job_qiyue
 def start_order_job(st,et):
     start = time.time()
 
@@ -27,6 +28,13 @@ def start_order_job(st,et):
     db_order_util_new.batch_save_order(platform_order_api_util_new.get_youshuge_order(st,et, account_list_youshuge))
     db_order_util_new.batch_save_order(platform_order_api_util_new.get_zhangdu_order(st,et, account_list_zhangdu))
 
+    start_order_job_wending(st,et)
+    start_order_job_qiyue(dt)
+
+
+
+
+
 
     print('订单同步执行时间（秒）：', int(time.time() - start))
 
@@ -36,10 +44,13 @@ if __name__ == '__main__':
 
     st = date_util.get_n_day(n=0, is_timestamp=1)
     et = date_util.getPreviousHourAndCurrentHourSecondTime()[1]
+    dt= date_util.get_n_day()
+
 
 
     if sys.argv.__len__()>1:
         st = date_util.get_n_day(n=-1, is_timestamp=1)
         et = date_util.get_n_day(n=0, is_timestamp=1)
+        dt=date_util.get_n_day(-1)
 
     start_order_job(st, et)

dgp/tests/get_order_qiyue_new.py

@@ -0,0 +1,184 @@
+# !/usr/bin/env python
+# -*- coding: utf-8 -*-
+# 20201120
+## 20201120新添加 七悦 平台接口
+## 七悦api接口地址 https://vip.zhangwenwenhua.com/uploads/attachments/open-api/open-api.pdf
+########################## 七悦订单接口，返回数据格式
+'''
+{'msg': 'ok', 'code': 0,
+    'data': {
+        'current_page': 1,
+        'data': [
+            {'id': 167100631495102464, 'money': 30, 'type': 1, 'state': 2, 'user_id': 101280121, 'channel_id': 14880, 'create_time': 1605062783, 'finish_time': 1605062790
+            , 'trade_no': '20201111104622_101280121_Ykhp', 'transaction_no': '4200000845202011115590974617', 'user_open_id': 'oxw0X6zGsL6dSX-Y8lsXMhPG1_cY'
+            , 'user_is_subscribe': 1, 'user_subscribe_time': 1605045625, 'user_create_time': 1605045625
+            , 'referral_url': 'https://wxaef9ff2c63776f6f.wenhuazw.com/index/book/chapter?book_id=10076083&sid=1007608300004&referral_id=727741'
+            , 'book_name': '我的超级老婆', 'book_keywords': '', 'wechat_app_id': 'wxaef9ff2c63776f6f', 'wechat_app_name': '炎兵文楼', 'channel_name': '炎兵文楼（趣程）'
+            , 'state_desc': '完成', 'type_desc': '书币充值'}
+            ,{...}]
+        ,'first_page_url': 'https://api.zhangwenwenhua.com/v1/orders?page=1'
+        , 'from': 1
+        , 'last_page': 1
+        , 'last_page_url': 'https://api.zhangwenwenhua.com/v1/orders?page=1'
+        , 'next_page_url': None
+        , 'path': 'https://api.zhangwenwenhua.com/v1/orders'
+        , 'per_page': '10'
+        , 'prev_page_url': None
+        , 'to': 5
+        , 'total': 5
+    }
+}
+'''
+#########################
+import time
+import requests
+
+from concurrent.futures import ProcessPoolExecutor
+import requests
+from util import date_util
+from util import platform_config_util  ## 账号配置
+from util import robust_util
+# from apscheduler.schedulers.blocking import BlockingScheduler
+from util.MySQLConnection import MySQLConnection
+
+
+def get_qiyue_account_list():
+    """
+    des cription:  七悦账号列表
+    return:       [['stage','token']] ->list
+    """
+    return platform_config_util.get_account_list('七悦', 'qiyue_account_config.csv')
+
+
+def get_qiyue_order_task(date, account):
+    order_list = ()
+
+    order_url = "https://api.zhangwenwenhua.com" + "/v1/orders"
+    stage = account[0]
+    token = account[1]
+    page = 1
+    size = 50
+
+
+
+    while True:
+
+        while True:
+            timestamp = int(time.time())
+            url = order_url + "?" + "token=" + str(token) + "&timestamp=" + str(timestamp) + "&page=" + str(
+                page) + "&size=" + str(size) + "&date=" + date
+            r = requests.get(url=url)
+            if r.status_code==200:
+                break
+            else:
+                print(r.text)
+                time.sleep(61)
+
+
+        response_result_json = r.json()
+
+        result_data = response_result_json['data']
+
+        total = result_data['total']
+        if total <= 0:
+            break
+        order_item_list = result_data['data']
+        for x in order_item_list:
+            if int(x['state']) != 2:
+                continue
+            y = {}
+
+            y['platform'] = '七悦'
+            y['channel'] = x['wechat_app_name']  ## 公众号名称
+            y['channel_id'] = x['channel_id']  ## 公众号id
+            y['from_novel'] = x['book_name']  ## 小说名称
+            y['user_id'] = x['user_id']  ## 付费用户uid
+            y['stage'] = stage  ## 期数
+            createTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(x['create_time']))
+            y['order_time'] = createTime  ## 订单生成时间
+            y['amount'] = x['money']  ## 原数据单位：元
+            uid_reg_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(x['user_create_time']))
+            y['reg_time'] = uid_reg_time  ## 用户注册时间
+            y['order_id'] = x['id']  ## 订单id
+            y['date'] = createTime
+            y = sorted(y.items(), key=lambda item: item[0])
+            y = dict(y)
+            y = tuple(y.values())
+            order_list = order_list + ((y),)
+
+        next_page_url = result_data['next_page_url']
+        if next_page_url is None:
+            break
+        page += 1
+
+    print(f"【{stage}】-【{order_list.__len__()}】条")
+    return order_list
+
+
+@robust_util.catch_exception
+def get_qiyue_order(dt, account_list):
+    total_order_list = ()
+    start_exec_seconds = date_util.getCurrentSecondTime()
+
+
+    for account in account_list:
+        order_list= get_qiyue_order_task(dt, account)
+        total_order_list = order_list + total_order_list
+
+    print('七悦订单数量：', len(total_order_list), '执行时长（秒）：', date_util.getCurrentSecondTime() - start_exec_seconds)
+    return total_order_list
+
+
+def batch_save_order(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 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_qiyue(dt):
+
+    account_list = get_qiyue_account_list()
+
+    batch_save_order(get_qiyue_order(dt, account_list))
+
+
+
+if __name__ == '__main__':
+
+    dt=date_util.get_n_day()
+    start_order_job_qiyue(dt)
+
+
+
+
+
+
+

dgp/tests/get_order_wending_new.py

@@ -0,0 +1,232 @@
+import hashlib
+import time
+import datetime
+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 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_wending_account_list():
+    sql = "select text from order_account_text where platform='文鼎'"
+    con = MySQLConnection()
+    data = con.query(sql)
+    li = []
+    for i in data:
+        a = i['text'].replace('\n', '').split(',')
+        li.append(a)
+    return li
+
+
+def get_wending_order(st, et, account_list):
+    total_order_list = ()
+    start_exec_seconds = date_util.getCurrentSecondTime()
+
+    futures = []
+    for account in account_list:
+        futures.append(get_wending_order_task(st, et, account))
+
+    for future in futures:
+
+        if len(future) > 0:
+            total_order_list = future + total_order_list
+
+    print('文鼎订单数量：', len(total_order_list), '执行时长（秒）：', date_util.getCurrentSecondTime() - start_exec_seconds)
+    return total_order_list
+
+
+def get_wd_account_siteid_list(account):
+    url = 'https://bi.reading.163.com/dist-api/siteList'
+
+    consumerkey = account[0]
+    secretkey = account[1]
+    stage = account[3]
+    timestamp = int(time.time() * 1000)
+
+    siteid_params = {
+        "consumerkey": consumerkey,
+        'secretkey': secretkey,
+        'timestamp': timestamp,
+    }
+    sorted_data = sorted(siteid_params.items(), reverse=False)
+    s = ""
+    for k, v in sorted_data:
+        s = s + str(k) + "=" + str(v)
+    sign = md5(s).lower()
+    siteid_params['sign'] = sign
+
+    consumerkey = siteid_params['consumerkey']
+    timestamp = siteid_params['timestamp']
+    parameter = 'consumerkey=' + str(consumerkey) + '&timestamp=' + str(timestamp) + '&sign=' + str(sign)
+    get_url = url + "?" + parameter
+
+    while True:
+        r = requests.get(url=get_url)
+        if r.status_code == 200:
+            break
+
+    try:
+        id_key_list = r.json()['data']
+    except:
+        return []
+    mpid_list = []
+    try:
+        for id_key_val in id_key_list:
+            mpid = dict(id_key_val)["mpId"]
+            mpid_list.append(mpid)
+    except Exception as e:
+        print(stage, '站点查询返回结果：', r.json())
+
+    return mpid_list
+
+
+def get_wending_json_object(url, params):
+    params['timestamp'] = int(time.time() * 1000)
+    sorted_data = sorted(params.items(), reverse=False)
+    s = ""
+    for k, v in sorted_data:
+        s = s + str(k) + "=" + str(v)
+    sign = md5(s).lower()
+    params['sign'] = sign
+
+    consumerkey = params['consumerkey']
+    secretkey = params['secretkey']
+    timestamp = params['timestamp']
+    siteid = params['siteid']
+    pageSize = params['pageSize']
+    starttime = params['starttime']
+    endtime = params['endtime']
+    page = params['page']
+    paystatus = params['paystatus']
+    ## +'&secretkey='+str(secretkey)
+    parameter = 'consumerkey=' + str(consumerkey) + '&timestamp=' + str(timestamp) + '&siteid=' + str(
+        siteid) + '&pageSize=' + str(pageSize) \
+                + '&starttime=' + str(starttime) + '&endtime=' + str(endtime) + '&page=' + str(
+        page) + '&paystatus=' + str(paystatus) + '&sign=' + str(sign)
+    global get_url
+    get_url = url + "?" + parameter
+
+    while True:
+        r = requests.get(url=get_url)
+        if r.status_code == 200:
+            break
+        else:
+            time.sleep(1)
+            print("请求连接出错，等待1s...")
+
+    response_result_json = r.json()
+    del params['sign']
+    return response_result_json
+
+
+def get_wending_order_task(st, et, account):
+    order_list = ()
+    url = 'https://bi.reading.163.com/dist-api/rechargeList'
+    consumerkey = account[0]
+    secretkey = account[1]
+    siteid = account[2]
+    stage = account[3]
+    siteid_list = get_wd_account_siteid_list(account)
+    print(siteid_list)
+    if len(siteid_list) == 0:
+        siteid_list.append(siteid)
+
+    starttime = getSelfDateStr(st, '%Y%m%d%H%M')
+    endtime = getSelfDateStr(et, '%Y%m%d%H%M')
+
+    for siteid in siteid_list:
+
+        page = 1
+        while True:
+            params = {
+                'consumerkey': consumerkey,
+                'secretkey': secretkey,
+                'timestamp': int(1601481600),
+                'siteid': siteid,
+                'pageSize': 1000,
+                'starttime': starttime,
+                'endtime': endtime,
+                'page': page,
+                'paystatus': 1}
+
+            response_result_json = get_wending_json_object(url, params)
+
+            order_item_list = response_result_json['data']['rechargeList']
+
+            for x in order_item_list:
+                y = {}
+
+                y['platform'] = '文鼎'
+                y['channel'] = x['wx_mpName']  ## 公众号名称
+                y['channel_id'] = x['wx_originalId']  ## 公众号id
+                y['from_novel'] = x['bookTitle']  ## 小说名称
+                y['user_id'] = x['userId']  ## 付费用户uid
+                y['stage'] = stage  ## 期数
+                createTime = time.strftime("%Y-%m-%d %H:%M:%S",
+                                           time.localtime(x['createTime'] // 1000))  ## 时间戳 》struct_time 》标准时间
+                y['order_time'] = createTime  ## 订单生成时间
+                y['amount'] = x['money'] / 100  ## 原数据单位：分
+                uid_reg_time = time.strftime("%Y-%m-%d %H:%M:%S",
+                                             time.localtime(x['userRegisterTime'] // 1000))  ## 13位时间戳 》标准时间
+                y['reg_time'] = uid_reg_time  ## 用户注册时间
+                y['order_id'] = x['ewTradeId']  ## 订单id
+                y['date'] = createTime
+                y = sorted(y.items(), key=lambda item: item[0])
+                y = dict(y)
+                y = tuple(y.values())
+                order_list = order_list + ((y),)
+
+            if len(order_item_list) < 1000:
+                break
+            else:
+                page += 1
+
+    print(f"文鼎数据日期-{starttime}到{endtime}-期数-{stage}-获取数据-{len(order_list)}条")
+    return order_list
+
+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_wending(st, et):
+    account_list = get_wending_account_list()
+    da = get_wending_order(st, et, account_list)
+    batch_save_order_new(da)
+
+
+if __name__ == '__main__':
+    start_order_job_wending()
+