import urllib.request import os import sys import zipfile import subprocess import shutil import configparser MYSQL_VERSION = "5.7.44" MYSQL_ZIP_NAME = f"mysql-{MYSQL_VERSION}-winx64.zip" MYSQL_DIR_NAME = f"mysql-{MYSQL_VERSION}-winx64" MYSQL_DOWNLOAD_URL = f"https://downloads.mysql.com/archives/get/p/23/file/{MYSQL_ZIP_NAME}" PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) MYSQL_ZIP_PATH = os.path.join(PROJECT_ROOT, MYSQL_ZIP_NAME) MYSQL_HOME = os.path.join(PROJECT_ROOT, "mysql") MYSQL_DATA = os.path.join(MYSQL_HOME, "data") MYSQL_BIN = os.path.join(MYSQL_HOME, "bin") def download(): print(f"下载 MySQL {MYSQL_VERSION} 到: {MYSQL_ZIP_PATH}") print(f"URL: {MYSQL_DOWNLOAD_URL}") if os.path.exists(MYSQL_ZIP_PATH): size = os.path.getsize(MYSQL_ZIP_PATH) if size > 100 * 1024 * 1024: print(f"文件已存在, 大小: {size / 1024 / 1024:.1f} MB, 跳过下载") return True else: print(f"文件已存在但大小异常 ({size / 1024 / 1024:.1f} MB), 重新下载") os.remove(MYSQL_ZIP_PATH) def progress(block_num, block_size, total_size): downloaded = block_num * block_size percent = min(downloaded / total_size * 100, 100) if total_size > 0 else 0 mb = downloaded / 1024 / 1024 total_mb = total_size / 1024 / 1024 sys.stdout.write(f"\r下载进度: {percent:.1f}% ({mb:.1f}/{total_mb:.1f} MB)") sys.stdout.flush() try: urllib.request.urlretrieve(MYSQL_DOWNLOAD_URL, MYSQL_ZIP_PATH, progress) print(f"\n下载完成! 文件大小: {os.path.getsize(MYSQL_ZIP_PATH) / 1024 / 1024:.1f} MB") return True except Exception as e: print(f"\n下载失败: {e}") return False def extract(): if os.path.exists(MYSQL_HOME) and os.path.exists(MYSQL_BIN): print(f"MySQL 目录已存在: {MYSQL_HOME}, 跳过解压") return True if not os.path.exists(MYSQL_ZIP_PATH): print("MySQL zip 文件不存在, 请先下载") return False print(f"正在解压 {MYSQL_ZIP_NAME} ...") temp_extract = os.path.join(PROJECT_ROOT, MYSQL_DIR_NAME) try: with zipfile.ZipFile(MYSQL_ZIP_PATH, 'r') as zf: zf.extractall(PROJECT_ROOT) print("解压完成") if os.path.exists(temp_extract): if os.path.exists(MYSQL_HOME): shutil.rmtree(MYSQL_HOME) os.rename(temp_extract, MYSQL_HOME) print(f"重命名为: {MYSQL_HOME}") return True except Exception as e: print(f"解压失败: {e}") return False def _read_mysql_port_from_config(): config_path = os.path.join(PROJECT_ROOT, "config.ini") if os.path.exists(config_path): cfg = configparser.ConfigParser() cfg.read(config_path, encoding='utf-8') return cfg.getint('mysql', 'port', fallback=33366) return 33366 def init_mysql(): if os.path.exists(MYSQL_DATA): print(f"MySQL 数据目录已存在: {MYSQL_DATA}, 跳过初始化") return True mysqld = os.path.join(MYSQL_BIN, "mysqld.exe") if not os.path.exists(mysqld): print(f"mysqld.exe 不存在: {mysqld}") return False print("正在初始化 MySQL 数据目录...") try: result = subprocess.run( [mysqld, "--initialize-insecure", "--basedir=" + MYSQL_HOME, "--datadir=" + MYSQL_DATA], capture_output=True, text=True, timeout=120 ) if result.returncode != 0: print(f"初始化输出: {result.stderr}") else: print("MySQL 数据目录初始化完成") return True except Exception as e: print(f"MySQL 初始化失败: {e}") return False def write_my_ini(): my_ini_path = os.path.join(MYSQL_HOME, "my.ini") port = _read_mysql_port_from_config() ini_content = f"""[mysqld] basedir={MYSQL_HOME} datadir={MYSQL_DATA} port={port} max_connections=200 max_allowed_packet=64M character-set-server=utf8mb4 collation-server=utf8mb4_general_ci default-storage-engine=InnoDB sql_mode=NO_ENGINE_SUBSTITUTION,STRICT_TRANS_TABLES [client] port={port} default-character-set=utf8mb4 [mysql] default-character-set=utf8mb4 """ with open(my_ini_path, 'w', encoding='utf-8') as f: f.write(ini_content) print(f"my.ini 已写入: {my_ini_path} (port={port})") return True def start_mysql_temp(): mysqld = os.path.join(MYSQL_BIN, "mysqld.exe") if not os.path.exists(mysqld): print(f"mysqld.exe 不存在: {mysqld}") return False print("临时启动 MySQL 以创建用户和数据库...") try: proc = subprocess.Popen( [mysqld, "--standalone", "--console"], stdout=subprocess.PIPE, stderr=subprocess.PIPE ) import time time.sleep(5) if proc.poll() is not None: print("MySQL 启动失败") return False print("MySQL 已临时启动, 正在配置用户和数据库...") _setup_users_and_db() print("正在停止临时 MySQL...") proc.terminate() proc.wait(timeout=10) print("临时 MySQL 已停止") return True except Exception as e: print(f"临时启动 MySQL 失败: {e}") return False def _setup_users_and_db(): mysql_exe = os.path.join(MYSQL_BIN, "mysql.exe") port = _read_mysql_port_from_config() config_path = os.path.join(PROJECT_ROOT, "config.ini") cfg = configparser.ConfigParser() cfg.read(config_path, encoding='utf-8') db_user = cfg.get('mysql', 'user', fallback='admin') db_password = cfg.get('mysql', 'password', fallback='admin') db_name = cfg.get('mysql', 'database', fallback='word_sign') sql_commands = f""" CREATE DATABASE IF NOT EXISTS `{db_name}` CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci; CREATE USER IF NOT EXISTS '{db_user}'@'127.0.0.1' IDENTIFIED BY '{db_password}'; CREATE USER IF NOT EXISTS '{db_user}'@'localhost' IDENTIFIED BY '{db_password}'; CREATE USER IF NOT EXISTS '{db_user}'@'%%' IDENTIFIED BY '{db_password}'; GRANT ALL PRIVILEGES ON `{db_name}`.* TO '{db_user}'@'127.0.0.1'; GRANT ALL PRIVILEGES ON `{db_name}`.* TO '{db_user}'@'localhost'; GRANT ALL PRIVILEGES ON `{db_name}`.* TO '{db_user}'@'%%'; GRANT ALL PRIVILEGES ON *.* TO '{db_user}'@'127.0.0.1' WITH GRANT OPTION; GRANT ALL PRIVILEGES ON *.* TO '{db_user}'@'localhost' WITH GRANT OPTION; GRANT ALL PRIVILEGES ON *.* TO '{db_user}'@'%%' WITH GRANT OPTION; FLUSH PRIVILEGES; """ sql_file = os.path.join(PROJECT_ROOT, "_temp_init.sql") with open(sql_file, 'w', encoding='utf-8') as f: f.write(sql_commands) try: result = subprocess.run( [mysql_exe, "-u", "root", "--skip-password", f"--port={port}", "-e", f"source {sql_file}"], capture_output=True, text=True, timeout=30 ) if result.returncode == 0: print(f"用户 '{db_user}' 和数据库 '{db_name}' 创建成功") else: print(f"创建用户/数据库输出: {result.stderr}") print("尝试使用 mysql 命令行...") result2 = subprocess.run( [mysql_exe, "-u", "root", f"--port={port}", f"--execute={sql_commands.strip()}"], capture_output=True, text=True, timeout=30 ) if result2.returncode == 0: print(f"用户 '{db_user}' 和数据库 '{db_name}' 创建成功 (方式2)") else: print(f"创建用户/数据库失败: {result2.stderr}") except Exception as e: print(f"创建用户/数据库异常: {e}") finally: if os.path.exists(sql_file): os.remove(sql_file) def main(): print("=" * 60) print(f" MySQL {MYSQL_VERSION} 绿色版 - 下载与初始化") print("=" * 60) print() if not download(): print("下载失败, 终止") return if not extract(): print("解压失败, 终止") return if not init_mysql(): print("初始化数据目录失败, 终止") return write_my_ini() if not start_mysql_temp(): print("MySQL 临时启动配置失败, 但基本安装已完成") print("首次启动 start.bat 时会自动完成剩余配置") print() print("=" * 60) print(" MySQL 安装完成!") print(f" 目录: {MYSQL_HOME}") print(f" 数据: {MYSQL_DATA}") print(" 使用 start.bat 启动服务") print("=" * 60) if __name__ == '__main__': main()