Spaces:
Running
Running
File size: 2,618 Bytes
2ecc4a7 | 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 | import re
import json
import json_repair
import threading
import html
from typing import Any, Dict, Tuple, Optional
import logging
logger = logging.getLogger(__name__)
def repair_json_with_module(json_content: str) -> Optional[Any]:
result_container = [None]
exception_container = [None]
def repair_thread():
try:
result_container[0] = json_repair.loads(json_content)
except Exception as e:
exception_container[0] = e
thread = threading.Thread(target=repair_thread)
thread.daemon = True
thread.start()
thread.join(timeout=10)
if thread.is_alive():
logger.warning("TIMEOUT: JSON repair took longer than 10 seconds")
return None
if exception_container[0]:
logger.warning(f"JSON repair failed: {exception_container[0]}")
return None
return result_container[0]
def extract_json_block(response_text: str) -> str:
"""Extracts JSON block from response text intelligently."""
# 1. Look for ```json ... ```
match = re.search(r"```json\s*([\s\S]*?)\s*```", response_text, re.IGNORECASE)
if match:
return match.group(1).strip()
# 2. Look for ``` ... ``` (optional json tag)
match = re.search(r"```\s*(?:json)?\s*([\s\S]*?)\s*```", response_text, re.IGNORECASE)
if match:
candidate = match.group(1).strip()
if candidate.lower().startswith('json'):
candidate = candidate[4:].strip()
return candidate
# 3. Look for **Answer**: ...
answer_match = re.search(r'\*\*Answer\*\*:\s*([\s\S]*)', response_text, re.IGNORECASE)
if answer_match:
return answer_match.group(1).strip()
# 4. Fallback to finding first { and last }
first_brace = response_text.find('{')
last_brace = response_text.rfind('}')
if first_brace != -1 and last_brace != -1 and last_brace > first_brace:
return response_text[first_brace:last_brace + 1]
return response_text.strip()
def repair_json(json_str: str) -> Optional[Dict[str, Any]]:
"""Combines extraction, cleaning and repair."""
try:
# Clean HTML entities and tags
json_str = html.unescape(json_str)
json_str = re.sub(r"<br\s*/?>", "\n", json_str)
json_str = json_str.strip()
# Try standard parse
try:
return json.loads(json_str)
except:
pass
# Try repair
return repair_json_with_module(json_str)
except Exception as e:
logger.error(f"Ultimate JSON repair failed: {e}")
return None
|