from model.DataBaseUtils import MysqlUtils
from model.DingTalkUtils import DingTalkDecorators, DingTalkUtils
from model.DateUtils import DateUtils
from model import ComUtils
import time
import json
import requests
db = MysqlUtils()
du = DateUtils()
@DingTalkDecorators("阳光")
def yangguang(start=None, end=None):
get_channel_info()
accounts = get_account("阳光")
if start:
start = start + ' 00:00:00'
end = end + ' 23:59:59'
else:
start = du.getTodayOrYestoday() + ' 00:00:00'
end = du.get_n_hours_ago(0)
client_id = 10008097
token = '2xa1d55tTPBjeEA8Ho'
if accounts.__len__() == 0:
return
else:
print(f"阳光账号数：{accounts.__len__()}")
for i in accounts:
stage = i[0]
vip_id = i[1]
print(stage, vip_id)
print(vip_id)
# get_yg_vip_channel(stage, vip_id, client_id, token)
get_yg_data(stage, vip_id, client_id, token, start, end)
print(check())
parse_order_data()
def get_yg_data(stage, vip_id, client_id, token, start, end):
url = "https://data.yifengaf.cn:443/channeldata/data/orders/list"
nonce = ComUtils.get_random_str()
timestamp = int(time.time())
signaure = ComUtils.sha1(str(token) + str(timestamp) + str(client_id) + str(nonce))
params = {
"client_id": client_id,
"token": token,
"nonce": nonce,
"timestamp": timestamp,
"signaure": signaure,
"vip_id": vip_id,
"start_time": start, # %Y-%m-%d %H:%i:%s:
"end_time": end
}
headers = {"Content-Type": "application/json"}
for i in range(5):
try:
r = requests.post(url=url, data=json.dumps(params), headers=headers, timeout=5)
break
except:
pass
raise
print(vip_id, r.text)
task_id = json.loads(r.text).get("data").get("task_id")
db.quchen_text.execute(
f"replace into yangguang_path(vip_id,task_id,stage,type) values ('{vip_id}','{task_id}','{stage}','order')")
def get_yg_vip_channel(stage, vip_id, client_id, token):
url = "https://data.yifengaf.cn:443/channeldata/data/account/list"
nonce = ComUtils.get_random_str()
timestamp = int(time.time())
signaure = ComUtils.sha1(str(token) + str(timestamp) + str(client_id) + str(nonce))
params = {
"client_id": client_id,
"token": token,
"nonce": nonce,
"timestamp": timestamp,
"signaure": signaure,
"vip_id": vip_id,
}
headers = {"Content-Type": "application/json"}
r = requests.post(url=url, data=json.dumps(params), headers=headers)
print(r.text)
task_id = json.loads(r.text).get("data").get("task_id")
db.quchen_text.execute(
f"replace into yangguang_path(vip_id,task_id,stage,type) values ('{vip_id}','{task_id}','{stage}','channel')")
def get_channel_info():
accounts = get_account("阳光")
client_id = 10008097
token = '2xa1d55tTPBjeEA8Ho'
for i in accounts:
stage = i[0]
vip_id = i[1]
print(vip_id)
get_yg_vip_channel(stage, vip_id, client_id, token)
def parse_order_data():
print(111)
accounts = get_account("阳光")
for i in accounts:
# print(i)
try:
vip_id = i[1]
stage = i[0]
data = parse_order(vip_id, stage)
save_data(data)
except Exception as e:
DingTalkUtils().send(msg='阳光出错vipid:' + str(vip_id), phone='15168342316')
print(e)
def parse_order(vip_id, stage):
print(vip_id)
url = db.quchen_text.getOne(f"select path from yangguang_path where type='channel' and vip_id={vip_id} ")
for i in range(5):
try:
r = requests.get(url, timeout=5).text
print(r)
break
except:
pass
raise
channel_di = {}
a = r.split('}')
for i in a[:-1]:
if i[-1] != '}':
b = json.loads(i + "}", strict=False)
else:
b = json.loads(i, strict=False)
channel_di[b["channel_id"]] = (b["wx_nickname"], b['app_id'])
# print(channel_di)
print(f'{stage} 有channel数：{len(channel_di)}')
info = db.quchen_text.getData(f"select stage,path from yangguang_path where type='order' and vip_id={vip_id}")
stage = info[0][0]
path = info[0][1]
for i in range(5):
try:
text = requests.get(path, timeout=5).text.replace('"referral_url":,', '')
print(text)
break
except Exception as e:
print('channel', e)
# raise
insert_data = []
for j in text.split("}")[:-1]:
if j[-1] != '}':
j = j + '}'
try:
di = json.loads(j, strict=False)
except Exception as e:
print(j)
print(e)
insert_data.append((
di["create_time"][:10],
stage,
'阳光',
channel_di[di['channel_id']][0],
di['channel_id'],
di['openid'],
di['create_time'],
di['user_createtime'],
di['money'],
di.get('book_name'),
di['merchant_id'],
2 if di['state'] == "完成" else 1,
di['user_id'],
channel_di[di['channel_id']][1],
1 if di['type'] == '书币充值' else 2,
di['merchant_id'],
di['transaction_id']
))
# print(insert_data)
# exit(0)
print("订单数：" + str(insert_data.__len__()))
save_data(insert_data)
def save_data(data):
sql = """replace into ods_order(date,stage,platform,channel,channel_id,user_id,order_time,
reg_time,amount,from_novel,order_id,status,platform_user_id,wechat_app_id,order_type,
trade_no,transaction_no) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """
# print(sql)
db.quchen_text.executeMany(sql, data)
def check():
x = 1
while True:
a = db.quchen_text.getOne("select count(1) from yangguang_path where type ='order' and path is null")
print(f" 回调接口 待处理数量 {a} ")
if a == 0:
info = '回调完成'
break
time.sleep(60)
x += 1
if x > 10:
DingTalkUtils().send('阳光订单回调延时10min', '15168342316')
info = '回调未完成'
break
return info
def get_account(plactform, id=None):
op = f" and id={id} " if id else ''
data = db.quchen_text.getData(f"select text from order_account_text where platform='{plactform}' {op}")
new_data = []
for i in data:
new_data.append(i[0].replace('\n', '').split(","))
return new_data
def daily_yg():
st = du.get_n_days(-10)
et = du.get_n_days(-1)
yangguang(st, et)
if __name__ == '__main__':
# get_channel_info()
# exit(0)
yangguang(start=du.get_n_days(-29), end=du.get_n_days(0))
# yangguang('2021-05-28','2021-05-28')
# daily_yg()
# for i in du.split_date2('2020-06-28','2020-11-03',30):
# print(i)
# yangguang(i[0], i[1])
# parse_order_data()
# get_channel_info()
# yangguang(start=du.get_n_days(-1),end=du.get_n_days(0))
parse_order('29600', stage='趣程27期')