Files
ai-contract-platform/backend/app/services/contract_generation_service.py
T
2026-04-24 13:44:17 +08:00

718 lines
25 KiB
Python

import json
import re
import io
import base64
from typing import Any, Dict, List, Optional, Tuple
from datetime import datetime
from uuid import uuid4
from enum import Enum
from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from app.models.generated_contract import GeneratedContract
from app.repositories.contract_template_repository import ContractTemplateRepository
from app.repositories.generated_contract_repository import GeneratedContractRepository
from app.services.file_service import FileService
class ExportFormat(str, Enum):
"""合同导出格式"""
DOCX = "docx"
PDF = "pdf"
HTML = "html"
TXT = "txt"
class ContractGenerationService:
def __init__(self, db: Session):
self.db = db
self.repository = ContractTemplateRepository(db)
self.generated_contract_repository = GeneratedContractRepository(db)
self.file_service = FileService(db)
def list_available_templates(
self,
current_user: Dict[str, Any],
contract_type: Optional[str] = None,
) -> List[Dict]:
"""获取可用的合同模板列表"""
tenant_id = current_user["tenant_id"]
# 获取启用的模板
templates = self.repository.list_enabled(tenant_id, contract_type)
return [
{
"id": template.id,
"name": template.name,
"contract_type": template.contract_type,
"description": template.description,
"variables": self._parse_variables(template.variables_json),
"preview_excerpt": self._get_preview_excerpt(template.content),
}
for template in templates
]
def generate_contract(
self,
current_user: Dict[str, Any],
template_id: str,
variable_values: Dict[str, str],
export_format: ExportFormat = ExportFormat.DOCX,
options: Optional[Dict[str, Any]] = None,
) -> Dict:
"""根据模板和变量值生成合同(完整版)"""
tenant_id = current_user["tenant_id"]
options = options or {}
# 获取模板
template = self.repository.get_by_id(template_id, tenant_id)
if not template:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="模板不存在或无权访问"
)
if not template.enabled:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="模板已停用"
)
# 解析模板变量
template_variables = self._parse_variables(template.variables_json)
# 验证变量值
self._validate_variable_values(template_variables, variable_values)
# 替换模板内容
generated_content = self._replace_variables(template.content, variable_values)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
# 根据导出格式生成文件
file_content, file_name = self._export_contract(
generated_content,
template.name,
timestamp,
export_format,
options
)
file_item = self.file_service.create_file_record(
current_user=current_user,
file_name=file_name,
content=file_content,
file_type="contract",
mime_type=self._get_mime_type(export_format),
require_admin=False,
)
generated_contract = GeneratedContract(
id=f"contract-{uuid4().hex[:12]}",
tenant_id=tenant_id,
template_id=template_id,
template_name=template.name,
contract_type=template.contract_type,
generated_by=current_user["id"],
generated_by_name=current_user.get("display_name", current_user.get("username")),
export_format=export_format.value,
file_id=file_item["id"],
file_name=file_name,
mime_type=self._get_mime_type(export_format),
file_size=len(file_content),
generated_content=generated_content,
variable_values_json=variable_values,
export_options_json=options,
file_content_base64=base64.b64encode(file_content).decode("utf-8"),
status="generated",
)
created_contract = self.generated_contract_repository.create(generated_contract)
return self._to_generated_contract_item(created_contract)
def list_generated_contracts(
self,
current_user: Dict[str, Any],
page: int = 1,
page_size: int = 10,
contract_type: Optional[str] = None,
keyword: Optional[str] = None,
) -> Dict[str, Any]:
tenant_id = current_user["tenant_id"]
items = self.generated_contract_repository.list_paginated(
tenant_id=tenant_id,
page=page,
page_size=page_size,
contract_type=contract_type,
keyword=keyword,
)
total = self.generated_contract_repository.count(
tenant_id=tenant_id,
contract_type=contract_type,
keyword=keyword,
)
return {
"items": [self._to_generated_contract_list_item(item) for item in items],
"total": total,
"page": page,
"page_size": page_size,
}
def get_generated_contract(self, current_user: Dict[str, Any], contract_id: str) -> Dict:
tenant_id = current_user["tenant_id"]
generated_contract = self.generated_contract_repository.get_by_id(contract_id, tenant_id)
if not generated_contract:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="合同不存在或无权访问")
return self._to_generated_contract_item(generated_contract)
def get_generated_contract_preview(self, current_user: Dict[str, Any], contract_id: str) -> Dict:
tenant_id = current_user["tenant_id"]
generated_contract = self.generated_contract_repository.get_by_id(contract_id, tenant_id)
if not generated_contract:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="合同不存在或无权访问")
return {
"id": generated_contract.id,
"template_name": generated_contract.template_name,
"contract_type": generated_contract.contract_type,
"preview_content": generated_contract.generated_content,
"export_format": generated_contract.export_format,
"file_name": generated_contract.file_name,
"generated_at": generated_contract.created_at.isoformat(),
"preview_url": f"/api/v1/contracts/{generated_contract.id}/preview",
"download_url": f"/api/v1/contracts/{generated_contract.id}/download",
}
def get_generated_contract_download(self, current_user: Dict[str, Any], contract_id: str) -> Dict:
tenant_id = current_user["tenant_id"]
generated_contract = self.generated_contract_repository.get_by_id(contract_id, tenant_id)
if not generated_contract:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="合同不存在或无权访问")
if generated_contract.file_id:
file_payload = self.file_service.get_download_payload(current_user, generated_contract.file_id)
return {
"id": generated_contract.id,
"file_id": generated_contract.file_id,
"file_name": file_payload["file_name"],
"mime_type": file_payload["mime_type"],
"content": file_payload["content"],
}
return {
"id": generated_contract.id,
"file_name": generated_contract.file_name,
"mime_type": generated_contract.mime_type,
"content": base64.b64decode(generated_contract.file_content_base64),
}
def preview_contract(
self,
current_user: Dict[str, Any],
template_id: str,
variable_values: Dict[str, str],
) -> Dict:
"""预览合同生成结果(不保存)"""
tenant_id = current_user["tenant_id"]
# 获取模板
template = self.repository.get_by_id(template_id, tenant_id)
if not template:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="模板不存在或无权访问"
)
# 解析模板变量
template_variables = self._parse_variables(template.variables_json)
# 验证变量值
self._validate_variable_values(template_variables, variable_values, preview=True)
# 替换模板内容
generated_content = self._replace_variables(template.content, variable_values)
return {
"preview_content": generated_content,
"template_name": template.name,
"variable_summary": self._get_variable_summary(template_variables, variable_values),
}
def _get_mime_type(self, export_format: ExportFormat) -> str:
if export_format == ExportFormat.DOCX:
return "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
if export_format == ExportFormat.PDF:
return "application/pdf"
if export_format == ExportFormat.HTML:
return "text/html; charset=utf-8"
return "text/plain; charset=utf-8"
def _to_generated_contract_item(self, generated_contract: GeneratedContract) -> Dict:
return {
"id": generated_contract.id,
"template_id": generated_contract.template_id,
"template_name": generated_contract.template_name,
"contract_type": generated_contract.contract_type,
"file_id": generated_contract.file_id,
"generated_content": generated_contract.generated_content,
"variable_values": generated_contract.variable_values_json or {},
"generated_at": generated_contract.created_at.isoformat(),
"export_format": generated_contract.export_format,
"file_name": generated_contract.file_name,
"file_size": generated_contract.file_size,
"file_content_base64": generated_contract.file_content_base64,
"preview_url": f"/api/v1/contracts/{generated_contract.id}/preview",
"download_url": f"/api/v1/contracts/{generated_contract.id}/download",
"metadata": {
"generated_by": generated_contract.generated_by,
"generated_by_name": generated_contract.generated_by_name,
"tenant_id": generated_contract.tenant_id,
"original_template": generated_contract.template_name,
"export_options": generated_contract.export_options_json or {},
"variables_used": len(generated_contract.variable_values_json or {}),
"asset_file_id": generated_contract.file_id,
}
}
def _to_generated_contract_list_item(self, generated_contract: GeneratedContract) -> Dict:
item = self._to_generated_contract_item(generated_contract)
return {
"id": item["id"],
"template_name": item["template_name"],
"contract_type": item["contract_type"],
"file_id": item["file_id"],
"generated_at": item["generated_at"],
"export_format": item["export_format"],
"file_name": item["file_name"],
"file_size": item["file_size"],
"preview_url": item["preview_url"],
"download_url": item["download_url"],
"metadata": {
"generated_by_name": item["metadata"]["generated_by_name"],
"asset_file_id": item["metadata"]["asset_file_id"],
},
}
def _parse_variables(self, variables_json: Any) -> List[Dict]:
"""解析模板变量JSON"""
if not variables_json:
return []
if isinstance(variables_json, list):
variables = variables_json
else:
try:
variables = json.loads(variables_json)
except (json.JSONDecodeError, TypeError):
return []
if not isinstance(variables, list):
return []
parsed_variables = []
for var in variables:
if isinstance(var, dict) and "name" in var:
parsed_variables.append({
"name": var["name"],
"label": var.get("label", var["name"]),
"required": var.get("required", True),
"default_value": var.get("default_value", ""),
"description": var.get("description", ""),
})
return parsed_variables
def _validate_variable_values(
self,
template_variables: List[Dict],
variable_values: Dict[str, str],
preview: bool = False,
) -> None:
"""验证变量值"""
# 检查必填变量
for var in template_variables:
var_name = var["name"]
is_required = var["required"]
if is_required and var_name not in variable_values:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"必填变量 '{var['label']}' 未提供"
)
# 检查变量值是否为空(如果是必填)
if is_required and var_name in variable_values:
value = variable_values[var_name]
if not value or not value.strip():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"变量 '{var['label']}' 不能为空"
)
# 检查是否有未知变量
template_var_names = {var["name"] for var in template_variables}
provided_var_names = set(variable_values.keys())
unknown_vars = provided_var_names - template_var_names
if unknown_vars and not preview:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"未知变量: {', '.join(unknown_vars)}"
)
def _replace_variables(self, template_content: str, variable_values: Dict[str, str]) -> str:
"""替换模板中的变量占位符"""
content = template_content
# 替换所有 {{variable_name}} 格式的占位符
for var_name, value in variable_values.items():
placeholder = f"{{{{{var_name}}}}}"
content = content.replace(placeholder, value)
# 清理未替换的占位符(用空字符串替换)
remaining_placeholders = re.findall(r'\{\{([^}]+)\}\}', content)
for placeholder in remaining_placeholders:
content = content.replace(f"{{{{{placeholder}}}}}", "")
return content
def _get_preview_excerpt(self, content: str, length: int = 100) -> str:
"""获取模板内容预览"""
# 移除变量占位符
preview = re.sub(r'\{\{[^}]+\}\}', '[变量]', content)
# 截取指定长度
if len(preview) > length:
preview = preview[:length] + "..."
return preview
def _get_variable_summary(
self,
template_variables: List[Dict],
variable_values: Dict[str, str],
) -> List[Dict]:
"""获取变量填写情况摘要"""
summary = []
for var in template_variables:
var_name = var["name"]
value = variable_values.get(var_name, "")
summary.append({
"name": var_name,
"label": var["label"],
"required": var["required"],
"provided": var_name in variable_values,
"value_preview": self._get_value_preview(value),
"is_empty": not value or not value.strip(),
})
return summary
def _get_value_preview(self, value: str, max_length: int = 50) -> str:
"""获取变量值预览"""
if not value:
return "(空)"
value = str(value).strip()
if len(value) > max_length:
return value[:max_length] + "..."
return value
def _export_contract(
self,
content: str,
template_name: str,
timestamp: str,
export_format: ExportFormat,
options: Dict[str, Any],
) -> Tuple[bytes, str]:
"""导出合同到不同格式"""
# 清理模板名称用于文件名
safe_name = re.sub(r'[^\w\s-]', '', template_name)
safe_name = re.sub(r'[-\s]+', '_', safe_name).strip('-_').lower()
if export_format == ExportFormat.DOCX:
return self._export_to_docx(content, safe_name, timestamp, options)
elif export_format == ExportFormat.PDF:
return self._export_to_pdf(content, safe_name, timestamp, options)
elif export_format == ExportFormat.HTML:
return self._export_to_html(content, safe_name, timestamp, options)
else: # TXT
return self._export_to_txt(content, safe_name, timestamp, options)
def _export_to_docx(
self,
content: str,
safe_name: str,
timestamp: str,
options: Dict[str, Any],
) -> Tuple[bytes, str]:
"""导出为DOCX格式"""
try:
# 尝试导入python-docx
from docx import Document
from docx.shared import Inches, Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.enum.style import WD_STYLE_TYPE
# 创建新文档
doc = Document()
# 设置文档样式
style = doc.styles['Normal']
style.font.name = '宋体'
style.font.size = Pt(12)
# 添加标题
title = doc.add_heading('合同文件', 0)
title.alignment = WD_ALIGN_PARAGRAPH.CENTER
# 添加生成信息
info_para = doc.add_paragraph()
info_para.add_run(f'生成时间: {timestamp}').italic = True
info_para.alignment = WD_ALIGN_PARAGRAPH.RIGHT
# 添加分隔线
doc.add_paragraph('-' * 50)
# 添加合同内容
# 按段落分割内容
paragraphs = content.split('\n')
for para_text in paragraphs:
if para_text.strip():
para = doc.add_paragraph(para_text.strip())
para.style = doc.styles['Normal']
# 添加页脚
section = doc.sections[0]
footer = section.footer
footer_para = footer.paragraphs[0]
footer_para.text = f"AI合同平台生成 - {safe_name}"
footer_para.alignment = WD_ALIGN_PARAGRAPH.CENTER
# 保存到内存
import io
output = io.BytesIO()
doc.save(output)
file_content = output.getvalue()
file_name = f"{safe_name}_{timestamp}.docx"
return file_content, file_name
except ImportError:
# 如果python-docx不可用,使用简单的文本格式
return self._export_to_txt(content, safe_name, timestamp, options)
def _export_to_pdf(
self,
content: str,
safe_name: str,
timestamp: str,
options: Dict[str, Any],
) -> Tuple[bytes, str]:
"""导出为PDF格式"""
try:
# 尝试导入reportlab
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
from reportlab.lib.units import cm
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer
from reportlab.lib.enums import TA_CENTER, TA_RIGHT
import io
# 创建内存中的PDF
buffer = io.BytesIO()
# 创建文档
doc = SimpleDocTemplate(
buffer,
pagesize=A4,
rightMargin=2*cm,
leftMargin=2*cm,
topMargin=2*cm,
bottomMargin=2*cm
)
# 准备内容
styles = getSampleStyleSheet()
story = []
# 标题
title_style = styles['Title']
title_style.alignment = TA_CENTER
story.append(Paragraph('合同文件', title_style))
story.append(Spacer(1, 0.5*cm))
# 生成信息
info_style = styles['Normal']
info_style.alignment = TA_RIGHT
story.append(Paragraph(f'生成时间: {timestamp}', info_style))
story.append(Spacer(1, 0.5*cm))
# 分隔线
story.append(Paragraph('-' * 50, styles['Normal']))
story.append(Spacer(1, 0.5*cm))
# 合同内容
paragraphs = content.split('\n')
for para_text in paragraphs:
if para_text.strip():
story.append(Paragraph(para_text.strip(), styles['Normal']))
story.append(Spacer(1, 0.2*cm))
# 页脚
footer_style = styles['Normal']
footer_style.alignment = TA_CENTER
story.append(Spacer(1, 1*cm))
story.append(Paragraph(f"AI合同平台生成 - {safe_name}", footer_style))
# 构建PDF
doc.build(story)
file_content = buffer.getvalue()
buffer.close()
file_name = f"{safe_name}_{timestamp}.pdf"
return file_content, file_name
except ImportError:
# 如果reportlab不可用,使用HTML格式
return self._export_to_html(content, safe_name, timestamp, options)
def _export_to_html(
self,
content: str,
safe_name: str,
timestamp: str,
options: Dict[str, Any],
) -> Tuple[bytes, str]:
"""导出为HTML格式"""
html_template = f"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>合同文件 - {safe_name}</title>
<style>
body {{
font-family: 'SimSun', '宋体', serif;
line-height: 1.6;
margin: 0;
padding: 20px;
background-color: #f8f9fa;
}}
.container {{
max-width: 800px;
margin: 0 auto;
background-color: white;
padding: 40px;
box-shadow: 0 0 20px rgba(0,0,0,0.1);
border-radius: 8px;
}}
.header {{
text-align: center;
margin-bottom: 30px;
border-bottom: 2px solid #1890ff;
padding-bottom: 20px;
}}
.title {{
font-size: 28px;
color: #1890ff;
margin-bottom: 10px;
}}
.timestamp {{
text-align: right;
color: #666;
font-style: italic;
margin-bottom: 20px;
}}
.content {{
font-size: 14px;
white-space: pre-wrap;
margin-bottom: 30px;
}}
.footer {{
text-align: center;
color: #999;
font-size: 12px;
border-top: 1px solid #eee;
padding-top: 20px;
margin-top: 30px;
}}
.separator {{
border-top: 1px dashed #ddd;
margin: 20px 0;
}}
@media print {{
body {{
background-color: white;
padding: 0;
}}
.container {{
box-shadow: none;
padding: 0;
}}
}}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1 class="title">合同文件</h1>
</div>
<div class="timestamp">
生成时间: {timestamp}
</div>
<div class="separator"></div>
<div class="content">
{content}
</div>
<div class="separator"></div>
<div class="footer">
AI合同平台生成 - {safe_name}
</div>
</div>
</body>
</html>"""
file_content = html_template.encode('utf-8')
file_name = f"{safe_name}_{timestamp}.html"
return file_content, file_name
def _export_to_txt(
self,
content: str,
safe_name: str,
timestamp: str,
options: Dict[str, Any],
) -> Tuple[bytes, str]:
"""导出为TXT格式"""
txt_content = f"""合同文件
{'=' * 50}
生成时间: {timestamp}
{'=' * 50}
{content}
{'=' * 50}
AI合同平台生成 - {safe_name}
"""
file_content = txt_content.encode('utf-8')
file_name = f"{safe_name}_{timestamp}.txt"
return file_content, file_name