alien_py_cloud/alien_store/services/contract_server.py

from sqlalchemy.ext.asyncio import AsyncSession
from alien_store.repositories.contract_repo import ContractRepository
from alien_store.schemas.request.contract_store import TemplatesCreate


class ContractServer:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.esign_repo = ContractRepository(db)

    async def create_template(self, template_data: TemplatesCreate):
        await self.esign_repo.create(template_data)
        return {
            "message": "模板创建成功"
        }

    async def list_by_store(self, store_id: int):
        return await self.esign_repo.get_by_store_id(store_id)

    async def get_store_reason(self, store_id: int) -> str | None:
        return await self.esign_repo.check_store_status(store_id)

    async def get_contract_item_by_sign_flow_id(self, sign_flow_id: str):
        return await self.esign_repo.get_contract_item_by_sign_flow_id(sign_flow_id)

    async def update_contract_items(self, row_id: int, items: list) -> bool:
        return await self.esign_repo.update_contract_items(row_id, items)

    async def list_all_paged(
        self,
        page: int,
        page_size: int = 10,
        store_name: str | None = None,
        merchant_name: str | None = None,
        signing_status: str | None = None,
        business_segment: str | None = None,
        store_status: str | None = None,
        expiry_start=None,
        expiry_end=None,
    ):
        items, total = await self.esign_repo.get_all_paged(
            page,
            page_size,
            store_name=store_name,
            merchant_name=merchant_name,
            signing_status=signing_status,
            business_segment=business_segment,
            store_status=store_status,
            expiry_start=expiry_start,
            expiry_end=expiry_end,
        )
        return items, total

    async def mark_signed_by_phone(self, contact_phone: str, sign_flow_id: str, signing_time, contract_download_url):
        return await self.esign_repo.mark_signed_by_phone(contact_phone, sign_flow_id, signing_time, contract_download_url)

    async def update_sign_url(self, contact_phone: str, sign_flow_id: str, sign_url: str):
        return await self.esign_repo.update_sign_url(contact_phone, sign_flow_id, sign_url)

    async def append_contract_url(self, templates_data, contract_item: dict):
        return await self.esign_repo.append_contract_url(templates_data, contract_item)