ai-contract-platform/backend/app/api/contracts.py

from typing import Optional
from urllib.parse import quote

from fastapi import APIRouter, Depends, Form, Header, HTTPException, Request, Response
from sqlalchemy.orm import Session

from app.core.response import success_response
from app.db import get_db_session
from app.services.audit_log_service import AuditLogService
from app.services.auth_service import AuthService
from app.services.contract_generation_service import ContractGenerationService, ExportFormat

router = APIRouter(prefix="/contracts", tags=["contracts"])
ROUTE_PERMISSION_SCOPE = "contract_generation"
ROUTE_PERMISSION_ACTIONS = {
    "list": "contracts:list",
    "templates": "contracts:templates:list",
    "generate": "contracts:generate",
    "preview": "contracts:preview",
    "detail": "contracts:detail",
    "preview_generated": "contracts:generated:preview",
    "download": "contracts:download",
}


def get_contract_generation_service(db: Session = Depends(get_db_session)) -> ContractGenerationService:
    return ContractGenerationService(db)


def get_audit_log_service(db: Session = Depends(get_db_session)) -> AuditLogService:
    return AuditLogService(db)


def require_current_user(
    authorization: Optional[str] = Header(default=None),
    db: Session = Depends(get_db_session),
):
    return AuthService(db).require_current_user(authorization)


@router.get("")
def list_generated_contracts(
    page: int = 1,
    page_size: int = 10,
    contract_type: Optional[str] = None,
    keyword: Optional[str] = None,
    contract_generation_service: ContractGenerationService = Depends(get_contract_generation_service),
    current_user: dict = Depends(require_current_user),
):
    """获取已生成合同列表"""
    return success_response(
        contract_generation_service.list_generated_contracts(
            current_user=current_user,
            page=page,
            page_size=page_size,
            contract_type=contract_type,
            keyword=keyword,
        )
    )


@router.get("/templates")
def list_available_templates(
    contract_type: Optional[str] = None,
    contract_generation_service: ContractGenerationService = Depends(get_contract_generation_service),
    current_user: dict = Depends(require_current_user),
):
    """获取可用的合同模板列表"""
    return success_response(contract_generation_service.list_available_templates(current_user, contract_type))


@router.post("/generate")
def generate_contract(
    request: Request,
    template_id: str = Form(...),
    variable_values_json: str = Form(...),
    export_format: str = Form(default="docx"),
    options_json: Optional[str] = Form(default=None),
    contract_generation_service: ContractGenerationService = Depends(get_contract_generation_service),
    audit_log_service: AuditLogService = Depends(get_audit_log_service),
    current_user: dict = Depends(require_current_user),
):
    """生成合同"""
    import json

    try:
        # 解析变量值
        variable_values = json.loads(variable_values_json)
        if not isinstance(variable_values, dict):
            raise HTTPException(status_code=400, detail="变量值必须是JSON对象")

        # 解析选项
        options = {}
        if options_json:
            options = json.loads(options_json)
            if not isinstance(options, dict):
                raise HTTPException(status_code=400, detail="选项必须是JSON对象")

        # 验证导出格式
        try:
            export_format_enum = ExportFormat(export_format.lower())
        except ValueError:
            raise HTTPException(status_code=400, detail=f"不支持的导出格式: {export_format}")

        # 生成合同
        result = contract_generation_service.generate_contract(
            current_user=current_user,
            template_id=template_id,
            variable_values=variable_values,
            export_format=export_format_enum,
            options=options,
        )

        # 记录审计日志
        audit_log_service.record_event(
            tenant_id=current_user.get("tenant_id"),
            user_id=current_user.get("id"),
            action="contract_generation",
            resource_type="contract",
            resource_id=result["id"],
            detail_json={
                "template_id": template_id,
                "template_name": result["template_name"],
                "contract_type": result["contract_type"],
                "export_format": export_format,
                "variables_count": len(variable_values),
                "file_name": result["file_name"],
            },
            ip_address=request.client.host if request.client else None,
            user_agent=request.headers.get("user-agent"),
        )

        return success_response(result)

    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="JSON解析错误")
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"合同生成失败: {str(e)}")


@router.post("/preview")
def preview_contract(
    template_id: str = Form(...),
    variable_values_json: str = Form(...),
    contract_generation_service: ContractGenerationService = Depends(get_contract_generation_service),
    current_user: dict = Depends(require_current_user),
):
    """预览合同生成结果"""
    import json

    try:
        # 解析变量值
        variable_values = json.loads(variable_values_json)
        if not isinstance(variable_values, dict):
            raise HTTPException(status_code=400, detail="变量值必须是JSON对象")

        # 预览合同
        result = contract_generation_service.preview_contract(
            current_user=current_user,
            template_id=template_id,
            variable_values=variable_values,
        )

        return success_response(result)

    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="JSON解析错误")
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"合同预览失败: {str(e)}")


@router.get("/{contract_id}")
def get_generated_contract(
    contract_id: str,
    contract_generation_service: ContractGenerationService = Depends(get_contract_generation_service),
    current_user: dict = Depends(require_current_user),
):
    """获取已生成合同详情"""
    return success_response(contract_generation_service.get_generated_contract(current_user, contract_id))


@router.get("/{contract_id}/preview")
def get_contract_preview(
    contract_id: str,
    contract_generation_service: ContractGenerationService = Depends(get_contract_generation_service),
    current_user: dict = Depends(require_current_user),
):
    """获取已生成合同的真实预览"""
    return success_response(contract_generation_service.get_generated_contract_preview(current_user, contract_id))


@router.get("/{contract_id}/download")
def download_contract(
    contract_id: str,
    request: Request,
    contract_generation_service: ContractGenerationService = Depends(get_contract_generation_service),
    audit_log_service: AuditLogService = Depends(get_audit_log_service),
    current_user: dict = Depends(require_current_user),
):
    """下载已生成合同文件"""
    result = contract_generation_service.get_generated_contract_download(current_user, contract_id)

    audit_log_service.record_event(
        tenant_id=current_user.get("tenant_id"),
        user_id=current_user.get("id"),
        action="download_contract",
        resource_type="contract",
        resource_id=contract_id,
        detail_json={
            "file_name": result["file_name"],
            "mime_type": result["mime_type"],
        },
        ip_address=request.client.host if request.client else None,
        user_agent=request.headers.get("user-agent"),
    )

    headers = {"Content-Disposition": f"attachment; filename*=UTF-8''{quote(result['file_name'])}"}
    return Response(content=result["content"], media_type=result["mime_type"], headers=headers)