import os, sys from flask_sqlalchemy import SQLAlchemy from flask_socketio import SocketIO from flask_cors import CORS from db.course_list import CourseList from .services.user_db_service import UserList from .services.my_function import MyFunction from flask_pymongo import PyMongo from qcloud_cos import CosConfig from qcloud_cos import CosS3Client from typing import Optional mongo = PyMongo() socketio = SocketIO(cors_allowed_origins="*",max_payload_size=52428800, async_mode="eventlet") # 不直接传 app cors = CORS() # ===== 你的全局对象 ===== # Backboard from backboardManager import BackBoardManager # 你已有的类 backboard_manager = BackBoardManager() # 用户映射(全局内存结构) userid_recorder = {} # 你代码里有使用 users_list = UserList() uuid2username = {} # 你的映射,或换成数据库访问层 username2uuid = {} course_list = CourseList() user_uuid2UserClass = {} user_uuid2ase_client = {} user_uuid2chatmanager = {} def init_extensions(app): # 第三方 socketio.init_app( app, cors_allowed_origins='*',#app.config["VSCODE_WEB_URL"], ping_timeout=app.config["SOCKETIO_PING_TIMEOUT"], ping_interval=app.config["SOCKETIO_PING_INTERVAL"], max_payload_size=50000000, async_mode="eventlet" ) cors.init_app( app, resources={r"/*": {"origins": "*"}},#app.config["VSCODE_WEB_URL"]}}, supports_credentials=True, ) # 把自定义对象挂到 app.extensions,供各处通过 current_app 访问 app.extensions["users_list"] = users_list app.extensions["course_list"] = course_list app.extensions["backboard_manager"] = backboard_manager app.extensions["uuid2username"] = uuid2username app.extensions["username2uuid"] = username2uuid app.extensions["userid_recorder"] = userid_recorder app.extensions["user_uuid2UserClass"] = user_uuid2UserClass app.extensions["my_function"] = MyFunction() app.extensions["user_uuid2ase_client"] = user_uuid2ase_client app.extensions["user_uuid2chatmanager"] = user_uuid2chatmanager mongo.init_app(app,uri=app.config["MONGO_URI"]) app.extensions["mongo"] = mongo cos_config = CosConfig( Region=app.config["TENCENT_COS"]["region"], SecretId=app.config["TENCENT_COS"]["secret_id"], SecretKey=app.config["TENCENT_COS"]["secret_key"], ) cos_client = CosS3Client(cos_config) app.extensions["cos_client"] = cos_client def register_namespaces(app): """把所有 Socket.IO namespaces 注册到 socketio""" # 延迟导入以避免循环 from .sockets.namespaces import VSCodeNamespace, AgentNamespace socketio.on_namespace(VSCodeNamespace("/vscode")) socketio.on_namespace(AgentNamespace("/agent")) def close_extensions(): # 按“最可能阻塞”的优先级依次停止 try: if user_uuid2chatmanager: for uuid in user_uuid2chatmanager: user_uuid2chatmanager[uuid].disconnect(uuid) except Exception as e: print("Failed to shutdown scheduler: %s", e)