306 lines
9.9 KiB
Python
306 lines
9.9 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
import datetime
|
|
import logging
|
|
|
|
# 配置日志
|
|
logging.basicConfig(
|
|
filename='migration.log',
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger('migration')
|
|
|
|
# 数据文件路径
|
|
DATA_DIR = 'data'
|
|
DB_PATH = os.path.join(DATA_DIR, 'warhammer.db')
|
|
USERS_FILE = os.path.join(DATA_DIR, 'users.json')
|
|
GAME_STATES_FILE = os.path.join(DATA_DIR, 'game_states.json')
|
|
CARDS_FILE = os.path.join(DATA_DIR, 'cards.json')
|
|
CARD_VOTES_FILE = os.path.join(DATA_DIR, 'card_votes.json')
|
|
CUSTOM_CARDS_FILE = os.path.join(DATA_DIR, 'custom_cards.json')
|
|
SESSIONS_FILE = os.path.join(DATA_DIR, 'sessions.json')
|
|
|
|
def create_tables(conn):
|
|
"""创建数据库表结构"""
|
|
cursor = conn.cursor()
|
|
|
|
# 用户表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id TEXT PRIMARY KEY,
|
|
username TEXT UNIQUE NOT NULL,
|
|
password TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
high_score INTEGER DEFAULT 0,
|
|
total_games INTEGER DEFAULT 0,
|
|
last_year INTEGER DEFAULT 41000,
|
|
last_game_time TEXT
|
|
)
|
|
''')
|
|
|
|
# 游戏状态表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS game_states (
|
|
id TEXT PRIMARY KEY,
|
|
user_id TEXT NOT NULL,
|
|
game_data TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id)
|
|
)
|
|
''')
|
|
|
|
# 自定义卡牌表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS custom_cards (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL,
|
|
character_data TEXT NOT NULL,
|
|
description TEXT NOT NULL,
|
|
option_a TEXT NOT NULL,
|
|
option_b TEXT NOT NULL,
|
|
creator_id TEXT NOT NULL,
|
|
creator_name TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY (creator_id) REFERENCES users(id)
|
|
)
|
|
''')
|
|
|
|
# 卡牌投票表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS card_votes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
card_id TEXT NOT NULL,
|
|
user_id TEXT NOT NULL,
|
|
vote_type TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY (card_id) REFERENCES custom_cards(id),
|
|
FOREIGN KEY (user_id) REFERENCES users(id),
|
|
UNIQUE(card_id, user_id)
|
|
)
|
|
''')
|
|
|
|
# 卡牌投票统计表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS card_vote_stats (
|
|
card_id TEXT PRIMARY KEY,
|
|
upvotes INTEGER DEFAULT 0,
|
|
downvotes INTEGER DEFAULT 0,
|
|
FOREIGN KEY (card_id) REFERENCES custom_cards(id)
|
|
)
|
|
''')
|
|
|
|
# 会话表
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS sessions (
|
|
id TEXT PRIMARY KEY,
|
|
user_id TEXT NOT NULL,
|
|
login_time TEXT NOT NULL,
|
|
last_activity TEXT NOT NULL,
|
|
ip_address TEXT,
|
|
status TEXT,
|
|
FOREIGN KEY (user_id) REFERENCES users(id)
|
|
)
|
|
''')
|
|
|
|
# 卡牌数据表 - 只存储一条记录，因为这是游戏基础数据
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS game_cards (
|
|
id INTEGER PRIMARY KEY,
|
|
cards_data TEXT NOT NULL
|
|
)
|
|
''')
|
|
|
|
conn.commit()
|
|
logger.info("数据库表创建完成")
|
|
|
|
def load_json_data(file_path, default_value=None):
|
|
"""从JSON文件安全地加载数据"""
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError) as e:
|
|
logger.warning(f"加载文件 {file_path} 失败: {str(e)}")
|
|
return default_value if default_value is not None else {}
|
|
|
|
def migrate_users(conn, users_data):
|
|
"""迁移用户数据"""
|
|
cursor = conn.cursor()
|
|
|
|
for user in users_data:
|
|
cursor.execute(
|
|
"INSERT OR REPLACE INTO users (id, username, password, created_at, high_score, total_games, last_year, last_game_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(
|
|
user['id'],
|
|
user['username'],
|
|
user['password'],
|
|
user.get('created_at', datetime.datetime.utcnow().isoformat()),
|
|
user.get('high_score', 0),
|
|
user.get('total_games', 0),
|
|
user.get('last_year', 41000),
|
|
user.get('last_game_time', None)
|
|
)
|
|
)
|
|
|
|
conn.commit()
|
|
logger.info(f"用户数据迁移完成，共 {len(users_data)} 条记录")
|
|
|
|
def migrate_game_states(conn, game_states_data):
|
|
"""迁移游戏状态数据"""
|
|
cursor = conn.cursor()
|
|
|
|
for state in game_states_data:
|
|
cursor.execute(
|
|
"INSERT OR REPLACE INTO game_states (id, user_id, game_data, created_at, updated_at) VALUES (?, ?, ?, ?, ?)",
|
|
(
|
|
state['id'],
|
|
state['user_id'],
|
|
json.dumps(state['game_data'], ensure_ascii=False),
|
|
state.get('created_at', datetime.datetime.utcnow().isoformat()),
|
|
state.get('updated_at', datetime.datetime.utcnow().isoformat())
|
|
)
|
|
)
|
|
|
|
conn.commit()
|
|
logger.info(f"游戏状态数据迁移完成，共 {len(game_states_data)} 条记录")
|
|
|
|
def migrate_custom_cards(conn, custom_cards_data):
|
|
"""迁移自定义卡牌数据"""
|
|
cursor = conn.cursor()
|
|
cards = custom_cards_data.get('cards', [])
|
|
|
|
for card in cards:
|
|
cursor.execute(
|
|
"INSERT OR REPLACE INTO custom_cards (id, title, character_data, description, option_a, option_b, creator_id, creator_name, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(
|
|
card['id'],
|
|
card['title'],
|
|
json.dumps(card['character'], ensure_ascii=False),
|
|
card['description'],
|
|
json.dumps(card['option_a'], ensure_ascii=False),
|
|
json.dumps(card['option_b'], ensure_ascii=False),
|
|
card['creator_id'],
|
|
card['creator_name'],
|
|
card.get('created_at', datetime.datetime.utcnow().isoformat())
|
|
)
|
|
)
|
|
|
|
conn.commit()
|
|
logger.info(f"自定义卡牌数据迁移完成，共 {len(cards)} 条记录")
|
|
|
|
def migrate_card_votes(conn, card_votes_data):
|
|
"""迁移卡牌投票数据"""
|
|
cursor = conn.cursor()
|
|
|
|
# 迁移用户投票记录
|
|
user_votes = card_votes_data.get('user_votes', {})
|
|
for user_id, votes in user_votes.items():
|
|
for card_id, vote_type in votes.items():
|
|
cursor.execute(
|
|
"INSERT OR REPLACE INTO card_votes (card_id, user_id, vote_type, created_at) VALUES (?, ?, ?, ?)",
|
|
(
|
|
card_id,
|
|
user_id,
|
|
vote_type,
|
|
datetime.datetime.utcnow().isoformat()
|
|
)
|
|
)
|
|
|
|
# 迁移卡牌投票统计
|
|
card_votes = card_votes_data.get('card_votes', {})
|
|
for card_id, votes in card_votes.items():
|
|
cursor.execute(
|
|
"INSERT OR REPLACE INTO card_vote_stats (card_id, upvotes, downvotes) VALUES (?, ?, ?)",
|
|
(
|
|
card_id,
|
|
votes.get('upvotes', 0),
|
|
votes.get('downvotes', 0)
|
|
)
|
|
)
|
|
|
|
conn.commit()
|
|
logger.info(f"卡牌投票数据迁移完成")
|
|
|
|
def migrate_sessions(conn, sessions_data):
|
|
"""迁移会话数据"""
|
|
cursor = conn.cursor()
|
|
|
|
active_sessions = sessions_data.get('active_sessions', [])
|
|
for session in active_sessions:
|
|
cursor.execute(
|
|
"INSERT OR REPLACE INTO sessions (id, user_id, login_time, last_activity, ip_address, status) VALUES (?, ?, ?, ?, ?, ?)",
|
|
(
|
|
str(hash(session['user_id'] + session.get('login_time', ''))), # 创建唯一ID
|
|
session['user_id'],
|
|
session.get('login_time', datetime.datetime.utcnow().isoformat()),
|
|
session.get('last_activity', datetime.datetime.utcnow().isoformat()),
|
|
session.get('ip_address', ''),
|
|
session.get('status', 'idle')
|
|
)
|
|
)
|
|
|
|
conn.commit()
|
|
logger.info(f"会话数据迁移完成，共 {len(active_sessions)} 条记录")
|
|
|
|
def migrate_cards_data(conn, cards_data):
|
|
"""迁移卡牌基础数据"""
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute(
|
|
"INSERT OR REPLACE INTO game_cards (id, cards_data) VALUES (?, ?)",
|
|
(1, json.dumps(cards_data, ensure_ascii=False))
|
|
)
|
|
|
|
conn.commit()
|
|
logger.info(f"卡牌基础数据迁移完成")
|
|
|
|
def main():
|
|
"""主迁移函数"""
|
|
logger.info("开始数据迁移")
|
|
|
|
# 检查数据库文件路径
|
|
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
|
|
|
|
# 创建/连接到数据库
|
|
conn = sqlite3.connect(DB_PATH)
|
|
|
|
try:
|
|
# 创建表结构
|
|
create_tables(conn)
|
|
|
|
# 加载并迁移用户数据
|
|
users_data = load_json_data(USERS_FILE, [])
|
|
migrate_users(conn, users_data)
|
|
|
|
# 加载并迁移游戏状态数据
|
|
game_states_data = load_json_data(GAME_STATES_FILE, [])
|
|
migrate_game_states(conn, game_states_data)
|
|
|
|
# 加载并迁移自定义卡牌数据
|
|
custom_cards_data = load_json_data(CUSTOM_CARDS_FILE, {"cards": []})
|
|
migrate_custom_cards(conn, custom_cards_data)
|
|
|
|
# 加载并迁移卡牌投票数据
|
|
card_votes_data = load_json_data(CARD_VOTES_FILE, {"user_votes": {}, "card_votes": {}})
|
|
migrate_card_votes(conn, card_votes_data)
|
|
|
|
# 加载并迁移会话数据
|
|
sessions_data = load_json_data(SESSIONS_FILE, {"active_sessions": []})
|
|
migrate_sessions(conn, sessions_data)
|
|
|
|
# 加载并迁移卡牌基础数据
|
|
cards_data = load_json_data(CARDS_FILE, {})
|
|
migrate_cards_data(conn, cards_data)
|
|
|
|
logger.info("数据迁移成功完成")
|
|
except Exception as e:
|
|
logger.error(f"迁移失败: {str(e)}")
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|