File size: 7,053 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 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | """
文档处理服务适配器
提供文档处理服务的统一接口实现
"""
from typing import Any
from apps.core.interfaces import IDocumentProcessingService
class DocumentProcessingServiceAdapter(IDocumentProcessingService):
"""
文档处理服务适配器
实现 IDocumentProcessingService 接口，提供文档处理的核心功能
"""
def __init__(self) -> None:
"""
初始化文档处理服务适配器
支持依赖注入模式，所有依赖通过延迟加载获取
"""
pass
def extract_text_from_pdf(
self, file_path: str, limit: int | None = None, preview_page: int | None = None
) -> dict[str, Any]:
"""
从PDF文件提取文本
Args:
file_path: PDF文件路径
limit: 文本长度限制
preview_page: 预览页码
Returns:
包含文本和预览图的字典
"""
try:
from apps.automation.services.document.document_processing import process_pdf
image_url, text = process_pdf(file_path, limit, preview_page)
return {"text": text, "image_url": image_url, "file_path": file_path, "file_type": "pdf"}
except Exception as e:
from apps.core.exceptions import ValidationException
raise ValidationException(
message=f"PDF文件处理失败: {e}",
code="PDF_PROCESSING_FAILED",
errors={"error_message": str(e)},
) from e
def extract_text_from_docx(self, file_path: str, limit: int | None = None) -> str:
"""
从DOCX文件提取文本
Args:
file_path: DOCX文件路径
limit: 文本长度限制
Returns:
提取的文本内容
"""
try:
from apps.automation.services.document.document_processing import extract_docx_text
return extract_docx_text(file_path, limit=limit)
except Exception as e:
from apps.core.exceptions import ValidationException
raise ValidationException(
message=f"DOCX文件处理失败: {e}",
code="DOCX_PROCESSING_FAILED",
errors={"error_message": str(e)},
) from e
def extract_text_from_image(self, file_path: str, limit: int | None = None) -> str:
"""
从图片文件提取文本（OCR）
Args:
file_path: 图片文件路径
limit: 文本长度限制
Returns:
OCR识别的文本内容
"""
try:
from apps.automation.services.document.document_processing import extract_text_from_image_with_rapidocr
text = extract_text_from_image_with_rapidocr(file_path)
if limit is not None and len(text) > limit:
text = text[:limit]
return text
except Exception as e:
from apps.core.exceptions import ValidationException
raise ValidationException(
message=f"图片OCR处理失败: {e}",
code="IMAGE_OCR_FAILED",
errors={"error_message": str(e)},
) from e
def process_uploaded_document(
self, uploaded_file: Any, limit: int | None = None, preview_page: int | None = None
) -> dict[str, Any]:
"""
处理上传的文档
Args:
uploaded_file: 上传的文件对象
limit: 文本长度限制
preview_page: 预览页码
Returns:
包含文本和预览信息的字典
"""
try:
from apps.automation.services.document.document_processing import process_uploaded_document
result = process_uploaded_document(uploaded_file, limit=limit, preview_page=preview_page)
return {
"text": result.text,
"image_url": result.image_url,
"file_name": uploaded_file.name if hasattr(uploaded_file, "name") else None,
"file_size": uploaded_file.size if hasattr(uploaded_file, "size") else None,
}
except Exception as e:
from apps.core.exceptions import ValidationException
raise ValidationException(
message=f"文档内容提取失败: {e}",
code="DOCUMENT_CONTENT_EXTRACTION_FAILED",
errors={"error_message": str(e)},
) from e
def extract_document_content_by_path(
self, file_path: str, limit: int | None = None, preview_page: int | None = None
) -> dict[str, Any]:
"""
根据文件路径提取文档内容
Args:
file_path: 文件路径
limit: 文本长度限制
preview_page: 预览页码
Returns:
包含文本和预览信息的字典
"""
try:
from apps.automation.services.document.document_processing import extract_document_content
result = extract_document_content(file_path, limit=limit, preview_page=preview_page)
return {"text": result.text, "image_url": result.image_url, "file_path": file_path}
except Exception as e:
from apps.core.exceptions import ValidationException
raise ValidationException(
message=f"文档内容提取失败: {e!s}",
code="DOCUMENT_CONTENT_EXTRACTION_FAILED",
errors={"file_path": file_path, "error": str(e)},
) from e
# 内部方法版本，供其他模块调用
def extract_text_from_pdf_internal(
self, file_path: str, limit: int | None = None, preview_page: int | None = None
) -> dict[str, Any]:
"""从PDF文件提取文本（内部接口，无权限检查）"""
return self.extract_text_from_pdf(file_path, limit, preview_page)
def extract_text_from_docx_internal(self, file_path: str, limit: int | None = None) -> str:
"""从DOCX文件提取文本（内部接口，无权限检查）"""
return self.extract_text_from_docx(file_path, limit)
def extract_text_from_image_internal(self, file_path: str, limit: int | None = None) -> str:
"""从图片文件提取文本（内部接口，无权限检查）"""
return self.extract_text_from_image(file_path, limit)
def process_uploaded_document_internal(
self, uploaded_file: Any, limit: int | None = None, preview_page: int | None = None
) -> dict[str, Any]:
"""处理上传的文档（内部接口，无权限检查）"""
return self.process_uploaded_document(uploaded_file, limit, preview_page)
def extract_document_content_by_path_internal(
self, file_path: str, limit: int | None = None, preview_page: int | None = None
) -> dict[str, Any]:
"""根据文件路径提取文档内容（内部接口，无权限检查）"""
return self.extract_document_content_by_path(file_path, limit, preview_page)
|