File size: 1,475 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 | """Module for tasks."""
from datetime import date
from typing import Any
from apps.client.services.identity_extraction import IdentityExtractionService
from apps.client.services.storage import delete_media_file, to_media_abs
def execute_identity_doc_recognition(file_path: str, doc_type: str) -> dict[str, Any]:
abs_path = to_media_abs(file_path)
try:
content = abs_path.read_bytes()
service = IdentityExtractionService()
result = service.extract(content, doc_type)
return {
"doc_type": result.doc_type,
"extracted_data": result.extracted_data,
"confidence": result.confidence,
}
finally:
delete_media_file(file_path)
def recognize_expiry_date_task(doc_id: int) -> dict[str, Any]:
"""识别证件到期日期并更新数据库。"""
from apps.client.services.client_identity_doc_service import ClientIdentityDocService
doc_service = ClientIdentityDocService()
doc = doc_service.get_identity_doc(doc_id)
abs_path = to_media_abs(doc.file_path)
content = abs_path.read_bytes()
service = IdentityExtractionService()
result = service.extract(content, doc.doc_type)
expiry_str: str | None = result.extracted_data.get("expiry_date")
if expiry_str:
expiry_date = date.fromisoformat(expiry_str)
doc_service.update_expiry_date(doc_id, expiry_date)
return {"status": "success", "doc_id": doc_id, "expiry_date": expiry_str}
|