From 17bee2f91889b28c6fb35bcaf9b0e7111356968c Mon Sep 17 00:00:00 2001 From: HuangHai <10402852@qq.com> Date: Mon, 1 Sep 2025 07:40:41 +0800 Subject: [PATCH] 'commit' --- dsLightRag/Util/ShiTiRecognizer.py | 191 +++++++++++++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 dsLightRag/Util/ShiTiRecognizer.py diff --git a/dsLightRag/Util/ShiTiRecognizer.py b/dsLightRag/Util/ShiTiRecognizer.py new file mode 100644 index 00000000..22d70cdd --- /dev/null +++ b/dsLightRag/Util/ShiTiRecognizer.py @@ -0,0 +1,191 @@ +import json +import logging +from typing import Dict, Any + +from alibabacloud_credentials.client import Client as CredentialClient +from alibabacloud_credentials.models import Config as CredentialConfig +from alibabacloud_ocr_api20210707 import models as OcrModels +from alibabacloud_ocr_api20210707.client import Client as OcrClient +from alibabacloud_tea_openapi import models as OpenApiModels +from alibabacloud_tea_util import models as UtilModels +from alibabacloud_tea_util.client import Client as UtilClient + +from Config.Config import ALY_AK, ALY_SK + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + + +class ShiTiRecognizer: + """阿里云OCR试题识别服务封装类""" + + # 默认配置 + DEFAULT_ENDPOINT = 'ocr-api.cn-hangzhou.aliyuncs.com' + + def __init__(self, access_key_id: str = None, access_key_secret: str = None, endpoint: str = None): + """ + 初始化试题识别器 + + Args: + access_key_id: 阿里云访问密钥ID + access_key_secret: 阿里云访问密钥Secret + endpoint: OCR服务端点 + """ + self.access_key_id = access_key_id or ALY_AK + self.access_key_secret = access_key_secret or ALY_SK + self.endpoint = endpoint or self.DEFAULT_ENDPOINT + self._client = None + + @property + def client(self) -> OcrClient: + """懒加载方式创建OCR客户端""" + if self._client is None: + self._client = self._create_client() + return self._client + + def _create_client(self) -> OcrClient: + """创建OCR客户端""" + try: + credential_config = CredentialConfig( + type='access_key', + access_key_id=self.access_key_id, + access_key_secret=self.access_key_secret + ) + credential = CredentialClient(config=credential_config) + config = OpenApiModels.Config( + credential=credential, + endpoint=self.endpoint + ) + return OcrClient(config) + except Exception as e: + logger.error(f"创建OCR客户端失败: {str(e)}") + raise + + def recognize_question(self, image_url: str) -> Dict[str, Any]: + """ + 识别图片中的试题内容 + + Args: + image_url: 图片URL + + Returns: + 识别结果字典 + """ + if not image_url: + raise ValueError("图片URL不能为空") + + logger.info(f"开始识别试题，图片URL: {image_url}") + + request = OcrModels.RecognizeEduQuestionOcrRequest() + request.url = image_url + runtime = UtilModels.RuntimeOptions() + + try: + response = self.client.recognize_edu_question_ocr_with_options(request, runtime) + result = self._parse_response(response) + logger.info("试题识别成功") + return result + except Exception as error: + logger.error(f"试题识别失败: {str(error)}") + self._handle_error(error) + return {"error": str(error)} + + def _parse_response(self, response) -> Dict[str, Any]: + """解析API响应""" + if not response or not response.body: + return {"error": "API返回空响应"} + + try: + # 获取响应体对象 + body = response.body + + # 检查响应体是否有to_map方法，这是阿里云SDK中常用的对象转换方法 + if hasattr(body, 'to_map'): + body_map = body.to_map() + + # 检查Data字段是否为字符串，如果是则尝试解析为JSON + if "Data" in body_map and isinstance(body_map["Data"], str): + try: + body_map["Data"] = json.loads(body_map["Data"]) + except json.JSONDecodeError: + logger.warning("Data字段不是有效的JSON字符串") + + return body_map + + # 如果没有to_map方法，尝试直接获取属性 + result = { + "request_id": getattr(body, 'request_id', ''), + "code": getattr(body, 'code', ''), + "message": getattr(body, 'message', ''), + "Data": None + } + + # 处理Data字段 + data = getattr(body, 'data', None) + if data: + # 如果Data对象有to_map方法，使用它 + if hasattr(data, 'to_map'): + result["Data"] = data.to_map() + else: + # 否则手动构建Data字典 + result["Data"] = { + "content": getattr(data, 'content', ''), + "score": getattr(data, 'score', 0), + "question_info": getattr(data, 'question_info', ''), + "angle": getattr(data, 'angle', 0) + } + + return result + except Exception as e: + logger.warning(f"响应解析失败，返回原始数据: {str(e)}") + return {"raw_data": str(response.body)} + + def _handle_error(self, error: Exception) -> None: + """处理API错误""" + error_message = getattr(error, 'message', str(error)) + logger.error(f"错误信息: {error_message}") + + if hasattr(error, 'data') and error.data: + recommend = error.data.get("Recommend") + if recommend: + logger.info(f"诊断建议: {recommend}") + + # 在实际项目中，这里可以添加更复杂的错误处理逻辑 + UtilClient.assert_as_string(error_message) + + + +if __name__ == '__main__': + """主函数，演示试题识别功能""" + try: + recognizer = ShiTiRecognizer() + # 传入固定的图片URL + image_url = "https://dsideal.obs.cn-north-1.myhuaweicloud.com/HuangHai/Backup/ShiTi.jpg" + result = recognizer.recognize_question(image_url) + + print("识别结果:") + print(json.dumps(result, indent=2, ensure_ascii=False)) + + # 如果需要，可以在这里添加结果处理逻辑 + if "Data" in result and result["Data"]: + # 尝试多种方式获取content字段 + content = "" + if isinstance(result["Data"], dict): + content = result["Data"].get("content", "") + elif hasattr(result["Data"], "get"): + content = result["Data"].get("content", "") + + if content: + print(f"\n识别的试题内容: {content}") + else: + print("\n未识别到试题内容") + else: + print("\n响应中没有有效的试题数据") + + except Exception as e: + logger.error(f"程序执行失败: {str(e)}") + print(f"错误: {str(e)}")