Files
shanxia2.0/backend/app/crud/pilot_service.py
T
2026-05-28 07:37:16 +08:00

4331 lines
157 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""中试服务管理 CRUD(细胞产品类型/规格/价格体系/项目价格、渠道信息、体检机构)"""
from __future__ import annotations
import json
import math
from datetime import datetime
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.crud.collection_protocol import urls_to_attachment_list
from app.models.models import (
BizCellProductType,
BizCellProductSpec,
BizCellProductPrice,
BizCellProductProjectSystem,
BizCellProductProjectSystemItem,
BizChannel,
BizChannelAccount,
BizChannelAgreement,
BizChannelAgreementProductPrice,
BizChannelInvoice,
BizChannelMailAddress,
BizChannelProductQuota,
BizChannelServiceOrder,
BizChannelServiceOrderProduct,
BizChannelServiceOrderTxn,
BizCheckupInstitution,
BizCheckupInstitutionInvoice,
BizCheckupInstitutionProject,
BizCheckupInstitutionPackage,
BizCheckupInstitutionPackageItem,
SysUser,
)
def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
val = (value or "").strip()
if len(val) == 10:
return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00"
return val
def _user_nick_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]:
ids = list({x for x in user_ids if x})
if not ids:
return {}
rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all()
return {u.id: (u.nick_name or "").strip() for u in rows}
# ────────────────────────────────────────────
# Entity 1: BizCellProductType 细胞产品类型
# ────────────────────────────────────────────
def _row_to_dict_cell_product_type(row) -> Dict[str, Any]:
return {
"id": row.id,
"productTypeCode": row.product_type_code or "",
"productTypeName": row.product_type_name or "",
"tenantNo": row.tenant_no or "",
"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_cell_product_types(db, *, page=1, page_size=10, keyword=None, tenant_no=None, user_info=None):
def _filtered():
q0 = db.query(BizCellProductType).filter(BizCellProductType.del_flag == 0)
if tenant_no:
q0 = q0.filter(BizCellProductType.tenant_no == tenant_no)
kw = (keyword or "").strip()
if kw:
like = f"%{kw}%"
q0 = q0.filter(
or_(
BizCellProductType.product_type_code.like(like),
BizCellProductType.product_type_name.like(like),
)
)
if user_info:
from app.core.data_scope import apply_data_scope
q0 = apply_data_scope(q0, BizCellProductType, user_info, db, tenant_no)
return q0
total = _filtered().order_by(None).with_entities(func.count(BizCellProductType.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = _filtered().order_by(BizCellProductType.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
return [_row_to_dict_cell_product_type(x) for x in rows], int(total)
def get_cell_product_type(db, type_id, tenant_no=None):
q = db.query(BizCellProductType).filter(and_(BizCellProductType.id == type_id, BizCellProductType.del_flag == 0))
if tenant_no:
q = q.filter(BizCellProductType.tenant_no == tenant_no)
return q.first()
def cell_product_type_code_exists(db, code, tenant_no, exclude_id=None):
q = db.query(BizCellProductType).filter(
and_(BizCellProductType.product_type_code == code, BizCellProductType.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCellProductType.tenant_no == tenant_no)
if exclude_id:
q = q.filter(BizCellProductType.id != exclude_id)
return db.query(q.exists()).scalar()
def create_cell_product_type(db, *, product_type_code, product_type_name, tenant_no="zkxw", order_num=None, remark=None, user_id=None):
if cell_product_type_code_exists(db, product_type_code, tenant_no):
raise ValueError(f"产品类型编码 {product_type_code} 已存在")
obj = BizCellProductType(
id=generate_uuid(),
product_type_code=product_type_code,
product_type_name=product_type_name,
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_cell_product_type(db, type_id, *, product_type_code=None, product_type_name=None, order_num=None, remark=None, user_id=None, tenant_no=None):
obj = get_cell_product_type(db, type_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
if product_type_code is not None and product_type_code != obj.product_type_code:
if cell_product_type_code_exists(db, product_type_code, tenant_no, exclude_id=type_id):
raise ValueError(f"产品类型编码 {product_type_code} 已存在")
obj.product_type_code = product_type_code
if product_type_name is not None:
obj.product_type_name = product_type_name
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_cell_product_type(db, type_id, tenant_no=None):
obj = get_cell_product_type(db, type_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 2: BizCellProductSpec 细胞产品规格
# ────────────────────────────────────────────
def _row_to_dict_cell_product_spec(row) -> Dict[str, Any]:
return {
"id": row.id,
"specCode": row.spec_code or "",
"specName": row.spec_name or "",
"tenantNo": row.tenant_no or "",
"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_cell_product_specs(db, *, page=1, page_size=10, keyword=None, tenant_no=None, user_info=None):
def _filtered():
q0 = db.query(BizCellProductSpec).filter(BizCellProductSpec.del_flag == 0)
if tenant_no:
q0 = q0.filter(BizCellProductSpec.tenant_no == tenant_no)
kw = (keyword or "").strip()
if kw:
like = f"%{kw}%"
q0 = q0.filter(
or_(
BizCellProductSpec.spec_code.like(like),
BizCellProductSpec.spec_name.like(like),
)
)
if user_info:
from app.core.data_scope import apply_data_scope
q0 = apply_data_scope(q0, BizCellProductSpec, user_info, db, tenant_no)
return q0
total = _filtered().order_by(None).with_entities(func.count(BizCellProductSpec.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = _filtered().order_by(BizCellProductSpec.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
return [_row_to_dict_cell_product_spec(x) for x in rows], int(total)
def get_cell_product_spec(db, spec_id, tenant_no=None):
q = db.query(BizCellProductSpec).filter(and_(BizCellProductSpec.id == spec_id, BizCellProductSpec.del_flag == 0))
if tenant_no:
q = q.filter(BizCellProductSpec.tenant_no == tenant_no)
return q.first()
def cell_product_spec_code_exists(db, code, tenant_no, exclude_id=None):
q = db.query(BizCellProductSpec).filter(
and_(BizCellProductSpec.spec_code == code, BizCellProductSpec.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCellProductSpec.tenant_no == tenant_no)
if exclude_id:
q = q.filter(BizCellProductSpec.id != exclude_id)
return db.query(q.exists()).scalar()
def create_cell_product_spec(db, *, spec_code, spec_name, tenant_no="zkxw", order_num=None, remark=None, user_id=None):
if cell_product_spec_code_exists(db, spec_code, tenant_no):
raise ValueError(f"规格编码 {spec_code} 已存在")
obj = BizCellProductSpec(
id=generate_uuid(),
spec_code=spec_code,
spec_name=spec_name,
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_cell_product_spec(db, spec_id, *, spec_code=None, spec_name=None, order_num=None, remark=None, user_id=None, tenant_no=None):
obj = get_cell_product_spec(db, spec_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
if spec_code is not None and spec_code != obj.spec_code:
if cell_product_spec_code_exists(db, spec_code, tenant_no, exclude_id=spec_id):
raise ValueError(f"规格编码 {spec_code} 已存在")
obj.spec_code = spec_code
if spec_name is not None:
obj.spec_name = spec_name
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_cell_product_spec(db, spec_id, tenant_no=None):
obj = get_cell_product_spec(db, spec_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 3: BizCellProductPrice 细胞产品价格
# ────────────────────────────────────────────
def _row_to_dict_cell_product_price(row) -> Dict[str, Any]:
return {
"id": row.id,
"projectName": row.project_name or "",
"projectCode": row.project_code or "",
"cellTypeCode": row.cell_type_code or "",
"cellTypeName": row.cell_type_name or "",
"specText": row.spec_text or "",
"specId": row.spec_id or "",
"cellProductTypeId": row.cell_product_type_id or "",
"unitPrice": float(row.unit_price) if row.unit_price is not None else None,
"tenantNo": row.tenant_no or "",
"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_cell_product_prices(db, *, page=1, page_size=10, keyword=None, tenant_no=None, cell_product_type_id=None, user_info=None):
def _filtered():
q0 = db.query(BizCellProductPrice).filter(BizCellProductPrice.del_flag == 0)
if tenant_no:
q0 = q0.filter(BizCellProductPrice.tenant_no == tenant_no)
if cell_product_type_id:
q0 = q0.filter(BizCellProductPrice.cell_product_type_id == cell_product_type_id)
kw = (keyword or "").strip()
if kw:
like = f"%{kw}%"
q0 = q0.filter(
or_(
BizCellProductPrice.project_name.like(like),
BizCellProductPrice.project_code.like(like),
BizCellProductPrice.cell_type_name.like(like),
)
)
if user_info:
from app.core.data_scope import apply_data_scope
q0 = apply_data_scope(q0, BizCellProductPrice, user_info, db, tenant_no)
return q0
total = _filtered().order_by(None).with_entities(func.count(BizCellProductPrice.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = _filtered().order_by(BizCellProductPrice.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
return [_row_to_dict_cell_product_price(x) for x in rows], int(total)
def get_cell_product_price(db, price_id, tenant_no=None):
q = db.query(BizCellProductPrice).filter(and_(BizCellProductPrice.id == price_id, BizCellProductPrice.del_flag == 0))
if tenant_no:
q = q.filter(BizCellProductPrice.tenant_no == tenant_no)
return q.first()
def cell_product_price_code_exists(db, code, tenant_no, exclude_id=None):
q = db.query(BizCellProductPrice).filter(
and_(BizCellProductPrice.project_code == code, BizCellProductPrice.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCellProductPrice.tenant_no == tenant_no)
if exclude_id:
q = q.filter(BizCellProductPrice.id != exclude_id)
return db.query(q.exists()).scalar()
def create_cell_product_price(
db,
*,
project_name,
project_code,
cell_type_code,
cell_type_name,
spec_text,
spec_id,
cell_product_type_id,
unit_price,
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
if cell_product_price_code_exists(db, project_code, tenant_no):
raise ValueError(f"项目编码 {project_code} 已存在")
obj = BizCellProductPrice(
id=generate_uuid(),
project_name=project_name,
project_code=project_code,
cell_type_code=cell_type_code,
cell_type_name=cell_type_name,
spec_text=spec_text or "",
spec_id=spec_id or "",
cell_product_type_id=cell_product_type_id or "",
unit_price=unit_price,
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_cell_product_price(
db,
price_id,
*,
project_name=None,
project_code=None,
cell_type_code=None,
cell_type_name=None,
spec_text=None,
spec_id=None,
cell_product_type_id=None,
unit_price=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
obj = get_cell_product_price(db, price_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
if project_code is not None and project_code != obj.project_code:
if cell_product_price_code_exists(db, project_code, tenant_no, exclude_id=price_id):
raise ValueError(f"项目编码 {project_code} 已存在")
obj.project_code = project_code
if project_name is not None:
obj.project_name = project_name
if cell_type_code is not None:
obj.cell_type_code = cell_type_code
if cell_type_name is not None:
obj.cell_type_name = cell_type_name
if spec_text is not None:
obj.spec_text = spec_text
if spec_id is not None:
obj.spec_id = spec_id
if cell_product_type_id is not None:
obj.cell_product_type_id = cell_product_type_id
if unit_price is not None:
obj.unit_price = unit_price
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_cell_product_price(db, price_id, tenant_no=None):
obj = get_cell_product_price(db, price_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 4: BizCellProductProjectSystem 项目价格体系(主)
# ────────────────────────────────────────────
def _row_to_dict_project_system(row) -> Dict[str, Any]:
return {
"id": row.id,
"systemName": row.system_name or "",
"systemCode": row.system_code or "",
"status": row.status or "1",
"tenantNo": row.tenant_no or "",
"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_project_systems(db, *, page=1, page_size=10, keyword=None, tenant_no=None, user_info=None):
def _filtered():
q0 = db.query(BizCellProductProjectSystem).filter(BizCellProductProjectSystem.del_flag == 0)
if tenant_no:
q0 = q0.filter(BizCellProductProjectSystem.tenant_no == tenant_no)
kw = (keyword or "").strip()
if kw:
like = f"%{kw}%"
q0 = q0.filter(
or_(
BizCellProductProjectSystem.system_code.like(like),
BizCellProductProjectSystem.system_name.like(like),
)
)
if user_info:
from app.core.data_scope import apply_data_scope
q0 = apply_data_scope(q0, BizCellProductProjectSystem, user_info, db, tenant_no)
return q0
total = _filtered().order_by(None).with_entities(func.count(BizCellProductProjectSystem.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = _filtered().order_by(BizCellProductProjectSystem.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
return [_row_to_dict_project_system(x) for x in rows], int(total)
def get_project_system(db, system_id, tenant_no=None):
q = db.query(BizCellProductProjectSystem).filter(
and_(BizCellProductProjectSystem.id == system_id, BizCellProductProjectSystem.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCellProductProjectSystem.tenant_no == tenant_no)
return q.first()
def project_system_code_exists(db, code, tenant_no, exclude_id=None):
q = db.query(BizCellProductProjectSystem).filter(
and_(BizCellProductProjectSystem.system_code == code, BizCellProductProjectSystem.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCellProductProjectSystem.tenant_no == tenant_no)
if exclude_id:
q = q.filter(BizCellProductProjectSystem.id != exclude_id)
return db.query(q.exists()).scalar()
def create_project_system(db, *, system_name, system_code, status="1", tenant_no="zkxw", order_num=None, remark=None, user_id=None):
if project_system_code_exists(db, system_code, tenant_no):
raise ValueError(f"体系编码 {system_code} 已存在")
obj = BizCellProductProjectSystem(
id=generate_uuid(),
system_name=system_name,
system_code=system_code,
status=status or "1",
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_project_system(db, system_id, *, system_name=None, system_code=None, status=None, order_num=None, remark=None, user_id=None, tenant_no=None):
obj = get_project_system(db, system_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
if system_code is not None and system_code != obj.system_code:
if project_system_code_exists(db, system_code, tenant_no, exclude_id=system_id):
raise ValueError(f"体系编码 {system_code} 已存在")
obj.system_code = system_code
if system_name is not None:
obj.system_name = system_name
if status is not None:
obj.status = status
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_project_system(db, system_id, tenant_no=None):
obj = get_project_system(db, system_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 5: BizCellProductProjectSystemItem 项目价格体系明细
# ────────────────────────────────────────────
def _row_to_dict_project_system_item(row) -> Dict[str, Any]:
return {
"id": row.id,
"projectSystemId": row.project_system_id or "",
"sourcePriceId": row.source_price_id or "",
"projectName": row.project_name or "",
"projectCode": row.project_code or "",
"cellTypeCode": row.cell_type_code or "",
"cellTypeName": row.cell_type_name or "",
"specText": row.spec_text or "",
"specId": row.spec_id or "",
"cellProductTypeId": row.cell_product_type_id or "",
"unitPrice": float(row.unit_price) if row.unit_price is not None else None,
"tenantNo": row.tenant_no or "",
"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_project_system_items(db, project_system_id, tenant_no=None, user_info=None):
q = db.query(BizCellProductProjectSystemItem).filter(
and_(
BizCellProductProjectSystemItem.project_system_id == project_system_id,
BizCellProductProjectSystemItem.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizCellProductProjectSystemItem.tenant_no == tenant_no)
if user_info:
from app.core.data_scope import apply_data_scope
q = apply_data_scope(q, BizCellProductProjectSystemItem, user_info, db, tenant_no)
rows = q.order_by(
BizCellProductProjectSystemItem.order_num.asc(),
BizCellProductProjectSystemItem.create_time.asc(),
).all()
return [_row_to_dict_project_system_item(x) for x in rows]
def create_project_system_item(
db,
*,
project_system_id,
source_price_id="",
project_name,
project_code,
cell_type_code,
cell_type_name,
spec_text="",
spec_id="",
cell_product_type_id="",
unit_price,
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
obj = BizCellProductProjectSystemItem(
id=generate_uuid(),
project_system_id=project_system_id,
source_price_id=source_price_id,
project_name=project_name,
project_code=project_code,
cell_type_code=cell_type_code,
cell_type_name=cell_type_name,
spec_text=spec_text or "",
spec_id=spec_id or "",
cell_product_type_id=cell_product_type_id or "",
unit_price=unit_price,
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_project_system_item(
db,
item_id,
*,
source_price_id=None,
project_name=None,
project_code=None,
cell_type_code=None,
cell_type_name=None,
spec_text=None,
spec_id=None,
cell_product_type_id=None,
unit_price=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
q = db.query(BizCellProductProjectSystemItem).filter(
and_(BizCellProductProjectSystemItem.id == item_id, BizCellProductProjectSystemItem.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCellProductProjectSystemItem.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
if source_price_id is not None:
obj.source_price_id = source_price_id
if project_name is not None:
obj.project_name = project_name
if project_code is not None:
obj.project_code = project_code
if cell_type_code is not None:
obj.cell_type_code = cell_type_code
if cell_type_name is not None:
obj.cell_type_name = cell_type_name
if spec_text is not None:
obj.spec_text = spec_text
if spec_id is not None:
obj.spec_id = spec_id
if cell_product_type_id is not None:
obj.cell_product_type_id = cell_product_type_id
if unit_price is not None:
obj.unit_price = unit_price
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_project_system_item(db, item_id, tenant_no=None):
q = db.query(BizCellProductProjectSystemItem).filter(
and_(BizCellProductProjectSystemItem.id == item_id, BizCellProductProjectSystemItem.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCellProductProjectSystemItem.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 6: BizChannel 渠道信息
# ────────────────────────────────────────────
def _row_to_dict_channel(
row,
db: Optional[Session] = None,
tenant_no: Optional[str] = None,
) -> Dict[str, Any]:
tn = tenant_no or row.tenant_no
return {
"id": row.id,
"channelName": row.channel_name or "",
"creditCode": row.credit_code or "",
"channelType": row.channel_type or "",
"contactName": row.contact_name or "",
"contactPhone": row.contact_phone or "",
"contactEmail": row.contact_email or "",
"protocolAttachments": row.protocol_attachments or "",
"attachmentFiles": urls_to_attachment_list(row.protocol_attachments, db, tn) if db else [],
"tenantNo": row.tenant_no or "",
"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_channels(db, *, page=1, page_size=10, keyword=None, tenant_no=None, user_info=None):
def _filtered():
q0 = db.query(BizChannel).filter(BizChannel.del_flag == 0)
if tenant_no:
q0 = q0.filter(BizChannel.tenant_no == tenant_no)
kw = (keyword or "").strip()
if kw:
like = f"%{kw}%"
q0 = q0.filter(
or_(
BizChannel.channel_name.like(like),
BizChannel.credit_code.like(like),
BizChannel.contact_name.like(like),
)
)
if user_info:
from app.core.data_scope import apply_data_scope
q0 = apply_data_scope(q0, BizChannel, user_info, db, tenant_no)
return q0
total = _filtered().order_by(None).with_entities(func.count(BizChannel.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = _filtered().order_by(BizChannel.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
return [_row_to_dict_channel(x, db, tenant_no) for x in rows], int(total)
def get_channel(db, channel_id, tenant_no=None):
q = db.query(BizChannel).filter(and_(BizChannel.id == channel_id, BizChannel.del_flag == 0))
if tenant_no:
q = q.filter(BizChannel.tenant_no == tenant_no)
return q.first()
def create_channel(
db,
*,
channel_name,
credit_code=None,
channel_type=None,
contact_name=None,
contact_phone=None,
contact_email=None,
protocol_attachments=None,
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
obj = BizChannel(
id=generate_uuid(),
channel_name=channel_name,
credit_code=credit_code or "",
channel_type=channel_type or "",
contact_name=contact_name or "",
contact_phone=contact_phone or "",
contact_email=contact_email or "",
protocol_attachments=protocol_attachments or "",
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_channel(
db,
channel_id,
*,
channel_name=None,
credit_code=None,
channel_type=None,
contact_name=None,
contact_phone=None,
contact_email=None,
protocol_attachments=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
obj = get_channel(db, channel_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
if channel_name is not None:
obj.channel_name = channel_name
if credit_code is not None:
obj.credit_code = credit_code
if channel_type is not None:
obj.channel_type = channel_type
if contact_name is not None:
obj.contact_name = contact_name
if contact_phone is not None:
obj.contact_phone = contact_phone
if contact_email is not None:
obj.contact_email = contact_email
if protocol_attachments is not None:
obj.protocol_attachments = protocol_attachments
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_channel(db, channel_id, tenant_no=None):
obj = get_channel(db, channel_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 7: BizChannelInvoice 渠道开票信息
# ────────────────────────────────────────────
def _row_to_dict_channel_invoice(row) -> Dict[str, Any]:
return {
"id": row.id,
"channelId": row.channel_id or "",
"companyName": row.company_name or "",
"taxNo": row.tax_no or "",
"companyAddress": row.company_address or "",
"companyPhone": row.company_phone or "",
"bankName": row.bank_name or "",
"bankAccount": row.bank_account or "",
"tenantNo": row.tenant_no or "",
"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_channel_invoices(db, channel_id, tenant_no=None, user_info=None):
q = db.query(BizChannelInvoice).filter(
and_(
BizChannelInvoice.channel_id == channel_id,
BizChannelInvoice.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelInvoice.tenant_no == tenant_no)
if user_info:
from app.core.data_scope import apply_data_scope
q = apply_data_scope(q, BizChannelInvoice, user_info, db, tenant_no)
rows = q.order_by(BizChannelInvoice.create_time.desc()).all()
return [_row_to_dict_channel_invoice(x) for x in rows]
def create_channel_invoice(
db,
*,
channel_id,
company_name,
tax_no=None,
company_address=None,
company_phone=None,
bank_name=None,
bank_account=None,
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
obj = BizChannelInvoice(
id=generate_uuid(),
channel_id=channel_id,
company_name=company_name,
tax_no=tax_no or "",
company_address=company_address or "",
company_phone=company_phone or "",
bank_name=bank_name or "",
bank_account=bank_account or "",
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_channel_invoice(
db,
invoice_id,
*,
company_name=None,
tax_no=None,
company_address=None,
company_phone=None,
bank_name=None,
bank_account=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
q = db.query(BizChannelInvoice).filter(
and_(BizChannelInvoice.id == invoice_id, BizChannelInvoice.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelInvoice.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
if company_name is not None:
obj.company_name = company_name
if tax_no is not None:
obj.tax_no = tax_no
if company_address is not None:
obj.company_address = company_address
if company_phone is not None:
obj.company_phone = company_phone
if bank_name is not None:
obj.bank_name = bank_name
if bank_account is not None:
obj.bank_account = bank_account
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_channel_invoice(db, invoice_id, tenant_no=None):
q = db.query(BizChannelInvoice).filter(
and_(BizChannelInvoice.id == invoice_id, BizChannelInvoice.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelInvoice.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 8: BizChannelMailAddress 渠道邮寄地址
# ────────────────────────────────────────────
def _row_to_dict_channel_mail_address(row) -> Dict[str, Any]:
return {
"id": row.id,
"channelId": row.channel_id or "",
"mailAddress": row.mail_address or "",
"mailContact": row.mail_contact or "",
"mailPhone": row.mail_phone or "",
"tenantNo": row.tenant_no or "",
"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_channel_mail_addresses(db, channel_id, tenant_no=None, user_info=None):
q = db.query(BizChannelMailAddress).filter(
and_(
BizChannelMailAddress.channel_id == channel_id,
BizChannelMailAddress.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelMailAddress.tenant_no == tenant_no)
if user_info:
from app.core.data_scope import apply_data_scope
q = apply_data_scope(q, BizChannelMailAddress, user_info, db, tenant_no)
rows = q.order_by(BizChannelMailAddress.create_time.desc()).all()
return [_row_to_dict_channel_mail_address(x) for x in rows]
def create_channel_mail_address(
db,
*,
channel_id,
mail_address,
mail_contact=None,
mail_phone=None,
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
obj = BizChannelMailAddress(
id=generate_uuid(),
channel_id=channel_id,
mail_address=mail_address,
mail_contact=mail_contact or "",
mail_phone=mail_phone or "",
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_channel_mail_address(
db,
address_id,
*,
mail_address=None,
mail_contact=None,
mail_phone=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
q = db.query(BizChannelMailAddress).filter(
and_(BizChannelMailAddress.id == address_id, BizChannelMailAddress.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelMailAddress.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
if mail_address is not None:
obj.mail_address = mail_address
if mail_contact is not None:
obj.mail_contact = mail_contact
if mail_phone is not None:
obj.mail_phone = mail_phone
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_channel_mail_address(db, address_id, tenant_no=None):
q = db.query(BizChannelMailAddress).filter(
and_(BizChannelMailAddress.id == address_id, BizChannelMailAddress.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelMailAddress.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 9: BizChannelAgreement 渠道协议
# ────────────────────────────────────────────
def _row_to_dict_channel_agreement(row) -> Dict[str, Any]:
def _fmt_date(v):
"""CHAR(8) YYYYMMDD → YYYY-MM-DD"""
if not v or len(v) != 8:
return v or ""
return f"{v[:4]}-{v[4:6]}-{v[6:8]}"
return {
"id": row.id,
"channelId": row.channel_id or "",
"agreementType": row.agreement_type or "",
"businessType": row.business_type or "",
"settleMode": row.settle_mode or "",
"creditLimitAmount": float(row.credit_limit_amount) if row.credit_limit_amount is not None else None,
"effectiveDate": _fmt_date(row.effective_date),
"expireDate": _fmt_date(row.expire_date),
"agreementStatus": row.agreement_status or "",
"tenantNo": row.tenant_no or "",
"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_channel_agreements(db, channel_id, tenant_no=None, user_info=None):
q = db.query(BizChannelAgreement).filter(
and_(
BizChannelAgreement.channel_id == channel_id,
BizChannelAgreement.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
if user_info:
from app.core.data_scope import apply_data_scope
q = apply_data_scope(q, BizChannelAgreement, user_info, db, tenant_no)
rows = q.order_by(BizChannelAgreement.create_time.desc()).all()
return [_row_to_dict_channel_agreement(x) for x in rows]
def create_channel_agreement(
db,
*,
channel_id,
agreement_type,
business_type=None,
settle_mode=None,
credit_limit_amount=None,
effective_date=None,
expire_date=None,
agreement_status="dsx",
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
_ed = (effective_date or "").replace("-", "") if effective_date else ""
_xd = (expire_date or "").replace("-", "") if expire_date else ""
obj = BizChannelAgreement(
id=generate_uuid(),
channel_id=channel_id,
agreement_type=agreement_type,
business_type=business_type or "",
settle_mode=settle_mode or "",
credit_limit_amount=credit_limit_amount,
effective_date=_ed,
expire_date=_xd,
agreement_status=agreement_status or "dsx",
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_channel_agreement(
db,
agreement_id,
*,
agreement_type=None,
business_type=None,
settle_mode=None,
credit_limit_amount=None,
effective_date=None,
expire_date=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
q = db.query(BizChannelAgreement).filter(
and_(BizChannelAgreement.id == agreement_id, BizChannelAgreement.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
if agreement_type is not None:
obj.agreement_type = agreement_type
if business_type is not None:
obj.business_type = business_type
if settle_mode is not None:
obj.settle_mode = settle_mode
if credit_limit_amount is not None:
obj.credit_limit_amount = credit_limit_amount
if effective_date is not None:
obj.effective_date = effective_date.replace("-", "") if effective_date else ""
if expire_date is not None:
obj.expire_date = expire_date.replace("-", "") if expire_date else ""
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_channel_agreement(db, agreement_id, tenant_no=None):
q = db.query(BizChannelAgreement).filter(
and_(BizChannelAgreement.id == agreement_id, BizChannelAgreement.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
def activate_channel_agreement(db, agreement_id, tenant_no=None, user_id=None):
"""生效协议:仅待生效(dsx)可变为生效中(sxz)"""
q = db.query(BizChannelAgreement).filter(
and_(BizChannelAgreement.id == agreement_id, BizChannelAgreement.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
if (obj.agreement_status or "").strip() != "dsx":
raise ValueError("仅待生效状态的协议可执行生效操作")
obj.agreement_status = "sxz"
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def void_channel_agreement(db, agreement_id, tenant_no=None, user_id=None):
"""作废协议:设置 agreement_status='yzf'"""
q = db.query(BizChannelAgreement).filter(
and_(BizChannelAgreement.id == agreement_id, BizChannelAgreement.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
obj.agreement_status = "yzf"
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
# ────────────────────────────────────────────
# Entity 10: BizChannelAgreementProductPrice 渠道协议产品价格
# ────────────────────────────────────────────
def _row_to_dict_channel_agreement_price(row) -> Dict[str, Any]:
return {
"id": row.id,
"channelAgreementId": row.channel_agreement_id or "",
"projectName": row.project_name or "",
"projectCode": row.project_code or "",
"cellTypeCode": row.cell_type_code or "",
"cellTypeName": row.cell_type_name or "",
"specText": row.spec_text or "",
"originalPrice": float(row.original_price) if row.original_price is not None else None,
"agreementPrice": float(row.agreement_price) if row.agreement_price is not None else None,
"tenantNo": row.tenant_no or "",
"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 _channel_agreement_price_item_key(
project_code=None,
project_name=None,
cell_type_code=None,
cell_type_name=None,
spec_text=None,
) -> str:
code = (project_code or project_name or "").strip().lower()
cell = (cell_type_code or cell_type_name or "").strip().lower()
spec = (spec_text or "").strip().lower()
return f"{code}|{cell}|{spec}"
def _channel_agreement_price_exists(
db,
channel_agreement_id,
*,
project_code=None,
project_name=None,
cell_type_code=None,
cell_type_name=None,
spec_text=None,
tenant_no=None,
exclude_price_id=None,
):
target_key = _channel_agreement_price_item_key(
project_code=project_code,
project_name=project_name,
cell_type_code=cell_type_code,
cell_type_name=cell_type_name,
spec_text=spec_text,
)
q = db.query(BizChannelAgreementProductPrice).filter(
and_(
BizChannelAgreementProductPrice.channel_agreement_id == channel_agreement_id,
BizChannelAgreementProductPrice.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelAgreementProductPrice.tenant_no == tenant_no)
if exclude_price_id:
q = q.filter(BizChannelAgreementProductPrice.id != exclude_price_id)
for row in q.all():
if _channel_agreement_price_item_key(
project_code=row.project_code,
project_name=row.project_name,
cell_type_code=row.cell_type_code,
cell_type_name=row.cell_type_name,
spec_text=row.spec_text,
) == target_key:
return True
return False
def list_channel_agreement_prices(db, channel_agreement_id, tenant_no=None, user_info=None):
q = db.query(BizChannelAgreementProductPrice).filter(
and_(
BizChannelAgreementProductPrice.channel_agreement_id == channel_agreement_id,
BizChannelAgreementProductPrice.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelAgreementProductPrice.tenant_no == tenant_no)
if user_info:
from app.core.data_scope import apply_data_scope
q = apply_data_scope(q, BizChannelAgreementProductPrice, user_info, db, tenant_no)
rows = q.order_by(BizChannelAgreementProductPrice.create_time.desc()).all()
return [_row_to_dict_channel_agreement_price(x) for x in rows]
def create_channel_agreement_price(
db,
*,
channel_agreement_id,
project_name=None,
project_code=None,
cell_type_code=None,
cell_type_name=None,
spec_text=None,
original_price=None,
agreement_price=None,
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
if _channel_agreement_price_exists(
db,
channel_agreement_id,
project_code=project_code,
project_name=project_name,
cell_type_code=cell_type_code,
cell_type_name=cell_type_name,
spec_text=spec_text,
tenant_no=tenant_no,
):
raise ValueError("该项目信息已存在,不可重复添加")
obj = BizChannelAgreementProductPrice(
id=generate_uuid(),
channel_agreement_id=channel_agreement_id,
project_name=project_name or "",
project_code=project_code or "",
cell_type_code=cell_type_code or "",
cell_type_name=cell_type_name or "",
spec_text=spec_text or "",
original_price=original_price,
agreement_price=agreement_price,
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_channel_agreement_price(
db,
price_id,
*,
project_name=None,
project_code=None,
cell_type_code=None,
cell_type_name=None,
spec_text=None,
original_price=None,
agreement_price=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
q = db.query(BizChannelAgreementProductPrice).filter(
and_(BizChannelAgreementProductPrice.id == price_id, BizChannelAgreementProductPrice.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelAgreementProductPrice.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
next_project_name = project_name if project_name is not None else obj.project_name
next_project_code = project_code if project_code is not None else obj.project_code
next_cell_type_code = cell_type_code if cell_type_code is not None else obj.cell_type_code
next_cell_type_name = cell_type_name if cell_type_name is not None else obj.cell_type_name
next_spec_text = spec_text if spec_text is not None else obj.spec_text
if _channel_agreement_price_exists(
db,
obj.channel_agreement_id,
project_code=next_project_code,
project_name=next_project_name,
cell_type_code=next_cell_type_code,
cell_type_name=next_cell_type_name,
spec_text=next_spec_text,
tenant_no=tenant_no,
exclude_price_id=price_id,
):
raise ValueError("该项目信息已存在,不可重复添加")
if project_name is not None:
obj.project_name = project_name
if project_code is not None:
obj.project_code = project_code
if cell_type_code is not None:
obj.cell_type_code = cell_type_code
if cell_type_name is not None:
obj.cell_type_name = cell_type_name
if spec_text is not None:
obj.spec_text = spec_text
if original_price is not None:
obj.original_price = original_price
if agreement_price is not None:
obj.agreement_price = agreement_price
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_channel_agreement_price(db, price_id, tenant_no=None):
q = db.query(BizChannelAgreementProductPrice).filter(
and_(BizChannelAgreementProductPrice.id == price_id, BizChannelAgreementProductPrice.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelAgreementProductPrice.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 11: BizCheckupInstitution 体检机构
# ────────────────────────────────────────────
def _row_to_dict_checkup_institution(row) -> Dict[str, Any]:
return {
"id": row.id,
"institutionName": row.institution_name or "",
"institutionAddress": row.institution_address or "",
"contactName": row.contact_name or "",
"contactPhone": row.contact_phone or "",
"status": row.status or "1",
"tenantNo": row.tenant_no or "",
"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_checkup_institutions(db, *, page=1, page_size=10, keyword=None, status=None, tenant_no=None, user_info=None):
def _filtered():
q0 = db.query(BizCheckupInstitution).filter(BizCheckupInstitution.del_flag == 0)
if tenant_no:
q0 = q0.filter(BizCheckupInstitution.tenant_no == tenant_no)
kw = (keyword or "").strip()
if kw:
like = f"%{kw}%"
q0 = q0.filter(
or_(
BizCheckupInstitution.institution_name.like(like),
BizCheckupInstitution.contact_name.like(like),
)
)
st = (status or "").strip()
if st in ("0", "1"):
q0 = q0.filter(BizCheckupInstitution.status == st)
if user_info:
from app.core.data_scope import apply_data_scope
q0 = apply_data_scope(q0, BizCheckupInstitution, user_info, db, tenant_no)
return q0
total = _filtered().order_by(None).with_entities(func.count(BizCheckupInstitution.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = _filtered().order_by(BizCheckupInstitution.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
return [_row_to_dict_checkup_institution(x) for x in rows], int(total)
def get_checkup_institution(db, institution_id, tenant_no=None):
q = db.query(BizCheckupInstitution).filter(
and_(BizCheckupInstitution.id == institution_id, BizCheckupInstitution.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCheckupInstitution.tenant_no == tenant_no)
return q.first()
def create_checkup_institution(
db,
*,
institution_name,
institution_address=None,
contact_name=None,
contact_phone=None,
status="1",
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
obj = BizCheckupInstitution(
id=generate_uuid(),
institution_name=institution_name,
institution_address=institution_address or "",
contact_name=contact_name or "",
contact_phone=contact_phone or "",
status=status or "1",
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_checkup_institution(
db,
institution_id,
*,
institution_name=None,
institution_address=None,
contact_name=None,
contact_phone=None,
status=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
obj = get_checkup_institution(db, institution_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
if institution_name is not None:
obj.institution_name = institution_name
if institution_address is not None:
obj.institution_address = institution_address
if contact_name is not None:
obj.contact_name = contact_name
if contact_phone is not None:
obj.contact_phone = contact_phone
if status is not None:
obj.status = status
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_checkup_institution(db, institution_id, tenant_no=None):
obj = get_checkup_institution(db, institution_id, tenant_no)
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 12: BizCheckupInstitutionInvoice 体检机构开票信息
# ────────────────────────────────────────────
def _row_to_dict_checkup_institution_invoice(row) -> Dict[str, Any]:
return {
"id": row.id,
"checkupInstitutionId": row.checkup_institution_id or "",
"companyName": row.company_name or "",
"taxNo": row.tax_no or "",
"companyAddress": row.company_address or "",
"companyPhone": row.company_phone or "",
"bankName": row.bank_name or "",
"bankAccount": row.bank_account or "",
"tenantNo": row.tenant_no or "",
"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_checkup_institution_invoices(db, checkup_institution_id, tenant_no=None, user_info=None):
q = db.query(BizCheckupInstitutionInvoice).filter(
and_(
BizCheckupInstitutionInvoice.checkup_institution_id == checkup_institution_id,
BizCheckupInstitutionInvoice.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionInvoice.tenant_no == tenant_no)
if user_info:
from app.core.data_scope import apply_data_scope
q = apply_data_scope(q, BizCheckupInstitutionInvoice, user_info, db, tenant_no)
rows = q.order_by(BizCheckupInstitutionInvoice.create_time.desc()).all()
return [_row_to_dict_checkup_institution_invoice(x) for x in rows]
def create_checkup_institution_invoice(
db,
*,
checkup_institution_id,
company_name,
tax_no=None,
company_address=None,
company_phone=None,
bank_name=None,
bank_account=None,
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
obj = BizCheckupInstitutionInvoice(
id=generate_uuid(),
checkup_institution_id=checkup_institution_id,
company_name=company_name,
tax_no=tax_no or "",
company_address=company_address or "",
company_phone=company_phone or "",
bank_name=bank_name or "",
bank_account=bank_account or "",
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_checkup_institution_invoice(
db,
invoice_id,
*,
company_name=None,
tax_no=None,
company_address=None,
company_phone=None,
bank_name=None,
bank_account=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
q = db.query(BizCheckupInstitutionInvoice).filter(
and_(BizCheckupInstitutionInvoice.id == invoice_id, BizCheckupInstitutionInvoice.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionInvoice.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
if company_name is not None:
obj.company_name = company_name
if tax_no is not None:
obj.tax_no = tax_no
if company_address is not None:
obj.company_address = company_address
if company_phone is not None:
obj.company_phone = company_phone
if bank_name is not None:
obj.bank_name = bank_name
if bank_account is not None:
obj.bank_account = bank_account
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_checkup_institution_invoice(db, invoice_id, tenant_no=None):
q = db.query(BizCheckupInstitutionInvoice).filter(
and_(BizCheckupInstitutionInvoice.id == invoice_id, BizCheckupInstitutionInvoice.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionInvoice.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 13: BizCheckupInstitutionProject 体检机构项目及价格
# ────────────────────────────────────────────
def _row_to_dict_checkup_institution_project(row) -> Dict[str, Any]:
return {
"id": row.id,
"checkupInstitutionId": row.checkup_institution_id or "",
"projectName": row.project_name or "",
"projectFee": float(row.project_fee) if row.project_fee is not None else None,
"status": row.status or "1",
"tenantNo": row.tenant_no or "",
"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_checkup_institution_projects(db, checkup_institution_id, tenant_no=None, user_info=None):
q = db.query(BizCheckupInstitutionProject).filter(
and_(
BizCheckupInstitutionProject.checkup_institution_id == checkup_institution_id,
BizCheckupInstitutionProject.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionProject.tenant_no == tenant_no)
if user_info:
from app.core.data_scope import apply_data_scope
q = apply_data_scope(q, BizCheckupInstitutionProject, user_info, db, tenant_no)
rows = q.order_by(BizCheckupInstitutionProject.create_time.desc()).all()
return [_row_to_dict_checkup_institution_project(x) for x in rows]
def create_checkup_institution_project(
db,
*,
checkup_institution_id,
project_name,
project_fee=None,
status="1",
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
obj = BizCheckupInstitutionProject(
id=generate_uuid(),
checkup_institution_id=checkup_institution_id,
project_name=project_name,
project_fee=project_fee,
status=status or "1",
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_checkup_institution_project(
db,
project_id,
*,
project_name=None,
project_fee=None,
status=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
q = db.query(BizCheckupInstitutionProject).filter(
and_(BizCheckupInstitutionProject.id == project_id, BizCheckupInstitutionProject.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionProject.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
if project_name is not None:
obj.project_name = project_name
if project_fee is not None:
obj.project_fee = project_fee
if status is not None:
obj.status = status
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = remark
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_checkup_institution_project(db, project_id, tenant_no=None):
q = db.query(BizCheckupInstitutionProject).filter(
and_(BizCheckupInstitutionProject.id == project_id, BizCheckupInstitutionProject.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionProject.tenant_no == tenant_no)
obj = q.first()
if not obj:
raise ValueError("记录不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ────────────────────────────────────────────
# Entity 14: BizCheckupInstitutionPackage 体检机构项目套餐
# ────────────────────────────────────────────
def _sum_package_item_fees(items: List[Dict[str, Any]]) -> float:
total = 0.0
for it in items:
fee = it.get("project_fee")
if fee is None and "projectFee" in it:
fee = it.get("projectFee")
total += float(fee) if fee is not None else 0.0
return total
def _row_to_dict_checkup_institution_package(
row,
*,
total_fee: Optional[float] = None,
item_count: Optional[int] = None,
) -> Dict[str, Any]:
d = {
"id": row.id,
"checkupInstitutionId": row.checkup_institution_id or "",
"packageName": row.package_name or "",
"packageCode": row.package_code or "",
"status": row.status or "1",
"tenantNo": row.tenant_no or "",
"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 "",
}
if total_fee is not None:
d["totalFee"] = total_fee
if item_count is not None:
d["itemCount"] = item_count
return d
def _row_to_dict_checkup_institution_package_item(
row,
*,
source_project_deleted: bool = False,
source_project_status: Optional[str] = None,
) -> Dict[str, Any]:
return {
"id": row.id,
"packageId": row.package_id or "",
"checkupInstitutionProjectId": row.checkup_institution_project_id or "",
"projectName": row.project_name or "",
"projectFee": float(row.project_fee) if row.project_fee is not None else None,
"status": source_project_status if source_project_status is not None else "1",
"sourceProjectDeleted": source_project_deleted,
"tenantNo": row.tenant_no or "",
"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 checkup_package_code_exists(db, institution_id, code, tenant_no, exclude_id=None):
q = db.query(BizCheckupInstitutionPackage).filter(
and_(
BizCheckupInstitutionPackage.checkup_institution_id == institution_id,
BizCheckupInstitutionPackage.package_code == code,
BizCheckupInstitutionPackage.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionPackage.tenant_no == tenant_no)
if exclude_id:
q = q.filter(BizCheckupInstitutionPackage.id != exclude_id)
return db.query(q.exists()).scalar()
def _get_checkup_project_for_package(db, project_id, institution_id, tenant_no=None):
q = db.query(BizCheckupInstitutionProject).filter(
BizCheckupInstitutionProject.id == project_id,
BizCheckupInstitutionProject.checkup_institution_id == institution_id,
)
if tenant_no:
q = q.filter(BizCheckupInstitutionProject.tenant_no == tenant_no)
return q.first()
def _validate_package_items_for_save(
db,
institution_id,
items,
tenant_no=None,
existing_item_project_ids=None,
):
if not items:
raise ValueError("套餐至少包含一个体检项目")
existing_ids = set(existing_item_project_ids or [])
seen = set()
for it in items:
pid = it.get("checkup_institution_project_id") or it.get("checkupInstitutionProjectId")
if not pid:
raise ValueError("套餐明细缺少机构项目")
if pid in seen:
raise ValueError("套餐内不能重复添加同一体检项目")
seen.add(pid)
proj = _get_checkup_project_for_package(db, pid, institution_id, tenant_no)
if not proj:
raise ValueError("存在无效的体检项目")
if pid in existing_ids:
continue
if proj.del_flag != 0:
raise ValueError("不能新添加已删除的体检项目")
if (proj.status or "1") != "1":
raise ValueError("套餐不能包含已停用的体检项目")
def _list_package_items_raw(db, package_id, tenant_no=None):
q = db.query(BizCheckupInstitutionPackageItem).filter(
and_(
BizCheckupInstitutionPackageItem.package_id == package_id,
BizCheckupInstitutionPackageItem.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionPackageItem.tenant_no == tenant_no)
return q.order_by(BizCheckupInstitutionPackageItem.order_num.asc(), BizCheckupInstitutionPackageItem.create_time.asc()).all()
def _package_items_to_dicts(db, institution_id, item_rows, tenant_no=None):
result = []
for row in item_rows:
proj = _get_checkup_project_for_package(
db, row.checkup_institution_project_id, institution_id, tenant_no
)
deleted = proj is None or proj.del_flag != 0
proj_status = (proj.status or "1") if proj else "1"
result.append(
_row_to_dict_checkup_institution_package_item(
row,
source_project_deleted=deleted,
source_project_status=proj_status,
)
)
return result
def _insert_package_items(
db,
*,
package_id,
institution_id,
items,
tenant_no,
user_id=None,
):
for idx, it in enumerate(items):
pid = it.get("checkup_institution_project_id") or it.get("checkupInstitutionProjectId")
proj = _get_checkup_project_for_package(db, pid, institution_id, tenant_no)
name = (it.get("project_name") or it.get("projectName") or "").strip()
if not name and proj:
name = proj.project_name or ""
fee = it.get("project_fee")
if fee is None:
fee = it.get("projectFee")
obj = BizCheckupInstitutionPackageItem(
id=generate_uuid(),
package_id=package_id,
checkup_institution_project_id=pid,
project_name=name,
project_fee=fee,
tenant_no=tenant_no,
order_num=it.get("order_num") if it.get("order_num") is not None else it.get("orderNum") or (idx + 1),
remark=it.get("remark") or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
def list_checkup_institution_packages(db, checkup_institution_id, tenant_no=None, user_info=None):
q = db.query(BizCheckupInstitutionPackage).filter(
and_(
BizCheckupInstitutionPackage.checkup_institution_id == checkup_institution_id,
BizCheckupInstitutionPackage.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionPackage.tenant_no == tenant_no)
if user_info:
from app.core.data_scope import apply_data_scope
q = apply_data_scope(q, BizCheckupInstitutionPackage, user_info, db, tenant_no)
rows = q.order_by(BizCheckupInstitutionPackage.create_time.desc()).all()
out = []
for row in rows:
item_rows = _list_package_items_raw(db, row.id, tenant_no)
fees = [float(x.project_fee) if x.project_fee is not None else 0.0 for x in item_rows]
out.append(
_row_to_dict_checkup_institution_package(
row,
total_fee=sum(fees),
item_count=len(item_rows),
)
)
return out
def get_checkup_institution_package(db, package_id, tenant_no=None):
q = db.query(BizCheckupInstitutionPackage).filter(
and_(BizCheckupInstitutionPackage.id == package_id, BizCheckupInstitutionPackage.del_flag == 0)
)
if tenant_no:
q = q.filter(BizCheckupInstitutionPackage.tenant_no == tenant_no)
return q.first()
def get_checkup_institution_package_detail(db, package_id, tenant_no=None):
pkg = get_checkup_institution_package(db, package_id, tenant_no)
if not pkg:
return None
item_rows = _list_package_items_raw(db, package_id, tenant_no)
items = _package_items_to_dicts(db, pkg.checkup_institution_id, item_rows, tenant_no)
fees = [float(x.project_fee) if x.project_fee is not None else 0.0 for x in item_rows]
d = _row_to_dict_checkup_institution_package(pkg, total_fee=sum(fees), item_count=len(item_rows))
d["items"] = items
return d
def create_checkup_institution_package(
db,
*,
checkup_institution_id,
package_name,
package_code,
status="1",
items,
tenant_no="zkxw",
order_num=None,
remark=None,
user_id=None,
):
code = (package_code or "").strip()
if not code:
raise ValueError("套餐编码不能为空")
if checkup_package_code_exists(db, checkup_institution_id, code, tenant_no):
raise ValueError(f"套餐编码 {code} 已存在")
_validate_package_items_for_save(db, checkup_institution_id, items, tenant_no)
pkg = BizCheckupInstitutionPackage(
id=generate_uuid(),
checkup_institution_id=checkup_institution_id,
package_name=package_name,
package_code=code,
status=status or "1",
tenant_no=tenant_no,
order_num=order_num or 0,
remark=remark or "",
create_by=user_id,
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(pkg)
db.flush()
_insert_package_items(
db,
package_id=pkg.id,
institution_id=checkup_institution_id,
items=items,
tenant_no=tenant_no,
user_id=user_id,
)
db.commit()
db.refresh(pkg)
return get_checkup_institution_package_detail(db, pkg.id, tenant_no)
def update_checkup_institution_package(
db,
package_id,
*,
package_name=None,
package_code=None,
status=None,
items=None,
order_num=None,
remark=None,
user_id=None,
tenant_no=None,
):
pkg = get_checkup_institution_package(db, package_id, tenant_no)
if not pkg:
raise ValueError("记录不存在")
institution_id = pkg.checkup_institution_id
if package_code is not None:
code = package_code.strip()
if not code:
raise ValueError("套餐编码不能为空")
if code != pkg.package_code and checkup_package_code_exists(
db, institution_id, code, tenant_no, exclude_id=package_id
):
raise ValueError(f"套餐编码 {code} 已存在")
pkg.package_code = code
if package_name is not None:
pkg.package_name = package_name
if status is not None:
pkg.status = status
if order_num is not None:
pkg.order_num = order_num
if remark is not None:
pkg.remark = remark
if items is not None:
old_rows = _list_package_items_raw(db, package_id, tenant_no)
existing_pids = {r.checkup_institution_project_id for r in old_rows}
_validate_package_items_for_save(
db,
institution_id,
items,
tenant_no,
existing_item_project_ids=existing_pids,
)
# 全量替换:物理删除旧明细,避免 uk(package_id, project_id) 与软删行冲突
q_items = db.query(BizCheckupInstitutionPackageItem).filter(
BizCheckupInstitutionPackageItem.package_id == package_id
)
if tenant_no:
q_items = q_items.filter(BizCheckupInstitutionPackageItem.tenant_no == tenant_no)
for row in q_items.all():
db.delete(row)
db.flush()
_insert_package_items(
db,
package_id=package_id,
institution_id=institution_id,
items=items,
tenant_no=tenant_no or pkg.tenant_no,
user_id=user_id,
)
pkg.update_by = user_id
pkg.update_time = datetime.now()
db.add(pkg)
db.commit()
return get_checkup_institution_package_detail(db, package_id, tenant_no)
def soft_delete_checkup_institution_package(db, package_id, tenant_no=None):
pkg = get_checkup_institution_package(db, package_id, tenant_no)
if not pkg:
raise ValueError("记录不存在")
now = datetime.now()
pkg.del_flag = 1
pkg.update_time = now
db.add(pkg)
for row in _list_package_items_raw(db, package_id, tenant_no):
row.del_flag = 1
row.update_time = now
db.add(row)
db.commit()
# ────────────────────────────────────────────
# Shared utility
# ────────────────────────────────────────────
def total_pages(total, page_size):
if total <= 0:
return 0
return int(math.ceil(total / float(page_size)))
# ────────────────────────────────────────────
# Entity 15: BizChannelAccount 渠道账户
# ────────────────────────────────────────────
def _row_to_dict_channel_account(row) -> Dict[str, Any]:
return {
"id": row.id,
"channelId": row.channel_id or "",
"balanceAmount": float(row.balance_amount) if row.balance_amount else 0,
"tenantNo": row.tenant_no 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 get_channel_account(db, channel_id, tenant_no=None):
"""只读查询渠道账户,不加行锁"""
q = db.query(BizChannelAccount).filter(
and_(BizChannelAccount.channel_id == channel_id, BizChannelAccount.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelAccount.tenant_no == tenant_no)
return q.first()
def _get_or_create_channel_account(db, channel_id, tenant_no, user_id):
"""获取或创建渠道账户(带行锁 with_for_update)"""
q = db.query(BizChannelAccount).filter(
and_(BizChannelAccount.channel_id == channel_id, BizChannelAccount.del_flag == 0)
).with_for_update()
if tenant_no:
q = q.filter(BizChannelAccount.tenant_no == tenant_no)
obj = q.first()
if obj:
return obj
obj = BizChannelAccount(
id=generate_uuid(),
channel_id=channel_id,
balance_amount=0,
tenant_no=tenant_no or "zkxw",
order_num=0,
remark="",
create_by=user_id,
create_dept_id="",
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.flush()
return obj
# ────────────────────────────────────────────
# Entity 16: BizChannelProductQuota 渠道产品份数
# ────────────────────────────────────────────
def _row_to_dict_channel_product_quota(row) -> Dict[str, Any]:
return {
"id": row.id,
"channelId": row.channel_id or "",
"projectCode": row.project_code or "",
"projectName": row.project_name or "",
"cellTypeCode": row.cell_type_code or "",
"cellTypeName": row.cell_type_name or "",
"specText": row.spec_text or "",
"totalQty": float(row.total_qty) if row.total_qty else 0,
"usedQty": float(row.used_qty) if row.used_qty else 0,
"remainingQty": float(row.remaining_qty) if row.remaining_qty else 0,
"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 _upsert_channel_product_quota(
db,
channel_id,
project_code,
project_name,
cell_type_code,
cell_type_name,
spec_text,
add_qty,
tenant_no,
user_id,
):
"""增加份数:已存在则累加,不存在则新建"""
q = db.query(BizChannelProductQuota).filter(
and_(
BizChannelProductQuota.channel_id == channel_id,
BizChannelProductQuota.project_code == (project_code or ""),
BizChannelProductQuota.cell_type_code == (cell_type_code or ""),
BizChannelProductQuota.spec_text == (spec_text or ""),
BizChannelProductQuota.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelProductQuota.tenant_no == tenant_no)
obj = q.first()
if obj:
obj.total_qty = (float(obj.total_qty) if obj.total_qty else 0) + add_qty
obj.remaining_qty = (float(obj.remaining_qty) if obj.remaining_qty else 0) + add_qty
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.flush()
return obj
obj = BizChannelProductQuota(
id=generate_uuid(),
channel_id=channel_id,
project_code=project_code or "",
project_name=project_name or "",
cell_type_code=cell_type_code or "",
cell_type_name=cell_type_name or "",
spec_text=spec_text or "",
total_qty=add_qty,
used_qty=0,
remaining_qty=add_qty,
tenant_no=tenant_no or "zkxw",
order_num=0,
remark="",
create_by=user_id,
create_dept_id="",
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.flush()
return obj
def _reduce_channel_product_quota(db, channel_id, project_code, cell_type_code, spec_text, reduce_qty, user_id):
"""扣减份数:验证 remaining_qty >= reduce_qty"""
q = db.query(BizChannelProductQuota).filter(
and_(
BizChannelProductQuota.channel_id == channel_id,
BizChannelProductQuota.project_code == (project_code or ""),
BizChannelProductQuota.cell_type_code == (cell_type_code or ""),
BizChannelProductQuota.spec_text == (spec_text or ""),
BizChannelProductQuota.del_flag == 0,
)
)
obj = q.first()
if not obj:
raise ValueError("份数记录不存在")
remaining = float(obj.remaining_qty) if obj.remaining_qty else 0
if remaining < reduce_qty:
raise ValueError(f"剩余份数不足,当前剩余 {remaining},需扣减 {reduce_qty}")
obj.total_qty = (float(obj.total_qty) if obj.total_qty else 0) - reduce_qty
obj.remaining_qty = remaining - reduce_qty
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.flush()
return obj
def list_channel_product_quotas(db, channel_id, tenant_no=None):
"""查询渠道下所有产品份数"""
q = db.query(BizChannelProductQuota).filter(
and_(
BizChannelProductQuota.channel_id == channel_id,
BizChannelProductQuota.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelProductQuota.tenant_no == tenant_no)
rows = q.order_by(BizChannelProductQuota.create_time.desc()).all()
return [_row_to_dict_channel_product_quota(x) for x in rows]
# ────────────────────────────────────────────
# Entity 17: BizChannelServiceOrder 渠道服务订单
# ────────────────────────────────────────────
def _calc_gmcp_purchase_total_from_db(db, order_id, tenant_no=None) -> float:
"""从订单产品持仓汇总购买总额(协议单价 × 累计购买份数)。"""
q = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
total = 0.0
for prod in q.all():
price = float(prod.agreement_unit_price or 0)
qty = float(prod.total_buy_qty or 0)
total += price * qty
return total
def _batch_gmcp_purchase_totals(db, order_ids, tenant_no=None) -> Dict[str, float]:
if not order_ids:
return {}
amount_expr = BizChannelServiceOrderProduct.agreement_unit_price * BizChannelServiceOrderProduct.total_buy_qty
q = db.query(
BizChannelServiceOrderProduct.service_order_id,
func.coalesce(func.sum(amount_expr), 0),
).filter(
BizChannelServiceOrderProduct.del_flag == 0,
BizChannelServiceOrderProduct.service_order_id.in_(order_ids),
)
if tenant_no:
q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
return {row[0]: float(row[1] or 0) for row in q.group_by(BizChannelServiceOrderProduct.service_order_id).all()}
def _calc_gmcp_current_holding_amount_from_db(db, order_id, tenant_no=None) -> float:
"""从订单产品持仓汇总当前余额(协议单价 × 当前持有份数)。"""
q = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
total = 0.0
for prod in q.all():
price = float(prod.agreement_unit_price or 0)
qty = float(prod.current_qty or 0)
total += price * qty
return total
def _batch_gmcp_current_holding_amounts(db, order_ids, tenant_no=None) -> Dict[str, float]:
if not order_ids:
return {}
amount_expr = BizChannelServiceOrderProduct.agreement_unit_price * BizChannelServiceOrderProduct.current_qty
q = db.query(
BizChannelServiceOrderProduct.service_order_id,
func.coalesce(func.sum(amount_expr), 0),
).filter(
BizChannelServiceOrderProduct.del_flag == 0,
BizChannelServiceOrderProduct.service_order_id.in_(order_ids),
)
if tenant_no:
q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
return {row[0]: float(row[1] or 0) for row in q.group_by(BizChannelServiceOrderProduct.service_order_id).all()}
def _resolve_service_order_purchase_total(row, db=None, tenant_no=None, purchase_total_override=None) -> float:
stored = float(row.purchase_total_amount) if row.purchase_total_amount else 0
if (row.order_type or "") != "gmcp":
return stored
if purchase_total_override is not None:
return float(purchase_total_override)
if db:
computed = _calc_gmcp_purchase_total_from_db(db, row.id, tenant_no or row.tenant_no)
if computed > 0 or not row.purchase_total_amount:
return computed
return stored
def _resolve_service_order_current_balance(
row, db=None, tenant_no=None, current_balance_override=None
) -> float:
if (row.order_type or "") == "gmcp":
if current_balance_override is not None:
return float(current_balance_override)
if db:
return _calc_gmcp_current_holding_amount_from_db(db, row.id, tenant_no or row.tenant_no)
return 0.0
return float(row.current_balance) if row.current_balance else 0
def _row_to_dict_service_order(
row,
db=None,
tenant_no=None,
*,
purchase_total_override=None,
current_balance_override=None,
) -> Dict[str, Any]:
tn = tenant_no or row.tenant_no
purchase_total = _resolve_service_order_purchase_total(
row, db, tn, purchase_total_override=purchase_total_override
)
current_balance = _resolve_service_order_current_balance(
row, db, tn, current_balance_override=current_balance_override
)
d = {
"id": row.id,
"channelId": row.channel_id or "",
"channelName": row.channel_name or "",
"orderName": row.order_name or "",
"deptName": row.dept_name or "",
"orderType": row.order_type or "",
"prepayAmount": float(row.prepay_amount) if row.prepay_amount else 0,
"purchaseTotalAmount": purchase_total,
"orderStatus": row.order_status or "",
"channelAgreementId": row.channel_agreement_id or "",
"totalRechargeAmount": float(row.total_recharge_amount) if row.total_recharge_amount else 0,
"totalRefundAmount": float(row.total_refund_amount) if row.total_refund_amount else 0,
"currentBalance": current_balance,
"balanceBefore": float(row.balance_before) if row.balance_before else 0,
"balanceAfter": float(row.balance_after) if row.balance_after else 0,
"financeSubmitBy": row.finance_submit_by or "",
"financeSubmitTime": row.finance_submit_time.strftime("%Y-%m-%d %H:%M:%S") if row.finance_submit_time else "",
"financeSubmitRemark": row.finance_submit_remark or "",
"financeSubmitAttachments": row.finance_submit_attachments or "",
"financeSubmitAttachmentFiles": urls_to_attachment_list(row.finance_submit_attachments, db, tn) if db and row.finance_submit_attachments else [],
"financeConfirmTime": row.finance_confirm_time.strftime("%Y-%m-%d %H:%M:%S") if row.finance_confirm_time else "",
"financeConfirmAttachments": row.finance_confirm_attachments or "",
"financeConfirmAttachmentFiles": urls_to_attachment_list(row.finance_confirm_attachments, db, tn) if db and row.finance_confirm_attachments else [],
"financeRemark": row.finance_remark or "",
"refundApplyBy": row.refund_apply_by or "",
"refundApplyTime": row.refund_apply_time.strftime("%Y-%m-%d %H:%M:%S") if row.refund_apply_time else "",
"refundAmount": float(row.refund_amount) if row.refund_amount else 0,
"refundRemark": row.refund_remark or "",
"refundAttachments": row.refund_attachments or "",
"refundAttachmentFiles": urls_to_attachment_list(row.refund_attachments, db, tn) if db and row.refund_attachments else [],
"refundFinanceConfirmTime": row.refund_finance_confirm_time.strftime("%Y-%m-%d %H:%M:%S") if row.refund_finance_confirm_time else "",
"refundFinanceConfirmAttachments": row.refund_finance_confirm_attachments or "",
"refundFinanceConfirmAttachmentFiles": urls_to_attachment_list(row.refund_finance_confirm_attachments, db, tn) if db and row.refund_finance_confirm_attachments else [],
"refundFinanceRemark": row.refund_finance_remark or "",
"tenantNo": row.tenant_no or "",
"remark": row.remark or "",
"createBy": row.create_by 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 "",
}
return d
def _row_to_dict_service_order_product(row) -> Dict[str, Any]:
return {
"id": row.id,
"serviceOrderId": row.service_order_id or "",
"projectCode": row.project_code or "",
"projectName": row.project_name or "",
"cellType": row.cell_type or "",
"specText": row.spec_text or "",
"agreementUnitPrice": float(row.agreement_unit_price) if row.agreement_unit_price else 0,
"buyQty": float(row.buy_qty) if row.buy_qty else 0,
"buyAmount": float(row.buy_amount) if row.buy_amount else 0,
"totalBuyQty": float(row.total_buy_qty) if row.total_buy_qty else 0,
"totalUnsubscribeQty": float(row.total_unsubscribe_qty) if row.total_unsubscribe_qty else 0,
"currentQty": float(row.current_qty) if row.current_qty else 0,
"totalUnsubscribeAmount": float(row.total_unsubscribe_amount) if row.total_unsubscribe_amount else 0,
}
def _normalize_txn_product_item(p: Dict[str, Any]) -> Dict[str, Any]:
"""流水 products_json 项统一为 camelCase(入库时为 snake_case)。"""
qty = p.get("qty")
if qty is None:
qty = p.get("buyQty") or p.get("buy_qty")
unit_price = p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0
qty_val = float(qty or 0)
amount = p.get("amount")
if amount is None and qty_val:
amount = float(unit_price or 0) * qty_val
return {
"projectCode": p.get("projectCode") or p.get("project_code") or "",
"projectName": p.get("projectName") or p.get("project_name") or "",
"cellType": p.get("cellType") or p.get("cell_type") or "",
"specText": p.get("specText") or p.get("spec_text") or "",
"agreementUnitPrice": float(unit_price or 0),
"qty": qty_val,
"amount": float(amount or 0),
}
def _enrich_txn_products_from_order(
db,
service_order_id: str,
products: List[Dict[str, Any]],
tenant_no=None,
) -> List[Dict[str, Any]]:
"""退订等流水可能仅存 projectCode,从订单持仓补全项目名与规格。"""
if not db or not service_order_id or not products:
return products
holdings = get_service_order_products(db, service_order_id, tenant_no)
if not holdings:
return products
by_code_spec = {
(h.get("projectCode") or "", h.get("specText") or ""): h for h in holdings
}
by_code: Dict[str, List[Dict[str, Any]]] = {}
for h in holdings:
code = h.get("projectCode") or ""
by_code.setdefault(code, []).append(h)
enriched: List[Dict[str, Any]] = []
for p in products:
item = dict(p)
code = item.get("projectCode") or ""
spec = item.get("specText") or ""
holding = by_code_spec.get((code, spec))
if not holding and code:
matches = by_code.get(code, [])
if len(matches) == 1:
holding = matches[0]
if holding:
if not (item.get("projectName") or "").strip():
item["projectName"] = holding.get("projectName") or ""
if not (item.get("specText") or "").strip():
item["specText"] = holding.get("specText") or ""
if not (item.get("cellType") or "").strip():
item["cellType"] = holding.get("cellType") or ""
enriched.append(item)
return enriched
def _row_to_dict_service_order_txn(row, db=None, tenant_no=None) -> Dict[str, Any]:
tn = tenant_no or row.tenant_no
d = {
"id": row.id,
"serviceOrderId": row.service_order_id or "",
"txnType": row.txn_type or "",
"amount": float(row.amount) if row.amount else 0,
"txnStatus": row.txn_status or "",
"productsJson": row.products_json or "",
"remark": row.remark or "",
"refundAttachments": row.refund_attachments or "",
"refundAttachmentFiles": urls_to_attachment_list(row.refund_attachments, db, tn) if db and row.refund_attachments else [],
"financeSubmitBy": row.finance_submit_by or "",
"financeSubmitTime": row.finance_submit_time.strftime("%Y-%m-%d %H:%M:%S") if row.finance_submit_time else "",
"financeSubmitRemark": row.finance_submit_remark or "",
"financeSubmitAttachments": row.finance_submit_attachments or "",
"financeSubmitAttachmentFiles": urls_to_attachment_list(row.finance_submit_attachments, db, tn) if db and row.finance_submit_attachments else [],
"financeConfirmTime": row.finance_confirm_time.strftime("%Y-%m-%d %H:%M:%S") if row.finance_confirm_time else "",
"financeConfirmBy": row.finance_confirm_by or "",
"financeConfirmAttachments": row.finance_confirm_attachments or "",
"financeConfirmAttachmentFiles": urls_to_attachment_list(row.finance_confirm_attachments, db, tn) if db and row.finance_confirm_attachments else [],
"financeRemark": row.finance_remark or "",
"voidBy": row.void_by or "",
"voidTime": row.void_time.strftime("%Y-%m-%d %H:%M:%S") if row.void_time else "",
"balanceBefore": float(row.balance_before) if row.balance_before else 0,
"balanceAfter": float(row.balance_after) if row.balance_after else 0,
"tenantNo": row.tenant_no or "",
"createBy": row.create_by 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 "",
}
if row.products_json:
try:
raw_products = json.loads(row.products_json)
products = [_normalize_txn_product_item(x) for x in raw_products]
if db and row.service_order_id:
products = _enrich_txn_products_from_order(
db, row.service_order_id, products, tenant_no
)
d["products"] = products
except Exception:
d["products"] = []
else:
d["products"] = []
return d
def _service_order_txn_to_response_dict(
db,
txn,
tenant_no=None,
*,
order=None,
name_map=None,
) -> Dict[str, Any]:
"""流水响应字典:含订单渠道信息与提交人/创建人昵称。"""
d = _row_to_dict_service_order_txn(txn, db, tenant_no)
if order is None and txn.service_order_id:
order = get_service_order(db, txn.service_order_id, tenant_no)
if order:
d["channelName"] = order.channel_name or ""
d["orderName"] = order.order_name or ""
d["deptName"] = order.dept_name or ""
d["orderType"] = order.order_type or ""
confirm_uid = txn.finance_confirm_by or (
txn.update_by
if (txn.txn_status or "") == "yqr" and txn.finance_confirm_time and not txn.finance_confirm_by
else None
)
void_uid = txn.void_by or (
txn.update_by
if (txn.txn_status or "") == "yzf"
and txn.finance_submit_time
and not txn.void_by
else None
)
if name_map is None:
user_ids = [x for x in (txn.create_by, txn.finance_submit_by, confirm_uid, void_uid) if x]
name_map = _user_nick_name_map(db, user_ids) if db and user_ids else {}
d["createByName"] = name_map.get(txn.create_by or "", "") or (txn.create_by or "")
d["financeSubmitByName"] = name_map.get(txn.finance_submit_by or "", "") or (
txn.finance_submit_by or ""
)
d["financeConfirmByName"] = name_map.get(confirm_uid or "", "") or (confirm_uid or "")
if not d.get("voidTime") and (txn.txn_status or "") == "yzf" and txn.update_time:
d["voidTime"] = txn.update_time.strftime("%Y-%m-%d %H:%M:%S")
d["voidByName"] = name_map.get(void_uid or "", "") or (void_uid or "")
if not (d.get("financeSubmitRemark") or "").strip() and (d.get("remark") or "").strip():
d["financeSubmitRemark"] = (d.get("remark") or "").strip()
_enrich_finance_operate_fields(d, txn)
return d
def _enrich_finance_operate_fields(d: Dict[str, Any], txn) -> None:
"""财务页「已处理」展示:操作类型、操作人、操作时间。"""
status = txn.txn_status or ""
if status == "yqr" and txn.finance_confirm_time:
d["financeOperateType"] = "confirm"
d["financeOperateTypeLabel"] = "财务确认"
d["financeOperateByName"] = d.get("financeConfirmByName") or d.get("financeConfirmBy") or ""
d["financeOperateTime"] = d.get("financeConfirmTime") or ""
elif status == "yzf" and txn.finance_submit_time:
d["financeOperateType"] = "void"
d["financeOperateTypeLabel"] = "财务作废"
d["financeOperateByName"] = d.get("voidByName") or d.get("voidBy") or ""
d["financeOperateTime"] = d.get("voidTime") or ""
else:
d["financeOperateType"] = ""
d["financeOperateTypeLabel"] = ""
d["financeOperateByName"] = ""
d["financeOperateTime"] = ""
def list_channel_service_orders(
db,
*,
page=1,
page_size=10,
order_type=None,
order_status=None,
channel_id=None,
keyword=None,
create_time_start=None,
create_time_end=None,
tenant_no=None,
user_info=None,
):
def _filtered():
q0 = db.query(BizChannelServiceOrder).filter(BizChannelServiceOrder.del_flag == 0)
if tenant_no:
q0 = q0.filter(BizChannelServiceOrder.tenant_no == tenant_no)
ot = (order_type or "").strip()
if ot:
q0 = q0.filter(BizChannelServiceOrder.order_type == ot)
# Support comma-separated order_status
os_raw = (order_status or "").strip()
if os_raw:
statuses = [s.strip() for s in os_raw.split(",") if s.strip()]
if statuses:
q0 = q0.filter(BizChannelServiceOrder.order_status.in_(statuses))
if channel_id:
q0 = q0.filter(BizChannelServiceOrder.channel_id == channel_id)
kw = (keyword or "").strip()
if kw:
like = f"%{kw}%"
q0 = q0.filter(
or_(
BizChannelServiceOrder.channel_name.like(like),
BizChannelServiceOrder.order_name.like(like),
BizChannelServiceOrder.dept_name.like(like),
)
)
if create_time_start and create_time_start.strip():
q0 = q0.filter(
BizChannelServiceOrder.create_time
>= _normalize_date_bound(create_time_start.strip())
)
if create_time_end and create_time_end.strip():
q0 = q0.filter(
BizChannelServiceOrder.create_time
<= _normalize_date_bound(create_time_end.strip(), end_of_day=True)
)
if user_info:
from app.core.data_scope import apply_data_scope
q0 = apply_data_scope(q0, BizChannelServiceOrder, user_info, db, tenant_no)
return q0
total = _filtered().order_by(None).with_entities(func.count(BizChannelServiceOrder.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = (
_filtered()
.order_by(BizChannelServiceOrder.create_time.desc())
.offset((page - 1) * page_size)
.limit(page_size)
.all()
)
user_ids: List[str] = []
for x in rows:
if x.create_by:
user_ids.append(x.create_by)
if x.finance_submit_by:
user_ids.append(x.finance_submit_by)
name_map = _user_nick_name_map(db, user_ids) if db else {}
gmcp_ids = [x.id for x in rows if (x.order_type or "") == "gmcp"]
purchase_total_map = _batch_gmcp_purchase_totals(db, gmcp_ids, tenant_no) if db and gmcp_ids else {}
holding_amount_map = (
_batch_gmcp_current_holding_amounts(db, gmcp_ids, tenant_no) if db and gmcp_ids else {}
)
return [
{
**_row_to_dict_service_order(
x,
db,
tenant_no,
purchase_total_override=purchase_total_map.get(x.id),
current_balance_override=holding_amount_map.get(x.id),
),
"createByName": name_map.get(x.create_by or "", "") or (x.create_by or ""),
"financeSubmitByName": name_map.get(x.finance_submit_by or "", "") or (x.finance_submit_by or ""),
}
for x in rows
], int(total)
def get_service_order(db, order_id, tenant_no=None):
q = db.query(BizChannelServiceOrder).filter(
and_(BizChannelServiceOrder.id == order_id, BizChannelServiceOrder.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelServiceOrder.tenant_no == tenant_no)
return q.first()
def get_service_order_products(db, service_order_id, tenant_no=None):
q = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == service_order_id,
BizChannelServiceOrderProduct.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
rows = q.order_by(BizChannelServiceOrderProduct.create_time.asc()).all()
return [_row_to_dict_service_order_product(x) for x in rows]
# ────────────────────────────────────────────
# Service Order: Create / Update / Delete
# ────────────────────────────────────────────
def _calc_gmcp_purchase_total(products):
total = 0.0
for p in products:
price = float(p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0)
qty = float(p.get("buyQty") or p.get("buy_qty") or 0)
total += price * qty
return total
def _gmcp_has_purchase_lock(db, order_id, tenant_no=None) -> bool:
"""是否存在购买记录(未作废 purchase 流水,或历史建单写入的产品行)。"""
q = db.query(BizChannelServiceOrderTxn).filter(
and_(
BizChannelServiceOrderTxn.service_order_id == order_id,
BizChannelServiceOrderTxn.txn_type == "purchase",
BizChannelServiceOrderTxn.txn_status != "yzf",
BizChannelServiceOrderTxn.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelServiceOrderTxn.tenant_no == tenant_no)
if q.first():
return True
pq = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.del_flag == 0,
)
)
if tenant_no:
pq = pq.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
for row in pq.all():
if float(row.buy_qty or 0) > 0 or float(row.total_buy_qty or 0) > 0:
return True
return False
def gmcp_agreement_editable(db, order, tenant_no=None) -> bool:
if order.order_type != "gmcp" or order.order_status == "yzf":
return False
return not _gmcp_has_purchase_lock(db, order.id, tenant_no or order.tenant_no)
def _resolve_agreement_price_cell_type(db, channel_agreement_id, project_code, spec_text, cell_type_fallback=""):
price_row = (
db.query(BizChannelAgreementProductPrice)
.filter(
and_(
BizChannelAgreementProductPrice.channel_agreement_id == channel_agreement_id,
BizChannelAgreementProductPrice.project_code == (project_code or ""),
BizChannelAgreementProductPrice.spec_text == (spec_text or ""),
BizChannelAgreementProductPrice.del_flag == 0,
)
)
).first()
if price_row:
return price_row.cell_type_code or "", price_row.cell_type_name or cell_type_fallback
return "", cell_type_fallback
def _save_gmcp_product_rows(
db,
order_id,
products,
tenant_no,
user_id,
now=None,
):
"""仅保存购买订单产品明细,不调整份数库存"""
now = now or datetime.now()
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
project_name = p.get("projectName") or p.get("project_name") or ""
cell_type = p.get("cellType") or p.get("cell_type") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
agreement_unit_price = float(p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0)
buy_qty = float(p.get("buyQty") or p.get("buy_qty") or 0)
buy_amount = float(p.get("buyAmount") or p.get("buy_amount") or agreement_unit_price * buy_qty)
prod_row = BizChannelServiceOrderProduct(
id=generate_uuid(),
service_order_id=order_id,
project_code=project_code,
project_name=project_name,
cell_type=cell_type,
spec_text=spec_text,
agreement_unit_price=agreement_unit_price,
buy_qty=buy_qty,
buy_amount=buy_amount,
tenant_no=tenant_no or "zkxw",
order_num=0,
remark="",
create_by=user_id,
create_dept_id="",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
db.add(prod_row)
db.flush()
def _replace_gmcp_product_rows(db, order_id, products, tenant_no, user_id):
now = datetime.now()
db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.del_flag == 0,
)
).update({"del_flag": 1, "update_time": now}, synchronize_session=False)
_save_gmcp_product_rows(db, order_id, products, tenant_no, user_id, now=now)
def _apply_gmcp_order_quotas(db, order, tenant_no, user_id):
"""财务确认收款后,按订单明细增加渠道产品份数"""
products = get_service_order_products(db, order.id, tenant_no)
for prod in products:
project_code = prod.get("projectCode") or ""
spec_text = prod.get("specText") or ""
cell_type_code, cell_type_name = _resolve_agreement_price_cell_type(
db,
order.channel_agreement_id,
project_code,
spec_text,
prod.get("cellType") or "",
)
_upsert_channel_product_quota(
db,
channel_id=order.channel_id,
project_code=project_code,
project_name=prod.get("projectName") or "",
cell_type_code=cell_type_code,
cell_type_name=cell_type_name,
spec_text=spec_text,
add_qty=float(prod.get("buyQty") or 0),
tenant_no=tenant_no or order.tenant_no,
user_id=user_id,
)
def _get_valid_channel_agreement(db, channel_agreement_id, channel_id, tenant_no=None):
"""校验渠道协议存在、属于当前渠道且已生效"""
aid = (channel_agreement_id or "").strip()
if not aid:
raise ValueError("请选择渠道协议")
q = db.query(BizChannelAgreement).filter(
and_(
BizChannelAgreement.id == aid,
BizChannelAgreement.channel_id == channel_id,
BizChannelAgreement.del_flag == 0,
)
)
if tenant_no:
q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
agreement = q.first()
if not agreement:
raise ValueError("渠道协议不存在或不属于当前渠道")
if (agreement.agreement_status or "").strip() != "sxz":
raise ValueError("协议未生效,无法使用")
return agreement
def create_channel_service_order(
db,
*,
channel_id,
order_type,
channel_name=None,
order_name=None,
dept_name=None,
prepay_amount=None,
channel_agreement_id=None,
products=None,
tenant_no=None,
remark=None,
user_id=None,
):
# Validate channel exists
channel = get_channel(db, channel_id, tenant_no)
if not channel:
raise ValueError("渠道不存在")
now = datetime.now()
balance_before = None
balance_after = None
purchase_total_amount = None
order_status = "dsx"
if order_type == "yfk":
# 预存款订单:创建后为待生效,充值金额通过「发起充值」流水录入
_get_valid_channel_agreement(db, channel_agreement_id, channel_id, tenant_no)
order_status = "dsx"
prepay_amount = None
elif order_type == "gmcp":
# 购买型:建单仅绑定渠道与协议;产品通过 purchase 流水录入并经财务确认
if not channel_agreement_id:
raise ValueError("请选择渠道协议")
_get_valid_channel_agreement(db, channel_agreement_id, channel_id, tenant_no)
purchase_total_amount = _calc_gmcp_purchase_total(products) if products else None
order_status = "dsx"
else:
raise ValueError(f"不支持的订单类型: {order_type}")
order_id = generate_uuid()
order = BizChannelServiceOrder(
id=order_id,
channel_id=channel_id,
channel_name=channel_name or channel.channel_name or "",
order_name=(order_name or "").strip() or None,
dept_name=dept_name or "",
order_type=order_type,
prepay_amount=None,
purchase_total_amount=purchase_total_amount,
order_status=order_status,
channel_agreement_id=channel_agreement_id,
balance_before=balance_before,
balance_after=balance_after,
tenant_no=tenant_no or "zkxw",
order_num=0,
remark=remark or "",
create_by=user_id,
create_dept_id="",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
db.add(order)
db.flush()
db.commit()
db.refresh(order)
return order
def update_channel_service_order(db, order_id, *, tenant_no=None, user_id=None, products=None, **kwargs):
"""更新订单基本信息;gmcp 产品仅通过 purchase 流水维护"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_status == "yzf":
raise ValueError("已作废订单不可编辑")
if kwargs.get("channel_id"):
raise ValueError("渠道不可修改")
if products is not None and order.order_type == "gmcp":
raise ValueError("请通过「购买产品」操作录入产品明细")
if order.order_type in ("yfk", "gmcp"):
if order.order_status not in ("dsx", "sxz"):
raise ValueError("当前状态不允许编辑")
else:
raise ValueError("不支持的订单类型")
if "dept_name" in kwargs and kwargs["dept_name"] is not None:
order.dept_name = kwargs["dept_name"]
if "order_name" in kwargs and kwargs["order_name"] is not None:
order.order_name = (kwargs["order_name"] or "").strip() or None
if "remark" in kwargs and kwargs["remark"] is not None:
order.remark = kwargs["remark"]
if order.order_type == "yfk":
if "channel_agreement_id" in kwargs and kwargs["channel_agreement_id"] is not None:
new_agreement_id = kwargs["channel_agreement_id"]
if new_agreement_id != (order.channel_agreement_id or ""):
if order.order_status != "dsx":
raise ValueError("已有业务流水,不可修改渠道协议")
_get_valid_channel_agreement(
db, new_agreement_id, order.channel_id, tenant_no
)
order.channel_agreement_id = new_agreement_id
elif order.order_type == "gmcp":
if "channel_agreement_id" in kwargs and kwargs["channel_agreement_id"] is not None:
new_agreement_id = kwargs["channel_agreement_id"]
if new_agreement_id != (order.channel_agreement_id or ""):
if _gmcp_has_purchase_lock(db, order_id, tenant_no):
raise ValueError("已有购买记录,不可修改渠道协议")
_get_valid_channel_agreement(
db, new_agreement_id, order.channel_id, tenant_no
)
order.channel_agreement_id = new_agreement_id
order.update_by = user_id
order.update_time = datetime.now()
db.add(order)
db.commit()
db.refresh(order)
return order
def soft_delete_service_order(db, order_id, tenant_no=None):
"""仅允许待生效状态下删除订单"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_status != "dsx":
raise ValueError("仅待生效订单可删除")
order.del_flag = 1
order.update_time = datetime.now()
db.add(order)
db.commit()
# ────────────────────────────────────────────
# Service Order: Status Transitions
# ────────────────────────────────────────────
def submit_finance_confirm(
db,
order_id,
*,
finance_submit_remark=None,
finance_submit_attachments=None,
tenant_no=None,
user_id=None,
):
"""发起财务确认:type=yfk/gmcp + status=dsx → dqr"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_type not in ("yfk", "gmcp"):
raise ValueError("仅预存款或购买产品订单可发起财务确认")
if order.order_status != "dsx":
raise ValueError("仅待生效状态可发起财务确认")
now = datetime.now()
order.order_status = "dqr"
order.finance_submit_remark = finance_submit_remark or ""
order.finance_submit_attachments = finance_submit_attachments or ""
order.finance_submit_time = now
order.finance_submit_by = user_id
order.update_by = user_id
order.update_time = now
db.add(order)
db.commit()
db.refresh(order)
return order
def confirm_receipt(db, order_id, *, finance_remark=None, finance_confirm_attachments=None, tenant_no=None, user_id=None):
"""确认到账:yfk dqr→yqr 增加余额;gmcp dqr→sxz 增加产品份数"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_type not in ("yfk", "gmcp"):
raise ValueError("仅预存款或购买产品订单可确认到账")
if order.order_status != "dqr":
raise ValueError("仅待确认状态可确认到账")
now = datetime.now()
if order.order_type == "yfk":
account = _get_or_create_channel_account(db, order.channel_id, tenant_no or order.tenant_no, user_id)
balance_before = float(account.balance_amount) if account.balance_amount else 0
add_amount = float(order.prepay_amount) if order.prepay_amount else 0
account.balance_amount = balance_before + add_amount
balance_after = float(account.balance_amount)
account.update_by = user_id
account.update_time = now
db.add(account)
db.flush()
order.order_status = "yqr"
order.balance_before = balance_before
order.balance_after = balance_after
else:
_apply_gmcp_order_quotas(db, order, tenant_no, user_id)
order.order_status = "sxz"
order.finance_confirm_time = now
order.finance_confirm_attachments = finance_confirm_attachments or ""
order.finance_remark = finance_remark or ""
order.update_by = user_id
order.update_time = now
db.add(order)
db.commit()
db.refresh(order)
return order
def submit_refund(db, order_id, *, refund_amount=None, refund_remark=None, refund_attachments=None, tenant_no=None, user_id=None):
"""申请退款:type=yfk + status=yqr → tfqrz"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_type != "yfk":
raise ValueError("仅预存款订单可申请退款")
if order.order_status != "yqr":
raise ValueError("仅已确认状态可申请退款")
now = datetime.now()
# Default refund_amount to prepay_amount
if refund_amount is None:
refund_amount = float(order.prepay_amount) if order.prepay_amount else 0
else:
refund_amount = float(refund_amount)
prepay = float(order.prepay_amount) if order.prepay_amount else 0
if refund_amount > prepay:
raise ValueError(f"退款金额不能大于预存金额 {prepay}")
order.order_status = "tfqrz"
order.refund_apply_by = user_id
order.refund_apply_time = now
order.refund_amount = refund_amount
order.refund_remark = refund_remark or ""
order.refund_attachments = refund_attachments or ""
order.update_by = user_id
order.update_time = now
db.add(order)
db.commit()
db.refresh(order)
return order
def confirm_refund(db, order_id, *, refund_finance_remark=None, refund_finance_confirm_attachments=None, tenant_no=None, user_id=None):
"""确认退款:status=tfqrz → ytf,扣减余额"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_status != "tfqrz":
raise ValueError("仅退款待确认状态可确认退款")
now = datetime.now()
account = _get_or_create_channel_account(db, order.channel_id, tenant_no or order.tenant_no, user_id)
balance_before = float(account.balance_amount) if account.balance_amount else 0
deduct = float(order.refund_amount) if order.refund_amount else 0
account.balance_amount = balance_before - deduct
balance_after = float(account.balance_amount)
account.update_by = user_id
account.update_time = now
db.add(account)
db.flush()
order.order_status = "ytf"
order.refund_finance_confirm_time = now
order.refund_finance_confirm_attachments = refund_finance_confirm_attachments or ""
order.refund_finance_remark = refund_finance_remark or ""
order.balance_before = balance_before
order.balance_after = balance_after
order.update_by = user_id
order.update_time = now
db.add(order)
db.commit()
db.refresh(order)
return order
def void_service_order(db, order_id, *, tenant_no=None, user_id=None):
"""作废订单:type=gmcp + status=sxz → yzf,扣减已生效的产品份数"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_type != "gmcp":
raise ValueError("仅购买产品订单可作废")
if order.order_status != "sxz":
raise ValueError("仅生效中状态可作废")
now = datetime.now()
# Reduce product quotas
products = get_service_order_products(db, order_id, tenant_no)
for prod in products:
cell_type_code, _ = _resolve_agreement_price_cell_type(
db,
order.channel_agreement_id,
prod.get("projectCode") or "",
prod.get("specText") or "",
prod.get("cellType") or "",
)
_reduce_channel_product_quota(
db,
channel_id=order.channel_id,
project_code=prod.get("projectCode") or "",
cell_type_code=cell_type_code,
spec_text=prod.get("specText") or "",
reduce_qty=prod.get("buyQty") or 0,
user_id=user_id,
)
order.order_status = "yzf"
order.update_by = user_id
order.update_time = now
db.add(order)
db.commit()
db.refresh(order)
return order
# ────────────────────────────────────────────
# Service Order Txn: 操作流水 CRUD
# ────────────────────────────────────────────
def list_service_order_txns(
db,
*,
service_order_id=None,
txn_status=None,
txn_type=None,
keyword=None,
create_time_start=None,
create_time_end=None,
finance_done_only=False,
page=1,
page_size=20,
tenant_no=None,
):
"""查询操作流水列表。finance_done_only:财务已处理(已确认或财务侧作废)。"""
q = db.query(BizChannelServiceOrderTxn).filter(BizChannelServiceOrderTxn.del_flag == 0)
if tenant_no:
q = q.filter(BizChannelServiceOrderTxn.tenant_no == tenant_no)
if service_order_id:
q = q.filter(BizChannelServiceOrderTxn.service_order_id == service_order_id)
if finance_done_only:
q = q.filter(
or_(
BizChannelServiceOrderTxn.txn_status == "yqr",
and_(
BizChannelServiceOrderTxn.txn_status == "yzf",
BizChannelServiceOrderTxn.finance_submit_time.isnot(None),
),
)
)
else:
ss = (txn_status or "").strip()
if ss:
statuses = [s.strip() for s in ss.split(",") if s.strip()]
if statuses:
q = q.filter(BizChannelServiceOrderTxn.txn_status.in_(statuses))
tt = (txn_type or "").strip()
if tt:
types = [t.strip() for t in tt.split(",") if t.strip()]
if types:
q = q.filter(BizChannelServiceOrderTxn.txn_type.in_(types))
kw = (keyword or "").strip()
if kw:
like = f"%{kw}%"
q = q.outerjoin(
BizChannelServiceOrder,
and_(
BizChannelServiceOrder.id == BizChannelServiceOrderTxn.service_order_id,
BizChannelServiceOrder.del_flag == 0,
),
).filter(
or_(
BizChannelServiceOrder.channel_name.like(like),
BizChannelServiceOrder.order_name.like(like),
BizChannelServiceOrder.dept_name.like(like),
)
)
if create_time_start and create_time_start.strip():
q = q.filter(
BizChannelServiceOrderTxn.create_time
>= _normalize_date_bound(create_time_start.strip())
)
if create_time_end and create_time_end.strip():
q = q.filter(
BizChannelServiceOrderTxn.create_time
<= _normalize_date_bound(create_time_end.strip(), end_of_day=True)
)
total = q.order_by(None).with_entities(func.count(BizChannelServiceOrderTxn.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
if finance_done_only:
operate_time = func.coalesce(
BizChannelServiceOrderTxn.finance_confirm_time,
BizChannelServiceOrderTxn.void_time,
)
rows = (
q.order_by(operate_time.desc(), BizChannelServiceOrderTxn.create_time.desc())
.offset((page - 1) * page_size)
.limit(page_size)
.all()
)
else:
rows = (
q.order_by(BizChannelServiceOrderTxn.create_time.desc())
.offset((page - 1) * page_size)
.limit(page_size)
.all()
)
if not rows:
return [], int(total)
order_ids = list({r.service_order_id for r in rows if r.service_order_id})
orders: Dict[str, Any] = {}
if order_ids:
oq = db.query(BizChannelServiceOrder).filter(
BizChannelServiceOrder.id.in_(order_ids),
BizChannelServiceOrder.del_flag == 0,
)
if tenant_no:
oq = oq.filter(BizChannelServiceOrder.tenant_no == tenant_no)
for o in oq.all():
orders[o.id] = o
user_ids: List[str] = []
for r in rows:
if r.create_by:
user_ids.append(r.create_by)
if r.finance_submit_by:
user_ids.append(r.finance_submit_by)
if r.finance_confirm_by:
user_ids.append(r.finance_confirm_by)
elif (r.txn_status or "") == "yqr" and r.update_by:
user_ids.append(r.update_by)
if r.void_by:
user_ids.append(r.void_by)
elif (r.txn_status or "") == "yzf" and r.finance_submit_time and r.update_by:
user_ids.append(r.update_by)
name_map = _user_nick_name_map(db, user_ids) if db and user_ids else {}
items = [
_service_order_txn_to_response_dict(
db,
r,
tenant_no,
order=orders.get(r.service_order_id),
name_map=name_map,
)
for r in rows
]
return items, int(total)
def get_service_order_txn(db, txn_id, *, tenant_no=None):
"""获取单条流水"""
q = db.query(BizChannelServiceOrderTxn).filter(
and_(BizChannelServiceOrderTxn.id == txn_id, BizChannelServiceOrderTxn.del_flag == 0)
)
if tenant_no:
q = q.filter(BizChannelServiceOrderTxn.tenant_no == tenant_no)
return q.first()
def create_recharge_txn(
db,
order_id,
*,
amount,
remark=None,
finance_submit_attachments=None,
tenant_no=None,
user_id=None,
):
"""创建充值流水(yfk订单),状态 dsx;附件预存于 finance_submit_attachments 供后续财务确认"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_type != "yfk":
raise ValueError("仅充值型订单可发起充值")
if order.order_status not in ("dsx", "sxz"):
raise ValueError("订单当前状态不允许充值")
amount = float(amount)
if amount <= 0:
raise ValueError("充值金额必须大于0")
now = datetime.now()
txn = BizChannelServiceOrderTxn(
id=generate_uuid(),
service_order_id=order_id,
txn_type="recharge",
amount=amount,
txn_status="dsx",
remark=remark or "",
finance_submit_attachments=(finance_submit_attachments or "").strip() or None,
tenant_no=tenant_no or order.tenant_no,
order_num=0,
create_by=user_id,
create_dept_id="",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
db.add(txn)
db.commit()
db.refresh(txn)
return txn
def create_refund_txn(db, order_id, *, amount, remark=None, attachments=None, tenant_no=None, user_id=None):
"""创建退款流水(yfk订单),amount <= current_balance"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_type != "yfk":
raise ValueError("仅充值型订单可申请退款")
if order.order_status != "sxz":
raise ValueError("仅生效中订单可申请退款")
amount = float(amount)
if amount <= 0:
raise ValueError("退款金额必须大于0")
current_balance = float(order.current_balance) if order.current_balance else 0
if amount > current_balance:
raise ValueError(f"退款金额不能大于当前余额 {current_balance}")
now = datetime.now()
txn = BizChannelServiceOrderTxn(
id=generate_uuid(),
service_order_id=order_id,
txn_type="refund",
amount=amount,
txn_status="dsx",
remark=remark or "",
refund_attachments=attachments or "",
tenant_no=tenant_no or order.tenant_no,
order_num=0,
create_by=user_id,
create_dept_id="",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
db.add(txn)
db.commit()
db.refresh(txn)
return txn
def create_purchase_txn(db, order_id, *, products, remark=None, tenant_no=None, user_id=None):
"""创建购买产品流水(gmcp订单),同时创建/更新产品持仓"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_type != "gmcp":
raise ValueError("仅购买型订单可购买产品")
if order.order_status not in ("dsx", "sxz"):
raise ValueError("订单当前状态不允许购买")
if not products:
raise ValueError("产品明细不能为空")
now = datetime.now()
tn = tenant_no or order.tenant_no
total_amount = 0.0
# Upsert product holding rows
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
project_name = p.get("projectName") or p.get("project_name") or ""
cell_type = p.get("cellType") or p.get("cell_type") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
agreement_unit_price = float(p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0)
qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
amt = float(p.get("amount") or agreement_unit_price * qty)
total_amount += amt
# Find or create product holding row
existing = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.project_code == project_code,
BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
BizChannelServiceOrderProduct.del_flag == 0,
)
).first()
if existing:
existing.total_buy_qty = float(existing.total_buy_qty or 0) + qty
existing.current_qty = float(existing.current_qty or 0) + qty
existing.update_by = user_id
existing.update_time = now
db.add(existing)
else:
prod_row = BizChannelServiceOrderProduct(
id=generate_uuid(),
service_order_id=order_id,
project_code=project_code,
project_name=project_name,
cell_type=cell_type,
spec_text=spec_text,
agreement_unit_price=agreement_unit_price,
buy_qty=qty,
buy_amount=amt,
total_buy_qty=qty,
total_unsubscribe_qty=0,
current_qty=qty,
total_unsubscribe_amount=0,
tenant_no=tn,
order_num=0,
remark="",
create_by=user_id,
create_dept_id="",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
db.add(prod_row)
db.flush()
products_json = json.dumps(products, ensure_ascii=False)
txn = BizChannelServiceOrderTxn(
id=generate_uuid(),
service_order_id=order_id,
txn_type="purchase",
amount=total_amount,
txn_status="dsx",
products_json=products_json,
remark=remark or "",
tenant_no=tn,
order_num=0,
create_by=user_id,
create_dept_id="",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
order.purchase_total_amount = float(order.purchase_total_amount or 0) + total_amount
order.update_by = user_id
order.update_time = now
db.add(order)
db.add(txn)
db.commit()
db.refresh(txn)
return txn
def create_unsubscribe_txn(db, order_id, *, products, remark=None, tenant_no=None, user_id=None):
"""创建退订产品流水(gmcp订单),校验退订份数 <= current_qty"""
order = get_service_order(db, order_id, tenant_no)
if not order:
raise ValueError("订单不存在")
if order.order_type != "gmcp":
raise ValueError("仅购买型订单可退订产品")
if order.order_status != "sxz":
raise ValueError("仅生效中订单可退订")
if not products:
raise ValueError("退订产品明细不能为空")
now = datetime.now()
tn = tenant_no or order.tenant_no
total_amount = 0.0
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
qty = float(p.get("qty") or 0)
if qty <= 0:
raise ValueError("退订数量必须大于0")
# Find existing product holding
existing = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.project_code == project_code,
BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
BizChannelServiceOrderProduct.del_flag == 0,
)
).first()
if not existing:
raise ValueError(f"产品 {project_code} 不存在于持仓中")
current_qty = float(existing.current_qty or 0)
if qty > current_qty:
raise ValueError(f"退订数量 {qty} 超过当前持有数量 {current_qty}")
unit_price = float(existing.agreement_unit_price or 0)
amt = qty * unit_price
total_amount += amt
existing.total_unsubscribe_qty = float(existing.total_unsubscribe_qty or 0) + qty
existing.current_qty = current_qty - qty
existing.total_unsubscribe_amount = float(existing.total_unsubscribe_amount or 0) + amt
existing.update_by = user_id
existing.update_time = now
db.add(existing)
db.flush()
products_json = json.dumps(products, ensure_ascii=False)
txn = BizChannelServiceOrderTxn(
id=generate_uuid(),
service_order_id=order_id,
txn_type="unsubscribe",
amount=total_amount,
txn_status="dsx",
products_json=products_json,
remark=remark or "",
tenant_no=tn,
order_num=0,
create_by=user_id,
create_dept_id="",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
db.add(txn)
db.commit()
db.refresh(txn)
return txn
def _rollback_purchase_txn_holdings(db, order_id, products, user_id, now=None):
now = now or datetime.now()
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
existing = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.project_code == project_code,
BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
BizChannelServiceOrderProduct.del_flag == 0,
)
).first()
if existing:
existing.total_buy_qty = float(existing.total_buy_qty or 0) - qty
existing.current_qty = float(existing.current_qty or 0) - qty
existing.update_by = user_id
existing.update_time = now
db.add(existing)
def _apply_purchase_txn_holdings(db, order_id, products, tenant_no, user_id, now=None):
now = now or datetime.now()
tn = tenant_no
total_amount = 0.0
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
project_name = p.get("projectName") or p.get("project_name") or ""
cell_type = p.get("cellType") or p.get("cell_type") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
agreement_unit_price = float(p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0)
qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
if qty <= 0:
raise ValueError("购买份数必须大于0")
amt = float(p.get("amount") or agreement_unit_price * qty)
total_amount += amt
existing = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.project_code == project_code,
BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
BizChannelServiceOrderProduct.del_flag == 0,
)
).first()
if existing:
existing.total_buy_qty = float(existing.total_buy_qty or 0) + qty
existing.current_qty = float(existing.current_qty or 0) + qty
existing.update_by = user_id
existing.update_time = now
db.add(existing)
else:
prod_row = BizChannelServiceOrderProduct(
id=generate_uuid(),
service_order_id=order_id,
project_code=project_code,
project_name=project_name,
cell_type=cell_type,
spec_text=spec_text,
agreement_unit_price=agreement_unit_price,
buy_qty=qty,
buy_amount=amt,
total_buy_qty=qty,
total_unsubscribe_qty=0,
current_qty=qty,
total_unsubscribe_amount=0,
tenant_no=tn,
order_num=0,
remark="",
create_by=user_id,
create_dept_id="",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
db.add(prod_row)
db.flush()
return total_amount
def _rollback_unsubscribe_txn_holdings(db, order_id, products, user_id, now=None):
now = now or datetime.now()
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
qty = float(p.get("qty") or 0)
existing = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.project_code == project_code,
BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
BizChannelServiceOrderProduct.del_flag == 0,
)
).first()
if existing:
unit_price = float(existing.agreement_unit_price or 0)
existing.total_unsubscribe_qty = float(existing.total_unsubscribe_qty or 0) - qty
existing.current_qty = float(existing.current_qty or 0) + qty
existing.total_unsubscribe_amount = float(existing.total_unsubscribe_amount or 0) - (qty * unit_price)
existing.update_by = user_id
existing.update_time = now
db.add(existing)
def _apply_unsubscribe_txn_holdings(db, order_id, products, user_id, now=None):
now = now or datetime.now()
total_amount = 0.0
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
qty = float(p.get("qty") or 0)
if qty <= 0:
raise ValueError("退订数量必须大于0")
existing = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order_id,
BizChannelServiceOrderProduct.project_code == project_code,
BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
BizChannelServiceOrderProduct.del_flag == 0,
)
).first()
if not existing:
raise ValueError(f"产品 {project_code} 不存在于持仓中")
current_qty = float(existing.current_qty or 0)
if qty > current_qty:
raise ValueError(f"退订数量 {qty} 超过当前持有数量 {current_qty}")
unit_price = float(existing.agreement_unit_price or 0)
amt = qty * unit_price
total_amount += amt
existing.total_unsubscribe_qty = float(existing.total_unsubscribe_qty or 0) + qty
existing.current_qty = current_qty - qty
existing.total_unsubscribe_amount = float(existing.total_unsubscribe_amount or 0) + amt
existing.update_by = user_id
existing.update_time = now
db.add(existing)
db.flush()
return total_amount
def update_service_order_txn(
db,
txn_id,
*,
amount=None,
remark=None,
finance_submit_attachments=None,
refund_attachments=None,
products=None,
tenant_no=None,
user_id=None,
):
"""更新待生效流水:仅 dsx;充值/退款可改金额与附件,购买/退订可改产品明细与备注"""
txn = get_service_order_txn(db, txn_id, tenant_no=tenant_no)
if not txn:
raise ValueError("流水不存在")
if txn.txn_status != "dsx":
raise ValueError("仅待生效流水可编辑")
now = datetime.now()
txn_type = (txn.txn_type or "").strip()
if amount is not None:
amt = float(amount)
if amt <= 0:
raise ValueError("金额必须大于0")
if txn_type in ("recharge", "refund"):
if txn_type == "refund":
order = get_service_order(db, txn.service_order_id, tenant_no)
if not order:
raise ValueError("关联订单不存在")
current_balance = float(order.current_balance) if order.current_balance else 0
if amt > current_balance:
raise ValueError(f"退款金额不能大于当前余额 {current_balance}")
txn.amount = amt
else:
raise ValueError("购买/退订流水不可直接修改金额,请调整产品明细")
if products is not None:
if txn_type not in ("purchase", "unsubscribe"):
raise ValueError("仅购买/退订流水可修改产品明细")
if not products:
raise ValueError("产品明细不能为空")
order = get_service_order(db, txn.service_order_id, tenant_no)
if not order:
raise ValueError("关联订单不存在")
old_products = json.loads(txn.products_json) if txn.products_json else []
if txn_type == "purchase":
_rollback_purchase_txn_holdings(db, order.id, old_products, user_id, now)
total_amount = _apply_purchase_txn_holdings(
db, order.id, products, tenant_no or order.tenant_no, user_id, now
)
else:
_rollback_unsubscribe_txn_holdings(db, order.id, old_products, user_id, now)
total_amount = _apply_unsubscribe_txn_holdings(db, order.id, products, user_id, now)
txn.amount = total_amount
txn.products_json = json.dumps(products, ensure_ascii=False)
if remark is not None:
txn.remark = remark or ""
if finance_submit_attachments is not None:
txn.finance_submit_attachments = (finance_submit_attachments or "").strip() or None
if refund_attachments is not None:
if txn_type != "refund":
raise ValueError("仅退款流水可修改退款附件")
txn.refund_attachments = (refund_attachments or "").strip() or None
txn.update_by = user_id
txn.update_time = now
db.add(txn)
db.commit()
db.refresh(txn)
return txn
def submit_txn_finance(db, txn_id, *, finance_submit_remark=None, finance_submit_attachments=None, tenant_no=None, user_id=None):
"""发起流水财务确认:dsx → dqr"""
txn = get_service_order_txn(db, txn_id, tenant_no=tenant_no)
if not txn:
raise ValueError("流水不存在")
if txn.txn_status != "dsx":
raise ValueError("仅待生效流水可发起确认")
now = datetime.now()
txn.txn_status = "dqr"
txn.finance_submit_by = user_id
txn.finance_submit_time = now
submit_remark = (finance_submit_remark or "").strip()
if not submit_remark:
submit_remark = (txn.remark or "").strip()
txn.finance_submit_remark = submit_remark
if finance_submit_attachments:
txn.finance_submit_attachments = finance_submit_attachments
elif not (txn.finance_submit_attachments or "").strip():
if (txn.txn_type or "").strip() == "refund" and (txn.refund_attachments or "").strip():
txn.finance_submit_attachments = txn.refund_attachments
else:
txn.finance_submit_attachments = ""
txn.update_by = user_id
txn.update_time = now
db.add(txn)
db.commit()
db.refresh(txn)
return txn
def confirm_txn(db, txn_id, *, finance_remark=None, finance_confirm_attachments=None, tenant_no=None, user_id=None):
"""财务确认流水:dqr → yqr,更新主表汇总和产品持仓,更新渠道账户余额"""
txn = get_service_order_txn(db, txn_id, tenant_no=tenant_no)
if not txn:
raise ValueError("流水不存在")
if txn.txn_status != "dqr":
raise ValueError("仅待确认流水可进行财务确认")
order = get_service_order(db, txn.service_order_id, tenant_no)
if not order:
raise ValueError("关联订单不存在")
now = datetime.now()
if txn.txn_type == "recharge":
# Increase channel account balance
account = _get_or_create_channel_account(db, order.channel_id, tenant_no or order.tenant_no, user_id)
balance_before = float(account.balance_amount) if account.balance_amount else 0
add_amount = float(txn.amount) if txn.amount else 0
account.balance_amount = balance_before + add_amount
balance_after = float(account.balance_amount)
account.update_by = user_id
account.update_time = now
db.add(account)
db.flush()
# Update order summary
order.total_recharge_amount = float(order.total_recharge_amount or 0) + add_amount
order.current_balance = float(order.current_balance or 0) + add_amount
txn.balance_before = balance_before
txn.balance_after = balance_after
elif txn.txn_type == "refund":
# Deduct channel account balance
account = _get_or_create_channel_account(db, order.channel_id, tenant_no or order.tenant_no, user_id)
balance_before = float(account.balance_amount) if account.balance_amount else 0
deduct_amount = float(txn.amount) if txn.amount else 0
account.balance_amount = balance_before - deduct_amount
balance_after = float(account.balance_amount)
account.update_by = user_id
account.update_time = now
db.add(account)
db.flush()
# Update order summary
order.total_refund_amount = float(order.total_refund_amount or 0) + deduct_amount
order.current_balance = float(order.current_balance or 0) - deduct_amount
txn.balance_before = balance_before
txn.balance_after = balance_after
elif txn.txn_type == "purchase":
# Products already updated at create time, just update channel quotas
products = json.loads(txn.products_json) if txn.products_json else []
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
cell_type_code, _ = _resolve_agreement_price_cell_type(
db, order.channel_agreement_id, project_code, spec_text, p.get("cellType") or p.get("cell_type") or ""
)
_upsert_channel_product_quota(
db,
channel_id=order.channel_id,
project_code=project_code,
project_name=p.get("projectName") or p.get("project_name") or "",
cell_type_code=cell_type_code,
cell_type_name=p.get("cellType") or p.get("cell_type") or "",
spec_text=spec_text,
add_qty=qty,
tenant_no=tenant_no or order.tenant_no,
user_id=user_id,
)
elif txn.txn_type == "unsubscribe":
# Products already updated at create time, just reduce channel quotas
products = json.loads(txn.products_json) if txn.products_json else []
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
qty = float(p.get("qty") or 0)
cell_type_code, _ = _resolve_agreement_price_cell_type(
db, order.channel_agreement_id, project_code, spec_text, p.get("cellType") or p.get("cell_type") or ""
)
_reduce_channel_product_quota(
db,
channel_id=order.channel_id,
project_code=project_code,
cell_type_code=cell_type_code,
spec_text=spec_text,
reduce_qty=qty,
user_id=user_id,
)
# Update txn status
txn.txn_status = "yqr"
txn.finance_confirm_time = now
txn.finance_confirm_by = user_id
txn.finance_confirm_attachments = finance_confirm_attachments or ""
txn.finance_remark = finance_remark or ""
txn.update_by = user_id
txn.update_time = now
# Transition order to sxz if currently dsx
if order.order_status == "dsx":
order.order_status = "sxz"
order.update_by = user_id
order.update_time = now
db.add(txn)
db.add(order)
db.commit()
db.refresh(txn)
return txn
def void_txn(db, txn_id, *, tenant_no=None, user_id=None):
"""作废流水:dsx/dqr → yzf。如果是 purchase/unsubscribe 还需要回滚产品持仓"""
txn = get_service_order_txn(db, txn_id, tenant_no=tenant_no)
if not txn:
raise ValueError("流水不存在")
if txn.txn_status not in ("dsx", "dqr"):
raise ValueError("仅待生效或待财务确认流水可作废")
if (txn.txn_type or "").strip() == "settlement":
raise ValueError("输注计划结算流水不可作废")
now = datetime.now()
order = get_service_order(db, txn.service_order_id, tenant_no)
if txn.txn_type == "purchase" and order:
# Rollback product holdings
products = json.loads(txn.products_json) if txn.products_json else []
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
existing = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order.id,
BizChannelServiceOrderProduct.project_code == project_code,
BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
BizChannelServiceOrderProduct.del_flag == 0,
)
).first()
if existing:
existing.total_buy_qty = float(existing.total_buy_qty or 0) - qty
existing.current_qty = float(existing.current_qty or 0) - qty
existing.update_by = user_id
existing.update_time = now
db.add(existing)
txn_amount = float(txn.amount or 0)
if txn_amount > 0:
order.purchase_total_amount = max(
0.0,
float(order.purchase_total_amount or 0) - txn_amount,
)
order.update_by = user_id
order.update_time = now
db.add(order)
elif txn.txn_type == "unsubscribe" and order:
# Rollback unsubscribe: restore product holdings
products = json.loads(txn.products_json) if txn.products_json else []
for p in products:
project_code = p.get("projectCode") or p.get("project_code") or ""
spec_text = p.get("specText") or p.get("spec_text") or ""
qty = float(p.get("qty") or 0)
unit_price = 0
existing = db.query(BizChannelServiceOrderProduct).filter(
and_(
BizChannelServiceOrderProduct.service_order_id == order.id,
BizChannelServiceOrderProduct.project_code == project_code,
BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
BizChannelServiceOrderProduct.del_flag == 0,
)
).first()
if existing:
unit_price = float(existing.agreement_unit_price or 0)
existing.total_unsubscribe_qty = float(existing.total_unsubscribe_qty or 0) - qty
existing.current_qty = float(existing.current_qty or 0) + qty
existing.total_unsubscribe_amount = float(existing.total_unsubscribe_amount or 0) - (qty * unit_price)
existing.update_by = user_id
existing.update_time = now
db.add(existing)
txn.txn_status = "yzf"
txn.void_by = user_id
txn.void_time = now
txn.update_by = user_id
txn.update_time = now
db.add(txn)
db.commit()
db.refresh(txn)
return txn