python脚本实战---修改时间戳与接口请求

涉及模块

 1、python日期与时间戳的转换2. 获取非当前日期时间戳3. 更新sql数据4. 接口调用

需求点

修改data_user_cpl_reward表中数据A为前一日数据
调用清除缓存接口

导入模块

import mysql.connector
import requests,json
import time
from datetime import datetime,timedelta

获取昨日当前时间戳

def yestoday():current_time = datetime.today()Adjust_time = current_time - timedelta(days=1)#时间前推一天untime = time.mktime(Adjust_time.timetuple())#mktime() 函数执行与gmtime(), localtime()相反的操作，它接收struct_time对象作为参数，返回用秒数来表示时间的浮点数return  int(untime)
yt_time = yestoday()

查询当前时间

def nowTime():now = time.time()N_t = time.localtime(now)N_tt = time.strftime("%Y-%m-%d %H:%M:%S",N_t)#时间戳转具体日期return N_tt
current_time = nowTime()
print("当前时间为:",current_time)

连接数据库

config = {'host':'47.XX.XX.33','user':'XX','password':'NewXXXXXXTIFIED1','port':'3306','database':'XXX','charset':'utf8'
}

检查数据库状态

try:cnn = mysql.connector.connect(**config)print("数据库连接成功")
except mysql.connector.Error as e:print("数据库连接失败")

更新sql表中时间戳

def update_cpa():cursor = cnn.cursor(buffered=True)try:update = "update data_user_cpa_task set rewardTypeTime = %s where userId = %s and taskId = %s and rewardType = 4"#此处含有多个占位符，不可使用cursor.execute直接拼参，需要单独传值sql = cursor.execute(update,(yt_time,userId,taskId))cnn.commit()except mysql.connector.Error as q:print("更新数据报错",q)

查询调整后时间

def select_Cpa_Time():cursor = cnn.cursor(buffered=True)sql = "SELECT rewardTypeTime from data_user_cpa_task WHERE userId = %s and taskId = %s and rewardType = 4"sql_select = cursor.execute(sql,(userId,taskId))repare = cursor.fetchone()if repare == None:print("查询无此数据，请核对后再试！")else:stamp = repare[0]Array_time = time.localtime(stamp)now_time = time.strftime("%Y-%m-%d %H:%M:%S",Array_time)print("已成功更新cpa时间戳:", stamp,"cpa更新后时间：",now_time)cursor.close()cnn.close()

调取缓存接口

github_url = 'http://XXX.com/v1/test/refresh-tomorrow-tasks-cache'
header = {'Content-Type':'application/json','mk':'o8293nHdiwnjNjksnv','isObj':'1','platform':'2','system':'{"adChannel":"","androidId":"cd87566612901e90","channelId":"1","deviceBrand":"OPPO","deviceType":"PCHM10","imei":"","imeiAll":"","imsi":"","isEmulator":0,"isRoot":0,"joinId":1,"macAddress":"b4:a5:ac:2f:6a:65","oaid":"4C7C0DE1F0274A5194F4E406AA9CAAED752a7dee774a8f4b62831ebbe3566ea8","platform":1,"systemVersion":"11","systemVersionCode":30,"token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJ4aWFvemh1byIsImF1ZCI6InhpYW96aHVvIiwiaWF0IjoxNjI2NzYxMTgzLCJleHAiOjE2MjkzNTMxODMsImRhdGEiOnsidXNlcklkIjoxMDAwMTY3NSwiZGV2aWNlSWQiOjMxMjM2OH19.nK4w_wfDEq3uRFmcWxPDFrE2WOFLUrsYBpRW5_zMCzo","uuid":"984d7dd3-660b-4f3e-b48e-5ebd5f1e2661","versionCode":13,"versionName":"1.1.2","ymId":"8e3f928cc666b6b7e813352577749623od"}'}
data = {"userId":userId,"platform":2}
data_json = json.dumps(data)
r = requests.post(github_url,headers = header,data=data_json)
print (r.json())

总体代码展示

# coding=UTF-8
import mysql.connector
import requests,json
import time
from datetime import datetime,timedelta
userId = 10001634
taskId = 181173#获取昨日当前时间戳
def yestoday():current_time = datetime.today()Adjust_time = current_time - timedelta(days=1)#时间前推一天untime = time.mktime(Adjust_time.timetuple())#mktime() 函数执行与gmtime(), localtime()相反的操作，它接收struct_time对象作为参数，返回用秒数来表示时间的浮点数return  int(untime)
yt_time = yestoday()
#查询当前时间
def nowTime():now = time.time()N_t = time.localtime(now)N_tt = time.strftime("%Y-%m-%d %H:%M:%S",N_t)#时间戳转具体日期return N_tt
current_time = nowTime()
print("当前时间为:",current_time)#连接数据库
config = {'host':'47.00.00.00','user':'newtest','password':'********','port':'3306','database':'crazy_bounty','charset':'utf8'
}
#检查数据库状态
try:cnn = mysql.connector.connect(**config)print("数据库连接成功")
except mysql.connector.Error as e:print("数据库连接失败")
#更新CPA时间戳
def update_cpa():cursor = cnn.cursor(buffered=True)try:update = "update data_user_cpa_task set rewardTypeTime = %s where userId = %s and taskId = %s and rewardType = 4"#此处含有多个占位符，不可使用cursor.execute直接拼参，需要单独传值sql = cursor.execute(update,(yt_time,userId,taskId))cnn.commit()except mysql.connector.Error as q:print("更新数据报错",q)
#更新CPL时间戳
def update_cpl():cursor = cnn.cursor(buffered=True)try:update = "update data_user_cpl_reward set rewardTypeTime = %s where userId = %s and taskId = %s and rewardType = 4"cpl_sql = cursor.execute(update,(yt_time,userId,taskId))cnn.commit()except mysql.connector.Error as l:print("更新CPL数据出错",l)
#查询CPA调整后时间
def select_Cpa_Time():cursor = cnn.cursor(buffered=True)sql = "SELECT rewardTypeTime from data_user_cpa_task WHERE userId = %s and taskId = %s and rewardType = 4"sql_select = cursor.execute(sql,(userId,taskId))repare = cursor.fetchone()if repare == None:print("查询无此数据，请核对后再试！")else:stamp = repare[0]Array_time = time.localtime(stamp)now_time = time.strftime("%Y-%m-%d %H:%M:%S",Array_time)print("已成功更新cpa时间戳:", stamp,"cpa更新后时间：",now_time)cursor.close()cnn.close()
#查询CPL调整后时间
def select_Cpl_Time():cursor = cnn.cursor(buffered=True)sql = "SELECT rewardTypeTime from data_user_cpl_reward WHERE userId = %s and taskId = %s and rewardType = 4"sql_select = cursor.execute(sql,(userId,taskId))repare = cursor.fetchone()if repare == None:print("查询无此数据，请核对后再试！")else:stamp = repare[0]Arrary_Time = time.localtime(stamp)now_Time = time.strftime("%Y-%m-%d %H:%M:%S",Arrary_Time)print("已成功更新时间戳",stamp,"更新后时间：",now_Time)cursor.close()cnn.close()#执行cpa任务
update_cpa()
select_Cpa_Time()#执行cpl任务
update_cpl()
select_Cpl_Time()#调取缓存接口
github_url = 'http://xxx.com/v1/test/refresh-tomorrow-tasks-cache'
header = {'Content-Type':'application/json','mk':'o8293nHdiwnjNjksnv','isObj':'1','platform':'2','system':'{"adChannel":"","androidId":"cd87566612901e90","channelId":"1","deviceBrand":"OPPO","deviceType":"PCHM10","imei":"","imeiAll":"","imsi":"","isEmulator":0,"isRoot":0,"joinId":1,"macAddress":"b4:a5:ac:2f:6a:65","oaid":"4C7C0DE1F0274A5194F4E406AA9CAAED752a7dee774a8f4b62831ebbe3566ea8","platform":1,"systemVersion":"11","systemVersionCode":30,"token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJ4aWFvemh1byIsImF1ZCI6InhpYW96aHVvIiwiaWF0IjoxNjI2NzYxMTgzLCJleHAiOjE2MjkzNTMxODMsImRhdGEiOnsidXNlcklkIjoxMDAwMTY3NSwiZGV2aWNlSWQiOjMxMjM2OH19.nK4w_wfDEq3uRFmcWxPDFrE2WOFLUrsYBpRW5_zMCzo","uuid":"984d7dd3-660b-4f3e-b48e-5ebd5f1e2661","versionCode":13,"versionName":"1.1.2","ymId":"8e3f928cc666b6b7e813352577749623od"}'}
data = {"userId":userId,"platform":2}
data_json = json.dumps(data)
r = requests.post(github_url,headers = header,data=data_json)
print (r.json())

输出结果展示

python脚本实战---修改时间戳与接口请求相关推荐

Python脚本调用腾讯地图接口
Python脚本调用腾讯地图接口腾讯地图提供的有公开的webServerAPI,BAT在很多领域三分天下,地图领域也是如此.腾讯地图在几年以前还是烂的掉渣,自从腾讯的很多生态合作伙伴使用了之后,现状 ...
Qt与Python脚本实战之一(爬虫)
Qt与Python脚本实战之一(爬虫) Python环境安装 1.Python版本选择(2.7 or 3.6.x) 版本不同区别蛮大 2.安装pip 一个Python包管理工具类似nodejs的np ...
Python接口自动化实战 ( 第一阶段) - 封装接口请求类和异常处理
1.封装http接口请求已经实现了一个简单的接口请求,接下来就要考虑封装这个请求,在后面的用例中,只需要传递参数(URL ,Params,cookie,heade,method 等)每次去调用这个请 ...
使用python脚本批量修改vc工程文件
作者:朱金灿来源:clever101的专栏为什么大多数人学不会人工智能编程?>>> 问题描述 vc工程文件中的很多属性设置,比如输出目录.中间目录和输出文件名等等.使用vs ...
python猜密码游戏规则_【python笔记三】python脚本实战---数字密码小游戏
描述:A输入一个数值,输入时该数值已隐藏无法看到,让B猜该数字为多少,数值过小或过大都会提示,并会在最后猜对时统计出共猜了几次. 脚本内容: #!/usr/bin/env python #FileNa ...
Python脚本-批量修改文件名
Notice1:os.listdir(path) filenames = os.listdir(path) os.listdir() 方法用于返回指定的文件夹包含的文件或文件夹的名字的列表.这个列表以 ...
Python爬虫实战02：分析Ajax请求并抓取今日头条街拍
1 目标网站分析首先我们打开今日头条网站,搜索街拍,点击图集,这里每就是我们要爬取的目录,我们称为索引页.1 点开一个标题,进去,称为详情页.2这里面的图是我们所要爬取的.比如这里可以点击图片,共 ...
Python脚本实战【大麦网】抢票程序，我终于可以去看我杰哥的演唱会啦！
前言大麦网,是中国综合类现场娱乐票务营销平台,业务覆盖演唱会. 话剧.音乐剧.体育赛事等领域. 但是因为票数有限,还有黄牛们不能丢了饭碗,所以导致了,很多人都抢不到票那么,今天带大家用Python ...
使用python脚本批量修改Arcgis的数据字段类型或者名称
参考其他的文章,放上代码,直接使用arcgis自带的python IDE运行,可解决中文字符的问题. # -- coding: utf-8 -- import arcpy import sys fro ...

python脚本实战---修改时间戳与接口请求

python脚本实战---修改时间戳与接口请求相关推荐

最新文章

热门文章