diff --git a/apps/api/app/services/jobs/result_projection.py b/apps/api/app/services/jobs/result_projection.py index 199625580..b39166c9d 100644 --- a/apps/api/app/services/jobs/result_projection.py +++ b/apps/api/app/services/jobs/result_projection.py @@ -15,6 +15,7 @@ JobStatusValue = Literal[ "pending", "waiting-file", "running", "converting", "done", "failed" ] +_CHARGED_BILLING_STATUS: str = "charged" def build_error_response( @@ -112,6 +113,17 @@ def _resolve_duration_seconds(job: Any) -> float | None: return None +def _resolve_credits_spent(job: Any) -> float: + if to_job_status_value(job.status) == "failed": + return 0.0 + + billing_status = getattr(job, "billing_status", None) + if billing_status != _CHARGED_BILLING_STATUS: + return 0.0 + + return MicroDollar(getattr(job, "credits_charged", 0) or 0).to_credit() + + async def _resolve_result_delivery( job: Any, ) -> tuple[dict[str, Any] | None, str | None, datetime]: @@ -162,9 +174,5 @@ async def build_job_result_response( model=parsing_params.get("model"), ocr_enabled=parsing_params.get("ocr_enabled"), duration_seconds=_resolve_duration_seconds(job), - credits_spent=( - MicroDollar(job.credits_charged).to_credit() - if hasattr(job, "credits_charged") - else 0 - ), + credits_spent=_resolve_credits_spent(job), ) diff --git a/apps/api/tests/contract/test_job_read_contract.py b/apps/api/tests/contract/test_job_read_contract.py index 7e52f3b68..ef6c68617 100644 --- a/apps/api/tests/contract/test_job_read_contract.py +++ b/apps/api/tests/contract/test_job_read_contract.py @@ -132,6 +132,46 @@ async def test_should_return_job_details_for_an_existing_waiting_file_job( assert response_json["credits_spent"] == 0.0 +@pytest.mark.asyncio +async def test_should_report_zero_credits_spent_for_refunded_failed_job( + developer_api_client_factory: Callable[ + [], AbstractAsyncContextManager[AsyncClient] + ], +) -> None: + async with developer_api_client_factory() as api_client: + created_job = await _create_waiting_file_job(api_client) + job_id = cast(str, created_job["job_id"]) + + await ContractDatabase.execute( + """ + UPDATE jobs + SET + status = 'failed', + error_code = 'INVALID_ARGUMENT', + error_message = 'Invalid file: the uploaded .docx file is not a valid Word document. Please check the file and upload again.', + credits_charged = 15000, + billing_status = 'refunded' + WHERE job_id = :job_id + """, + {"job_id": job_id}, + ) + + response = await api_client.get(f"/api/v1/jobs/{job_id}") + + assert response.status_code == 200 + + response_json = cast(dict[str, object], response.json()) + error = cast(dict[str, object], response_json["error"]) + + assert response_json["status"] == "failed" + assert error["code"] == "INVALID_ARGUMENT" + assert ( + error["message"] + == "Invalid file: the uploaded .docx file is not a valid Word document. Please check the file and upload again." + ) + assert response_json["credits_spent"] == 0.0 + + @pytest.mark.asyncio async def test_should_return_not_found_when_requesting_an_unknown_job( developer_api_client_factory: Callable[ diff --git a/apps/worker/app/services/document_parser/orchestration/office_container_validator.py b/apps/worker/app/services/document_parser/orchestration/office_container_validator.py new file mode 100644 index 000000000..3009603c1 --- /dev/null +++ b/apps/worker/app/services/document_parser/orchestration/office_container_validator.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +import zipfile +from dataclasses import dataclass + +from app.services.document_parser.orchestration.format_router import DocumentFormat + +from shared.core.exceptions.domain_exceptions import ValidationException + + +@dataclass(frozen=True) +class _OfficeContainerRequirement: + extension: str + document_label: str + required_member: str + + @property + def user_message(self) -> str: + return ( + f"Invalid file: the uploaded {self.extension} file is not a valid " + f"{self.document_label}. Please check the file and upload again." + ) + + @property + def violation_description(self) -> str: + return ( + f"Expected a valid {self.extension[1:].upper()} ZIP package containing " + f"{self.required_member}" + ) + + +_CONTENT_TYPES_MEMBER: str = "[Content_Types].xml" +_OFFICE_CONTAINER_REQUIREMENTS: dict[ + DocumentFormat, + _OfficeContainerRequirement, +] = { + DocumentFormat.DOCX: _OfficeContainerRequirement( + extension=".docx", + document_label="Word document", + required_member="word/document.xml", + ), + DocumentFormat.XLSX: _OfficeContainerRequirement( + extension=".xlsx", + document_label="Excel workbook", + required_member="xl/workbook.xml", + ), + DocumentFormat.PPTX: _OfficeContainerRequirement( + extension=".pptx", + document_label="PowerPoint presentation", + required_member="ppt/presentation.xml", + ), +} + + +def validate_office_container( + file_path: str, + document_format: DocumentFormat, +) -> None: + """Validate OOXML containers before dispatching to archive-based parsers.""" + requirement = _OFFICE_CONTAINER_REQUIREMENTS.get(document_format) + if requirement is None: + return + + if not zipfile.is_zipfile(file_path): + _raise_invalid_office_file(requirement) + + try: + with zipfile.ZipFile(file_path, "r") as archive: + member_names = set(archive.namelist()) + except zipfile.BadZipFile as exc: + raise _build_invalid_office_file_exception(requirement) from exc + + if ( + _CONTENT_TYPES_MEMBER not in member_names + or requirement.required_member not in member_names + ): + _raise_invalid_office_file(requirement) + + +def _raise_invalid_office_file(requirement: _OfficeContainerRequirement) -> None: + raise _build_invalid_office_file_exception(requirement) + + +def _build_invalid_office_file_exception( + requirement: _OfficeContainerRequirement, +) -> ValidationException: + return ValidationException( + user_message=requirement.user_message, + violations=[ + { + "field": "file", + "description": requirement.violation_description, + } + ], + ) + + +__all__ = ["validate_office_container"] diff --git a/apps/worker/app/services/document_parser/orchestration/route_parse.py b/apps/worker/app/services/document_parser/orchestration/route_parse.py index 68bb57e73..1c05fd442 100644 --- a/apps/worker/app/services/document_parser/orchestration/route_parse.py +++ b/apps/worker/app/services/document_parser/orchestration/route_parse.py @@ -2,6 +2,9 @@ get_document_parse_adapter, resolve_document_format, ) +from app.services.document_parser.orchestration.office_container_validator import ( + validate_office_container, +) from app.services.document_parser.orchestration.parse_output import ParseOutput from app.services.document_parser.orchestration.parse_session import ParseSession @@ -9,5 +12,6 @@ def route_document_parse(session: ParseSession) -> ParseOutput: """Route a parser session to the correct adapter and return its output.""" document_format = resolve_document_format(session.file_full_path) + validate_office_container(session.file_full_path, document_format) adapter = get_document_parse_adapter(document_format) return adapter.parse(session) diff --git a/apps/worker/tests/contract/test_parse_task_contract.py b/apps/worker/tests/contract/test_parse_task_contract.py index 59cde7310..696c5e766 100644 --- a/apps/worker/tests/contract/test_parse_task_contract.py +++ b/apps/worker/tests/contract/test_parse_task_contract.py @@ -483,6 +483,58 @@ def test_parse_task_should_refund_charged_job_when_uploaded_file_cannot_be_parse ] +def test_parse_task_should_report_invalid_docx_as_client_file_error( + worker_contract_environment: None, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + contract = WorkerParseContract.create() + contract.use_workspace_root(monkeypatch, tmp_path) + contract.use_billing(monkeypatch, is_enabled=True) + + invalid_docx_path = tmp_path / "invalid.docx" + invalid_docx_path.write_bytes(b"this is not a docx package") + job = contract.create_file_job( + source_file_name="contract-invalid.docx", + job_id_prefix="job_invalid_docx", + ) + contract.upload_source_file( + local_file_path=invalid_docx_path, + s3_key=job["s3_key"], + ) + + celery_result = contract.enqueue_parse_task( + job_id=job["job_id"], + user_id=job["user_id"], + ) + + assert celery_result.failed() + assert contract.find_task_workspaces(tmp_path, job["job_id"]) == [] + + job_row = contract.observe_job_status(job["job_id"]) + assert job_row["status"] == "failed" + assert job_row["billing_status"] == "refunded" + assert job_row["credits_charged"] == int(contract.settings.MICRO_DOLLARS_PER_PAGE) + assert job_row["error_code"] == "INVALID_ARGUMENT" + assert ( + job_row["error_message"] + == "Invalid file: the uploaded .docx file is not a valid Word document. Please check the file and upload again." + ) + assert contract.count_job_results(job["job_id"]) == 0 + + metadata = contract.get_job_metadata(job["job_id"]) + assert metadata["error_details"] == { + "violations": [ + { + "field": "file", + "description": ( + "Expected a valid DOCX ZIP package containing word/document.xml" + ), + } + ] + } + + def test_should_reject_pdf_when_page_count_exceeds_configured_limit( worker_contract_environment: None, monkeypatch: pytest.MonkeyPatch, diff --git a/packages/shared-python/shared/core/state_machine/service_sync.py b/packages/shared-python/shared/core/state_machine/service_sync.py index 6d8ba1cc1..c820ef4b0 100644 --- a/packages/shared-python/shared/core/state_machine/service_sync.py +++ b/packages/shared-python/shared/core/state_machine/service_sync.py @@ -36,7 +36,10 @@ ) from shared.models.database.job import Job from shared.models.database.job_state_audit_log import JobStateAuditLog -from shared.services.redis.redis_sync_service import SyncRedisServiceFactory +from shared.services.redis.redis_sync_service import ( + SyncJobMetadataService, + SyncRedisServiceFactory, +) from shared.services.redis.key_builder import RedisKeyType, redis_key_builder @@ -412,6 +415,7 @@ def _update_job_error( "error_message": error_message, "error_code": error_code, } + metadata_updates: Dict[str, Any] = {} if error_details: import json as _json @@ -425,8 +429,14 @@ def _update_job_error( pg_array(["error_details"]), cast(literal(_json.dumps(error_details)), JSONB), ) + metadata_updates["error_details"] = error_details db.execute(update(Job).where(Job.job_id == job_id).values(**update_values)) + if metadata_updates: + SyncJobMetadataService(self.redis).update_metadata( + job_id, + metadata_updates, + ) except Exception as e: logger.error(f"Failed to update Job {job_id} error info: {e}")