"""
预览服务器模块
提供浏览器预览功能，支持文件监听和热重载。
"""
import sys
import queue
import webbrowser
import random
from pathlib import Path
from threading import Thread
try:
from flask import Flask, Response
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
except ImportError:
Flask = None
Observer = None
FileSystemEventHandler = None
from core.presentation import Presentation
from renderers.html_renderer import HtmlRenderer
from loaders.yaml_loader import YAMLError
from utils import log_info, log_error
# HTML 模板
HTML_TEMPLATE = """
YAML Preview
{{ slides_html }}
"""
ERROR_TEMPLATE = """
预览错误
"""
# 全局变量
app = None
change_queue = None
current_yaml_file = None
current_template_dir = None
class YAMLChangeHandler:
"""文件变化处理器"""
def on_modified(self, event):
if event.src_path.endswith('.yaml'):
log_info(f"检测到文件变化: {event.src_path}")
if change_queue:
change_queue.put('reload')
def generate_preview_html(yaml_file, template_dir):
"""生成完整的预览 HTML 页面"""
try:
pres = Presentation(yaml_file, template_dir)
renderer = HtmlRenderer()
slides_html = ""
for i, slide_data in enumerate(pres.data.get('slides', [])):
rendered = pres.render_slide(slide_data)
slides_html += renderer.render_slide(rendered, i, Path(yaml_file).parent)
return HTML_TEMPLATE.replace('{{ slides_html }}', slides_html)
except YAMLError as e:
return ERROR_TEMPLATE.replace('{{ error }}', str(e))
def create_flask_app():
"""创建 Flask 应用"""
flask_app = Flask(__name__)
@flask_app.route('/')
def index():
"""主页面"""
try:
return generate_preview_html(current_yaml_file, current_template_dir)
except Exception as e:
return ERROR_TEMPLATE.replace('{{ error }}', f"生成预览失败: {str(e)}")
@flask_app.route('/events')
def events():
"""SSE 事件流"""
def event_stream():
while True:
change_queue.get()
yield 'data: reload\\n\\n'
return Response(event_stream(), mimetype='text/event-stream')
return flask_app
def start_preview_server(yaml_file, template_dir, port, host='127.0.0.1', open_browser=True):
"""启动预览服务器
Args:
yaml_file: YAML 文件路径
template_dir: 模板目录路径
port: 服务器端口
host: 主机地址（默认：127.0.0.1）
open_browser: 是否自动打开浏览器（默认：True）
"""
global app, change_queue, current_yaml_file, current_template_dir
if Flask is None:
log_error("预览功能需要 flask 和 watchdog 依赖")
log_error("请确保使用 uv 运行脚本，依赖会自动安装")
sys.exit(1)
current_yaml_file = yaml_file
current_template_dir = template_dir
change_queue = queue.Queue()
# 创建 Flask 应用
app = create_flask_app()
# 启动文件监听
if FileSystemEventHandler:
handler = YAMLChangeHandler()
if hasattr(handler, 'on_modified'):
# 创建一个简单的事件处理器
class SimpleHandler(FileSystemEventHandler):
def on_modified(self, event):
handler.on_modified(event)
observer = Observer()
observer.schedule(SimpleHandler(), str(Path(yaml_file).parent), recursive=False)
observer.start()
# 输出日志
log_info(f"正在监听: {yaml_file}")
log_info(f"预览地址: http://{host}:{port}")
log_info("按 Ctrl+C 停止")
# 自动打开浏览器（如果启用）
if open_browser:
Thread(target=lambda: webbrowser.open(f'http://localhost:{port}')).start()
# 启动 Flask
try:
app.run(host=host, port=port, debug=False, threaded=True)
except OSError as e:
if 'Address already in use' in str(e):
log_error(f"端口 {port} 已被占用")
log_error(f"请使用 --port 参数指定其他端口，例如: --port {port + 1}")
else:
log_error(f"启动服务器失败: {str(e)}")
sys.exit(1)
except KeyboardInterrupt:
if 'observer' in locals():
observer.stop()
observer.join()
log_info("已停止")