"""CRM 报价单 CRUD — 对照 CrmQuoteServiceImpl.java（P2 全量）""" from __future__ import annotations import math import secrets 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 exists, func, or_ from sqlalchemy.orm import Query, Session, aliased from app.core.delete_guard import assert_row_deletable_by_creator from app.core.utils import generate_uuid from app.crud.crm.contact import get_contact from app.crud.crm.customer import get_customer from app.crud.crm.opportunity import get_opportunity from app.crud.crm.product import resolve_product_id_for_quote_item from app.services.crm_supplier_hazard import assert_products_allowed_for_supplier from app.crud.crm.quote_approval import ( can_user_approve, create_approval_records, delete_pending_approvals, ) from app.crud.crm.quote_product import ( batch_insert_quote_products, delete_products_by_quote_id, list_products_by_quote_ids, list_quote_products, ) from app.crud.crm.quote_share_log import record_visit from app.crud.crm.supplier_info import get_supplier_info from app.models.crm_models import CrmContract, CrmCustomer, CrmQuote, CrmQuoteApproval from app.models.models import SysUser from app.services.crm_quote_pdf import generate_quote_pdf from app.crud.crm.data_scope_helpers import apply_crm_owner_scope, ensure_crm_owner_row_in_scope _FMT = "%Y-%m-%d %H:%M:%S" _DATE_FMT = "%Y-%m-%d" def _ensure_quote_in_scope(db: Session, row: CrmQuote, user_info, tenant_no: str) -> None: ensure_crm_owner_row_in_scope(db, row, user_info, tenant_no, message="无权访问该报价单") def _get_quote_in_scope( db: Session, quote_id: str, *, tenant_no: str, user_info=None, ) -> CrmQuote: obj = get_quote(db, quote_id, tenant_no=tenant_no) if not obj: raise ValueError("报价单不存在") _ensure_quote_in_scope(db, obj, user_info, tenant_no) return obj 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 _row_to_dict(row: CrmQuote) -> Dict[str, Any]: """ORM → camelCase（单点真相）""" return { "id": row.id, "quoteNo": row.quote_no or "", "quoteTitle": row.quote_title or "", "quoteType": row.quote_type or "sales", "customerId": row.customer_id or "", "customerName": row.customer_name or "", "opportunityId": row.opportunity_id or "", "contactId": row.contact_id or "", "contactName": row.contact_name or "", "contactMobile": row.contact_mobile or "", "contactEmail": row.contact_email or "", "version": row.version or "V1.0", "parentId": row.parent_id or "0", "isLatest": row.is_latest if row.is_latest is not None else 1, "totalAmount": float(row.total_amount) if row.total_amount is not None else 0.0, "discountRate": float(row.discount_rate) if row.discount_rate is not None else 0.0, "discountAmount": float(row.discount_amount) if row.discount_amount is not None else 0.0, "finalAmount": float(row.final_amount) if row.final_amount is not None else 0.0, "validityDays": row.validity_days if row.validity_days is not None else 30, "quoteDate": row.quote_date.strftime(_DATE_FMT) if row.quote_date else "", "expireDate": row.expire_date.strftime(_DATE_FMT) if row.expire_date else "", "status": row.status or "draft", "rejectReason": row.reject_reason or "", "pdfUrl": row.pdf_url or "", "pdfGenerated": row.pdf_generated if row.pdf_generated is not None else 0, "shareLink": row.share_link or "", "shareCode": row.share_code or "", "sharePassword": row.share_password or "", "shareExpireTime": row.share_expire_time.strftime(_FMT) if row.share_expire_time else "", "viewCount": row.view_count if row.view_count is not None else 0, "lastViewTime": row.last_view_time.strftime(_FMT) if row.last_view_time else "", "ownerUserId": row.owner_user_id or "", "ownerUserName": row.owner_user_name or "", "supplierId": row.supplier_id or "", "supplierName": row.supplier_name or "", "contractId": row.contract_id or "", "paymentTerms": row.payment_terms or "", "deliveryTerms": row.delivery_terms or "", "freight": float(row.freight) if row.freight is not None else 0.0, "barrelDeposit": float(getattr(row, "barrel_deposit", None) or 0), "attachments": row.attachments or "", "templateContent": row.template_content or "", "createBy": row.create_by or "", "createTime": row.create_time.strftime(_FMT) if row.create_time else "", "updateTime": row.update_time.strftime(_FMT) if row.update_time else "", "remark": row.remark or "", } def quote_row_to_dict(row: CrmQuote) -> Dict[str, Any]: return _row_to_dict(row) def _build_quote_filters( db: Session, q: Query, filters: dict, *, tenant_no: str, ) -> Query: """Java: buildQueryWrapper""" q = q.filter(CrmQuote.tenant_no == tenant_no, CrmQuote.del_flag == 0) quote_no = (_get_field(filters, "quote_no", "quoteNo") or "").strip() if quote_no: q = q.filter(CrmQuote.quote_no.like(f"%{quote_no}%")) keyword = (_get_field(filters, "keyword") or "").strip() quote_title = (_get_field(filters, "quote_title", "quoteTitle") or "").strip() customer_name = (_get_field(filters, "customer_name", "customerName") or "").strip() if keyword: # 报价编号 / 报价主题 / 客户名称 / 客户名称拼音首字母 like = f"%{keyword}%" pinyin_like = f"%{keyword.lower()}%" customer_pinyin_match = exists().where( CrmCustomer.id == CrmQuote.customer_id, CrmCustomer.del_flag == 0, CrmCustomer.tenant_no == tenant_no, func.lower(func.coalesce(CrmCustomer.name_pinyin_abbr, "")).like(pinyin_like), ) q = q.filter( or_( CrmQuote.quote_no.like(like), CrmQuote.quote_title.like(like), CrmQuote.customer_name.like(like), customer_pinyin_match, ) ) elif quote_title and customer_name and quote_title == customer_name: q = q.filter( or_( CrmQuote.quote_title.like(f"%{quote_title}%"), CrmQuote.customer_name.like(f"%{customer_name}%"), ) ) else: if quote_title: q = q.filter(CrmQuote.quote_title.like(f"%{quote_title}%")) if customer_name: q = q.filter(CrmQuote.customer_name.like(f"%{customer_name}%")) customer_id = (_get_field(filters, "customer_id", "customerId") or "").strip() if customer_id: q = q.filter(CrmQuote.customer_id == customer_id) opportunity_id = (_get_field(filters, "opportunity_id", "opportunityId") or "").strip() if opportunity_id: q = q.filter(CrmQuote.opportunity_id == opportunity_id) status = (_get_field(filters, "status") or "").strip() if status: q = q.filter(CrmQuote.status == status) exclude_status = (_get_field(filters, "exclude_status", "excludeStatus") or "").strip() if exclude_status: q = q.filter(CrmQuote.status != exclude_status) quote_type = (_get_field(filters, "quote_type", "quoteType") or "").strip() if quote_type: q = q.filter(CrmQuote.quote_type == quote_type) owner_user_id = (_get_field(filters, "owner_user_id", "ownerUserId") or "").strip() if owner_user_id: q = q.filter(CrmQuote.owner_user_id == owner_user_id) is_latest = _get_field(filters, "is_latest", "isLatest") if is_latest is not None and is_latest != "": q = q.filter(CrmQuote.is_latest == int(is_latest)) min_amount = _get_field(filters, "min_amount", "minAmount") if min_amount is not None and min_amount != "": q = q.filter(CrmQuote.final_amount >= _decimal_value(min_amount)) max_amount = _get_field(filters, "max_amount", "maxAmount") if max_amount is not None and max_amount != "": q = q.filter(CrmQuote.final_amount <= _decimal_value(max_amount)) quote_date_begin = _parse_optional_date(_get_field(filters, "quote_date_begin", "quoteDateBegin")) if quote_date_begin: q = q.filter(CrmQuote.quote_date >= quote_date_begin) quote_date_end = _parse_optional_date(_get_field(filters, "quote_date_end", "quoteDateEnd")) if quote_date_end: q = q.filter(CrmQuote.quote_date <= quote_date_end) begin_time = _parse_optional_datetime(_get_field(filters, "begin_time", "beginTime")) end_time = _parse_optional_datetime(_get_field(filters, "end_time", "endTime")) if begin_time and end_time: q = q.filter(CrmQuote.create_time >= begin_time, CrmQuote.create_time <= end_time) elif begin_time: q = q.filter(CrmQuote.create_time >= begin_time) elif end_time: q = q.filter(CrmQuote.create_time <= end_time) pending_approval_user_id = ( _get_field(filters, "pending_approval_user_id", "pendingApprovalUserId") or "" ).strip() if pending_approval_user_id: # 待审批 Tab：仅返回当前轮到该用户审批的报价单（顺序审批，前序步骤均已通过） q = q.filter(CrmQuote.status == "submitted") user_approval = aliased(CrmQuoteApproval) prior_approval = aliased(CrmQuoteApproval) q = q.filter( exists().where( user_approval.quote_id == CrmQuote.id, user_approval.approver_id == pending_approval_user_id, user_approval.approval_result == "pending", user_approval.del_flag == 0, ~exists().where( prior_approval.quote_id == user_approval.quote_id, prior_approval.step < user_approval.step, prior_approval.approval_result != "approved", prior_approval.del_flag == 0, ), ) ) approved_by_user_id = (_get_field(filters, "approved_by_user_id", "approvedByUserId") or "").strip() if approved_by_user_id: q = q.filter( CrmQuote.id.in_( db.query(CrmQuoteApproval.quote_id) .filter( CrmQuoteApproval.approver_id == approved_by_user_id, CrmQuoteApproval.approval_result.in_(("approved", "rejected")), CrmQuoteApproval.del_flag == 0, ) .distinct() ) ) return q.order_by(CrmQuote.create_time.desc()) def _fill_contract_linked_count(db: Session, rows: List[Dict[str, Any]]) -> None: """Java: fillContractLinkedCount""" quote_ids = {str(d.get("id") or "").strip() for d in rows if d.get("id")} quote_ids.discard("") if not quote_ids: return contracts = ( db.query(CrmContract.quote_id) .filter( CrmContract.quote_id.in_(quote_ids), CrmContract.del_flag == 0, ) .all() ) count_map: Dict[str, int] = defaultdict(int) for (qid,) in contracts: if qid: count_map[qid] += 1 for d in rows: d["contractLinkedCount"] = count_map.get(str(d.get("id") or ""), 0) def calculate_quote_amount( entity: CrmQuote, products: List[dict], ) -> None: """Java: calculateAmount""" if not products: entity.total_amount = Decimal("0") entity.discount_amount = Decimal("0") entity.final_amount = Decimal("0") return total_amount = Decimal("0") for product in products: price = _decimal_value(_get_field(product, "price")) quantity = _decimal_value(_get_field(product, "quantity")) discount = _decimal_value(_get_field(product, "discount"), Decimal("0")) subtotal = price * quantity * (Decimal("1") - discount / Decimal("100")) total_amount += subtotal entity.total_amount = total_amount discount_rate = _decimal_value(entity.discount_rate, Decimal("0")) discount_amount = total_amount * discount_rate / Decimal("100") entity.discount_amount = discount_amount entity.final_amount = total_amount - discount_amount def _resolve_quote_product_ids( db: Session, *, tenant_no: str, products: List[dict], user_id: str = None, ) -> None: """Java: resolveQuoteProductIds""" if not products: return 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 generate_quote_no( db: Session, tenant_no: str, *, supplier_id: str = None, quote_date: date = None, ) -> str: """Java: generateQuoteNo — {供方编码}DD-yyyyMMdd + 3位序号""" 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 = quote_date or date.today() date_str = date_to_use.strftime("%Y%m%d") prefix = f"{supplier_code}DD-{date_str}" last = ( db.query(CrmQuote) .filter( CrmQuote.tenant_no == tenant_no, CrmQuote.del_flag == 0, CrmQuote.quote_no.like(f"{prefix}%"), CrmQuote.quote_date == date_to_use, ) .order_by(CrmQuote.quote_no.desc()) .first() ) sequence = 1 if last and last.quote_no and len(last.quote_no) >= len(prefix) + 3: try: sequence = int(last.quote_no[len(prefix) :]) + 1 except ValueError: sequence = 1 for retry in range(10): candidate = f"{prefix}{sequence + retry:03d}" exists = ( db.query(CrmQuote.id) .filter(CrmQuote.quote_no == candidate, CrmQuote.del_flag == 0) .first() ) if not exists: return candidate raise ValueError("生成报价单编号失败，请稍后重试") def generate_quote_title( quote_no: str, quote_type: str, supplier_short_name: str, customer_short_name: str, ) -> str: """Java: 自动生成报价主题""" if quote_type == "purchase": return f"{customer_short_name}-{supplier_short_name}（{quote_no}）" return f"{supplier_short_name}-{customer_short_name}（{quote_no}）" 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 _enrich_contact_fields(db: Session, entity: CrmQuote) -> None: if not entity.contact_id: return contact = get_contact(db, entity.contact_id) if contact: entity.contact_mobile = contact.mobile or entity.contact_mobile or "" entity.contact_email = contact.email or entity.contact_email or "" if not entity.contact_name: entity.contact_name = contact.contact_name or "" def _enrich_supplier_name(db: Session, entity: CrmQuote, tenant_no: str) -> None: if not entity.supplier_id: return supplier = get_supplier_info(db, entity.supplier_id, tenant_no=tenant_no) if supplier: entity.supplier_name = supplier.supplier_name or "" def get_quote( db: Session, quote_id: str, *, tenant_no: str = None, ) -> Optional[CrmQuote]: q = db.query(CrmQuote).filter(CrmQuote.id == quote_id, CrmQuote.del_flag == 0) if tenant_no: q = q.filter(CrmQuote.tenant_no == tenant_no) return q.first() def get_quote_detail( db: Session, quote_id: str, *, tenant_no: str = None, user_info=None, ) -> Dict[str, Any]: """Java: queryById""" obj = get_quote(db, quote_id, tenant_no=tenant_no) if not obj: raise ValueError("报价单不存在") _ensure_quote_in_scope(db, obj, user_info, tenant_no or obj.tenant_no) detail = _row_to_dict(obj) detail["productList"] = list_quote_products(db, quote_id, tenant_no=tenant_no or obj.tenant_no) if obj.opportunity_id: opp = get_opportunity(db, obj.opportunity_id, tenant_no=tenant_no or obj.tenant_no) if opp: detail["opportunityName"] = opp.get("opportunityName") or "" if obj.contact_id and (not detail.get("contactMobile") or not detail.get("contactEmail")): contact = get_contact(db, obj.contact_id) if contact: if not detail.get("contactMobile"): detail["contactMobile"] = contact.mobile or "" if not detail.get("contactEmail"): detail["contactEmail"] = contact.email or "" if obj.owner_user_id and not detail.get("ownerUserMobile"): owner = db.query(SysUser).filter(SysUser.id == obj.owner_user_id, SysUser.del_flag == 0).first() if owner and owner.phone_no: detail["ownerUserMobile"] = owner.phone_no if obj.supplier_id: supplier = get_supplier_info(db, obj.supplier_id, tenant_no=tenant_no or obj.tenant_no) if supplier: if supplier.supplier_name: detail["supplierName"] = supplier.supplier_name detail["supplierCode"] = supplier.supplier_code or "" detail["supplierShortName"] = supplier.short_name or "" return detail def list_quotes( db: Session, tenant_no: str, *, page: int = 1, page_size: int = 10, filters: dict = None, current_user_id: str = None, user_info=None, ) -> Tuple[List[Dict[str, Any]], int]: """Java: queryPageList""" q = db.query(CrmQuote) q = _build_quote_filters(db, q, filters or {}, tenant_no=tenant_no) if user_info: q = apply_crm_owner_scope(q, CrmQuote, user_info, db, tenant_no) total = q.order_by(None).with_entities(func.count(CrmQuote.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) 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") == "submitted" and current_user_id: item["canApprove"] = can_user_approve(db, item["id"], current_user_id) else: item["canApprove"] = False if result: quote_ids = [str(item["id"]) for item in result if item.get("id")] products_map = list_products_by_quote_ids(db, quote_ids, tenant_no=tenant_no) for item in result: item["productList"] = products_map.get(str(item.get("id") or ""), []) _fill_contract_linked_count(db, result) return result, int(total) def create_quote( 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] customer_id = (_get_field(data, "customer_id", "customerId") or "").strip() if not customer_id: raise ValueError("客户ID不能为空") customer = get_customer(db, customer_id) if customer and tenant_no and customer.tenant_no != tenant_no: raise ValueError("客户不存在") supplier_id = (_get_field(data, "supplier_id", "supplierId") or "").strip() or None supplier_short_name = _get_supplier_short_name(db, supplier_id or "", tenant_no) assert_products_allowed_for_supplier( db, tenant_no=tenant_no, supplier_id=str(supplier_id) if supplier_id else None, products=product_rows or [], ) quote_date = _parse_optional_date(_get_field(data, "quote_date", "quoteDate")) or date.today() quote_type = (_get_field(data, "quote_type", "quoteType") or "sales").strip() or "sales" validity_days = int(_get_field(data, "validity_days", "validityDays") or 30) quote_no = (_get_field(data, "quote_no", "quoteNo") or "").strip() if not quote_no or "***" in quote_no: quote_no = generate_quote_no(db, tenant_no, supplier_id=supplier_id, quote_date=quote_date) customer_short_name = (customer.short_name or "") if customer else "" customer_name = (customer.customer_name or "") if customer else (_get_field(data, "customer_name", "customerName") or "") quote_title = (_get_field(data, "quote_title", "quoteTitle") or "").strip() if not quote_title: quote_title = generate_quote_title(quote_no, quote_type, supplier_short_name, customer_short_name) owner_user_id = (_get_field(data, "owner_user_id", "ownerUserId") or user_id or "").strip() owner_user_name = _resolve_user_display_name(db, owner_user_id) if owner_user_id else "" now = datetime.now() entity = CrmQuote( id=generate_uuid(), quote_no=quote_no, quote_title=quote_title, quote_type=quote_type, customer_id=customer_id, customer_name=customer_name, opportunity_id=_get_field(data, "opportunity_id", "opportunityId") or None, contact_id=_get_field(data, "contact_id", "contactId") or None, contact_name=_get_field(data, "contact_name", "contactName") or "", version=_get_field(data, "version") or "V1.0", parent_id=_get_field(data, "parent_id", "parentId") or "0", is_latest=int(_get_field(data, "is_latest", "isLatest") if _get_field(data, "is_latest", "isLatest") is not None else 1), discount_rate=_decimal_value(_get_field(data, "discount_rate", "discountRate"), Decimal("0")), validity_days=validity_days, quote_date=quote_date, expire_date=quote_date + timedelta(days=validity_days), status=_get_field(data, "status") or "draft", owner_user_id=owner_user_id or user_id or "", owner_user_name=owner_user_name, supplier_id=supplier_id, supplier_name=_get_field(data, "supplier_name", "supplierName") or "", payment_terms=_get_field(data, "payment_terms", "paymentTerms") or "", delivery_terms=_get_field(data, "delivery_terms", "deliveryTerms") or "", freight=_decimal_value(_get_field(data, "freight"), Decimal("0")), barrel_deposit=_decimal_value(_get_field(data, "barrel_deposit", "barrelDeposit"), Decimal("0")), attachments=_get_field(data, "attachments") or "", tenant_no=tenant_no, order_num=_get_field(data, "order_num", "orderNum"), remark=_get_field(data, "remark") or "", create_by=user_id, create_dept_id=dept_id or "", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) _enrich_contact_fields(db, entity) _enrich_supplier_name(db, entity, tenant_no) _resolve_quote_product_ids(db, tenant_no=tenant_no, products=product_rows, user_id=user_id) calculate_quote_amount(entity, product_rows) db.add(entity) db.flush() if product_rows: batch_insert_quote_products( db, tenant_no=tenant_no, quote_id=entity.id, products=product_rows, user_id=user_id, dept_id=dept_id, ) db.commit() db.refresh(entity) return get_quote_detail(db, entity.id, tenant_no=tenant_no) # 允许编辑的报价单状态：草稿 / 已提交 / 审批通过 / 审批失败；保存后统一回草稿 QUOTE_EDITABLE_STATUSES = frozenset({"draft", "submitted", "accepted", "rejected"}) def assert_quote_editable_for_update(status: str) -> None: s = (status or "").strip() or "draft" if s not in QUOTE_EDITABLE_STATUSES: raise ValueError("仅草稿、已提交、审批通过或审批失败的报价单才能修改") def update_quote( db: Session, quote_id: str, *, tenant_no: str, data: dict, user_id: str = None, dept_id: str = "", user_info=None, ) -> Dict[str, Any]: """Java: updateByBo""" old = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not old: raise ValueError("报价单不存在") prev_status = (old.status or "").strip() or "draft" assert_quote_editable_for_update(prev_status) if prev_status == "submitted": delete_pending_approvals(db, quote_id) products = _get_field(data, "products", "productList") product_rows = [dict(p) for p in products] if products is not None else None quote_type = (_get_field(data, "quote_type", "quoteType") or old.quote_type or "sales").strip() old.quote_type = quote_type orig_customer_id = old.customer_id orig_supplier_id = old.supplier_id orig_quote_date = old.quote_date for field_map in ( (("quote_title", "quoteTitle"), "quote_title"), (("customer_id", "customerId"), "customer_id"), (("customer_name", "customerName"), "customer_name"), (("opportunity_id", "opportunityId"), "opportunity_id"), (("contact_id", "contactId"), "contact_id"), (("contact_name", "contactName"), "contact_name"), (("supplier_name", "supplierName"), "supplier_name"), (("payment_terms", "paymentTerms"), "payment_terms"), (("delivery_terms", "deliveryTerms"), "delivery_terms"), (("attachments",), "attachments"), (("remark",), "remark"), ): val = _get_field(data, *field_map[0]) if val is not None: setattr(old, field_map[1], val) if _get_field(data, "discount_rate", "discountRate") is not None: old.discount_rate = _decimal_value(_get_field(data, "discount_rate", "discountRate")) if _get_field(data, "validity_days", "validityDays") is not None: old.validity_days = int(_get_field(data, "validity_days", "validityDays")) if _get_field(data, "freight") is not None: old.freight = _decimal_value(_get_field(data, "freight")) if _get_field(data, "barrel_deposit", "barrelDeposit") is not None: old.barrel_deposit = _decimal_value(_get_field(data, "barrel_deposit", "barrelDeposit")) new_quote_date = _parse_optional_date(_get_field(data, "quote_date", "quoteDate")) quote_date_changed = new_quote_date is not None and new_quote_date != orig_quote_date if new_quote_date: old.quote_date = new_quote_date if _get_field(data, "supplier_id", "supplierId") is not None: old.supplier_id = (_get_field(data, "supplier_id", "supplierId") or "").strip() or None customer_changed = _get_field(data, "customer_id", "customerId") is not None and ( orig_customer_id != (_get_field(data, "customer_id", "customerId") or "").strip() ) supplier_changed = _get_field(data, "supplier_id", "supplierId") is not None and ( orig_supplier_id != old.supplier_id ) customer = get_customer(db, old.customer_id) if old.customer_id else None supplier_short_name = _get_supplier_short_name(db, old.supplier_id or "", tenant_no) if quote_date_changed and old.quote_date: old.quote_no = generate_quote_no( db, tenant_no, supplier_id=old.supplier_id, quote_date=old.quote_date ) customer_short_name = (customer.short_name or "") if customer else (old.customer_name or "") old.quote_title = generate_quote_title(old.quote_no, quote_type, supplier_short_name, customer_short_name) elif customer_changed or supplier_changed: quote_no = old.quote_no if supplier_changed and quote_no and "DD-" in quote_no: dd_index = quote_no.index("DD-") date_and_seq = quote_no[dd_index:] new_supplier_code = "SX" if old.supplier_id: supplier = get_supplier_info(db, old.supplier_id, tenant_no=tenant_no) if supplier and (supplier.supplier_code or "").strip(): new_supplier_code = supplier.supplier_code.strip() quote_no = f"{new_supplier_code}{date_and_seq}" old.quote_no = quote_no customer_short_name = (customer.short_name or "") if customer else "" current_quote_no = old.quote_no or quote_no old.quote_title = generate_quote_title(current_quote_no, quote_type, supplier_short_name, customer_short_name) if old.validity_days and old.quote_date: old.expire_date = old.quote_date + timedelta(days=int(old.validity_days)) _enrich_contact_fields(db, old) _enrich_supplier_name(db, old, tenant_no) if supplier_changed: old.template_content = None products_to_check = ( product_rows if product_rows is not None else list_quote_products(db, quote_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_quote_product_ids(db, tenant_no=tenant_no, products=product_rows, user_id=user_id) calculate_quote_amount(old, product_rows) else: calculate_quote_amount(old, list_quote_products(db, quote_id, tenant_no=tenant_no)) # 编辑保存后统一回草稿（含已提交 / 审批通过 / 审批失败） old.status = "draft" if hasattr(old, "reject_reason"): old.reject_reason = None old.update_by = user_id old.update_time = datetime.now() db.flush() if product_rows is not None: delete_products_by_quote_id(db, quote_id) if product_rows: batch_insert_quote_products( db, tenant_no=tenant_no, quote_id=quote_id, products=product_rows, user_id=user_id, dept_id=dept_id, ) db.commit() return get_quote_detail(db, quote_id, tenant_no=tenant_no) def soft_delete_quotes( db: Session, quote_ids: List[str], *, tenant_no: str, validate: bool = True, user_info=None, ) -> int: """Java: deleteWithValidByIds""" count = 0 for qid in quote_ids: entity = _get_quote_in_scope(db, qid, tenant_no=tenant_no, user_info=user_info) if not entity: continue if validate: if entity.status != "draft": raise ValueError("只能删除草稿状态的报价单") if entity.contract_id: 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_quote( db: Session, quote_id: str, *, tenant_no: str, approver_ids: List[str] = None, remark: str = "", user_id: str = None, dept_id: str = "", user_info=None, ) -> bool: """Java: submitQuote""" entity = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not entity: raise ValueError("报价单不存在") if entity.status != "draft": raise ValueError("只有草稿状态的报价单才能提交") entity.status = "submitted" entity.update_time = datetime.now() db.flush() if approver_ids: create_approval_records( db, tenant_no=tenant_no, quote_id=quote_id, approver_ids=approver_ids, user_id=user_id, dept_id=dept_id, ) try: detail = get_quote_detail(db, quote_id, tenant_no=tenant_no) pdf_url = generate_quote_pdf(detail, detail.get("productList") or []) entity.pdf_url = pdf_url entity.pdf_generated = 1 except Exception: pass db.commit() return True def save_template( db: Session, quote_id: str, template_content: str, *, tenant_no: str, user_info=None, ) -> bool: """Java: saveTemplate""" entity = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not entity: raise ValueError("报价单不存在") entity.template_content = template_content entity.update_time = datetime.now() db.commit() return True def get_template( db: Session, quote_id: str, *, tenant_no: str, user_info=None, ) -> str: """Java: getTemplate""" entity = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not entity: raise ValueError("报价单不存在") return entity.template_content or "" def generate_pdf( db: Session, quote_id: str, *, tenant_no: str, user_info=None, ) -> str: """Java: generatePdf""" entity = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not entity: raise ValueError("报价单不存在") detail = get_quote_detail(db, quote_id, tenant_no=tenant_no) pdf_url = generate_quote_pdf(detail, detail.get("productList") or []) entity.pdf_url = pdf_url entity.pdf_generated = 1 entity.update_time = datetime.now() db.commit() return pdf_url def generate_share( db: Session, quote_id: str, *, tenant_no: str, validity_days: int = None, password: str = "", user_info=None, ) -> str: """Java: generateShare""" entity = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not entity: raise ValueError("报价单不存在") share_code = secrets.token_hex(4)[:8] share_expire_time = None if validity_days and validity_days > 0: share_expire_time = datetime.now() + timedelta(days=validity_days) share_link = f"/share/{share_code}" entity.share_code = share_code entity.share_link = share_link entity.share_password = password or None entity.share_expire_time = share_expire_time entity.update_time = datetime.now() db.commit() return share_link def view_by_share_code( db: Session, share_code: str, *, password: str = "", ip_address: str = "", user_agent: str = "", ) -> Dict[str, Any]: """Java: viewByShareCode""" entity = ( db.query(CrmQuote) .filter(CrmQuote.share_code == share_code, CrmQuote.del_flag == 0) .first() ) if not entity: raise ValueError("分享链接无效") if entity.share_expire_time and datetime.now() > entity.share_expire_time: raise ValueError("分享链接已过期") if (entity.share_password or "").strip(): if not password or entity.share_password != password: raise ValueError("分享密码错误") if entity.expire_date and date.today() > entity.expire_date and entity.status != "expired": entity.status = "expired" entity.view_count = (entity.view_count or 0) + 1 entity.last_view_time = datetime.now() if entity.status in ("submitted", "accepted"): entity.status = "viewed" entity.update_time = datetime.now() db.flush() try: record_visit( db, tenant_no=entity.tenant_no, quote_id=entity.id, share_code=share_code, ip_address=ip_address, user_agent=user_agent, ) except Exception: pass db.commit() detail = _row_to_dict(entity) detail["productList"] = list_quote_products(db, entity.id, tenant_no=entity.tenant_no) return detail def accept_quote( db: Session, quote_id: str, *, tenant_no: str, remark: str = "", user_info=None, ) -> bool: """Java: acceptQuote""" entity = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not entity: raise ValueError("报价单不存在") entity.status = "accepted" if remark: entity.remark = remark entity.update_time = datetime.now() db.commit() return True def reject_quote( db: Session, quote_id: str, *, tenant_no: str, reject_reason: str = "", remark: str = "", user_info=None, ) -> bool: """Java: rejectQuote""" entity = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not entity: raise ValueError("报价单不存在") entity.status = "rejected" entity.reject_reason = reject_reason or "" if remark: entity.remark = remark entity.update_time = datetime.now() db.commit() return True def to_contract( db: Session, quote_id: str, *, tenant_no: str, contract_no: str = "", contract_title: str = "", sign_date: date = None, start_date: date = None, end_date: date = None, user_id: str = None, dept_id: str = "", user_info=None, ) -> str: """Java: toContract — 返回 contractId""" sign_date = _parse_optional_date(sign_date) start_date = _parse_optional_date(start_date) end_date = _parse_optional_date(end_date) quote = get_quote(db, quote_id, tenant_no=tenant_no) if not quote: raise ValueError("报价单不存在") if quote.status != "accepted": raise ValueError("只有已接受的报价单才能转合同") if quote.contract_id: raise ValueError("该报价单已转为合同,不能重复转换") products = list_quote_products(db, quote_id, tenant_no=tenant_no) if not products: raise ValueError("报价单没有产品明细,无法转合同") from app.crud.crm.contract import create_from_quote payload: Dict[str, Any] = { "quoteId": quote_id, "customerId": quote.customer_id, "opportunityId": quote.opportunity_id, "amount": float(quote.final_amount or quote.total_amount or 0), "ownerUserId": quote.owner_user_id or user_id or "", "supplierId": quote.supplier_id, "paymentTerms": quote.payment_terms, "deliveryTerms": quote.delivery_terms, } if contract_no: payload["contractNo"] = contract_no if contract_title: payload["contractName"] = contract_title if sign_date: payload["signDate"] = sign_date.strftime(_DATE_FMT) if start_date: payload["startDate"] = start_date.strftime(_DATE_FMT) if end_date: payload["endDate"] = end_date.strftime(_DATE_FMT) qtype = (quote.quote_type or "sales").strip() payload["contractType"] = "purchase" if qtype == "purchase" else "sale" return create_from_quote( db, quote_id, tenant_no=tenant_no, data=payload, user_id=user_id, dept_id=dept_id, ) def create_new_version( db: Session, quote_id: str, *, tenant_no: str, version: str, remark: str = "", user_id: str = None, dept_id: str = "", user_info=None, ) -> str: """Java: createNewVersion — 返回新报价单 id""" old = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not old: raise ValueError("报价单不存在") validity_days = old.validity_days or 30 quote_date = date.today() now = datetime.now() new_entity = CrmQuote( id=generate_uuid(), quote_no=old.quote_no, quote_title=old.quote_title, quote_type=old.quote_type or "sales", customer_id=old.customer_id, customer_name=old.customer_name, opportunity_id=old.opportunity_id, contact_id=old.contact_id, contact_name=old.contact_name, contact_mobile=old.contact_mobile, contact_email=old.contact_email, version=version, parent_id=quote_id, is_latest=1, discount_rate=old.discount_rate, validity_days=validity_days, quote_date=quote_date, expire_date=quote_date + timedelta(days=validity_days), status="draft", owner_user_id=old.owner_user_id, owner_user_name=old.owner_user_name, supplier_id=old.supplier_id, supplier_name=old.supplier_name, payment_terms=old.payment_terms, delivery_terms=old.delivery_terms, freight=old.freight, barrel_deposit=getattr(old, "barrel_deposit", None) or Decimal("0"), tenant_no=tenant_no, remark=remark or old.remark, create_by=user_id, create_dept_id=dept_id or "", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) old.is_latest = 0 old.update_time = now old_products = list_quote_products(db, quote_id, tenant_no=tenant_no) calculate_quote_amount(new_entity, old_products) db.add(new_entity) db.flush() if old_products: batch_insert_quote_products( db, tenant_no=tenant_no, quote_id=new_entity.id, products=old_products, user_id=user_id, dept_id=dept_id, ) db.commit() return new_entity.id def query_versions( db: Session, quote_no: str, *, tenant_no: str, user_info=None, ) -> List[Dict[str, Any]]: """Java: queryVersions""" rows = ( db.query(CrmQuote) .filter( CrmQuote.quote_no == quote_no, CrmQuote.tenant_no == tenant_no, CrmQuote.del_flag == 0, ) .order_by(CrmQuote.create_time.desc()) .all() ) return [_row_to_dict(r) for r in rows] def withdraw_quote_approval( db: Session, quote_id: str, *, tenant_no: str, remark: str = "", user_id: str = None, user_info=None, ) -> bool: """Java: CrmQuoteServiceImpl.withdrawApproval（校验创建人）""" entity = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not entity: raise ValueError("报价单不存在") if entity.status != "submitted": raise ValueError("只能撤回已提交的报价单") if user_id and entity.create_by and entity.create_by != user_id: raise ValueError("只有创建人才能撤回审批") entity.status = "draft" entity.update_time = datetime.now() if remark: entity.remark = remark delete_pending_approvals(db, quote_id) db.commit() return True def invalid_quote( db: Session, quote_id: str, *, tenant_no: str, remark: str = "", user_info=None, ) -> bool: """Java: invalidQuote""" entity = _get_quote_in_scope(db, quote_id, tenant_no=tenant_no, user_info=user_info) if not entity: raise ValueError("报价单不存在") if entity.status == "invalid": raise ValueError("报价单已作废，不能重复作废") entity.status = "invalid" if remark: current = entity.remark or "" entity.remark = f"{current}\n【作废原因】{remark}".strip() if current else f"【作废原因】{remark}" 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]: """Java: statisticsOverview""" q = db.query(CrmQuote).filter(CrmQuote.tenant_no == tenant_no, CrmQuote.del_flag == 0) if owner_user_id: q = q.filter(CrmQuote.owner_user_id == owner_user_id) begin_dt = _parse_optional_datetime(begin_time) end_dt = _parse_optional_datetime(end_time) if begin_dt: q = q.filter(CrmQuote.create_time >= begin_dt) if end_dt: q = q.filter(CrmQuote.create_time <= end_dt) if user_info: q = apply_crm_owner_scope(q, CrmQuote, user_info, db, tenant_no) all_quotes = q.all() total_count = len(all_quotes) def _count_status(status: str) -> int: return sum(1 for item in all_quotes if item.status == status) draft_count = _count_status("draft") submitted_count = _count_status("submitted") accepted_count = _count_status("accepted") rejected_count = _count_status("rejected") expired_count = _count_status("expired") total_amount = sum(_decimal_value(item.final_amount) for item in all_quotes) accepted_amount = sum( _decimal_value(item.final_amount) for item in all_quotes if item.status == "accepted" ) accept_rate = (accepted_count * 100.0 / total_count) if total_count else 0.0 avg_amount = ( (total_amount / Decimal(total_count)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) if total_count else Decimal("0") ) return { "totalCount": total_count, "draftCount": draft_count, "submittedCount": submitted_count, "acceptedCount": accepted_count, "rejectedCount": rejected_count, "expiredCount": expired_count, "totalAmount": float(total_amount), "acceptedAmount": float(accepted_amount), "acceptRate": f"{accept_rate:.2f}", "avgAmount": float(avg_amount), } def statistics_status( db: Session, tenant_no: str, user_info=None, ) -> List[Dict[str, Any]]: """Java: statisticsStatus""" status_array = ("draft", "submitted", "accepted", "rejected", "expired", "contract") status_names = ("草稿", "已提交", "已接受", "已拒绝", "已过期", "已转合同") base_q = db.query(CrmQuote).filter(CrmQuote.tenant_no == tenant_no, CrmQuote.del_flag == 0) if user_info: base_q = apply_crm_owner_scope(base_q, CrmQuote, user_info, db, tenant_no) total = base_q.count() result = [] for status, status_name in zip(status_array, status_names): q = base_q.filter(CrmQuote.status == status) rows = q.all() count = len(rows) amount = sum(_decimal_value(r.final_amount) for r in rows) percent = (count * 100.0 / total) if total else 0.0 result.append( { "status": status, "statusName": status_name, "count": count, "amount": float(amount), "percent": f"{percent:.2f}", } ) return result def _period_key(dt: datetime, period_type: str) -> str: if period_type.lower() == "year": return dt.strftime("%Y年") if period_type.lower() == "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]]: """Java: statisticsTrend""" q = db.query(CrmQuote).filter(CrmQuote.tenant_no == tenant_no, CrmQuote.del_flag == 0) begin_dt = _parse_optional_datetime(begin_time) end_dt = _parse_optional_datetime(end_time) if begin_dt: q = q.filter(CrmQuote.create_time >= begin_dt) if end_dt: q = q.filter(CrmQuote.create_time <= end_dt) if user_info: q = apply_crm_owner_scope(q, CrmQuote, user_info, db, tenant_no) all_quotes = q.all() if not all_quotes: return [] grouped: Dict[str, List[CrmQuote]] = defaultdict(list) for quote in all_quotes: create_time = quote.create_time or datetime.now() if quote.create_time is None and quote.quote_date: create_time = datetime.combine(quote.quote_date, datetime.min.time()) grouped[_period_key(create_time, period_type)].append(quote) result = [] for period, period_quotes in grouped.items(): accepted_count = sum( 1 for item in period_quotes if item.status in ("accepted", "contract") ) total_amount = sum(_decimal_value(item.final_amount) for item in period_quotes) accepted_amount = sum( _decimal_value(item.final_amount) for item in period_quotes if item.status in ("accepted", "contract") ) accept_rate = (accepted_count * 100.0 / len(period_quotes)) if period_quotes else 0.0 result.append( { "period": period, "totalCount": len(period_quotes), "acceptedCount": accepted_count, "totalAmount": float(total_amount), "acceptedAmount": float(accepted_amount), "acceptRate": f"{accept_rate:.2f}", } ) return result