start_server.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
服务器启动脚本
用于启动Flask API服务器和Celery worker
支持守护进程模式，可在后台长时间运行 (Linux专用)
"""
import os
import sys
import subprocess
import signal
import threading
import time
import argparse
import atexit
import json
from pathlib import Path

# PID文件路径
PID_FILE = "cardquery_server.pid"
STATUS_FILE = "cardquery_status.json"

class DaemonServer:
    """守护进程服务器类"""
    
    def __init__(self):
        self.processes = []
        self.celery_process = None
        self.flask_process = None
        self.running = True
        
    def write_pid_file(self):
        """写入PID文件"""
        pid_data = {
            'main_pid': os.getpid(),
            'celery_pid': self.celery_process.pid if self.celery_process else None,
            'flask_pid': self.flask_process.pid if self.flask_process else None,
            'start_time': time.time()
        }
        
        with open(PID_FILE, 'w') as f:
            json.dump(pid_data, f, indent=2)
            
        # 注册退出时清理PID文件
        atexit.register(self.cleanup_pid_file)
    
    def cleanup_pid_file(self):
        """清理PID文件"""
        try:
            if os.path.exists(PID_FILE):
                os.remove(PID_FILE)
            if os.path.exists(STATUS_FILE):
                os.remove(STATUS_FILE)
        except Exception:
            pass
    
    def update_status(self, status):
        """更新状态文件"""
        status_data = {
            'status': status,
            'timestamp': time.time(),
            'celery_running': self.celery_process.poll() is None if self.celery_process else False,
            'flask_running': self.flask_process.poll() is None if self.flask_process else False
        }
        
        try:
            with open(STATUS_FILE, 'w') as f:
                json.dump(status_data, f, indent=2)
        except Exception:
            pass

    def start_celery_worker(self):
        """启动Celery worker"""
        print("正在启动 Celery worker...")
        cmd = [
            sys.executable, "-m", "celery", 
            "-A", "verification_service.celery", 
            "worker", 
            "--loglevel=info",
            "--concurrency=2"
        ]
        log_dir = "/home/cxy/log/"
        os.makedirs(log_dir, exist_ok=True)
        # 在守护进程模式下重定向输出
        if hasattr(self, 'daemon_mode') and self.daemon_mode:
            with open(os.path.join(log_dir, 'celery.log'), 'a') as log_file:
                self.celery_process = subprocess.Popen(
                    cmd, 
                    stdout=log_file, 
                    stderr=log_file,
                    stdin=subprocess.DEVNULL
                )
        else:
            self.celery_process = subprocess.Popen(cmd)
            
        return self.celery_process

    def start_flask_app(self):
        """启动Flask应用"""
        print("正在启动 Flask API 服务器...")
        cmd = [sys.executable, "app.py"]
        log_dir = "/home/cxy/log/"
        os.makedirs(log_dir, exist_ok=True)
        # 在守护进程模式下重定向输出
        if hasattr(self, 'daemon_mode') and self.daemon_mode:
            with open(os.path.join(log_dir, 'flask.log'), 'a') as log_file:
                self.flask_process = subprocess.Popen(
                    cmd,
                    stdout=log_file,
                    stderr=log_file,
                    stdin=subprocess.DEVNULL
                )
        else:
            self.flask_process = subprocess.Popen(cmd)
            
        return self.flask_process

    def start_scheduler(self):
        """启动定时任务调度器"""
        print("正在启动定时任务调度器...")
        try:
            from scheduler_service import start_scheduler
            start_scheduler()
            print("✅ 定时任务调度器启动成功")
        except Exception as e:
            print(f"❌ 启动定时任务调度器失败: {e}")
            print("   定时任务功能将不可用")

    def signal_handler(self, sig, frame):
        """信号处理器，用于优雅关闭"""
        print("\n收到停止信号，正在关闭服务器...")
        self.running = False
        self.stop_all_processes()
        sys.exit(0)

    def stop_all_processes(self):
        """停止所有进程"""
        print("正在停止所有服务...")
        self.update_status("stopping")
        
        # 停止定时任务调度器
        try:
            print("停止定时任务调度器...")
            from scheduler_service import stop_scheduler
            stop_scheduler()
        except Exception as e:
            print(f"停止定时任务调度器时出错: {e}")
        
        # 停止所有进程
        for name, process in [("Celery Worker", self.celery_process), ("Flask API", self.flask_process)]:
            if process:
                try:
                    print(f"停止 {name}...")
                    process.terminate()
                    # 等待进程优雅退出
                    try:
                        process.wait(timeout=10)
                    except subprocess.TimeoutExpired:
                        print(f"强制停止 {name}...")
                        process.kill()
                except Exception as e:
                    print(f"停止 {name} 时出错: {e}")
        
        self.cleanup_pid_file()
        print("✅ 服务器已停止")

    def check_processes(self):
        """检查进程状态"""
        while self.running:
            celery_running = self.celery_process and self.celery_process.poll() is None
            flask_running = self.flask_process and self.flask_process.poll() is None
            
            if not celery_running and self.celery_process:
                print(f"\n❌ Celery Worker 已停止运行 (退出码: {self.celery_process.returncode})")
                self.running = False
                break
                
            if not flask_running and self.flask_process:
                print(f"\n❌ Flask API 已停止运行 (退出码: {self.flask_process.returncode})")
                self.running = False
                break
            
            self.update_status("running")
            time.sleep(5)
        
        if self.running:  # 如果是因为进程异常退出
            self.stop_all_processes()

    def daemonize(self):
        """创建守护进程"""
        try:
            # 第一次fork
            pid = os.fork()
            if pid > 0:
                sys.exit(0)  # 父进程退出
        except OSError as e:
            print(f"第一次fork失败: {e}")
            sys.exit(1)

        # 脱离父进程环境
        # os.chdir("/")  # 注释掉原有切换到根目录
        os.chdir(os.path.dirname(os.path.abspath(__file__)))  # 切换到项目目录，保证相对路径可用
        os.setsid()
        os.umask(0)

        log_dir = "/home/cxy/log/"
        os.makedirs(log_dir, exist_ok=True)

        # 重定向标准文件描述符到指定日志文件
        with open('/dev/null', 'r') as f:
            os.dup2(f.fileno(), sys.stdin.fileno())
        with open(os.path.join(log_dir, 'daemon.log'), 'a+') as f:
            os.dup2(f.fileno(), sys.stdout.fileno())
        with open(os.path.join(log_dir, 'daemon.log'), 'a+') as f:
            os.dup2(f.fileno(), sys.stderr.fileno())

    def start_server(self, daemon_mode=False):
        """启动服务器"""
        self.daemon_mode = daemon_mode
        
        print("=" * 50)
        print("        CardQuery API 服务器启动器")
        print("=" * 50)
        
        # 检查必需的文件
        required_files = ['app.py', 'verification_service.py']
        missing_files = []
        
        for file in required_files:
            if not Path(file).exists():
                missing_files.append(file)
        
        if missing_files:
            print("❌ 缺少必需的文件:")
            for file in missing_files:
                print(f"   - {file}")
            print("\n请确保所有必需的文件都存在后再启动服务器。")
            return False
        
        # 如果是守护进程模式，进行daemon化
        if daemon_mode:
            print("正在创建守护进程...")
            self.daemonize()
        
        # 设置信号处理器
        signal.signal(signal.SIGINT, self.signal_handler)
        signal.signal(signal.SIGTERM, self.signal_handler)
        
        try:
            # 启动Celery worker
            self.start_celery_worker()
            
            # 等待一下确保Celery启动
            time.sleep(3)
            
            # 启动Flask应用
            self.start_flask_app()
            
            # 启动定时任务调度器
            self.start_scheduler()
            
            # 写入PID文件
            self.write_pid_file()
            
            print("\n✅ 服务器启动成功!")
            print("\n服务状态:")
            print("  - Flask API: http://localhost:5000")
            print("  - Celery Worker: 运行中")
            print("  - Redis: 请确保 Redis 服务正在运行")
            print("  - 定时任务调度器: 运行中")
            print("\n🔧 新功能:")
            print("  - 用户宿舍信息存储 (JSON格式)")
            print("  - 用户注册和登录管理")
            print("  - 用户统计信息查询")
            print("  - 定时水电费查询 (每天早上9点)")
            print("  - 查询历史数据库存储")
            
            if daemon_mode:
                print("  - 运行模式: 守护进程 (后台运行)")
                print(f"  - PID文件: {PID_FILE}")
                print(f"  - 日志文件: daemon.log, celery.log, flask.log")
            else:
                print("  - 运行模式: 前台运行")
                print("\n按 Ctrl+C 停止服务器")
            
            print("=" * 50)
            
            self.update_status("running")
            
            # 监控进程状态
            self.check_processes()
            
        except Exception as e:
            print(f"启动服务器时出错: {e}")
            self.stop_all_processes()
            return False
        
        return True

def get_server_status():
    """获取服务器状态"""
    if not os.path.exists(PID_FILE):
        return None
    
    try:
        with open(PID_FILE, 'r') as f:
            pid_data = json.load(f)
        
        # 检查主进程是否还在运行
        main_pid = pid_data.get('main_pid')
        if main_pid:
            try:
                os.kill(main_pid, 0)  # 检查进程是否存在
                return pid_data
            except OSError:
                # 进程不存在，清理PID文件
                os.remove(PID_FILE)
                return None
    except Exception:
        return None

def stop_server():
    """停止服务器"""
    pid_data = get_server_status()
    if not pid_data:
        print("❌ 服务器未运行或PID文件不存在")
        return False
    
    main_pid = pid_data.get('main_pid')
    if main_pid:
        try:
            print(f"正在停止服务器 (PID: {main_pid})...")
            os.kill(main_pid, signal.SIGTERM)
            
            # 等待进程停止
            for _ in range(30):  # 等待最多30秒
                if not get_server_status():
                    print("✅ 服务器已停止")
                    return True
                time.sleep(1)
            
            # 如果还没停止，强制停止
            print("强制停止服务器...")
            os.kill(main_pid, signal.SIGKILL)
            time.sleep(2)
            
            if os.path.exists(PID_FILE):
                os.remove(PID_FILE)
            
            print("✅ 服务器已强制停止")
            return True
            
        except OSError as e:
            print(f"停止服务器时出错: {e}")
            return False
    
    return False

def status_server():
    """显示服务器状态"""
    pid_data = get_server_status()
    if not pid_data:
        print("❌ 服务器未运行")
        return
    
    print("=" * 50)
    print("        CardQuery API 服务器状态")
    print("=" * 50)
    print(f"✅ 服务器正在运行")
    print(f"  - 主进程PID: {pid_data.get('main_pid')}")
    print(f"  - Celery PID: {pid_data.get('celery_pid', 'N/A')}")
    print(f"  - Flask PID: {pid_data.get('flask_pid', 'N/A')}")
    
    start_time = pid_data.get('start_time')
    if start_time:
        uptime = time.time() - start_time
        hours = int(uptime // 3600)
        minutes = int((uptime % 3600) // 60)
        print(f"  - 运行时间: {hours}小时 {minutes}分钟")
    
    # 显示状态文件信息
    if os.path.exists(STATUS_FILE):
        try:
            with open(STATUS_FILE, 'r') as f:
                status_data = json.load(f)
            print(f"  - 状态: {status_data.get('status', 'unknown')}")
            print(f"  - Celery状态: {'运行中' if status_data.get('celery_running') else '已停止'}")
            print(f"  - Flask状态: {'运行中' if status_data.get('flask_running') else '已停止'}")
        except Exception:
            pass
    
    print("  - API地址: http://localhost:5000")
    print("=" * 50)

def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='CardQuery API 服务器管理器 (Linux)')
    parser.add_argument('action', nargs='?', default='start', 
                       choices=['start', 'stop', 'restart', 'status', 'daemon'],
                       help='执行的操作 (start|stop|restart|status|daemon)')
    
    args = parser.parse_args()
    
    if args.action == 'start':
        # 检查是否已经在运行
        if get_server_status():
            print("❌ 服务器已经在运行，使用 'python start_server.py stop' 先停止服务")
            return
        
        server = DaemonServer()
        server.start_server(daemon_mode=False)
        
    elif args.action == 'daemon':
        # 检查是否已经在运行
        if get_server_status():
            print("❌ 服务器已经在运行，使用 'python start_server.py stop' 先停止服务")
            return
        
        print("正在启动守护进程模式...")
        server = DaemonServer()
        server.start_server(daemon_mode=True)
        
    elif args.action == 'stop':
        stop_server()
        
    elif args.action == 'restart':
        print("正在重启服务器...")
        stop_server()
        time.sleep(2)
        server = DaemonServer()
        server.start_server(daemon_mode=False)
        
    elif args.action == 'status':
        status_server()

if __name__ == "__main__":
    main()