File size: 6,269 Bytes
b4f2238 | 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 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | """响应解析 Mixin"""
import json
import logging
import re
from datetime import datetime
from typing import Any, cast
logger = logging.getLogger("apps.document_recognition")
class ResponseParserMixin:
"""Ollama 响应解析 Mixin"""
def _normalize_case_number(self, case_number: str) -> str:
raise NotImplementedError
def _parse_summons_response(self, response: dict[str, Any]) -> dict[str, Any]:
"""解析传票信息提取响应"""
result: dict[str, Any] = {"case_number": None, "court_time": None}
try:
if "message" not in response or "content" not in response["message"]:
logger.warning("Ollama 响应格式异常")
return result
content = response["message"]["content"]
parsed = self._extract_json_from_response(content)
if parsed is None:
logger.warning(f"无法从响应中提取 JSON: {content[:200]}")
return result
case_number = parsed.get("case_number")
if case_number and case_number.lower() != "null":
result["case_number"] = self._normalize_case_number(case_number)
court_time = parsed.get("court_time")
if court_time and court_time.lower() != "null":
result["court_time"] = self._parse_datetime(court_time)
return result
except Exception as e:
logger.warning(f"解析传票响应失败: {e!s}")
return result
def _parse_execution_response(self, response: dict[str, Any]) -> dict[str, Any]:
"""解析执行裁定书信息提取响应"""
result: dict[str, Any] = {"case_number": None, "preservation_deadline": None}
try:
if "message" not in response or "content" not in response["message"]:
logger.warning("Ollama 响应格式异常")
return result
content = response["message"]["content"]
parsed = self._extract_json_from_response(content)
if parsed is None:
logger.warning(f"无法从响应中提取 JSON: {content[:200]}")
return result
case_number = parsed.get("case_number")
if case_number and case_number.lower() != "null":
result["case_number"] = self._normalize_case_number(case_number)
deadline = parsed.get("preservation_deadline")
if deadline and deadline.lower() != "null":
result["preservation_deadline"] = self._parse_date(deadline)
return result
except Exception as e:
logger.warning(f"解析执行裁定书响应失败: {e!s}")
return result
def _extract_json_from_response(self, content: str) -> dict[str, Any] | None:
"""从响应内容中提取 JSON"""
content = content.strip()
try:
return cast(dict[str, Any], json.loads(content))
except json.JSONDecodeError:
pass
start_idx = content.find("{")
end_idx = content.rfind("}")
if start_idx != -1 and end_idx != -1 and end_idx > start_idx:
try:
return cast(dict[str, Any], json.loads(content[start_idx : end_idx + 1]))
except json.JSONDecodeError:
pass
if "```json" in content:
try:
json_start = content.index("```json") + 7
json_end = content.index("```", json_start)
return cast(dict[str, Any], json.loads(content[json_start:json_end].strip()))
except (ValueError, json.JSONDecodeError):
pass
if "```" in content:
try:
json_start = content.index("```") + 3
json_end = content.index("```", json_start)
return cast(dict[str, Any], json.loads(content[json_start:json_end].strip()))
except (ValueError, json.JSONDecodeError):
pass
return None
def _parse_datetime(self, datetime_str: str) -> datetime | None:
"""解析日期时间字符串"""
if not datetime_str:
return None
datetime_str = datetime_str.strip()
formats = [
"%Y-%m-%d %H:%M",
"%Y-%m-%d %H:%M:%S",
"%Y年%m月%d日 %H:%M",
"%Y年%m月%d日 %H时%M分",
"%Y年%m月%d日%H时%M分",
"%Y/%m/%d %H:%M",
"%Y.%m.%d %H:%M",
]
for fmt in formats:
try:
return datetime.strptime(datetime_str, fmt)
except ValueError:
continue
pattern_cn = r"(\d{4})年(\d{1,2})月(\d{1,2})日\s*(\d{1,2})时(\d{1,2})分?"
match = re.search(pattern_cn, datetime_str)
if match:
try:
year, month, day, hour, minute = map(int, match.groups())
return datetime(year, month, day, hour, minute)
except ValueError:
pass
pattern_std = r"(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})\s+(\d{1,2}):(\d{1,2})"
match = re.search(pattern_std, datetime_str)
if match:
try:
year, month, day, hour, minute = map(int, match.groups())
return datetime(year, month, day, hour, minute)
except ValueError:
pass
logger.warning(f"无法解析日期时间: {datetime_str}")
return None
def _parse_date(self, date_str: str) -> datetime | None:
"""解析日期字符串"""
if not date_str:
return None
date_str = date_str.strip()
formats = ["%Y-%m-%d", "%Y年%m月%d日", "%Y/%m/%d", "%Y.%m.%d"]
for fmt in formats:
try:
return datetime.strptime(date_str, fmt)
except ValueError:
continue
pattern = r"(\d{4})[-年/.](\d{1,2})[-月/.](\d{1,2})"
match = re.search(pattern, date_str)
if match:
try:
year, month, day = map(int, match.groups())
return datetime(year, month, day)
except ValueError:
pass
logger.warning(f"无法解析日期: {date_str}")
return None
|