alien_py_cloud/alien_contract/services/contract_server.py

import datetime
import json
import logging
import os
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from alien_contract.repositories.contract_repo import ContractRepository
from alien_contract.schemas.request.contract import BundleCreateRequest
from alien_contract.infrastructure.esign import main as esign_main
from alien_contract.infrastructure.esign.contract_builder import build_contract_items, ContractBuildError
from alien_contract.infrastructure.esign.main import sign_url, file_download_url

LOG_DIR = os.path.join("common", "logs", "alien_contract")
os.makedirs(LOG_DIR, exist_ok=True)


def _init_logger():
    logger = logging.getLogger("alien_contract_service")
    if logger.handlers:
        return logger
    logger.setLevel(logging.INFO)
    fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s %(message)s")
    info_handler = logging.FileHandler(os.path.join(LOG_DIR, "info.log"), encoding="utf-8")
    info_handler.setLevel(logging.INFO)
    info_handler.setFormatter(fmt)
    error_handler = logging.FileHandler(os.path.join(LOG_DIR, "error.log"), encoding="utf-8")
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(fmt)
    logger.addHandler(info_handler)
    logger.addHandler(error_handler)
    return logger


logger = _init_logger()

# e签宝 v3 回调 action 说明：
#   SIGN_MISSON_COMPLETE 单个签署任务完成（按签署人维度，平台方自动盖章也会触发，signResult=2）
#   SIGN_FLOW_COMPLETE / SIGN_FLOW_FINISH 整个签署流程结束
# 注意：绝不能用"单个任务完成"判定整份合同已签署，必须以流程真实状态 signFlowStatus 为准。
SIGN_FLOW_FINISH_ACTIONS = {
    "SIGN_FLOW_FINISH",
    "SIGN_FLOW_COMPLETE",
}

# e签宝 流程状态 signFlowStatus: 0草稿 1签署中 2完成 3撤销 4终止 5过期 6删除 7拒签
ESIGN_FLOW_STATUS_COMPLETED = 2

# 与数据库 NOW() 时区(东八区)保持一致，避免应用容器在 UTC 时写入早 8 小时的时间
CN_TZ = datetime.timezone(datetime.timedelta(hours=8))


def _ms_to_naive_cn(ms) -> "datetime.datetime | None":
    """e签宝时间戳(epoch毫秒)转换为东八区(无时区)datetime"""
    if not ms:
        return None
    try:
        return datetime.datetime.fromtimestamp(ms / 1000, tz=CN_TZ).replace(tzinfo=None)
    except Exception:
        return None


BUNDLE_CONFIGS = {
    "STORE_STANDARD": [
        ("store_agreement", "店铺入驻协议", 1),
        ("alipay_auth", "支付宝授权函", 0),
        ("wechat_pay_commitment", "微信支付承诺函", 0),
    ],
    "LAWYER_STANDARD": [
        ("lawyer_agreement", "律所入驻协议", 1),
        ("alipay_auth", "支付宝授权函", 0),
        ("wechat_pay_commitment", "微信支付承诺函", 0),
    ],
}

DEFAULT_BUNDLE_BY_SUBJECT = {
    "store": "STORE_STANDARD",
    "lawyer": "LAWYER_STANDARD",
}


def _extract_download_url(download_resp: str) -> tuple[str | None, Any]:
    try:
        download_json = json.loads(download_resp)
    except json.JSONDecodeError as exc:
        return None, {"error": f"e签宝返回非 JSON: {exc}", "resp": download_resp}

    data = download_json.get("data") if isinstance(download_json, dict) else None
    files = data.get("files") if isinstance(data, dict) else None
    if isinstance(files, list) and files:
        first_file = files[0]
        if isinstance(first_file, dict):
            download_url = first_file.get("downloadUrl")
            if download_url:
                return download_url, download_json
    return None, download_json


class ContractCenterService:
    def __init__(self, db: AsyncSession):
        self.repo = ContractRepository(db)

    async def create_bundle(self, req: BundleCreateRequest) -> dict:
        bundle_type = DEFAULT_BUNDLE_BY_SUBJECT[req.subject_type]
        configs = BUNDLE_CONFIGS.get(bundle_type)
        if not configs:
            return {"success": False, "message": "不支持的合同包类型", "raw": {"bundle_type": bundle_type}}
        try:
            items = build_contract_items(
                configs=configs,
                template_name=req.subject_name,
                signer_name=req.subject_name,
                signer_id_num=req.ord_id,
                psn_account=req.contact_phone,
                psn_name=req.contact_name,
            )
        except ContractBuildError as exc:
            return {"success": False, "message": exc.message, "raw": exc.raw}

        bundle = await self.repo.create_bundle(
            {
                "subject_type": req.subject_type,
                "subject_id": req.subject_id,
                "subject_name": req.subject_name,
                "business_segment": req.business_segment,
                "contact_name": req.contact_name,
                "contact_phone": req.contact_phone,
                "ord_id": req.ord_id,
                "bundle_type": bundle_type,
                "status": "未签署",
            }
        )
        documents = await self.repo.create_documents(bundle.id, items)
        for document in documents:
            try:
                sign_resp = sign_url(document.sign_flow_id, req.contact_phone)
                sign_json = json.loads(sign_resp)
                sign_data = sign_json.get("data") if isinstance(sign_json, dict) else None
                result_sign_url = sign_data.get("url") if isinstance(sign_data, dict) else None
            except Exception:
                await self.repo.rollback()
                return {
                    "success": False,
                    "message": f"{document.contract_name}创建成功但签署链接获取失败",
                    "raw": {"contract_type": document.contract_type, "sign_flow_id": document.sign_flow_id},
                }
            if not result_sign_url:
                await self.repo.rollback()
                return {
                    "success": False,
                    "message": f"{document.contract_name}创建成功但签署链接缺失",
                    "raw": {"contract_type": document.contract_type, "sign_flow_id": document.sign_flow_id, "resp": sign_json},
                }
            await self.repo.update_document_urls(document.id, sign_url=result_sign_url)
            document.sign_url = result_sign_url
        primary_doc = next((doc for doc in documents if doc.is_primary == 1), documents[0])
        await self.repo.set_primary_document(bundle.id, primary_doc.id)
        await self.repo.commit()
        return {
            "success": True,
            "message": "合同包创建成功",
            "bundle_id": bundle.id,
            "primary_sign_flow_id": primary_doc.sign_flow_id,
            "created_contracts": [
                {
                    "contract_type": d.contract_type,
                    "contract_name": d.contract_name,
                    "sign_flow_id": d.sign_flow_id,
                    "file_id": d.file_id,
                    "contract_url": d.template_url,
                    "sign_url": d.sign_url,
                }
                for d in documents
            ],
        }

    async def list_bundles(self, subject_type: str, subject_id: int, page: int, page_size: int, *, doc_status: int | None = None) -> dict:
        bundles, total = await self.repo.list_bundles(subject_type, subject_id, page, page_size)
        ids = [b.id for b in bundles]
        docs_map = await self.repo.list_documents_by_bundle_ids(ids, doc_status=doc_status)
        items = []
        for b in bundles:
            docs = docs_map.get(b.id, [])
            items.append(
                {
                    "id": b.id,
                    "subject_type": b.subject_type,
                    "subject_id": b.subject_id,
                    "subject_name": b.subject_name,
                    "business_segment": b.business_segment,
                    "contact_name": b.contact_name,
                    "contact_phone": b.contact_phone,
                    "ord_id": b.ord_id,
                    "bundle_type": b.bundle_type,
                    "status": b.status,
                    "primary_document_id": b.primary_document_id,
                    "documents": [
                        {
                            "id": d.id,
                            "contract_type": d.contract_type,
                            "contract_name": d.contract_name,
                            "is_primary": d.is_primary,
                            "status": d.status,
                            "sign_flow_id": d.sign_flow_id,
                            "file_id": d.file_id,
                            "template_url": d.template_url,
                            "sign_url": d.sign_url,
                            "download_url": d.download_url,
                            "signing_time": d.signing_time,
                            "effective_time": d.effective_time,
                            "expiry_time": d.expiry_time,
                        }
                        for d in docs
                    ],
                }
            )
        total_pages = (total + page_size - 1) // page_size if total > 0 else 0
        return {"items": items, "total": total, "page": page, "page_size": page_size, "total_pages": total_pages}

    async def get_document_detail(self, sign_flow_id: str) -> dict:
        document, bundle = await self.repo.get_document_and_bundle(sign_flow_id)
        if not document:
            return {"success": False, "message": "未找到合同"}
        if document.status == 0:
            return await self._get_pending_detail(document, bundle)
        return await self._get_signed_detail(document, bundle)

    async def _get_pending_detail(self, document, bundle):
        try:
            detail_resp = esign_main.get_contract_detail(document.file_id)
            detail_json = json.loads(detail_resp)
            data = detail_json.get("data") if isinstance(detail_json, dict) else None
            contract_url_val = data.get("fileDownloadUrl") if isinstance(data, dict) else None
            if not contract_url_val and isinstance(detail_json, dict):
                contract_url_val = detail_json.get("fileDownloadUrl")
        except Exception as exc:
            return {"success": False, "message": "获取合同链接失败", "raw": str(exc)}
        if not contract_url_val:
            return {"success": False, "message": "e签宝返回缺少合同链接", "raw": detail_resp}

        await self.repo.update_document_urls(document.id, template_url=contract_url_val)
        try:
            sign_resp = sign_url(document.sign_flow_id, bundle.contact_phone)
            sign_json = json.loads(sign_resp)
            sign_data = sign_json.get("data") if isinstance(sign_json, dict) else None
            result_sign_url = sign_data.get("url") if isinstance(sign_data, dict) else None
        except Exception as exc:
            return {"success": False, "message": "获取签署链接失败", "raw": str(exc)}
        if not result_sign_url:
            return {"success": False, "message": "e签宝返回缺少签署链接", "raw": sign_json}

        await self.repo.update_document_urls(document.id, sign_url=result_sign_url)
        await self.repo.commit()
        return {
            "status": 0,
            "contract_url": contract_url_val,
            "sign_url": result_sign_url,
            "sign_flow_id": document.sign_flow_id,
        }

    async def _get_signed_detail(self, document, _bundle):
        try:
            download_resp = file_download_url(document.sign_flow_id)
        except Exception as exc:
            cached_url = document.download_url or None
            if cached_url:
                return {
                    "status": 1,
                    "contract_url": cached_url,
                    "contract_download_url": cached_url,
                    "sign_flow_id": document.sign_flow_id,
                }
            return {"success": False, "message": "获取合同下载链接失败", "raw": str(exc)}
        contract_download_url, raw = _extract_download_url(download_resp)
        if not contract_download_url:
            cached_url = document.download_url or None
            if cached_url:
                return {
                    "status": 1,
                    "contract_url": cached_url,
                    "contract_download_url": cached_url,
                    "sign_flow_id": document.sign_flow_id,
                }
            logger.error(
                "file_download_url missing downloadUrl sign_flow_id=%s resp=%s",
                document.sign_flow_id,
                download_resp,
            )
            return {"success": False, "message": "合同已签署，下载文件生成中，请稍后重试", "raw": raw}
        await self.repo.update_document_urls(document.id, template_url=contract_download_url, download_url=contract_download_url)
        await self.repo.commit()
        return {
            "status": 1,
            "contract_url": contract_download_url,
            "contract_download_url": contract_download_url,
            "sign_flow_id": document.sign_flow_id,
        }

    async def process_esign_callback(self, payload: dict) -> dict:
        sign_result = payload.get("signResult")
        sign_flow_id = payload.get("signFlowId")
        action = payload.get("action")
        operator_mobile = (
            payload.get("operator", {})
            .get("psnAccount", {})
            .get("accountMobile")
        )
        if not sign_flow_id:
            logger.info(
                "esign_callback_event %s",
                json.dumps(
                    {
                        "result": "ignored",
                        "reason": "missing_signFlowId",
                        "action": action,
                        "sign_result": sign_result,
                        "operator_mobile": operator_mobile,
                    },
                    ensure_ascii=False,
                ),
            )
            return {"success": True, "code": "200", "msg": "ignored_missing_signFlowId"}

        document, bundle = await self.repo.get_document_and_bundle(sign_flow_id)
        if not document:
            logger.info(
                "esign_callback_event %s",
                json.dumps(
                    {
                        "result": "ignored",
                        "reason": "unknown_signFlowId",
                        "sign_flow_id": sign_flow_id,
                        "action": action,
                        "sign_result": sign_result,
                        "operator_mobile": operator_mobile,
                    },
                    ensure_ascii=False,
                ),
            )
            return {"success": True, "code": "200", "msg": "ignored_unknown_signFlowId"}

        event_type = f"esign_callback:{action or 'UNKNOWN'}"
        await self.repo.create_event(bundle.id, document.id, sign_flow_id, event_type[:50], payload)

        # 不能凭"单个签署任务完成(SIGN_MISSON_COMPLETE / signResult=2)"判定整份合同已签署，
        # 因为平台方自动盖章也会触发该回调。以 e签宝 流程真实状态 signFlowStatus 为准。
        flow_status, finish_ms = self._query_flow_status(sign_flow_id)

        # 兜底：查询失败时，仅当为"流程结束"类回调且报文明确为已完成(2)才认定完成
        if flow_status is None and action in SIGN_FLOW_FINISH_ACTIONS:
            payload_status = payload.get("signFlowStatus") or payload.get("flowStatus")
            if payload_status in (2, "2"):
                flow_status = ESIGN_FLOW_STATUS_COMPLETED

        mark_signed = flow_status == ESIGN_FLOW_STATUS_COMPLETED

        logger.info(
            "esign_callback_event %s",
            json.dumps(
                {
                    "result": "mark_signed" if mark_signed else "ignored",
                    "sign_flow_id": sign_flow_id,
                    "bundle_id": bundle.id,
                    "document_id": document.id,
                    "contract_type": document.contract_type,
                    "action": action,
                    "sign_result": sign_result,
                    "flow_status": flow_status,
                    "operator_mobile": operator_mobile,
                },
                ensure_ascii=False,
            ),
        )

        if mark_signed:
            ts_ms = finish_ms or payload.get("operateTime") or payload.get("timestamp")
            signing_dt = _ms_to_naive_cn(ts_ms)
            effective_dt = expiry_dt = None
            if signing_dt:
                effective_dt = (signing_dt + datetime.timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
                expiry_dt = effective_dt + datetime.timedelta(days=365)
            contract_download_url = None
            try:
                download_resp = file_download_url(sign_flow_id)
                contract_download_url, raw_download_resp = _extract_download_url(download_resp)
            except Exception as exc:
                raw_download_resp = str(exc)
            if not contract_download_url:
                logger.error(
                    "file_download_url missing downloadUrl on callback sign_flow_id=%s resp=%s",
                    sign_flow_id,
                    raw_download_resp,
                )
            await self.repo.mark_document_signed(document.id, signing_dt, effective_dt, expiry_dt, contract_download_url)
            await self.repo.recalc_bundle_status(bundle.id)
            await self.repo.commit()
            return {"success": True, "code": "200", "msg": "success"}

        await self.repo.commit()
        return {"success": True, "code": "200", "msg": f"ignored_flow_status_{flow_status}"}

    def _query_flow_status(self, sign_flow_id: str) -> tuple[int | None, int | None]:
        """查询 e签宝 流程真实状态，返回 (signFlowStatus, signFlowFinishTime毫秒)；失败返回 (None, None)"""
        try:
            detail_resp = esign_main.query_sign_flow_detail(sign_flow_id)
            detail_json = json.loads(detail_resp)
        except Exception as exc:
            logger.error("query_sign_flow_detail error sign_flow_id=%s err=%s", sign_flow_id, exc)
            return None, None
        data = detail_json.get("data") if isinstance(detail_json, dict) else None
        if not isinstance(data, dict):
            logger.error("query_sign_flow_detail no data sign_flow_id=%s resp=%s", sign_flow_id, detail_json)
            return None, None
        return data.get("signFlowStatus"), data.get("signFlowFinishTime")