background_watcher.py

#!/usr/bin/env python3
"""
独立的后台文件监听服务
可以在没有Claude Code的情况下持续监听文件变更并创建检查点
"""

import os
import sys
import asyncio
import signal
import logging
from pathlib import Path
from datetime import datetime
from src.checkpoint_manager import CheckpointManager
from src.file_watcher import FileWatcher

class BackgroundCheckpointService:
    def __init__(self, project_root: str = None):
        self.project_root = project_root or self.detect_project_root()
        self.checkpoint_manager = CheckpointManager(self.project_root)
        self.file_watcher = None
        self.running = True
        
        # 设置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(f'{self.project_root}/.checkpoints/watcher.log'),
                logging.StreamHandler(sys.stdout)
            ]
        )
        self.logger = logging.getLogger(__name__)
        
    def detect_project_root(self):
        """检测项目根目录"""
        current_dir = Path(os.getcwd())
        project_markers = [
            '.git', '.gitignore', 
            'package.json', 'pyproject.toml', 'setup.py', 'requirements.txt',
            'Cargo.toml', 'go.mod', 'pom.xml', 'build.gradle',
            'CLAUDE.md', 'README.md'
        ]
        
        for parent in [current_dir] + list(current_dir.parents):
            for marker in project_markers:
                if (parent / marker).exists():
                    return str(parent)
        return str(current_dir)
    
    def setup_signal_handlers(self):
        """设置优雅退出信号处理"""
        signal.signal(signal.SIGTERM, self._signal_handler)
        signal.signal(signal.SIGINT, self._signal_handler)
    
    def _signal_handler(self, signum, frame):
        """处理退出信号"""
        self.logger.info(f"收到信号 {signum}，准备优雅退出...")
        self.running = False
    
    async def auto_checkpoint_callback(self, event_type: str, description: str):
        """文件变更回调，创建自动检查点"""
        try:
            self.logger.info(f"检测到文件变更: {description}")
            checkpoint_id = self.checkpoint_manager.create_checkpoint(
                f"后台自动保存: {description}", 
                auto=True
            )
            self.logger.info(f"✅ 创建检查点: {checkpoint_id}")
            
            # 检查并清理旧检查点
            checkpoints = self.checkpoint_manager.list_checkpoints()
            if len(checkpoints) > self.checkpoint_manager.config.get('max_checkpoints', 20):
                self.logger.info("开始清理旧检查点...")
                self.checkpoint_manager.cleanup_old_checkpoints()
                
        except Exception as e:
            self.logger.error(f"创建检查点失败: {e}")
    
    async def start_service(self):
        """启动后台监听服务"""
        self.logger.info(f"🚀 启动后台检查点服务")
        self.logger.info(f"📂 监听目录: {self.project_root}")
        
        # 创建启动检查点
        startup_checkpoint = self.checkpoint_manager.create_checkpoint(
            "后台服务启动时自动保存", 
            auto=True
        )
        self.logger.info(f"📸 启动检查点: {startup_checkpoint}")
        
        # 启动文件监听
        self.file_watcher = FileWatcher(
            self.project_root,
            self.auto_checkpoint_callback
        )
        await self.file_watcher.start()
        self.logger.info("👀 文件监听已启动")
        
        # 保持服务运行
        try:
            while self.running:
                await asyncio.sleep(1)
        except KeyboardInterrupt:
            self.logger.info("收到键盘中断")
        finally:
            await self.stop_service()
    
    async def stop_service(self):
        """停止后台服务"""
        self.logger.info("⏹️  停止后台服务...")
        
        if self.file_watcher:
            await self.file_watcher.stop()
            self.logger.info("文件监听已停止")
        
        # 创建退出检查点
        try:
            exit_checkpoint = self.checkpoint_manager.create_checkpoint(
                "后台服务退出时自动保存",
                auto=True
            )
            self.logger.info(f"📸 退出检查点: {exit_checkpoint}")
        except Exception as e:
            self.logger.error(f"创建退出检查点失败: {e}")
        
        self.logger.info("✅ 后台服务已停止")

def main():
    """主函数"""
    import argparse
    
    parser = argparse.ArgumentParser(description='后台检查点监听服务')
    parser.add_argument('--project-root', '-p', 
                       help='项目根目录路径，默认自动检测')
    parser.add_argument('--daemon', '-d', action='store_true',
                       help='以守护进程模式运行')
    
    args = parser.parse_args()
    
    if args.daemon:
        # 简单的守护进程实现
        pid = os.fork()
        if pid > 0:
            # 父进程退出
            print(f"后台服务已启动，PID: {pid}")
            print(f"日志文件: {args.project_root or os.getcwd()}/.checkpoints/watcher.log")
            print("使用 'kill {pid}' 停止服务")
            sys.exit(0)
    
    # 创建并启动服务
    service = BackgroundCheckpointService(args.project_root)
    service.setup_signal_handlers()
    
    try:
        asyncio.run(service.start_service())
    except Exception as e:
        print(f"服务启动失败: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()