123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
import json
|
||
import sqlite3
|
||
import os
|
||
|
||
# 获取当前脚本所在目录
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
|
||
# 定义JSON文件路径
|
||
json_file_path = os.path.join(script_dir, 'data_test_predict', 'betting_predictions_final_1_20.json')
|
||
|
||
# 定义SQLite数据库路径
|
||
db_path = os.path.join(script_dir, 'bet_records.db')
|
||
|
||
# 确保目录存在
|
||
os.makedirs(os.path.join(script_dir, 'data_test_predict'), exist_ok=True)
|
||
|
||
def create_table():
|
||
"""创建SQLite表"""
|
||
conn = sqlite3.connect(db_path)
|
||
cursor = conn.cursor()
|
||
|
||
# 创建bet_records表，包含id、bet_time和bet_data字段
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS bet_records (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
bet_time TEXT NOT NULL,
|
||
bet_data TEXT NOT NULL
|
||
)
|
||
''')
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
print("SQLite表创建成功")
|
||
|
||
def process_bet_data():
|
||
"""处理投注数据并保存到SQLite"""
|
||
try:
|
||
# 读取JSON文件
|
||
with open(json_file_path, 'r', encoding='utf-8') as f:
|
||
bet_data_array = json.load(f)
|
||
|
||
print(f"成功读取投注预测数据，共{len(bet_data_array)}条记录")
|
||
|
||
# 连接SQLite数据库
|
||
conn = sqlite3.connect(db_path)
|
||
cursor = conn.cursor()
|
||
|
||
# 处理每条投注数据
|
||
for i, bet_data in enumerate(bet_data_array):
|
||
bet_time = bet_data.get('time', '')
|
||
result = bet_data.get('result', {})
|
||
|
||
print(f"处理投注时间: {bet_time}")
|
||
|
||
# 提取有效投注数据（result中值不为null的对象）
|
||
valid_bet_data = {}
|
||
for pos, pos_data in result.items():
|
||
for num, value in pos_data.items():
|
||
if value is not None:
|
||
if pos not in valid_bet_data:
|
||
valid_bet_data[pos] = {}
|
||
valid_bet_data[pos][num] = value
|
||
|
||
# 将有效投注数据转换为JSON字符串
|
||
bet_data_json = json.dumps(valid_bet_data, ensure_ascii=False)
|
||
|
||
# 插入数据到SQLite表
|
||
cursor.execute(
|
||
"INSERT INTO bet_records (bet_time, bet_data) VALUES (?, ?)",
|
||
(bet_time, bet_data_json)
|
||
)
|
||
|
||
print(f" - 有效投注数据: {bet_data_json}")
|
||
|
||
# 提交事务
|
||
conn.commit()
|
||
print(f"成功保存{len(bet_data_array)}条投注记录到SQLite数据库")
|
||
|
||
except FileNotFoundError:
|
||
print(f"错误: JSON文件不存在 - {json_file_path}")
|
||
except json.JSONDecodeError:
|
||
print("错误: JSON文件格式不正确")
|
||
except Exception as e:
|
||
print(f"错误: {str(e)}")
|
||
finally:
|
||
if 'conn' in locals():
|
||
conn.close()
|
||
|
||
def verify_data():
|
||
"""验证数据是否成功保存"""
|
||
try:
|
||
conn = sqlite3.connect(db_path)
|
||
cursor = conn.cursor()
|
||
|
||
# 查询记录数
|
||
cursor.execute("SELECT COUNT(*) FROM bet_records")
|
||
count = cursor.fetchone()[0]
|
||
print(f"SQLite数据库中共有{count}条投注记录")
|
||
|
||
# 查询前5条记录
|
||
cursor.execute("SELECT id, bet_time, bet_data FROM bet_records LIMIT 5")
|
||
records = cursor.fetchall()
|
||
|
||
print("前5条记录:")
|
||
for record in records:
|
||
print(f"ID: {record[0]}")
|
||
print(f"投注时间: {record[1]}")
|
||
print(f"投注数据: {record[2]}")
|
||
print("-" * 50)
|
||
|
||
except Exception as e:
|
||
print(f"验证数据时出错: {str(e)}")
|
||
finally:
|
||
if 'conn' in locals():
|
||
conn.close()
|
||
|
||
if __name__ == "__main__":
|
||
print("开始处理投注数据并保存到SQLite数据库...")
|
||
create_table()
|
||
process_bet_data()
|
||
verify_data()
|
||
print("处理完成")
|