Upload 7 files

Dockerfile

@@ -0,0 +1,38 @@
+# 第一阶段：构建阶段
+FROM python:3.10-slim AS builder
+
+# 设置工作目录
+WORKDIR /app
+
+# 复制依赖文件
+COPY requirements.txt .
+
+# 安装构建依赖
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    libssl-dev \
+    libffi-dev \
+    python3-dev \
+    && apt-get clean \
+    && pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir -r requirements.txt
+
+# 第二阶段：运行阶段
+FROM python:3.10-slim
+
+# 设置工作目录
+WORKDIR /app
+
+# 从构建阶段复制必要的文件
+COPY --from=builder /usr/local/lib/python3.10/site-packages /usr/local/lib/python3.10/site-packages
+COPY --from=builder /usr/local/bin /usr/local/bin
+COPY . .
+
+# 设置环境变量
+ENV PYTHONUNBUFFERED=1
+
+# 暴露 Flask 端口
+EXPOSE 5000
+
+# 运行 Flask 应用
+CMD ["gunicorn", "-b", "0.0.0.0:5000", "app:app"]

app.py

@@ -0,0 +1,251 @@
+from flask import Flask, request, jsonify, Response
+import subprocess
+import sys
+import threading
+import time
+import requests
+from functools import wraps
+import urllib.parse
+import re
+import json
+import base64
+import os
+import glob
+
+app = Flask(__name__)
+
+# 从环境变量中读取配置，如果未设置则使用默认值
+OUTPUT_BASE64 = os.getenv('OUTPUT_BASE64', 'True').lower() == 'true'
+AUTH_TOKEN = os.getenv('AUTH_TOKEN', '')
+# HOST = os.getenv('HOST', '127.0.0.1')
+HOST = os.getenv('HOST', '0.0.0.0')
+PORT = int(os.getenv('PORT', 5000))
+TIMEOUT = int(os.getenv('TIMEOUT', 30))
+REMOVE_PYTHON_CODE_BLOCK = os.getenv('REMOVE_PYTHON_CODE_BLOCK', 'True').lower() == 'true'
+
+# 配置变量，决定是否启用单元测试，默认启用
+ENABLE_UNIT_TESTS = False
+
+def auth_required(f):
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        if AUTH_TOKEN:
+            auth_header = request.headers.get('Authorization')
+            if not auth_header or not auth_header.startswith('Bearer '):
+                return jsonify({'error': 'Authorization header is missing or invalid'}), 401
+            token = auth_header.split('Bearer ')[1]
+            if token != AUTH_TOKEN:
+                return jsonify({'error': 'Invalid token'}), 401
+        return f(*args, **kwargs)
+    return decorated
+
+def extract_code(code):
+    if REMOVE_PYTHON_CODE_BLOCK:
+        match = re.search(r'```python\s*(.*?)\s*```', code, re.DOTALL)
+        if match:
+            return match.group(1)
+    return code
+
+def run_code(language, code, variables, timeout):
+    try:
+        # 根据语言类型选择解释器
+        if language == 'python':
+            # 提取代码中的有效部分
+            code = extract_code(code)
+            # 将变量注入到代码环境中
+            var_definitions = '\n'.join(f"{key} = {value}" for key, value in variables.items())
+            code_with_vars = f"{var_definitions}\n{code}"
+            # 使用subprocess运行代码
+            result = subprocess.run([sys.executable, '-c', code_with_vars], capture_output=True, text=True, timeout=timeout)
+            return result.stdout, result.stderr
+        else:
+            return None, "Unsupported language"
+    except subprocess.TimeoutExpired:
+        return None, "Code execution timed out"
+    except Exception as e:
+        return None, str(e)
+
+def convert_image_to_base64(image_path):
+    with open(image_path, "rb") as image_file:
+        encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
+    return encoded_string
+
+def unescape_string(s):
+    return s.encode('utf-8').decode('unicode_escape')
+
+@app.route('/runcode', methods=['POST', 'GET'])
+@auth_required
+def run_code_endpoint():
+    if request.method == 'POST':
+        data = request.get_json()
+        if not data:
+            return jsonify({'error': 'Invalid JSON'}), 400
+        language = data.get('languageType', 'python')
+        variables = data.get('variables', {})
+        code = data.get('code', '')
+    elif request.method == 'GET':
+        query_string = request.query_string.decode('utf-8')
+        print("Query String:",query_string)
+        # 检查query_string中是否有"+"
+        if '+' in query_string:
+            # query_string的+替换为空格
+            query_string = re.sub(r'\+', ' ', query_string)
+            print("Query String:",query_string)
+
+        # 如果没有+，则直接输出query_string
+        else:
+            print("Query String:",query_string)
+
+
+        # 使用正则表达式提取参数
+        language_match = re.search(r'languageType=("?)([^"&]+)\1', query_string)
+        variables_match = re.search(r'variables=("?)([^"&]+)\1', query_string)
+        code_match = re.search(r'code=("?)([^"&]+)\1', query_string)
+        print("code_match:",code_match)
+        if not language_match or not variables_match or not code_match:
+            return jsonify({'error': 'Invalid parameters'}), 400
+        
+
+        
+        language = urllib.parse.unquote(language_match.group(2))
+        variables = urllib.parse.unquote(variables_match.group(2))
+        code = urllib.parse.unquote(code_match.group(2))
+        print("code:",code)
+        
+        # 处理转义字符
+        language = unescape_string(language)
+        variables = unescape_string(variables)
+        code = unescape_string(code)
+        print("code:",code)
+        
+        # # 使用正则表达式将 + 替换为空格
+        # code = re.sub(r'\+', ' ', code)
+        # print("code:",code)
+
+
+        try:
+            variables = json.loads(variables)
+        except json.JSONDecodeError:
+            return jsonify({'error': 'Invalid variables format'}), 400
+        
+        # 打印解析后的JSON参数
+        print("Parsed JSON Parameters:")
+        print(json.dumps({
+            'languageType': language,
+            'variables': variables,
+            'code': code
+        }, indent=4))
+    else:
+        return jsonify({'error': 'Unsupported HTTP method'}), 405
+
+    # 运行代码
+    stdout, stderr = run_code(language, code, variables, TIMEOUT)
+
+    if stderr:
+        return jsonify({'error': stderr}), 400
+    else:
+        # 查找当前目录中的所有图片文件
+        image_files = glob.glob('*.png') + glob.glob('*.jpg') + glob.glob('*.jpeg') + glob.glob('*.gif')
+        
+        if image_files:
+            image_data = {}
+            for image_path in image_files:
+                encoded_image = convert_image_to_base64(image_path)
+                image_data[image_path] = encoded_image
+                os.remove(image_path)  # 删除生成的图片文件
+
+            if not OUTPUT_BASE64:
+                # 如果 OUTPUT_BASE64 不是 True，则输出原始图片
+                html_content = ''.join([f'<img src="{image_path}" alt="{image_path}" />' for image_path in image_data.keys()])
+                return Response(html_content, mimetype='text/html')
+
+            else:
+                return jsonify({'output': stdout, 'images': image_data}), 200
+        else:
+            return jsonify({'output': stdout}), 200
+
+def run_test_case(test_case):
+    url = f'http://{HOST}:{PORT}/runcode'
+    if test_case['method'] == 'GET':
+        params = {
+            'languageType': test_case['language'],
+            'variables': test_case['variables'],
+            'code': test_case['code']
+        }
+        headers = {'Authorization': f'Bearer {AUTH_TOKEN}'} if AUTH_TOKEN else {}
+        print("Request Params:")
+        print(params)
+
+        response = requests.get(url, params=params, headers=headers)
+    elif test_case['method'] == 'POST':
+        data = {
+            'languageType': test_case['language'],
+            'variables': test_case['variables'],
+            'code': test_case['code']
+        }
+        headers = {'Authorization': f'Bearer {AUTH_TOKEN}'} if AUTH_TOKEN else {}
+        print("Request Data:")
+        print(data)
+
+        response = requests.post(url, json=data, headers=headers)
+    else:
+        raise ValueError("Unsupported HTTP method in test case")
+
+    # 打印返回JSON
+    print("Response JSON:")
+    print(response.json())
+
+def test_run_code_endpoint():
+    test_cases = [
+        {
+            'method': 'GET',
+            'language': 'python',
+            'variables': str({}),
+            'code': "print('Hello, World!')"
+        },
+        {
+            'method': 'GET',
+            'language': '"python"',
+            'variables': str({}),
+            'code': '"print(\'Hello, World!\')"'
+        },
+        {
+            'method': 'GET',
+            'language': 'python',
+            'variables': str({}),
+            'code': "print('Hello%20World')"
+        },
+        {
+            'method': 'POST',
+            'language': 'python',
+            'variables': {},
+            'code': "print(5 ** 11)"
+        },
+        {
+            'method': 'POST',
+            'language': 'python',
+            'variables': {'m': 7, 'n': 4},
+            'code': "print(m * n)"
+        },
+        {
+            'method': 'POST',
+            'language': 'python',
+            'variables': {},
+            'code': "```python\nprint('Hello from code block!')\n```"
+        }
+    ]
+
+    for test_case in test_cases:
+        run_test_case(test_case)
+
+if __name__ == '__main__':
+    # 启动Flask应用
+    threading.Thread(target=app.run, kwargs={'host': HOST, 'port': PORT, 'debug': False}).start()
+
+    # 等待Flask应用启动
+    time.sleep(1)
+
+    # 运行单元测试
+    if ENABLE_UNIT_TESTS:
+        test_run_code_endpoint()
+        print("All tests executed!")

build.py

@@ -0,0 +1,27 @@
+import yaml
+import subprocess
+
+# 读取配置文件，指定编码为 utf-8
+with open('config.yaml', 'r', encoding='utf-8') as file:
+    config = yaml.safe_load(file)
+
+DEPENDENCIES = config['dependencies']
+INTERPRETER_IMAGE = config['interpreter_image']
+
+# 动态生成 requirements.txt
+def generate_requirements(dependencies):
+    with open('requirements.txt', 'w', encoding='utf-8') as f:
+        for dep in dependencies:
+            f.write(f"{dep}\n")
+
+generate_requirements(DEPENDENCIES)
+print("Generated requirements.txt with specified dependencies.")
+
+# 构建 Docker 镜像
+def build_docker_image(image_name):
+    build_command = ["docker", "build", "-t", image_name, "."]
+    subprocess.run(build_command, check=True)
+    print(f"Built Docker image {image_name}")
+
+build_docker_image(INTERPRETER_IMAGE)
+

center.py

@@ -0,0 +1,1392 @@
+# # # # import yaml
+# # # # import subprocess
+# # # # import uuid
+# # # # from flask import Flask, request, jsonify
+# # # # from queue import Queue
+# # # # import requests
+# # # # import threading
+# # # # import time
+# # # # import os
+# # # # import atexit 
+# # # # import base64
+# # # # import signal
+# # # # from sqlalchemy import create_engine, Column, String, LargeBinary, Integer
+# # # # from sqlalchemy.orm import declarative_base, sessionmaker
+# # # # from sqlalchemy.exc import OperationalError, IntegrityError
+# # # # import psycopg2
+
+# # # # app = Flask(__name__)
+
+# # # # # 读取配置文件，指定编码为 utf-8
+# # # # with open('config.yaml', 'r', encoding='utf-8') as file:
+# # # #     config = yaml.safe_load(file)
+
+# # # # DOMAIN = config['domain']
+# # # # INTERPRETER_IMAGE = config['interpreter_image']
+# # # # PORT_START = config['interpreter_port_range']['start']
+# # # # PORT_END = config['interpreter_port_range']['end']
+# # # # DEPENDENCIES = config['dependencies']
+# # # # RESOURCE_LIMITS = config.get('resource_limits', {})
+# # # # MEMORY_LIMIT = RESOURCE_LIMITS.get('memory')
+# # # # CPU_LIMIT = RESOURCE_LIMITS.get('cpus')
+# # # # POSTGRES = config['postgres']
+# # # # TIMEOUT = config.get('timeout', 30)  # 默认超时时间为30秒
+# # # # SCHEDULER_PORT = config['scheduler_port']  # 新增调度中心端口
+
+# # # # # 容器管理
+# # # # containers = []
+# # # # ports = list(range(PORT_START, PORT_END + 1))
+# # # # lock = threading.Lock()
+# # # # request_queue = Queue()
+# # # # result_dict = {}  # 用于存储结果的字典
+# # # # current_container_index = 0
+
+# # # # # 检查 PostgreSQL 容器是否存在
+# # # # POSTGRES_CONTAINER_NAME = "postgres_code_interpreter"
+# # # # POSTGRES_DATA_DIR = os.path.join(os.getcwd(), 'pgdata')
+
+# # # # def is_postgres_container_running():
+# # # #     result = subprocess.run(["docker", "ps", "-a", "--filter", f"name={POSTGRES_CONTAINER_NAME}", "--format", "{{.Names}}"], capture_output=True, text=True)
+# # # #     return POSTGRES_CONTAINER_NAME in result.stdout.strip().split('\n')
+
+# # # # def wait_for_postgres_ready():
+# # # #     retry_attempts = 10
+# # # #     retry_delay = 5  # seconds
+# # # #     for attempt in range(retry_attempts):
+# # # #         try:
+# # # #             conn = psycopg2.connect(
+# # # #                 dbname=POSTGRES['db'],
+# # # #                 user=POSTGRES['user'],
+# # # #                 password=POSTGRES['password'],
+# # # #                 host=POSTGRES['host'],
+# # # #                 port=POSTGRES['port']
+# # # #             )
+# # # #             conn.close()
+# # # #             print("PostgreSQL is ready")
+# # # #             return True
+# # # #         except psycopg2.OperationalError as e:
+# # # #             print(f"Waiting for PostgreSQL to be ready: {e}")
+# # # #             time.sleep(retry_delay)
+# # # #     return False
+
+# # # # if not is_postgres_container_running():
+# # # #     if not os.path.exists(POSTGRES_DATA_DIR):
+# # # #         os.makedirs(POSTGRES_DATA_DIR)
+
+# # # #     subprocess.run([
+# # # #         "docker", "run", "--name", POSTGRES_CONTAINER_NAME, "-d",
+# # # #         "-e", f"POSTGRES_USER={POSTGRES['user']}",
+# # # #         "-e", f"POSTGRES_PASSWORD={POSTGRES['password']}",
+# # # #         "-e", f"POSTGRES_DB={POSTGRES['db']}",
+# # # #         "-v", f"{POSTGRES_DATA_DIR}:/var/lib/postgresql/data",
+# # # #         "-p", f"{POSTGRES['port']}:5432",
+# # # #         "postgres"
+# # # #     ], check=True)
+
+# # # #     if not wait_for_postgres_ready():
+# # # #         print("Failed to connect to PostgreSQL after multiple attempts.")
+# # # #         exit(1)
+
+# # # # # PostgreSQL 配置
+# # # # DATABASE_URL = f"postgresql://{POSTGRES['user']}:{POSTGRES['password']}@{POSTGRES['host']}:{POSTGRES['port']}/{POSTGRES['db']}"
+# # # # engine = None
+# # # # Base = declarative_base()
+# # # # Session = None
+# # # # session = None
+
+# # # # def init_db():
+# # # #     global engine, Session, session
+# # # #     retry_attempts = 5
+# # # #     retry_delay = 5  # seconds
+# # # #     for attempt in range(retry_attempts):
+# # # #         try:
+# # # #             engine = create_engine(DATABASE_URL)
+# # # #             Session = sessionmaker(bind=engine)
+# # # #             session = Session()
+# # # #             Base.metadata.create_all(engine)  # 确保数据库模式已创建
+# # # #             print("Database connection successful")
+# # # #             break
+# # # #         except OperationalError as e:
+# # # #             print(f"Database connection failed: {e}")
+# # # #             if attempt < retry_attempts - 1:
+# # # #                 print(f"Retrying in {retry_delay} seconds...")
+# # # #                 time.sleep(retry_delay)
+# # # #             else:
+# # # #                 print("Failed to connect to the database after multiple attempts.")
+# # # #                 raise
+
+# # # # class Image(Base):
+# # # #     __tablename__ = 'images'
+# # # #     id = Column(Integer, primary_key=True, autoincrement=True)
+# # # #     filename = Column(String, unique=True, nullable=False)
+# # # #     data = Column(LargeBinary, nullable=False)
+
+# # # # init_db()
+
+# # # # # 启动代码解释器容器
+# # # # def start_container(port):
+# # # #     container_name = f"code_interpreter_docker_{uuid.uuid4()}"
+# # # #     run_command = [
+# # # #         "docker", "run", "--name", container_name, "-d", "-p", f"{port}:5000"
+# # # #     ]
+# # # #     if MEMORY_LIMIT:
+# # # #         run_command.extend(["--memory", MEMORY_LIMIT])
+# # # #     if CPU_LIMIT:
+# # # #         run_command.extend(["--cpus", CPU_LIMIT])
+# # # #     run_command.append(INTERPRETER_IMAGE)
+    
+# # # #     subprocess.run(run_command, check=True)
+# # # #     print(f"Started container {container_name} on port {port}")
+# # # #     return container_name, port
+
+# # # # def stop_container(container_name):
+# # # #     subprocess.run(["docker", "stop", container_name])
+# # # #     subprocess.run(["docker", "rm", container_name])
+# # # #     print(f"Stopped and removed container {container_name}")
+
+# # # # def start_containers():
+# # # #     for port in ports:
+# # # #         container_name, port = start_container(port)
+# # # #         containers.append((container_name, port))
+
+# # # # def stop_containers():
+# # # #     for container_name, _ in containers:
+# # # #         stop_container(container_name)
+
+# # # # @app.route('/runcode', methods=['POST', 'GET'])
+# # # # def run_code():
+# # # #     if request.method == 'POST':
+# # # #         data = request.get_json()
+# # # #         if not data:
+# # # #             return jsonify({'error': 'Invalid JSON'}), 400
+# # # #     elif request.method == 'GET':
+# # # #         query_string = request.query_string.decode('utf-8')
+# # # #         data = {'query_string': query_string}
+# # # #     else:
+# # # #         return jsonify({'error': 'Invalid request method'}), 405
+
+# # # #     request_id = str(uuid.uuid4())  # 生成一个唯一的请求ID
+# # # #     request_queue.put((request_id, data))
+
+# # # #     start_time = time.time()
+# # # #     while time.time() - start_time < 10:  # 最多等待10秒
+# # # #         if request_id in result_dict:
+# # # #             output = result_dict.pop(request_id)
+# # # #             # 处理返回结果，替换 base64 数据为链接
+# # # #             if 'images' in output:
+# # # #                 try:
+# # # #                     output = process_images(output)
+# # # #                 except Exception as e:
+# # # #                     return jsonify({'error': str(e)}), 500
+# # # #             return jsonify(output), 200
+
+# # # #         time.sleep(0.1)
+
+# # # #     return jsonify({'error': 'Request timed out'}), 504
+
+# # # # def process_images(output):
+# # # #     images = output['images']
+# # # #     for filename, base64_data in images.items():
+# # # #         image_data = base64.b64decode(base64_data)
+# # # #         unique_filename = f"{uuid.uuid4()}_{filename}"
+# # # #         image_record = Image(filename=unique_filename, data=image_data)
+# # # #         try:
+# # # #             session.add(image_record)
+# # # #             session.commit()
+# # # #         except IntegrityError as e:
+# # # #             session.rollback()
+# # # #             print(f"Database error: {e}")
+# # # #             continue
+# # # #         if DOMAIN:
+# # # #             images[filename] = f"https://{DOMAIN}/image/{unique_filename}"
+# # # #         else:
+# # # #             images[filename] = f"http://127.0.0.1:{SCHEDULER_PORT}/image/{unique_filename}"
+# # # #     return output
+
+
+# # # # @app.route('/image/<filename>', methods=['GET'])
+# # # # def serve_image(filename):
+# # # #     image_record = session.query(Image).filter_by(filename=filename).first()
+# # # #     if image_record:
+# # # #         return image_record.data, 200, {'Content-Type': 'image/png'}
+# # # #     return jsonify({'error': 'Image not found'}), 404
+
+# # # # def handle_requests():
+# # # #     global current_container_index
+# # # #     while True:
+# # # #         request_id, data = request_queue.get()
+# # # #         if data is None:
+# # # #             break
+
+# # # #         with lock:
+# # # #             container_name, port = containers[current_container_index]
+# # # #             current_container_index = (current_container_index + 1) % len(containers)
+
+# # # #         try:
+# # # #             if 'query_string' in data:
+# # # #                 # GET 请求处理
+# # # #                 response = requests.get(f"http://localhost:{port}/runcode?{data['query_string']}", timeout=TIMEOUT)
+# # # #             else:
+# # # #                 # POST 请求处理
+# # # #                 response = requests.post(f"http://localhost:{port}/runcode", json=data, timeout=TIMEOUT)
+# # # #             output = response.json()
+# # # #         except requests.exceptions.Timeout:
+# # # #             output = {'error': 'Code execution timed out'}
+# # # #             reset_container(container_name, port)
+# # # #         except Exception as e:
+# # # #             output = {'error': str(e)}
+        
+# # # #         result_dict[request_id] = output  # 将结果放入结果字典
+
+# # # #         request_queue.task_done()
+# # # #         print(f"Finished processing request {request_id}: {output}")
+
+# # # # def reset_container(container_name, port):
+# # # #     print(f"Resetting container {container_name}")
+# # # #     stop_container(container_name)
+# # # #     new_container_name, _ = start_container(port)
+# # # #     with lock:
+# # # #         for i, (name, p) in enumerate(containers):
+# # # #             if name == container_name:
+# # # #                 containers[i] = (new_container_name, port)
+# # # #                 break
+
+# # # # def signal_handler(signal, frame):
+# # # #     print('Stopping containers and exiting program...')
+# # # #     stop_containers()
+# # # #     if session:
+# # # #         session.close()
+# # # #     engine.dispose()
+# # # #     os._exit(0)
+
+# # # # signal.signal(signal.SIGINT, signal_handler)
+
+# # # # worker_thread = threading.Thread(target=handle_requests)
+# # # # worker_thread.start()
+
+# # # # start_containers()
+# # # # atexit.register(stop_containers)
+
+# # # # if __name__ == '__main__':
+# # # #     app.run(port=SCHEDULER_PORT, threaded=True)  # 使用调度中心端口
+
+
+
+# # # import yaml
+# # # import subprocess
+# # # import uuid
+# # # from flask import Flask, request, jsonify
+# # # from queue import Queue
+# # # import requests
+# # # import threading
+# # # import time
+# # # import os
+# # # import atexit
+# # # import base64
+# # # import signal
+# # # from sqlalchemy import create_engine, Column, String, LargeBinary, Integer
+# # # from sqlalchemy.orm import declarative_base, sessionmaker
+# # # from sqlalchemy.exc import OperationalError, IntegrityError
+# # # import psycopg2
+
+# # # app = Flask(__name__)
+
+# # # # 读取配置文件，指定编码为 utf-8
+# # # with open('config.yaml', 'r', encoding='utf-8') as file:
+# # #     config = yaml.safe_load(file)
+
+# # # DOMAIN = config['domain']
+# # # INTERPRETER_IMAGE = config['interpreter_image']
+# # # PORT_START = config['interpreter_port_range']['start']
+# # # PORT_END = config['interpreter_port_range']['end']
+# # # DEPENDENCIES = config['dependencies']
+# # # RESOURCE_LIMITS = config.get('resource_limits', {})
+# # # MEMORY_LIMIT = RESOURCE_LIMITS.get('memory')
+# # # CPU_LIMIT = RESOURCE_LIMITS.get('cpus')
+# # # POSTGRES = config['postgres']
+# # # TIMEOUT = config.get('timeout', 30)  # 默认超时时间为30秒
+# # # SCHEDULER_PORT = config['scheduler_port']  # 新增调度中心端口
+
+# # # # 容器管理
+# # # containers = []
+# # # ports = list(range(PORT_START, PORT_END + 1))
+# # # lock = threading.Lock()
+# # # request_queue = Queue()
+# # # result_dict = {}  # 用于存储结果的字典
+# # # current_container_index = 0
+
+# # # # 检查 PostgreSQL 容器是否存在
+# # # POSTGRES_CONTAINER_NAME = "postgres_code_interpreter"
+# # # POSTGRES_DATA_DIR = os.path.join(os.getcwd(), 'pgdata')
+
+# # # def is_postgres_container_running():
+# # #     result = subprocess.run(["docker", "ps", "-a", "--filter", f"name={POSTGRES_CONTAINER_NAME}", "--format", "{{.Names}}"], capture_output=True, text=True)
+# # #     return POSTGRES_CONTAINER_NAME in result.stdout.strip().split('\n')
+
+# # # def wait_for_postgres_ready():
+# # #     retry_attempts = 10
+# # #     retry_delay = 5  # seconds
+# # #     for attempt in range(retry_attempts):
+# # #         try:
+# # #             conn = psycopg2.connect(
+# # #                 dbname=POSTGRES['db'],
+# # #                 user=POSTGRES['user'],
+# # #                 password=POSTGRES['password'],
+# # #                 host=POSTGRES['host'],
+# # #                 port=POSTGRES['port']
+# # #             )
+# # #             conn.close()
+# # #             print("PostgreSQL is ready")
+# # #             return True
+# # #         except psycopg2.OperationalError as e:
+# # #             print(f"Waiting for PostgreSQL to be ready: {e}")
+# # #             time.sleep(retry_delay)
+# # #     return False
+
+# # # if not is_postgres_container_running():
+# # #     if not os.path.exists(POSTGRES_DATA_DIR):
+# # #         os.makedirs(POSTGRES_DATA_DIR)
+
+# # #     subprocess.run([
+# # #         "docker", "run", "--name", POSTGRES_CONTAINER_NAME, "-d",
+# # #         "-e", f"POSTGRES_USER={POSTGRES['user']}",
+# # #         "-e", f"POSTGRES_PASSWORD={POSTGRES['password']}",
+# # #         "-e", f"POSTGRES_DB={POSTGRES['db']}",
+# # #         "-v", f"{POSTGRES_DATA_DIR}:/var/lib/postgresql/data",
+# # #         "-p", f"{POSTGRES['port']}:5432",
+# # #         "postgres"
+# # #     ], check=True)
+
+# # #     if not wait_for_postgres_ready():
+# # #         print("Failed to connect to PostgreSQL after multiple attempts.")
+# # #         exit(1)
+
+# # # # PostgreSQL 配置
+# # # DATABASE_URL = f"postgresql://{POSTGRES['user']}:{POSTGRES['password']}@{POSTGRES['host']}:{POSTGRES['port']}/{POSTGRES['db']}"
+# # # engine = None
+# # # Base = declarative_base()
+# # # Session = None
+# # # session = None
+
+# # # def init_db():
+# # #     global engine, Session, session
+# # #     retry_attempts = 5
+# # #     retry_delay = 5  # seconds
+# # #     for attempt in range(retry_attempts):
+# # #         try:
+# # #             engine = create_engine(DATABASE_URL)
+# # #             Session = sessionmaker(bind=engine)
+# # #             session = Session()
+# # #             Base.metadata.create_all(engine)  # 确保数据库模式已创建
+# # #             print("Database connection successful")
+# # #             break
+# # #         except OperationalError as e:
+# # #             print(f"Database connection failed: {e}")
+# # #             if attempt < retry_attempts - 1:
+# # #                 print(f"Retrying in {retry_delay} seconds...")
+# # #                 time.sleep(retry_delay)
+# # #             else:
+# # #                 print("Failed to connect to the database after multiple attempts.")
+# # #                 raise
+
+# # # class Image(Base):
+# # #     __tablename__ = 'images'
+# # #     id = Column(Integer, primary_key=True, autoincrement=True)
+# # #     filename = Column(String, unique=True, nullable=False)
+# # #     data = Column(LargeBinary, nullable=False)
+
+# # # init_db()
+
+# # # # 启动代码解释器容器
+# # # def start_container(port):
+# # #     container_name = f"code_interpreter_docker_{uuid.uuid4()}"
+# # #     run_command = [
+# # #         "docker", "run", "--name", container_name, "-d", "-p", f"{port}:5000"
+# # #     ]
+# # #     if MEMORY_LIMIT:
+# # #         run_command.extend(["--memory", MEMORY_LIMIT])
+# # #     if CPU_LIMIT:
+# # #         run_command.extend(["--cpus", CPU_LIMIT])
+# # #     run_command.append(INTERPRETER_IMAGE)
+    
+# # #     subprocess.run(run_command, check=True)
+# # #     print(f"Started container {container_name} on port {port}")
+# # #     return container_name, port
+
+# # # def stop_container(container_name):
+# # #     subprocess.run(["docker", "stop", container_name])
+# # #     subprocess.run(["docker", "rm", container_name])
+# # #     print(f"Stopped and removed container {container_name}")
+
+# # # def start_containers():
+# # #     for port in ports:
+# # #         container_name, port = start_container(port)
+# # #         containers.append((container_name, port))
+
+# # # def stop_containers():
+# # #     for container_name, _ in containers:
+# # #         stop_container(container_name)
+
+# # # @app.route('/runcode', methods=['POST', 'GET'])
+# # # def run_code():
+# # #     if request.method == 'POST':
+# # #         data = request.get_json()
+# # #         if not data:
+# # #             return jsonify({'error': 'Invalid JSON'}), 400
+# # #     elif request.method == 'GET':
+# # #         query_string = request.query_string.decode('utf-8')
+# # #         data = {'query_string': query_string}
+# # #     else:
+# # #         return jsonify({'error': 'Invalid request method'}), 405
+
+# # #     request_id = str(uuid.uuid4())  # 生成一个唯一的请求ID
+# # #     request_queue.put((request_id, data))
+
+# # #     start_time = time.time()
+# # #     while time.time() - start_time < 10:  # 最多等待10秒
+# # #         if request_id in result_dict:
+# # #             output = result_dict.pop(request_id)
+# # #             # 处理返回结果，替换 base64 数据为链接
+# # #             if 'images' in output:
+# # #                 try:
+# # #                     output = process_images(output)
+# # #                 except Exception as e:
+# # #                     return jsonify({'error': str(e)}), 500
+# # #             return jsonify(output), 200
+
+# # #         time.sleep(0.1)
+
+# # #     return jsonify({'error': 'Request timed out'}), 504
+
+# # # def process_images(output):
+# # #     images = output['images']
+# # #     for filename, base64_data in images.items():
+# # #         image_data = base64.b64decode(base64_data)
+# # #         unique_filename = f"{uuid.uuid4()}_{filename}"
+# # #         image_record = Image(filename=unique_filename, data=image_data)
+# # #         try:
+# # #             session.add(image_record)
+# # #             session.commit()
+# # #         except IntegrityError as e:
+# # #             session.rollback()
+# # #             print(f"Database error: {e}")
+# # #             continue
+# # #         if DOMAIN:
+# # #             images[filename] = f"https://{DOMAIN}/image/{unique_filename}"
+# # #         else:
+# # #             images[filename] = f"http://127.0.0.1:{SCHEDULER_PORT}/image/{unique_filename}"
+# # #     return output
+
+
+# # # @app.route('/image/<filename>', methods=['GET'])
+# # # def serve_image(filename):
+# # #     image_record = session.query(Image).filter_by(filename=filename).first()
+# # #     if image_record:
+# # #         return image_record.data, 200, {'Content-Type': 'image/png'}
+# # #     return jsonify({'error': 'Image not found'}), 404
+
+# # # def handle_requests():
+# # #     global current_container_index
+# # #     while True:
+# # #         request_id, data = request_queue.get()
+# # #         if data is None:
+# # #             break
+
+# # #         with lock:
+# # #             container_name, port = containers[current_container_index]
+# # #             current_container_index = (current_container_index + 1) % len(containers)
+
+# # #         try:
+# # #             if 'query_string' in data:
+# # #                 # GET 请求处理
+# # #                 response = requests.get(f"http://localhost:{port}/runcode?{data['query_string']}", timeout=TIMEOUT)
+# # #             else:
+# # #                 # POST 请求处理
+# # #                 response = requests.post(f"http://localhost:{port}/runcode", json=data, timeout=TIMEOUT)
+# # #             output = response.json()
+# # #         except requests.exceptions.Timeout:
+# # #             output = {'error': 'Code execution timed out'}
+# # #             reset_container(container_name, port)
+# # #         except Exception as e:
+# # #             output = {'error': str(e)}
+        
+# # #         result_dict[request_id] = output  # 将结果放入结果字典
+
+# # #         request_queue.task_done()
+# # #         print(f"Finished processing request {request_id}: {output}")
+
+# # #         # Reset container after each request
+# # #         reset_container(container_name, port)
+
+# # # def reset_container(container_name, port):
+# # #     print(f"Resetting container {container_name}")
+# # #     stop_container(container_name)
+# # #     new_container_name, _ = start_container(port)
+# # #     with lock:
+# # #         for i, (name, p) in enumerate(containers):
+# # #             if name == container_name:
+# # #                 containers[i] = (new_container_name, port)
+# # #                 break
+
+# # # def signal_handler(signal, frame):
+# # #     print('Stopping containers and exiting program...')
+# # #     stop_containers()
+# # #     if session:
+# # #         session.close()
+# # #     engine.dispose()
+# # #     os._exit(0)
+
+# # # signal.signal(signal.SIGINT, signal_handler)
+
+# # # worker_thread = threading.Thread(target=handle_requests)
+# # # worker_thread.start()
+
+# # # start_containers()
+# # # atexit.register(stop_containers)
+
+# # # if __name__ == '__main__':
+# # #     app.run(port=SCHEDULER_PORT, threaded=True)  # 使用调度中心端口
+
+
+
+# # import yaml
+# # import subprocess
+# # import uuid
+# # from flask import Flask, request, jsonify
+# # from queue import Queue
+# # import requests
+# # import threading
+# # import time
+# # import os
+# # import atexit
+# # import base64
+# # import signal
+# # from sqlalchemy import create_engine, Column, String, LargeBinary, Integer
+# # from sqlalchemy.orm import declarative_base, sessionmaker
+# # from sqlalchemy.exc import OperationalError, IntegrityError
+# # import psycopg2
+
+# # app = Flask(__name__)
+
+# # # 读取配置文件，指定编码为 utf-8
+# # with open('config.yaml', 'r', encoding='utf-8') as file:
+# #     config = yaml.safe_load(file)
+
+# # DOMAIN = config['domain']
+# # INTERPRETER_IMAGE = config['interpreter_image']
+# # PORT_START = config['interpreter_port_range']['start']
+# # PORT_END = config['interpreter_port_range']['end']
+# # DEPENDENCIES = config['dependencies']
+# # RESOURCE_LIMITS = config.get('resource_limits', {})
+# # MEMORY_LIMIT = RESOURCE_LIMITS.get('memory')
+# # CPU_LIMIT = RESOURCE_LIMITS.get('cpus')
+# # POSTGRES = config['postgres']
+# # TIMEOUT = config.get('timeout', 30)  # 默认超时时间为30秒
+# # TIMEOUT_SECONDS = config.get('timeout_seconds', 60)  # 从配置中读取超时时间
+# # SCHEDULER_PORT = config['scheduler_port']  # 新增调度中心端口
+
+# # # 容器管理
+# # containers = []
+# # ports = list(range(PORT_START, PORT_END + 1))
+# # lock = threading.Lock()
+# # request_queue = Queue()
+# # result_dict = {}  # 用于存储结果的字典
+# # current_container_index = 0
+# # semaphore = threading.Semaphore(len(ports))  # 控制并发请求的信号量
+
+# # # 检查 PostgreSQL 容器是否存在
+# # POSTGRES_CONTAINER_NAME = "postgres_code_interpreter"
+# # POSTGRES_DATA_DIR = os.path.join(os.getcwd(), 'pgdata')
+
+# # def is_postgres_container_running():
+# #     result = subprocess.run(["docker", "ps", "-a", "--filter", f"name={POSTGRES_CONTAINER_NAME}", "--format", "{{.Names}}"], capture_output=True, text=True)
+# #     return POSTGRES_CONTAINER_NAME in result.stdout.strip().split('\n')
+
+# # def wait_for_postgres_ready():
+# #     retry_attempts = 10
+# #     retry_delay = 5  # seconds
+# #     for attempt in range(retry_attempts):
+# #         try:
+# #             conn = psycopg2.connect(
+# #                 dbname=POSTGRES['db'],
+# #                 user=POSTGRES['user'],
+# #                 password=POSTGRES['password'],
+# #                 host=POSTGRES['host'],
+# #                 port=POSTGRES['port']
+# #             )
+# #             conn.close()
+# #             print("PostgreSQL is ready")
+# #             return True
+# #         except psycopg2.OperationalError as e:
+# #             print(f"Waiting for PostgreSQL to be ready: {e}")
+# #             time.sleep(retry_delay)
+# #     return False
+
+# # if not is_postgres_container_running():
+# #     if not os.path.exists(POSTGRES_DATA_DIR):
+# #         os.makedirs(POSTGRES_DATA_DIR)
+
+# #     subprocess.run([
+# #         "docker", "run", "--name", POSTGRES_CONTAINER_NAME, "-d",
+# #         "-e", f"POSTGRES_USER={POSTGRES['user']}",
+# #         "-e", f"POSTGRES_PASSWORD={POSTGRES['password']}",
+# #         "-e", f"POSTGRES_DB={POSTGRES['db']}",
+# #         "-v", f"{POSTGRES_DATA_DIR}:/var/lib/postgresql/data",
+# #         "-p", f"{POSTGRES['port']}:5432",
+# #         "postgres"
+# #     ], check=True)
+
+# #     if not wait_for_postgres_ready():
+# #         print("Failed to connect to PostgreSQL after multiple attempts.")
+# #         exit(1)
+
+# # # PostgreSQL 配置
+# # DATABASE_URL = f"postgresql://{POSTGRES['user']}:{POSTGRES['password']}@{POSTGRES['host']}:{POSTGRES['port']}/{POSTGRES['db']}"
+# # engine = None
+# # Base = declarative_base()
+# # Session = None
+# # session = None
+
+# # def init_db():
+# #     global engine, Session, session
+# #     retry_attempts = 5
+# #     retry_delay = 5  # seconds
+# #     for attempt in range(retry_attempts):
+# #         try:
+# #             engine = create_engine(DATABASE_URL)
+# #             Session = sessionmaker(bind=engine)
+# #             session = Session()
+# #             Base.metadata.create_all(engine)  # 确保数据库模式已创建
+# #             print("Database connection successful")
+# #             break
+# #         except OperationalError as e:
+# #             print(f"Database connection failed: {e}")
+# #             if attempt < retry_attempts - 1:
+# #                 print(f"Retrying in {retry_delay} seconds...")
+# #                 time.sleep(retry_delay)
+# #             else:
+# #                 print("Failed to connect to the database after multiple attempts.")
+# #                 raise
+
+# # class Image(Base):
+# #     __tablename__ = 'images'
+# #     id = Column(Integer, primary_key=True, autoincrement=True)
+# #     filename = Column(String, unique=True, nullable=False)
+# #     data = Column(LargeBinary, nullable=False)
+
+# # init_db()
+
+# # # 启动代码解释器容器
+# # def start_container(port):
+# #     container_name = f"code_interpreter_docker_{uuid.uuid4()}"
+# #     run_command = [
+# #         "docker", "run", "--name", container_name, "-d", "-p", f"{port}:5000"
+# #     ]
+# #     if MEMORY_LIMIT:
+# #         run_command.extend(["--memory", MEMORY_LIMIT])
+# #     if CPU_LIMIT:
+# #         run_command.extend(["--cpus", CPU_LIMIT])
+# #     run_command.append(INTERPRETER_IMAGE)
+    
+# #     subprocess.run(run_command, check=True)
+# #     print(f"Started container {container_name} on port {port}")
+# #     return container_name, port
+
+# # def stop_container(container_name):
+# #     subprocess.run(["docker", "stop", container_name])
+# #     subprocess.run(["docker", "rm", container_name])
+# #     print(f"Stopped and removed container {container_name}")
+
+# # def start_containers():
+# #     for port in ports:
+# #         container_name, port = start_container(port)
+# #         containers.append((container_name, port))
+
+# # def stop_containers():
+# #     for container_name, _ in containers:
+# #         stop_container(container_name)
+
+# # @app.route('/runcode', methods=['POST', 'GET'])
+# # def run_code():
+# #     if request.method == 'POST':
+# #         data = request.get_json()
+# #         if not data:
+# #             return jsonify({'error': 'Invalid JSON'}), 400
+# #     elif request.method == 'GET':
+# #         query_string = request.query_string.decode('utf-8')
+# #         data = {'query_string': query_string}
+# #     else:
+# #         return jsonify({'error': 'Invalid request method'}), 405
+
+# #     request_id = str(uuid.uuid4())  # 生成一个唯一的请求ID
+# #     request_queue.put((request_id, data))
+
+# #     start_time = time.time()
+# #     while time.time() - start_time < 10:  # 最多等待10秒
+# #         if request_id in result_dict:
+# #             output = result_dict.pop(request_id)
+# #             # 处理返回结果，替换 base64 数据为链接
+# #             if 'images' in output:
+# #                 try:
+# #                     output = process_images(output)
+# #                 except Exception as e:
+# #                     return jsonify({'error': str(e)}), 500
+# #             return jsonify(output), 200
+
+# #         time.sleep(0.1)
+
+# #     return jsonify({'error': 'Request timed out'}), 504
+
+# # def process_images(output):
+# #     images = output['images']
+# #     for filename, base64_data in images.items():
+# #         image_data = base64.b64decode(base64_data)
+# #         unique_filename = f"{uuid.uuid4()}_{filename}"
+# #         image_record = Image(filename=unique_filename, data=image_data)
+# #         try:
+# #             session.add(image_record)
+# #             session.commit()
+# #         except IntegrityError as e:
+# #             session.rollback()
+# #             print(f"Database error: {e}")
+# #             continue
+# #         if DOMAIN:
+# #             images[filename] = f"https://{DOMAIN}/image/{unique_filename}"
+# #         else:
+# #             images[filename] = f"http://127.0.0.1:{SCHEDULER_PORT}/image/{unique_filename}"
+# #     return output
+
+# # @app.route('/image/<filename>', methods=['GET'])
+# # def serve_image(filename):
+# #     image_record = session.query(Image).filter_by(filename=filename).first()
+# #     if image_record:
+# #         return image_record.data, 200, {'Content-Type': 'image/png'}
+# #     return jsonify({'error': 'Image not found'}), 404
+
+# # def handle_requests():
+# #     global current_container_index
+# #     while True:
+# #         request_id, data = request_queue.get()
+# #         if data is None:
+# #             break
+
+# #         semaphore.acquire()  # 获取信号量，确保不超过并发限制
+
+# #         with lock:
+# #             container_name, port = containers[current_container_index]
+# #             current_container_index = (current_container_index + 1) % len(containers)
+
+# #         try:
+# #             if 'query_string' in data:
+# #                 response = requests.get(f"http://localhost:{port}/runcode?{data['query_string']}", timeout=TIMEOUT_SECONDS)
+# #             else:
+# #                 response = requests.post(f"http://localhost:{port}/runcode", json=data, timeout=TIMEOUT_SECONDS)
+# #             output = response.json()
+# #         except requests.exceptions.Timeout:
+# #             output = {'error': 'Code execution timed out'}
+# #         except Exception as e:
+# #             output = {'error': str(e)}
+        
+# #         result_dict[request_id] = output  # 将结果放入结果字典
+
+# #         request_queue.task_done()
+# #         print(f"Finished processing request {request_id}: {output}")
+
+# #         # 异步重置容器
+# #         reset_container(container_name, port)
+
+# # def reset_container(container_name, port):
+# #     threading.Thread(target=_reset_container, args=(container_name, port)).start()
+
+# # def _reset_container(container_name, port):
+# #     print(f"Resetting container {container_name}")
+# #     stop_container(container_name)
+# #     new_container_name, _ = start_container(port)
+# #     with lock:
+# #         for i, (name, p) in enumerate(containers):
+# #             if name == container_name:
+# #                 containers[i] = (new_container_name, port)
+# #                 break
+# #     semaphore.release()  # 释放信号量
+
+# # def signal_handler(signal, frame):
+# #     print('Stopping containers and exiting program...')
+# #     stop_containers()
+# #     if session:
+# #         session.close()
+# #     engine.dispose()
+# #     os._exit(0)
+
+# # signal.signal(signal.SIGINT, signal_handler)
+
+# # worker_thread = threading.Thread(target=handle_requests)
+# # worker_thread.start()
+
+# # start_containers()
+# # atexit.register(stop_containers)
+
+# # if __name__ == '__main__':
+# #     app.run(port=SCHEDULER_PORT, threaded=True)  # 使用调度中心端口
+
+
+
+# import yaml
+# import subprocess
+# import uuid
+# from flask import Flask, request, jsonify
+# from queue import Queue
+# import requests
+# import threading
+# import time
+# import os
+# import atexit
+# import base64
+# import signal
+# from sqlalchemy import create_engine, Column, String, LargeBinary, Integer
+# from sqlalchemy.orm import declarative_base, sessionmaker
+# from sqlalchemy.exc import OperationalError, IntegrityError
+# import psycopg2
+
+# app = Flask(__name__)
+
+# # 读取配置文件，指定编码为 utf-8
+# with open('config.yaml', 'r', encoding='utf-8') as file:
+#     config = yaml.safe_load(file)
+
+# DOMAIN = config['domain']
+# INTERPRETER_IMAGE = config['interpreter_image']
+# PORT_START = config['interpreter_port_range']['start']
+# PORT_END = config['interpreter_port_range']['end']
+# DEPENDENCIES = config['dependencies']
+# RESOURCE_LIMITS = config.get('resource_limits', {})
+# MEMORY_LIMIT = RESOURCE_LIMITS.get('memory')
+# CPU_LIMIT = RESOURCE_LIMITS.get('cpus')
+# POSTGRES = config['postgres']
+# TIMEOUT = config.get('timeout', 30)  # 默认超时时间为30秒
+# TIMEOUT_SECONDS = config.get('timeout_seconds', 60)  # 从配置中读取超时时间
+# SCHEDULER_PORT = config['scheduler_port']  # 新增调度中心端口
+
+# # 容器管理
+# containers = []
+# ports = list(range(PORT_START, PORT_END + 1))
+# lock = threading.Lock()
+# request_queue = Queue()
+# result_dict = {}  # 用于存储结果的字典
+# current_container_index = 0
+# semaphore = threading.Semaphore(len(ports))  # 控制并发请求的信号量
+
+# # 检查 PostgreSQL 容器是否存在
+# POSTGRES_CONTAINER_NAME = "postgres_code_interpreter"
+# POSTGRES_DATA_DIR = os.path.join(os.getcwd(), 'pgdata')
+
+# def is_postgres_container_running():
+#     result = subprocess.run(["docker", "ps", "-a", "--filter", f"name={POSTGRES_CONTAINER_NAME}", "--format", "{{.Names}}"], capture_output=True, text=True)
+#     return POSTGRES_CONTAINER_NAME in result.stdout.strip().split('\n')
+
+# def wait_for_postgres_ready():
+#     retry_attempts = 10
+#     retry_delay = 5  # seconds
+#     for attempt in range(retry_attempts):
+#         try:
+#             conn = psycopg2.connect(
+#                 dbname=POSTGRES['db'],
+#                 user=POSTGRES['user'],
+#                 password=POSTGRES['password'],
+#                 host=POSTGRES['host'],
+#                 port=POSTGRES['port']
+#             )
+#             conn.close()
+#             print("PostgreSQL is ready")
+#             return True
+#         except psycopg2.OperationalError as e:
+#             print(f"Waiting for PostgreSQL to be ready: {e}")
+#             time.sleep(retry_delay)
+#     return False
+
+# if not is_postgres_container_running():
+#     if not os.path.exists(POSTGRES_DATA_DIR):
+#         os.makedirs(POSTGRES_DATA_DIR)
+
+#     subprocess.run([
+#         "docker", "run", "--name", POSTGRES_CONTAINER_NAME, "-d",
+#         "-e", f"POSTGRES_USER={POSTGRES['user']}",
+#         "-e", f"POSTGRES_PASSWORD={POSTGRES['password']}",
+#         "-e", f"POSTGRES_DB={POSTGRES['db']}",
+#         "-v", f"{POSTGRES_DATA_DIR}:/var/lib/postgresql/data",
+#         "-p", f"{POSTGRES['port']}:5432",
+#         "postgres"
+#     ], check=True)
+
+#     if not wait_for_postgres_ready():
+#         print("Failed to connect to PostgreSQL after multiple attempts.")
+#         exit(1)
+
+# # PostgreSQL 配置
+# DATABASE_URL = f"postgresql://{POSTGRES['user']}:{POSTGRES['password']}@{POSTGRES['host']}:{POSTGRES['port']}/{POSTGRES['db']}"
+# engine = None
+# Base = declarative_base()
+# Session = None
+# session = None
+
+# def init_db():
+#     global engine, Session, session
+#     retry_attempts = 5
+#     retry_delay = 5  # seconds
+#     for attempt in range(retry_attempts):
+#         try:
+#             engine = create_engine(DATABASE_URL)
+#             Session = sessionmaker(bind=engine)
+#             session = Session()
+#             Base.metadata.create_all(engine)  # 确保数据库模式已创建
+#             print("Database connection successful")
+#             break
+#         except OperationalError as e:
+#             print(f"Database connection failed: {e}")
+#             if attempt < retry_attempts - 1:
+#                 print(f"Retrying in {retry_delay} seconds...")
+#                 time.sleep(retry_delay)
+#             else:
+#                 print("Failed to connect to the database after multiple attempts.")
+#                 raise
+
+# class Image(Base):
+#     __tablename__ = 'images'
+#     id = Column(Integer, primary_key=True, autoincrement=True)
+#     filename = Column(String, unique=True, nullable=False)
+#     data = Column(LargeBinary, nullable=False)
+
+# init_db()
+
+# # 启动代码解释器容器
+# def start_container(port):
+#     container_name = f"code_interpreter_docker_{uuid.uuid4()}"
+#     run_command = [
+#         "docker", "run", "--name", container_name, "-d", "-p", f"{port}:5000"
+#     ]
+#     if MEMORY_LIMIT:
+#         run_command.extend(["--memory", MEMORY_LIMIT])
+#     if CPU_LIMIT:
+#         run_command.extend(["--cpus", CPU_LIMIT])
+#     run_command.append(INTERPRETER_IMAGE)
+    
+#     subprocess.run(run_command, check=True)
+#     print(f"Started container {container_name} on port {port}")
+#     return container_name, port
+
+# def stop_container(container_name):
+#     subprocess.run(["docker", "stop", container_name])
+#     subprocess.run(["docker", "rm", container_name])
+#     print(f"Stopped and removed container {container_name}")
+
+# def start_containers():
+#     for port in ports:
+#         container_name, port = start_container(port)
+#         containers.append((container_name, port))
+
+# def stop_containers():
+#     for container_name, _ in containers:
+#         stop_container(container_name)
+
+# @app.route('/runcode', methods=['POST', 'GET'])
+# def run_code():
+#     if request.method == 'POST':
+#         data = request.get_json()
+#         if not data:
+#             return jsonify({'error': 'Invalid JSON'}), 400
+#     elif request.method == 'GET':
+#         query_string = request.query_string.decode('utf-8')
+#         data = {'query_string': query_string}
+#     else:
+#         return jsonify({'error': 'Invalid request method'}), 405
+
+#     request_id = str(uuid.uuid4())  # 生成一个唯一的请求ID
+#     request_queue.put((request_id, data))
+
+#     start_time = time.time()
+#     while time.time() - start_time < 10:  # 最多等待10秒
+#         if request_id in result_dict:
+#             output = result_dict.pop(request_id)
+#             # 处理返回结果，替换 base64 数据为链接
+#             if 'images' in output:
+#                 try:
+#                     output = process_images(output)
+#                 except Exception as e:
+#                     return jsonify({'error': str(e)}), 500
+#             return jsonify(output), 200
+
+#         time.sleep(0.1)
+
+#     return jsonify({'error': 'Request timed out'}), 504
+
+# def process_images(output):
+#     images = output['images']
+#     for filename, base64_data in images.items():
+#         image_data = base64.b64decode(base64_data)
+#         unique_filename = f"{uuid.uuid4()}_{filename}"
+#         image_record = Image(filename=unique_filename, data=image_data)
+#         try:
+#             session.add(image_record)
+#             session.commit()
+#         except IntegrityError as e:
+#             session.rollback()
+#             print(f"Database error: {e}")
+#             continue
+#         if DOMAIN:
+#             images[filename] = f"https://{DOMAIN}/image/{unique_filename}"
+#         else:
+#             images[filename] = f"http://127.0.0.1:{SCHEDULER_PORT}/image/{unique_filename}"
+#     return output
+
+# @app.route('/image/<filename>', methods=['GET'])
+# def serve_image(filename):
+#     image_record = session.query(Image).filter_by(filename=filename).first()
+#     if image_record:
+#         return image_record.data, 200, {'Content-Type': 'image/png'}
+#     return jsonify({'error': 'Image not found'}), 404
+
+# def handle_requests():
+#     global current_container_index
+#     while True:
+#         request_id, data = request_queue.get()
+#         if data is None:
+#             break
+
+#         semaphore.acquire()  # 获取信号量，确保不超过并发限制
+
+#         with lock:
+#             container_name, port = containers[current_container_index]
+#             current_container_index = (current_container_index + 1) % len(containers)
+
+#         try:
+#             if 'query_string' in data:
+#                 response = requests.get(f"http://localhost:{port}/runcode?{data['query_string']}", timeout=TIMEOUT_SECONDS)
+#             else:
+#                 response = requests.post(f"http://localhost:{port}/runcode", json=data, timeout=TIMEOUT_SECONDS)
+#             output = response.json()
+#         except requests.exceptions.Timeout:
+#             output = {'error': 'Code execution timed out'}
+#         except Exception as e:
+#             output = {'error': str(e)}
+        
+#         result_dict[request_id] = output  # 将结果放入结果字典
+
+#         request_queue.task_done()
+#         print(f"Finished processing request {request_id}: {output}")
+
+#         # 异步重置容器
+#         reset_container(container_name, port)
+
+# def reset_container(container_name, port):
+#     threading.Thread(target=_reset_container, args=(container_name, port)).start()
+
+# def _reset_container(container_name, port):
+#     print(f"Resetting container {container_name}")
+#     stop_container(container_name)
+#     new_container_name, _ = start_container(port)
+#     with lock:
+#         for i, (name, p) in enumerate(containers):
+#             if name == container_name:
+#                 containers[i] = (new_container_name, port)
+#                 break
+#     semaphore.release()  # 释放信号量
+
+# def signal_handler(signal, frame):
+#     print('Stopping containers and exiting program...')
+#     stop_containers()
+#     if session:
+#         session.close()
+#     engine.dispose()
+#     os._exit(0)
+
+# signal.signal(signal.SIGINT, signal_handler)
+
+# # 启动多个工作线程来处理请求
+# for _ in range(len(ports)):
+#     threading.Thread(target=handle_requests).start()
+
+# start_containers()
+# atexit.register(stop_containers)
+
+# if __name__ == '__main__':
+#     app.run(port=SCHEDULER_PORT, threaded=True)  # 使用调度中心端口
+
+
+import yaml
+import subprocess
+import uuid
+from flask import Flask, request, jsonify
+from queue import Queue
+import requests
+import threading
+import time
+import os
+import atexit
+import base64
+import signal
+from sqlalchemy import create_engine, Column, String, LargeBinary, Integer
+from sqlalchemy.orm import declarative_base, sessionmaker
+from sqlalchemy.exc import OperationalError, IntegrityError
+import psycopg2
+
+app = Flask(__name__)
+
+# 读取配置文件，指定编码为 utf-8
+with open('config.yaml', 'r', encoding='utf-8') as file:
+    config = yaml.safe_load(file)
+
+DOMAIN = config['domain']
+INTERPRETER_IMAGE = config['interpreter_image']
+PORT_START = config['interpreter_port_range']['start']
+PORT_END = config['interpreter_port_range']['end']
+DEPENDENCIES = config['dependencies']
+RESOURCE_LIMITS = config.get('resource_limits', {})
+MEMORY_LIMIT = RESOURCE_LIMITS.get('memory')
+CPU_LIMIT = RESOURCE_LIMITS.get('cpus')
+POSTGRES = config['postgres']
+TIMEOUT = config.get('timeout', 30)  # 默认超时时间为30秒
+TIMEOUT_SECONDS = config.get('timeout_seconds', 60)  # 从配置中读取超时时间
+SCHEDULER_PORT = config['scheduler_port']  # 新增调度中心端口
+
+# 容器管理
+containers = []
+ports = list(range(PORT_START, PORT_END + 1))
+lock = threading.Lock()
+request_queue = Queue()
+result_dict = {}  # 用于存储结果的字典
+current_container_index = 0
+semaphore = threading.Semaphore(len(ports))  # 控制并发请求的信号量
+
+# 检查 PostgreSQL 容器是否存在
+POSTGRES_CONTAINER_NAME = "postgres_code_interpreter"
+POSTGRES_DATA_DIR = os.path.join(os.getcwd(), 'pgdata')
+
+def is_postgres_container_running():
+    result = subprocess.run(["docker", "ps", "-a", "--filter", f"name={POSTGRES_CONTAINER_NAME}", "--format", "{{.Names}}"], capture_output=True, text=True)
+    return POSTGRES_CONTAINER_NAME in result.stdout.strip().split('\n')
+
+def wait_for_postgres_ready():
+    retry_attempts = 10
+    retry_delay = 5  # seconds
+    for attempt in range(retry_attempts):
+        try:
+            conn = psycopg2.connect(
+                dbname=POSTGRES['db'],
+                user=POSTGRES['user'],
+                password=POSTGRES['password'],
+                host=POSTGRES['host'],
+                port=POSTGRES['port']
+            )
+            conn.close()
+            print("PostgreSQL is ready")
+            return True
+        except psycopg2.OperationalError as e:
+            print(f"Waiting for PostgreSQL to be ready: {e}")
+            time.sleep(retry_delay)
+    return False
+
+if not is_postgres_container_running():
+    if not os.path.exists(POSTGRES_DATA_DIR):
+        os.makedirs(POSTGRES_DATA_DIR)
+
+    subprocess.run([
+        "docker", "run", "--name", POSTGRES_CONTAINER_NAME, "-d",
+        "-e", f"POSTGRES_USER={POSTGRES['user']}",
+        "-e", f"POSTGRES_PASSWORD={POSTGRES['password']}",
+        "-e", f"POSTGRES_DB={POSTGRES['db']}",
+        "-v", f"{POSTGRES_DATA_DIR}:/var/lib/postgresql/data",
+        "-p", f"{POSTGRES['port']}:5432",
+        "postgres"
+    ], check=True)
+
+    if not wait_for_postgres_ready():
+        print("Failed to connect to PostgreSQL after multiple attempts.")
+        exit(1)
+
+# PostgreSQL 配置
+DATABASE_URL = f"postgresql://{POSTGRES['user']}:{POSTGRES['password']}@{POSTGRES['host']}:{POSTGRES['port']}/{POSTGRES['db']}"
+engine = None
+Base = declarative_base()
+Session = None
+session = None
+
+def init_db():
+    global engine, Session, session
+    retry_attempts = 5
+    retry_delay = 5  # seconds
+    for attempt in range(retry_attempts):
+        try:
+            engine = create_engine(DATABASE_URL)
+            Session = sessionmaker(bind=engine)
+            session = Session()
+            Base.metadata.create_all(engine)  # 确保数据库模式已创建
+            print("Database connection successful")
+            break
+        except OperationalError as e:
+            print(f"Database connection failed: {e}")
+            if attempt < retry_attempts - 1:
+                print(f"Retrying in {retry_delay} seconds...")
+                time.sleep(retry_delay)
+            else:
+                print("Failed to connect to the database after multiple attempts.")
+                raise
+
+class Image(Base):
+    __tablename__ = 'images'
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    filename = Column(String, unique=True, nullable=False)
+    data = Column(LargeBinary, nullable=False)
+
+init_db()
+
+# 启动代码解释器容器
+def start_container(port):
+    container_name = f"code_interpreter_docker_{uuid.uuid4()}"
+    run_command = [
+        "docker", "run", "--name", container_name, "-d", "-p", f"{port}:5000"
+    ]
+    if MEMORY_LIMIT:
+        run_command.extend(["--memory", MEMORY_LIMIT])
+    if CPU_LIMIT:
+        run_command.extend(["--cpus", CPU_LIMIT])
+    run_command.append(INTERPRETER_IMAGE)
+    
+    subprocess.run(run_command, check=True)
+    print(f"Started container {container_name} on port {port}")
+    return container_name, port
+
+def stop_container(container_name):
+    subprocess.run(["docker", "stop", container_name])
+    subprocess.run(["docker", "rm", container_name])
+    print(f"Stopped and removed container {container_name}")
+
+def start_containers():
+    for port in ports:
+        container_name, port = start_container(port)
+        containers.append((container_name, port))
+
+def stop_containers():
+    for container_name, _ in containers:
+        stop_container(container_name)
+
+@app.route('/runcode', methods=['POST', 'GET'])
+def run_code():
+    if request.method == 'POST':
+        data = request.get_json()
+        if not data:
+            return jsonify({'error': 'Invalid JSON'}), 400
+    elif request.method == 'GET':
+        query_string = request.query_string.decode('utf-8')
+        data = {'query_string': query_string}
+    else:
+        return jsonify({'error': 'Invalid request method'}), 405
+
+    request_id = str(uuid.uuid4())  # 生成一个唯一的请求ID
+    request_queue.put((request_id, data))
+
+    start_time = time.time()
+    while time.time() - start_time < 10:  # 最多等待10秒
+        if request_id in result_dict:
+            output = result_dict.pop(request_id)
+            # 处理返回结果，替换 base64 数据为链接
+            if 'images' in output:
+                try:
+                    output = process_images(output)
+                except Exception as e:
+                    return jsonify({'error': str(e)}), 500
+            return jsonify(output), 200
+
+        time.sleep(0.1)
+
+    return jsonify({'error': 'Request timed out'}), 504
+
+def process_images(output):
+    images = output['images']
+    for filename, base64_data in images.items():
+        image_data = base64.b64decode(base64_data)
+        unique_filename = f"{uuid.uuid4()}_{filename}"
+        image_record = Image(filename=unique_filename, data=image_data)
+        try:
+            session.add(image_record)
+            session.commit()
+        except IntegrityError as e:
+            session.rollback()
+            print(f"Database error: {e}")
+            continue
+        if DOMAIN:
+            images[filename] = f"https://{DOMAIN}/image/{unique_filename}"
+        else:
+            images[filename] = f"http://127.0.0.1:{SCHEDULER_PORT}/image/{unique_filename}"
+    return output
+
+@app.route('/image/<filename>', methods=['GET'])
+def serve_image(filename):
+    image_record = session.query(Image).filter_by(filename=filename).first()
+    if image_record:
+        return image_record.data, 200, {'Content-Type': 'image/png'}
+    return jsonify({'error': 'Image not found'}), 404
+
+def handle_requests():
+    global current_container_index
+    while True:
+        request_id, data = request_queue.get()
+        if data is None:
+            break
+
+        semaphore.acquire()  # 获取信号量，确保不超过并发限制
+
+        with lock:
+            container_name, port = containers[current_container_index]
+            current_container_index = (current_container_index + 1) % len(containers)
+
+        try:
+            if 'query_string' in data:
+                response = requests.get(f"http://localhost:{port}/runcode?{data['query_string']}", timeout=TIMEOUT_SECONDS)
+            else:
+                response = requests.post(f"http://localhost:{port}/runcode", json=data, timeout=TIMEOUT_SECONDS)
+            output = response.json()
+        except requests.exceptions.Timeout:
+            output = {'error': 'Code execution timed out'}
+        except Exception as e:
+            output = {'error': str(e)}
+        
+        result_dict[request_id] = output  # 将结果放入结果字典
+
+        request_queue.task_done()
+        print(f"Finished processing request {request_id}: {output}")
+
+        # 异步重置容器
+        reset_container(container_name, port)
+
+def reset_container(container_name, port):
+    threading.Thread(target=_reset_container, args=(container_name, port)).start()
+
+def _reset_container(container_name, port):
+    print(f"Resetting container {container_name}")
+    stop_container(container_name)
+    new_container_name, _ = start_container(port)
+    with lock:
+        for i, (name, p) in enumerate(containers):
+            if name == container_name:
+                containers[i] = (new_container_name, port)
+                break
+    semaphore.release()  # 释放信号量
+
+def signal_handler(signal, frame):
+    print('Stopping containers and exiting program...')
+    stop_containers()
+    if session:
+        session.close()
+    engine.dispose()
+    os._exit(0)
+
+signal.signal(signal.SIGINT, signal_handler)
+
+# 启动多个工作线程来处理请求
+for _ in range(len(ports)):
+    threading.Thread(target=handle_requests).start()
+
+start_containers()
+atexit.register(stop_containers)
+
+if __name__ == '__main__':
+    app.run(port=SCHEDULER_PORT, threaded=True)  # 使用调度中心端口

concurrent_test.py

@@ -0,0 +1,32 @@
+import aiohttp
+import asyncio
+
+async def post_request(session, url, headers, json_data):
+    async with session.post(url, headers=headers, json=json_data) as response:
+        response_text = await response.text()
+        print(response_text)
+        return response_text
+
+async def main(concurrent_requests):
+    url = 'http://127.0.0.1:8000/runcode'
+    headers = {
+        'User-Agent': 'Apifox/1.0.0 (https://apifox.com)',
+        'Content-Type': 'application/json',
+        'Authorization': 'Bearer 114514',
+        'Accept': '*/*',
+        'Host': '127.0.0.1:8000',
+        'Connection': 'keep-alive'
+    }
+    json_data = {
+        "languageType": "python",
+        "variables": {},
+        "code": "```python\nimport time\nprint('Hello from code block!')\ntime.sleep(3)\n```"
+    }
+
+    async with aiohttp.ClientSession() as session:
+        tasks = [post_request(session, url, headers, json_data) for _ in range(concurrent_requests)]
+        await asyncio.gather(*tasks)
+
+if __name__ == '__main__':
+    concurrent_requests = 5  # 设置并发数
+    asyncio.run(main(concurrent_requests))

config.yaml

@@ -0,0 +1,44 @@
+domain: "" #填写域名后对外提供链接将使用域名替代 ip:scheduler_port
+interpreter_image: "leezhuuu/code_interpreter"
+scheduler_port: 8000 # 调度中心端口
+interpreter_port_range: # 设置端口范围为 8001-8003，端口范围决定容器数量
+  start: 8001
+  end: 8003
+dependencies: # 定义容器的依赖项，以下为默认依赖项,可以根据需要自行更改
+  - numpy
+  - pandas
+  - scipy
+  - matplotlib
+  - seaborn
+  - scikit-learn
+  - opencv-python
+  - opencv-python-headless
+  - Pillow
+  - requests
+  - Flask
+  - pyyaml
+  - sympy
+  - plotly
+  - bokeh
+  - statsmodels
+  - jupyter
+  - ipython
+  - jupyterlab
+  - pytest
+  - hypothesis
+  - Flask-Cors
+  - Werkzeug
+  - Gunicorn
+resource_limits:
+  memory: ""  # 可以自定义内存限制，例如 "2g"、"500m" 等，不填则默认不进行限制
+  cpus: ""   # 可以自定义 CPU 限制，例如 "1.5"、"0.5" 等，不填则默认不进行限制
+timeout_seconds: 60  # 设置超时时间为 1 分钟（60 秒）
+postgres:
+  user: "user" # 数据库用户名,请自行更改
+  password: "password" # 数据库密码,请自行更改
+  db: "code_interpreter_db"
+  host: "localhost"
+  port: "5432" # 数据库端口,请自行更改
+mode: "docker" # 选择运行模式，可以是 "docker" 或 "k8s"
+kubeconfig_path: "/path/to/kubeconfig.yaml" # Kubernetes 配置文件路径
+k8s_interpreter_yaml: "/path/to/code-interpreter.yaml" # Kubernetes 部署配置文件路径

requirements.txt

@@ -0,0 +1,24 @@
+numpy
+pandas
+scipy
+matplotlib
+seaborn
+scikit-learn
+opencv-python
+opencv-python-headless
+Pillow
+requests
+Flask
+pyyaml
+sympy
+plotly
+bokeh
+statsmodels
+jupyter
+ipython
+jupyterlab
+pytest
+hypothesis
+Flask-Cors
+Werkzeug
+Gunicorn