File size: 4,379 Bytes
78c0e6a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """
Board XML to JSON processor.
Updated: uses pooled connections, no threading.
"""
import json
import re
from typing import Optional, Dict, Any, List
from http_pool import board_processor_session
class BoardProcessor:
def __init__(self, api_url: str = "https://corvo-ai-claude-4-6-opus.hf.space/chat"):
self.api_url = api_url
self.system_prompt = self._load_system_prompt()
def _load_system_prompt(self) -> str:
"""Load system prompt from system.txt file"""
try:
with open('system.txt', 'r', encoding='utf-8') as f:
return f.read()
except FileNotFoundError:
return """You are a converter that transforms XML board data into JSON format.
Convert the given XML to JSON following these rules:
1. Extract all elements from the XML
2. Convert them to proper JSON structure
3. Wrap your JSON output in <json></json> tags
Example:
Input: <board><note color="yellow" x="100" y="200">Hello</note></board>
Output: <json>{"board": {"notes": [{"color": "yellow", "x": 100, "y": 200, "text": "Hello"}]}}</json>
Always wrap your final JSON output in <json></json> tags."""
def _call_ai_api(
self,
user_input: Optional[str] = None,
chat_history: Optional[List[Dict[str, Any]]] = None,
temperature: float = 0.9,
top_p: float = 0.95,
max_tokens: Optional[int] = None
) -> str:
"""Call the AI API using pooled connection."""
payload = {
"user_input": user_input,
"chat_history": chat_history or [],
"temperature": temperature,
"top_p": top_p,
"max_tokens": max_tokens
}
try:
response = board_processor_session.post(
self.api_url, json=payload, timeout=120
)
response.raise_for_status()
result = response.json()
return result.get("assistant_response", "")
except Exception as e:
raise Exception(f"API request failed: {str(e)}")
def _extract_json_from_response(self, response: str) -> str:
"""Extract JSON from <json></json> tags in the AI response."""
print("\n📋 DEBUG - Raw AI Response:")
print("=" * 80)
print(response[:500] + ("..." if len(response) > 500 else ""))
print("=" * 80)
patterns = [
r'<json>(.*?)</json>',
r'```json\s*(.*?)\s*```',
r'```\s*(.*?)\s*```',
]
for pattern in patterns:
json_match = re.search(pattern, response, re.DOTALL)
if json_match:
json_str = json_match.group(1).strip()
try:
json.loads(json_str)
return json_str
except json.JSONDecodeError:
continue
try:
json_match = re.search(r'\{.*\}', response, re.DOTALL)
if json_match:
json_str = json_match.group(0).strip()
json.loads(json_str)
return json_str
except json.JSONDecodeError:
pass
# Try array
try:
json_match = re.search(r'\[.*\]', response, re.DOTALL)
if json_match:
json_str = json_match.group(0).strip()
json.loads(json_str)
return json_str
except json.JSONDecodeError:
pass
raise ValueError("No valid JSON found in AI response.")
def convert_xml_to_json(
self,
xml_text: str,
temperature: float = 0.9,
top_p: float = 0.95,
max_tokens: Optional[int] = None
) -> str:
"""Convert XML text to Board JSON text using pooled connection."""
chat_history = [
{"role": "system", "content": self.system_prompt}
]
ai_response = self._call_ai_api(
user_input=xml_text,
chat_history=chat_history,
temperature=temperature,
top_p=top_p,
max_tokens=max_tokens
)
print(xml_text[:200] + ("..." if len(xml_text) > 200 else ""))
json_text = self._extract_json_from_response(ai_response)
return json_text |