#!/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 datetime
import hashlib
import math
import time
from concurrent.futures import ProcessPoolExecutor
from urllib import parse
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 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):
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 = 1
##接口鉴权参数
client_id = account[0]
token = account[1]
##订单接口参数必填
channel_id = account[3]
##需要保存的参数
stage = account[2]
channel = account[4]
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//30 == 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"数据日期-{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 start_order_job():
start_exec_seconds = date_util.getCurrentSecondTime()
st_unix, et_unix = date_util.getPreviousHourAndCurrentHourSecondTime(start_exec_seconds)
# 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))
print('订单同步执行时间（秒）：', date_util.getCurrentSecondTime() - start_exec_seconds)
#start_order_job()
start_job_time = '2020-11-10 01:20:00'
if __name__ == '__main__':
scheduler = BlockingScheduler()
scheduler.add_job(start_order_job, 'interval', days =1 ,start_date=start_job_time)
#scheduler.add_job(start_order_job, 'interval',days =1,hours = 2,minutes = 0,seconds = 0)
#线上是24h执行一次
scheduler.start()