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()

SIGN_FLOW_FINISH_ACTIONS = {
    "SIGN_FLOW_FINISH",
    "SIGN_FLOW_COMPLETE",
}

NON_FINAL_SIGN_ACTIONS = {
    "OPERATOR_COMPLETE_SIGN",
    "OPERATOR_READ",
    "OPERATOR_VIEW",
}

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)

        mark_signed = bool(action in SIGN_FLOW_FINISH_ACTIONS or (sign_result == 2 and action not in NON_FINAL_SIGN_ACTIONS))

        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,
                    "operator_mobile": operator_mobile,
                },
                ensure_ascii=False,
            ),
        )

        if mark_signed:
            ts_ms = payload.get("operateTime") or payload.get("timestamp")
            signing_dt = None
            if ts_ms:
                try:
                    signing_dt = datetime.datetime.fromtimestamp(ts_ms / 1000)
                except Exception:
                    signing_dt = None
            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()
        if action:
            return {"success": True, "code": "200", "msg": f"ignored_action_{action}"}
        return {"success": True, "code": "200", "msg": f"ignored_signResult_{sign_result}"}