2320 lines
79 KiB
Python
2320 lines
79 KiB
Python
"""中试服务管理 CRUD（细胞产品类型/规格/价格体系/项目价格、渠道信息、体检机构）"""
|
|
from __future__ import annotations
|
|
|
|
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,
|
|
BizChannelAgreement,
|
|
BizChannelAgreementProductPrice,
|
|
BizChannelInvoice,
|
|
BizChannelMailAddress,
|
|
BizCheckupInstitution,
|
|
BizCheckupInstitutionInvoice,
|
|
BizCheckupInstitutionProject,
|
|
BizCheckupInstitutionPackage,
|
|
BizCheckupInstitutionPackageItem,
|
|
)
|
|
|
|
|
|
# ────────────────────────────────────────────
|
|
# 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)))
|