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()