2956 lines
106 KiB
Python
2956 lines
106 KiB
Python
"""CRM 合同 CRUD — 对照 CrmContractServiceImpl.java（列表/详情 P1）"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import re
|
||
from collections import defaultdict
|
||
from datetime import date, datetime, timedelta
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from sqlalchemy import and_, exists, func, or_
|
||
from sqlalchemy.orm import Query, Session
|
||
|
||
from app.core.delete_guard import assert_row_deletable_by_creator
|
||
from app.core.utils import generate_uuid
|
||
from app.crud.crm.data_scope_helpers import apply_crm_owner_scope, ensure_crm_owner_row_in_scope, apply_crm_create_by_scope, ensure_crm_create_by_row_in_scope
|
||
from app.crud.crm.contact import get_contact
|
||
from app.crud.crm.contract_approval import (
|
||
can_user_approve,
|
||
create_approval_records,
|
||
delete_pending_approvals,
|
||
invalidate_all_approvals,
|
||
)
|
||
from app.crud.crm.contract_product import (
|
||
batch_insert_contract_products,
|
||
delete_products_by_contract_id,
|
||
list_contract_products,
|
||
list_products_by_contract_ids,
|
||
)
|
||
from app.crud.crm.customer import get_customer
|
||
from app.crud.crm.invoice_info import list_invoice_info_by_customer
|
||
from app.services.crm.list_sort import resolve_datetime_order_by
|
||
from app.services.crm_settlement_clause import default_settlement_clause
|
||
from app.services.crm_supplier_hazard import assert_products_allowed_for_supplier
|
||
from app.crud.crm.product import resolve_product_id_for_quote_item
|
||
from app.crud.crm.quote import get_quote
|
||
from app.crud.crm.quote_product import list_quote_products
|
||
from app.crud.crm.supplier_info import get_supplier_info
|
||
from app.models.crm_models import (
|
||
CrmContract,
|
||
CrmContractApproval,
|
||
CrmContractChange,
|
||
CrmContractProduct,
|
||
CrmContractTemplate,
|
||
CrmOpportunity,
|
||
CrmProduct,
|
||
CrmQuote,
|
||
)
|
||
from app.models.models import SysUser
|
||
|
||
_FMT = "%Y-%m-%d %H:%M:%S"
|
||
|
||
def _ensure_contract_in_scope(db: Session, row: CrmContract, user_info, tenant_no: str) -> None:
|
||
ensure_crm_owner_row_in_scope(db, row, user_info, tenant_no, message="无权访问该合同")
|
||
|
||
|
||
def _get_contract_row_in_scope(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
user_info=None,
|
||
) -> CrmContract:
|
||
row = get_contract_row(db, contract_id, tenant_no=tenant_no)
|
||
if not row:
|
||
raise ValueError("合同不存在")
|
||
_ensure_contract_in_scope(db, row, user_info, tenant_no)
|
||
return row
|
||
|
||
|
||
_DATE_FMT = "%Y-%m-%d"
|
||
|
||
|
||
def total_pages(total: int, page_size: int) -> int:
|
||
if page_size <= 0:
|
||
return 0
|
||
return math.ceil(total / page_size)
|
||
|
||
|
||
def _decimal_value(value: Any, default: Decimal = Decimal("0")) -> Decimal:
|
||
if value is None or value == "":
|
||
return default
|
||
if isinstance(value, Decimal):
|
||
return value
|
||
return Decimal(str(value))
|
||
|
||
|
||
def _parse_optional_date(value: Any) -> Optional[date]:
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, date) and not isinstance(value, datetime):
|
||
return value
|
||
if isinstance(value, datetime):
|
||
return value.date()
|
||
text = str(value).strip()
|
||
if not text:
|
||
return None
|
||
for fmt in (_DATE_FMT, _FMT):
|
||
try:
|
||
return datetime.strptime(text, fmt).date()
|
||
except ValueError:
|
||
continue
|
||
raise ValueError(f"日期格式无效: {text}")
|
||
|
||
|
||
def _parse_optional_datetime(value: Any) -> Optional[datetime]:
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, datetime):
|
||
return value
|
||
text = str(value).strip()
|
||
if not text:
|
||
return None
|
||
for fmt in (_FMT, _DATE_FMT):
|
||
try:
|
||
return datetime.strptime(text, fmt)
|
||
except ValueError:
|
||
continue
|
||
raise ValueError(f"日期时间格式无效: {text}")
|
||
|
||
|
||
def _get_field(data: dict, *keys: str, default=None):
|
||
for key in keys:
|
||
if key in data and data[key] is not None:
|
||
return data[key]
|
||
return default
|
||
|
||
|
||
def _resolve_user_display_name(db: Session, user_id: Optional[str]) -> str:
|
||
uid = (user_id or "").strip()
|
||
if not uid:
|
||
return ""
|
||
u = db.query(SysUser).filter(SysUser.id == uid, SysUser.del_flag == 0).first()
|
||
if not u:
|
||
return ""
|
||
return (u.nick_name or u.user_no or "").strip()
|
||
|
||
|
||
def _is_blank(value: Any) -> bool:
|
||
return value is None or str(value).strip() == ""
|
||
|
||
|
||
def _first_non_blank(*values: Any) -> str:
|
||
for value in values:
|
||
text = "" if value is None else str(value).strip()
|
||
if text:
|
||
return text
|
||
return ""
|
||
|
||
|
||
def _get_default_invoice_info(db: Session, *, tenant_no: str, customer_id: str) -> Optional[dict]:
|
||
rows = list_invoice_info_by_customer(db, tenant_no, customer_id)
|
||
if not rows:
|
||
return None
|
||
for row in rows:
|
||
if str(row.get("isDefault") or "").strip() == "1":
|
||
return row
|
||
return rows[0]
|
||
|
||
|
||
def _build_customer_address(customer: Any) -> str:
|
||
if not customer:
|
||
return ""
|
||
address_parts = [
|
||
_first_non_blank(getattr(customer, "province", "")),
|
||
_first_non_blank(getattr(customer, "city", "")),
|
||
_first_non_blank(getattr(customer, "district", "")),
|
||
_first_non_blank(getattr(customer, "address", "")),
|
||
]
|
||
return "".join([part for part in address_parts if part])
|
||
|
||
|
||
def _row_to_dict(row: CrmContract) -> Dict[str, Any]:
|
||
"""ORM → camelCase（单点真相）"""
|
||
return {
|
||
"id": row.id,
|
||
"contractNo": row.contract_no or "",
|
||
"contractName": row.contract_name or "",
|
||
"customerId": row.customer_id or "",
|
||
"customerName": row.customer_name or "",
|
||
"opportunityId": row.opportunity_id or "",
|
||
"quoteId": row.quote_id or "",
|
||
"parentContractId": row.parent_contract_id or "",
|
||
"renewalCount": row.renewal_count if row.renewal_count is not None else 0,
|
||
"templateId": row.template_id or "",
|
||
"contractTemplate": row.contract_template or "",
|
||
"contactId": row.contact_id or "",
|
||
"contactName": row.contact_name or "",
|
||
"contactMobile": row.contact_mobile or "",
|
||
"contractType": row.contract_type or "",
|
||
"contractNature": row.contract_nature or "",
|
||
"amount": float(row.amount) if row.amount is not None else 0.0,
|
||
"currency": row.currency or "CNY",
|
||
"taxRate": float(row.tax_rate) if row.tax_rate is not None else 0.0,
|
||
"amountWithTax": float(row.amount_with_tax) if row.amount_with_tax is not None else 0.0,
|
||
"freight": float(row.freight) if getattr(row, "freight", None) is not None else 0.0,
|
||
"barrelDeposit": float(getattr(row, "barrel_deposit", None) or 0),
|
||
"status": row.status or "draft",
|
||
"approvalStatus": row.approval_status or "",
|
||
"signDate": row.sign_date.strftime(_DATE_FMT) if row.sign_date else "",
|
||
"actualSignDate": row.actual_sign_date.strftime(_DATE_FMT) if row.actual_sign_date else "",
|
||
"startDate": row.start_date.strftime(_DATE_FMT) if row.start_date else "",
|
||
"endDate": row.end_date.strftime(_DATE_FMT) if row.end_date else "",
|
||
"contractPeriod": row.contract_period,
|
||
"performanceStatus": row.performance_status or "",
|
||
"performanceRate": float(row.performance_rate) if row.performance_rate is not None else 0.0,
|
||
"customerSatisfaction": row.customer_satisfaction,
|
||
"approvalFlowId": row.approval_flow_id or "",
|
||
"contractFile": row.contract_file or "",
|
||
"attachmentFiles": row.attachment_files or "",
|
||
"ownerUserId": row.owner_user_id or "",
|
||
"ownerUserName": row.owner_user_name or "",
|
||
"receivedAmount": float(row.received_amount) if row.received_amount is not None else 0.0,
|
||
"unreceivedAmount": float(row.unreceived_amount) if row.unreceived_amount is not None else 0.0,
|
||
"invoiceAmount": float(row.invoice_amount) if row.invoice_amount is not None else 0.0,
|
||
"uninvoiceAmount": float(row.uninvoice_amount) if row.uninvoice_amount is not None else 0.0,
|
||
"partyA": row.party_a or "",
|
||
"partyAAddress": row.party_a_address or "",
|
||
"partyAContact": row.party_a_contact or "",
|
||
"partyAMobile": row.party_a_mobile or "",
|
||
"partyALegalPerson": row.party_a_legal_person or "",
|
||
"partyAAgent": row.party_a_agent or "",
|
||
"partyAPhoneFax": row.party_a_phone_fax or "",
|
||
"partyABank": row.party_a_bank or "",
|
||
"partyAAccount": row.party_a_account or "",
|
||
"supplierId": row.supplier_id or "",
|
||
"settlementMethod": row.settlement_method or "",
|
||
"partyB": row.party_b or "",
|
||
"partyBAddress": row.party_b_address or "",
|
||
"partyBContact": row.party_b_contact or "",
|
||
"partyBMobile": row.party_b_mobile or "",
|
||
"partyBLegalPerson": row.party_b_legal_person or "",
|
||
"partyBAgent": row.party_b_agent or "",
|
||
"partyBPhoneFax": row.party_b_phone_fax or "",
|
||
"partyBBank": row.party_b_bank or "",
|
||
"partyBAccount": row.party_b_account or "",
|
||
"paymentTerms": row.payment_terms or "",
|
||
"deliveryTerms": row.delivery_terms or "",
|
||
"contractTerms": row.contract_terms or "",
|
||
"invalidReason": row.invalid_reason or "",
|
||
"invalidTime": row.invalid_time.strftime(_FMT) if row.invalid_time else "",
|
||
"templateContent": row.template_content or "",
|
||
"settlementClauseDrafts": row.settlement_clause_drafts or "",
|
||
"createBy": row.create_by or "",
|
||
"createTime": row.create_time.strftime(_FMT) if row.create_time else "",
|
||
"updateBy": row.update_by or "",
|
||
"updateTime": row.update_time.strftime(_FMT) if row.update_time else "",
|
||
"remark": row.remark or "",
|
||
}
|
||
|
||
|
||
def contract_row_to_dict(row: CrmContract) -> Dict[str, Any]:
|
||
return _row_to_dict(row)
|
||
|
||
|
||
def _build_contract_filters(
|
||
db: Session,
|
||
q: Query,
|
||
filters: dict,
|
||
*,
|
||
tenant_no: str,
|
||
) -> Query:
|
||
"""Java: buildQueryWrapper"""
|
||
from app.services.crm.contract_list_filters import build_contract_column_filters
|
||
|
||
q = q.filter(CrmContract.tenant_no == tenant_no, CrmContract.del_flag == 0)
|
||
|
||
keyword = (_get_field(filters, "keyword") or "").strip()
|
||
if keyword:
|
||
like = f"%{keyword}%"
|
||
pinyin_like = f"%{keyword.lower()}%"
|
||
product_match = (
|
||
db.query(CrmContractProduct.id)
|
||
.outerjoin(
|
||
CrmProduct,
|
||
and_(
|
||
CrmProduct.id == CrmContractProduct.product_id,
|
||
CrmProduct.del_flag == 0,
|
||
CrmProduct.tenant_no == tenant_no,
|
||
),
|
||
)
|
||
.filter(
|
||
CrmContractProduct.contract_id == CrmContract.id,
|
||
CrmContractProduct.tenant_no == tenant_no,
|
||
CrmContractProduct.del_flag == 0,
|
||
or_(
|
||
CrmContractProduct.product_name.like(like),
|
||
func.lower(func.coalesce(CrmProduct.product_name, "")).like(
|
||
f"%{keyword.lower()}%"
|
||
),
|
||
func.lower(func.coalesce(CrmProduct.name_pinyin_abbr, "")).like(
|
||
pinyin_like
|
||
),
|
||
),
|
||
)
|
||
.exists()
|
||
)
|
||
q = q.filter(
|
||
or_(
|
||
CrmContract.contract_no.like(like),
|
||
product_match,
|
||
)
|
||
)
|
||
|
||
contract_no = (_get_field(filters, "contract_no", "contractNo") or "").strip()
|
||
contract_name = (_get_field(filters, "contract_name", "contractName") or "").strip()
|
||
customer_name = (_get_field(filters, "customer_name", "customerName") or "").strip()
|
||
legacy_combined = False
|
||
if (
|
||
contract_no
|
||
and contract_name
|
||
and customer_name
|
||
and contract_no == contract_name == customer_name
|
||
):
|
||
legacy_keyword = contract_no
|
||
q = q.filter(
|
||
or_(
|
||
CrmContract.contract_no.like(f"%{legacy_keyword}%"),
|
||
CrmContract.contract_name.like(f"%{legacy_keyword}%"),
|
||
CrmContract.customer_name.like(f"%{legacy_keyword}%"),
|
||
)
|
||
)
|
||
legacy_combined = True
|
||
|
||
customer_id = (_get_field(filters, "customer_id", "customerId") or "").strip()
|
||
if customer_id:
|
||
q = q.filter(CrmContract.customer_id == customer_id)
|
||
|
||
opportunity_id = (_get_field(filters, "opportunity_id", "opportunityId") or "").strip()
|
||
if opportunity_id:
|
||
q = q.filter(CrmContract.opportunity_id == opportunity_id)
|
||
|
||
quote_id = (_get_field(filters, "quote_id", "quoteId") or "").strip()
|
||
if quote_id:
|
||
q = q.filter(CrmContract.quote_id == quote_id)
|
||
|
||
exclude_status = (_get_field(filters, "exclude_status", "excludeStatus") or "").strip()
|
||
if exclude_status:
|
||
q = q.filter(CrmContract.status != exclude_status)
|
||
|
||
# 列头漏斗 / 旧单值字段
|
||
col_params = {
|
||
"contract_no": None if (keyword or legacy_combined) else _get_field(filters, "contract_no", "contractNo"),
|
||
"contract_name": None if legacy_combined else _get_field(filters, "contract_name", "contractName"),
|
||
"customer_name": None if legacy_combined else _get_field(filters, "customer_name", "customerName"),
|
||
"party_a": _get_field(filters, "party_a", "partyA"),
|
||
"party_b": _get_field(filters, "party_b", "partyB"),
|
||
"settlement_method": _get_field(filters, "settlement_method", "settlementMethod"),
|
||
"owner_user_name": _get_field(filters, "owner_user_name", "ownerUserName"),
|
||
"remark": _get_field(filters, "remark"),
|
||
"contact_name": _get_field(filters, "contact_name", "contactName"),
|
||
"contact_mobile": _get_field(filters, "contact_mobile", "contactMobile"),
|
||
"payment_method": _get_field(filters, "payment_method", "paymentMethod"),
|
||
"payment_terms": _get_field(filters, "payment_terms", "paymentTerms"),
|
||
"status": _get_field(filters, "status"),
|
||
"status_in": _get_field(filters, "status_in", "statusIn"),
|
||
"contract_type": _get_field(filters, "contract_type", "contractType"),
|
||
"contract_type_in": _get_field(filters, "contract_type_in", "contractTypeIn"),
|
||
"approval_status": _get_field(filters, "approval_status", "approvalStatus"),
|
||
"approval_status_in": _get_field(filters, "approval_status_in", "approvalStatusIn"),
|
||
"owner_user_id": _get_field(filters, "owner_user_id", "ownerUserId"),
|
||
"begin_time": _get_field(filters, "begin_time", "beginTime"),
|
||
"end_time": _get_field(filters, "end_time", "endTime"),
|
||
"sign_date_begin": _get_field(filters, "sign_date_begin", "signDateBegin"),
|
||
"sign_date_end": _get_field(filters, "sign_date_end", "signDateEnd"),
|
||
"sign_date_from": _get_field(filters, "sign_date_from", "signDateFrom"),
|
||
"sign_date_to": _get_field(filters, "sign_date_to", "signDateTo"),
|
||
"start_date_from": _get_field(filters, "start_date_from", "startDateFrom"),
|
||
"start_date_to": _get_field(filters, "start_date_to", "startDateTo"),
|
||
"end_date_from": _get_field(filters, "end_date_from", "endDateFrom"),
|
||
"end_date_to": _get_field(filters, "end_date_to", "endDateTo"),
|
||
"update_time_from": _get_field(filters, "update_time_from", "updateTimeFrom"),
|
||
"update_time_to": _get_field(filters, "update_time_to", "updateTimeTo"),
|
||
"amount_op": _get_field(filters, "amount_op", "amountOp"),
|
||
"amount": _get_field(filters, "amount"),
|
||
"amount_min": _get_field(filters, "amount_min", "amountMin"),
|
||
"amount_max": _get_field(filters, "amount_max", "amountMax"),
|
||
"freight_op": _get_field(filters, "freight_op", "freightOp"),
|
||
"freight_min": _get_field(filters, "freight_min", "freightMin"),
|
||
"freight_max": _get_field(filters, "freight_max", "freightMax"),
|
||
"barrel_deposit_op": _get_field(filters, "barrel_deposit_op", "barrelDepositOp"),
|
||
"barrel_deposit_min": _get_field(filters, "barrel_deposit_min", "barrelDepositMin"),
|
||
"barrel_deposit_max": _get_field(filters, "barrel_deposit_max", "barrelDepositMax"),
|
||
"unreceived_amount_op": _get_field(filters, "unreceived_amount_op", "unreceivedAmountOp"),
|
||
"unreceived_amount_min": _get_field(filters, "unreceived_amount_min", "unreceivedAmountMin"),
|
||
"unreceived_amount_max": _get_field(filters, "unreceived_amount_max", "unreceivedAmountMax"),
|
||
"invoice_amount_op": _get_field(filters, "invoice_amount_op", "invoiceAmountOp"),
|
||
"invoice_amount_min": _get_field(filters, "invoice_amount_min", "invoiceAmountMin"),
|
||
"invoice_amount_max": _get_field(filters, "invoice_amount_max", "invoiceAmountMax"),
|
||
"uninvoice_amount_op": _get_field(filters, "uninvoice_amount_op", "uninvoiceAmountOp"),
|
||
"uninvoice_amount_min": _get_field(filters, "uninvoice_amount_min", "uninvoiceAmountMin"),
|
||
"uninvoice_amount_max": _get_field(filters, "uninvoice_amount_max", "uninvoiceAmountMax"),
|
||
"renewal_count_op": _get_field(filters, "renewal_count_op", "renewalCountOp"),
|
||
"renewal_count": _get_field(filters, "renewal_count", "renewalCount"),
|
||
"renewal_count_min": _get_field(filters, "renewal_count_min", "renewalCountMin"),
|
||
"renewal_count_max": _get_field(filters, "renewal_count_max", "renewalCountMax"),
|
||
"min_amount": _get_field(filters, "min_amount", "minAmount"),
|
||
"max_amount": _get_field(filters, "max_amount", "maxAmount"),
|
||
}
|
||
for expr in build_contract_column_filters(**col_params):
|
||
q = q.filter(expr)
|
||
|
||
pending_approval_user_id = (
|
||
_get_field(filters, "pending_approval_user_id", "pendingApprovalUserId") or ""
|
||
).strip()
|
||
if pending_approval_user_id:
|
||
# 仅返回当前处于审批中且审批状态为待审批的合同，避免历史审批记录混入草稿/已审批数据
|
||
q = q.filter(
|
||
CrmContract.status == "approving",
|
||
CrmContract.approval_status == "pending",
|
||
)
|
||
q = q.filter(
|
||
CrmContract.id.in_(
|
||
db.query(CrmContractApproval.contract_id)
|
||
.filter(
|
||
CrmContractApproval.approver_id == pending_approval_user_id,
|
||
CrmContractApproval.approval_result == "pending",
|
||
CrmContractApproval.is_current == 1,
|
||
CrmContractApproval.del_flag == 0,
|
||
)
|
||
.distinct()
|
||
)
|
||
)
|
||
|
||
approved_by_user_id = (
|
||
_get_field(filters, "approved_by_user_id", "approvedByUserId") or ""
|
||
).strip()
|
||
if approved_by_user_id:
|
||
# 已审批列表应只看审批流已结束的合同，排除仍在审批中的记录
|
||
q = q.filter(CrmContract.approval_status.in_(("approved", "rejected")))
|
||
contract_ids = [
|
||
cid
|
||
for (cid,) in db.query(CrmContractApproval.contract_id)
|
||
.filter(
|
||
CrmContractApproval.approver_id == approved_by_user_id,
|
||
CrmContractApproval.approval_result.in_(("approved", "rejected")),
|
||
CrmContractApproval.del_flag == 0,
|
||
)
|
||
.distinct()
|
||
.all()
|
||
if cid
|
||
]
|
||
if contract_ids:
|
||
q = q.filter(CrmContract.id.in_(contract_ids))
|
||
else:
|
||
q = q.filter(CrmContract.id == "__none__")
|
||
|
||
default_order = CrmContract.create_time.desc()
|
||
whitelist = {
|
||
"signDate": CrmContract.sign_date,
|
||
"startDate": CrmContract.start_date,
|
||
"endDate": CrmContract.end_date,
|
||
"createTime": CrmContract.create_time,
|
||
"updateTime": CrmContract.update_time,
|
||
}
|
||
sort_field = _get_field(filters, "sort_field", "sortField")
|
||
sort_order = _get_field(filters, "sort_order", "sortOrder")
|
||
return q.order_by(
|
||
resolve_datetime_order_by(sort_field, sort_order, whitelist, default_order)
|
||
)
|
||
|
||
|
||
def list_contracts(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
filters: dict = None,
|
||
current_user_id: str = None,
|
||
user_info=None,
|
||
include_products: bool = False,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
"""Java: queryPageList"""
|
||
q = db.query(CrmContract)
|
||
q = _build_contract_filters(db, q, filters or {}, tenant_no=tenant_no)
|
||
if user_info:
|
||
q = apply_crm_owner_scope(q, CrmContract, user_info, db, tenant_no)
|
||
total = q.order_by(None).with_entities(func.count(CrmContract.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 500)
|
||
rows = q.offset((page - 1) * page_size).limit(page_size).all()
|
||
result = [_row_to_dict(r) for r in rows]
|
||
|
||
for item in result:
|
||
if item.get("status") == "approving" and current_user_id:
|
||
item["canApprove"] = can_user_approve(db, item["id"], current_user_id)
|
||
else:
|
||
item["canApprove"] = False
|
||
if item.get("ownerUserId") and not item.get("ownerUserName"):
|
||
item["ownerUserName"] = _resolve_user_display_name(db, item["ownerUserId"])
|
||
if item.get("createBy") and not item.get("createByName"):
|
||
item["createByName"] = _resolve_user_display_name(db, item["createBy"])
|
||
|
||
if include_products and result:
|
||
contract_ids = [str(item["id"]) for item in result if item.get("id")]
|
||
products_map = list_products_by_contract_ids(db, contract_ids, tenant_no=tenant_no)
|
||
for item in result:
|
||
item["products"] = products_map.get(str(item.get("id") or ""), [])
|
||
|
||
return result, int(total)
|
||
|
||
|
||
def get_contract(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
current_user_id: str = None,
|
||
user_info=None,
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""Java: queryById"""
|
||
row = (
|
||
db.query(CrmContract)
|
||
.filter(
|
||
CrmContract.id == contract_id,
|
||
CrmContract.tenant_no == tenant_no,
|
||
CrmContract.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not row:
|
||
return None
|
||
_ensure_contract_in_scope(db, row, user_info, tenant_no)
|
||
|
||
detail = _row_to_dict(row)
|
||
detail["createByName"] = _resolve_user_display_name(db, detail.get("createBy"))
|
||
detail["updateByName"] = _resolve_user_display_name(db, detail.get("updateBy"))
|
||
if detail.get("ownerUserId") and not detail.get("ownerUserName"):
|
||
detail["ownerUserName"] = _resolve_user_display_name(db, detail["ownerUserId"])
|
||
|
||
supplier_id = (detail.get("supplierId") or "").strip()
|
||
supplier = None
|
||
if supplier_id:
|
||
supplier = get_supplier_info(db, supplier_id, tenant_no=tenant_no)
|
||
if supplier:
|
||
detail["supplierName"] = supplier.supplier_name or ""
|
||
|
||
customer = None
|
||
customer_id = (detail.get("customerId") or "").strip()
|
||
if customer_id:
|
||
customer = get_customer(db, customer_id)
|
||
|
||
# 查询兜底：历史空数据按基础资料补展示，不改变原始库存量
|
||
if supplier:
|
||
if _is_blank(detail.get("partyA")):
|
||
detail["partyA"] = _first_non_blank(getattr(supplier, "supplier_name", ""))
|
||
if _is_blank(detail.get("partyAAddress")):
|
||
detail["partyAAddress"] = _first_non_blank(getattr(supplier, "supplier_address", ""))
|
||
if _is_blank(detail.get("partyALegalPerson")):
|
||
detail["partyALegalPerson"] = _first_non_blank(getattr(supplier, "legal_person", ""))
|
||
if _is_blank(detail.get("partyAAgent")):
|
||
detail["partyAAgent"] = _first_non_blank(getattr(supplier, "agent", ""))
|
||
if _is_blank(detail.get("partyAPhoneFax")):
|
||
detail["partyAPhoneFax"] = _first_non_blank(getattr(supplier, "phone_fax", ""))
|
||
if _is_blank(detail.get("partyABank")):
|
||
detail["partyABank"] = _first_non_blank(getattr(supplier, "bank_name", ""))
|
||
if _is_blank(detail.get("partyAAccount")):
|
||
detail["partyAAccount"] = _first_non_blank(getattr(supplier, "bank_account", ""))
|
||
|
||
if customer:
|
||
inv = _get_default_invoice_info(db, tenant_no=tenant_no, customer_id=customer_id)
|
||
if _is_blank(detail.get("partyB")):
|
||
detail["partyB"] = _first_non_blank(
|
||
inv.get("companyName") if inv else "",
|
||
customer.customer_name,
|
||
)
|
||
if _is_blank(detail.get("partyBAddress")):
|
||
detail["partyBAddress"] = _first_non_blank(
|
||
inv.get("companyAddress") if inv else "",
|
||
_build_customer_address(customer),
|
||
)
|
||
if _is_blank(detail.get("partyBContact")):
|
||
detail["partyBContact"] = _first_non_blank(detail.get("contactName"))
|
||
if _is_blank(detail.get("partyBMobile")):
|
||
detail["partyBMobile"] = _first_non_blank(
|
||
inv.get("phoneFax") if inv else "",
|
||
detail.get("contactMobile"),
|
||
customer.mobile,
|
||
customer.telephone,
|
||
)
|
||
if _is_blank(detail.get("partyBLegalPerson")):
|
||
detail["partyBLegalPerson"] = _first_non_blank(inv.get("legalRepresentative") if inv else "")
|
||
if _is_blank(detail.get("partyBPhoneFax")):
|
||
detail["partyBPhoneFax"] = _first_non_blank(
|
||
inv.get("phoneFax") if inv else "",
|
||
customer.telephone,
|
||
)
|
||
if _is_blank(detail.get("partyBBank")):
|
||
detail["partyBBank"] = _first_non_blank(inv.get("bankName") if inv else "")
|
||
if _is_blank(detail.get("partyBAccount")):
|
||
detail["partyBAccount"] = _first_non_blank(inv.get("bankAccount") if inv else "")
|
||
|
||
detail["products"] = list_contract_products(
|
||
db,
|
||
contract_id,
|
||
tenant_no=tenant_no,
|
||
include_price_tiers=True,
|
||
)
|
||
from app.services.pl_contract_plan_qty import (
|
||
aggregate_contract_product_plan_qtys,
|
||
attach_plan_qty_fields_to_products,
|
||
)
|
||
|
||
qty_map = aggregate_contract_product_plan_qtys(
|
||
db, tenant_no=tenant_no, contract_id=contract_id
|
||
)
|
||
attach_plan_qty_fields_to_products(detail["products"], qty_map)
|
||
|
||
if current_user_id and detail.get("status") == "approving":
|
||
detail["canApprove"] = can_user_approve(db, contract_id, current_user_id)
|
||
else:
|
||
detail["canApprove"] = False
|
||
|
||
return detail
|
||
|
||
|
||
def get_contract_detail(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
current_user_id: str = None,
|
||
user_info=None,
|
||
) -> Dict[str, Any]:
|
||
detail = get_contract(
|
||
db,
|
||
contract_id,
|
||
tenant_no=tenant_no,
|
||
current_user_id=current_user_id,
|
||
user_info=user_info,
|
||
)
|
||
if not detail:
|
||
raise ValueError("合同不存在")
|
||
return detail
|
||
|
||
|
||
def get_contract_row(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
) -> Optional[CrmContract]:
|
||
return (
|
||
db.query(CrmContract)
|
||
.filter(
|
||
CrmContract.id == contract_id,
|
||
CrmContract.tenant_no == tenant_no,
|
||
CrmContract.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def _get_supplier_short_name(db: Session, supplier_id: str, tenant_no: str) -> str:
|
||
if not supplier_id:
|
||
return "山峡"
|
||
supplier = get_supplier_info(db, supplier_id, tenant_no=tenant_no)
|
||
if not supplier:
|
||
return "山峡"
|
||
if (supplier.short_name or "").strip():
|
||
return supplier.short_name.strip()
|
||
if (supplier.supplier_name or "").strip():
|
||
return supplier.supplier_name.strip()
|
||
return "山峡"
|
||
|
||
|
||
def _build_contract_name_from_parties(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
contract_no: str,
|
||
party_a: str = "",
|
||
party_b: str = "",
|
||
supplier_id: str = "",
|
||
customer_id: str = "",
|
||
customer_name: str = "",
|
||
customer_short_name: str = "",
|
||
user_contract_name: str = "",
|
||
) -> str:
|
||
"""根据供需方面板与主数据生成/校正合同名称。"""
|
||
supplier = get_supplier_info(db, supplier_id, tenant_no=tenant_no) if supplier_id else None
|
||
supplier_name = (getattr(supplier, "supplier_name", None) or "").strip() if supplier else ""
|
||
supplier_short_name = _get_supplier_short_name(db, supplier_id or "", tenant_no)
|
||
|
||
cust_name = (customer_name or "").strip()
|
||
cust_short = (customer_short_name or "").strip()
|
||
invoice_name = ""
|
||
if customer_id:
|
||
# 始终从客户主数据取简称，避免前端未带 shortName 时落成全称
|
||
customer = get_customer(db, customer_id)
|
||
if customer:
|
||
cust_name = cust_name or (customer.customer_name or "").strip()
|
||
db_short = (customer.short_name or "").strip()
|
||
if db_short:
|
||
cust_short = db_short
|
||
inv = _get_default_invoice_info(db, tenant_no=tenant_no, customer_id=customer_id)
|
||
if inv:
|
||
invoice_name = (inv.get("companyName") or "").strip()
|
||
|
||
return generate_contract_name(
|
||
contract_no or "",
|
||
party_a=party_a or "",
|
||
party_b=party_b or "",
|
||
supplier_name=supplier_name,
|
||
supplier_short_name=supplier_short_name,
|
||
customer_name=cust_name,
|
||
customer_short_name=cust_short,
|
||
customer_invoice_name=invoice_name,
|
||
user_contract_name=user_contract_name or "",
|
||
)
|
||
|
||
|
||
def generate_contract_no(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
supplier_id: str = None,
|
||
sign_date: date = None,
|
||
) -> str:
|
||
"""Java: generateContractNo — {供方编码}/X-JT{yyyyMMdd}{3位序号}
|
||
|
||
流水与撞号检测须含软删：uk_contract_tenant_no 对 (contract_no, tenant_no) 全局唯一。
|
||
"""
|
||
supplier_code = "SX"
|
||
if supplier_id:
|
||
supplier = get_supplier_info(db, supplier_id, tenant_no=tenant_no)
|
||
if supplier and (supplier.supplier_code or "").strip():
|
||
supplier_code = supplier.supplier_code.strip()
|
||
|
||
date_to_use = sign_date or date.today()
|
||
date_str = date_to_use.strftime("%Y%m%d")
|
||
prefix = f"{supplier_code}/X-JT{date_str}"
|
||
|
||
# 含软删：已软删编号仍占唯一索引，不能复用
|
||
last = (
|
||
db.query(CrmContract)
|
||
.filter(
|
||
CrmContract.tenant_no == tenant_no,
|
||
CrmContract.contract_no.like(f"{prefix}%"),
|
||
)
|
||
.order_by(CrmContract.contract_no.desc())
|
||
.first()
|
||
)
|
||
|
||
sequence = 1
|
||
if last and last.contract_no and len(last.contract_no) >= len(prefix) + 3:
|
||
try:
|
||
sequence = int(last.contract_no[len(prefix) :]) + 1
|
||
except ValueError:
|
||
sequence = 1
|
||
|
||
for retry in range(50):
|
||
candidate = f"{prefix}{sequence + retry:03d}"
|
||
if not contract_no_exists(db, tenant_no, candidate):
|
||
return candidate
|
||
|
||
raise ValueError("生成合同编号失败，请稍后重试")
|
||
|
||
|
||
def contract_no_exists(
|
||
db: Session,
|
||
tenant_no: str,
|
||
contract_no: str,
|
||
*,
|
||
exclude_id: str = None,
|
||
) -> bool:
|
||
"""租户内合同编号是否已占用（含软删，保证全局唯一）。"""
|
||
q = db.query(CrmContract.id).filter(
|
||
CrmContract.tenant_no == tenant_no,
|
||
CrmContract.contract_no == contract_no,
|
||
)
|
||
if exclude_id:
|
||
q = q.filter(CrmContract.id != exclude_id)
|
||
return q.first() is not None
|
||
|
||
|
||
_CONTRACT_NO_STD_RE = re.compile(r"^([A-Z]+)/X-JT(\d{8})(\d{3})$")
|
||
|
||
|
||
def replace_contract_no_date(contract_no: str, sign_date: date) -> Optional[str]:
|
||
"""标准编号替换日期段，保留序号；非标准返回 None。"""
|
||
if not contract_no or not sign_date:
|
||
return None
|
||
m = _CONTRACT_NO_STD_RE.match(contract_no.strip())
|
||
if not m:
|
||
return None
|
||
supplier_code, _old_date, seq = m.group(1), m.group(2), m.group(3)
|
||
date_str = sign_date.strftime("%Y%m%d")
|
||
return f"{supplier_code}/X-JT{date_str}{seq}"
|
||
|
||
|
||
def align_contract_no_to_sign_date(
|
||
db: Session,
|
||
tenant_no: str,
|
||
contract_no: str,
|
||
sign_date: date,
|
||
*,
|
||
exclude_id: str = None,
|
||
) -> str:
|
||
"""将标准合同编号日期段对齐签订日；撞号则序号顺延（含软删查重）。
|
||
|
||
非标准编号原样返回。无签订日则原样返回。
|
||
"""
|
||
if not (contract_no or "").strip():
|
||
return contract_no or ""
|
||
if not sign_date:
|
||
return contract_no.strip()
|
||
|
||
replaced = replace_contract_no_date(contract_no, sign_date)
|
||
if replaced is None:
|
||
return contract_no.strip()
|
||
|
||
m = _CONTRACT_NO_STD_RE.match(replaced)
|
||
if not m:
|
||
return replaced
|
||
supplier_code, date_str, seq = m.group(1), m.group(2), m.group(3)
|
||
start_seq = int(seq)
|
||
for retry in range(50):
|
||
candidate = f"{supplier_code}/X-JT{date_str}{start_seq + retry:03d}"
|
||
if not contract_no_exists(db, tenant_no, candidate, exclude_id=exclude_id):
|
||
return candidate
|
||
raise ValueError("生成合同编号失败，请稍后重试")
|
||
|
||
|
||
def resolve_party_short_name(
|
||
party_name: str,
|
||
*,
|
||
supplier_name: str = "",
|
||
supplier_short_name: str = "",
|
||
customer_name: str = "",
|
||
customer_short_name: str = "",
|
||
customer_invoice_name: str = "",
|
||
) -> str:
|
||
"""按面板单位全称精确匹配交易主体/客户主数据，取 shortName；对不上则用全称。"""
|
||
party = (party_name or "").strip()
|
||
if not party:
|
||
return ""
|
||
|
||
s_name = (supplier_name or "").strip()
|
||
s_short = (supplier_short_name or "").strip()
|
||
if s_name and party == s_name:
|
||
return s_short or s_name
|
||
if s_short and party == s_short:
|
||
return s_short
|
||
|
||
c_name = (customer_name or "").strip()
|
||
c_short = (customer_short_name or "").strip()
|
||
c_inv = (customer_invoice_name or "").strip()
|
||
if c_name and party == c_name:
|
||
return c_short or c_name
|
||
if c_short and party == c_short:
|
||
return c_short
|
||
if c_inv and party == c_inv:
|
||
return c_short or c_inv
|
||
|
||
return party
|
||
|
||
|
||
def _strip_contract_name_no_suffix(name: str) -> str:
|
||
return re.sub(r"（[^）]+）$", "", (name or "").strip()).strip()
|
||
|
||
|
||
def _auto_contract_name_body(
|
||
party_a: str = "",
|
||
party_b: str = "",
|
||
*,
|
||
supplier_name: str = "",
|
||
supplier_short_name: str = "",
|
||
customer_name: str = "",
|
||
customer_short_name: str = "",
|
||
customer_invoice_name: str = "",
|
||
) -> str:
|
||
a = resolve_party_short_name(
|
||
party_a,
|
||
supplier_name=supplier_name,
|
||
supplier_short_name=supplier_short_name,
|
||
customer_name=customer_name,
|
||
customer_short_name=customer_short_name,
|
||
customer_invoice_name=customer_invoice_name,
|
||
)
|
||
b = resolve_party_short_name(
|
||
party_b,
|
||
supplier_name=supplier_name,
|
||
supplier_short_name=supplier_short_name,
|
||
customer_name=customer_name,
|
||
customer_short_name=customer_short_name,
|
||
customer_invoice_name=customer_invoice_name,
|
||
)
|
||
return "-".join(x for x in (a, b) if x)
|
||
|
||
|
||
def _user_contract_name_uses_party_full_name(
|
||
cleaned_user_name: str,
|
||
party_a: str = "",
|
||
party_b: str = "",
|
||
*,
|
||
supplier_name: str = "",
|
||
customer_name: str = "",
|
||
customer_invoice_name: str = "",
|
||
) -> bool:
|
||
"""前端预览误用单位全称时，名称片段会等于面板/客户/开票全称。"""
|
||
parts = [p.strip() for p in (cleaned_user_name or "").split("-") if p.strip()]
|
||
if not parts:
|
||
return False
|
||
full_names = {
|
||
(party_a or "").strip(),
|
||
(party_b or "").strip(),
|
||
(supplier_name or "").strip(),
|
||
(customer_name or "").strip(),
|
||
(customer_invoice_name or "").strip(),
|
||
}
|
||
full_names.discard("")
|
||
return any(part in full_names for part in parts)
|
||
|
||
|
||
def generate_contract_name(
|
||
contract_no: str,
|
||
party_a: str = "",
|
||
party_b: str = "",
|
||
*,
|
||
supplier_name: str = "",
|
||
supplier_short_name: str = "",
|
||
customer_name: str = "",
|
||
customer_short_name: str = "",
|
||
customer_invoice_name: str = "",
|
||
user_contract_name: str = "",
|
||
contract_type: str = "", # 保留兼容旧调用，命名不再按类型反序
|
||
) -> str:
|
||
"""按面板供方(partyA)-需方(partyB)生成：{供方简称}-{需方简称}（{合同编号}）。
|
||
|
||
- 名称为空：按主数据简称自动生成
|
||
- 名称含单位全称（前端误预览）：仍按简称重算
|
||
- 续签 / 真正手填自定义名：保留主体，仅校正末尾括号内合同编号
|
||
"""
|
||
auto_body = _auto_contract_name_body(
|
||
party_a,
|
||
party_b,
|
||
supplier_name=supplier_name,
|
||
supplier_short_name=supplier_short_name,
|
||
customer_name=customer_name,
|
||
customer_short_name=customer_short_name,
|
||
customer_invoice_name=customer_invoice_name,
|
||
)
|
||
|
||
def _with_no(body: str) -> str:
|
||
if not body:
|
||
return f"合同（{contract_no}）" if contract_no else "合同"
|
||
return f"{body}（{contract_no}）" if contract_no else body
|
||
|
||
user_raw = (user_contract_name or "").strip()
|
||
if not user_raw:
|
||
return _with_no(auto_body)
|
||
|
||
# 续签名称：保留整段（含原编号括号），仅必要时追加当前编号
|
||
if "-续签" in user_raw:
|
||
cleaned = _strip_contract_name_no_suffix(user_raw)
|
||
# 若原文已以当前编号结尾则保持；否则在整段后续签名后挂编号
|
||
if contract_no and user_raw.endswith(f"（{contract_no}）"):
|
||
return user_raw
|
||
if contract_no and f"（{contract_no}）" not in cleaned:
|
||
# 续签名通常为「原名-续签N」，原名里已有旧编号括号，不能 strip 掉
|
||
return f"{user_raw}（{contract_no}）" if not user_raw.endswith(f"（{contract_no}）") else user_raw
|
||
return user_raw
|
||
|
||
cleaned = _strip_contract_name_no_suffix(user_raw)
|
||
if (
|
||
cleaned == auto_body
|
||
or _user_contract_name_uses_party_full_name(
|
||
cleaned,
|
||
party_a,
|
||
party_b,
|
||
supplier_name=supplier_name,
|
||
customer_name=customer_name,
|
||
customer_invoice_name=customer_invoice_name,
|
||
)
|
||
):
|
||
return _with_no(auto_body)
|
||
|
||
if contract_no and f"（{contract_no}）" not in cleaned:
|
||
return f"{cleaned}（{contract_no}）"
|
||
return cleaned
|
||
|
||
|
||
# 框架类合同类型（含历史 framework，兼容未刷完数据）
|
||
FRAMEWORK_CONTRACT_TYPES = frozenset({"framework", "framework_sale", "framework_purchase"})
|
||
|
||
|
||
def is_framework_contract_type(contract_type: Optional[str]) -> bool:
|
||
return (contract_type or "").strip() in FRAMEWORK_CONTRACT_TYPES
|
||
|
||
|
||
def _validate_contract_type_and_products(data: dict) -> None:
|
||
contract_type = (_get_field(data, "contract_type", "contractType") or "").strip()
|
||
if contract_type in ("sale", "purchase") or is_framework_contract_type(contract_type):
|
||
products = _get_field(data, "products", "productList") or []
|
||
if not products:
|
||
raise ValueError("销售合同、采购合同或框架合同必须至少添加一个产品")
|
||
|
||
|
||
def _normalize_framework_contract_data(data: dict) -> None:
|
||
if not is_framework_contract_type(_get_field(data, "contract_type", "contractType")):
|
||
return
|
||
data["amount"] = 0
|
||
data["amountWithTax"] = 0
|
||
products = _get_field(data, "products", "productList") or []
|
||
for product in products:
|
||
if not isinstance(product, dict):
|
||
continue
|
||
quantity_mode = (_get_field(product, "quantity_mode", "quantityMode") or "customer_demand").strip()
|
||
price_mode = (_get_field(product, "price_mode", "priceMode") or "market").strip()
|
||
product["quantityMode"] = quantity_mode
|
||
product["priceMode"] = price_mode
|
||
if quantity_mode == "customer_demand":
|
||
product["quantity"] = 0
|
||
if price_mode == "market":
|
||
product["price"] = 0
|
||
product["subtotal"] = 0
|
||
product["taxAmount"] = 0
|
||
|
||
|
||
def _validate_product_prices_by_contract_type(data: dict) -> None:
|
||
contract_type = (_get_field(data, "contract_type", "contractType") or "").strip()
|
||
if contract_type not in ("sale", "purchase"):
|
||
return
|
||
products = _get_field(data, "products", "productList") or []
|
||
for product in products:
|
||
price = _decimal_value(_get_field(product, "price"))
|
||
subtotal = _decimal_value(_get_field(product, "subtotal"))
|
||
if price < Decimal("0"):
|
||
raise ValueError("销售/采购合同产品的含税单价不能为负数")
|
||
if subtotal < Decimal("0"):
|
||
raise ValueError("销售/采购合同产品的含税总价不能为负数")
|
||
|
||
|
||
def _validate_sale_purchase_contract_amount(data: dict) -> None:
|
||
contract_type = (_get_field(data, "contract_type", "contractType") or "").strip()
|
||
if contract_type not in ("sale", "purchase"):
|
||
return
|
||
amount = _decimal_value(_get_field(data, "amount"))
|
||
if amount < Decimal("0"):
|
||
raise ValueError("销售合同或采购合同的合同金额不能为负数")
|
||
|
||
|
||
def _resolve_contract_product_ids(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
products: List[dict],
|
||
user_id: str = None,
|
||
) -> None:
|
||
for product in products:
|
||
pid = resolve_product_id_for_quote_item(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
product_data=product,
|
||
user_id=user_id,
|
||
)
|
||
product["productId"] = pid
|
||
product["product_id"] = pid
|
||
|
||
|
||
def _calc_amount_with_tax(amount: Decimal, tax_rate: Decimal) -> Decimal:
|
||
if tax_rate is None:
|
||
tax_rate = Decimal("0")
|
||
return amount * (Decimal("1") + tax_rate / Decimal("100"))
|
||
|
||
|
||
def _apply_contract_entity_fields(
|
||
entity: CrmContract,
|
||
data: dict,
|
||
*,
|
||
db: Session,
|
||
tenant_no: str,
|
||
user_id: str = None,
|
||
is_create: bool = True,
|
||
) -> None:
|
||
field_map = (
|
||
(("contract_name", "contractName"), "contract_name"),
|
||
(("customer_id", "customerId"), "customer_id"),
|
||
(("customer_name", "customerName"), "customer_name"),
|
||
(("opportunity_id", "opportunityId"), "opportunity_id"),
|
||
(("quote_id", "quoteId"), "quote_id"),
|
||
(("parent_contract_id", "parentContractId"), "parent_contract_id"),
|
||
(("template_id", "templateId"), "template_id"),
|
||
(("contract_template", "contractTemplate"), "contract_template"),
|
||
(("contact_id", "contactId"), "contact_id"),
|
||
(("contract_type", "contractType"), "contract_type"),
|
||
(("contract_nature", "contractNature"), "contract_nature"),
|
||
(("currency",), "currency"),
|
||
(("performance_status", "performanceStatus"), "performance_status"),
|
||
(("contract_file", "contractFile"), "contract_file"),
|
||
(("attachment_files", "attachmentFiles"), "attachment_files"),
|
||
(("owner_user_id", "ownerUserId"), "owner_user_id"),
|
||
(("party_a", "partyA"), "party_a"),
|
||
(("party_a_address", "partyAAddress"), "party_a_address"),
|
||
(("party_a_contact", "partyAContact"), "party_a_contact"),
|
||
(("party_a_mobile", "partyAMobile"), "party_a_mobile"),
|
||
(("party_a_legal_person", "partyALegalPerson"), "party_a_legal_person"),
|
||
(("party_a_agent", "partyAAgent"), "party_a_agent"),
|
||
(("party_a_phone_fax", "partyAPhoneFax"), "party_a_phone_fax"),
|
||
(("party_a_bank", "partyABank"), "party_a_bank"),
|
||
(("party_a_account", "partyAAccount"), "party_a_account"),
|
||
(("supplier_id", "supplierId"), "supplier_id"),
|
||
(("settlement_method", "settlementMethod"), "settlement_method"),
|
||
(("party_b", "partyB"), "party_b"),
|
||
(("party_b_address", "partyBAddress"), "party_b_address"),
|
||
(("party_b_contact", "partyBContact"), "party_b_contact"),
|
||
(("party_b_mobile", "partyBMobile"), "party_b_mobile"),
|
||
(("party_b_legal_person", "partyBLegalPerson"), "party_b_legal_person"),
|
||
(("party_b_agent", "partyBAgent"), "party_b_agent"),
|
||
(("party_b_phone_fax", "partyBPhoneFax"), "party_b_phone_fax"),
|
||
(("party_b_bank", "partyBBank"), "party_b_bank"),
|
||
(("party_b_account", "partyBAccount"), "party_b_account"),
|
||
(("payment_terms", "paymentTerms"), "payment_terms"),
|
||
(("delivery_terms", "deliveryTerms"), "delivery_terms"),
|
||
(("contract_terms", "contractTerms"), "contract_terms"),
|
||
(("template_content", "templateContent"), "template_content"),
|
||
(("settlement_clause_drafts", "settlementClauseDrafts"), "settlement_clause_drafts"),
|
||
(("remark",), "remark"),
|
||
(("order_num", "orderNum"), "order_num"),
|
||
)
|
||
for keys, attr in field_map:
|
||
val = _get_field(data, *keys)
|
||
if val is not None:
|
||
setattr(entity, attr, val)
|
||
|
||
if _get_field(data, "amount") is not None:
|
||
entity.amount = _decimal_value(_get_field(data, "amount"))
|
||
if _get_field(data, "tax_rate", "taxRate") is not None:
|
||
entity.tax_rate = _decimal_value(_get_field(data, "tax_rate", "taxRate"))
|
||
if _get_field(data, "amount_with_tax", "amountWithTax") is not None:
|
||
entity.amount_with_tax = _decimal_value(_get_field(data, "amount_with_tax", "amountWithTax"))
|
||
elif entity.amount is not None and entity.tax_rate is not None:
|
||
entity.amount_with_tax = _calc_amount_with_tax(entity.amount, entity.tax_rate)
|
||
if _get_field(data, "freight") is not None:
|
||
entity.freight = _decimal_value(_get_field(data, "freight"))
|
||
if _get_field(data, "barrel_deposit", "barrelDeposit") is not None:
|
||
entity.barrel_deposit = _decimal_value(_get_field(data, "barrel_deposit", "barrelDeposit"))
|
||
|
||
for date_key, attr in (
|
||
(("sign_date", "signDate"), "sign_date"),
|
||
(("actual_sign_date", "actualSignDate"), "actual_sign_date"),
|
||
(("start_date", "startDate"), "start_date"),
|
||
(("end_date", "endDate"), "end_date"),
|
||
):
|
||
val = _get_field(data, *date_key)
|
||
if val is not None:
|
||
setattr(entity, attr, _parse_optional_date(val))
|
||
|
||
if _get_field(data, "contract_period", "contractPeriod") is not None:
|
||
entity.contract_period = int(_get_field(data, "contract_period", "contractPeriod"))
|
||
if _get_field(data, "performance_rate", "performanceRate") is not None:
|
||
entity.performance_rate = _decimal_value(_get_field(data, "performance_rate", "performanceRate"))
|
||
if _get_field(data, "customer_satisfaction", "customerSatisfaction") is not None:
|
||
entity.customer_satisfaction = int(_get_field(data, "customer_satisfaction", "customerSatisfaction"))
|
||
if _get_field(data, "renewal_count", "renewalCount") is not None:
|
||
entity.renewal_count = int(_get_field(data, "renewal_count", "renewalCount"))
|
||
|
||
if is_create and not entity.status:
|
||
entity.status = "draft"
|
||
if entity.status == "draft":
|
||
entity.approval_status = None
|
||
|
||
customer_id = (entity.customer_id or "").strip()
|
||
customer = None
|
||
if customer_id:
|
||
customer = get_customer(db, customer_id)
|
||
if customer and tenant_no and customer.tenant_no != tenant_no:
|
||
raise ValueError("客户不存在")
|
||
if customer:
|
||
entity.customer_name = customer.customer_name or entity.customer_name or ""
|
||
|
||
if entity.contact_id:
|
||
contact = get_contact(db, entity.contact_id)
|
||
if contact:
|
||
entity.contact_name = contact.contact_name or entity.contact_name or ""
|
||
entity.contact_mobile = contact.mobile or entity.contact_mobile or ""
|
||
|
||
owner_id = (entity.owner_user_id or user_id or "").strip()
|
||
if owner_id:
|
||
entity.owner_user_id = owner_id
|
||
entity.owner_user_name = _resolve_user_display_name(db, owner_id)
|
||
|
||
supplier = None
|
||
supplier_id = (entity.supplier_id or "").strip()
|
||
if supplier_id:
|
||
supplier = get_supplier_info(db, supplier_id, tenant_no=tenant_no)
|
||
|
||
# 后端兜底：当新增/编辑未显式传入供需方信息时，按供应商与客户基础资料补齐，避免写入空数据
|
||
if supplier:
|
||
if _is_blank(entity.party_a):
|
||
entity.party_a = _first_non_blank(getattr(supplier, "supplier_name", ""))
|
||
if _is_blank(entity.party_a_address):
|
||
entity.party_a_address = _first_non_blank(getattr(supplier, "supplier_address", ""))
|
||
if _is_blank(entity.party_a_legal_person):
|
||
entity.party_a_legal_person = _first_non_blank(getattr(supplier, "legal_person", ""))
|
||
if _is_blank(entity.party_a_agent):
|
||
entity.party_a_agent = _first_non_blank(getattr(supplier, "agent", ""))
|
||
if _is_blank(entity.party_a_phone_fax):
|
||
entity.party_a_phone_fax = _first_non_blank(getattr(supplier, "phone_fax", ""))
|
||
if _is_blank(entity.party_a_bank):
|
||
entity.party_a_bank = _first_non_blank(getattr(supplier, "bank_name", ""))
|
||
if _is_blank(entity.party_a_account):
|
||
entity.party_a_account = _first_non_blank(getattr(supplier, "bank_account", ""))
|
||
|
||
if customer:
|
||
inv = _get_default_invoice_info(db, tenant_no=tenant_no, customer_id=customer_id)
|
||
if _is_blank(entity.party_b):
|
||
entity.party_b = _first_non_blank(
|
||
inv.get("companyName") if inv else "",
|
||
customer.customer_name,
|
||
)
|
||
if _is_blank(entity.party_b_address):
|
||
entity.party_b_address = _first_non_blank(
|
||
inv.get("companyAddress") if inv else "",
|
||
_build_customer_address(customer),
|
||
)
|
||
if _is_blank(entity.party_b_contact):
|
||
entity.party_b_contact = _first_non_blank(entity.contact_name)
|
||
if _is_blank(entity.party_b_mobile):
|
||
entity.party_b_mobile = _first_non_blank(
|
||
inv.get("phoneFax") if inv else "",
|
||
entity.contact_mobile,
|
||
customer.mobile,
|
||
customer.telephone,
|
||
)
|
||
if _is_blank(entity.party_b_legal_person):
|
||
entity.party_b_legal_person = _first_non_blank(inv.get("legalRepresentative") if inv else "")
|
||
if _is_blank(entity.party_b_phone_fax):
|
||
entity.party_b_phone_fax = _first_non_blank(
|
||
inv.get("phoneFax") if inv else "",
|
||
customer.telephone,
|
||
)
|
||
if _is_blank(entity.party_b_bank):
|
||
entity.party_b_bank = _first_non_blank(inv.get("bankName") if inv else "")
|
||
if _is_blank(entity.party_b_account):
|
||
entity.party_b_account = _first_non_blank(inv.get("bankAccount") if inv else "")
|
||
|
||
received = entity.received_amount if entity.received_amount is not None else Decimal("0")
|
||
invoice = entity.invoice_amount if entity.invoice_amount is not None else Decimal("0")
|
||
amount = entity.amount if entity.amount is not None else Decimal("0")
|
||
entity.received_amount = received
|
||
entity.invoice_amount = invoice
|
||
entity.unreceived_amount = amount - received
|
||
entity.uninvoice_amount = amount - invoice
|
||
|
||
|
||
def create_contract(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
data: dict,
|
||
user_id: str = None,
|
||
dept_id: str = "",
|
||
) -> Dict[str, Any]:
|
||
"""Java: insertByBo"""
|
||
products = _get_field(data, "products", "productList") or []
|
||
if not isinstance(products, list):
|
||
raise ValueError("产品明细格式无效")
|
||
product_rows = [dict(p) for p in products]
|
||
|
||
_validate_contract_type_and_products(data)
|
||
_normalize_framework_contract_data(data)
|
||
_validate_product_prices_by_contract_type(data)
|
||
_validate_sale_purchase_contract_amount(data)
|
||
|
||
customer_id = (_get_field(data, "customer_id", "customerId") or "").strip()
|
||
if not customer_id:
|
||
raise ValueError("客户ID不能为空")
|
||
|
||
supplier_id = (_get_field(data, "supplier_id", "supplierId") or "").strip() or None
|
||
|
||
assert_products_allowed_for_supplier(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
supplier_id=str(supplier_id) if supplier_id else None,
|
||
products=product_rows or [],
|
||
)
|
||
|
||
contract_type = (_get_field(data, "contract_type", "contractType") or "sale").strip() or "sale"
|
||
sign_date = _parse_optional_date(_get_field(data, "sign_date", "signDate")) or date.today()
|
||
|
||
contract_no = (_get_field(data, "contract_no", "contractNo") or "").strip()
|
||
if not contract_no:
|
||
contract_no = generate_contract_no(db, tenant_no, supplier_id=supplier_id, sign_date=sign_date)
|
||
else:
|
||
contract_no = align_contract_no_to_sign_date(db, tenant_no, contract_no, sign_date)
|
||
if contract_no_exists(db, tenant_no, contract_no):
|
||
raise ValueError(f"合同编号已存在：{contract_no}")
|
||
|
||
user_contract_name = (_get_field(data, "contract_name", "contractName") or "").strip()
|
||
|
||
now = datetime.now()
|
||
entity = CrmContract(
|
||
id=generate_uuid(),
|
||
contract_no=contract_no,
|
||
contract_name="",
|
||
customer_id=customer_id,
|
||
status="draft",
|
||
contract_type=contract_type,
|
||
amount=_decimal_value(_get_field(data, "amount")),
|
||
tax_rate=_decimal_value(_get_field(data, "tax_rate", "taxRate"), Decimal("6.00")),
|
||
freight=_decimal_value(_get_field(data, "freight"), Decimal("0")),
|
||
barrel_deposit=_decimal_value(_get_field(data, "barrel_deposit", "barrelDeposit"), Decimal("0")),
|
||
currency=_get_field(data, "currency") or "CNY",
|
||
sign_date=sign_date,
|
||
tenant_no=tenant_no,
|
||
create_by=user_id,
|
||
create_dept_id=dept_id or "",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
received_amount=Decimal("0"),
|
||
invoice_amount=Decimal("0"),
|
||
)
|
||
_apply_contract_entity_fields(
|
||
entity, data, db=db, tenant_no=tenant_no, user_id=user_id, is_create=True
|
||
)
|
||
entity.contract_no = contract_no
|
||
entity.contract_name = _build_contract_name_from_parties(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
contract_no=contract_no,
|
||
party_a=entity.party_a or "",
|
||
party_b=entity.party_b or "",
|
||
supplier_id=entity.supplier_id or supplier_id or "",
|
||
customer_id=entity.customer_id or customer_id,
|
||
customer_name=entity.customer_name or "",
|
||
user_contract_name=user_contract_name,
|
||
)
|
||
|
||
_resolve_contract_product_ids(db, tenant_no=tenant_no, products=product_rows, user_id=user_id)
|
||
db.add(entity)
|
||
db.flush()
|
||
|
||
if product_rows:
|
||
batch_insert_contract_products(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
contract_id=entity.id,
|
||
products=product_rows,
|
||
contract_type=contract_type,
|
||
user_id=user_id,
|
||
dept_id=dept_id,
|
||
)
|
||
|
||
db.commit()
|
||
db.refresh(entity)
|
||
return get_contract_detail(db, entity.id, tenant_no=tenant_no)
|
||
|
||
|
||
# 禁止再编辑的合同状态：仅已完成；其余状态可编辑，保存后统一回草稿
|
||
CONTRACT_NON_EDITABLE_STATUSES = frozenset({"finished"})
|
||
# 兼容旧测试/引用：可编辑 = 非已完成（含草稿 / 审批中 / 已审批 / 执行中 / 已作废）
|
||
CONTRACT_EDITABLE_STATUSES = frozenset(
|
||
{"draft", "approving", "approved", "executing", "invalid"}
|
||
)
|
||
|
||
|
||
def assert_contract_editable_for_update(status: str) -> None:
|
||
s = (status or "").strip() or "draft"
|
||
if s in CONTRACT_NON_EDITABLE_STATUSES:
|
||
raise ValueError("已完成的合同不能修改")
|
||
|
||
|
||
def update_contract(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
data: dict,
|
||
user_id: str = None,
|
||
dept_id: str = "",
|
||
user_info=None,
|
||
) -> Dict[str, Any]:
|
||
"""Java: updateByBo"""
|
||
old = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not old:
|
||
raise ValueError("合同不存在")
|
||
prev_status = (old.status or "").strip() or "draft"
|
||
assert_contract_editable_for_update(prev_status)
|
||
|
||
products = _get_field(data, "products", "productList")
|
||
product_rows = [dict(p) for p in products] if products is not None else None
|
||
|
||
work_data = dict(data)
|
||
if product_rows is not None:
|
||
work_data["products"] = product_rows
|
||
_validate_contract_type_and_products(work_data)
|
||
_normalize_framework_contract_data(work_data)
|
||
_validate_product_prices_by_contract_type(work_data)
|
||
_validate_sale_purchase_contract_amount(work_data)
|
||
|
||
contract_type = (_get_field(data, "contract_type", "contractType") or old.contract_type or "sale").strip()
|
||
old.contract_type = contract_type
|
||
|
||
# 再编辑回草稿：作废全部历史审批（含已通过/已拒绝），避免再提审后 canApprove 命中旧记录
|
||
invalidate_all_approvals(db, contract_id)
|
||
|
||
_apply_contract_entity_fields(
|
||
old, data, db=db, tenant_no=tenant_no, user_id=user_id, is_create=False
|
||
)
|
||
|
||
# 编辑保存后统一回草稿（含审批中 / 已审批 / 执行中 / 已作废）
|
||
old.status = "draft"
|
||
old.approval_status = None
|
||
|
||
products_to_check = (
|
||
product_rows
|
||
if product_rows is not None
|
||
else list_contract_products(db, contract_id, tenant_no=tenant_no)
|
||
)
|
||
assert_products_allowed_for_supplier(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
supplier_id=str(old.supplier_id) if old.supplier_id else None,
|
||
products=products_to_check or [],
|
||
)
|
||
|
||
if product_rows is not None:
|
||
_resolve_contract_product_ids(db, tenant_no=tenant_no, products=product_rows, user_id=user_id)
|
||
|
||
sign_date = old.sign_date or date.today()
|
||
submitted_no = (_get_field(data, "contract_no", "contractNo") or "").strip() or (
|
||
old.contract_no or ""
|
||
).strip()
|
||
if submitted_no:
|
||
aligned_no = align_contract_no_to_sign_date(
|
||
db,
|
||
tenant_no,
|
||
submitted_no,
|
||
sign_date,
|
||
exclude_id=old.id,
|
||
)
|
||
if aligned_no != (old.contract_no or ""):
|
||
if contract_no_exists(db, tenant_no, aligned_no, exclude_id=old.id):
|
||
raise ValueError(f"合同编号已存在：{aligned_no}")
|
||
old.contract_no = aligned_no
|
||
elif submitted_no != (old.contract_no or ""):
|
||
if contract_no_exists(db, tenant_no, submitted_no, exclude_id=old.id):
|
||
raise ValueError(f"合同编号已存在：{submitted_no}")
|
||
old.contract_no = submitted_no
|
||
|
||
user_name = (_get_field(data, "contract_name", "contractName") or old.contract_name or "").strip()
|
||
old.contract_name = _build_contract_name_from_parties(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
contract_no=old.contract_no or "",
|
||
party_a=old.party_a or "",
|
||
party_b=old.party_b or "",
|
||
supplier_id=old.supplier_id or "",
|
||
customer_id=old.customer_id or "",
|
||
customer_name=old.customer_name or "",
|
||
user_contract_name=user_name,
|
||
)
|
||
|
||
old.update_by = user_id
|
||
old.update_time = datetime.now()
|
||
db.flush()
|
||
|
||
if product_rows is not None:
|
||
delete_products_by_contract_id(db, contract_id)
|
||
if product_rows:
|
||
batch_insert_contract_products(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
contract_id=contract_id,
|
||
products=product_rows,
|
||
contract_type=contract_type,
|
||
user_id=user_id,
|
||
dept_id=dept_id,
|
||
)
|
||
|
||
db.commit()
|
||
return get_contract_detail(db, contract_id, tenant_no=tenant_no)
|
||
|
||
|
||
def soft_delete_contracts(
|
||
db: Session,
|
||
contract_ids: List[str],
|
||
*,
|
||
tenant_no: str,
|
||
validate: bool = True,
|
||
user_info=None,
|
||
) -> int:
|
||
"""Java: deleteWithValidByIds — 软删除"""
|
||
count = 0
|
||
for cid in contract_ids:
|
||
entity = _get_contract_row_in_scope(db, cid, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
continue
|
||
if validate and entity.status != "draft":
|
||
raise ValueError("只能删除草稿状态的合同")
|
||
assert_row_deletable_by_creator(entity, user_info)
|
||
entity.del_flag = 1
|
||
entity.update_time = datetime.now()
|
||
count += 1
|
||
db.commit()
|
||
return count
|
||
|
||
|
||
def validate_framework_contract_for_submit_approval(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
contract: CrmContract = None,
|
||
) -> None:
|
||
"""框架合同提交审批校验：仅要求有产品明细；调价区间为可选，不强制。"""
|
||
entity = contract or get_contract_row(db, contract_id, tenant_no=tenant_no)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
if not is_framework_contract_type(entity.contract_type):
|
||
return
|
||
|
||
products = list_contract_products(db, contract_id, tenant_no=tenant_no)
|
||
if not products:
|
||
raise ValueError("框架合同必须维护产品明细")
|
||
|
||
|
||
def submit_contract_approval(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
approver_ids: List[str] = None,
|
||
remark: str = "",
|
||
user_id: str = None,
|
||
dept_id: str = "",
|
||
user_info=None,
|
||
) -> bool:
|
||
"""Java: submitApproval"""
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
if entity.status != "draft":
|
||
raise ValueError("只有草稿状态的合同才能提交审批")
|
||
|
||
validate_framework_contract_for_submit_approval(
|
||
db, contract_id, tenant_no=tenant_no, contract=entity
|
||
)
|
||
|
||
entity.status = "approving"
|
||
entity.approval_status = "pending"
|
||
if remark:
|
||
entity.remark = remark
|
||
entity.update_time = datetime.now()
|
||
db.flush()
|
||
|
||
if approver_ids:
|
||
# 重新提审前清掉残留审批记录，保证仅当前轮有效
|
||
invalidate_all_approvals(db, contract_id)
|
||
create_approval_records(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
contract_id=contract_id,
|
||
approver_ids=approver_ids,
|
||
user_id=user_id,
|
||
dept_id=dept_id,
|
||
)
|
||
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def batch_submit_contract_approval(
|
||
db: Session,
|
||
contract_ids: List[str],
|
||
*,
|
||
tenant_no: str,
|
||
approver_ids: List[str] = None,
|
||
remark: str = "",
|
||
user_id: str = None,
|
||
dept_id: str = "",
|
||
user_info=None,
|
||
) -> int:
|
||
"""Java: batchSubmitApproval"""
|
||
success = 0
|
||
for cid in contract_ids:
|
||
try:
|
||
submit_contract_approval(
|
||
db,
|
||
cid,
|
||
tenant_no=tenant_no,
|
||
approver_ids=approver_ids,
|
||
remark=remark,
|
||
user_id=user_id,
|
||
dept_id=dept_id,
|
||
)
|
||
success += 1
|
||
except ValueError:
|
||
continue
|
||
return success
|
||
|
||
|
||
def withdraw_contract_approval(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
remark: str = "",
|
||
user_id: str = None,
|
||
user_info=None,
|
||
) -> bool:
|
||
"""Java: withdrawApproval"""
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
if entity.status != "approving":
|
||
raise ValueError("只有审批中的合同才能撤回")
|
||
|
||
entity.status = "draft"
|
||
entity.approval_status = None
|
||
if remark:
|
||
entity.remark = remark
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
delete_pending_approvals(db, contract_id)
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def start_execute(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
remark: str = "",
|
||
user_id: str = None,
|
||
user_info=None,
|
||
) -> bool:
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
if entity.status != "approved":
|
||
raise ValueError("只有已审批的合同才能开始执行")
|
||
entity.status = "executing"
|
||
if remark:
|
||
entity.remark = remark
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def finish_contract(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
remark: str = "",
|
||
user_id: str = None,
|
||
user_info=None,
|
||
) -> bool:
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
if entity.status != "executing":
|
||
raise ValueError("只有执行中的合同才能完成")
|
||
entity.status = "finished"
|
||
if remark:
|
||
entity.remark = remark
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def invalid_contract(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
invalid_reason: str,
|
||
remark: str = "",
|
||
user_id: str = None,
|
||
user_info=None,
|
||
) -> bool:
|
||
if not (invalid_reason or "").strip():
|
||
raise ValueError("作废原因不能为空")
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
if entity.status == "finished":
|
||
raise ValueError("已完成的合同不能作废")
|
||
entity.status = "invalid"
|
||
entity.invalid_reason = invalid_reason
|
||
entity.invalid_time = datetime.now()
|
||
if remark:
|
||
entity.remark = remark
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def save_template(
|
||
db: Session,
|
||
contract_id: str,
|
||
template_content: str,
|
||
*,
|
||
tenant_no: str,
|
||
settlement_clause_drafts: str = None,
|
||
user_info=None,
|
||
) -> bool:
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
|
||
drafts = settlement_clause_drafts
|
||
if not (drafts or "").strip():
|
||
method = (entity.settlement_method or "").strip() or "arrive_payment"
|
||
drafts = json.dumps(
|
||
{method: default_settlement_clause(method)},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
entity.template_content = template_content or ""
|
||
entity.settlement_clause_drafts = drafts
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def get_template(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
user_info=None,
|
||
) -> str:
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
return entity.template_content or ""
|
||
|
||
|
||
def upload_file(
|
||
db: Session,
|
||
contract_id: str,
|
||
file_url: str,
|
||
*,
|
||
tenant_no: str,
|
||
user_info=None,
|
||
) -> bool:
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
entity.contract_file = file_url or ""
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def upload_attachments(
|
||
db: Session,
|
||
contract_id: str,
|
||
oss_ids: str,
|
||
*,
|
||
tenant_no: str,
|
||
user_info=None,
|
||
) -> bool:
|
||
if not (oss_ids or "").strip():
|
||
raise ValueError("附件ossId不能为空")
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
|
||
existing = (entity.attachment_files or "").strip()
|
||
new_parts = [x.strip() for x in oss_ids.split(",") if x.strip()]
|
||
merged = set(existing.split(",")) if existing and not existing.startswith("[") else set()
|
||
merged.discard("")
|
||
merged.update(new_parts)
|
||
entity.attachment_files = ",".join(sorted(merged))
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def upload_attachments_with_category(
|
||
db: Session,
|
||
contract_id: str,
|
||
attachments: List[dict],
|
||
*,
|
||
tenant_no: str,
|
||
user_info=None,
|
||
) -> bool:
|
||
"""按入参整表覆盖合同扫描件；attachments=[] 表示清空。"""
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
|
||
saved: List[dict] = []
|
||
seen = set()
|
||
for att in attachments or []:
|
||
oss_id = str(att.get("ossId") or att.get("oss_id") or "").strip()
|
||
if not oss_id or oss_id in seen:
|
||
continue
|
||
seen.add(oss_id)
|
||
saved.append({
|
||
"ossId": oss_id,
|
||
"category": str(att.get("category") or "other"),
|
||
})
|
||
|
||
entity.attachment_files = json.dumps(saved, ensure_ascii=False)
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def statistics_overview(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
owner_user_id: str = None,
|
||
begin_time: str = None,
|
||
end_time: str = None,
|
||
user_info=None,
|
||
) -> Dict[str, Any]:
|
||
q = db.query(CrmContract).filter(CrmContract.tenant_no == tenant_no, CrmContract.del_flag == 0)
|
||
if owner_user_id:
|
||
q = q.filter(CrmContract.owner_user_id == owner_user_id)
|
||
begin_dt = _parse_optional_datetime(begin_time)
|
||
end_dt = _parse_optional_datetime(end_time)
|
||
if begin_dt and end_dt:
|
||
q = q.filter(CrmContract.create_time >= begin_dt, CrmContract.create_time <= end_dt)
|
||
elif begin_dt:
|
||
q = q.filter(CrmContract.create_time >= begin_dt)
|
||
elif end_dt:
|
||
q = q.filter(CrmContract.create_time <= end_dt)
|
||
if user_info:
|
||
q = apply_crm_owner_scope(q, CrmContract, user_info, db, tenant_no)
|
||
|
||
contracts = q.all()
|
||
total_count = len(contracts)
|
||
|
||
def _count(status: str) -> int:
|
||
return sum(1 for c in contracts if c.status == status)
|
||
|
||
total_amount = sum(_decimal_value(c.amount) for c in contracts)
|
||
received_amount = sum(_decimal_value(c.received_amount) for c in contracts)
|
||
unreceived_amount = sum(_decimal_value(c.unreceived_amount) for c in contracts)
|
||
invoice_amount = sum(_decimal_value(c.invoice_amount) for c in contracts)
|
||
uninvoice_amount = sum(_decimal_value(c.uninvoice_amount) for c in contracts)
|
||
|
||
received_rate = (
|
||
(received_amount / total_amount * Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
||
if total_amount > 0
|
||
else Decimal("0")
|
||
)
|
||
invoice_rate = (
|
||
(invoice_amount / total_amount * Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
||
if total_amount > 0
|
||
else Decimal("0")
|
||
)
|
||
|
||
return {
|
||
"totalCount": total_count,
|
||
"draftCount": _count("draft"),
|
||
"approvingCount": _count("approving"),
|
||
"executingCount": _count("executing"),
|
||
"finishedCount": _count("finished"),
|
||
"invalidCount": _count("invalid"),
|
||
"totalAmount": float(total_amount),
|
||
"receivedAmount": float(received_amount),
|
||
"unreceivedAmount": float(unreceived_amount),
|
||
"receivedRate": float(received_rate),
|
||
"invoiceAmount": float(invoice_amount),
|
||
"uninvoiceAmount": float(uninvoice_amount),
|
||
"invoiceRate": float(invoice_rate),
|
||
}
|
||
|
||
|
||
def statistics_by_status(db: Session, tenant_no: str, *, user_info=None) -> List[Dict[str, Any]]:
|
||
q = db.query(CrmContract).filter(CrmContract.tenant_no == tenant_no, CrmContract.del_flag == 0)
|
||
if user_info:
|
||
q = apply_crm_owner_scope(q, CrmContract, user_info, db, tenant_no)
|
||
contracts = q.all()
|
||
total_count = len(contracts)
|
||
total_amount = sum(_decimal_value(c.amount) for c in contracts)
|
||
|
||
status_map = (
|
||
("draft", "草稿"),
|
||
("approving", "审批中"),
|
||
("approved", "已审批"),
|
||
("executing", "执行中"),
|
||
("finished", "已完成"),
|
||
("invalid", "已作废"),
|
||
)
|
||
result = []
|
||
for status, status_name in status_map:
|
||
subset = [c for c in contracts if c.status == status]
|
||
count = len(subset)
|
||
amount = sum(_decimal_value(c.amount) for c in subset)
|
||
percent = (count * 100.0 / total_count) if total_count else 0.0
|
||
result.append(
|
||
{
|
||
"status": status,
|
||
"statusName": status_name,
|
||
"count": count,
|
||
"amount": float(amount),
|
||
"percent": f"{percent:.2f}",
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def statistics_by_type(db: Session, tenant_no: str, *, user_info=None) -> List[Dict[str, Any]]:
|
||
q = db.query(CrmContract).filter(CrmContract.tenant_no == tenant_no, CrmContract.del_flag == 0)
|
||
if user_info:
|
||
q = apply_crm_owner_scope(q, CrmContract, user_info, db, tenant_no)
|
||
contracts = q.all()
|
||
total_count = len(contracts)
|
||
type_map = (
|
||
("sale", "销售合同"),
|
||
("service", "服务合同"),
|
||
("purchase", "采购合同"),
|
||
("framework_sale", "框架销售合同"),
|
||
("framework_purchase", "框架采购合同"),
|
||
("framework", "框架合同"), # 历史兼容
|
||
("supplement", "补充协议"),
|
||
("lease", "租赁合同"),
|
||
("other", "其他"),
|
||
)
|
||
result = []
|
||
for contract_type, type_name in type_map:
|
||
subset = [c for c in contracts if c.contract_type == contract_type]
|
||
count = len(subset)
|
||
if count == 0:
|
||
continue
|
||
amount = sum(_decimal_value(c.amount) for c in subset)
|
||
percent = (count * 100.0 / total_count) if total_count else 0.0
|
||
result.append(
|
||
{
|
||
"contractType": contract_type,
|
||
"typeName": type_name,
|
||
"count": count,
|
||
"amount": float(amount),
|
||
"percent": f"{percent:.2f}",
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def _contract_period_key(dt: datetime, period_type: str) -> str:
|
||
if period_type == "year":
|
||
return dt.strftime("%Y")
|
||
if period_type == "quarter":
|
||
quarter = (dt.month - 1) // 3 + 1
|
||
return f"{dt.year}-Q{quarter}"
|
||
return dt.strftime("%Y-%m")
|
||
|
||
|
||
def statistics_trend(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
period_type: str = "month",
|
||
begin_time: str = None,
|
||
end_time: str = None,
|
||
user_info=None,
|
||
) -> List[Dict[str, Any]]:
|
||
q = db.query(CrmContract).filter(CrmContract.tenant_no == tenant_no, CrmContract.del_flag == 0)
|
||
begin_dt = _parse_optional_datetime(begin_time)
|
||
end_dt = _parse_optional_datetime(end_time)
|
||
if begin_dt and end_dt:
|
||
q = q.filter(CrmContract.create_time >= begin_dt, CrmContract.create_time <= end_dt)
|
||
if user_info:
|
||
q = apply_crm_owner_scope(q, CrmContract, user_info, db, tenant_no)
|
||
contracts = q.all()
|
||
if not contracts:
|
||
return []
|
||
|
||
grouped: Dict[str, List[CrmContract]] = defaultdict(list)
|
||
for contract in contracts:
|
||
create_time = contract.create_time or datetime.now()
|
||
grouped[_contract_period_key(create_time, period_type or "month")].append(contract)
|
||
|
||
result = []
|
||
for period, subset in sorted(grouped.items()):
|
||
total_amount = sum(_decimal_value(c.amount) for c in subset)
|
||
received_amount = sum(_decimal_value(c.received_amount) for c in subset)
|
||
received_rate = (
|
||
(received_amount / total_amount * Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
||
if total_amount > 0
|
||
else Decimal("0")
|
||
)
|
||
result.append(
|
||
{
|
||
"period": period,
|
||
"totalCount": len(subset),
|
||
"totalAmount": float(total_amount),
|
||
"receivedAmount": float(received_amount),
|
||
"receivedRate": float(received_rate),
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def query_expiring(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
days: int = 30,
|
||
user_info=None,
|
||
) -> List[Dict[str, Any]]:
|
||
if not days or days <= 0:
|
||
days = 30
|
||
today = date.today()
|
||
end_date = today + timedelta(days=days)
|
||
rows = (
|
||
db.query(CrmContract)
|
||
.filter(
|
||
CrmContract.tenant_no == tenant_no,
|
||
CrmContract.del_flag == 0,
|
||
CrmContract.status == "executing",
|
||
CrmContract.end_date.isnot(None),
|
||
CrmContract.end_date <= end_date,
|
||
CrmContract.end_date >= today,
|
||
)
|
||
.order_by(CrmContract.end_date.asc())
|
||
.all()
|
||
)
|
||
return [_row_to_dict(r) for r in rows]
|
||
|
||
|
||
def _quote_products_to_contract_products(quote_products: List[dict]) -> List[dict]:
|
||
rows = []
|
||
for item in quote_products:
|
||
rows.append(
|
||
{
|
||
"productId": item.get("productId") or item.get("product_id"),
|
||
"productName": item.get("productName") or item.get("product_name"),
|
||
"brandMode": item.get("brandMode") or item.get("brand_mode"),
|
||
"brandId": item.get("brandId") or item.get("brand_id"),
|
||
"brandName": item.get("brandName") or item.get("brand_name"),
|
||
"productNo": item.get("productNo") or item.get("product_no"),
|
||
"categoryId": item.get("categoryId") or item.get("category_id"),
|
||
"categoryName": item.get("categoryName") or item.get("category_name"),
|
||
"model": item.get("model"),
|
||
"unit": item.get("unit"),
|
||
"quantity": item.get("quantity"),
|
||
"price": item.get("price"),
|
||
"discount": item.get("discount"),
|
||
"taxRate": item.get("taxRate") or item.get("tax_rate"),
|
||
"taxAmount": item.get("taxAmount") or item.get("tax_amount"),
|
||
"subtotal": item.get("subtotal"),
|
||
"packageSpec": item.get("packageSpec") or item.get("package_spec"),
|
||
"deliveryMethod": item.get("deliveryMethod") or item.get("delivery_method"),
|
||
"description": item.get("description"),
|
||
"sort": item.get("sort"),
|
||
"remark": item.get("remark"),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
def create_from_quote(
|
||
db: Session,
|
||
quote_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
data: dict,
|
||
user_id: str = None,
|
||
dept_id: str = "",
|
||
user_info=None,
|
||
) -> str:
|
||
"""Java: createFromQuote — 返回合同 id"""
|
||
quote = get_quote(db, quote_id, tenant_no=tenant_no)
|
||
if not quote:
|
||
raise ValueError("报价单不存在")
|
||
|
||
payload = dict(data)
|
||
payload.setdefault("quoteId", quote_id)
|
||
payload.setdefault("customerId", quote.customer_id)
|
||
if quote.opportunity_id:
|
||
payload.setdefault("opportunityId", quote.opportunity_id)
|
||
if payload.get("amount") is None:
|
||
payload["amount"] = float(quote.final_amount or quote.total_amount or 0)
|
||
if payload.get("freight") is None:
|
||
payload["freight"] = float(quote.freight or 0)
|
||
if payload.get("barrelDeposit") is None and payload.get("barrel_deposit") is None:
|
||
payload["barrelDeposit"] = float(getattr(quote, "barrel_deposit", None) or 0)
|
||
if not payload.get("ownerUserId"):
|
||
payload["ownerUserId"] = quote.owner_user_id or user_id or ""
|
||
if quote.supplier_id and not payload.get("supplierId"):
|
||
payload["supplierId"] = quote.supplier_id
|
||
|
||
products = _get_field(payload, "products", "productList")
|
||
if not products:
|
||
quote_products = list_quote_products(db, quote_id, tenant_no=tenant_no)
|
||
payload["products"] = _quote_products_to_contract_products(quote_products)
|
||
|
||
if not payload.get("contractType"):
|
||
qtype = (quote.quote_type or "sales").strip()
|
||
payload["contractType"] = "purchase" if qtype == "purchase" else "sale"
|
||
|
||
detail = create_contract(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
data=payload,
|
||
user_id=user_id,
|
||
dept_id=dept_id,
|
||
)
|
||
contract_id = detail["id"]
|
||
|
||
quote_row = (
|
||
db.query(CrmQuote)
|
||
.filter(CrmQuote.id == quote_id, CrmQuote.tenant_no == tenant_no, CrmQuote.del_flag == 0)
|
||
.first()
|
||
)
|
||
if quote_row:
|
||
quote_row.status = "converted"
|
||
quote_row.contract_id = contract_id
|
||
quote_row.update_time = datetime.now()
|
||
db.commit()
|
||
|
||
return contract_id
|
||
|
||
|
||
def renew_contract(
|
||
db: Session,
|
||
parent_contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
data: dict,
|
||
user_id: str = None,
|
||
dept_id: str = "",
|
||
user_info=None,
|
||
) -> str:
|
||
"""Java: renewContract — 基础续签"""
|
||
parent = get_contract_row(db, parent_contract_id, tenant_no=tenant_no)
|
||
if not parent:
|
||
raise ValueError("原合同不存在")
|
||
|
||
renewal_count = (parent.renewal_count or 0) + 1
|
||
payload = dict(data)
|
||
payload.setdefault("customerId", parent.customer_id)
|
||
payload.setdefault("contactId", parent.contact_id)
|
||
payload.setdefault("contractType", parent.contract_type)
|
||
payload.setdefault("ownerUserId", parent.owner_user_id)
|
||
payload.setdefault("supplierId", payload.get("supplierId") or parent.supplier_id)
|
||
payload["parentContractId"] = parent_contract_id
|
||
payload["renewalCount"] = renewal_count
|
||
payload["contractNature"] = "renewal"
|
||
|
||
if not payload.get("contractName"):
|
||
payload["contractName"] = f"{parent.contract_name or '合同'}-续签{renewal_count}"
|
||
|
||
payload.setdefault("partyA", parent.party_a)
|
||
payload.setdefault("partyAAddress", parent.party_a_address)
|
||
payload.setdefault("partyAContact", parent.party_a_contact)
|
||
payload.setdefault("partyAMobile", parent.party_a_mobile)
|
||
payload.setdefault("partyB", parent.party_b)
|
||
payload.setdefault("partyBAddress", parent.party_b_address)
|
||
payload.setdefault("partyBContact", parent.party_b_contact)
|
||
payload.setdefault("partyBMobile", parent.party_b_mobile)
|
||
payload.setdefault("paymentTerms", payload.get("paymentTerms") or parent.payment_terms)
|
||
payload.setdefault("deliveryTerms", payload.get("deliveryTerms") or parent.delivery_terms)
|
||
payload.setdefault("contractTerms", payload.get("contractTerms") or parent.contract_terms)
|
||
|
||
if payload.get("amount") is None:
|
||
payload["amount"] = float(parent.amount or 0)
|
||
if payload.get("taxRate") is None and parent.tax_rate is not None:
|
||
payload["taxRate"] = float(parent.tax_rate)
|
||
|
||
products = _get_field(payload, "products", "productList")
|
||
if not products:
|
||
payload["products"] = list_contract_products(db, parent_contract_id, tenant_no=tenant_no)
|
||
|
||
detail = create_contract(db, tenant_no=tenant_no, data=payload, user_id=user_id, dept_id=dept_id)
|
||
return detail["id"]
|
||
|
||
|
||
def export_contracts(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
filters: dict = None,
|
||
user_info=None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""Java: queryList for export"""
|
||
q = db.query(CrmContract)
|
||
q = _build_contract_filters(db, q, filters or {}, tenant_no=tenant_no)
|
||
rows = q.limit(10000).all()
|
||
return [_row_to_dict(r) for r in rows]
|
||
|
||
|
||
# --- 子模块：变更 / 模板（列表与详情） ---
|
||
|
||
|
||
def _change_row_to_dict(row: CrmContractChange) -> Dict[str, Any]:
|
||
return {
|
||
"id": row.id,
|
||
"contractId": row.contract_id or "",
|
||
"contractNo": row.contract_no or "",
|
||
"changeType": row.change_type or "",
|
||
"changeReason": row.change_reason or "",
|
||
"changeContent": row.change_content or "",
|
||
"oldValue": row.old_value or "",
|
||
"newValue": row.new_value or "",
|
||
"changeAmount": float(row.change_amount) if row.change_amount is not None else None,
|
||
"approvalStatus": row.approval_status or "",
|
||
"approverId": row.approver_id or "",
|
||
"approverName": row.approver_name or "",
|
||
"approvalTime": row.approval_time.strftime(_FMT) if row.approval_time else "",
|
||
"approvalOpinion": row.approval_opinion or "",
|
||
"attachmentFiles": row.attachment_files or "",
|
||
"remark": row.remark or "",
|
||
"createTime": row.create_time.strftime(_FMT) if row.create_time else "",
|
||
}
|
||
|
||
|
||
def list_contract_changes(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
filters: dict = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
from app.services.crm.contract_change_list_filters import build_contract_change_column_filters
|
||
|
||
q = db.query(CrmContractChange).filter(
|
||
CrmContractChange.tenant_no == tenant_no,
|
||
CrmContractChange.del_flag == 0,
|
||
)
|
||
if user_info:
|
||
q = apply_crm_create_by_scope(q, CrmContractChange, user_info, db, tenant_no)
|
||
filters = filters or {}
|
||
contract_id = (filters.get("contractId") or filters.get("contract_id") or "").strip()
|
||
if contract_id:
|
||
q = q.filter(CrmContractChange.contract_id == contract_id)
|
||
|
||
keyword = (_get_field(filters, "keyword") or "").strip()
|
||
if keyword:
|
||
like = f"%{keyword}%"
|
||
q = q.filter(
|
||
or_(
|
||
CrmContractChange.contract_no.like(like),
|
||
CrmContractChange.change_reason.like(like),
|
||
exists().where(
|
||
CrmContract.id == CrmContractChange.contract_id,
|
||
CrmContract.del_flag == 0,
|
||
CrmContract.contract_name.like(like),
|
||
),
|
||
)
|
||
)
|
||
|
||
for expr in build_contract_change_column_filters(**filters):
|
||
q = q.filter(expr)
|
||
|
||
total = q.order_by(None).with_entities(func.count(CrmContractChange.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 500)
|
||
rows = q.order_by(CrmContractChange.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
|
||
return _enrich_change_rows(db, [_change_row_to_dict(r) for r in rows]), int(total)
|
||
|
||
|
||
def get_contract_change_detail(
|
||
db: Session,
|
||
change_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
user_info=None,
|
||
) -> Dict[str, Any]:
|
||
row = (
|
||
db.query(CrmContractChange)
|
||
.filter(
|
||
CrmContractChange.id == change_id,
|
||
CrmContractChange.tenant_no == tenant_no,
|
||
CrmContractChange.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not row:
|
||
raise ValueError("合同变更记录不存在")
|
||
ensure_crm_create_by_row_in_scope(
|
||
db, row, user_info, tenant_no, message="无权访问该合同变更"
|
||
)
|
||
return _change_row_to_dict(row)
|
||
|
||
|
||
def _template_row_to_dict(row: CrmContractTemplate) -> Dict[str, Any]:
|
||
return {
|
||
"id": row.id,
|
||
"templateNo": row.template_no or "",
|
||
"templateName": row.template_name or "",
|
||
"contractType": row.contract_type or "",
|
||
"templateType": row.contract_type or "",
|
||
"templateContent": row.template_content or "",
|
||
"paymentTerms": row.payment_terms or "",
|
||
"deliveryTerms": row.delivery_terms or "",
|
||
"contractTerms": row.contract_terms or "",
|
||
"partyB": row.party_b or "",
|
||
"partyBAddress": row.party_b_address or "",
|
||
"partyBContact": row.party_b_contact or "",
|
||
"partyBMobile": row.party_b_mobile or "",
|
||
"defaultCurrency": row.default_currency or "CNY",
|
||
"defaultTaxRate": float(row.default_tax_rate) if row.default_tax_rate is not None else 6.0,
|
||
"defaultPeriod": row.default_period,
|
||
"isActive": row.is_active if row.is_active is not None else 1,
|
||
"usageCount": row.usage_count if row.usage_count is not None else 0,
|
||
"sort": row.sort if row.sort is not None else 0,
|
||
"remark": row.remark or "",
|
||
"createTime": row.create_time.strftime(_FMT) if row.create_time else "",
|
||
}
|
||
|
||
|
||
def list_contract_templates(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
filters: dict = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
from app.services.crm.contract_template_list_filters import (
|
||
build_contract_template_column_filters,
|
||
)
|
||
|
||
q = db.query(CrmContractTemplate).filter(
|
||
CrmContractTemplate.tenant_no == tenant_no,
|
||
CrmContractTemplate.del_flag == 0,
|
||
)
|
||
if user_info:
|
||
q = apply_crm_create_by_scope(q, CrmContractTemplate, user_info, db, tenant_no)
|
||
filters = filters or {}
|
||
|
||
keyword = (_get_field(filters, "keyword") or "").strip()
|
||
if keyword:
|
||
like = f"%{keyword}%"
|
||
q = q.filter(
|
||
or_(
|
||
CrmContractTemplate.template_no.like(like),
|
||
CrmContractTemplate.template_name.like(like),
|
||
)
|
||
)
|
||
|
||
for expr in build_contract_template_column_filters(**filters):
|
||
q = q.filter(expr)
|
||
|
||
total = q.order_by(None).with_entities(func.count(CrmContractTemplate.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 500)
|
||
rows = q.order_by(CrmContractTemplate.sort.asc(), CrmContractTemplate.create_time.desc()).offset(
|
||
(page - 1) * page_size
|
||
).limit(page_size).all()
|
||
return [_template_row_to_dict(r) for r in rows], int(total)
|
||
|
||
|
||
def get_contract_template_detail(
|
||
db: Session,
|
||
template_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
) -> Dict[str, Any]:
|
||
row = (
|
||
db.query(CrmContractTemplate)
|
||
.filter(
|
||
CrmContractTemplate.id == template_id,
|
||
CrmContractTemplate.tenant_no == tenant_no,
|
||
CrmContractTemplate.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not row:
|
||
raise ValueError("合同模板不存在")
|
||
return _template_row_to_dict(row)
|
||
|
||
|
||
def _opportunity_products_to_contract_products(items: List[dict]) -> List[dict]:
|
||
rows = []
|
||
for item in items:
|
||
rows.append(
|
||
{
|
||
"productId": item.get("productId") or item.get("product_id"),
|
||
"productName": item.get("productName") or item.get("product_name"),
|
||
"productNo": item.get("productNo") or item.get("product_no") or item.get("productCode"),
|
||
"categoryId": item.get("categoryId") or item.get("category_id"),
|
||
"categoryName": item.get("categoryName") or item.get("category_name"),
|
||
"model": item.get("model"),
|
||
"unit": item.get("unit"),
|
||
"quantity": item.get("quantity"),
|
||
"price": item.get("price"),
|
||
"discount": item.get("discount"),
|
||
"taxRate": item.get("taxRate") or item.get("tax_rate"),
|
||
"subtotal": item.get("subtotal"),
|
||
"description": item.get("description"),
|
||
"sort": item.get("sort") or item.get("sortOrder"),
|
||
"remark": item.get("remark"),
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
def create_from_opportunity(
|
||
db: Session,
|
||
opportunity_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
data: dict,
|
||
user_id: str = None,
|
||
dept_id: str = "",
|
||
user_info=None,
|
||
) -> str:
|
||
"""Java: createFromOpportunity — 返回合同 id"""
|
||
opp_row = (
|
||
db.query(CrmOpportunity)
|
||
.filter(
|
||
CrmOpportunity.id == opportunity_id,
|
||
CrmOpportunity.tenant_no == tenant_no,
|
||
CrmOpportunity.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not opp_row:
|
||
raise ValueError("商机不存在")
|
||
|
||
payload = dict(data)
|
||
payload.setdefault("opportunityId", opportunity_id)
|
||
payload.setdefault("customerId", opp_row.customer_id)
|
||
if payload.get("amount") is None:
|
||
payload["amount"] = float(opp_row.amount or 0)
|
||
if not payload.get("ownerUserId"):
|
||
payload["ownerUserId"] = opp_row.owner_user_id or user_id or ""
|
||
|
||
products = _get_field(payload, "products", "productList")
|
||
if not products:
|
||
from app.crud.crm.opportunity_product import list_products_by_opportunity
|
||
|
||
opp_products = list_products_by_opportunity(db, opportunity_id, tenant_no=tenant_no)
|
||
payload["products"] = _opportunity_products_to_contract_products(opp_products)
|
||
|
||
detail = create_contract(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
data=payload,
|
||
user_id=user_id,
|
||
dept_id=dept_id,
|
||
)
|
||
contract_id = detail["id"]
|
||
|
||
opp_row.stage = "won"
|
||
opp_row.update_time = datetime.now()
|
||
db.commit()
|
||
return contract_id
|
||
|
||
|
||
def sign_contract(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
actual_sign_date: str,
|
||
contract_file: str,
|
||
remark: str = "",
|
||
user_id: str = None,
|
||
user_info=None,
|
||
) -> bool:
|
||
"""Java: signContract"""
|
||
entity = _get_contract_row_in_scope(db, contract_id, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
raise ValueError("合同不存在")
|
||
if entity.status != "approved":
|
||
raise ValueError("只有审批通过的合同才能签订")
|
||
|
||
entity.actual_sign_date = _parse_optional_date(actual_sign_date)
|
||
entity.contract_file = (contract_file or "").strip()
|
||
entity.status = "executing"
|
||
entity.performance_status = "normal"
|
||
entity.performance_rate = Decimal("0")
|
||
if remark:
|
||
entity.remark = remark
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def batch_assign_owner(
|
||
db: Session,
|
||
contract_ids: List[str],
|
||
*,
|
||
tenant_no: str,
|
||
owner_user_id: str,
|
||
remark: str = "",
|
||
user_id: str = None,
|
||
user_info=None,
|
||
) -> int:
|
||
"""Java: batchAssignOwner"""
|
||
if not contract_ids:
|
||
raise ValueError("合同ID列表不能为空")
|
||
owner_id = (owner_user_id or "").strip()
|
||
if not owner_id:
|
||
raise ValueError("负责人ID不能为空")
|
||
|
||
success = 0
|
||
owner_name = _resolve_user_display_name(db, owner_id)
|
||
for cid in contract_ids:
|
||
entity = _get_contract_row_in_scope(db, cid, tenant_no=tenant_no, user_info=user_info)
|
||
if not entity:
|
||
continue
|
||
entity.owner_user_id = owner_id
|
||
entity.owner_user_name = owner_name
|
||
if remark:
|
||
entity.remark = remark
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
success += 1
|
||
db.commit()
|
||
return success
|
||
|
||
|
||
def download_contract_pdf(
|
||
db: Session,
|
||
contract_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
user_info=None,
|
||
) -> str:
|
||
"""Java: downloadContractPdf — 返回合同文件 URL/路径"""
|
||
detail = get_contract_detail(db, contract_id, tenant_no=tenant_no)
|
||
if not detail:
|
||
raise ValueError("合同不存在")
|
||
file_path = (detail.get("contractFile") or "").strip()
|
||
if not file_path:
|
||
raise ValueError("合同PDF文件不存在")
|
||
return file_path
|
||
|
||
|
||
def generate_template_no(db: Session, tenant_no: str) -> str:
|
||
"""Java: generateTemplateNo — TMPL{yyyyMMdd}{4位序号}"""
|
||
date_str = datetime.now().strftime("%Y%m%d")
|
||
prefix = f"TMPL{date_str}"
|
||
last = (
|
||
db.query(CrmContractTemplate)
|
||
.filter(
|
||
CrmContractTemplate.tenant_no == tenant_no,
|
||
CrmContractTemplate.del_flag == 0,
|
||
CrmContractTemplate.template_no.like(f"{prefix}%"),
|
||
)
|
||
.order_by(CrmContractTemplate.template_no.desc())
|
||
.first()
|
||
)
|
||
sequence = 1
|
||
if last and last.template_no and len(last.template_no) >= len(prefix) + 4:
|
||
try:
|
||
sequence = int(last.template_no[len(prefix) :]) + 1
|
||
except ValueError:
|
||
sequence = 1
|
||
for retry in range(10):
|
||
candidate = f"{prefix}{sequence + retry:04d}"
|
||
exists = (
|
||
db.query(CrmContractTemplate.id)
|
||
.filter(CrmContractTemplate.template_no == candidate, CrmContractTemplate.del_flag == 0)
|
||
.first()
|
||
)
|
||
if not exists:
|
||
return candidate
|
||
raise ValueError("生成模板编号失败，请稍后重试")
|
||
|
||
|
||
def _get_change_row(db: Session, change_id: str, *, tenant_no: str) -> Optional[CrmContractChange]:
|
||
return (
|
||
db.query(CrmContractChange)
|
||
.filter(
|
||
CrmContractChange.id == change_id,
|
||
CrmContractChange.tenant_no == tenant_no,
|
||
CrmContractChange.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def _enrich_change_rows(db: Session, rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
contract_ids = {row.get("contractId") for row in rows if row.get("contractId")}
|
||
if not contract_ids:
|
||
return rows
|
||
contracts = (
|
||
db.query(CrmContract.id, CrmContract.contract_name)
|
||
.filter(CrmContract.id.in_(contract_ids), CrmContract.del_flag == 0)
|
||
.all()
|
||
)
|
||
name_map = {item.id: item.contract_name or "" for item in contracts}
|
||
for row in rows:
|
||
row["contractName"] = name_map.get(row.get("contractId"), "")
|
||
return rows
|
||
|
||
|
||
def create_contract_change(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
data: dict,
|
||
user_id: str = None,
|
||
dept_id: str = "",
|
||
) -> Dict[str, Any]:
|
||
contract_id = (_get_field(data, "contract_id", "contractId") or "").strip()
|
||
if not contract_id:
|
||
raise ValueError("合同ID不能为空")
|
||
contract = get_contract_row(db, contract_id, tenant_no=tenant_no)
|
||
if not contract:
|
||
raise ValueError("合同不存在")
|
||
|
||
change_type = (_get_field(data, "change_type", "changeType") or "").strip()
|
||
change_reason = (_get_field(data, "change_reason", "changeReason") or "").strip()
|
||
if not change_type:
|
||
raise ValueError("变更类型不能为空")
|
||
if not change_reason:
|
||
raise ValueError("变更原因不能为空")
|
||
|
||
now = datetime.now()
|
||
entity = CrmContractChange(
|
||
id=generate_uuid(),
|
||
contract_id=contract_id,
|
||
contract_no=(_get_field(data, "contract_no", "contractNo") or contract.contract_no or "").strip(),
|
||
change_type=change_type,
|
||
change_reason=change_reason,
|
||
change_content=_get_field(data, "change_content", "changeContent") or "",
|
||
old_value=_get_field(data, "old_value", "oldValue") or "",
|
||
new_value=_get_field(data, "new_value", "newValue") or "",
|
||
change_amount=_decimal_value(_get_field(data, "change_amount", "changeAmount"))
|
||
if _get_field(data, "change_amount", "changeAmount") is not None
|
||
else None,
|
||
approval_status="pending",
|
||
attachment_files=_get_field(data, "attachment_files", "attachmentFiles") or "",
|
||
remark=_get_field(data, "remark") or "",
|
||
tenant_no=tenant_no,
|
||
create_by=user_id,
|
||
create_dept_id=dept_id or "",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(entity)
|
||
db.commit()
|
||
return get_contract_change_detail(db, entity.id, tenant_no=tenant_no)
|
||
|
||
|
||
def update_contract_change(
|
||
db: Session,
|
||
change_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
data: dict,
|
||
user_id: str = None,
|
||
user_info=None,
|
||
) -> Dict[str, Any]:
|
||
entity = _get_change_row(db, change_id, tenant_no=tenant_no)
|
||
if not entity:
|
||
raise ValueError("合同变更记录不存在")
|
||
ensure_crm_create_by_row_in_scope(
|
||
db, entity, user_info, tenant_no, message="无权访问该合同变更"
|
||
)
|
||
if entity.approval_status != "pending":
|
||
raise ValueError("只能修改待审批状态的变更记录")
|
||
|
||
for keys, attr in (
|
||
(("change_type", "changeType"), "change_type"),
|
||
(("change_reason", "changeReason"), "change_reason"),
|
||
(("change_content", "changeContent"), "change_content"),
|
||
(("old_value", "oldValue"), "old_value"),
|
||
(("new_value", "newValue"), "new_value"),
|
||
(("attachment_files", "attachmentFiles"), "attachment_files"),
|
||
(("remark",), "remark"),
|
||
):
|
||
val = _get_field(data, *keys)
|
||
if val is not None:
|
||
setattr(entity, attr, val)
|
||
if _get_field(data, "change_amount", "changeAmount") is not None:
|
||
entity.change_amount = _decimal_value(_get_field(data, "change_amount", "changeAmount"))
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return get_contract_change_detail(db, change_id, tenant_no=tenant_no)
|
||
|
||
|
||
def soft_delete_contract_changes(
|
||
db: Session,
|
||
change_ids: List[str],
|
||
*,
|
||
tenant_no: str,
|
||
validate: bool = True,
|
||
user_info=None,
|
||
) -> int:
|
||
count = 0
|
||
for cid in change_ids:
|
||
entity = _get_change_row(db, cid, tenant_no=tenant_no)
|
||
if not entity:
|
||
continue
|
||
ensure_crm_create_by_row_in_scope(
|
||
db, entity, user_info, tenant_no, message="无权访问该合同变更"
|
||
)
|
||
if validate and entity.approval_status != "pending":
|
||
raise ValueError("只能删除待审批状态的变更记录")
|
||
assert_row_deletable_by_creator(entity, user_info)
|
||
entity.del_flag = 1
|
||
entity.update_time = datetime.now()
|
||
count += 1
|
||
db.commit()
|
||
return count
|
||
|
||
|
||
def submit_change_approval(
|
||
db: Session,
|
||
change_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
approver_ids: List[str] = None,
|
||
remark: str = "",
|
||
user_id: str = None,
|
||
) -> bool:
|
||
"""Java Controller TODO — 保持 pending，记录备注"""
|
||
entity = _get_change_row(db, change_id, tenant_no=tenant_no)
|
||
if not entity:
|
||
raise ValueError("变更记录不存在")
|
||
if entity.approval_status != "pending":
|
||
raise ValueError("只能提交待审批状态的变更记录")
|
||
if remark:
|
||
entity.remark = remark
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def withdraw_change(db: Session, change_id: str, *, tenant_no: str, user_id: str = None) -> bool:
|
||
entity = _get_change_row(db, change_id, tenant_no=tenant_no)
|
||
if not entity:
|
||
raise ValueError("变更记录不存在")
|
||
if entity.approval_status != "pending":
|
||
raise ValueError("只能撤回待审批状态的变更记录")
|
||
entity.del_flag = 1
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def execute_change(db: Session, change_id: str, *, tenant_no: str, user_id: str = None) -> bool:
|
||
entity = _get_change_row(db, change_id, tenant_no=tenant_no)
|
||
if not entity:
|
||
raise ValueError("变更记录不存在")
|
||
if entity.approval_status != "approved":
|
||
raise ValueError("只能执行已通过审批的变更")
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def export_contract_changes(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
filters: dict = None,
|
||
) -> List[Dict[str, Any]]:
|
||
rows, _ = list_contract_changes(
|
||
db,
|
||
tenant_no,
|
||
page=1,
|
||
page_size=10000,
|
||
filters=filters or {},
|
||
)
|
||
return _enrich_change_rows(db, rows)
|
||
|
||
|
||
def create_contract_template(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
data: dict,
|
||
user_id: str = None,
|
||
dept_id: str = "",
|
||
) -> Dict[str, Any]:
|
||
template_name = (_get_field(data, "template_name", "templateName") or "").strip()
|
||
if not template_name:
|
||
raise ValueError("模板名称不能为空")
|
||
|
||
template_no = (_get_field(data, "template_no", "templateNo") or "").strip()
|
||
if not template_no:
|
||
template_no = generate_template_no(db, tenant_no)
|
||
|
||
now = datetime.now()
|
||
entity = CrmContractTemplate(
|
||
id=generate_uuid(),
|
||
template_no=template_no,
|
||
template_name=template_name,
|
||
contract_type=_get_field(data, "contract_type", "contractType") or "",
|
||
template_content=_get_field(data, "template_content", "templateContent") or "",
|
||
payment_terms=_get_field(data, "payment_terms", "paymentTerms") or "",
|
||
delivery_terms=_get_field(data, "delivery_terms", "deliveryTerms") or "",
|
||
contract_terms=_get_field(data, "contract_terms", "contractTerms") or "",
|
||
party_b=_get_field(data, "party_b", "partyB") or "",
|
||
party_b_address=_get_field(data, "party_b_address", "partyBAddress") or "",
|
||
party_b_contact=_get_field(data, "party_b_contact", "partyBContact") or "",
|
||
party_b_mobile=_get_field(data, "party_b_mobile", "partyBMobile") or "",
|
||
default_currency=_get_field(data, "default_currency", "defaultCurrency") or "CNY",
|
||
default_tax_rate=_decimal_value(_get_field(data, "default_tax_rate", "defaultTaxRate"), Decimal("6.00")),
|
||
default_period=int(_get_field(data, "default_period", "defaultPeriod") or 12),
|
||
is_active=int(_get_field(data, "is_active", "isActive") if _get_field(data, "is_active", "isActive") is not None else 1),
|
||
usage_count=0,
|
||
sort=int(_get_field(data, "sort") or 0),
|
||
remark=_get_field(data, "remark") or "",
|
||
tenant_no=tenant_no,
|
||
create_by=user_id,
|
||
create_dept_id=dept_id or "",
|
||
create_time=now,
|
||
update_by=user_id,
|
||
update_time=now,
|
||
del_flag=0,
|
||
)
|
||
db.add(entity)
|
||
db.commit()
|
||
return get_contract_template_detail(db, entity.id, tenant_no=tenant_no)
|
||
|
||
|
||
def update_contract_template(
|
||
db: Session,
|
||
template_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
data: dict,
|
||
user_id: str = None,
|
||
user_info=None,
|
||
) -> Dict[str, Any]:
|
||
entity = (
|
||
db.query(CrmContractTemplate)
|
||
.filter(
|
||
CrmContractTemplate.id == template_id,
|
||
CrmContractTemplate.tenant_no == tenant_no,
|
||
CrmContractTemplate.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not entity:
|
||
raise ValueError("合同模板不存在")
|
||
ensure_crm_create_by_row_in_scope(
|
||
db, entity, user_info, tenant_no, message="无权访问该合同模板"
|
||
)
|
||
|
||
field_map = (
|
||
(("template_no", "templateNo"), "template_no"),
|
||
(("template_name", "templateName"), "template_name"),
|
||
(("contract_type", "contractType"), "contract_type"),
|
||
(("template_content", "templateContent"), "template_content"),
|
||
(("payment_terms", "paymentTerms"), "payment_terms"),
|
||
(("delivery_terms", "deliveryTerms"), "delivery_terms"),
|
||
(("contract_terms", "contractTerms"), "contract_terms"),
|
||
(("party_b", "partyB"), "party_b"),
|
||
(("party_b_address", "partyBAddress"), "party_b_address"),
|
||
(("party_b_contact", "partyBContact"), "party_b_contact"),
|
||
(("party_b_mobile", "partyBMobile"), "party_b_mobile"),
|
||
(("default_currency", "defaultCurrency"), "default_currency"),
|
||
(("remark",), "remark"),
|
||
)
|
||
for keys, attr in field_map:
|
||
val = _get_field(data, *keys)
|
||
if val is not None:
|
||
setattr(entity, attr, val)
|
||
if _get_field(data, "default_tax_rate", "defaultTaxRate") is not None:
|
||
entity.default_tax_rate = _decimal_value(_get_field(data, "default_tax_rate", "defaultTaxRate"))
|
||
if _get_field(data, "default_period", "defaultPeriod") is not None:
|
||
entity.default_period = int(_get_field(data, "default_period", "defaultPeriod"))
|
||
if _get_field(data, "is_active", "isActive") is not None:
|
||
entity.is_active = int(_get_field(data, "is_active", "isActive"))
|
||
if _get_field(data, "sort") is not None:
|
||
entity.sort = int(_get_field(data, "sort"))
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return get_contract_template_detail(db, template_id, tenant_no=tenant_no)
|
||
|
||
|
||
def soft_delete_contract_templates(
|
||
db: Session,
|
||
template_ids: List[str],
|
||
*,
|
||
tenant_no: str,
|
||
validate: bool = True,
|
||
user_info=None,
|
||
) -> int:
|
||
count = 0
|
||
for tid in template_ids:
|
||
entity = (
|
||
db.query(CrmContractTemplate)
|
||
.filter(
|
||
CrmContractTemplate.id == tid,
|
||
CrmContractTemplate.tenant_no == tenant_no,
|
||
CrmContractTemplate.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not entity:
|
||
continue
|
||
ensure_crm_create_by_row_in_scope(
|
||
db, entity, user_info, tenant_no, message="无权访问该合同模板"
|
||
)
|
||
if validate and (entity.usage_count or 0) > 0:
|
||
raise ValueError(f"模板【{entity.template_name}】已被使用，不能删除")
|
||
assert_row_deletable_by_creator(entity, user_info)
|
||
entity.del_flag = 1
|
||
entity.update_time = datetime.now()
|
||
count += 1
|
||
db.commit()
|
||
return count
|
||
|
||
|
||
def toggle_contract_template_active(
|
||
db: Session,
|
||
template_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
is_active: int,
|
||
user_id: str = None,
|
||
) -> bool:
|
||
entity = (
|
||
db.query(CrmContractTemplate)
|
||
.filter(
|
||
CrmContractTemplate.id == template_id,
|
||
CrmContractTemplate.tenant_no == tenant_no,
|
||
CrmContractTemplate.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not entity:
|
||
raise ValueError("合同模板不存在")
|
||
entity.is_active = 1 if int(is_active) == 1 else 0
|
||
entity.update_by = user_id
|
||
entity.update_time = datetime.now()
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def export_contract_templates(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
filters: dict = None,
|
||
) -> List[Dict[str, Any]]:
|
||
rows, _ = list_contract_templates(
|
||
db,
|
||
tenant_no,
|
||
page=1,
|
||
page_size=10000,
|
||
filters=filters or {},
|
||
)
|
||
return rows
|