Update app.py

app.py

@@ -1,251 +1,250 @@
-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!")
+from flask import Flask, request, jsonify, Response, render_template_string
+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)
+        
+        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
+
+@app.route('/', methods=['GET'])
+def welcome_page():
+    return render_template_string('<h1>欢迎使用lee\'s code-interpreter</h1>')
+
+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!")