HHY
f7a555dd6c
后端: - 客户模块 CRUD+搜索 (api/services/repositories 分层) - 合同模块: FG单号规则/意向单转正式单/订单创建自动快照客户信息 - 优化下料: FFD首递减装箱算法(利用率统计/余料/未排入清单) 前端 (Vue3+TS+Element Plus): - 7页面: 登录/工作台/订单中心/合同/客户/样板库/优化下料 - 自研二维窗型画布渲染器: 解析drawing_data渲染玻璃区域/外框/ 扇开向标记/尺寸标注, 支持缩放平移复位 - 样板库与订单联动(用此样板下单自动带图纸) - axios统一envelope拦截/登录守卫/付费墙提示 工程: - SPA history 回退路由, 构建产物挂载后端 - 全接口回归 6/6 通过
56 lines
2.3 KiB
Python
56 lines
2.3 KiB
Python
"""合同数据访问层。"""
|
|
import datetime
|
|
|
|
from sqlalchemy import func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from huamenchuang.models.contract import Contract
|
|
|
|
STATUS_LABELS = {0: "意向单", 1: "正式单", 2: "已完成", 3: "已作废"}
|
|
|
|
|
|
class ContractRepository:
|
|
def __init__(self, db: AsyncSession) -> None:
|
|
self.db = db
|
|
|
|
async def next_contract_no(self, company_id: int) -> str:
|
|
"""合同号：FG+YYYYMMDD+3位流水（对齐目标站 FG20260818009 风格）。"""
|
|
today = datetime.date.today().strftime("%Y%m%d")
|
|
prefix = f"FG{today}"
|
|
stmt = select(Contract.contract_no).where(
|
|
Contract.company_id == company_id, Contract.contract_no.like(f"{prefix}%")
|
|
).order_by(Contract.contract_no.desc()).limit(1)
|
|
last = (await self.db.execute(stmt)).scalar_one_or_none()
|
|
seq = int(last[-3:]) + 1 if last else 1
|
|
return f"{prefix}{seq:03d}"
|
|
|
|
async def get(self, contract_id: int, company_id: int) -> Contract | None:
|
|
stmt = select(Contract).where(Contract.id == contract_id, Contract.company_id == company_id)
|
|
return (await self.db.execute(stmt)).scalar_one_or_none()
|
|
|
|
async def search(self, company_id: int, page: int, page_size: int, keyword: str = "",
|
|
status: int | None = None) -> tuple[list[Contract], int]:
|
|
conds = [Contract.company_id == company_id]
|
|
if status is not None:
|
|
conds.append(Contract.status == status)
|
|
if kw := keyword.strip():
|
|
like = f"%{kw}%"
|
|
conds.append(or_(Contract.contract_no.like(like), Contract.customer_name.like(like), Contract.phone.like(like)))
|
|
where = *conds,
|
|
total = (await self.db.execute(select(func.count()).select_from(Contract).where(*where))).scalar_one()
|
|
stmt = (select(Contract).where(*where).order_by(Contract.id.desc())
|
|
.offset((page - 1) * page_size).limit(page_size))
|
|
return list((await self.db.execute(stmt)).scalars()), total
|
|
|
|
async def create(self, contract: Contract) -> Contract:
|
|
self.db.add(contract)
|
|
await self.db.flush()
|
|
return contract
|
|
|
|
async def delete(self, contract: Contract) -> None:
|
|
await self.db.delete(contract)
|
|
|
|
|
|
def status_label(status: int) -> str:
|
|
return STATUS_LABELS.get(status, "未知")
|