diff --git a/app/adapters/delivery_providers/cse/client.py b/app/adapters/delivery_providers/cse/client.py index 73c4915..91e940c 100644 --- a/app/adapters/delivery_providers/cse/client.py +++ b/app/adapters/delivery_providers/cse/client.py @@ -21,12 +21,17 @@ from app.adapters.delivery_providers.cse.mapper import ( map_cse_calc_response_for_tariff_code, ) from app.adapters.delivery_providers.cse.order_mapper import ( + CSEOrderInfo, CSEOrderRegistrationParams, CSEOrderRegistrationResult, build_calc_body_for_calculation, build_calc_body_for_payment, + build_tracking_body_for_order, + build_waybill_print_form_body, + map_cse_tracking_response, map_cse_save_order_request, map_cse_save_order_response, + map_cse_waybill_print_form_response, split_tariff_code, ) from app.adapters.delivery_providers.cse.soap import ( @@ -113,6 +118,30 @@ class CSEClient: except CSEMappingError as exc: raise CSEClientError(str(exc)) from exc + async def get_order(self, order_number: str) -> CSEOrderInfo: + root = await self._post( + "Tracking", + build_tracking_body_for_order(order_number), + request_error_message="CSE Tracking request was rejected with status", + ) + try: + return map_cse_tracking_response(root) + except CSEMappingError as exc: + raise CSEClientError(str(exc)) from exc + + async def download_waybill_pdf(self, waybill_number: str) -> bytes: + root = await self._post( + "GetFormsForDocuments", + build_waybill_print_form_body(waybill_number), + request_error_message=( + "CSE GetFormsForDocuments request was rejected with status" + ), + ) + try: + return map_cse_waybill_print_form_response(root) + except CSEMappingError as exc: + raise CSEClientError(str(exc)) from exc + async def _post( self, operation: str, @@ -152,7 +181,8 @@ class CSEClient: status_code=response.status_code, ) raise CSEClientError( - f"CSE {operation} request failed with status {response.status_code}." + f"CSE {operation} request failed with status " + f"{response.status_code}." ) if 400 <= response.status_code < 500: @@ -312,3 +342,9 @@ class CSEProvider(DeliveryProvider): self, request: InitPaymentRequest, order_uuid: str ) -> CSEOrderRegistrationResult: return await self._client.save_order(request, order_uuid) + + async def get_order(self, order_number: str) -> CSEOrderInfo: + return await self._client.get_order(order_number) + + async def download_waybill_pdf(self, waybill_number: str) -> bytes: + return await self._client.download_waybill_pdf(waybill_number) diff --git a/app/adapters/delivery_providers/cse/order_mapper.py b/app/adapters/delivery_providers/cse/order_mapper.py index bf62b73..bc39c44 100644 --- a/app/adapters/delivery_providers/cse/order_mapper.py +++ b/app/adapters/delivery_providers/cse/order_mapper.py @@ -1,5 +1,7 @@ -"""CSE Calc and SaveDocuments payload mappers.""" +"""CSE Calc, SaveDocuments and waybill payload mappers.""" +import base64 +import binascii from dataclasses import dataclass from app.adapters.delivery_providers.cse.constants import ( @@ -32,6 +34,13 @@ class CSEOrderRegistrationResult: order_number: str +@dataclass(frozen=True) +class CSEOrderInfo: + order_number: str + status_code: str | None + waybill_number: str | None + + def build_calc_body_for_calculation( request: DeliveryCalculationRequest, delivery_type: str = "", @@ -176,6 +185,73 @@ def map_cse_save_order_response(root: Element) -> CSEOrderRegistrationResult: raise CSEMappingError("CSE SaveDocuments response is missing document Number.") +def build_tracking_body_for_order(order_number: str) -> dict[str, Element]: + return { + "documents": Element( + key="Documents", + items=[Element(key=order_number)], + ), + "parameters": Element( + key="parameters", + items=[ + make_field("DocumentType", "Order"), + make_field("OnlySelectedType", True, "boolean"), + ], + ), + } + + +def map_cse_tracking_response(root: Element) -> CSEOrderInfo: + for document in root.items: + order_number = document.property_value("Number") or document.key + if not order_number: + continue + return CSEOrderInfo( + order_number=order_number, + status_code=_latest_tracking_status(document), + waybill_number=_extract_waybill_number(document), + ) + raise CSEMappingError("CSE Tracking response is missing document data.") + + +def build_waybill_print_form_body(waybill_number: str) -> dict[str, Element]: + return { + "documents": Element( + key="Documents", + items=[Element(key=waybill_number)], + ), + "parameters": Element( + key="parameters", + items=[ + make_field("DocumentType", "waybill"), + make_field("Type", "print"), + make_field( + "Name", + "Универсальная печатная " + "форма документа НАКЛАДНАЯ", + ), + make_field("Format", "pdf"), + make_field("OnlySelectedType", True, "boolean"), + ], + ), + } + + +def map_cse_waybill_print_form_response(root: Element) -> bytes: + for document in root.items: + bdata = document.bdata + if not bdata: + continue + try: + normalized_bdata = "".join(bdata.split()) + return base64.b64decode(normalized_bdata, validate=True) + except (binascii.Error, ValueError) as exc: + raise CSEMappingError( + "CSE GetFormsForDocuments response contains invalid BData." + ) from exc + raise CSEMappingError("CSE GetFormsForDocuments response is missing BData.") + + def _parcel_type(request: DeliveryCalculationRequest) -> str | None: return request.parcel_type.value if request.parcel_type is not None else None @@ -190,3 +266,30 @@ def _compose_address(address: Address) -> str: def _format_decimal(value: float) -> str: return format(value, "g") + + +def _latest_tracking_status(document: Element) -> str | None: + dated: list[tuple[str, str]] = [] + for state in document.items: + if not state.key: + continue + date_time = state.property_value("DateTime") or "" + dated.append((date_time, state.key)) + if not dated: + return None + dated.sort(key=lambda item: item[0]) + return dated[-1][1] + + +def _extract_waybill_number(document: Element) -> str | None: + for table in document.tables: + if table.key != "Waybills": + continue + for waybill in table.items: + document_type = waybill.property_value("DocumentType") + if document_type is not None and document_type.lower() != "waybill": + continue + number = waybill.property_value("Number") or waybill.key + if number: + return number + return None diff --git a/app/adapters/delivery_providers/cse/soap.py b/app/adapters/delivery_providers/cse/soap.py index 775ef4d..d4ca74f 100644 --- a/app/adapters/delivery_providers/cse/soap.py +++ b/app/adapters/delivery_providers/cse/soap.py @@ -18,6 +18,7 @@ SOAP_NS = "http://www.w3.org/2003/05/soap-envelope" # Child tags of an Element that hold nested Element lists. _LIST_TAGS = ("Fields", "List", "Tables", "Properties") _SCALAR_TAGS = ("Key", "Value", "ValueType") +_BINARY_TAG = "BData" @dataclass @@ -27,6 +28,7 @@ class Element: key: str | None = None value: str | None = None value_type: str | None = None + bdata: str | None = None fields: list["Element"] = field(default_factory=list) items: list["Element"] = field(default_factory=list) # tables: list["Element"] = field(default_factory=list) @@ -159,6 +161,8 @@ def _parse_element(node: ET.Element) -> Element: element.tables.append(_parse_element(child)) elif local == "Properties": element.properties.append(_parse_element(child)) + elif local == _BINARY_TAG: + element.bdata = (child.text or "").strip() return element diff --git a/app/repositories/order/repository.py b/app/repositories/order/repository.py index c069952..e1cf722 100644 --- a/app/repositories/order/repository.py +++ b/app/repositories/order/repository.py @@ -6,7 +6,7 @@ from dataclasses import dataclass from datetime import datetime from typing import Any -from sqlalchemy import select +from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.domain.cdek_polling import TERMINAL_ORDER_STATUSES @@ -95,17 +95,23 @@ class OrderRepository: *, limit: int, ) -> Sequence[Order]: + cdek_pending = and_( + Order.provider == "cdek", + Order.provider_order_id.is_not(None), + Order.provider_waybill_url.is_(None), + ( + Order.provider_order_status.is_(None) + | Order.provider_order_status.not_in(TERMINAL_ORDER_STATUSES) + ), + ) + cse_pending = and_( + Order.provider == "cse", + Order.provider_order_id.is_not(None), + Order.provider_waybill_id.is_(None), + ) statement = ( select(Order) - .where( - Order.provider == "cdek", - Order.provider_order_id.is_not(None), - Order.provider_waybill_url.is_(None), - ( - Order.provider_order_status.is_(None) - | Order.provider_order_status.not_in(TERMINAL_ORDER_STATUSES) - ), - ) + .where(or_(cdek_pending, cse_pending)) .order_by(Order.provider_polled_at.asc().nulls_first()) .limit(limit) .with_for_update(skip_locked=True) @@ -173,11 +179,18 @@ class OrderRepository: *, limit: int, ) -> Sequence[Order]: + cdek_ready = and_( + Order.provider == "cdek", + Order.provider_waybill_url.is_not(None), + ) + cse_ready = and_( + Order.provider == "cse", + Order.provider_waybill_id.is_not(None), + ) statement = ( select(Order) .where( - Order.provider == "cdek", - Order.provider_waybill_url.is_not(None), + or_(cdek_ready, cse_ready), Order.waybill_email_sent_at.is_(None), ) .order_by(Order.created_at.asc()) diff --git a/app/services/waybill_email_sender.py b/app/services/waybill_email_sender.py index 5603068..8a37896 100644 --- a/app/services/waybill_email_sender.py +++ b/app/services/waybill_email_sender.py @@ -1,7 +1,7 @@ -"""Background service that e-mails CDEK waybill PDFs to customers.""" +"""Background service that e-mails provider waybill PDFs to customers.""" import asyncio -from collections.abc import Callable, Sequence +from collections.abc import Callable, Mapping, Sequence from contextlib import AbstractAsyncContextManager from dataclasses import dataclass from datetime import datetime, timezone @@ -15,15 +15,18 @@ logger = structlog.get_logger(__name__) _EMAIL_SUBJECT_TEMPLATE = "Накладная по заказу {order_uuid}" _EMAIL_BODY_TEMPLATE = ( "Здравствуйте!\n\n" - "По вашему заказу {order_uuid} сформирована транспортная накладная CDEK.\n" + "По вашему заказу {order_uuid} сформирована " + "транспортная накладная.\n" "PDF-файл накладной приложен к этому письму.\n" +) +_EMAIL_BODY_URL_LINE_TEMPLATE = ( "Также накладная доступна по ссылке: {waybill_url}\n" ) _ATTACHMENT_FILENAME_TEMPLATE = "waybill_{order_uuid}.pdf" class WaybillPDFDownloaderProtocol(Protocol): - async def download_waybill_pdf(self, url: str) -> bytes: ... + async def download_waybill_pdf(self, order: "OrderRecord") -> bytes: ... class EmailSenderProtocol(Protocol): @@ -40,7 +43,9 @@ class EmailSenderProtocol(Protocol): class OrderRecord(Protocol): order_uuid: str + provider: str account_email: str + provider_waybill_id: str | None provider_waybill_url: str | None @@ -72,13 +77,13 @@ class WaybillEmailSenderService: self, *, order_repository: WaybillEmailSenderRepositoryProtocol, - waybill_downloader: WaybillPDFDownloaderProtocol, + waybill_downloaders: Mapping[str, WaybillPDFDownloaderProtocol], email_sender: EmailSenderProtocol, batch_size: int, datetime_now: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: self._repository = order_repository - self._waybill_downloader = waybill_downloader + self._waybill_downloaders = dict(waybill_downloaders) self._email_sender = email_sender self._batch_size = batch_size self._datetime_now = datetime_now @@ -99,6 +104,7 @@ class WaybillEmailSenderService: logger.exception( "waybill_email_order_failed", order_uuid=order.order_uuid, + provider=order.provider, account_email=order.account_email, ) return SendBatchSummary( @@ -133,16 +139,17 @@ class WaybillEmailSenderService: continue async def _handle_order(self, session: object, order: OrderRecord) -> None: - waybill_url = order.provider_waybill_url - if waybill_url is None: + if order.provider_waybill_url is None and order.provider_waybill_id is None: return - pdf_bytes = await self._waybill_downloader.download_waybill_pdf(waybill_url) + downloader = self._resolve_downloader(order.provider) + pdf_bytes = await downloader.download_waybill_pdf(order) subject = _EMAIL_SUBJECT_TEMPLATE.format(order_uuid=order.order_uuid) - body = _EMAIL_BODY_TEMPLATE.format( - order_uuid=order.order_uuid, - waybill_url=waybill_url, - ) + body = _EMAIL_BODY_TEMPLATE.format(order_uuid=order.order_uuid) + if order.provider_waybill_url is not None: + body += _EMAIL_BODY_URL_LINE_TEMPLATE.format( + waybill_url=order.provider_waybill_url, + ) filename = _ATTACHMENT_FILENAME_TEMPLATE.format(order_uuid=order.order_uuid) await self._email_sender.send_email( @@ -162,6 +169,13 @@ class WaybillEmailSenderService: logger.info( "waybill_email_sent", order_uuid=order.order_uuid, + provider=order.provider, account_email=order.account_email, sent_at=sent_at, ) + + def _resolve_downloader(self, provider: str) -> WaybillPDFDownloaderProtocol: + downloader = self._waybill_downloaders.get(provider) + if downloader is None: + raise RuntimeError(f"Waybill downloader is not configured for {provider}.") + return downloader diff --git a/app/services/waybill_poller.py b/app/services/waybill_poller.py index 19d3e6c..0681b9d 100644 --- a/app/services/waybill_poller.py +++ b/app/services/waybill_poller.py @@ -1,7 +1,7 @@ -"""Background service that polls CDEK for waybill updates.""" +"""Background service that polls providers for waybill updates.""" import asyncio -from collections.abc import Callable, Sequence +from collections.abc import Callable, Mapping, Sequence from contextlib import AbstractAsyncContextManager from dataclasses import dataclass from datetime import datetime, timezone @@ -9,24 +9,20 @@ from typing import Protocol import structlog -from app.adapters.delivery_providers.cdek.order_mapper import ( - CDEKOrderInfo, - CDEKWaybillInfo, -) - logger = structlog.get_logger(__name__) -class CDEKOrderInfoAdapterProtocol(Protocol): - async def get_order(self, cdek_order_uuid: str) -> CDEKOrderInfo: ... +class OrderInfoAdapterProtocol(Protocol): + async def get_order(self, provider_order_id: str) -> object: ... -class CDEKWaybillInfoAdapterProtocol(Protocol): - async def get_waybill(self, cdek_waybill_uuid: str) -> CDEKWaybillInfo: ... +class WaybillInfoAdapterProtocol(Protocol): + async def get_waybill(self, provider_waybill_id: str) -> object: ... class OrderRecord(Protocol): order_uuid: str + provider: str provider_order_id: str | None provider_waybill_id: str | None @@ -70,14 +66,14 @@ class WaybillPollerService: self, *, order_repository: WaybillPollerRepositoryProtocol, - order_info_adapter: CDEKOrderInfoAdapterProtocol, - waybill_info_adapter: CDEKWaybillInfoAdapterProtocol, + order_info_adapters: Mapping[str, OrderInfoAdapterProtocol], + waybill_info_adapters: Mapping[str, WaybillInfoAdapterProtocol], batch_size: int, datetime_now: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: self._repository = order_repository - self._order_info_adapter = order_info_adapter - self._waybill_info_adapter = waybill_info_adapter + self._order_info_adapters = dict(order_info_adapters) + self._waybill_info_adapters = dict(waybill_info_adapters) self._batch_size = batch_size self._datetime_now = datetime_now @@ -97,6 +93,7 @@ class WaybillPollerService: logger.exception( "waybill_poll_order_failed", order_uuid=order.order_uuid, + provider=order.provider, provider_order_id=order.provider_order_id, provider_waybill_id=order.provider_waybill_id, ) @@ -137,35 +134,74 @@ class WaybillPollerService: provider_order_id = order.provider_order_id if provider_order_id is None: return - info = await self._order_info_adapter.get_order(provider_order_id) + adapter = self._resolve_order_info_adapter(order.provider) + info = await adapter.get_order(provider_order_id) + status_code = _extract_status_code(info) + waybill_id = _extract_waybill_id(info) await self._repository.record_order_poll( session, order_uuid=order.order_uuid, - order_status=info.status_code, - waybill_uuid=info.waybill_uuid, + order_status=status_code, + waybill_uuid=waybill_id, polled_at=polled_at, ) logger.info( "waybill_poll_order_result", order_uuid=order.order_uuid, + provider=order.provider, provider_order_id=provider_order_id, - provider_order_status=info.status_code, - provider_waybill_id=info.waybill_uuid, + provider_order_status=status_code, + provider_waybill_id=waybill_id, ) return - waybill = await self._waybill_info_adapter.get_waybill( - order.provider_waybill_id - ) + adapter = self._resolve_waybill_info_adapter(order.provider) + waybill = await adapter.get_waybill(order.provider_waybill_id) + waybill_url = _extract_waybill_url(waybill) await self._repository.record_waybill_poll( session, order_uuid=order.order_uuid, - waybill_url=waybill.url, + waybill_url=waybill_url, polled_at=polled_at, ) logger.info( "waybill_poll_waybill_result", order_uuid=order.order_uuid, + provider=order.provider, provider_waybill_id=order.provider_waybill_id, - provider_waybill_url=waybill.url, + provider_waybill_url=waybill_url, ) + + def _resolve_order_info_adapter(self, provider: str) -> OrderInfoAdapterProtocol: + adapter = self._order_info_adapters.get(provider) + if adapter is None: + raise RuntimeError(f"Order info adapter is not configured for {provider}.") + return adapter + + def _resolve_waybill_info_adapter( + self, provider: str + ) -> WaybillInfoAdapterProtocol: + adapter = self._waybill_info_adapters.get(provider) + if adapter is None: + raise RuntimeError( + f"Waybill info adapter is not configured for {provider}." + ) + return adapter + + +def _extract_status_code(info: object) -> str | None: + value = getattr(info, "status_code", None) + return value if isinstance(value, str) and value else None + + +def _extract_waybill_id(info: object) -> str | None: + for attr in ("waybill_id", "waybill_uuid", "waybill_number"): + value = getattr(info, attr, None) + if isinstance(value, str) and value: + return value + return None + + +def _extract_waybill_url(info: object) -> str | None: + value = getattr(info, "url", None) + return value if isinstance(value, str) and value else None diff --git a/app/workers/waybill_email_sender.py b/app/workers/waybill_email_sender.py index d1fbe7f..2b33e90 100644 --- a/app/workers/waybill_email_sender.py +++ b/app/workers/waybill_email_sender.py @@ -1,13 +1,21 @@ -"""Background worker that e-mails CDEK waybill PDFs.""" +"""Background worker that e-mails provider waybill PDFs.""" import asyncio import signal +from dataclasses import dataclass import httpx import structlog +from app.adapters.delivery_providers.registry import ( + resolve_delivery_provider_timeout_seconds, +) from app.adapters.delivery_providers.cdek.auth import CDEKAuthClient from app.adapters.delivery_providers.cdek.client import CDEKClient +from app.adapters.delivery_providers.cse.client import CSEClient +from app.adapters.delivery_providers.cse.order_mapper import ( + CSEOrderRegistrationParams, +) from app.adapters.email import SMTPEmailSender from app.adapters.postgres.engine import ( create_postgres_engine, @@ -21,24 +29,71 @@ from app.services.waybill_email_sender import WaybillEmailSenderService logger = structlog.get_logger(__name__) +@dataclass(frozen=True) +class CDEKWaybillPDFDownloader: + client: CDEKClient + + async def download_waybill_pdf(self, order: object) -> bytes: + url = getattr(order, "provider_waybill_url", None) + if not isinstance(url, str) or not url: + raise RuntimeError("CDEK waybill URL is missing.") + return await self.client.download_waybill_pdf(url) + + +@dataclass(frozen=True) +class CSEWaybillPDFDownloader: + client: CSEClient + + async def download_waybill_pdf(self, order: object) -> bytes: + waybill_number = getattr(order, "provider_waybill_id", None) + if not isinstance(waybill_number, str) or not waybill_number: + raise RuntimeError("CSE waybill number is missing.") + return await self.client.download_waybill_pdf(waybill_number) + + async def _run(settings: Settings, stop_event: asyncio.Event) -> None: + http_client = httpx.AsyncClient( + timeout=resolve_delivery_provider_timeout_seconds(settings.delivery_providers) + ) + waybill_downloaders: dict[str, object] = {} + cdek_config = settings.delivery_providers.cdek - http_client = httpx.AsyncClient(timeout=cdek_config.timeout_seconds) - auth_client = CDEKAuthClient( - http_client=http_client, - base_url=cdek_config.base_url, - client_id=cdek_config.client_id, - client_secret=cdek_config.client_secret, - timeout_seconds=cdek_config.timeout_seconds, - ) - cdek_client = CDEKClient( - http_client=http_client, - auth_client=auth_client, - base_url=cdek_config.base_url, - timeout_seconds=cdek_config.timeout_seconds, - retry_attempts=cdek_config.retry_attempts, - retry_backoff_seconds=cdek_config.retry_backoff_seconds, - ) + if cdek_config.enabled: + auth_client = CDEKAuthClient( + http_client=http_client, + base_url=cdek_config.base_url, + client_id=cdek_config.client_id, + client_secret=cdek_config.client_secret, + timeout_seconds=cdek_config.timeout_seconds, + ) + cdek_client = CDEKClient( + http_client=http_client, + auth_client=auth_client, + base_url=cdek_config.base_url, + timeout_seconds=cdek_config.timeout_seconds, + retry_attempts=cdek_config.retry_attempts, + retry_backoff_seconds=cdek_config.retry_backoff_seconds, + ) + waybill_downloaders["cdek"] = CDEKWaybillPDFDownloader(cdek_client) + + cse_config = settings.delivery_providers.cse + if cse_config.enabled: + cse_client = CSEClient( + http_client=http_client, + base_url=cse_config.base_url, + login=cse_config.login, + password=cse_config.password, + registration_params=CSEOrderRegistrationParams( + payer=cse_config.payer, + payment_method=cse_config.payment_method, + shipping_method=cse_config.shipping_method, + ), + timeout_seconds=cse_config.timeout_seconds, + retry_attempts=cse_config.retry_attempts, + retry_backoff_seconds=cse_config.retry_backoff_seconds, + ) + waybill_downloaders["cse"] = CSEWaybillPDFDownloader(cse_client) + email_sender = SMTPEmailSender( smtp_host=settings.email.smtp_host, smtp_port=settings.email.smtp_port, @@ -53,7 +108,7 @@ async def _run(settings: Settings, stop_event: asyncio.Event) -> None: repository = OrderRepository(session_factory=session_factory) service = WaybillEmailSenderService( order_repository=repository, - waybill_downloader=cdek_client, + waybill_downloaders=waybill_downloaders, email_sender=email_sender, batch_size=settings.waybill_email_sender.batch_size, ) diff --git a/app/workers/waybill_poller.py b/app/workers/waybill_poller.py index 189c66a..650fbd1 100644 --- a/app/workers/waybill_poller.py +++ b/app/workers/waybill_poller.py @@ -1,4 +1,4 @@ -"""Background worker that polls CDEK for waybill updates.""" +"""Background worker that polls delivery providers for waybill updates.""" import asyncio import signal @@ -6,8 +6,15 @@ import signal import httpx import structlog +from app.adapters.delivery_providers.registry import ( + resolve_delivery_provider_timeout_seconds, +) from app.adapters.delivery_providers.cdek.auth import CDEKAuthClient from app.adapters.delivery_providers.cdek.client import CDEKClient +from app.adapters.delivery_providers.cse.client import CSEClient +from app.adapters.delivery_providers.cse.order_mapper import ( + CSEOrderRegistrationParams, +) from app.adapters.postgres.engine import ( create_postgres_engine, create_postgres_session_factory, @@ -21,30 +28,57 @@ logger = structlog.get_logger(__name__) async def _run(settings: Settings, stop_event: asyncio.Event) -> None: + http_client = httpx.AsyncClient( + timeout=resolve_delivery_provider_timeout_seconds(settings.delivery_providers) + ) + order_info_adapters: dict[str, object] = {} + waybill_info_adapters: dict[str, object] = {} + cdek_config = settings.delivery_providers.cdek - http_client = httpx.AsyncClient(timeout=cdek_config.timeout_seconds) - auth_client = CDEKAuthClient( - http_client=http_client, - base_url=cdek_config.base_url, - client_id=cdek_config.client_id, - client_secret=cdek_config.client_secret, - timeout_seconds=cdek_config.timeout_seconds, - ) - cdek_client = CDEKClient( - http_client=http_client, - auth_client=auth_client, - base_url=cdek_config.base_url, - timeout_seconds=cdek_config.timeout_seconds, - retry_attempts=cdek_config.retry_attempts, - retry_backoff_seconds=cdek_config.retry_backoff_seconds, - ) + if cdek_config.enabled: + auth_client = CDEKAuthClient( + http_client=http_client, + base_url=cdek_config.base_url, + client_id=cdek_config.client_id, + client_secret=cdek_config.client_secret, + timeout_seconds=cdek_config.timeout_seconds, + ) + cdek_client = CDEKClient( + http_client=http_client, + auth_client=auth_client, + base_url=cdek_config.base_url, + timeout_seconds=cdek_config.timeout_seconds, + retry_attempts=cdek_config.retry_attempts, + retry_backoff_seconds=cdek_config.retry_backoff_seconds, + ) + order_info_adapters["cdek"] = cdek_client + waybill_info_adapters["cdek"] = cdek_client + + cse_config = settings.delivery_providers.cse + if cse_config.enabled: + cse_client = CSEClient( + http_client=http_client, + base_url=cse_config.base_url, + login=cse_config.login, + password=cse_config.password, + registration_params=CSEOrderRegistrationParams( + payer=cse_config.payer, + payment_method=cse_config.payment_method, + shipping_method=cse_config.shipping_method, + ), + timeout_seconds=cse_config.timeout_seconds, + retry_attempts=cse_config.retry_attempts, + retry_backoff_seconds=cse_config.retry_backoff_seconds, + ) + order_info_adapters["cse"] = cse_client + engine = create_postgres_engine(settings.postgres) session_factory = create_postgres_session_factory(engine) repository = OrderRepository(session_factory=session_factory) service = WaybillPollerService( order_repository=repository, - order_info_adapter=cdek_client, - waybill_info_adapter=cdek_client, + order_info_adapters=order_info_adapters, + waybill_info_adapters=waybill_info_adapters, batch_size=settings.waybill_poller.batch_size, ) logger.info( diff --git a/tests/adapters/delivery_providers/cse/test_cse_adapter.py b/tests/adapters/delivery_providers/cse/test_cse_adapter.py index c2b07e0..1fe416b 100644 --- a/tests/adapters/delivery_providers/cse/test_cse_adapter.py +++ b/tests/adapters/delivery_providers/cse/test_cse_adapter.py @@ -70,6 +70,62 @@ _SAVE_RESPONSE = """ """ +_TRACKING_RESPONSE = """ + + + + + Tracking + + CSE-000123 + Order + NumberCSE-000123 + + Заказ принят, идет обработка заказа. + + DateTime2026-06-01T10:00:00 + + + + Накладная оформлена. + + DateTime2026-06-01T10:02:00 + + + + Waybills + + 496-AA-1676378 + + DocumentTypeWaybill + + + Number496-AA-1676378 + + + + + + + +""" + +_FORM_RESPONSE = """ + + + + + GetPrintForms + + 496-AA-1676378 + FormFormatPDF + JVBERg== + + + + +""" + _ERROR_RESPONSE = """ @@ -253,6 +309,40 @@ def test_provider_register_order_returns_document_number() -> None: assert http_client.calls[0]["url"] == "http://lk-test.cse.ru/1c/ws/web1c.1cws" +def test_provider_get_order_extracts_waybill_number_from_tracking() -> None: + response = httpx.Response(200, text=_TRACKING_RESPONSE) + client, http_client = _build_client([response]) + provider = CSEProvider(client) + + result = asyncio.run(provider.get_order("CSE-000123")) + + assert result.order_number == "CSE-000123" + assert result.status_code == "Накладная оформлена." + assert result.waybill_number == "496-AA-1676378" + content = http_client.calls[0]["content"].decode("utf-8") + assert "Tracking" in content + assert "DocumentType" in content + assert "Order" in content + + +def test_provider_download_waybill_pdf_uses_print_form() -> None: + response = httpx.Response(200, text=_FORM_RESPONSE) + client, http_client = _build_client([response]) + provider = CSEProvider(client) + + pdf = asyncio.run(provider.download_waybill_pdf("496-AA-1676378")) + + assert pdf == b"%PDF" + content = http_client.calls[0]["content"].decode("utf-8") + assert "GetFormsForDocuments" in content + assert "DocumentType" in content + assert "waybill" in content + assert "Type" in content + assert "print" in content + assert "Format" in content + assert "pdf" in content + + def test_save_order_request_uses_selected_tariff_urgency() -> None: request = make_init_payment_request( systemData={ diff --git a/tests/repositories/order/test_repository.py b/tests/repositories/order/test_repository.py index 4a42f3f..692d94a 100644 --- a/tests/repositories/order/test_repository.py +++ b/tests/repositories/order/test_repository.py @@ -312,15 +312,22 @@ def test_list_orders_pending_waybill_returns_orders_without_url() -> None: ) await _seed_order( repository, - order_uuid="not-cdek", + order_uuid="cse-pending", provider="cse", provider_order_id="cse-1", ) + await _seed_order( + repository, + order_uuid="cse-with-waybill", + provider="cse", + provider_order_id="cse-2", + provider_waybill_id="496-AA-1676378", + ) async with repository.session() as session: orders = await repository.list_orders_pending_waybill(session, limit=10) - assert [order.order_uuid for order in orders] == ["pending"] + assert [order.order_uuid for order in orders] == ["pending", "cse-pending"] asyncio.run(_with_repository(run)) @@ -449,7 +456,7 @@ def test_record_waybill_poll_sets_url_only_when_previously_null() -> None: asyncio.run(_with_repository(run)) -def test_list_orders_pending_waybill_email_returns_orders_with_url_and_no_sent_at() -> None: +def test_list_orders_pending_waybill_email_returns_ready_orders() -> None: async def run( repository: OrderRepository, _session_factory: async_sessionmaker[AsyncSession], @@ -474,6 +481,19 @@ def test_list_orders_pending_waybill_email_returns_orders_with_url_and_no_sent_a provider_waybill_id="w3", provider_waybill_url="https://cdek.test/3.pdf", ) + await _seed_order( + repository, + order_uuid="cse-ready", + provider="cse", + provider_order_id="cse-o1", + provider_waybill_id="496-AA-1676378", + ) + await _seed_order( + repository, + order_uuid="cse-no-waybill", + provider="cse", + provider_order_id="cse-o2", + ) async with repository.session() as session: sent = await repository.get_order_by_order_uuid(session, "already-sent") assert sent is not None @@ -486,7 +506,7 @@ def test_list_orders_pending_waybill_email_returns_orders_with_url_and_no_sent_a session, limit=10 ) - assert [order.order_uuid for order in orders] == ["ready"] + assert [order.order_uuid for order in orders] == ["ready", "cse-ready"] asyncio.run(_with_repository(run)) diff --git a/tests/services/test_waybill_email_sender.py b/tests/services/test_waybill_email_sender.py index b8e4448..caf7330 100644 --- a/tests/services/test_waybill_email_sender.py +++ b/tests/services/test_waybill_email_sender.py @@ -10,6 +10,8 @@ from app.services.waybill_email_sender import WaybillEmailSenderService class StoredOrder: order_uuid: str account_email: str + provider: str = "cdek" + provider_waybill_id: str | None = None provider_waybill_url: str | None = None waybill_email_sent_at: datetime | None = None @@ -41,7 +43,10 @@ class StubRepository: return [ order for order in self._orders.values() - if order.provider_waybill_url is not None + if ( + order.provider_waybill_url is not None + or order.provider_waybill_id is not None + ) and order.waybill_email_sent_at is None ] @@ -68,9 +73,11 @@ class StubDownloader: self._results = results self.calls: list[str] = [] - async def download_waybill_pdf(self, url: str) -> bytes: - self.calls.append(url) - result = self._results[url] + async def download_waybill_pdf(self, order: StoredOrder) -> bytes: + locator = order.provider_waybill_url or order.provider_waybill_id + assert locator is not None + self.calls.append(locator) + result = self._results[locator] if isinstance(result, Exception): raise result return result @@ -114,7 +121,7 @@ def _make_service( ) -> WaybillEmailSenderService: return WaybillEmailSenderService( order_repository=repository, - waybill_downloader=downloader or StubDownloader({}), + waybill_downloaders={"cdek": downloader or StubDownloader({})}, email_sender=email_sender or StubEmailSender(), batch_size=10, datetime_now=lambda: _SENT_AT, @@ -151,6 +158,35 @@ def test_poll_once_downloads_pdf_sends_email_and_marks_sent() -> None: assert order.waybill_email_sent_at == _SENT_AT +def test_poll_once_sends_cse_waybill_pdf_without_url() -> None: + order = StoredOrder( + order_uuid="o-1", + provider="cse", + account_email="client@example.com", + provider_waybill_id="496-AA-1676378", + ) + repo = StubRepository([order]) + downloader = StubDownloader({"496-AA-1676378": b"%PDF"}) + email_sender = StubEmailSender() + service = WaybillEmailSenderService( + order_repository=repo, + waybill_downloaders={"cse": downloader}, + email_sender=email_sender, + batch_size=10, + datetime_now=lambda: _SENT_AT, + ) + + summary = asyncio.run(service.poll_once()) + + assert summary.processed == 1 + assert summary.succeeded == 1 + assert downloader.calls == ["496-AA-1676378"] + assert len(email_sender.calls) == 1 + assert "https://" not in email_sender.calls[0]["body"] + assert email_sender.calls[0]["attachment_bytes"] == b"%PDF" + assert order.waybill_email_sent_at == _SENT_AT + + def test_poll_once_download_error_keeps_order_pending_and_skips_send() -> None: order = StoredOrder( order_uuid="o-1", diff --git a/tests/services/test_waybill_poller.py b/tests/services/test_waybill_poller.py index 26c3e32..62b6e85 100644 --- a/tests/services/test_waybill_poller.py +++ b/tests/services/test_waybill_poller.py @@ -7,12 +7,14 @@ from app.adapters.delivery_providers.cdek.order_mapper import ( CDEKOrderInfo, CDEKWaybillInfo, ) +from app.adapters.delivery_providers.cse.order_mapper import CSEOrderInfo from app.services.waybill_poller import WaybillPollerService @dataclass class StoredOrder: order_uuid: str + provider: str = "cdek" provider_order_id: str | None = None provider_order_status: str | None = None provider_waybill_id: str | None = None @@ -129,6 +131,9 @@ class StubWaybillInfoAdapter: _POLLED_AT = datetime(2026, 5, 24, 12, 0, tzinfo=timezone.utc) +_CSE_WAYBILL_CREATED_STATUS = ( + "На основании заказа оформлена накладная." +) def _make_service( @@ -139,8 +144,8 @@ def _make_service( ) -> WaybillPollerService: return WaybillPollerService( order_repository=repository, - order_info_adapter=order_info or StubOrderInfoAdapter({}), - waybill_info_adapter=waybill_info or StubWaybillInfoAdapter({}), + order_info_adapters={"cdek": order_info or StubOrderInfoAdapter({})}, + waybill_info_adapters={"cdek": waybill_info or StubWaybillInfoAdapter({})}, batch_size=10, datetime_now=lambda: _POLLED_AT, ) @@ -194,6 +199,39 @@ def test_poll_once_fetches_waybill_info_when_waybill_uuid_is_present() -> None: assert order.provider_polled_at == _POLLED_AT +def test_poll_once_fetches_cse_waybill_number_from_order_tracking() -> None: + order = StoredOrder( + order_uuid="o", + provider="cse", + provider_order_id="CSE-000123", + ) + repo = StubRepository([order]) + order_info = StubOrderInfoAdapter( + { + "CSE-000123": CSEOrderInfo( + order_number="CSE-000123", + status_code=_CSE_WAYBILL_CREATED_STATUS, + waybill_number="496-AA-1676378", + ) + } + ) + service = WaybillPollerService( + order_repository=repo, + order_info_adapters={"cse": order_info}, + waybill_info_adapters={}, + batch_size=10, + datetime_now=lambda: _POLLED_AT, + ) + + summary = asyncio.run(service.poll_once()) + + assert summary.processed == 1 and summary.succeeded == 1 and summary.failed == 0 + assert order_info.calls == ["CSE-000123"] + assert order.provider_order_status == _CSE_WAYBILL_CREATED_STATUS + assert order.provider_waybill_id == "496-AA-1676378" + assert order.provider_polled_at == _POLLED_AT + + def test_poll_once_records_terminal_status_without_waybill() -> None: order = StoredOrder(order_uuid="o", provider_order_id="cdek-o") repo = StubRepository([order]) diff --git a/tests/workers/test_waybill_email_sender_main.py b/tests/workers/test_waybill_email_sender_main.py index bdca3d3..85a5830 100644 --- a/tests/workers/test_waybill_email_sender_main.py +++ b/tests/workers/test_waybill_email_sender_main.py @@ -64,6 +64,7 @@ def test_run_exits_when_stop_event_is_set() -> None: ), patch("app.workers.waybill_email_sender.CDEKAuthClient"), patch("app.workers.waybill_email_sender.CDEKClient"), + patch("app.workers.waybill_email_sender.CSEClient"), patch("app.workers.waybill_email_sender.SMTPEmailSender"), patch( "app.workers.waybill_email_sender.create_postgres_engine", diff --git a/tests/workers/test_waybill_poller_main.py b/tests/workers/test_waybill_poller_main.py index af28ce3..025998f 100644 --- a/tests/workers/test_waybill_poller_main.py +++ b/tests/workers/test_waybill_poller_main.py @@ -69,6 +69,7 @@ def test_run_exits_when_stop_event_is_set() -> None: ), patch("app.workers.waybill_poller.CDEKAuthClient"), patch("app.workers.waybill_poller.CDEKClient"), + patch("app.workers.waybill_poller.CSEClient"), patch( "app.workers.waybill_poller.create_postgres_engine", return_value=engine_instance,