801 lines
24 KiB
Python
801 lines
24 KiB
Python
"""CRM 产品模块 CRUD（产品分类 + 产品 + 附件）"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import datetime
|
||
from decimal import Decimal
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from sqlalchemy import and_, func, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.utils import generate_uuid
|
||
from app.models.crm_models import CrmProduct, CrmProductCategory, SysOss
|
||
|
||
|
||
# ── 工具函数 ──────────────────────────────────────────────────
|
||
|
||
|
||
def total_pages(total: int, page_size: int) -> int:
|
||
"""计算总页数"""
|
||
if page_size <= 0:
|
||
return 0
|
||
return math.ceil(total / page_size)
|
||
|
||
|
||
# ── 产品分类 CRUD ────────────────────────────────────────────
|
||
|
||
|
||
def _category_row_to_dict(row: CrmProductCategory) -> Dict[str, Any]:
|
||
"""分类 ORM 对象转 dict（camelCase 键名）"""
|
||
return {
|
||
"id": row.id,
|
||
"parentId": row.parent_id,
|
||
"categoryName": row.category_name,
|
||
"categoryCode": row.category_code or "",
|
||
"level": row.level or 1,
|
||
"status": row.status,
|
||
"orderNum": row.order_num if row.order_num is not None else 0,
|
||
"remark": row.remark or "",
|
||
"createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
|
||
"updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
|
||
}
|
||
|
||
|
||
def list_categories(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
category_name: Optional[str] = None,
|
||
status: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""获取分类（扁平列表，支持名称模糊与状态筛选）"""
|
||
q = db.query(CrmProductCategory).filter(
|
||
CrmProductCategory.tenant_no == tenant_no,
|
||
CrmProductCategory.del_flag == 0,
|
||
)
|
||
name_kw = (category_name or "").strip()
|
||
if name_kw:
|
||
q = q.filter(CrmProductCategory.category_name.like(f"%{name_kw}%"))
|
||
if status is not None and str(status).strip() != "":
|
||
q = q.filter(CrmProductCategory.status == str(status).strip())
|
||
rows = q.order_by(
|
||
CrmProductCategory.order_num.asc(),
|
||
CrmProductCategory.create_time.asc(),
|
||
).all()
|
||
return [_category_row_to_dict(r) for r in rows]
|
||
|
||
|
||
def _build_category_tree(
|
||
items: List[Dict[str, Any]], parent_id: str = "0"
|
||
) -> List[Dict[str, Any]]:
|
||
"""按 parentId 将扁平分类列表组装为嵌套树"""
|
||
children = [item for item in items if (item.get("parentId") or "0") == parent_id]
|
||
for child in children:
|
||
child["children"] = _build_category_tree(items, child["id"])
|
||
children.sort(key=lambda x: (x.get("orderNum", 0) or 0))
|
||
return children
|
||
|
||
|
||
def _collect_category_with_ancestors(
|
||
items: List[Dict[str, Any]], seed_ids: set[str]
|
||
) -> List[Dict[str, Any]]:
|
||
"""保留匹配节点及其全部上级，供检索后仍能正确组树。"""
|
||
by_id = {item["id"]: item for item in items if item.get("id")}
|
||
keep: set[str] = set()
|
||
for sid in seed_ids:
|
||
cur = (sid or "").strip()
|
||
while cur and cur != "0":
|
||
if cur not in by_id:
|
||
break
|
||
keep.add(cur)
|
||
cur = (by_id[cur].get("parentId") or "0").strip()
|
||
return [by_id[i] for i in keep if i in by_id]
|
||
|
||
|
||
def list_category_tree(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
category_name: Optional[str] = None,
|
||
status: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""获取产品分类树（嵌套 children，供树形表格/树选择器使用）"""
|
||
name_kw = (category_name or "").strip()
|
||
status_val = str(status).strip() if status is not None else ""
|
||
has_filter = bool(name_kw) or bool(status_val)
|
||
|
||
if not has_filter:
|
||
flat = list_categories(db, tenant_no)
|
||
return _build_category_tree(flat)
|
||
|
||
matched = list_categories(
|
||
db,
|
||
tenant_no,
|
||
category_name=category_name,
|
||
status=status,
|
||
)
|
||
if not matched:
|
||
return []
|
||
|
||
all_flat = list_categories(db, tenant_no)
|
||
seed_ids = {m["id"] for m in matched if m.get("id")}
|
||
expanded = _collect_category_with_ancestors(all_flat, seed_ids)
|
||
return _build_category_tree(expanded)
|
||
|
||
|
||
def get_category(db: Session, category_id: str) -> Optional[CrmProductCategory]:
|
||
"""根据ID获取分类"""
|
||
return (
|
||
db.query(CrmProductCategory)
|
||
.filter(
|
||
CrmProductCategory.id == category_id,
|
||
CrmProductCategory.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def create_category(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
user_id: str = None,
|
||
**kwargs,
|
||
) -> CrmProductCategory:
|
||
"""创建分类"""
|
||
parent_id = kwargs.get("parent_id", "0") or "0"
|
||
level = kwargs.get("level") or 1
|
||
|
||
# 自动计算 level
|
||
if parent_id != "0":
|
||
parent = get_category(db, parent_id)
|
||
if parent:
|
||
level = (parent.level or 0) + 1
|
||
|
||
obj = CrmProductCategory(
|
||
id=generate_uuid(),
|
||
parent_id=parent_id,
|
||
category_name=kwargs.get("category_name"),
|
||
category_code=kwargs.get("category_code"),
|
||
level=level,
|
||
status=kwargs.get("status", "1"),
|
||
tenant_no=tenant_no,
|
||
order_num=kwargs.get("order_num"),
|
||
remark=kwargs.get("remark"),
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def update_category(
|
||
db: Session,
|
||
category_id: str,
|
||
*,
|
||
user_id: str = None,
|
||
**kwargs,
|
||
) -> CrmProductCategory:
|
||
"""更新分类"""
|
||
obj = get_category(db, category_id)
|
||
if not obj:
|
||
raise ValueError("分类不存在")
|
||
|
||
# 逐字段更新
|
||
for field in ("category_name", "category_code", "status", "order_num", "remark"):
|
||
if field in kwargs and kwargs[field] is not None:
|
||
setattr(obj, field, kwargs[field])
|
||
|
||
# parent_id 变了则重新计算 level
|
||
if "parent_id" in kwargs and kwargs["parent_id"] is not None:
|
||
new_parent_id = kwargs["parent_id"]
|
||
obj.parent_id = new_parent_id
|
||
if new_parent_id == "0":
|
||
obj.level = 1
|
||
else:
|
||
parent = get_category(db, new_parent_id)
|
||
if parent:
|
||
obj.level = (parent.level or 0) + 1
|
||
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_category(db: Session, category_id: str) -> None:
|
||
"""软删除分类"""
|
||
obj = get_category(db, category_id)
|
||
if not obj:
|
||
raise ValueError("分类不存在")
|
||
|
||
# 检查是否有子分类
|
||
has_children = (
|
||
db.query(CrmProductCategory)
|
||
.filter(
|
||
CrmProductCategory.parent_id == category_id,
|
||
CrmProductCategory.del_flag == 0,
|
||
)
|
||
.first()
|
||
is not None
|
||
)
|
||
if has_children:
|
||
raise ValueError("该分类下存在子分类/产品，无法删除")
|
||
|
||
# 检查是否有产品引用
|
||
has_products = (
|
||
db.query(CrmProduct)
|
||
.filter(
|
||
CrmProduct.category_id == category_id,
|
||
CrmProduct.del_flag == 0,
|
||
)
|
||
.first()
|
||
is not None
|
||
)
|
||
if has_products:
|
||
raise ValueError("该分类下存在子分类/产品，无法删除")
|
||
|
||
obj.del_flag = 1
|
||
db.commit()
|
||
|
||
|
||
def category_code_exists(
|
||
db: Session,
|
||
category_code: str,
|
||
tenant_no: str,
|
||
exclude_id: str = None,
|
||
) -> bool:
|
||
"""检查分类编码是否已存在"""
|
||
q = db.query(CrmProductCategory).filter(
|
||
CrmProductCategory.category_code == category_code,
|
||
CrmProductCategory.tenant_no == tenant_no,
|
||
CrmProductCategory.del_flag == 0,
|
||
)
|
||
if exclude_id:
|
||
q = q.filter(CrmProductCategory.id != exclude_id)
|
||
return q.first() is not None
|
||
|
||
|
||
# ── 产品 CRUD ────────────────────────────────────────────────
|
||
|
||
|
||
def _product_row_to_dict(row: CrmProduct) -> Dict[str, Any]:
|
||
"""产品 ORM 对象转 dict（camelCase 键名）"""
|
||
return {
|
||
"id": row.id,
|
||
"productNo": row.product_no,
|
||
"productName": row.product_name,
|
||
"categoryId": row.category_id or "",
|
||
"categoryName": row.category_name or "",
|
||
"model": row.model or "",
|
||
"unit": row.unit,
|
||
"barcode": row.barcode or "",
|
||
"price": float(row.price) if row.price else 0.0,
|
||
"costPrice": float(row.cost_price) if row.cost_price else 0.0,
|
||
"cost": float(row.cost_price) if row.cost_price else 0.0,
|
||
"stock": row.stock if row.stock is not None else 0,
|
||
"lowStock": row.low_stock if row.low_stock is not None else 10,
|
||
"taxRate": float(row.tax_rate) if row.tax_rate is not None else 13.0,
|
||
"isToxic": int(row.is_toxic or 0),
|
||
"isExplosive": int(row.is_explosive or 0),
|
||
"status": row.status,
|
||
"description": row.description or "",
|
||
"images": row.images or "",
|
||
"saleCount": row.sale_count if row.sale_count is not None else 0,
|
||
"orderNum": row.order_num if row.order_num is not None else 0,
|
||
"remark": row.remark or "",
|
||
"createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
|
||
"updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
|
||
}
|
||
|
||
|
||
def list_products(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: str = None,
|
||
category_id: str = None,
|
||
status: str = None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
"""分页查询产品列表"""
|
||
q = db.query(CrmProduct).filter(
|
||
CrmProduct.tenant_no == tenant_no,
|
||
CrmProduct.del_flag == 0,
|
||
)
|
||
|
||
# keyword 筛选
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
q = q.filter(or_(CrmProduct.product_name.like(like), CrmProduct.product_no.like(like)))
|
||
|
||
# category_id 筛选
|
||
if category_id:
|
||
q = q.filter(CrmProduct.category_id == category_id)
|
||
|
||
# status 筛选
|
||
if status:
|
||
q = q.filter(CrmProduct.status == status)
|
||
|
||
total = q.order_by(None).with_entities(func.count(CrmProduct.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
|
||
rows = (
|
||
q.order_by(CrmProduct.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
return [_product_row_to_dict(r) for r in rows], int(total)
|
||
|
||
|
||
def get_product(db: Session, product_id: str) -> Optional[CrmProduct]:
|
||
"""根据ID获取产品"""
|
||
return (
|
||
db.query(CrmProduct)
|
||
.filter(
|
||
CrmProduct.id == product_id,
|
||
CrmProduct.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def get_product_by_no(db: Session, product_no: str, tenant_no: str) -> Optional[CrmProduct]:
|
||
"""根据编号获取产品"""
|
||
return (
|
||
db.query(CrmProduct)
|
||
.filter(
|
||
CrmProduct.product_no == product_no,
|
||
CrmProduct.tenant_no == tenant_no,
|
||
CrmProduct.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def generate_product_no(db: Session, tenant_no: str) -> str:
|
||
"""生成产品编号（格式：CP + yyyyMMdd + 3位序号，如 CP20260530001）"""
|
||
today_str = datetime.now().strftime("%Y%m%d")
|
||
prefix = f"CP{today_str}"
|
||
|
||
# 查询今天已有的最大编号
|
||
max_no = (
|
||
db.query(CrmProduct.product_no)
|
||
.filter(
|
||
CrmProduct.product_no.like(f"{prefix}%"),
|
||
CrmProduct.tenant_no == tenant_no,
|
||
)
|
||
.order_by(CrmProduct.product_no.desc())
|
||
.first()
|
||
)
|
||
|
||
if max_no and max_no[0]:
|
||
seq_str = max_no[0][len(prefix):]
|
||
try:
|
||
seq = int(seq_str) + 1
|
||
except ValueError:
|
||
seq = 1
|
||
else:
|
||
seq = 1
|
||
|
||
return f"{prefix}{seq:03d}"
|
||
|
||
|
||
def generate_barcode(db: Session, tenant_no: str) -> str:
|
||
"""生成唯一条码（纯数字：yyyyMMdd + 6位序号，如 20260606000001）"""
|
||
today_str = datetime.now().strftime("%Y%m%d")
|
||
prefix = today_str
|
||
|
||
max_barcode = (
|
||
db.query(CrmProduct.barcode)
|
||
.filter(
|
||
CrmProduct.barcode.like(f"{prefix}%"),
|
||
CrmProduct.tenant_no == tenant_no,
|
||
CrmProduct.del_flag == 0,
|
||
)
|
||
.order_by(CrmProduct.barcode.desc())
|
||
.first()
|
||
)
|
||
|
||
if max_barcode and max_barcode[0]:
|
||
seq_str = max_barcode[0][len(prefix):]
|
||
try:
|
||
seq = int(seq_str) + 1
|
||
except ValueError:
|
||
seq = 1
|
||
else:
|
||
seq = 1
|
||
|
||
return f"{prefix}{seq:06d}"
|
||
|
||
|
||
def create_product(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
user_id: str = None,
|
||
**kwargs,
|
||
) -> CrmProduct:
|
||
"""创建产品"""
|
||
product_no = kwargs.get("product_no")
|
||
if not product_no:
|
||
product_no = generate_product_no(db, tenant_no)
|
||
|
||
barcode = _normalize_text(kwargs.get("barcode"))
|
||
if not barcode:
|
||
barcode = generate_barcode(db, tenant_no)
|
||
elif barcode_exists(db, barcode, tenant_no):
|
||
raise ValueError(f"条码 {barcode} 已存在")
|
||
|
||
# 如果传了 category_id，查分类表冗余写入 category_name
|
||
category_id = kwargs.get("category_id")
|
||
category_name = kwargs.get("category_name")
|
||
if category_id and not category_name:
|
||
cat = get_category(db, category_id)
|
||
if cat:
|
||
category_name = cat.category_name
|
||
|
||
obj = CrmProduct(
|
||
id=generate_uuid(),
|
||
product_no=product_no,
|
||
product_name=kwargs.get("product_name"),
|
||
category_id=category_id,
|
||
category_name=category_name or "",
|
||
model=kwargs.get("model"),
|
||
unit=kwargs.get("unit", "件"),
|
||
barcode=barcode,
|
||
price=kwargs.get("price", 0),
|
||
cost_price=kwargs.get("cost_price"),
|
||
stock=kwargs.get("stock", 0),
|
||
low_stock=kwargs.get("low_stock", 10),
|
||
tax_rate=kwargs.get("tax_rate", Decimal("13.00")),
|
||
is_toxic=int(kwargs.get("is_toxic") or 0),
|
||
is_explosive=int(kwargs.get("is_explosive") or 0),
|
||
status=kwargs.get("status", "on"),
|
||
description=kwargs.get("description"),
|
||
images=kwargs.get("images"),
|
||
sale_count=kwargs.get("sale_count", 0),
|
||
tenant_no=tenant_no,
|
||
order_num=kwargs.get("order_num"),
|
||
remark=kwargs.get("remark"),
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def update_product(
|
||
db: Session,
|
||
product_id: str,
|
||
*,
|
||
user_id: str = None,
|
||
**kwargs,
|
||
) -> CrmProduct:
|
||
"""更新产品"""
|
||
obj = get_product(db, product_id)
|
||
if not obj:
|
||
raise ValueError("产品不存在")
|
||
|
||
# 逐字段更新
|
||
for field in (
|
||
"product_no",
|
||
"product_name",
|
||
"model",
|
||
"unit",
|
||
"price",
|
||
"cost_price",
|
||
"stock",
|
||
"low_stock",
|
||
"tax_rate",
|
||
"is_toxic",
|
||
"is_explosive",
|
||
"status",
|
||
"description",
|
||
"images",
|
||
"sale_count",
|
||
"order_num",
|
||
"remark",
|
||
):
|
||
if field in kwargs and kwargs[field] is not None:
|
||
setattr(obj, field, kwargs[field])
|
||
|
||
if "barcode" in kwargs:
|
||
barcode = _normalize_text(kwargs.get("barcode"))
|
||
if barcode and barcode_exists(db, barcode, tenant_no=obj.tenant_no, exclude_id=product_id):
|
||
raise ValueError(f"条码 {barcode} 已存在")
|
||
obj.barcode = barcode or None
|
||
|
||
# category_id 变了则同步更新 category_name
|
||
if "category_id" in kwargs and kwargs["category_id"] is not None:
|
||
cat_id = kwargs["category_id"]
|
||
obj.category_id = cat_id
|
||
if cat_id:
|
||
cat = get_category(db, cat_id)
|
||
obj.category_name = cat.category_name if cat else ""
|
||
else:
|
||
obj.category_name = ""
|
||
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_product(db: Session, product_id: str) -> None:
|
||
"""软删除产品"""
|
||
obj = get_product(db, product_id)
|
||
if not obj:
|
||
raise ValueError("产品不存在")
|
||
obj.del_flag = 1
|
||
db.commit()
|
||
|
||
|
||
def product_no_exists(
|
||
db: Session,
|
||
product_no: str,
|
||
tenant_no: str,
|
||
exclude_id: str = None,
|
||
) -> bool:
|
||
"""检查产品编号是否已存在"""
|
||
q = db.query(CrmProduct).filter(
|
||
CrmProduct.product_no == product_no,
|
||
CrmProduct.tenant_no == tenant_no,
|
||
CrmProduct.del_flag == 0,
|
||
)
|
||
if exclude_id:
|
||
q = q.filter(CrmProduct.id != exclude_id)
|
||
return q.first() is not None
|
||
|
||
|
||
def barcode_exists(
|
||
db: Session,
|
||
barcode: str,
|
||
tenant_no: str,
|
||
exclude_id: str = None,
|
||
) -> bool:
|
||
"""检查条码是否已存在"""
|
||
q = db.query(CrmProduct).filter(
|
||
CrmProduct.barcode == barcode,
|
||
CrmProduct.tenant_no == tenant_no,
|
||
CrmProduct.del_flag == 0,
|
||
)
|
||
if exclude_id:
|
||
q = q.filter(CrmProduct.id != exclude_id)
|
||
return q.first() is not None
|
||
|
||
|
||
def list_product_options(
|
||
db: Session,
|
||
tenant_no: str,
|
||
*,
|
||
keyword: str = None,
|
||
category_id: str = None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""获取产品下拉选项（不分页，返回全量）"""
|
||
q = db.query(CrmProduct).filter(
|
||
CrmProduct.tenant_no == tenant_no,
|
||
CrmProduct.del_flag == 0,
|
||
CrmProduct.status == "on",
|
||
)
|
||
if keyword:
|
||
like = f"%{keyword}%"
|
||
q = q.filter(or_(CrmProduct.product_name.like(like), CrmProduct.product_no.like(like)))
|
||
if category_id:
|
||
q = q.filter(CrmProduct.category_id == category_id)
|
||
rows = q.order_by(CrmProduct.create_time.desc()).all()
|
||
return [
|
||
{
|
||
"id": r.id,
|
||
"label": r.product_name,
|
||
"value": r.id,
|
||
"productNo": r.product_no,
|
||
"price": float(r.price) if r.price else 0.0,
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
# ── 附件 CRUD ────────────────────────────────────────────────
|
||
|
||
|
||
def _oss_row_to_dict(row: SysOss) -> Dict[str, Any]:
|
||
"""附件 ORM 对象转 dict"""
|
||
return {
|
||
"id": row.id,
|
||
"bizType": row.biz_type or "",
|
||
"bizId": row.biz_id or "",
|
||
"fileName": row.file_name or "",
|
||
"originalName": row.original_name or "",
|
||
"fileSuffix": row.file_suffix or "",
|
||
"url": row.url or "",
|
||
"fileSize": row.file_size or 0,
|
||
"fileType": row.file_type or "",
|
||
"attachType": row.attach_type or "attachment",
|
||
"sortOrder": row.sort_order or 0,
|
||
"createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
|
||
}
|
||
|
||
|
||
def list_oss_by_biz(db: Session, biz_type: str, biz_id: str, tenant_no: str) -> List[Dict[str, Any]]:
|
||
"""按业务类型+ID 查询附件列表（不分页）"""
|
||
rows = (
|
||
db.query(SysOss)
|
||
.filter(
|
||
SysOss.biz_type == biz_type,
|
||
SysOss.biz_id == biz_id,
|
||
SysOss.tenant_no == tenant_no,
|
||
SysOss.del_flag == 0,
|
||
)
|
||
.order_by(SysOss.sort_order.asc())
|
||
.all()
|
||
)
|
||
return [_oss_row_to_dict(r) for r in rows]
|
||
|
||
|
||
def create_oss(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
user_id: str = None,
|
||
**kwargs,
|
||
) -> SysOss:
|
||
"""创建附件记录"""
|
||
obj = SysOss(
|
||
id=generate_uuid(),
|
||
tenant_no=tenant_no,
|
||
biz_type=kwargs.get("biz_type", ""),
|
||
biz_id=kwargs.get("biz_id", ""),
|
||
file_name=kwargs.get("file_name", ""),
|
||
original_name=kwargs.get("original_name", ""),
|
||
file_suffix=kwargs.get("file_suffix", ""),
|
||
url=kwargs.get("url", ""),
|
||
file_size=kwargs.get("file_size", 0),
|
||
file_type=kwargs.get("file_type", ""),
|
||
attach_type=kwargs.get("attach_type", "attachment"),
|
||
sort_order=kwargs.get("sort_order", 0),
|
||
service=kwargs.get("service", "minio"),
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_oss(db: Session, oss_id: str) -> None:
|
||
"""软删除附件"""
|
||
obj = db.query(SysOss).filter(SysOss.id == oss_id, SysOss.del_flag == 0).first()
|
||
if obj:
|
||
obj.del_flag = 1
|
||
db.commit()
|
||
|
||
|
||
# ── 报价行产品 ID 解析（对照 CrmProductServiceImpl） ─────────────
|
||
|
||
|
||
QUOTE_AUTO_PRODUCT_NO_PREFIX = "SXWL"
|
||
QUOTE_AUTO_PRODUCT_NO_RANDOM_LEN = 8
|
||
PRODUCT_NO_GENERATE_MAX_RETRIES = 10
|
||
|
||
|
||
def _normalize_text(value: Any) -> Optional[str]:
|
||
if value is None:
|
||
return None
|
||
trimmed = str(value).strip()
|
||
return trimmed if trimmed else None
|
||
|
||
|
||
def _get_quote_item_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 generate_auto_product_no(db: Session, tenant_no: str) -> str:
|
||
"""Java: generateAutoProductNo — SXWL + 8位随机大写字符"""
|
||
import secrets
|
||
import string
|
||
|
||
alphabet = string.ascii_uppercase + string.digits
|
||
for _ in range(PRODUCT_NO_GENERATE_MAX_RETRIES):
|
||
suffix = "".join(
|
||
secrets.choice(alphabet) for _ in range(QUOTE_AUTO_PRODUCT_NO_RANDOM_LEN)
|
||
)
|
||
candidate = f"{QUOTE_AUTO_PRODUCT_NO_PREFIX}{suffix}"
|
||
if not product_no_exists(db, candidate, tenant_no):
|
||
return candidate
|
||
raise ValueError("自动生成产品编号失败，请稍后重试")
|
||
|
||
|
||
def resolve_product_id_for_quote_item(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
product_data: dict,
|
||
user_id: str = None,
|
||
) -> str:
|
||
"""Java: resolveProductIdForQuoteItem（CrmProduct 无 tax_rate，跳过税率匹配）"""
|
||
if not product_data:
|
||
raise ValueError("产品明细不能为空")
|
||
|
||
existing_id = (_get_quote_item_field(product_data, "product_id", "productId") or "").strip()
|
||
if existing_id:
|
||
found = get_product(db, existing_id)
|
||
if found and found.tenant_no == tenant_no:
|
||
return existing_id
|
||
|
||
product_name = _normalize_text(_get_quote_item_field(product_data, "product_name", "productName"))
|
||
if not product_name:
|
||
raise ValueError("产品名称不能为空")
|
||
model = _normalize_text(_get_quote_item_field(product_data, "model"))
|
||
unit = _normalize_text(_get_quote_item_field(product_data, "unit")) or "件"
|
||
category_id = (_get_quote_item_field(product_data, "category_id", "categoryId") or "").strip() or None
|
||
|
||
name_key = product_name.lower()
|
||
model_key = (model or "").lower()
|
||
unit_key = unit.lower()
|
||
|
||
candidates = (
|
||
db.query(CrmProduct)
|
||
.filter(
|
||
CrmProduct.tenant_no == tenant_no,
|
||
CrmProduct.del_flag == 0,
|
||
func.lower(func.trim(func.coalesce(CrmProduct.product_name, ""))) == name_key,
|
||
func.lower(func.trim(func.coalesce(CrmProduct.model, ""))) == model_key,
|
||
func.lower(func.trim(func.coalesce(CrmProduct.unit, ""))) == unit_key,
|
||
)
|
||
.order_by(CrmProduct.id.desc())
|
||
.all()
|
||
)
|
||
|
||
if candidates:
|
||
if category_id:
|
||
for candidate in candidates:
|
||
if candidate.category_id == category_id:
|
||
return candidate.id
|
||
return candidates[0].id
|
||
|
||
category_name = _normalize_text(_get_quote_item_field(product_data, "category_name", "categoryName"))
|
||
product_no = generate_auto_product_no(db, tenant_no)
|
||
category_name_val = category_name
|
||
if category_id and not category_name_val:
|
||
cat = get_category(db, category_id)
|
||
if cat:
|
||
category_name_val = cat.category_name
|
||
|
||
obj = CrmProduct(
|
||
id=generate_uuid(),
|
||
product_no=product_no,
|
||
product_name=product_name,
|
||
category_id=category_id,
|
||
category_name=category_name_val or "",
|
||
model=model,
|
||
unit=unit,
|
||
price=Decimal("0"),
|
||
status="on",
|
||
tenant_no=tenant_no,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.flush()
|
||
return obj.id
|