Spaces:
No application file
No application file
| import os | |
| import io | |
| import requests | |
| import tempfile | |
| import contextlib | |
| from typing import Tuple | |
| from tools.interfaces.interface_code_excutor import CodeExecutorInterface | |
| class PythonCodeExecutor(CodeExecutorInterface): | |
| """ | |
| Implémentation de CodeExecutorInterface pour exécuter du code Python | |
| avec un fichier externe (local ou distant). | |
| """ | |
| def __init__(self,huggingface_token:str,base_url:str): | |
| self.HUGGINGFACE_TOKEN=huggingface_token | |
| self.HUGGINGFACE_BASE_URL=base_url | |
| def run_code_with_file(self, code_and_file: str) -> str: | |
| try: | |
| code, mode, value = self._parse_code_and_file(code_and_file) | |
| file_path = self._resolve_file(mode, value) | |
| return self._execute_code(code, file_path) | |
| except Exception as e: | |
| return f"Erreur : {e}" | |
| def _parse_code_and_file(self, code_and_file: str) -> Tuple[str, str, str]: | |
| if "file_path=" in code_and_file: | |
| parts = code_and_file.split("file_path=") | |
| return parts[0].strip(), "file_path", parts[1].strip() | |
| elif "file_name=" in code_and_file: | |
| parts = code_and_file.split("file_name=") | |
| return parts[0].strip(), "file_name", parts[1].strip() | |
| else: | |
| raise ValueError("Aucun fichier spécifié (file_path= ou file_name= attendu).") | |
| def _resolve_file(self, mode: str, value: str) -> str: | |
| if mode == "file_path": | |
| if not os.path.exists(value): | |
| raise FileNotFoundError(f"Fichier local introuvable : {value}") | |
| return value | |
| elif mode == "file_name": | |
| headers = {"Authorization": f"Bearer {self.HUGGINGFACE_TOKEN}"} | |
| response = requests.get(self.HUGGINGFACE_BASE_URL + value, headers=headers) | |
| response.raise_for_status() | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(value)[-1]) as tmp: | |
| tmp.write(response.content) | |
| return tmp.name | |
| else: | |
| raise ValueError(f"Mode de fichier non supporté : {mode}") | |
| def _execute_code(self, code: str, file_path: str) -> str: | |
| output = io.StringIO() | |
| local_vars = {"file_path": file_path} | |
| with contextlib.redirect_stdout(output): | |
| exec(code, {}, local_vars) | |
| os.remove(file_path) if os.path.exists(file_path) and "tmp" in file_path else None | |
| result = output.getvalue() | |
| return result if result.strip() else "Code exécuté sans sortie." | |