import eventlet import logging # 配置 eventlet 日志,抑制 IOClosed 等错误日志 logging.getLogger('eventlet.hubs').setLevel(logging.CRITICAL) logging.getLogger('eventlet.greenio').setLevel(logging.CRITICAL) # 进行猴子补丁操作 eventlet.monkey_patch() # 配置 eventlet hub 以更优雅地处理 IOClosed 错误 try: from eventlet.hubs import hub from eventlet import hubs # 保存原始的 handle_error 方法 original_handle_error = hub.Hub.handle_error original_switch = hubs.hub.Hub.switch def custom_handle_error(self, context, type, value, tb): # 忽略 IOClosed 错误和相关的 EOFError error_str = str(value) if (type.__name__ == 'IOClosed' and 'Operation on closed file' in error_str) or \ (type.__name__ == 'EOFError') or \ (type.__name__ == 'OSError' and 'Bad file descriptor' in error_str): return # 其他错误调用原始方法 original_handle_error(self, context, type, value, tb) def custom_switch(self): try: return original_switch(self) except Exception as e: # 捕获 switch 方法中可能出现的 IOClosed 错误 error_str = str(e) if (type(e).__name__ == 'IOClosed' and 'Operation on closed file' in error_str) or \ (type(e).__name__ == 'EOFError') or \ (type(e).__name__ == 'OSError' and 'Bad file descriptor' in error_str): # 优雅处理,返回继续运行 return None # 其他异常重新抛出 raise # 替换原始方法 hub.Hub.handle_error = custom_handle_error hubs.hub.Hub.switch = custom_switch # 同样处理 greenio 模块的错误 from eventlet import greenio original_recv_loop = greenio.base.GreenSocket._recv_loop def custom_recv_loop(self, recv_func, recv_args, *args, **kwargs): try: return original_recv_loop(self, recv_func, recv_args, *args, **kwargs) except (hubs.IOClosed, EOFError, OSError) as e: # 优雅处理 IO 错误 if hasattr(e, 'errno') and e.errno in (9, 107): # Bad file descriptor or Operation on closed file return b'' raise greenio.base.GreenSocket._recv_loop = custom_recv_loop except Exception as e: # 如果修改失败,忽略错误 pass from flask import Flask from .views.routes import main_bp from .views.file import file_bp from .config import Config from .extensions import init_extensions,register_namespaces def create_app(): app = Flask(__name__, template_folder="../templates", static_folder="../static",static_url_path="/vsc-like/static") app.config.from_object(Config) @app.route('/') def hello(): return 'Hello, World!' # 注册蓝图 app.register_blueprint(main_bp) app.register_blueprint(file_bp) # 初始化SocketIO init_extensions(app) register_namespaces(app) return app