| import json |
| import re |
| from openai import OpenAI |
| from src.models import CodeResponse |
| from src.config import API_KEY, BASE_URL, MODEL |
|
|
|
|
| _client: OpenAI | None = None |
|
|
|
|
| def _get_client() -> OpenAI: |
| global _client |
| if _client is None: |
| _client = OpenAI(api_key=API_KEY, base_url=BASE_URL) |
| return _client |
|
|
|
|
| _SYSTEM_PROMPT = """You are an expert Python code generator. Given a user's intent, produce a runnable Python script. |
| |
| Rules: |
| - Output ONLY valid Python code that can execute in an isolated environment. |
| - Include all necessary imports. |
| - Use print() for any output you want the user to see. |
| - Do NOT use interactive functions like input(). |
| - Do NOT access files outside /tmp or the current directory. |
| - If you generate charts/plots/images, save them to /tmp with a descriptive filename and print the path at the end. |
| - If you need external packages, list them in requirements. |
| - The execution environment HAS full network access. You may use urllib, requests, aiohttp, httpx, etc. to fetch remote data. |
| - Keep the script self-contained and focused. |
| |
| Respond with valid JSON in exactly this format (no markdown, no code fences): |
| {"explanation": "brief explanation", "script": "python code here", "requirements": ["package1", "package2"]}""" |
|
|
|
|
| def generate_code(user_prompt: str, previous_error: str | None = None) -> CodeResponse: |
| client = _get_client() |
|
|
| messages = [{"role": "system", "content": _SYSTEM_PROMPT}] |
| if previous_error: |
| user_content = ( |
| f"The previous attempt failed:\n\n{previous_error}\n\n" |
| f"Fix the issue and rewrite. User intent: {user_prompt}" |
| ) |
| else: |
| user_content = user_prompt |
|
|
| messages.append({"role": "user", "content": user_content}) |
|
|
| response = client.chat.completions.create( |
| model=MODEL, |
| messages=messages, |
| max_tokens=4000, |
| temperature=0.1, |
| ) |
|
|
| raw = response.choices[0].message.content or "{}" |
| return _parse_json(raw, user_prompt) |
|
|
|
|
| def _parse_json(raw: str, fallback_prompt: str) -> CodeResponse: |
| cleaned = raw.strip() |
| if cleaned.startswith("```"): |
| cleaned = re.sub(r"^```(?:json)?\s*", "", cleaned) |
| cleaned = re.sub(r"\s*```$", "", cleaned) |
| try: |
| data = json.loads(cleaned) |
| return CodeResponse( |
| explanation=data.get("explanation", ""), |
| script=data.get("script", ""), |
| requirements=data.get("requirements", []), |
| ) |
| except json.JSONDecodeError: |
| code_match = re.search(r"```python\n?(.*?)```", raw, re.DOTALL) |
| if code_match: |
| return CodeResponse( |
| explanation="Extracted from code block", |
| script=code_match.group(1).strip(), |
| requirements=[], |
| ) |
| return CodeResponse( |
| explanation="Fallback: treated entire response as script", |
| script=raw, |
| requirements=[], |
| ) |
|
|