alien_py_cloud/alien_store/repositories/contract_repo.py

from sqlalchemy.ext.asyncio import AsyncSession
from alien_store.db.models.contract_store import ContractStore
import json
from datetime import datetime, timedelta


class ContractRepository:
    """合同数据访问层"""

    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_store_id(self, store_id: int):
        """根据店铺id查询所有合同"""
        result = await self.db.execute(
            ContractStore.__table__.select().where(ContractStore.store_id == store_id)
        )
        # 返回列表[dict]，避免 Pydantic 序列化 Row 对象出错
        return [dict(row) for row in result.mappings().all()]

    async def get_all(self):
        """查询所有合同"""
        result = await self.db.execute(ContractStore.__table__.select())
        return [dict(row) for row in result.mappings().all()]

    async def get_all_paged(self, page: int, page_size: int = 10):
        """分页查询所有合同"""
        offset = (page - 1) * page_size
        result = await self.db.execute(
            ContractStore.__table__.select().offset(offset).limit(page_size)
        )
        return [dict(row) for row in result.mappings().all()]

    async def create(self, user_data):
        """创建未签署合同模板"""
        db_templates = ContractStore(
            store_id=user_data.store_id,
            merchant_name=user_data.merchant_name,
            business_segment=user_data.business_segment,
            contact_phone=user_data.contact_phone,
            contract_url=user_data.contract_url,
            seal_url='0.0',
            signing_status='未签署'
        )
        self.db.add(db_templates)
        await self.db.commit()
        await self.db.refresh(db_templates)
        return db_templates

    async def mark_signed_by_phone(self, contact_phone: str, signing_time: datetime | None = None):
        """
        根据手机号将合同标记为已签署，并更新 contract_url 内的 status=1
        同时写入签署/生效/到期时间（签署时间=T，生效=T+1天，失效=生效+365天）
        """
        result = await self.db.execute(
            ContractStore.__table__.select().where(ContractStore.contact_phone == contact_phone)
        )
        rows = result.mappings().all()
        updated = False
        for row in rows:
            contract_url_raw = row.get("contract_url")
            items = None
            if contract_url_raw:
                try:
                    items = json.loads(contract_url_raw)
                except Exception:
                    items = None
            changed = False
            if isinstance(items, list):
                for item in items:
                    item["status"] = 1
                    changed = True
            # 时间处理
            signing_dt = signing_time
            effective_dt = expiry_dt = None
            if signing_dt:
                effective_dt = signing_dt + timedelta(days=1)
                expiry_dt = effective_dt + timedelta(days=365)
            if changed or True:
                await self.db.execute(
                    ContractStore.__table__.update()
                    .where(ContractStore.id == row["id"])
                    .values(
                        signing_status="已签署",
                        contract_url=json.dumps(items, ensure_ascii=False) if items else contract_url_raw,
                        signing_time=signing_dt,
                        effective_time=effective_dt,
                        expiry_time=expiry_dt,
                    )
                )
                updated = True
        if updated:
            await self.db.commit()
        return updated