222 lines
8.2 KiB
Python
222 lines
8.2 KiB
Python
|
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 FormulaRecognizer:
|
|||
|
|
"""阿里云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_formula(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.RecognizeEduFormulaRequest()
|
|||
|
|
request.url = image_url
|
|||
|
|
runtime = UtilModels.RuntimeOptions()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
response = self.client.recognize_edu_formula_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),
|
|||
|
|
"formula_info": getattr(data, 'formula_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)
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
async def recognize_formula_async(image_url: str) -> Dict[str, Any]:
|
|||
|
|
"""
|
|||
|
|
异步识别图片中的数学公式
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
image_url: 图片URL
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
识别结果字典
|
|||
|
|
"""
|
|||
|
|
if not image_url:
|
|||
|
|
raise ValueError("图片URL不能为空")
|
|||
|
|
|
|||
|
|
recognizer = FormulaRecognizer()
|
|||
|
|
logger.info(f"开始异步识别公式，图片URL: {image_url}")
|
|||
|
|
|
|||
|
|
request = OcrModels.RecognizeEduFormulaRequest()
|
|||
|
|
request.url = image_url
|
|||
|
|
runtime = UtilModels.RuntimeOptions()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
response = await recognizer.client.recognize_edu_formula_with_options_async(request, runtime)
|
|||
|
|
result = recognizer._parse_response(response)
|
|||
|
|
logger.info("异步公式识别成功")
|
|||
|
|
return result
|
|||
|
|
except Exception as error:
|
|||
|
|
logger.error(f"异步公式识别失败: {str(error)}")
|
|||
|
|
recognizer._handle_error(error)
|
|||
|
|
return {"error": str(error)}
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == '__main__':
|
|||
|
|
"""主函数，演示公式识别功能"""
|
|||
|
|
try:
|
|||
|
|
recognizer = FormulaRecognizer()
|
|||
|
|
# 传入固定的图片URL
|
|||
|
|
image_url = "https://dsideal.obs.cn-north-1.myhuaweicloud.com/HuangHai/Backup/GongShi.jpg"
|
|||
|
|
result = recognizer.recognize_formula(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)}")
|