data-str / json_processor.py
CORVO-AI's picture
Upload 15 files
d2a178c verified
import requests
import json
import re
from typing import Optional, Dict, Any, List
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:
# Provide a default system prompt if file not found
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 and return the assistant's response"""
payload = {
"user_input": user_input,
"chat_history": chat_history or [],
"temperature": temperature,
"top_p": top_p,
"max_tokens": max_tokens
}
try:
response = requests.post(self.api_url, json=payload, timeout=120)
response.raise_for_status()
result = response.json()
return result.get("assistant_response", "")
except requests.exceptions.RequestException 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 and return as text"""
# Print the raw response for debugging
print("\n๐Ÿ” DEBUG - Raw AI Response:")
print("=" * 80)
print(response)
print("=" * 80)
# Try multiple patterns to find JSON
patterns = [
r'<json>(.*?)</json>', # Standard <json></json> tags
r'```json\s*(.*?)\s*```', # Markdown code blocks
r'```\s*(.*?)\s*```', # Generic code blocks
]
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) # Validate
return json_str
except json.JSONDecodeError:
continue
# If no tags found, try to find JSON directly in the response
try:
# Look for JSON object pattern
json_match = re.search(r'\{.*\}', response, re.DOTALL)
if json_match:
json_str = json_match.group(0).strip()
json.loads(json_str) # Validate
return json_str
except json.JSONDecodeError:
pass
raise ValueError("No valid JSON found in AI response. Check the debug output above.")
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
Args:
xml_text: The XML string to convert
temperature: AI temperature parameter (default: 0.9)
top_p: AI top_p parameter (default: 0.95)
max_tokens: Maximum tokens for AI response (default: None)
Returns:
JSON string (as text)
"""
# Prepare chat history with system prompt
chat_history = [
{"role": "system", "content": self.system_prompt}
]
# Call AI API with XML input
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)
# Extract and return JSON as text
json_text = self._extract_json_from_response(ai_response)
return json_text
# Example usage
if __name__ == "__main__":
processor = BoardProcessor()
xml_input = """
<board>
<note color="yellow" x="100" y="200" w="280" h="140">ู‚ุงู†ูˆู† ู†ูŠูˆุชู† ุงู„ุซุงู†ูŠ:
ุงู„ู‚ูˆุฉ = ุงู„ูƒุชู„ุฉ ร— ุงู„ุชุณุงุฑุน
F = m ร— a</note>
<svg>ball</svg>
</board>
"""
try:
json_output = processor.convert_xml_to_json(xml_input)
print("\nโœ… Conversion successful!")
print("\nJSON Output (as text):")
print(json_output)
# Optionally pretty print
print("\n๐Ÿ“‹ Pretty printed:")
print(json.dumps(json.loads(json_output), indent=2, ensure_ascii=False))
except Exception as e:
print(f"\nโŒ Error: {str(e)}")