419 lines
9.8 KiB
Python
419 lines
9.8 KiB
Python
"""
|
|
股票分析系统部署脚本
|
|
提供系统部署、配置和环境设置功能
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import subprocess
|
|
import argparse
|
|
import platform
|
|
from pathlib import Path
|
|
|
|
|
|
class DeploymentManager:
|
|
"""部署管理器类"""
|
|
|
|
def __init__(self):
|
|
self.project_root = Path(__file__).parent
|
|
self.venv_path = self.project_root / 'venv'
|
|
self.requirements_file = self.project_root / 'requirements.txt'
|
|
self.config_dir = self.project_root / 'config'
|
|
self.logs_dir = self.project_root / 'logs'
|
|
self.data_dir = self.project_root / 'data'
|
|
|
|
self.is_windows = platform.system() == 'Windows'
|
|
self.is_linux = platform.system() == 'Linux'
|
|
self.is_macos = platform.system() == 'Darwin'
|
|
|
|
def check_environment(self):
|
|
"""检查运行环境"""
|
|
print("正在检查运行环境...")
|
|
|
|
# 检查Python版本
|
|
python_version = sys.version_info
|
|
if python_version < (3, 8):
|
|
print(f"错误: 需要Python 3.8或更高版本，当前版本: {python_version.major}.{python_version.minor}")
|
|
return False
|
|
|
|
print(f"✓ Python版本: {python_version.major}.{python_version.minor}.{python_version.micro}")
|
|
|
|
# 检查操作系统
|
|
print(f"✓ 操作系统: {platform.system()} {platform.release()}")
|
|
|
|
# 检查必要工具
|
|
tools = ['pip', 'git']
|
|
for tool in tools:
|
|
try:
|
|
subprocess.run([tool, '--version'], capture_output=True, check=True)
|
|
print(f"✓ {tool} 已安装")
|
|
except (subprocess.CalledProcessError, FileNotFoundError):
|
|
print(f"⚠ {tool} 未安装或不在PATH中")
|
|
|
|
return True
|
|
|
|
def create_virtual_environment(self):
|
|
"""创建虚拟环境"""
|
|
print("正在创建虚拟环境...")
|
|
|
|
if self.venv_path.exists():
|
|
print("虚拟环境已存在，跳过创建")
|
|
return True
|
|
|
|
try:
|
|
subprocess.run([
|
|
sys.executable, '-m', 'venv', str(self.venv_path)
|
|
], check=True, cwd=self.project_root)
|
|
|
|
print("✓ 虚拟环境创建成功")
|
|
return True
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"✗ 创建虚拟环境失败: {e}")
|
|
return False
|
|
|
|
def get_venv_python(self):
|
|
"""获取虚拟环境中的Python路径"""
|
|
if self.is_windows:
|
|
return self.venv_path / 'Scripts' / 'python.exe'
|
|
else:
|
|
return self.venv_path / 'bin' / 'python'
|
|
|
|
def get_venv_pip(self):
|
|
"""获取虚拟环境中的pip路径"""
|
|
if self.is_windows:
|
|
return self.venv_path / 'Scripts' / 'pip.exe'
|
|
else:
|
|
return self.venv_path / 'bin' / 'pip'
|
|
|
|
def install_dependencies(self):
|
|
"""安装依赖包"""
|
|
print("正在安装依赖包...")
|
|
|
|
if not self.requirements_file.exists():
|
|
print("✗ requirements.txt 文件不存在")
|
|
return False
|
|
|
|
try:
|
|
pip_path = self.get_venv_pip()
|
|
|
|
# 升级pip
|
|
subprocess.run([str(pip_path), 'install', '--upgrade', 'pip'], check=True)
|
|
|
|
# 安装依赖
|
|
subprocess.run([
|
|
str(pip_path), 'install', '-r', str(self.requirements_file)
|
|
], check=True)
|
|
|
|
print("✓ 依赖包安装成功")
|
|
return True
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"✗ 安装依赖包失败: {e}")
|
|
return False
|
|
|
|
def create_directories(self):
|
|
"""创建必要的目录"""
|
|
print("正在创建必要的目录...")
|
|
|
|
directories = [self.logs_dir, self.data_dir]
|
|
|
|
for directory in directories:
|
|
try:
|
|
directory.mkdir(exist_ok=True)
|
|
print(f"✓ 创建目录: {directory}")
|
|
except Exception as e:
|
|
print(f"✗ 创建目录失败 {directory}: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def setup_environment_variables(self):
|
|
"""设置环境变量"""
|
|
print("正在设置环境变量...")
|
|
|
|
# 创建.env文件
|
|
env_file = self.project_root / '.env'
|
|
|
|
if not env_file.exists():
|
|
try:
|
|
with open(env_file, 'w', encoding='utf-8') as f:
|
|
f.write("""# 股票分析系统环境配置
|
|
DATABASE_URL=sqlite:///stock_data.db
|
|
LOG_LEVEL=INFO
|
|
DEBUG=false
|
|
|
|
# 数据源配置
|
|
AKSHARE_BASE_URL=https://api.akshare.akfamily.xyz
|
|
BAOSTOCK_TIMEOUT=30
|
|
|
|
# 定时任务配置
|
|
SCHEDULER_ENABLED=true
|
|
DAILY_UPDATE_TIME=18:00
|
|
|
|
# 性能配置
|
|
MAX_CONCURRENT_TASKS=10
|
|
BATCH_SIZE=100
|
|
""")
|
|
|
|
print("✓ 环境配置文件创建成功")
|
|
|
|
except Exception as e:
|
|
print(f"✗ 创建环境配置文件失败: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def run_tests(self):
|
|
"""运行测试"""
|
|
print("正在运行测试...")
|
|
|
|
try:
|
|
python_path = self.get_venv_python()
|
|
|
|
result = subprocess.run([
|
|
str(python_path), '-m', 'pytest', 'tests/',
|
|
'-v', '--tb=short'
|
|
], cwd=self.project_root, capture_output=True, text=True)
|
|
|
|
if result.returncode == 0:
|
|
print("✓ 所有测试通过")
|
|
return True
|
|
else:
|
|
print("✗ 测试失败")
|
|
print(result.stdout)
|
|
print(result.stderr)
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"✗ 运行测试失败: {e}")
|
|
return False
|
|
|
|
def create_startup_scripts(self):
|
|
"""创建启动脚本"""
|
|
print("正在创建启动脚本...")
|
|
|
|
try:
|
|
python_path = self.get_venv_python()
|
|
run_script_path = self.project_root / 'run.py'
|
|
|
|
if self.is_windows:
|
|
# 创建Windows批处理文件
|
|
batch_content = f"""@echo off
|
|
cd /d "{self.project_root}"
|
|
"{python_path}" "{run_script_path}" %*
|
|
pause
|
|
"""
|
|
|
|
with open(self.project_root / 'start.bat', 'w', encoding='utf-8') as f:
|
|
f.write(batch_content)
|
|
|
|
print("✓ Windows启动脚本创建成功")
|
|
|
|
# 创建Linux/Mac启动脚本
|
|
shell_content = f"""#!/bin/bash
|
|
cd "{self.project_root}"
|
|
"{python_path}" "{run_script_path}" "$@"
|
|
"""
|
|
|
|
with open(self.project_root / 'start.sh', 'w', encoding='utf-8') as f:
|
|
f.write(shell_content)
|
|
|
|
# 设置执行权限
|
|
if not self.is_windows:
|
|
os.chmod(self.project_root / 'start.sh', 0o755)
|
|
|
|
print("✓ Linux/Mac启动脚本创建成功")
|
|
|
|
except Exception as e:
|
|
print(f"✗ 创建启动脚本失败: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def create_service_scripts(self):
|
|
"""创建服务脚本（用于生产环境）"""
|
|
print("正在创建服务脚本...")
|
|
|
|
try:
|
|
python_path = self.get_venv_python()
|
|
run_script_path = self.project_root / 'run.py'
|
|
|
|
if self.is_linux:
|
|
# 创建systemd服务文件
|
|
service_content = f"""[Unit]
|
|
Description=Stock Analysis System
|
|
After=network.target
|
|
|
|
[Service]
|
|
Type=simple
|
|
User=stock
|
|
Group=stock
|
|
WorkingDirectory={self.project_root}
|
|
ExecStart={python_path} {run_script_path} scheduler
|
|
Restart=always
|
|
RestartSec=10
|
|
|
|
[Install]
|
|
WantedBy=multi-user.target
|
|
"""
|
|
|
|
with open(self.project_root / 'stock-system.service', 'w', encoding='utf-8') as f:
|
|
f.write(service_content)
|
|
|
|
print("✓ systemd服务文件创建成功")
|
|
|
|
# 创建supervisor配置
|
|
supervisor_content = f"""[program:stock-system]
|
|
command={python_path} {run_script_path} scheduler
|
|
directory={self.project_root}
|
|
autostart=true
|
|
autorestart=true
|
|
user=stock
|
|
stdout_logfile={self.logs_dir}/supervisor.log
|
|
stderr_logfile={self.logs_dir}/supervisor_error.log
|
|
"""
|
|
|
|
with open(self.project_root / 'stock-system.conf', 'w', encoding='utf-8') as f:
|
|
f.write(supervisor_content)
|
|
|
|
print("✓ supervisor配置创建成功")
|
|
|
|
except Exception as e:
|
|
print(f"✗ 创建服务脚本失败: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def backup_database(self):
|
|
"""备份数据库"""
|
|
print("正在备份数据库...")
|
|
|
|
db_file = self.project_root / 'stock_data.db'
|
|
backup_dir = self.project_root / 'backups'
|
|
|
|
if not db_file.exists():
|
|
print("数据库文件不存在，跳过备份")
|
|
return True
|
|
|
|
try:
|
|
backup_dir.mkdir(exist_ok=True)
|
|
|
|
from datetime import datetime
|
|
backup_file = backup_dir / f'stock_data_backup_{datetime.now().strftime("%Y%m%d_%H%M%S")}.db'
|
|
|
|
import shutil
|
|
shutil.copy2(db_file, backup_file)
|
|
|
|
print(f"✓ 数据库备份成功: {backup_file}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ 数据库备份失败: {e}")
|
|
return False
|
|
|
|
def deploy(self, skip_tests=False, production=False):
|
|
"""执行完整部署流程"""
|
|
print("开始部署股票分析系统...")
|
|
print("=" * 50)
|
|
|
|
# 检查环境
|
|
if not self.check_environment():
|
|
return False
|
|
|
|
print("-" * 50)
|
|
|
|
# 创建虚拟环境
|
|
if not self.create_virtual_environment():
|
|
return False
|
|
|
|
print("-" * 50)
|
|
|
|
# 安装依赖
|
|
if not self.install_dependencies():
|
|
return False
|
|
|
|
print("-" * 50)
|
|
|
|
# 创建目录
|
|
if not self.create_directories():
|
|
return False
|
|
|
|
print("-" * 50)
|
|
|
|
# 设置环境变量
|
|
if not self.setup_environment_variables():
|
|
return False
|
|
|
|
print("-" * 50)
|
|
|
|
# 运行测试
|
|
if not skip_tests:
|
|
if not self.run_tests():
|
|
print("测试失败，是否继续部署? (y/N): ")
|
|
response = input().strip().lower()
|
|
if response != 'y':
|
|
return False
|
|
|
|
print("-" * 50)
|
|
|
|
# 创建启动脚本
|
|
if not self.create_startup_scripts():
|
|
return False
|
|
|
|
print("-" * 50)
|
|
|
|
# 生产环境额外配置
|
|
if production:
|
|
if not self.create_service_scripts():
|
|
return False
|
|
|
|
if not self.backup_database():
|
|
return False
|
|
|
|
print("=" * 50)
|
|
print("✓ 部署完成!")
|
|
|
|
# 显示使用说明
|
|
print("\n使用说明:")
|
|
print("1. 启动系统: " + ("start.bat" if self.is_windows else "./start.sh"))
|
|
print("2. 查看帮助: " + ("start.bat --help" if self.is_windows else "./start.sh --help"))
|
|
print("3. 初始化数据: " + ("start.bat init" if self.is_windows else "./start.sh init"))
|
|
|
|
return True
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
parser = argparse.ArgumentParser(description='股票分析系统部署工具')
|
|
|
|
parser.add_argument(
|
|
'--skip-tests',
|
|
action='store_true',
|
|
help='跳过测试'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--production',
|
|
action='store_true',
|
|
help='生产环境部署'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--backup-only',
|
|
action='store_true',
|
|
help='仅备份数据库'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
deployer = DeploymentManager()
|
|
|
|
if args.backup_only:
|
|
deployer.backup_database()
|
|
else:
|
|
deployer.deploy(skip_tests=args.skip_tests, production=args.production)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main() |