634 lines
26 KiB
Python
634 lines
26 KiB
Python
from datetime import datetime
|
|
from typing import Optional
|
|
from urllib.parse import quote
|
|
|
|
from fastapi import APIRouter, BackgroundTasks, Depends, Form, Header, HTTPException, Request, Response
|
|
from pydantic import BaseModel, Field
|
|
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
|
|
from app.services.tenant_service import TenantService
|
|
|
|
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",
|
|
"versions": "contracts:versions:list",
|
|
"version_detail": "contracts:versions:detail",
|
|
"version_create": "contracts:versions:create",
|
|
"version_diff": "contracts:versions:diff",
|
|
"version_download": "contracts:versions:download",
|
|
"lifecycle_update": "contracts:lifecycle:update",
|
|
"lifecycle_tasks": "contracts:lifecycle-tasks:list",
|
|
"lifecycle_task_sync": "contracts:lifecycle-tasks:sync",
|
|
"lifecycle_task_create": "contracts:lifecycle-tasks:create",
|
|
"lifecycle_task_update": "contracts:lifecycle-tasks:update",
|
|
"preview_generated": "contracts:generated:preview",
|
|
"download": "contracts:download",
|
|
}
|
|
|
|
|
|
class GeneratedContractVersionCreateRequest(BaseModel):
|
|
content: str = Field(..., min_length=1, max_length=1_000_000)
|
|
change_type: str = "editor_save"
|
|
change_summary: Optional[str] = None
|
|
source_version_id: Optional[str] = None
|
|
original_text: Optional[str] = None
|
|
replacement_text: Optional[str] = None
|
|
locate_status: str = "applied"
|
|
|
|
|
|
class GeneratedContractLifecycleUpdateRequest(BaseModel):
|
|
lifecycle_status: str = Field(..., min_length=1, max_length=32)
|
|
counterparty_name: Optional[str] = Field(default=None, max_length=120)
|
|
owner_name: Optional[str] = Field(default=None, max_length=100)
|
|
effective_at: Optional[datetime] = None
|
|
expires_at: Optional[datetime] = None
|
|
lifecycle_note: Optional[str] = Field(default=None, max_length=1000)
|
|
|
|
|
|
class ContractLifecycleTaskCreateRequest(BaseModel):
|
|
task_type: str = Field(..., min_length=1, max_length=64)
|
|
title: str = Field(..., min_length=1, max_length=200)
|
|
description: Optional[str] = Field(default=None, max_length=4000)
|
|
priority: str = Field(default="medium", max_length=32)
|
|
status_value: str = Field(default="open", max_length=32)
|
|
due_at: Optional[datetime] = None
|
|
evidence_file_id: Optional[str] = Field(default=None, max_length=64)
|
|
evidence_note: Optional[str] = Field(default=None, max_length=2000)
|
|
|
|
|
|
class ContractLifecycleTaskUpdateRequest(BaseModel):
|
|
task_type: Optional[str] = Field(default=None, max_length=64)
|
|
title: Optional[str] = Field(default=None, max_length=200)
|
|
description: Optional[str] = Field(default=None, max_length=4000)
|
|
priority: Optional[str] = Field(default=None, max_length=32)
|
|
status_value: Optional[str] = Field(default=None, max_length=32)
|
|
due_at: Optional[datetime] = None
|
|
evidence_file_id: Optional[str] = Field(default=None, max_length=64)
|
|
evidence_note: Optional[str] = Field(default=None, max_length=2000)
|
|
|
|
|
|
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 get_tenant_service(db: Session = Depends(get_db_session)) -> TenantService:
|
|
return TenantService(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,
|
|
document_type_code: Optional[str] = None,
|
|
lifecycle_status: 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,
|
|
document_type_code=document_type_code,
|
|
lifecycle_status=lifecycle_status,
|
|
keyword=keyword,
|
|
)
|
|
)
|
|
|
|
|
|
@router.get("/templates")
|
|
def list_available_templates(
|
|
contract_type: Optional[str] = None,
|
|
document_type_code: Optional[str] = None,
|
|
template_category: Optional[str] = None,
|
|
industry_category: Optional[str] = None,
|
|
contract_generation_service: ContractGenerationService = Depends(get_contract_generation_service),
|
|
current_user: dict = Depends(require_current_user),
|
|
):
|
|
kwargs = {
|
|
"current_user": current_user,
|
|
"contract_type": contract_type,
|
|
}
|
|
if document_type_code is not None:
|
|
kwargs["document_type_code"] = document_type_code
|
|
if template_category is not None:
|
|
kwargs["template_category"] = template_category
|
|
if industry_category is not None:
|
|
kwargs["industry_category"] = industry_category
|
|
return success_response(contract_generation_service.list_available_templates(**kwargs))
|
|
|
|
|
|
@router.post("/generate")
|
|
def generate_contract(
|
|
request: Request,
|
|
background_tasks: BackgroundTasks,
|
|
template_id: Optional[str] = Form(default=None),
|
|
variable_values_json: Optional[str] = Form(default=None),
|
|
export_format: str = Form(default="docx"),
|
|
options_json: Optional[str] = Form(default=None),
|
|
document_type_code: Optional[str] = Form(default=None),
|
|
elements_json: Optional[str] = Form(default=None),
|
|
element_schema_id: Optional[str] = Form(default=None),
|
|
element_schema_version_label: Optional[str] = Form(default=None),
|
|
contract_generation_service: ContractGenerationService = Depends(get_contract_generation_service),
|
|
audit_log_service: AuditLogService = Depends(get_audit_log_service),
|
|
tenant_service: TenantService = Depends(get_tenant_service),
|
|
current_user: dict = Depends(require_current_user),
|
|
):
|
|
import json
|
|
|
|
try:
|
|
variable_values = json.loads(variable_values_json) if variable_values_json else {}
|
|
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对象")
|
|
|
|
elements_snapshot = None
|
|
if elements_json:
|
|
elements_snapshot = json.loads(elements_json)
|
|
if not isinstance(elements_snapshot, dict):
|
|
raise HTTPException(status_code=400, detail="elements_json must be a JSON object")
|
|
|
|
try:
|
|
export_format_enum = ExportFormat(export_format.lower())
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail=f"不支持的导出格式: {export_format}")
|
|
|
|
tenant_service.assert_billing_allows(current_user, "generation")
|
|
|
|
if elements_snapshot is not None:
|
|
if not document_type_code:
|
|
raise HTTPException(status_code=400, detail="document_type_code is required when elements_json is provided")
|
|
result = contract_generation_service.generate_contract_from_elements(
|
|
current_user=current_user,
|
|
document_type_code=document_type_code,
|
|
elements=elements_snapshot,
|
|
export_format=export_format_enum,
|
|
options=options,
|
|
template_id=template_id,
|
|
variable_values=variable_values,
|
|
)
|
|
else:
|
|
if not template_id:
|
|
raise HTTPException(status_code=400, detail="template_id is required when elements_json is not provided")
|
|
if not variable_values_json:
|
|
raise HTTPException(status_code=400, detail="variable_values_json is required when elements_json is not provided")
|
|
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,
|
|
document_type_code=document_type_code,
|
|
elements_snapshot=elements_snapshot,
|
|
element_schema_id=element_schema_id,
|
|
element_schema_version_label=element_schema_version_label,
|
|
)
|
|
|
|
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": result.get("template_id") or template_id,
|
|
"template_name": result["template_name"],
|
|
"contract_type": result["contract_type"],
|
|
"document_type_code": result.get("document_type_code"),
|
|
"element_schema_id": result.get("metadata", {}).get("element_schema_id"),
|
|
"element_schema_version_label": result.get("metadata", {}).get("element_schema_version_label"),
|
|
"template_category": result.get("template_category"),
|
|
"industry_category": result.get("industry_category"),
|
|
"export_format": export_format,
|
|
"variables_count": len(result.get("variable_values") or variable_values),
|
|
"missing_field_count": len(result.get("missing_fields") or []),
|
|
"related_legal_article_count": len(result.get("related_legal_articles") or []),
|
|
"file_name": result["file_name"],
|
|
},
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
|
|
if result.get("id"):
|
|
background_tasks.add_task(
|
|
contract_generation_service._execute_deferred_auto_review,
|
|
result["id"],
|
|
current_user["tenant_id"],
|
|
current_user,
|
|
)
|
|
|
|
return success_response(result)
|
|
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(status_code=400, detail="JSON解析错误")
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"合同生成失败: {exc}")
|
|
|
|
|
|
@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 exc:
|
|
raise HTTPException(status_code=500, detail=f"合同预览失败: {exc}")
|
|
|
|
|
|
@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.patch("/{contract_id}/lifecycle")
|
|
def update_generated_contract_lifecycle(
|
|
contract_id: str,
|
|
payload: GeneratedContractLifecycleUpdateRequest,
|
|
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.update_generated_contract_lifecycle(
|
|
current_user=current_user,
|
|
contract_id=contract_id,
|
|
lifecycle_status=payload.lifecycle_status,
|
|
counterparty_name=payload.counterparty_name,
|
|
owner_name=payload.owner_name,
|
|
effective_at=payload.effective_at,
|
|
expires_at=payload.expires_at,
|
|
lifecycle_note=payload.lifecycle_note,
|
|
)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="contract_lifecycle_update",
|
|
resource_type="contract",
|
|
resource_id=contract_id,
|
|
detail_json={
|
|
"lifecycle_status": result.get("lifecycle", {}).get("status"),
|
|
"counterparty_name": result.get("lifecycle", {}).get("counterparty_name"),
|
|
"owner_name": result.get("lifecycle", {}).get("owner_name"),
|
|
"effective_at": result.get("lifecycle", {}).get("effective_at"),
|
|
"expires_at": result.get("lifecycle", {}).get("expires_at"),
|
|
"due_status": result.get("lifecycle", {}).get("due_status"),
|
|
},
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@router.get("/{contract_id}/lifecycle/tasks")
|
|
def list_contract_lifecycle_tasks(
|
|
contract_id: str,
|
|
status_value: Optional[str] = None,
|
|
task_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_contract_lifecycle_tasks(
|
|
current_user=current_user,
|
|
contract_id=contract_id,
|
|
status_value=status_value,
|
|
task_type=task_type,
|
|
)
|
|
)
|
|
|
|
|
|
@router.post("/{contract_id}/lifecycle/tasks/sync")
|
|
def sync_contract_lifecycle_tasks(
|
|
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.sync_contract_lifecycle_tasks(current_user=current_user, contract_id=contract_id)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="contract_lifecycle_task_sync",
|
|
resource_type="contract",
|
|
resource_id=contract_id,
|
|
detail_json={
|
|
"synced": result.get("synced"),
|
|
"reason": result.get("reason"),
|
|
"total": result.get("total"),
|
|
"summary": result.get("summary"),
|
|
},
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@router.post("/{contract_id}/lifecycle/tasks")
|
|
def create_contract_lifecycle_task(
|
|
contract_id: str,
|
|
payload: ContractLifecycleTaskCreateRequest,
|
|
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.create_contract_lifecycle_task(
|
|
current_user=current_user,
|
|
contract_id=contract_id,
|
|
task_type=payload.task_type,
|
|
title=payload.title,
|
|
description=payload.description,
|
|
priority=payload.priority,
|
|
status_value=payload.status_value,
|
|
due_at=payload.due_at,
|
|
evidence_file_id=payload.evidence_file_id,
|
|
evidence_note=payload.evidence_note,
|
|
)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="contract_lifecycle_task_create",
|
|
resource_type="contract_lifecycle_task",
|
|
resource_id=result.get("id"),
|
|
detail_json={
|
|
"contract_id": contract_id,
|
|
"task_type": result.get("task_type"),
|
|
"priority": result.get("priority"),
|
|
"status": result.get("status"),
|
|
"due_at": result.get("due_at"),
|
|
"evidence_file_id": result.get("evidence_file_id"),
|
|
},
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@router.put("/{contract_id}/lifecycle/tasks/{task_id}")
|
|
def update_contract_lifecycle_task(
|
|
contract_id: str,
|
|
task_id: str,
|
|
payload: ContractLifecycleTaskUpdateRequest,
|
|
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.update_contract_lifecycle_task(
|
|
current_user=current_user,
|
|
contract_id=contract_id,
|
|
task_id=task_id,
|
|
task_type=payload.task_type,
|
|
title=payload.title,
|
|
description=payload.description,
|
|
priority=payload.priority,
|
|
status_value=payload.status_value,
|
|
due_at=payload.due_at,
|
|
evidence_file_id=payload.evidence_file_id,
|
|
evidence_note=payload.evidence_note,
|
|
)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="contract_lifecycle_task_update",
|
|
resource_type="contract_lifecycle_task",
|
|
resource_id=task_id,
|
|
detail_json={
|
|
"contract_id": contract_id,
|
|
"task_type": result.get("task_type"),
|
|
"priority": result.get("priority"),
|
|
"status": result.get("status"),
|
|
"due_at": result.get("due_at"),
|
|
"evidence_file_id": result.get("evidence_file_id"),
|
|
},
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@router.get("/{contract_id}/versions")
|
|
def list_generated_contract_versions(
|
|
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.list_generated_contract_versions(current_user, contract_id))
|
|
|
|
|
|
@router.get("/{contract_id}/versions/diff")
|
|
def diff_generated_contract_versions(
|
|
contract_id: str,
|
|
request: Request,
|
|
from_version_id: Optional[str] = None,
|
|
to_version_id: Optional[str] = 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),
|
|
):
|
|
result = contract_generation_service.diff_generated_contract_versions(
|
|
current_user=current_user,
|
|
contract_id=contract_id,
|
|
from_version_id=from_version_id,
|
|
to_version_id=to_version_id,
|
|
)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="contract_version_diff",
|
|
resource_type="contract",
|
|
resource_id=contract_id,
|
|
detail_json={
|
|
"from_version": result.get("from_version", {}).get("version_label"),
|
|
"to_version": result.get("to_version", {}).get("version_label"),
|
|
"change_count": result.get("summary", {}).get("change_count"),
|
|
},
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@router.get("/{contract_id}/versions/{version_id}/download")
|
|
def download_generated_contract_version(
|
|
contract_id: str,
|
|
version_id: str,
|
|
request: Request,
|
|
export_format: str = "docx",
|
|
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),
|
|
):
|
|
try:
|
|
export_format_enum = ExportFormat(export_format.lower())
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail=f"不支持的导出格式: {export_format}")
|
|
|
|
result = contract_generation_service.export_generated_contract_version(
|
|
current_user=current_user,
|
|
contract_id=contract_id,
|
|
version_id=version_id,
|
|
export_format=export_format_enum,
|
|
)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="download_contract_version",
|
|
resource_type="contract_version",
|
|
resource_id=version_id,
|
|
detail_json={
|
|
"contract_id": contract_id,
|
|
"version_label": result.get("version", {}).get("version_label"),
|
|
"file_name": result["file_name"],
|
|
"mime_type": result["mime_type"],
|
|
"export_format": result.get("export_format"),
|
|
},
|
|
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)
|
|
|
|
|
|
@router.get("/{contract_id}/versions/{version_id}")
|
|
def get_generated_contract_version(
|
|
contract_id: str,
|
|
version_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_version(current_user, contract_id, version_id))
|
|
|
|
|
|
@router.post("/{contract_id}/versions")
|
|
def create_generated_contract_version(
|
|
contract_id: str,
|
|
payload: GeneratedContractVersionCreateRequest,
|
|
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.create_generated_contract_version(
|
|
current_user=current_user,
|
|
contract_id=contract_id,
|
|
content=payload.content,
|
|
change_type=payload.change_type,
|
|
change_summary=payload.change_summary,
|
|
source_version_id=payload.source_version_id,
|
|
original_text=payload.original_text,
|
|
replacement_text=payload.replacement_text,
|
|
locate_status=payload.locate_status,
|
|
)
|
|
audit_log_service.record_event(
|
|
tenant_id=current_user.get("tenant_id"),
|
|
user_id=current_user.get("id"),
|
|
action="contract_version_create",
|
|
resource_type="contract_version",
|
|
resource_id=result.get("id"),
|
|
detail_json={
|
|
"contract_id": contract_id,
|
|
"version_label": result.get("version_label"),
|
|
"change_type": result.get("change_type"),
|
|
"change_summary": result.get("change_summary"),
|
|
"content_length": len(payload.content or ""),
|
|
},
|
|
ip_address=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@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)
|