1、爬取数据功能
import requests
url = 'https://xxxxxxxxxxxxxxxxxxx' # API
res = requests.get(url)
if res:
print('获取成功')
res.encoding = 'utf-8'
page_data = res.json()
# print('请求响应结果:', page_data)
# 获取数据中的data
data = page_data.get("data")
else:
print('获取失败,请检查网络')
这里直接使用的是python的requests爬取数据,最终获得是python字典格式的数据
2、定时功能
import threading
import time
sync_time = 12 * 3600 # 12小时同步时间(秒)
# 计时器函数
def func_timer():
print("当前时间是", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
func_task()
global timer
timer = threading.Timer(sync_time, func_timer) # 每sync_time秒执行一次
# print(threading.active_count()) # 线程数量
timer.start()
timer = threading.Timer(1, func_timer)
timer.start()
print('定时器启动成功')
这里采用的是python的定时器功能
3、数据库功能
import cx_Oracle
database = cx_Oracle.connect('username','password','host/orcl') # 数据库连接
select_sql = 'select * from table' # 查询语句
delete_sql = 'delete from table' # 删除语句
insert_sql = 'insert into table(a,b,c) values(:a,:b,:c)' # 插入语句
# 数据库查询函数
def select_table(sql, db):
cr = db.cursor() # 创建游标
cr.execute(sql) # 执行语句
result = cr.fetchall() # 一次性返回所有结果集
print(result)
cr.close() # 关闭连接
# 数据库清除函数
def delete_table(sql, db):
cr = db.cursor()
cr.execute(sql)
cr.close()
db.commit()
# 数据库插入函数
def insert_table(sql, a_data, db):
cr = db.cursor()
for a in a_data:
# 强制类型转换
values = (str(a["a"]), str(a["b"]), str(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())))
cr.execute(sql, values)
print('插入完成')
cr.close()
db.commit() # 提交
db.close()
print('清除数据')
delete_table(delete_sql, database)
print('插入数据')
insert_table(insert_sql, a_data, database)
这里采用的是python的cx_Oracle包,首先创建数据库连接,然后创建数据库游标,执行相关的sql语句,待执行完成后,关闭游标,关闭连接。
注意:插入、删除、更新操作需要提交COMMIT才能写入数据库,不然无效。
参考链接:
1、Python定时任务 每隔一定时间运行一次函数/任务 python的定时任务_没有撤退可言.-CSDN博客_python定时器1秒执行一次
共有 0 条评论