diff --git a/back_end/settings.py b/back_end/settings.py index 899cce1f..43daa6af 100644 --- a/back_end/settings.py +++ b/back_end/settings.py @@ -253,6 +253,12 @@ SANDBOX_TIMEOUT = int(os.getenv('SANDBOX_TIMEOUT', '30')) # API 請求超時（秒） SANDBOX_API_KEY = os.getenv('SANDBOX_API_KEY', '') # API Key for authentication +# ==================== +# LLM Test Generation Service Configuration +# ==================== +LLM_TESTGEN_API_URL = os.getenv('LLM_TESTGEN_API_URL', 'http://34.81.90.111:8001') +LLM_TESTGEN_TIMEOUT = int(os.getenv('LLM_TESTGEN_TIMEOUT', '120')) # LLM 生成可能較慢，預設 120 秒 + # ==================== # Backend Configuration # ==================== diff --git a/problems/services/llm_testgen.py b/problems/services/llm_testgen.py new file mode 100644 index 00000000..3ef7ff0d --- /dev/null +++ b/problems/services/llm_testgen.py @@ -0,0 +1,352 @@ +""" +LLM Test Data Generation Service + +封裝與 LLM 測資生成服務的 API 互動邏輯 +""" + +import requests +import logging +from io import BytesIO +from typing import Optional, List, Dict, Any +from django.conf import settings + +logger = logging.getLogger(__name__) + +# LLM 測資生成服務設定 +LLM_TESTGEN_API_URL = getattr(settings, 'LLM_TESTGEN_API_URL', 'http://34.81.90.111:8001') +LLM_TESTGEN_TIMEOUT = getattr(settings, 'LLM_TESTGEN_TIMEOUT', 120) # 120 秒超時（LLM 生成可能較慢） + + +def get_solution_runtime(language: str) -> str: + """ + 將 Django 的語言名稱轉換成 LLM 服務的 runtime 格式 + + Django: 'c', 'cpp', 'python', 'java', 'javascript' + LLM Service: 'c', 'cpp', 'python', 'java' + """ + language_map = { + 'c': 'c', + 'cpp': 'cpp', + 'c++': 'cpp', + 'python': 'python', + 'py': 'python', + 'java': 'java', + 'javascript': 'python', # LLM 服務不支援 JS，暫時用 python + 'js': 'python', + } + return language_map.get(language.lower(), 'python') + + +def upload_solution(source_code: str, language: str) -> Dict[str, Any]: + """ + 上傳正解程式到 LLM 服務 + + Args: + source_code: 正解程式碼 + language: 程式語言 (c, cpp, python, java) + + Returns: + dict: { + 'ok': bool, + 'solution_id': str (如果成功), + 'error': str (如果失敗) + } + """ + try: + runtime = get_solution_runtime(language) + + # 建立檔案物件 + extension_map = { + 'c': 'c', + 'cpp': 'cpp', + 'python': 'py', + 'java': 'java', + } + ext = extension_map.get(runtime, 'txt') + filename = f'solution.{ext}' + + files = { + 'file': (filename, BytesIO(source_code.encode('utf-8')), 'text/plain') + } + data = { + 'runtime': runtime + } + + url = f'{LLM_TESTGEN_API_URL}/api/upload-solution' + logger.info(f'Uploading solution to LLM service: {url}') + + response = requests.post( + url, + files=files, + data=data, + timeout=LLM_TESTGEN_TIMEOUT + ) + + response.raise_for_status() + result = response.json() + + logger.info(f'Upload solution response: {result}') + return result + + except requests.exceptions.Timeout: + logger.error('Upload solution timeout') + return {'ok': False, 'error': 'LLM 服務連線逾時'} + except requests.exceptions.ConnectionError: + logger.error('Upload solution connection error') + return {'ok': False, 'error': 'LLM 服務連線失敗'} + except requests.exceptions.RequestException as e: + logger.error(f'Upload solution request error: {str(e)}') + return {'ok': False, 'error': f'請求錯誤: {str(e)}'} + except Exception as e: + logger.error(f'Upload solution error: {str(e)}') + return {'ok': False, 'error': f'未知錯誤: {str(e)}'} + + +def generate_testcases( + problem_statement: str, + input_spec: str, + output_spec: Optional[str] = None, + constraints: Optional[str] = None, + subtasks: Optional[List[Dict]] = None, + num_cases: Optional[int] = None, + mode: str = 'LLM_DIRECT', + solution_id: Optional[str] = None, + examples: Optional[List[Dict[str, str]]] = None +) -> Dict[str, Any]: + """ + 呼叫 LLM 服務生成測資 + + Args: + problem_statement: 題目敘述 + input_spec: 輸入格式說明 + output_spec: 輸出格式說明 + constraints: 限制條件 + subtasks: 子任務列表 [{'id': 1, 'name': 'subtask1', 'desc': '...', 'num': 5}] + num_cases: 測資數量（若無 subtasks） + mode: 模式 - 'LLM_INPUT_ONLY' 或 'LLM_DIRECT' + solution_id: 正解 ID（LLM_INPUT_ONLY 必填） + examples: 範例測資 [{'input': '...', 'output': '...'}] + + Returns: + dict: { + 'ok': bool, + 'data': {...} (如果成功), + 'error': str (如果失敗) + } + """ + try: + payload = { + 'problem_statement': problem_statement, + 'input_spec': input_spec, + 'mode': mode, + } + + if output_spec: + payload['output_spec'] = output_spec + + if constraints: + payload['constraints'] = constraints + + if subtasks: + payload['subtasks'] = subtasks + + if num_cases: + payload['num_cases'] = num_cases + + if solution_id: + payload['solution_id'] = solution_id + + if examples: + payload['examples'] = examples + + url = f'{LLM_TESTGEN_API_URL}/api/generate-testcases' + logger.info(f'Generating testcases via LLM service: {url}') + logger.debug(f'Generate request payload: {payload}') + + response = requests.post( + url, + json=payload, + timeout=LLM_TESTGEN_TIMEOUT + ) + + response.raise_for_status() + result = response.json() + + logger.info(f'Generate testcases response ok: {result.get("ok")}') + return result + + except requests.exceptions.Timeout: + logger.error('Generate testcases timeout') + return {'ok': False, 'error': 'LLM 服務連線逾時（生成測資可能需要較長時間）'} + except requests.exceptions.ConnectionError: + logger.error('Generate testcases connection error') + return {'ok': False, 'error': 'LLM 服務連線失敗'} + except requests.exceptions.RequestException as e: + logger.error(f'Generate testcases request error: {str(e)}') + return {'ok': False, 'error': f'請求錯誤: {str(e)}'} + except Exception as e: + logger.error(f'Generate testcases error: {str(e)}') + return {'ok': False, 'error': f'未知錯誤: {str(e)}'} + + +def list_solutions() -> Dict[str, Any]: + """ + 列出所有已上傳的正解 + + Returns: + dict: API 回應 + """ + try: + url = f'{LLM_TESTGEN_API_URL}/api/solutions' + response = requests.get(url, timeout=LLM_TESTGEN_TIMEOUT) + response.raise_for_status() + return response.json() + except Exception as e: + logger.error(f'List solutions error: {str(e)}') + return {'error': str(e)} + + +def delete_solution(solution_id: str) -> Dict[str, Any]: + """ + 刪除正解 + + Args: + solution_id: 正解 ID + + Returns: + dict: API 回應 + """ + try: + url = f'{LLM_TESTGEN_API_URL}/api/solutions/{solution_id}' + response = requests.delete(url, timeout=LLM_TESTGEN_TIMEOUT) + response.raise_for_status() + return response.json() + except Exception as e: + logger.error(f'Delete solution error: {str(e)}') + return {'error': str(e)} + + +def health_check() -> bool: + """ + 檢查 LLM 服務健康狀態 + + Returns: + bool: 服務是否正常 + """ + try: + url = f'{LLM_TESTGEN_API_URL}/health' + response = requests.get(url, timeout=10) + return response.status_code == 200 + except Exception: + return False + + +def generate_testcases_for_problem(problem) -> Dict[str, Any]: + """ + 根據題目資訊自動選擇模式並生成測資 + + Args: + problem: Problems model instance + + Returns: + dict: { + 'ok': bool, + 'mode': str, + 'data': {...}, + 'error': str (如果失敗) + } + """ + from problems.models import Problem_subtasks + + # 建構題目敘述 + problem_statement = problem.description + + # 輸入/輸出格式 + input_spec = problem.input_description or '' + output_spec = problem.output_description or '' + + # 限制條件（從 subtask_description 或其他欄位取得） + constraints = problem.subtask_description or problem.hint or '' + + # 範例測資 + examples = [] + if problem.sample_input and problem.sample_output: + examples.append({ + 'input': problem.sample_input, + 'output': problem.sample_output + }) + + # 取得 subtasks + subtasks_qs = Problem_subtasks.objects.filter(problem_id=problem.id).order_by('subtask_no') + subtasks = [] + + if subtasks_qs.exists(): + for st in subtasks_qs: + subtasks.append({ + 'id': st.subtask_no, + 'name': f'Subtask {st.subtask_no}', + 'desc': st.description or '', + 'num': st.num_testcases if hasattr(st, 'num_testcases') else 5 + }) + + # 決定生成模式 + has_solution = bool(problem.solution_code and problem.solution_code.strip()) + + if has_solution: + # 使用 LLM_INPUT_ONLY 模式：先上傳正解，再生成測資 + logger.info(f'Problem {problem.id} has solution code, using LLM_INPUT_ONLY mode') + + # 上傳正解 + upload_result = upload_solution( + source_code=problem.solution_code, + language=problem.solution_code_language or 'python' + ) + + if not upload_result.get('ok'): + return { + 'ok': False, + 'mode': 'LLM_INPUT_ONLY', + 'error': f"上傳正解失敗: {upload_result.get('error', '未知錯誤')}" + } + + solution_id = upload_result.get('solution_id') + + # 生成測資 + gen_result = generate_testcases( + problem_statement=problem_statement, + input_spec=input_spec, + output_spec=output_spec, + constraints=constraints, + subtasks=subtasks if subtasks else None, + num_cases=5 if not subtasks else None, + mode='LLM_INPUT_ONLY', + solution_id=solution_id, + examples=examples if examples else None + ) + + # 清理上傳的正解 + try: + delete_solution(solution_id) + except Exception as e: + logger.warning(f'Failed to delete solution {solution_id}: {str(e)}') + + gen_result['mode'] = 'LLM_INPUT_ONLY' + return gen_result + + else: + # 使用 LLM_DIRECT 模式：直接生成 input 和 output + logger.info(f'Problem {problem.id} has no solution code, using LLM_DIRECT mode') + + gen_result = generate_testcases( + problem_statement=problem_statement, + input_spec=input_spec, + output_spec=output_spec, + constraints=constraints, + subtasks=subtasks if subtasks else None, + num_cases=5 if not subtasks else None, + mode='LLM_DIRECT', + examples=examples if examples else None + ) + + gen_result['mode'] = 'LLM_DIRECT' + return gen_result diff --git a/problems/urls.py b/problems/urls.py index 3aa2d248..fd013d08 100644 --- a/problems/urls.py +++ b/problems/urls.py @@ -13,6 +13,12 @@ ProblemTestCaseListCreateView, ProblemTestCaseDetailView, ProblemTestCaseZipUploadView, ) from .views.sandbox import ProblemTestCasePackageView +from .views.llm_testgen import ( + LLMTestGenHealthView, + LLMTestGenGenerateView, + LLMTestGenCustomView, + LLMTestGenSaveView, +) router = DefaultRouter() router.register(r"problems", ProblemsViewSet, basename="problems") @@ -46,6 +52,11 @@ path("/checksum", ProblemTestCaseChecksumView.as_view(), name="problem-testcase-checksum"), path("/meta", ProblemTestCaseMetaView.as_view(), name="problem-testcase-meta"), path("/testdata", ProblemTestCasePackageView.as_view(), name="problem-testcase-package"), + # LLM 測資生成 API + path("llm-testgen/health", LLMTestGenHealthView.as_view(), name="llm-testgen-health"), + path("/llm-testgen/generate", LLMTestGenGenerateView.as_view(), name="llm-testgen-generate"), + path("/llm-testgen/custom", LLMTestGenCustomView.as_view(), name="llm-testgen-custom"), + path("/llm-testgen/save", LLMTestGenSaveView.as_view(), name="llm-testgen-save"), # 新標籤 API path("tags", TagListCreateView.as_view(), name="tag-list-create"), path("/tags", ProblemTagAddView.as_view(), name="problem-tag-add"), diff --git a/problems/views/llm_testgen.py b/problems/views/llm_testgen.py new file mode 100644 index 00000000..67ba2cba --- /dev/null +++ b/problems/views/llm_testgen.py @@ -0,0 +1,386 @@ +""" +LLM Test Data Generation Views + +提供 LLM 自動生成測資的 API endpoints +""" + +import logging +import json +from rest_framework import status +from rest_framework.views import APIView +from rest_framework.response import Response +from rest_framework.permissions import IsAuthenticated +from django.shortcuts import get_object_or_404 + +from problems.models import Problems, Problem_subtasks, Test_cases +from problems.services.llm_testgen import ( + generate_testcases_for_problem, + generate_testcases, + upload_solution, + delete_solution, + health_check, +) + +logger = logging.getLogger(__name__) + + +def api_response(data=None, message="OK", status_code=200): + """統一的 API 響應格式""" + status_str = "ok" if 200 <= status_code < 400 else "error" + return Response({ + "data": data, + "message": message, + "status": status_str, + }, status=status_code) + + +class LLMTestGenHealthView(APIView): + """ + GET /problem/llm-testgen/health + + 檢查 LLM 測資生成服務健康狀態 + """ + permission_classes = [IsAuthenticated] + + def get(self, request): + is_healthy = health_check() + + if is_healthy: + return api_response( + data={'status': 'healthy'}, + message='LLM 測資生成服務正常', + status_code=status.HTTP_200_OK + ) + else: + return api_response( + data={'status': 'unhealthy'}, + message='LLM 測資生成服務無法連線', + status_code=status.HTTP_503_SERVICE_UNAVAILABLE + ) + + +class LLMTestGenGenerateView(APIView): + """ + POST /problem/{pk}/llm-testgen/generate + + 根據題目資訊自動生成測資 + + Request Body (optional): + { + "num_cases": 5, // 覆蓋預設測資數量 + "subtasks": [ // 覆蓋 subtask 設定 + {"id": 1, "name": "Easy", "desc": "N <= 100", "num": 3}, + {"id": 2, "name": "Hard", "desc": "N <= 10000", "num": 5} + ] + } + + Response: + { + "data": { + "ok": true, + "mode": "LLM_INPUT_ONLY" | "LLM_DIRECT", + "testcases": [ + {"input": "...", "output": "...", "subtask_id": 1} + ] + }, + "message": "測資生成成功", + "status": "ok" + } + """ + permission_classes = [IsAuthenticated] + + def post(self, request, pk): + # 取得題目 + problem = get_object_or_404(Problems, pk=pk) + + # 權限檢查：只有題目創建者或管理員可以生成測資 + if problem.creator_id != request.user and not request.user.is_staff: + return api_response( + data=None, + message='沒有權限操作此題目', + status_code=status.HTTP_403_FORBIDDEN + ) + + # 檢查題目是否有基本資訊 + if not problem.description: + return api_response( + data=None, + message='題目缺少描述，無法生成測資', + status_code=status.HTTP_400_BAD_REQUEST + ) + + if not problem.input_description: + return api_response( + data=None, + message='題目缺少輸入格式說明，無法生成測資', + status_code=status.HTTP_400_BAD_REQUEST + ) + + logger.info(f'Generating testcases for problem {pk} by user {request.user.id}') + + try: + # 呼叫 LLM 服務生成測資 + result = generate_testcases_for_problem(problem) + + if result.get('ok'): + return api_response( + data={ + 'ok': True, + 'mode': result.get('mode'), + 'testcases': result.get('data', {}).get('testcases', []), + 'raw_response': result.get('data') + }, + message=f"測資生成成功（模式: {result.get('mode')}）", + status_code=status.HTTP_200_OK + ) + else: + return api_response( + data={'ok': False, 'error': result.get('error')}, + message=f"測資生成失敗: {result.get('error', '未知錯誤')}", + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR + ) + + except Exception as e: + logger.error(f'LLM testgen error for problem {pk}: {str(e)}', exc_info=True) + return api_response( + data={'ok': False, 'error': str(e)}, + message=f'測資生成失敗: {str(e)}', + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR + ) + + +class LLMTestGenCustomView(APIView): + """ + POST /problem/{pk}/llm-testgen/custom + + 使用自訂參數生成測資 + + Request Body: + { + "problem_statement": "...", // 可覆蓋題目敘述 + "input_spec": "...", // 可覆蓋輸入格式 + "output_spec": "...", // 可覆蓋輸出格式 + "constraints": "...", // 限制條件 + "mode": "LLM_DIRECT", // LLM_DIRECT 或 LLM_INPUT_ONLY + "num_cases": 5, // 測資數量 + "subtasks": [...], // 子任務設定 + "solution_code": "...", // 正解程式碼（LLM_INPUT_ONLY 必填） + "solution_language": "python" // 正解語言 + } + """ + permission_classes = [IsAuthenticated] + + def post(self, request, pk): + # 取得題目 + problem = get_object_or_404(Problems, pk=pk) + + # 權限檢查 + if problem.creator_id != request.user and not request.user.is_staff: + return api_response( + data=None, + message='沒有權限操作此題目', + status_code=status.HTTP_403_FORBIDDEN + ) + + data = request.data + + # 取得參數（優先使用請求中的參數，否則使用題目欄位） + problem_statement = data.get('problem_statement') or problem.description + input_spec = data.get('input_spec') or problem.input_description + output_spec = data.get('output_spec') or problem.output_description + constraints = data.get('constraints') or problem.subtask_description or '' + mode = data.get('mode', 'LLM_DIRECT') + num_cases = data.get('num_cases', 5) + subtasks = data.get('subtasks') + + # 範例測資 + examples = [] + if problem.sample_input and problem.sample_output: + examples.append({ + 'input': problem.sample_input, + 'output': problem.sample_output + }) + + # 驗證必填欄位 + if not problem_statement: + return api_response( + data=None, + message='缺少題目敘述', + status_code=status.HTTP_400_BAD_REQUEST + ) + + if not input_spec: + return api_response( + data=None, + message='缺少輸入格式說明', + status_code=status.HTTP_400_BAD_REQUEST + ) + + try: + solution_id = None + + # LLM_INPUT_ONLY 模式需要正解 + if mode == 'LLM_INPUT_ONLY': + solution_code = data.get('solution_code') or problem.solution_code + solution_language = data.get('solution_language') or problem.solution_code_language + + if not solution_code: + return api_response( + data=None, + message='LLM_INPUT_ONLY 模式需要提供正解程式碼', + status_code=status.HTTP_400_BAD_REQUEST + ) + + # 上傳正解 + upload_result = upload_solution(solution_code, solution_language or 'python') + + if not upload_result.get('ok'): + return api_response( + data={'ok': False, 'error': upload_result.get('error')}, + message=f"上傳正解失敗: {upload_result.get('error')}", + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR + ) + + solution_id = upload_result.get('solution_id') + + # 生成測資 + result = generate_testcases( + problem_statement=problem_statement, + input_spec=input_spec, + output_spec=output_spec, + constraints=constraints, + subtasks=subtasks, + num_cases=num_cases if not subtasks else None, + mode=mode, + solution_id=solution_id, + examples=examples if examples else None + ) + + # 清理上傳的正解 + if solution_id: + try: + delete_solution(solution_id) + except Exception as e: + logger.warning(f'Failed to delete solution {solution_id}: {str(e)}') + + if result.get('ok'): + return api_response( + data={ + 'ok': True, + 'mode': mode, + 'testcases': result.get('data', {}).get('testcases', []), + 'raw_response': result.get('data') + }, + message=f"測資生成成功（模式: {mode}）", + status_code=status.HTTP_200_OK + ) + else: + return api_response( + data={'ok': False, 'error': result.get('error')}, + message=f"測資生成失敗: {result.get('error', '未知錯誤')}", + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR + ) + + except Exception as e: + logger.error(f'LLM custom testgen error for problem {pk}: {str(e)}', exc_info=True) + return api_response( + data={'ok': False, 'error': str(e)}, + message=f'測資生成失敗: {str(e)}', + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR + ) + + +class LLMTestGenSaveView(APIView): + """ + POST /problem/{pk}/llm-testgen/save + + 將 LLM 生成的測資保存到題目 + + Request Body: + { + "testcases": [ + { + "input": "...", + "output": "...", + "subtask_id": 1 // optional, 對應到 subtask_no + } + ], + "create_subtasks": true // 是否自動建立不存在的 subtask + } + """ + permission_classes = [IsAuthenticated] + + def post(self, request, pk): + # 取得題目 + problem = get_object_or_404(Problems, pk=pk) + + # 權限檢查 + if problem.creator_id != request.user and not request.user.is_staff: + return api_response( + data=None, + message='沒有權限操作此題目', + status_code=status.HTTP_403_FORBIDDEN + ) + + data = request.data + testcases = data.get('testcases', []) + create_subtasks = data.get('create_subtasks', True) + + if not testcases: + return api_response( + data=None, + message='沒有測資可保存', + status_code=status.HTTP_400_BAD_REQUEST + ) + + try: + saved_count = 0 + created_subtasks = set() + + for tc in testcases: + input_data = tc.get('input', '') + output_data = tc.get('output', '') + subtask_no = tc.get('subtask_id', 1) + + # 確保 subtask 存在 + subtask, created = Problem_subtasks.objects.get_or_create( + problem_id=problem, + subtask_no=subtask_no, + defaults={ + 'score': 100 // len(set(t.get('subtask_id', 1) for t in testcases)), + 'description': f'Subtask {subtask_no}' + } + ) + + if created: + created_subtasks.add(subtask_no) + + # 計算 idx（該 subtask 下的測資數量 + 1） + existing_count = Test_cases.objects.filter(subtask_id=subtask).count() + + # 建立測資 + Test_cases.objects.create( + subtask_id=subtask, + idx=existing_count + 1, + input_data=input_data, + expected_output=output_data + ) + + saved_count += 1 + + return api_response( + data={ + 'saved_count': saved_count, + 'created_subtasks': list(created_subtasks) + }, + message=f'成功保存 {saved_count} 筆測資', + status_code=status.HTTP_201_CREATED + ) + + except Exception as e: + logger.error(f'Save testcases error for problem {pk}: {str(e)}', exc_info=True) + return api_response( + data=None, + message=f'保存測資失敗: {str(e)}', + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR + )