"""运营端接口""" import json import random import string import uuid from typing import Optional, List, Any, Tuple from datetime import datetime from decimal import Decimal from fastapi import APIRouter, Depends, HTTPException, Query, Request, UploadFile, File, Form from sqlalchemy.orm import Session from sqlalchemy import and_, or_, func from sqlalchemy.exc import IntegrityError from app.core.database import get_db from app.core.auth import get_current_user from app.core.security import get_password_hash, verify_password from app.core.response import success_response, error_response from app.core.storage import storage from app.models.models import ( SysUser, SysRegion, BasProductCategory, BasProductUnit, BasProduct, BizOrder, BizOrderItem, BizPriceList, BizPriceListItem, OrdClientSettlement, OrdClientSettlementItem, OrdSupplierSettlement, OrdSupplierSettlementItem, ) from app.api.upload import validate_file, ALLOWED_IMAGE_TYPES, MAX_IMAGE_SIZE from app.crud.region import region_crud from app.crud.category import category_crud from app.crud.unit import unit_crud from app.crud.user import user_crud from app.crud.password_history import password_history_crud from app.crud.product import product_crud from app.crud.address import address_crud from app.core.utils import generate_id router = APIRouter(prefix="/web/admin", tags=["运营端管理接口"]) def _numeric_to_float_optional(v) -> Optional[float]: if v is None: return None return float(v) def _category_path_display(db: Session, category_id: Optional[str]) -> str: """一级/二级分类用 / 连接（自底向上回溯父级）。""" if not category_id: return "" names: List[str] = [] cid = str(category_id).strip() seen = set() depth = 0 while cid and depth < 24 and cid not in seen: seen.add(cid) cat = category_crud.get(db, cid) if not cat: break names.insert(0, (getattr(cat, "name", None) or "").strip()) pid = getattr(cat, "parent_id", None) cid = str(pid).strip() if pid else "" depth += 1 return "/".join([n for n in names if n]) def _supplier_pickup_address_block(db: Session, supplier_id: Optional[str]) -> dict: """供应商默认取货地址；无默认时取第一条 pickup 类型地址。""" empty = {"fullAddress": None, "contactPerson": None, "contactPhone": None} if not supplier_id: return empty addr = address_crud.get_default(db, supplier_id, "pickup") if not addr: rows = address_crud.get_by_user_and_type(db, supplier_id, "pickup") addr = rows[0] if rows else None if not addr: return empty segs = [ getattr(addr, "province", None), getattr(addr, "city", None), getattr(addr, "district", None), getattr(addr, "detail_address", None), ] full = "".join(s for s in segs if s) return { "fullAddress": full or None, "contactPerson": addr.contact_person, "contactPhone": addr.contact_phone, } def require_admin(current_user: dict = Depends(get_current_user)): """验证运营人员权限""" if current_user.get("role") != "admin": raise HTTPException(status_code=403, detail="需要运营人员权限") return current_user def get_admin_region_filter(db: Session, current_user: dict) -> Optional[str]: """获取管理员的区域过滤条件 对于区域管理员，返回其所属区域ID（用于过滤数据） 对于超级管理员（无区域绑定），返回None（不过滤） """ user_id = current_user.get("user_id") if not user_id: return None user = user_crud.get(db, user_id) if not user: return None # 如果管理员绑定了区域ID，则只能查看该区域的数据 if user.region_id: return user.region_id return None def _effective_settlement_region_id( admin_region_id: Optional[str], tab_or_filter_region_id: Optional[str], ) -> Optional[str]: """结算相关接口：区域管理员固定为其绑定区域；超级管理员可按 Tab 的 regionId 筛选。""" if admin_region_id: return admin_region_id return tab_or_filter_region_id or None _REGION_MINUTES_UNSET = object() def _minutes_value_from_request(request: dict, camel: str, snake: str): """解析区域配送相关分钟配置；键不存在返回 _REGION_MINUTES_UNSET，空值为 None。""" if camel not in request and snake not in request: return _REGION_MINUTES_UNSET raw = request.get(camel, request.get(snake)) if raw is None or raw == "": return None try: n = int(raw) except (TypeError, ValueError): raise ValueError("预计配送时长、预警提前时间须为非负整数（分钟）") if n < 0: raise ValueError("预计配送时长、预警提前时间须为非负整数（分钟）") return n def _apply_region_minutes_to_data(request: dict, data: dict): """将 expectedDeliveryMinutes / deliveryWarnAdvanceMinutes 写入 data（字典为 create_data 或 update_data）。失败返回 error_response 的 dict。""" try: ed = _minutes_value_from_request( request, "expectedDeliveryMinutes", "expected_delivery_minutes" ) wa = _minutes_value_from_request( request, "deliveryWarnAdvanceMinutes", "delivery_warn_advance_minutes" ) except ValueError as e: return error_response(400, str(e)) if ed is not _REGION_MINUTES_UNSET: data["expected_delivery_minutes"] = ed if wa is not _REGION_MINUTES_UNSET: data["delivery_warn_advance_minutes"] = wa return None # ============ 区域管理 ============ @router.get("/regions", response_model=dict) def list_regions( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), keyword: Optional[str] = Query(None, description="关键词搜索"), isActive: Optional[bool] = Query(None, description="是否启用"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取区域列表""" query = db.query(SysRegion).filter(SysRegion.is_del == 0) if keyword: query = query.filter( (SysRegion.name.like(f"%{keyword}%")) | (SysRegion.code.like(f"%{keyword}%")) ) if isActive is not None: query = query.filter(SysRegion.is_active == isActive) total = query.count() skip = (page - 1) * pageSize regions = query.order_by(SysRegion.sort_order.asc()).offset(skip).limit(pageSize).all() return success_response({ "list": [{"id": r.id, "name": r.name, "code": r.code, "description": r.description, "sortOrder": r.sort_order, "expectedDeliveryMinutes": r.expected_delivery_minutes, "deliveryWarnAdvanceMinutes": r.delivery_warn_advance_minutes, "isActive": r.is_active, "createAt": str(r.create_at) if r.create_at else None} for r in regions], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/regions/active", response_model=dict) def list_active_regions( db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取所有启用区域（用于下拉选择）""" regions = region_crud.get_active_all(db) return success_response([{"id": r.id, "name": r.name, "code": r.code} for r in regions]) @router.get("/regions/{region_id}", response_model=dict) def get_region( region_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取区域详情""" region = region_crud.get(db, region_id) if not region: return error_response(404, "区域不存在") return success_response({ "id": region.id, "name": region.name, "code": region.code, "description": region.description, "sortOrder": region.sort_order, "expectedDeliveryMinutes": region.expected_delivery_minutes, "deliveryWarnAdvanceMinutes": region.delivery_warn_advance_minutes, "isActive": region.is_active, "createAt": str(region.create_at) if region.create_at else None, "createBy": region.create_by, "createName": region.create_name }) @router.post("/regions", response_model=dict) def create_region( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """创建区域""" # 检查编码唯一性 existing = region_crud.get_by_code(db, request.get("code")) if existing: return error_response(400, "区域编码已存在") # 字段名映射 create_data = { "name": request.get("name"), "code": request.get("code"), "description": request.get("description"), } if "sortOrder" in request or "sort_order" in request: create_data["sort_order"] = request.get("sortOrder", request.get("sort_order", 0)) if "isActive" in request or "is_active" in request: create_data["is_active"] = request.get("isActive", request.get("is_active", True)) err = _apply_region_minutes_to_data(request, create_data) if err is not None: return err creator = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} region = region_crud.create(db, create_data, creator) return success_response({ "id": region.id, "name": region.name, "code": region.code, "description": region.description, "sortOrder": region.sort_order, "expectedDeliveryMinutes": region.expected_delivery_minutes, "deliveryWarnAdvanceMinutes": region.delivery_warn_advance_minutes, "isActive": region.is_active, "createAt": str(region.create_at) if region.create_at else None, "createBy": region.create_by, "createName": region.create_name }) @router.put("/regions/{region_id}", response_model=dict) def update_region( region_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """更新区域""" # 检查编码唯一性（排除自己） if "code" in request and request["code"]: existing = region_crud.get_by_code(db, request["code"]) if existing and existing.id != region_id: return error_response(400, "区域编码已存在") # 字段名映射 update_data = { "name": request.get("name"), "code": request.get("code"), "description": request.get("description"), } if "sortOrder" in request or "sort_order" in request: update_data["sort_order"] = request.get("sortOrder", request.get("sort_order", 0)) if "isActive" in request or "is_active" in request: update_data["is_active"] = request.get("isActive", request.get("is_active", True)) err = _apply_region_minutes_to_data(request, update_data) if err is not None: return err updater = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} region = region_crud.update_with_history(db, region_id, update_data, updater) if not region: return error_response(404, "区域不存在") return success_response({ "id": region.id, "name": region.name, "code": region.code, "description": region.description, "sortOrder": region.sort_order, "expectedDeliveryMinutes": region.expected_delivery_minutes, "deliveryWarnAdvanceMinutes": region.delivery_warn_advance_minutes, "isActive": region.is_active }) @router.delete("/regions/{region_id}", response_model=dict) def delete_region( region_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """删除区域""" success = region_crud.soft_delete_with_history(db, region_id) if not success: return error_response(404, "区域不存在") return success_response({"id": region_id}) # ============ 商品分类管理 ============ def _category_price_coefficient_value(category: BasProductCategory) -> float: v = getattr(category, "price_coefficient", None) if v is None: return 0.0 return float(v) @router.get("/categories", response_model=dict) def list_categories( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), keyword: Optional[str] = Query(None, description="关键词搜索"), isActive: Optional[bool] = Query(None, description="是否启用"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取商品分类列表""" query = db.query(BasProductCategory).filter(BasProductCategory.is_del == 0) if keyword: query = query.filter(BasProductCategory.name.like(f"%{keyword}%")) if isActive is not None: query = query.filter(BasProductCategory.is_active == isActive) total = query.count() skip = (page - 1) * pageSize categories = query.order_by(BasProductCategory.sort_order.asc()).offset(skip).limit(pageSize).all() # 查询所有分类用于获取父分类名称 all_categories = db.query(BasProductCategory).filter(BasProductCategory.is_del == 0).all() parent_map = {c.id: c.name for c in all_categories} return success_response({ "list": [{ "id": c.id, "name": c.name, "parentId": c.parent_id, "parentName": parent_map.get(c.parent_id) if c.parent_id else None, "sortOrder": c.sort_order, "priceCoefficient": _category_price_coefficient_value(c), "isActive": c.is_active, "imageUrl": getattr(c, "image_url", None), } for c in categories], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/categories/tree", response_model=dict) def get_category_tree( db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取商品分类树形结构""" tree = category_crud.get_tree(db) return success_response(tree) @router.get("/categories/active", response_model=dict) def list_active_categories( db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取所有启用分类""" categories = category_crud.get_active_all(db) return success_response([{ "id": c.id, "name": c.name, "parentId": c.parent_id, "priceCoefficient": _category_price_coefficient_value(c), } for c in categories]) @router.get("/categories/{category_id}", response_model=dict) def get_category( category_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取分类详情""" category = category_crud.get(db, category_id) if not category: return error_response(404, "分类不存在") return success_response({ "id": category.id, "name": category.name, "parentId": category.parent_id, "sortOrder": category.sort_order, "priceCoefficient": _category_price_coefficient_value(category), "isActive": category.is_active, "imageUrl": getattr(category, "image_url", None), }) @router.post("/categories", response_model=dict) def create_category( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """创建商品分类""" creator = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} create_data = { "name": request.get("name"), } # 前端传参使用驼峰命名：parentId/sortOrder/isActive/imageUrl if "parentId" in request or "parent_id" in request: v = request.get("parentId", request.get("parent_id")) create_data["parent_id"] = v or None if "sortOrder" in request or "sort_order" in request: create_data["sort_order"] = request.get("sortOrder", request.get("sort_order")) if "isActive" in request or "is_active" in request: create_data["is_active"] = request.get("isActive", request.get("is_active")) if "imageUrl" in request or "image_url" in request: v = request.get("imageUrl", request.get("image_url")) create_data["image_url"] = v or None if "priceCoefficient" in request or "price_coefficient" in request: raw = request.get("priceCoefficient", request.get("price_coefficient")) if raw is not None and raw != "": try: create_data["price_coefficient"] = float(raw) except (TypeError, ValueError): return error_response(400, "价格系数格式不正确") category = category_crud.create(db, create_data, creator) return success_response({ "id": category.id, "name": category.name, "parentId": category.parent_id, "sortOrder": category.sort_order, "priceCoefficient": _category_price_coefficient_value(category), "isActive": category.is_active, "imageUrl": getattr(category, "image_url", None), }) @router.put("/categories/{category_id}", response_model=dict) def update_category( category_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """更新商品分类""" updater = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} update_data: dict = {} if "name" in request: update_data["name"] = request.get("name") if "parentId" in request or "parent_id" in request: v = request.get("parentId", request.get("parent_id")) update_data["parent_id"] = v or None if "sortOrder" in request or "sort_order" in request: update_data["sort_order"] = request.get("sortOrder", request.get("sort_order")) if "isActive" in request or "is_active" in request: update_data["is_active"] = request.get("isActive", request.get("is_active")) if "imageUrl" in request or "image_url" in request: v = request.get("imageUrl", request.get("image_url")) update_data["image_url"] = v or None if "priceCoefficient" in request or "price_coefficient" in request: raw = request.get("priceCoefficient", request.get("price_coefficient")) if raw is None or raw == "": update_data["price_coefficient"] = 0.0 else: try: update_data["price_coefficient"] = float(raw) except (TypeError, ValueError): return error_response(400, "价格系数格式不正确") category = category_crud.update_with_history(db, category_id, update_data, updater) if not category: return error_response(404, "分类不存在") return success_response({ "id": category.id, "name": category.name, "parentId": category.parent_id, "sortOrder": category.sort_order, "priceCoefficient": _category_price_coefficient_value(category), "isActive": category.is_active, "imageUrl": getattr(category, "image_url", None), }) @router.delete("/categories/{category_id}", response_model=dict) def delete_category( category_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """删除商品分类""" success = category_crud.soft_delete_with_history(db, category_id) if not success: return error_response(404, "分类不存在") return success_response({"id": category_id}) @router.post("/categories/upload-image", response_model=dict) async def upload_category_image( file: UploadFile = File(...), folder: str = Form("categories"), current_user: dict = Depends(require_admin), ): """上传商品分类图片（MinIO）并返回图片URL""" try: content = await validate_file(file, MAX_IMAGE_SIZE, ALLOWED_IMAGE_TYPES) original_filename = file.filename or "category.png" url = storage.upload_file(content, original_filename, folder) return success_response({"url": url}) except HTTPException: raise except Exception as e: return error_response(500, f"上传失败: {str(e)}") # ============ 商品单位管理 ============ @router.get("/units", response_model=dict) def list_units( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), keyword: Optional[str] = Query(None, description="关键词搜索"), isActive: Optional[bool] = Query(None, description="是否启用"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取商品单位列表""" query = db.query(BasProductUnit).filter(BasProductUnit.is_del == 0) if keyword: query = query.filter( (BasProductUnit.name.like(f"%{keyword}%")) | (BasProductUnit.code.like(f"%{keyword}%")) ) if isActive is not None: query = query.filter(BasProductUnit.is_active == isActive) total = query.count() skip = (page - 1) * pageSize units = query.order_by(BasProductUnit.sort_order.asc(), BasProductUnit.create_at.desc()).offset(skip).limit(pageSize).all() return success_response({ "list": [{"id": u.id, "name": u.name, "code": u.code, "isActive": u.is_active, "sortOrder": u.sort_order} for u in units], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/units/active", response_model=dict) def list_active_units( db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取所有启用单位""" units = unit_crud.get_active_all(db) return success_response([{"id": u.id, "name": u.name, "code": u.code} for u in units]) @router.get("/units/{unit_id}", response_model=dict) def get_unit( unit_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取单位详情""" unit = unit_crud.get(db, unit_id) if not unit: return error_response(404, "单位不存在") return success_response({ "id": unit.id, "name": unit.name, "code": unit.code, "isActive": unit.is_active, "sortOrder": unit.sort_order }) @router.post("/units", response_model=dict) def create_unit( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """创建商品单位""" # 检查编码唯一性 existing = unit_crud.get_by_code(db, request.get("code")) if existing: return error_response(400, "单位编码已存在") creator = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} unit = unit_crud.create(db, request, creator) return success_response({ "id": unit.id, "name": unit.name, "code": unit.code, "isActive": unit.is_active, "sortOrder": unit.sort_order }) @router.put("/units/{unit_id}", response_model=dict) def update_unit( unit_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """更新商品单位""" # 检查编码唯一性 if "code" in request and request["code"]: existing = unit_crud.get_by_code(db, request["code"]) if existing and existing.id != unit_id: return error_response(400, "单位编码已存在") updater = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} unit = unit_crud.update_with_history(db, unit_id, request, updater) if not unit: return error_response(404, "单位不存在") return success_response({ "id": unit.id, "name": unit.name, "code": unit.code, "isActive": unit.is_active, "sortOrder": unit.sort_order }) @router.delete("/units/{unit_id}", response_model=dict) def delete_unit( unit_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """删除商品单位""" success = unit_crud.soft_delete_with_history(db, unit_id) if not success: return error_response(404, "单位不存在") return success_response({"id": unit_id}) # ============ 用户管理 ============ # 角色定义常量 ROLE_OPTIONS = [ {"label": "运营管理员", "value": "admin"}, {"label": "供应商", "value": "supplier"}, {"label": "客户", "value": "client"}, {"label": "配送员", "value": "deliver"} ] @router.get("/roles", response_model=dict) def list_roles( current_user: dict = Depends(require_admin) ): """获取角色列表""" return success_response(ROLE_OPTIONS) @router.get("/users", response_model=dict) def list_users( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), role: Optional[str] = Query(None, description="角色筛选"), regionId: Optional[str] = Query(None, description="区域ID"), keyword: Optional[str] = Query(None, description="关键词搜索"), isActive: Optional[bool] = Query(None, description="是否激活"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取用户列表""" users, total = user_crud.get_multi_with_filters( db, role=role, region_id=regionId, keyword=keyword, is_active=isActive, skip=(page - 1) * pageSize, limit=pageSize ) region_map = {} for user in users: if user.region_id and user.region_id not in region_map: region = region_crud.get(db, user.region_id) if region: region_map[user.region_id] = region.name return success_response({ "list": [{ "id": u.id, "username": u.username, "realName": u.real_name, "phone": u.phone, "avatarUrl": (str(u.avatar_url).strip() if getattr(u, "avatar_url", None) else None) or None, "role": u.role, "regionId": u.region_id, "regionName": region_map.get(u.region_id), "isActive": u.is_active, "createAt": str(u.create_at) if u.create_at else None, "createName": getattr(u, "create_name", None) } for u in users], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/users/{user_id}", response_model=dict) def get_user( user_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取用户详情""" user = user_crud.get(db, user_id) if not user: return error_response(404, "用户不存在") region_name = None if user.region_id: region = region_crud.get(db, user.region_id) region_name = region.name if region else None return success_response({ "id": user.id, "username": user.username, "realName": user.real_name, "phone": user.phone, "avatarUrl": (str(user.avatar_url).strip() if getattr(user, "avatar_url", None) else None) or None, "role": user.role, "regionId": user.region_id, "regionName": region_name, "isActive": user.is_active, "createAt": str(user.create_at) if user.create_at else None, "createName": getattr(user, "create_name", None) }) @router.post("/users", response_model=dict) def create_user( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """创建用户""" # 检查用户名唯一性 existing = user_crud.get_by_username(db, request.get("username")) if existing: return error_response(400, "用户名已存在") creator = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} body = dict(request) if request else {} if body.get("avatarUrl") == "": body["avatarUrl"] = None try: user, initial_password = user_crud.create_with_password(db, body, creator) except ValueError as e: return error_response(400, str(e)) payload = { "id": user.id, "username": user.username, "realName": user.real_name, "avatarUrl": (str(user.avatar_url).strip() if getattr(user, "avatar_url", None) else None) or None, } if initial_password is not None: payload["initialPassword"] = initial_password return success_response(payload) @router.put("/users/{user_id}", response_model=dict) def update_user( user_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """更新用户""" updater = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} req = dict(request) if request else {} if req.get("avatarUrl") == "": req["avatarUrl"] = None user = user_crud.update_with_history(db, user_id, req, updater) if not user: return error_response(404, "用户不存在") region_name = None if user.region_id: region = region_crud.get(db, user.region_id) region_name = region.name if region else None return success_response({ "id": user.id, "username": user.username, "realName": user.real_name, "phone": user.phone, "avatarUrl": (str(user.avatar_url).strip() if getattr(user, "avatar_url", None) else None) or None, "role": user.role, "regionId": user.region_id, "regionName": region_name, "isActive": user.is_active }) @router.delete("/users/{user_id}", response_model=dict) def delete_user( user_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """删除用户""" # 不能删除自己 if user_id == current_user.get("user_id"): return error_response(400, "不能删除当前登录账号") success = user_crud.soft_delete_with_history(db, user_id) if not success: return error_response(404, "用户不存在") return success_response({"id": user_id}) @router.post("/users/upload-avatar", response_model=dict) async def upload_user_avatar( file: UploadFile = File(...), folder: str = Form("avatars"), current_user: dict = Depends(require_admin), ): """上传用户头像图片，返回可写入 avatarUrl 的地址""" try: content = await validate_file(file, MAX_IMAGE_SIZE, ALLOWED_IMAGE_TYPES) original_filename = file.filename or "avatar.png" url = storage.upload_file(content, original_filename, folder) return success_response({"url": url}) except HTTPException: raise except Exception as e: return error_response(500, f"上传失败: {str(e)}") @router.post("/users/{user_id}/reset-password", response_model=dict) def reset_user_password( user_id: str, request: Request, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """重置用户密码为服务端固定默认（明文不由请求传入，响应中也不返回密码）""" try: success = user_crud.reset_password_to_admin_default(db, user_id) except ValueError as e: return error_response(400, str(e)) if not success: return error_response(404, "用户不存在") ip_address = get_client_ip(request) password_history_crud.create_password_history( db, user_id=user_id, operation_type="reset", operator_id=current_user.get("user_id"), operator_name=current_user.get("real_name"), ip_address=ip_address ) return success_response({"ok": True}) @router.get("/users/{user_id}/password-history", response_model=dict) def get_user_password_history( user_id: str, page: int = Query(1, ge=1, description="页码"), page_size: int = Query(10, ge=1, le=50, description="每页条数"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取用户密码修改历史记录""" try: from app.models.models import SysPasswordHistory query = db.query(SysPasswordHistory).filter( SysPasswordHistory.user_id == user_id, SysPasswordHistory.is_del == 0 ) total = query.count() items = query.order_by(SysPasswordHistory.create_at.desc()).offset((page - 1) * page_size).limit(page_size).all() result = [] for item in items: result.append({ "id": item.id, "userId": item.user_id, "operationType": item.operation_type, "operationTypeText": "重置密码" if item.operation_type == "reset" else "修改密码" if item.operation_type == "change" else item.operation_type, "operatorName": item.operator_name, "operatorId": item.operator_id, "ipAddress": item.ip_address, "createAt": str(item.create_at) if item.create_at else None }) return success_response({ "list": result, "total": total, "page": page, "pageSize": page_size }) except Exception as e: return error_response(500, f"获取密码历史失败: {str(e)}") @router.get("/password-history", response_model=dict) def get_password_history_list( page: int = Query(1, ge=1, description="页码"), page_size: int = Query(10, ge=1, le=50, description="每页条数"), user_id: str = Query(None, description="用户ID"), operation_type: str = Query(None, description="操作类型"), keyword: str = Query(None, description="关键词搜索"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取密码修改历史记录列表（后台管理用）""" try: from app.models.models import SysPasswordHistory query = db.query(SysPasswordHistory).filter(SysPasswordHistory.is_del == 0) if user_id: query = query.filter(SysPasswordHistory.user_id == user_id) if operation_type: query = query.filter(SysPasswordHistory.operation_type == operation_type) if keyword: query = query.filter(SysPasswordHistory.operator_name.like(f"%{keyword}%")) total = query.count() items = query.order_by(SysPasswordHistory.create_at.desc()).offset((page - 1) * page_size).limit(page_size).all() result = [] for item in items: # 获取用户信息 user_info = {} user = user_crud.get(db, item.user_id) if user: user_info = { "userName": user.username, "realName": user.real_name, "role": user.role } result.append({ "id": item.id, "userId": item.user_id, "userInfo": user_info, "operationType": item.operation_type, "operationTypeText": "重置密码" if item.operation_type == "reset" else "修改密码" if item.operation_type == "change" else item.operation_type, "operatorName": item.operator_name, "operatorId": item.operator_id, "ipAddress": item.ip_address, "createAt": str(item.create_at) if item.create_at else None }) return success_response({ "list": result, "total": total, "page": page, "pageSize": page_size }) except Exception as e: return error_response(500, f"获取密码历史失败: {str(e)}") @router.post("/users/change-password", response_model=dict) def change_password( body: dict, request: Request, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """用户修改密码""" old_password = body.get("oldPassword") new_password = body.get("newPassword") if not old_password or not new_password: return error_response(400, "旧密码和新密码都不能为空") if len(new_password) < 6: return error_response(400, "新密码长度不能少于6位") user = user_crud.get(db, current_user.get("user_id")) if not user: return error_response(404, "用户不存在") if not verify_password(old_password, user.password_hash): return error_response(400, "旧密码错误") user.password_hash = get_password_hash(new_password) db.commit() ip_address = get_client_ip(request) password_history_crud.create_password_history( db, user_id=user.id, operation_type="modify", operator_id=user.id, operator_name=user.real_name, ip_address=ip_address ) return success_response({"message": "密码修改成功"}) # ============ 商品管理 ============ @router.get("/products", response_model=dict) def list_products( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=1000, description="每页条数"), categoryId: Optional[str] = Query(None, description="分类ID"), keyword: Optional[str] = Query(None, description="关键词搜索"), isActive: Optional[bool] = Query(None, description="是否启用"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取商品列表""" products, total = product_crud.get_multi_with_filters( db, category_id=categoryId, keyword=keyword, is_active=isActive, skip=(page - 1) * pageSize, limit=pageSize ) # 获取分类和单位名称映射 category_map = {} unit_map = {} for p in products: if p.category_id and p.category_id not in category_map: cat = category_crud.get(db, p.category_id) if cat: category_map[p.category_id] = cat.name if p.unit_id and p.unit_id not in unit_map: u = unit_crud.get(db, p.unit_id) if u: unit_map[p.unit_id] = u.name return success_response({ "list": [{ "id": p.id, "name": p.name, "skuCode": p.sku_code, "categoryId": p.category_id, "categoryName": category_map.get(p.category_id), "unitId": p.unit_id, "unitName": unit_map.get(p.unit_id), "measureValue": _numeric_to_float_optional(getattr(p, "measure_value", None)), "description": p.description, "detailContent": p.detail_content, "imageUrl": p.image_url, "priceCoefficient": _numeric_to_float_optional(getattr(p, "price_coefficient", None)) or 0.0, "isActive": p.is_active, "createAt": str(p.create_at) if p.create_at else None } for p in products], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/products/{product_id}", response_model=dict) def get_product( product_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取商品详情""" product = product_crud.get(db, product_id) if not product: return error_response(404, "商品不存在") category_name = None if product.category_id: cat = category_crud.get(db, product.category_id) category_name = cat.name if cat else None unit_name = None if product.unit_id: u = unit_crud.get(db, product.unit_id) unit_name = u.name if u else None return success_response({ "id": product.id, "name": product.name, "skuCode": product.sku_code, "categoryId": product.category_id, "categoryName": category_name, "unitId": product.unit_id, "unitName": unit_name, "measureValue": _numeric_to_float_optional(getattr(product, "measure_value", None)), "description": product.description, "detailContent": product.detail_content, "imageUrl": product.image_url, "priceCoefficient": _numeric_to_float_optional(getattr(product, "price_coefficient", None)) or 0.0, "isActive": product.is_active, "createAt": str(product.create_at) if product.create_at else None }) @router.get("/products/{product_id}/region-prices", response_model=dict) def get_product_region_prices( product_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """各启用区域下，当前生效价格清单中该商品的单价（无生效价则 price 为空）""" from app.crud.price_list import price_list_item_crud product = product_crud.get(db, product_id) if not product: return error_response(404, "商品不存在") regions = region_crud.get_active_all(db) admin_region_id = get_admin_region_filter(db, current_user) if admin_region_id: regions = [r for r in regions if r.id == admin_region_id] rows = [] for r in regions: item = price_list_item_crud.get_product_price(db, r.id, product_id) if item: unit_name = None if item.unit_id: u = unit_crud.get(db, item.unit_id) unit_name = u.name if u else None rows.append({ "regionId": r.id, "regionName": r.name, "price": float(item.price) if item.price is not None else None, "unitId": item.unit_id, "unitName": unit_name }) else: rows.append({ "regionId": r.id, "regionName": r.name, "price": None, "unitId": None, "unitName": None }) return success_response({"list": rows}) @router.post("/products", response_model=dict) def create_product( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """创建商品""" creator = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} try: product = product_crud.create(db, request, creator) except IntegrityError as e: db.rollback() err = str(getattr(e.orig, "args", e) or e) if "uk_sku_code" in err or "sku_code" in err.lower(): return error_response(400, "SKU 编码已存在，请修改或点击「自动生成」") raise return success_response({ "id": product.id, "name": product.name }) @router.post("/products/upload-image", response_model=dict) async def upload_product_image( file: UploadFile = File(...), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """上传商品图片/轮播图（支持图片和视频）""" try: # 验证文件类型 allowed_image_types = {"image/jpeg", "image/png", "image/gif", "image/webp", "image/heic", "image/heif"} allowed_video_types = {"video/mp4", "video/quicktime", "video/webm", "video/x-msvideo"} allowed_types = allowed_image_types | allowed_video_types if file.content_type not in allowed_types: return error_response(400, "不支持的文件格式，支持：JPG、PNG、GIF、WebP、HEIC、MP4") # 验证文件大小 (图片10MB，视频100MB) max_size = 100 * 1024 * 1024 if file.content_type.startswith("video/") else 10 * 1024 * 1024 content = b"" chunk_size = 1024 * 1024 while True: chunk = await file.read(chunk_size) if not chunk: break content += chunk if len(content) > max_size: return error_response(400, f"{'视频' if file.content_type.startswith('video/') else '图片'}大小不能超过{'100MB' if file.content_type.startswith('video/') else '10MB'}") await file.seek(0) # 确定文件后缀和上传目录 if file.content_type.startswith("video/"): ext = "mp4" if file.content_type == "video/mp4" else "mov" if file.content_type == "video/quicktime" else "webm" if file.content_type == "video/webm" else "avi" folder = "products/banners" filename = f"video_{uuid.uuid4().hex[:8]}.{ext}" else: ext = file.filename.rsplit(".", 1)[-1] if "." in (file.filename or "") else "jpg" folder = "products" filename = file.filename or f"image_{uuid.uuid4().hex[:8]}.{ext}" # 上传文件 url = storage.upload_file(content, filename, folder) return success_response({ "url": url, "filename": file.filename, "contentType": file.content_type, "size": len(content) }) except Exception as e: return error_response(500, f"上传失败: {str(e)}") @router.post("/products/upload-base64", response_model=dict) async def upload_product_image_base64( data: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """上传商品图片（Base64方式）""" try: base64_data = data.get("base64") if not base64_data: return error_response(400, "base64数据不能为空") # 去除前缀 if "," in base64_data: base64_data = base64_data.split(",")[1] # 验证图片大小 decoded_len = len(base64_data) * 3 // 4 if decoded_len > 5 * 1024 * 1024: return error_response(400, "图片大小不能超过5MB") url = storage.upload_base64(base64_data, "product.png", "products") return success_response({ "url": url, "filename": "product.png" }) except Exception as e: return error_response(500, f"上传失败: {str(e)}") @router.put("/products/{product_id}", response_model=dict) def update_product( product_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """更新商品""" updater = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} product = product_crud.update_with_history(db, product_id, request, updater) if not product: return error_response(404, "商品不存在") return success_response({ "id": product.id, "name": product.name, "skuCode": product.sku_code, "categoryId": product.category_id, "unitId": product.unit_id, "measureValue": _numeric_to_float_optional(getattr(product, "measure_value", None)), "description": product.description, "detailContent": product.detail_content, "imageUrl": product.image_url, "priceCoefficient": _numeric_to_float_optional(getattr(product, "price_coefficient", None)) or 0.0, "isActive": product.is_active }) @router.delete("/products/{product_id}", response_model=dict) def delete_product( product_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """删除商品""" success = product_crud.soft_delete_with_history(db, product_id) if not success: return error_response(404, "商品不存在") return success_response({"id": product_id}) # ============ 询价单管理 ============ @router.get("/inquiries", response_model=dict) def list_inquiries( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), regionId: Optional[str] = Query(None, description="区域ID"), status: Optional[str] = Query(None, description="状态: draft/published/closed"), keyword: Optional[str] = Query(None, description="关键词搜索"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取询价单列表""" from app.crud.inquiry import inquiry_crud # 区域管理员只能查看自己区域的询价单 admin_region_id = get_admin_region_filter(db, current_user) if admin_region_id and not regionId: regionId = admin_region_id elif admin_region_id and regionId != admin_region_id: return success_response({ "list": [], "total": 0, "page": page, "pageSize": pageSize, "totalPages": 0 }) inquiries, total = inquiry_crud.get_multi_with_filters( db, region_id=regionId, status=status, keyword=keyword, skip=(page - 1) * pageSize, limit=pageSize ) region_map = {} for inquiry in inquiries: if inquiry.region_id and inquiry.region_id not in region_map: region = region_crud.get(db, inquiry.region_id) if region: region_map[inquiry.region_id] = region.name return success_response({ "list": [{ "id": i.id, "regionId": i.region_id, "regionName": region_map.get(i.region_id), "title": i.title, "status": i.status, "publishTime": str(i.publish_time) if i.publish_time else None, "closeTime": str(i.close_time) if i.close_time else None, "remark": i.remark, "createAt": str(i.create_at) if i.create_at else None, "createName": i.create_name } for i in inquiries], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/inquiries/{inquiry_id}", response_model=dict) def get_inquiry( inquiry_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取询价单详情""" from app.crud.inquiry import inquiry_crud, inquiry_item_crud from app.crud.product import product_crud inquiry = inquiry_crud.get(db, inquiry_id) if not inquiry: return error_response(404, "询价单不存在") items = inquiry_item_crud.get_by_inquiry(db, inquiry_id) region_name = None if inquiry.region_id: region = region_crud.get(db, inquiry.region_id) region_name = region.name if region else None item_details = [] for item in items: product = product_crud.get(db, item.product_id) category_name = None if product and product.category_id: cat = category_crud.get(db, product.category_id) category_name = cat.name if cat else None unit_name = None if product and product.unit_id: u = unit_crud.get(db, product.unit_id) unit_name = u.name if u else None item_details.append({ "id": item.id, "productId": item.product_id, "productName": product.name if product else None, "categoryId": product.category_id if product else None, "categoryName": category_name, "unitName": unit_name, "description": product.description if product else None, "imageUrl": product.image_url if product else None, "createAt": str(item.create_at) if item.create_at else None }) return success_response({ "id": inquiry.id, "regionId": inquiry.region_id, "regionName": region_name, "title": inquiry.title, "status": inquiry.status, "publishTime": str(inquiry.publish_time) if inquiry.publish_time else None, "closeTime": str(inquiry.close_time) if inquiry.close_time else None, "remark": inquiry.remark, "createAt": str(inquiry.create_at) if inquiry.create_at else None, "createName": inquiry.create_name, "items": item_details }) @router.post("/inquiries", response_model=dict) def create_inquiry( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """创建询价单""" from app.crud.inquiry import inquiry_crud, inquiry_item_crud creator = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} inquiry_data = { "region_id": request.get("regionId"), "title": request.get("title"), "remark": request.get("remark"), "status": "draft" } inquiry = inquiry_crud.create(db, inquiry_data, creator) product_ids = request.get("productIds", []) if product_ids: inquiry_item_crud.create_items(db, inquiry.id, product_ids, creator) region_name = None if inquiry.region_id: region = region_crud.get(db, inquiry.region_id) region_name = region.name if region else None return success_response({ "id": inquiry.id, "regionId": inquiry.region_id, "regionName": region_name, "title": inquiry.title, "status": inquiry.status }) @router.put("/inquiries/{inquiry_id}", response_model=dict) def update_inquiry( inquiry_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """更新询价单""" from app.crud.inquiry import inquiry_crud, inquiry_item_crud inquiry = inquiry_crud.get(db, inquiry_id) if not inquiry: return error_response(404, "询价单不存在") if inquiry.status != "draft": return error_response(400, "只能编辑草稿状态的询价单") updater = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} update_data = {} if "title" in request: update_data["title"] = request["title"] if "remark" in request: update_data["remark"] = request["remark"] if update_data: inquiry = inquiry_crud.update_with_history(db, inquiry_id, update_data, updater) if "productIds" in request: inquiry_item_crud.update_items(db, inquiry_id, request["productIds"], updater) region_name = None if inquiry.region_id: region = region_crud.get(db, inquiry.region_id) region_name = region.name if region else None return success_response({ "id": inquiry.id, "regionId": inquiry.region_id, "regionName": region_name, "title": inquiry.title, "status": inquiry.status }) @router.delete("/inquiries/{inquiry_id}", response_model=dict) def delete_inquiry( inquiry_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """删除询价单""" from app.crud.inquiry import inquiry_crud, inquiry_item_crud inquiry = inquiry_crud.get(db, inquiry_id) if not inquiry: return error_response(404, "询价单不存在") if inquiry.status == "published": return error_response(400, "已发布的询价单不能删除") inquiry_item_crud.delete_by_inquiry(db, inquiry_id) inquiry_crud.soft_delete_with_history(db, inquiry_id) return success_response({"id": inquiry_id}) @router.post("/inquiries/{inquiry_id}/publish", response_model=dict) def publish_inquiry( inquiry_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """发布询价单""" from app.crud.inquiry import inquiry_crud, inquiry_item_crud from app.models.models import BizInquiry from sqlalchemy import and_ inquiry = inquiry_crud.get(db, inquiry_id) if not inquiry: return error_response(404, "询价单不存在") if inquiry.status != "draft": return error_response(400, "只能发布草稿状态的询价单") items = inquiry_item_crud.get_by_inquiry(db, inquiry_id) if not items: return error_response(400, "询价单必须包含至少一个商品") # 同区域仅允许一个发布中的询价单（并发下尽量避免双发布） try: # 锁住同区域发布中的记录（若存在）与当前记录 db.query(BizInquiry).filter( and_( BizInquiry.region_id == inquiry.region_id, BizInquiry.status == "published", BizInquiry.is_del == 0 ) ).with_for_update().all() db.query(BizInquiry).filter(BizInquiry.id == inquiry_id).with_for_update().all() except Exception: # with_for_update 在某些配置下可能不生效，但不影响基本校验 pass existing = inquiry_crud.get_current_published_by_region(db, inquiry.region_id) if existing and existing.id != inquiry_id: return error_response(400, "该区域已存在发布中的询价单，请先取消发布当前询价单后再发布新的询价单") inquiry = inquiry_crud.publish(db, inquiry_id) return success_response({ "id": inquiry.id, "status": inquiry.status, "publishTime": str(inquiry.publish_time) if inquiry.publish_time else None }) @router.post("/inquiries/{inquiry_id}/unpublish", response_model=dict) def unpublish_inquiry( inquiry_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """取消发布询价单（published -> draft）""" from app.crud.inquiry import inquiry_crud inquiry = inquiry_crud.get(db, inquiry_id) if not inquiry: return error_response(404, "询价单不存在") if inquiry.status != "published": return error_response(400, "只能取消发布已发布状态的询价单") inquiry = inquiry_crud.unpublish(db, inquiry_id) return success_response({ "id": inquiry.id, "status": inquiry.status }) @router.post("/inquiries/{inquiry_id}/close", response_model=dict) def close_inquiry( inquiry_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """关闭询价单""" from app.crud.inquiry import inquiry_crud inquiry = inquiry_crud.get(db, inquiry_id) if not inquiry: return error_response(404, "询价单不存在") if inquiry.status != "published": return error_response(400, "只能关闭已发布的询价单") inquiry = inquiry_crud.close(db, inquiry_id) return success_response({ "id": inquiry.id, "status": inquiry.status, "closeTime": str(inquiry.close_time) if inquiry.close_time else None }) # ============ 报价管理（无审批） ============ @router.get("/quotations", response_model=dict) def list_quotations( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), inquiryId: Optional[str] = Query(None, description="询价单ID"), regionId: Optional[str] = Query(None, description="区域ID"), status: Optional[str] = Query(None, description="状态: submitted/selected"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取报价单列表""" from app.crud.quotation import quotation_crud, quotation_item_crud from app.crud.inquiry import inquiry_crud from app.crud.user import user_crud # 区域管理员只能查看自己区域的报价单 admin_region_id = get_admin_region_filter(db, current_user) if admin_region_id and not regionId: regionId = admin_region_id elif admin_region_id and regionId != admin_region_id: return success_response({ "list": [], "total": 0, "page": page, "pageSize": pageSize, "totalPages": 0 }) quotations, total = quotation_crud.get_multi_with_filters( db, inquiry_id=inquiryId, region_id=regionId, status=status, skip=(page - 1) * pageSize, limit=pageSize ) region_map = {} inquiry_map = {} supplier_map = {} for q in quotations: if q.region_id and q.region_id not in region_map: region = region_crud.get(db, q.region_id) region_map[q.region_id] = region.name if region else None if q.inquiry_id and q.inquiry_id not in inquiry_map: inquiry = inquiry_crud.get(db, q.inquiry_id) inquiry_map[q.inquiry_id] = inquiry.title if inquiry else None if q.supplier_id and q.supplier_id not in supplier_map: supplier = user_crud.get(db, q.supplier_id) supplier_map[q.supplier_id] = supplier.real_name if supplier else None return success_response({ "list": [{ "id": q.id, "inquiryId": q.inquiry_id, "inquiryTitle": inquiry_map.get(q.inquiry_id), "supplierId": q.supplier_id, "supplierName": supplier_map.get(q.supplier_id), "regionId": q.region_id, "regionName": region_map.get(q.region_id), "status": q.status, "totalAmount": float(q.total_amount) if q.total_amount else 0, "submitTime": str(q.submit_time) if q.submit_time else None, "createAt": str(q.create_at) if q.create_at else None } for q in quotations], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/quotations/{quotation_id}", response_model=dict) def get_quotation( quotation_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取报价单详情""" from app.crud.quotation import quotation_crud, quotation_item_crud from app.crud.inquiry import inquiry_crud from app.crud.user import user_crud from app.crud.product import product_crud from app.crud.category import category_crud from app.models.models import BizPriceList, BizPriceListItem, BizInquiryItem from sqlalchemy import and_ quotation = quotation_crud.get(db, quotation_id) if not quotation: return error_response(404, "报价单不存在") # 询价单商品作为主表，确保供应商未报价商品也能展示 inquiry_items = db.query(BizInquiryItem).filter( and_( BizInquiryItem.inquiry_id == quotation.inquiry_id, BizInquiryItem.is_del == 0 ) ).order_by(BizInquiryItem.create_at.asc()).all() # 报价单项作为从表（按 product_id 映射） items = quotation_item_crud.get_by_quotation(db, quotation_id) q_item_by_pid: dict[str, object] = {} for it in items: pid = str(getattr(it, "product_id", "") or "") if pid and pid not in q_item_by_pid: q_item_by_pid[pid] = it # 若该报价已被选中并生成价格清单，则回显调价系数/最终售价（用于运营端查看） pl = db.query(BizPriceList).filter( and_( BizPriceList.quotation_id == quotation_id, BizPriceList.is_del == 0 ) ).order_by(BizPriceList.version.desc(), BizPriceList.published_time.desc()).first() # key: product_id, value: pricing_coefficient(%) used at publish-time # NOTE: older price lists may have NULL pricing_coefficient; we keep None to allow fallback to product.price_coefficient. coeff_map: dict[str, Optional[float]] = {} final_price_map: dict[str, float] = {} if pl: pl_items = db.query(BizPriceListItem).filter( and_( BizPriceListItem.price_list_id == pl.id, BizPriceListItem.is_del == 0 ) ).all() for pit in pl_items: pid = str(pit.product_id or "") if not pid: continue coeff_map[pid] = float(pit.pricing_coefficient) if getattr(pit, "pricing_coefficient", None) is not None else None final_price_map[pid] = float(pit.price) if pit.price else 0 region_name = None if quotation.region_id: region = region_crud.get(db, quotation.region_id) region_name = region.name if region else None inquiry_title = None if quotation.inquiry_id: inquiry = inquiry_crud.get(db, quotation.inquiry_id) inquiry_title = inquiry.title if inquiry else None supplier_name = None supplier_phone = None if quotation.supplier_id: supplier = user_crud.get(db, quotation.supplier_id) if supplier: supplier_name = supplier.real_name supplier_phone = supplier.phone item_details = [] for inq_item in inquiry_items: pid = str(getattr(inq_item, "product_id", "") or "") q_item = q_item_by_pid.get(pid) product = product_crud.get(db, pid) if pid else None default_coeff = _numeric_to_float_optional(getattr(product, "price_coefficient", None)) if product else None default_coeff = float(default_coeff) if default_coeff is not None else 0.0 category_name = None if product and getattr(product, "category_id", None): category = category_crud.get(db, product.category_id) category_name = category.name if category else None unit_id = None unit_name = None # unit 优先取报价单项（更准确），否则回退商品默认 unit_id if q_item and getattr(q_item, "unit_id", None): unit_id = getattr(q_item, "unit_id", None) elif product and getattr(product, "unit_id", None): unit_id = getattr(product, "unit_id", None) if unit_id: u = unit_crud.get(db, unit_id) unit_name = u.name if u else None supplier_price = float(getattr(q_item, "price", 0) or 0) if q_item else 0 pricing_coeff = coeff_map.get(pid, None) if pricing_coeff is None: pricing_coeff = default_coeff item_details.append({ "id": getattr(q_item, "id", None) if q_item else None, "productId": pid, "productName": product.name if product else None, "imageUrl": getattr(product, "image_url", None) if product else None, "categoryName": category_name, "unitId": unit_id, "unitName": unit_name, "measureValue": _numeric_to_float_optional(getattr(product, "measure_value", None)) if product else None, "supplierPrice": supplier_price, "pricingCoefficient": pricing_coeff, "finalPrice": final_price_map.get(pid, 0), # 兼容旧前端字段：price 仍返回供应商报价 "price": supplier_price }) return success_response({ "id": quotation.id, "inquiryId": quotation.inquiry_id, "inquiryTitle": inquiry_title, "supplierId": quotation.supplier_id, "supplierName": supplier_name, "supplierPhone": supplier_phone, "regionId": quotation.region_id, "regionName": region_name, "status": quotation.status, "totalAmount": float(quotation.total_amount) if quotation.total_amount else 0, "submitTime": str(quotation.submit_time) if quotation.submit_time else None, "createAt": str(quotation.create_at) if quotation.create_at else None, "items": item_details }) # ========================================= # 询价单报价聚合查询（用于询价报价管理页） # ========================================= @router.get("/inquiries/{inquiry_id}/quotations/latest-by-supplier", response_model=dict) def list_latest_quotations_by_supplier( inquiry_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """按询价单聚合：每个供应商最新一次报价（包含未报价供应商）""" from app.models.models import BizInquiry, BizQuotation, SysUser, BizInquiryItem from app.crud.quotation import quotation_item_crud from sqlalchemy import and_ inquiry = db.query(BizInquiry).filter(and_(BizInquiry.id == inquiry_id, BizInquiry.is_del == 0)).first() if not inquiry: return error_response(404, "询价单不存在") # 询价单商品作为主表：后续给每个供应商补齐这些商品行 inquiry_items = db.query(BizInquiryItem).filter( and_( BizInquiryItem.inquiry_id == inquiry_id, BizInquiryItem.is_del == 0 ) ).order_by(BizInquiryItem.create_at.asc()).all() # 该区域全部供应商（包含未报价） suppliers = db.query(SysUser).filter( and_( SysUser.role == "supplier", SysUser.region_id == inquiry.region_id, SysUser.is_del == 0, SysUser.is_active == True ) ).order_by(SysUser.create_at.asc()).all() # 查询该询价单下全部报价单，然后在内存中取每个供应商最新一条 quotations = db.query(BizQuotation).filter( and_( BizQuotation.inquiry_id == inquiry_id, BizQuotation.is_del == 0 ) ).order_by(BizQuotation.create_at.desc()).all() latest_by_supplier: dict[str, BizQuotation] = {} for q in quotations: if q.supplier_id not in latest_by_supplier: latest_by_supplier[q.supplier_id] = q # 组装返回：供应商维度 + 最新报价（如无则为空） result = [] for s in suppliers: q = latest_by_supplier.get(s.id) quotation_items = [] q_item_by_pid: dict[str, object] = {} if q: items = quotation_item_crud.get_by_quotation(db, q.id) for it in items: pid = str(getattr(it, "product_id", "") or "") if pid and pid not in q_item_by_pid: q_item_by_pid[pid] = it # LEFT JOIN：以询价单商品为主，补齐未报价商品 for inq_item in inquiry_items: pid = str(getattr(inq_item, "product_id", "") or "") it = q_item_by_pid.get(pid) product = product_crud.get(db, pid) if pid else None category_name = None if product and getattr(product, "category_id", None): cat = category_crud.get(db, product.category_id) category_name = cat.name if cat else None unit_id = None unit_name = None if it and getattr(it, "unit_id", None): unit_id = getattr(it, "unit_id", None) elif product and getattr(product, "unit_id", None): unit_id = getattr(product, "unit_id", None) if unit_id: u = unit_crud.get(db, unit_id) unit_name = u.name if u else None price_val = None if it is not None: try: price_val = float(getattr(it, "price", 0) or 0) except Exception: price_val = 0 quotation_items.append({ "id": getattr(it, "id", None) if it else None, "productId": pid, "productName": product.name if product else None, "categoryName": category_name, "unitId": unit_id, "unitName": unit_name, "measureValue": _numeric_to_float_optional(getattr(product, "measure_value", None)) if product else None, "imageUrl": getattr(product, "image_url", None) if product else None, # 没报价时保持 null，便于前端区分“未报价”与“0元” "price": price_val }) result.append({ "supplierId": s.id, "supplierName": s.real_name or s.username, "supplierPhone": s.phone, # 即使供应商未报价，也返回 quotation.items（=询价单全部商品），用于前端完整展示商品行 "quotation": { "id": q.id if q else None, "status": q.status if q else None, "totalAmount": float(q.total_amount) if (q and q.total_amount) else 0, "submitTime": str(q.submit_time) if (q and q.submit_time) else None, "createAt": str(q.create_at) if (q and q.create_at) else None, "items": quotation_items } }) return success_response({ "inquiryId": inquiry_id, "regionId": inquiry.region_id, "list": result, "total": len(result) }) @router.get("/inquiries/{inquiry_id}/quotations/history", response_model=dict) def list_quotation_history_by_inquiry( inquiry_id: str, page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), supplierId: Optional[str] = Query(None, description="供应商ID"), status: Optional[str] = Query(None, description="状态: submitted/selected"), startDate: Optional[str] = Query(None, description="开始日期(ISO)"), endDate: Optional[str] = Query(None, description="结束日期(ISO)"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """按询价单查看全部报价单（第三个tab）""" from app.models.models import BizQuotation, BizInquiry, BizQuotationItem, BizInquiryItem from app.crud.quotation import quotation_crud from app.crud.user import user_crud from sqlalchemy import and_ from datetime import datetime inquiry = db.query(BizInquiry).filter(and_(BizInquiry.id == inquiry_id, BizInquiry.is_del == 0)).first() if not inquiry: return error_response(404, "询价单不存在") query = db.query(BizQuotation).filter( and_( BizQuotation.is_del == 0, BizQuotation.inquiry_id == inquiry_id, BizQuotation.region_id == inquiry.region_id ) ) if supplierId: query = query.filter(BizQuotation.supplier_id == supplierId) if status: query = query.filter(BizQuotation.status == status) def _parse_dt(v: Optional[str]) -> Optional[datetime]: if not v: return None try: return datetime.fromisoformat(v.replace("Z", "+00:00")) except Exception: return None sd = _parse_dt(startDate) ed = _parse_dt(endDate) if sd: query = query.filter(BizQuotation.create_at >= sd) if ed: query = query.filter(BizQuotation.create_at <= ed) total = query.count() quotations = query.order_by(BizQuotation.create_at.desc()).offset((page - 1) * pageSize).limit(pageSize).all() supplier_map = {} for q in quotations: if q.supplier_id and q.supplier_id not in supplier_map: u = user_crud.get(db, q.supplier_id) supplier_map[q.supplier_id] = u.real_name if u else None q_ids = [q.id for q in quotations] partial_ids: set[str] = set() inquiry_pids = [ r[0] for r in db.query(BizInquiryItem.product_id) .filter(BizInquiryItem.inquiry_id == inquiry_id, BizInquiryItem.is_del == 0) .all() ] expected = set(inquiry_pids) if q_ids: item_rows = ( db.query(BizQuotationItem.quotation_id, BizQuotationItem.product_id, BizQuotationItem.price) .filter(BizQuotationItem.quotation_id.in_(q_ids), BizQuotationItem.is_del == 0) .all() ) by_quotation: dict[str, dict[str, object]] = {} for qid, pid, price in item_rows: by_quotation.setdefault(qid, {})[pid] = price for qid in q_ids: pmap = by_quotation.get(qid, {}) if expected: for pid in expected: pr = pmap.get(pid) try: pv = float(pr) if pr is not None else 0.0 except (TypeError, ValueError): pv = 0.0 if pv <= 0: partial_ids.add(qid) break else: for pr in pmap.values(): try: pv = float(pr) if pr is not None else 0.0 except (TypeError, ValueError): pv = 0.0 if pv <= 0: partial_ids.add(qid) break return success_response({ "list": [{ "id": q.id, "inquiryId": q.inquiry_id, "supplierId": q.supplier_id, "supplierName": supplier_map.get(q.supplier_id), "regionId": q.region_id, "status": q.status, "totalAmount": float(q.total_amount) if q.total_amount else 0, "submitTime": str(q.submit_time) if q.submit_time else None, "createAt": str(q.create_at) if q.create_at else None, "partialPriceMissing": q.id in partial_ids } for q in quotations], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/inquiries/{inquiry_id}/quotations/selected", response_model=dict) def get_selected_quotation_for_inquiry( inquiry_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取某询价单当前被选中的报价（若已生成价格清单，则从价格清单反查）""" from app.models.models import BizInquiry, BizQuotation, BizPriceList, BizPriceListItem, BizQuotationItem, BizInquiryItem from sqlalchemy import and_ from app.crud.user import user_crud inquiry = db.query(BizInquiry).filter(and_(BizInquiry.id == inquiry_id, BizInquiry.is_del == 0)).first() if not inquiry: return error_response(404, "询价单不存在") # 询价单商品清单（作为主表，确保完整展示） inquiry_items = db.query(BizInquiryItem).filter( and_( BizInquiryItem.inquiry_id == inquiry_id, BizInquiryItem.is_del == 0 ) ).order_by(BizInquiryItem.create_at.asc()).all() # 优先从价格清单查来源（允许后续价格已被新的询价覆盖 active） pl = db.query(BizPriceList).filter( and_( BizPriceList.inquiry_id == inquiry_id, BizPriceList.is_del == 0 ) ).order_by(BizPriceList.version.desc(), BizPriceList.published_time.desc()).first() if not pl: # 兜底：直接查 selected 报价 q = db.query(BizQuotation).filter( and_( BizQuotation.inquiry_id == inquiry_id, BizQuotation.status == "selected", BizQuotation.is_del == 0 ) ).order_by(BizQuotation.create_at.desc()).first() if not q: # 未选择任何供应商报价时，也返回询价单商品清单（用于前端完整展示） item_details = [] for inq_item in inquiry_items: pid = str(getattr(inq_item, "product_id", "") or "") product = product_crud.get(db, pid) if pid else None default_coeff = _numeric_to_float_optional(getattr(product, "price_coefficient", None)) if product else None default_coeff = float(default_coeff) if default_coeff is not None else 0.0 category_name = None if product and getattr(product, "category_id", None): cat = category_crud.get(db, product.category_id) category_name = cat.name if cat else None unit_id = getattr(product, "unit_id", None) if product else None unit_name = None if unit_id: u = unit_crud.get(db, unit_id) unit_name = u.name if u else None item_details.append({ "productId": pid, "productName": product.name if product else None, "skuCode": getattr(product, "sku_code", None) if product else None, "description": getattr(product, "description", None) if product else None, "imageUrl": getattr(product, "image_url", None) if product else None, "categoryName": category_name, "unitId": unit_id, "unitName": unit_name, "measureValue": _numeric_to_float_optional(getattr(product, "measure_value", None)) if product else None, # 未报价：返回 0 / 0，前端可按需展示为 '-' "supplierPrice": 0, "price": 0, "pricingCoefficient": default_coeff }) return success_response({ "quotationId": None, "quotationCreateAt": None, "quotationSubmitTime": None, "supplierId": None, "supplierName": None, "items": item_details }) supplier = user_crud.get(db, q.supplier_id) if q.supplier_id else None # 兜底分支也补齐询价单商品行：supplierPrice 从报价项拿，最终售价未知则为 0 supplier_price_map: dict[str, float] = {} q_items = db.query(BizQuotationItem).filter( and_(BizQuotationItem.quotation_id == q.id, BizQuotationItem.is_del == 0) ).all() for qi in q_items: if not qi.product_id: continue supplier_price_map[str(qi.product_id)] = float(qi.price) if qi.price else 0 item_details = [] for inq_item in inquiry_items: pid = str(getattr(inq_item, "product_id", "") or "") product = product_crud.get(db, pid) if pid else None default_coeff = _numeric_to_float_optional(getattr(product, "price_coefficient", None)) if product else None default_coeff = float(default_coeff) if default_coeff is not None else 0.0 category_name = None if product and getattr(product, "category_id", None): cat = category_crud.get(db, product.category_id) category_name = cat.name if cat else None unit_id = getattr(product, "unit_id", None) if product else None unit_name = None if unit_id: u = unit_crud.get(db, unit_id) unit_name = u.name if u else None item_details.append({ "productId": pid, "productName": product.name if product else None, "skuCode": getattr(product, "sku_code", None) if product else None, "description": getattr(product, "description", None) if product else None, "imageUrl": getattr(product, "image_url", None) if product else None, "categoryName": category_name, "unitId": unit_id, "unitName": unit_name, "measureValue": _numeric_to_float_optional(getattr(product, "measure_value", None)) if product else None, # 未报价商品返回 0（兜底分支无法区分 null/0；主流程会更准确） "supplierPrice": supplier_price_map.get(pid, 0), "price": 0, "pricingCoefficient": default_coeff }) return success_response({ "quotationId": q.id, "quotationCreateAt": str(q.create_at) if getattr(q, "create_at", None) else None, "quotationSubmitTime": str(q.submit_time) if getattr(q, "submit_time", None) else None, "supplierId": q.supplier_id, "supplierName": supplier.real_name if supplier else None, "items": item_details }) supplier = user_crud.get(db, pl.supplier_id) if pl.supplier_id else None quotation_create_at = None quotation_submit_time = None if pl.quotation_id: q = db.query(BizQuotation).filter( and_( BizQuotation.id == pl.quotation_id, BizQuotation.is_del == 0 ) ).first() if q: quotation_create_at = str(q.create_at) if getattr(q, "create_at", None) else None quotation_submit_time = str(q.submit_time) if getattr(q, "submit_time", None) else None items = db.query(BizPriceListItem).filter( and_(BizPriceListItem.price_list_id == pl.id, BizPriceListItem.is_del == 0) ).all() # price list item 按 product_id 映射（LEFT JOIN） pl_item_by_pid: dict[str, object] = {} for it in items: pid = str(getattr(it, "product_id", "") or "") if pid and pid not in pl_item_by_pid: pl_item_by_pid[pid] = it supplier_price_map: dict[str, float] = {} if pl.quotation_id: q_items = db.query(BizQuotationItem).filter( and_(BizQuotationItem.quotation_id == pl.quotation_id, BizQuotationItem.is_del == 0) ).all() for qi in q_items: if not qi.product_id: continue supplier_price_map[str(qi.product_id)] = float(qi.price) if qi.price else 0 item_details = [] for inq_item in inquiry_items: pid = str(getattr(inq_item, "product_id", "") or "") it = pl_item_by_pid.get(pid) product = product_crud.get(db, pid) if pid else None default_coeff = _numeric_to_float_optional(getattr(product, "price_coefficient", None)) if product else None default_coeff = float(default_coeff) if default_coeff is not None else 0.0 category_name = None if product and getattr(product, "category_id", None): cat = category_crud.get(db, product.category_id) category_name = cat.name if cat else None unit_id = None unit_name = None if it and getattr(it, "unit_id", None): unit_id = getattr(it, "unit_id", None) elif product and getattr(product, "unit_id", None): unit_id = getattr(product, "unit_id", None) if unit_id: u = unit_crud.get(db, unit_id) unit_name = u.name if u else None item_details.append({ "productId": pid, "productName": product.name if product else None, "skuCode": getattr(product, "sku_code", None) if product else None, "description": getattr(product, "description", None) if product else None, "imageUrl": getattr(product, "image_url", None) if product else None, "categoryName": category_name, "unitId": unit_id, "unitName": unit_name, "measureValue": _numeric_to_float_optional(getattr(product, "measure_value", None)) if product else None, "supplierPrice": supplier_price_map.get(pid, 0), # 最终售价来自 price list item；未生成/不存在时返回 0 "price": float(getattr(it, "price", 0) or 0) if it else 0, "pricingCoefficient": float(getattr(it, "pricing_coefficient", 0)) if (it and getattr(it, "pricing_coefficient", None) is not None) else default_coeff }) return success_response({ "priceListId": pl.id, "priceVersion": pl.version, "quotationId": pl.quotation_id, "quotationCreateAt": quotation_create_at, "quotationSubmitTime": quotation_submit_time, "supplierId": pl.supplier_id, "supplierName": supplier.real_name if supplier else None, "items": item_details }) @router.get("/quotation-compare", response_model=dict) def compare_quotations( inquiryId: str = Query(..., description="询价单ID"), supplierIds: str = Query(..., description="供应商ID，逗号分隔"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """报价对比（按询价单+供应商）""" from app.crud.quotation import quotation_crud, quotation_item_crud supplier_id_list = [x.strip() for x in supplierIds.split(",") if x.strip()] if len(supplier_id_list) < 2: return error_response(400, "至少选择两个供应商") quotations, _ = quotation_crud.get_multi_with_filters( db, inquiry_id=inquiryId, skip=0, limit=500 ) quotations = [q for q in quotations if q.supplier_id in supplier_id_list] if not quotations: return success_response([]) # 汇总每个商品在不同供应商下的价格 compare_map = {} for q in quotations: items = quotation_item_crud.get_by_quotation(db, q.id) for item in items: product = product_crud.get(db, item.product_id) unit_name = None if item.unit_id: u = unit_crud.get(db, item.unit_id) unit_name = u.name if u else None key = f"{item.product_id}:{item.unit_id or ''}" row = compare_map.get(key) if not row: row = { "productId": item.product_id, "productName": product.name if product else item.product_id, "unitId": item.unit_id, "unitName": unit_name, "prices": {}, "lowestPrice": None, "highestPrice": None } compare_map[key] = row price_value = float(item.price) if item.price is not None else 0 row["prices"][q.supplier_id] = price_value result = [] for _, row in compare_map.items(): price_values = [v for v in row["prices"].values() if v is not None] if price_values: row["lowestPrice"] = min(price_values) row["highestPrice"] = max(price_values) else: row["lowestPrice"] = 0 row["highestPrice"] = 0 result.append(row) result.sort(key=lambda x: (x.get("productName") or "", x.get("unitName") or "")) return success_response(result) @router.post("/quotations/{quotation_id}/approve", response_model=dict) def approve_quotation( quotation_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """报价无需审核（接口已停用）""" return error_response(410, "报价无需审核：请在“询价报价管理-最新报价对比”中直接选中供应商生效价格") @router.post("/quotations/{quotation_id}/reject", response_model=dict) def reject_quotation( quotation_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """报价无需审核（接口已停用）""" return error_response(410, "报价无需审核：请在“询价报价管理-最新报价对比”中直接选中供应商生效价格") # ========================================= # 价格清单管理 API (Web端) # ========================================= @router.get("/price-lists", response_model=dict) def list_price_lists( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), regionId: Optional[str] = Query(None, description="区域ID"), status: Optional[str] = Query(None, description="状态: draft/pending/active/expired"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取价格清单列表""" from app.crud.price_list import price_list_crud from app.crud.user import user_crud # 区域管理员只能查看自己区域的价格清单 admin_region_id = get_admin_region_filter(db, current_user) if admin_region_id and not regionId: regionId = admin_region_id elif admin_region_id and regionId != admin_region_id: return success_response({ "list": [], "total": 0, "page": page, "pageSize": pageSize, "totalPages": 0 }) price_lists, total = price_list_crud.get_multi_with_filters( db, region_id=regionId, status=status, skip=(page - 1) * pageSize, limit=pageSize ) region_map = {} supplier_map = {} publisher_map = {} for pl in price_lists: if pl.region_id and pl.region_id not in region_map: region = region_crud.get(db, pl.region_id) region_map[pl.region_id] = region.name if region else None if pl.supplier_id and pl.supplier_id not in supplier_map: supplier = user_crud.get(db, pl.supplier_id) supplier_map[pl.supplier_id] = supplier.real_name if supplier else None if pl.published_by and pl.published_by not in publisher_map: publisher = user_crud.get(db, pl.published_by) publisher_map[pl.published_by] = publisher.real_name if publisher else None return success_response({ "list": [{ "id": pl.id, "regionId": pl.region_id, "regionName": region_map.get(pl.region_id), "supplierId": pl.supplier_id, "supplierName": supplier_map.get(pl.supplier_id), "version": pl.version, "status": pl.status, "effectiveTime": str(pl.effective_time) if pl.effective_time else None, "publishedBy": pl.published_by, "publishedByName": publisher_map.get(pl.published_by), "publishedTime": str(pl.published_time) if pl.published_time else None, "remark": pl.remark, "createAt": str(pl.create_at) if pl.create_at else None } for pl in price_lists], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.post("/price-lists", response_model=dict) def create_price_list( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """创建价格清单""" from app.crud.price_list import price_list_crud from datetime import datetime region_id = request.get("regionId") supplier_id = request.get("supplierId") items = request.get("items", []) effective_type = request.get("effectiveType", "immediate") effective_time_str = request.get("effectiveTime") remark = request.get("remark") if not region_id: return error_response(400, "请选择区域") if not supplier_id: return error_response(400, "请选择供应商") if not items: return error_response(400, "请添加价格项") effective_time = None if effective_time_str: try: effective_time = datetime.fromisoformat(effective_time_str.replace("Z", "+00:00")) except: effective_time = None creator = { "user_id": current_user.get("user_id"), "real_name": current_user.get("real_name") } price_list = price_list_crud.create_with_items( db, obj_in={ "region_id": region_id, "supplier_id": supplier_id, "effective_time": effective_time, "remark": remark }, items=items, creator=creator ) # 如果是立即生效，直接发布 if effective_type == "immediate": price_list_crud.publish( db, price_list.id, current_user.get("user_id"), effective_type="immediate" ) return success_response({ "id": price_list.id, "message": "创建成功" }) @router.get("/price-lists/{price_list_id}", response_model=dict) def get_price_list( price_list_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取价格清单详情""" from app.crud.price_list import price_list_crud, price_list_item_crud from app.crud.user import user_crud price_list = price_list_crud.get(db, price_list_id) if not price_list: return error_response(404, "价格清单不存在") items = price_list_item_crud.get_by_price_list(db, price_list_id) # 获取关联信息 region_name = None if price_list.region_id: region = region_crud.get(db, price_list.region_id) region_name = region.name if region else None supplier_name = None if price_list.supplier_id: supplier = user_crud.get(db, price_list.supplier_id) supplier_name = supplier.real_name if supplier else None publisher_name = None if price_list.published_by: publisher = user_crud.get(db, price_list.published_by) publisher_name = publisher.real_name if publisher else None # 订单关联供应商（来自订单价格清单） from app.crud.price_list import price_list_crud price_list = price_list_crud.get(db, order.price_list_id) if order.price_list_id else None supplier_id_from_price_list = getattr(price_list, "supplier_id", None) if price_list else None supplier_name_from_price_list = None if supplier_id_from_price_list: _sup_u = user_crud.get(db, supplier_id_from_price_list) if _sup_u: supplier_name_from_price_list = _sup_u.real_name or _sup_u.username pl_ids_for_order = [order.price_list_id] if order.price_list_id else [] pli_map_for_order = _price_list_item_map(db, pl_ids_for_order) # 获取商品和单位信息 item_details = [] supplier_total_amount = 0.0 selling_total_amount = 0.0 for item in items: product = product_crud.get(db, item.product_id) category_name = None if product and product.category_id: cat = category_crud.get(db, product.category_id) category_name = cat.name if cat else None unit_name = None if item.unit_id: u = unit_crud.get(db, item.unit_id) unit_name = u.name if u else None qty = float(item.quantity) if item.quantity else 0 selling_u = float(item.unit_price) if item.unit_price else 0 selling_amt = float(item.amount) if item.amount else round(qty * selling_u, 2) pli = pli_map_for_order.get((order.price_list_id, item.product_id, item.unit_id)) supplier_u = _supplier_unit_yuan_from_pli(pli) if supplier_u is None: supplier_u = selling_u supplier_amt = round(qty * supplier_u, 2) supplier_total_amount += supplier_amt selling_total_amount += selling_amt item_details.append({ "id": item.id, "productId": item.product_id, "productName": product.name if product else None, "categoryName": category_name, "unitId": item.unit_id, "unitName": unit_name, "price": float(item.price) if item.price else 0, "imageUrl": product.image_url if product else None }) return success_response({ "id": price_list.id, "regionId": price_list.region_id, "regionName": region_name, "supplierId": price_list.supplier_id, "supplierName": supplier_name, "version": price_list.version, "status": price_list.status, "effectiveTime": str(price_list.effective_time) if price_list.effective_time else None, "publishedBy": price_list.published_by, "publishedByName": publisher_name, "publishedTime": str(price_list.published_time) if price_list.published_time else None, "remark": price_list.remark, "createAt": str(price_list.create_at) if price_list.create_at else None, "items": item_details }) @router.put("/price-lists/{price_list_id}", response_model=dict) def update_price_list( price_list_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """更新价格清单（只有草稿状态可编辑）""" from app.crud.price_list import price_list_crud from datetime import datetime price_list = price_list_crud.get(db, price_list_id) if not price_list: return error_response(404, "价格清单不存在") if price_list.status != "draft": return error_response(400, "只有草稿状态的价格清单可以编辑") items = request.get("items") effective_time_str = request.get("effectiveTime") remark = request.get("remark") obj_in = {} if effective_time_str: try: obj_in["effective_time"] = datetime.fromisoformat(effective_time_str.replace("Z", "+00:00")) except: pass if remark is not None: obj_in["remark"] = remark updater = { "user_id": current_user.get("user_id"), "real_name": current_user.get("real_name") } price_list_crud.update_with_items( db, price_list_id, obj_in, items=items, updater=updater ) return success_response({ "id": price_list_id, "message": "更新成功" }) @router.post("/price-lists/{price_list_id}/publish", response_model=dict) def publish_price_list( price_list_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """发布价格清单""" from app.crud.price_list import price_list_crud from datetime import datetime price_list = price_list_crud.get(db, price_list_id) if not price_list: return error_response(404, "价格清单不存在") if price_list.status != "draft": return error_response(400, "只有草稿状态的价格清单可以发布") effective_type = request.get("effectiveType", "immediate") effective_time_str = request.get("effectiveTime") effective_time = None if effective_time_str: try: effective_time = datetime.fromisoformat(effective_time_str.replace("Z", "+00:00")) except: pass price_list_crud.publish( db, price_list_id, current_user.get("user_id"), effective_time=effective_time, effective_type=effective_type ) return success_response({ "id": price_list_id, "message": "发布成功" }) @router.post("/price-lists/{price_list_id}/invalidate", response_model=dict) def invalidate_price_list( price_list_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """使价格清单失效""" from app.crud.price_list import price_list_crud price_list = price_list_crud.get(db, price_list_id) if not price_list: return error_response(404, "价格清单不存在") if price_list.status not in ["active", "pending"]: return error_response(400, "只有生效中或待生效的价格清单可以失效") price_list_crud.invalidate(db, price_list_id) return success_response({ "id": price_list_id, "message": "操作成功" }) @router.delete("/price-lists/{price_list_id}", response_model=dict) def delete_price_list( price_list_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """删除价格清单（只有草稿状态可删除）""" from app.crud.price_list import price_list_crud price_list = price_list_crud.get(db, price_list_id) if not price_list: return error_response(404, "价格清单不存在") if price_list.status != "draft": return error_response(400, "只有草稿状态的价格清单可以删除") price_list_crud.delete(db, price_list_id) return success_response({ "message": "删除成功" }) @router.get("/price-lists/product-history", response_model=dict) def get_product_price_history( regionId: str = Query(..., description="区域ID"), productId: str = Query(..., description="商品ID"), page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(20, ge=1, le=100, description="每页条数"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """按商品维度查询价格历史""" from app.crud.price_list import price_list_crud, price_list_item_crud product = product_crud.get(db, productId) if not product: return error_response(404, "商品不存在") items, total = price_list_item_crud.get_product_price_history( db, region_id=regionId, product_id=productId, skip=(page - 1) * pageSize, limit=pageSize ) # 获取单位信息 for item in items: if item.get("unit_id"): u = unit_crud.get(db, item["unit_id"]) item["unitName"] = u.name if u else None return success_response({ "productId": productId, "productName": product.name, "history": items, "total": total, "page": page, "pageSize": pageSize }) @router.get("/price-lists/generate-options", response_model=dict) def get_generate_price_list_options( regionId: str = Query(..., description="区域ID"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取生成价格清单的选项（该区域已提交、尚未被选中的报价单，报价无审核流程）""" from app.crud.quotation import quotation_crud, quotation_item_crud from app.crud.user import user_crud quotations, total = quotation_crud.get_multi_with_filters( db, region_id=regionId, status="submitted", skip=0, limit=100 ) result = [] for q in quotations: supplier = user_crud.get(db, q.supplier_id) items = quotation_item_crud.get_by_quotation(db, q.id) item_details = [] for item in items: product = product_crud.get(db, item.product_id) unit_name = None if item.unit_id: u = unit_crud.get(db, item.unit_id) unit_name = u.name if u else None item_details.append({ "productId": item.product_id, "productName": product.name if product else None, "unitId": item.unit_id, "unitName": unit_name, "price": float(item.price) if item.price else 0 }) result.append({ "quotationId": q.id, "quotationNo": f"QT{q.create_at.strftime('%Y%m%d%H%M%S') if q.create_at else ''}", "supplierId": q.supplier_id, "supplierName": supplier.real_name if supplier else None, "totalAmount": float(q.total_amount) if q.total_amount else 0, "items": item_details }) return success_response({ "list": result, "total": len(result) }) @router.get("/price-lists/compare", response_model=dict) def compare_price_list_versions( regionId: str = Query(..., description="区域ID"), version1Id: str = Query(..., description="版本1ID"), version2Id: str = Query(..., description="版本2ID"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """对比两个价格版本""" from app.crud.price_list import price_list_crud, price_list_item_crud # 获取两个版本的价格清单 price_list1 = price_list_crud.get(db, version1Id) price_list2 = price_list_crud.get(db, version2Id) if not price_list1 or not price_list2: return error_response(404, "价格版本不存在") if price_list1.region_id != regionId or price_list2.region_id != regionId: return error_response(400, "价格版本不属于该区域") # 获取两个版本的价格项 items1 = price_list_item_crud.get_by_price_list(db, version1Id) items2 = price_list_item_crud.get_by_price_list(db, version2Id) # 构建价格映射 price_map1 = {item.product_id: item for item in items1} price_map2 = {item.product_id: item for item in items2} # 获取所有涉及的商品 all_product_ids = set(price_map1.keys()) | set(price_map2.keys()) # 构建对比数据 comparison = [] for product_id in all_product_ids: item1 = price_map1.get(product_id) item2 = price_map2.get(product_id) product = product_crud.get(db, product_id) unit_name = None if item1 and item1.unit_id: u = unit_crud.get(db, item1.unit_id) unit_name = u.name if u else None elif item2 and item2.unit_id: u = unit_crud.get(db, item2.unit_id) unit_name = u.name if u else None price1 = float(item1.price) if item1 and item1.price else None price2 = float(item2.price) if item2 and item2.price else None # 计算差异 price_diff = None price_change = None change_rate = None if price1 is not None and price2 is not None: price_diff = price2 - price1 if price1 > 0: change_rate = round((price_diff / price1) * 100, 2) if price_diff > 0: price_change = "up" elif price_diff < 0: price_change = "down" else: price_change = "same" comparison.append({ "productId": product_id, "productName": product.name if product else None, "categoryId": product.category_id if product else None, "unitId": item1.unit_id if item1 else (item2.unit_id if item2 else None), "unitName": unit_name, "price1": price1, "price2": price2, "priceDiff": price_diff, "priceChange": price_change, "changeRate": change_rate, "existIn1": item1 is not None, "existIn2": item2 is not None }) # 计算汇总统计 total_products = len(comparison) price_up_count = sum(1 for c in comparison if c["priceChange"] == "up") price_down_count = sum(1 for c in comparison if c["priceChange"] == "down") price_same_count = sum(1 for c in comparison if c["priceChange"] == "same") new_product_count = sum(1 for c in comparison if not c["existIn1"] and c["existIn2"]) removed_product_count = sum(1 for c in comparison if c["existIn1"] and not c["existIn2"]) return success_response({ "version1": { "id": price_list1.id, "version": price_list1.version, "status": price_list1.status, "publishedTime": str(price_list1.published_time) if price_list1.published_time else None }, "version2": { "id": price_list2.id, "version": price_list2.version, "status": price_list2.status, "publishedTime": str(price_list2.published_time) if price_list2.published_time else None }, "statistics": { "totalProducts": total_products, "priceUp": price_up_count, "priceDown": price_down_count, "priceSame": price_same_count, "newProducts": new_product_count, "removedProducts": removed_product_count }, "comparison": comparison }) @router.get("/price-lists/history", response_model=dict) def get_price_list_history( regionId: str = Query(..., description="区域ID"), page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=50, description="每页条数"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取价格清单历史版本列表""" from app.crud.price_list import price_list_crud from app.crud.user import user_crud price_lists, total = price_list_crud.get_multi_with_filters( db, region_id=regionId, skip=(page - 1) * pageSize, limit=pageSize ) result = [] for pl in price_lists: supplier = user_crud.get(db, pl.supplier_id) region = region_crud.get(db, pl.region_id) result.append({ "id": pl.id, "version": pl.version, "status": pl.status, "statusText": "生效中" if pl.status == "active" else "待生效" if pl.status == "pending" else "已失效" if pl.status == "expired" else "草稿", "supplierId": pl.supplier_id, "supplierName": supplier.real_name if supplier else None, "regionName": region.name if region else None, "effectiveTime": str(pl.effective_time) if pl.effective_time else None, "publishedTime": str(pl.published_time) if pl.published_time else None, "remark": pl.remark, "createAt": str(pl.create_at) if pl.create_at else None }) return success_response({ "list": result, "total": total, "page": page, "pageSize": pageSize }) # ========================================= # 订单管理 API (Web端) # ========================================= # 订单状态映射 ORDER_STATUS_MAP = { "pending_payment": "待支付", "pending_review_cancel": "待审核取消", "pending_dispatch": "待分配配送", "delivering": "配送中", "pending_confirm": "待确认收货", "completed": "已完结", "cancelled": "已取消" } # 订单状态日志展示（含配送/支付维度的结构化 to_status） DELIVERY_STATUS_LOG_LABELS = { "pending_dispatch": "待分配配送", "delivering": "配送中", "delivered": "配送完成", } def _order_log_status_display_map() -> dict: m = dict(ORDER_STATUS_MAP) m["delivery.assign"] = "分配配送员" m["delivery.reassign"] = "更换配送员" for k, v in DELIVERY_STATUS_LOG_LABELS.items(): m[f"ds_{k}"] = f"配送状态：{v}" m["pay_pending_payment"] = "支付状态：待支付" m["pay_paid"] = "支付状态：已支付" m["pay_refunding"] = "支付状态：退款中" m["pay_refunded"] = "支付状态：已退款" _task_lbl = { "pending": "待分配", "assigned": "已分配", "supplier_stocking": "供应商备货中", "supplier_ready": "备货完成（待取货）", "picking": "取货中", "delivering": "配送中", "completed": "已完成", "cancelled": "已取消", } for k, v in _task_lbl.items(): m[f"dt_{k}"] = f"配送任务：{v}" return m ORDER_LOG_STATUS_MAP = _order_log_status_display_map() def _status_log_timeline_description(entry: dict) -> str: """与 web-admin 订单详情「操作记录」中对 reason/note 的展示规则保持一致。""" note = (entry.get("note") or "").strip() raw_reason = entry.get("reason") description = note or "" if raw_reason and str(raw_reason).strip().startswith("{"): try: o = json.loads(raw_reason) if isinstance(o, dict) and ( o.get("toDeliverName") is not None or o.get("fromDeliverName") is not None ): description = f"配送员：{o.get('fromDeliverName') or '—'} → {o.get('toDeliverName') or '—'}" if note: description += f"；{note}" except json.JSONDecodeError: if not description: description = str(raw_reason) elif raw_reason and not description: description = str(raw_reason) return description def _delivery_timeline_type(to_status: str) -> str: if to_status == "dt_completed": return "success" if to_status == "dt_cancelled": return "danger" return "primary" def _log_entry_in_delivery_timeline(entry: dict) -> bool: ts = entry.get("status") or "" if ts in ("delivery.assign", "delivery.reassign"): return True if ts.startswith("dt_") or ts.startswith("ds_"): return True return False def _build_order_delivery_timeline(dtask, log_details: List[dict], order) -> List[dict]: """ 配送进度以订单状态日志为权威来源（配送任务 dt_*、配送状态 ds_*、分配配送员等）， 与「操作记录」同源，避免两套拼接逻辑长期漂移。 """ events: List[dict] = [] if dtask is not None and dtask.create_at: dr = (getattr(dtask, "dispatch_remark", None) or "").strip() or None events.append({ "time": str(dtask.create_at), "title": "配送任务已创建", "description": dr or "", "type": "info", "hollow": False, }) timeline_statuses: set = set() for entry in log_details: if not _log_entry_in_delivery_timeline(entry): continue ts_key = entry.get("status") or "" timeline_statuses.add(ts_key) t = entry.get("createAt") if not t: continue from_t = (entry.get("fromStatusText") or "").strip() to_t = (entry.get("statusText") or "").strip() or ts_key or "-" title = f"{from_t} → {to_t}" if from_t else to_t events.append({ "time": t, "title": title, "description": _status_log_timeline_description(entry), "type": _delivery_timeline_type(ts_key), "hollow": False, }) if dtask is not None: if dtask.pickup_time and "dt_picking" not in timeline_statuses: pr = (dtask.pickup_remark or "").strip() or "" events.append({ "time": str(dtask.pickup_time), "title": "配送任务：取货中", "description": pr, "type": "primary", "hollow": False, }) if "dt_completed" not in timeline_statuses and ( dtask.status == "completed" or (getattr(dtask, "deliverer_status", None) == "delivered") ): t_end = dtask.delivery_time or getattr(order, "delivered_time", None) if t_end: dr = (dtask.delivery_remark or "").strip() or "" events.append({ "time": str(t_end), "title": "配送任务：已完成", "description": dr, "type": "success", "hollow": False, }) events = [e for e in events if e.get("time")] events.sort(key=lambda x: str(x.get("time") or "")) return events @router.get("/orders", response_model=dict) def list_orders( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), regionId: Optional[str] = Query(None, description="区域ID"), status: Optional[str] = Query(None, description="订单状态"), orderStatus: Optional[str] = Query(None, description="订单主状态"), deliveryStatus: Optional[str] = Query(None, description="配送状态"), payStatus: Optional[str] = Query(None, description="支付状态"), paymentMethod: Optional[str] = Query(None, description="支付方式: online/offline"), clientId: Optional[str] = Query(None, description="客户ID"), orderNo: Optional[str] = Query(None, description="订单号"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取订单列表""" from app.crud.order import order_crud, order_item_crud # 区域管理员只能查看自己区域的订单 admin_region_id = get_admin_region_filter(db, current_user) if admin_region_id and not regionId: # 如果是区域管理员且未指定区域，默认使用其所属区域 regionId = admin_region_id elif admin_region_id and regionId != admin_region_id: # 区域管理员不能查看其他区域的订单 return success_response({ "list": [], "total": 0, "page": page, "pageSize": pageSize, "totalPages": 0 }) orders, total = order_crud.get_multi_with_filters( db, region_id=regionId, status=status, order_status=orderStatus, delivery_status=deliveryStatus, pay_status=payStatus, payment_method=paymentMethod, client_id=clientId, order_no=orderNo, start_date=startDate, end_date=endDate, skip=(page - 1) * pageSize, limit=pageSize ) # 列表页：按订单行汇总供货总金额 / 售出总金额（与订单详情口径一致） order_ids_page = [o.id for o in orders] oid_to_order = {o.id: o for o in orders} items_by_oid: dict = {} if order_ids_page: for ln in ( db.query(BizOrderItem) .filter(BizOrderItem.order_id.in_(order_ids_page), BizOrderItem.is_del == 0) .all() ): items_by_oid.setdefault(ln.order_id, []).append(ln) pl_ids_page = list({o.price_list_id for o in orders if o.price_list_id}) pli_map_page = _price_list_item_map(db, pl_ids_page) if pl_ids_page else {} order_supplier_selling_totals: dict = {} for oid, lines in items_by_oid.items(): o = oid_to_order.get(oid) pl_id = o.price_list_id if o else None sup_acc = 0.0 sel_acc = 0.0 for item in lines: qty = float(item.quantity or 0) selling_u = float(item.unit_price or 0) selling_amt = ( float(item.amount) if item.amount is not None else round(qty * selling_u, 2) ) pli = pli_map_page.get((pl_id, item.product_id, item.unit_id)) if pl_id else None supplier_u = _supplier_unit_yuan_from_pli(pli) if supplier_u is None: supplier_u = selling_u sup_acc += round(qty * supplier_u, 2) sel_acc += selling_amt order_supplier_selling_totals[oid] = (round(sup_acc, 2), round(sel_acc, 2)) region_map = {} region_delivery_minutes_map = {} client_map = {} result = [] for order in orders: order_status = getattr(order, "order_status", None) or "pending_confirm" delivery_status = getattr(order, "delivery_status", None) or "pending_dispatch" pay_status = getattr(order, "pay_status", None) or "pending_payment" # 获取区域名称、预计配送时长 if order.region_id and order.region_id not in region_map: region = region_crud.get(db, order.region_id) region_map[order.region_id] = region.name if region else None region_delivery_minutes_map[order.region_id] = ( getattr(region, "expected_delivery_minutes", None) if region else None ) # 获取客户名称 if order.client_id and order.client_id not in client_map: client = user_crud.get(db, order.client_id) client_map[order.client_id] = client.real_name if client else None # 获取订单项数量 item_count = order_item_crud.get_count_by_order(db, order.id) _sup_total, _sel_total = order_supplier_selling_totals.get(order.id, (0.0, 0.0)) result.append({ "id": order.id, "orderNo": order.order_no, "clientId": order.client_id, "clientName": client_map.get(order.client_id), "regionId": order.region_id, "regionName": region_map.get(order.region_id), "regionExpectedDeliveryMinutes": region_delivery_minutes_map.get(order.region_id), "totalAmount": float(order.total_amount) if order.total_amount else 0, "supplierTotalAmount": _sup_total, "sellingTotalAmount": _sel_total, "paidAmount": float(order.paid_amount) if order.paid_amount else 0, "receivableAmount": float(order.receivable_amount) if order.receivable_amount else 0, "status": order.status, "orderStatus": order_status, "deliveryStatus": delivery_status, "payStatus": pay_status, "cancelStatus": getattr(order, "cancel_status", None), "paymentMethod": order_crud.normalize_payment_method(order.payment_method), "paymentStatus": "paid" if order.paid_amount and order.paid_amount > 0 else "unpaid", "itemCount": item_count, "createAt": str(order.create_at) if order.create_at else None, "paymentTime": str(order.payment_time) if order.payment_time else None }) return success_response({ "list": result, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/orders/{order_id}", response_model=dict) def get_order( order_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取订单详情""" from app.crud.order import order_crud, order_item_crud, order_status_log_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") # 获取订单项 items = order_item_crud.get_by_order(db, order_id) from app.crud.price_list import price_list_crud price_list = price_list_crud.get(db, order.price_list_id) if order.price_list_id else None supplier_id_from_price_list = getattr(price_list, "supplier_id", None) if price_list else None supplier_name_from_price_list = None if supplier_id_from_price_list: _sup_u = user_crud.get(db, supplier_id_from_price_list) if _sup_u: supplier_name_from_price_list = _sup_u.real_name or _sup_u.username pl_ids_for_order = [order.price_list_id] if order.price_list_id else [] pli_map_for_order = _price_list_item_map(db, pl_ids_for_order) from app.crud.price_list import price_list_crud price_list = price_list_crud.get(db, order.price_list_id) if order.price_list_id else None supplier_id_from_price_list = getattr(price_list, "supplier_id", None) if price_list else None supplier_name_from_price_list = None if supplier_id_from_price_list: _sup_u = user_crud.get(db, supplier_id_from_price_list) if _sup_u: supplier_name_from_price_list = _sup_u.real_name or _sup_u.username pl_ids_for_order = [order.price_list_id] if order.price_list_id else [] pli_map_for_order = _price_list_item_map(db, pl_ids_for_order) from app.crud.price_list import price_list_crud price_list = price_list_crud.get(db, order.price_list_id) if order.price_list_id else None supplier_id_from_price_list = getattr(price_list, "supplier_id", None) if price_list else None supplier_name_from_price_list = None if supplier_id_from_price_list: _sup_u = user_crud.get(db, supplier_id_from_price_list) if _sup_u: supplier_name_from_price_list = _sup_u.real_name or _sup_u.username pl_ids_for_order = [order.price_list_id] if order.price_list_id else [] pli_map_for_order = _price_list_item_map(db, pl_ids_for_order) # 获取状态日志 status_logs = order_status_log_crud.get_by_order(db, order_id) # 尝试锁定订单（防止并发操作） lock_info = None operator_id = current_user.get("user_id") try: order_crud.lock_order(db, order_id, operator_id) except ValueError as e: # 已被其他用户锁定，返回锁定信息但不阻止查看 lock_info = { "isLocked": True, "lockedBy": order.lock_operator_id, "lockedTime": str(order.lock_time) if order.lock_time else None, "message": str(e) } else: lock_info = {"isLocked": False} # 获取关联信息 region_name = None region_expected_delivery_minutes = None if order.region_id: region = region_crud.get(db, order.region_id) region_name = region.name if region else None region_expected_delivery_minutes = ( getattr(region, "expected_delivery_minutes", None) if region else None ) client_name = None if order.client_id: client = user_crud.get(db, order.client_id) client_name = client.real_name if client else None # 配送任务明细上的已送达量（订单行 delivered_quantity 可能未同步，展示时与任务明细取大） from app.crud.delivery import delivery_task_item_crud as _merge_task_delivered _line_delivered_from_task = _merge_task_delivered.get_delivered_quantity_by_order_item(db, order_id) # 获取商品和单位信息 item_details = [] supplier_total_amount = 0.0 selling_total_amount = 0.0 for item in items: product = product_crud.get(db, item.product_id) unit_name = None if item.unit_id: u = unit_crud.get(db, item.unit_id) unit_name = u.name if u else None category_id = None category_path = "" image_url = None measure_value = None if product: category_id = getattr(product, "category_id", None) if category_id: category_path = _category_path_display(db, str(category_id)) image_url = getattr(product, "image_url", None) measure_value = _numeric_to_float_optional(getattr(product, "measure_value", None)) _oi_dq = float(item.delivered_quantity) if item.delivered_quantity else 0.0 _ti_dq = float(_line_delivered_from_task.get(item.id, 0.0)) _merged_dq = max(_oi_dq, _ti_dq) qty = float(item.quantity) if item.quantity else 0 selling_u = float(item.unit_price) if item.unit_price else 0 selling_amt = float(item.amount) if item.amount else round(qty * selling_u, 2) pli = pli_map_for_order.get((order.price_list_id, item.product_id, item.unit_id)) supplier_u = _supplier_unit_yuan_from_pli(pli) if supplier_u is None: supplier_u = selling_u supplier_amt = round(qty * supplier_u, 2) supplier_total_amount += supplier_amt selling_total_amount += selling_amt item_details.append({ "id": item.id, "productId": item.product_id, "productName": product.name if product else None, "imageUrl": image_url, "categoryId": str(category_id) if category_id else None, "categoryName": category_path or None, "measureValue": measure_value, "unitId": item.unit_id, "unitName": unit_name, "supplierId": supplier_id_from_price_list, "supplierName": supplier_name_from_price_list, "quantity": qty, "unitPrice": selling_u, "amount": selling_amt, "sellingUnitPrice": selling_u, "sellingAmount": selling_amt, "supplierUnitPrice": supplier_u, "supplierAmount": supplier_amt, "deliveredQuantity": _merged_dq, "status": item.status }) order_status = getattr(order, "order_status", None) or "pending_confirm" delivery_status = getattr(order, "delivery_status", None) or "pending_dispatch" pay_status = getattr(order, "pay_status", None) or "pending_payment" # 处理状态日志（含配送任务、配送员分配等结构化记录） log_details = [] for log in status_logs: log_details.append({ "fromStatus": log.from_status, "fromStatusText": ORDER_LOG_STATUS_MAP.get( log.from_status, log.from_status or "" ), "status": log.to_status, "statusText": ORDER_LOG_STATUS_MAP.get(log.to_status, log.to_status), # BizOrderStatusLog 无 operator_name 字段，优先使用 create_name "operatorName": getattr(log, "create_name", None) or getattr(log, "operator_id", None), "operatorRole": log.operator_role, "reason": log.reason, "note": log.note, "createAt": str(log.create_at) if log.create_at else None }) deliver_id_out = None deliver_name = None deliver_phone = None expected_time = None delivery_dispatch_remark = None delivery_timeline = [] supplier_id_for_pickup = None supplier_name = None supplier_phone = None supplier_pickup = {"fullAddress": None, "contactPerson": None, "contactPhone": None} delivery_task_no = None delivery_task_status = None delivery_task_deliverer_status = None delivery_task_supplier_prepare_status = None delivery_pickup_time = None delivery_pickup_remark = None delivery_delivery_time = None delivery_delivery_remark = None delivery_prepare_images = [] delivery_pickup_images = [] delivery_deliver_images = [] from app.crud.delivery import delivery_task_crud as _order_delivery_task_crud dtask = _order_delivery_task_crud.get_by_order(db, order_id) if dtask: deliver_id_out = dtask.deliver_id delivery_task_no = dtask.task_no delivery_task_status = dtask.status delivery_task_deliverer_status = getattr(dtask, "deliverer_status", None) delivery_task_supplier_prepare_status = getattr(dtask, "supplier_prepare_status", None) delivery_pickup_time = str(dtask.pickup_time) if dtask.pickup_time else None delivery_pickup_remark = dtask.pickup_remark delivery_delivery_time = str(dtask.delivery_time) if dtask.delivery_time else None delivery_delivery_remark = dtask.delivery_remark if dtask.deliver_id: du = user_crud.get(db, dtask.deliver_id) if du: deliver_name = du.real_name deliver_phone = du.phone expected_time = str(dtask.effective_time) if dtask.effective_time else None delivery_dispatch_remark = getattr(dtask, "dispatch_remark", None) or None supplier_id_for_pickup = dtask.supplier_id delivery_timeline = _build_order_delivery_timeline(dtask, log_details, order) from app.crud.attachment import attachment_crud as _attachment_crud delivery_prepare_images = [ a.file_url for a in _attachment_crud.get_by_business(db, "delivery_prepare", dtask.id) if a.file_url ] delivery_pickup_images = [ a.file_url for a in _attachment_crud.get_by_business(db, "delivery_pickup", dtask.id) if a.file_url ] delivery_deliver_images = [ a.file_url for a in _attachment_crud.get_by_business(db, "delivery_deliver", dtask.id) if a.file_url ] # 兼容旧数据：历史仅写在任务字段中 if not delivery_pickup_images and getattr(dtask, "pickup_image_url", None): delivery_pickup_images = [ s.strip() for s in str(dtask.pickup_image_url).split(",") if s and str(s).strip() ] if not delivery_deliver_images and getattr(dtask, "delivery_image_url", None): delivery_deliver_images = [ s.strip() for s in str(dtask.delivery_image_url).split(",") if s and str(s).strip() ] else: from app.crud.price_list import price_list_crud _pl = price_list_crud.get(db, order.price_list_id) if _pl and getattr(_pl, "supplier_id", None): supplier_id_for_pickup = _pl.supplier_id delivery_timeline = _build_order_delivery_timeline(None, log_details, order) if supplier_id_for_pickup: _su = user_crud.get(db, supplier_id_for_pickup) if _su: supplier_name = _su.real_name supplier_phone = _su.phone supplier_pickup = _supplier_pickup_address_block(db, supplier_id_for_pickup) return success_response({ "id": order.id, "orderNo": order.order_no, "regionId": order.region_id, "regionName": region_name, "regionExpectedDeliveryMinutes": region_expected_delivery_minutes, "clientId": order.client_id, "clientName": client_name, "priceListId": order.price_list_id, "priceVersion": order.price_version, "status": order.status, "orderStatus": order_status, "deliveryStatus": delivery_status, "payStatus": pay_status, "paymentMethod": order_crud.normalize_payment_method(order.payment_method), "totalAmount": float(order.total_amount) if order.total_amount else 0, "sellingTotalAmount": round(selling_total_amount, 2), "supplierTotalAmount": round(supplier_total_amount, 2), "paidAmount": float(order.paid_amount) if order.paid_amount else 0, "receivableAmount": float(order.receivable_amount) if order.receivable_amount else 0, "refundAmount": float(order.refund_amount) if order.refund_amount else 0, "contactPerson": order.contact_person, "contactPhone": order.contact_phone, "deliveryAddress": order.delivery_address, "remark": order.remark, "cancelReason": order.cancel_reason, "cancelReviewNote": order.cancel_review_note, "createAt": str(order.create_at) if order.create_at else None, "paymentTime": str(order.payment_time) if order.payment_time else None, "completedTime": str(order.completed_time) if order.completed_time else None, "deliverId": deliver_id_out, "deliverName": deliver_name, "deliverPhone": deliver_phone, "expectedTime": expected_time, "deliveryDispatchRemark": delivery_dispatch_remark, "deliveryTimeline": delivery_timeline, "supplierId": supplier_id_for_pickup, "supplierName": supplier_name, "supplierPhone": supplier_phone, "supplierPickup": { "address": supplier_pickup.get("fullAddress"), "contactPerson": supplier_pickup.get("contactPerson"), "contactPhone": supplier_pickup.get("contactPhone"), }, "deliveryTaskNo": delivery_task_no, "deliveryTaskStatus": delivery_task_status, "deliveryTaskDelivererStatus": delivery_task_deliverer_status, "deliveryTaskSupplierPrepareStatus": delivery_task_supplier_prepare_status, "deliveryPickupTime": delivery_pickup_time, "deliveryPickupRemark": delivery_pickup_remark, "deliveryDeliveryTime": delivery_delivery_time, "deliveryDeliveryRemark": delivery_delivery_remark, "deliveryPrepareImages": delivery_prepare_images, "deliveryPickupImages": delivery_pickup_images, "deliveryDeliverImages": delivery_deliver_images, "items": item_details, "statusLogs": log_details, "lockInfo": lock_info }) @router.post("/orders/{order_id}/unlock", response_model=dict) def unlock_order( order_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """手动解锁订单""" from app.crud.order import order_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") operator_id = current_user.get("user_id") try: order_crud.unlock_order(db, order_id, operator_id) return success_response({"message": "解锁成功"}) except ValueError as e: return error_response(400, str(e)) @router.post("/orders/{order_id}/review-cancel", response_model=dict) def review_order_cancel( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """审核取消订单""" from app.crud.order import order_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") if getattr(order, "cancel_status", "none") != "pending_review": return error_response(400, "只有待审核取消的订单才能审核") approved = request.get("approved", False) review_note = request.get("reviewNote") reviewer_id = current_user.get("user_id") order = order_crud.review_cancel( db, order_id, approved=approved, reviewer_id=reviewer_id, review_note=review_note ) return success_response({ "id": order.id, "status": order.status, "orderStatus": getattr(order, "order_status", None), "deliveryStatus": getattr(order, "delivery_status", None), "payStatus": getattr(order, "pay_status", None), "cancelStatus": getattr(order, "cancel_status", None), "message": "审核完成" }) @router.get("/orders/pending-cancel", response_model=dict) def list_pending_cancel_orders( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取待审核取消的订单列表""" from app.crud.order import order_crud, order_item_crud orders, total = order_crud.get_multi_with_filters( db, cancel_status="pending_review", skip=(page - 1) * pageSize, limit=pageSize ) region_map = {} client_map = {} result = [] for order in orders: if order.region_id and order.region_id not in region_map: region = region_crud.get(db, order.region_id) region_map[order.region_id] = region.name if region else None if order.client_id and order.client_id not in client_map: client = user_crud.get(db, order.client_id) client_map[order.client_id] = client.real_name if client else None item_count = order_item_crud.get_count_by_order(db, order.id) result.append({ "id": order.id, "orderNo": order.order_no, "clientId": order.client_id, "clientName": client_map.get(order.client_id), "regionId": order.region_id, "regionName": region_map.get(order.region_id), "totalAmount": float(order.total_amount) if order.total_amount else 0, "paidAmount": float(order.paid_amount) if order.paid_amount else 0, "cancelReason": order.cancel_reason, "itemCount": item_count, "createAt": str(order.create_at) if order.create_at else None }) return success_response({ "list": result, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) # ========================================= # 订单锁定管理 API # ========================================= @router.post("/orders/{order_id}/lock", response_model=dict) def lock_order( order_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """锁定订单""" from app.crud.order import order_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") try: order = order_crud.lock_order( db, order_id=order_id, operator_id=current_user.get("user_id") ) return success_response({ "id": order.id, "lockOperatorId": order.lock_operator_id, "lockTime": str(order.lock_time) if order.lock_time else None, "message": "锁定成功" }) except ValueError as e: return error_response(400, str(e)) @router.post("/orders/{order_id}/unlock", response_model=dict) def unlock_order( order_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """解锁订单""" from app.crud.order import order_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") try: order = order_crud.unlock_order( db, order_id=order_id, operator_id=current_user.get("user_id") ) return success_response({ "id": order.id, "message": "解锁成功" }) except ValueError as e: return error_response(400, str(e)) # ========================================= # 订单调整管理 API # ========================================= @router.post("/orders/{order_id}/adjust", response_model=dict) def adjust_order( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """订单调整（增量/减量/删除/替代）""" import json from app.crud.order import order_crud, order_item_crud, order_adjust_crud from decimal import Decimal from app.models.models import BizDeliveryTask, BizDeliveryTaskItem order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") # 检查订单是否被其他操作员锁定 operator_id = current_user.get("user_id") if order.lock_operator_id and order.lock_operator_id != operator_id: lock_time = order.lock_time.strftime("%Y-%m-%d %H:%M:%S") if order.lock_time else "未知" return error_response(400, f"订单已被其他操作员于 {lock_time} 锁定，请先解锁后再操作") # 仅线下支付订单允许运营调整 if getattr(order, "payment_method", None) != "offline": return error_response(400, "仅线下支付订单支持运营调整") # 新状态模型：仅未完成/未取消的线下单支持调整 if getattr(order, "order_status", None) not in ["pending_payment", "pending_confirm"]: return error_response(400, "当前状态不支持调整") adjustments = request.get("adjustments", []) address_update = request.get("addressUpdate") or {} has_address_update = isinstance(address_update, dict) and len(address_update.keys()) > 0 if not adjustments and not has_address_update: return error_response(400, "请提供调整项或地址调整信息") operator_name = current_user.get("real_name") adjust_results = [] for adj in adjustments: order_item_id = adj.get("orderItemId") adjust_type = adj.get("adjustType") # increase/reduce/delete/substitute new_quantity = adj.get("newQuantity") new_product_id = adj.get("newProductId") reason = adj.get("reason") if not order_item_id or not adjust_type: continue if not reason: return error_response(400, "调整原因不能为空") # 获取订单项 item = order_item_crud.get(db, order_item_id) if not item or item.order_id != order_id: continue before_quantity = item.quantity before_amount = item.amount try: if adjust_type == "increase": # 增量 if new_quantity is None or new_quantity < 0: return error_response(400, "增量后数量不能小于0") if new_quantity < float(item.quantity): return error_response(400, "增量后数量不能小于原数量") new_amount = Decimal(str(new_quantity)) * item.unit_price order_item_crud.update_item_adjustment( db, order_item_id, new_quantity=Decimal(str(new_quantity)), new_amount=new_amount, new_status="increased", note=reason, updater_id=operator_id, updater_name=operator_name ) snapshot_items = order_item_crud.get_by_order(db, order_id) snapshot = [{ "orderItemId": it.id, "productId": it.product_id, "quantity": float(it.quantity) if it.quantity else 0, "status": it.status, } for it in snapshot_items if it.is_del == 0] order_adjust_crud.create_adjust_record( db, order_id=order_id, order_item_id=order_item_id, adjust_type="increase", operator_id=operator_id, operator_name=operator_name, before_quantity=before_quantity, after_quantity=Decimal(str(new_quantity)), before_amount=before_amount, after_amount=new_amount, difference_amount=new_amount - before_amount, reason=reason, snapshot_items_json=json.dumps(snapshot, ensure_ascii=False) ) adjust_results.append({ "orderItemId": order_item_id, "adjustType": "increase", "success": True }) elif adjust_type == "reduce": # 减量 if new_quantity is None or new_quantity < 0: return error_response(400, "减量后数量不能小于0") if new_quantity > float(item.quantity): return error_response(400, "减量后数量不能大于原数量") new_amount = Decimal(str(new_quantity)) * item.unit_price order_item_crud.mark_as_reduced( db, order_item_id, new_quantity=Decimal(str(new_quantity)), new_amount=new_amount, note=reason ) snapshot_items = order_item_crud.get_by_order(db, order_id) snapshot = [{ "orderItemId": it.id, "productId": it.product_id, "quantity": float(it.quantity) if it.quantity else 0, "status": it.status, } for it in snapshot_items if it.is_del == 0] # 创建调整记录 order_adjust_crud.create_adjust_record( db, order_id=order_id, order_item_id=order_item_id, adjust_type="reduce", operator_id=operator_id, operator_name=operator_name, before_quantity=before_quantity, after_quantity=Decimal(str(new_quantity)), before_amount=before_amount, after_amount=new_amount, difference_amount=new_amount - before_amount, reason=reason, snapshot_items_json=json.dumps(snapshot, ensure_ascii=False) ) adjust_results.append({ "orderItemId": order_item_id, "adjustType": "reduce", "success": True }) elif adjust_type == "delete": # 删除 order_item_crud.mark_as_deleted( db, order_item_id, note=reason ) snapshot_items = order_item_crud.get_by_order(db, order_id) snapshot = [{ "orderItemId": it.id, "productId": it.product_id, "quantity": float(it.quantity) if it.quantity else 0, "status": it.status, } for it in snapshot_items if it.is_del == 0] # 创建调整记录 order_adjust_crud.create_adjust_record( db, order_id=order_id, order_item_id=order_item_id, adjust_type="delete", operator_id=operator_id, operator_name=operator_name, before_quantity=before_quantity, after_quantity=Decimal("0"), before_amount=before_amount, after_amount=Decimal("0"), difference_amount=-before_amount, reason=reason, snapshot_items_json=json.dumps(snapshot, ensure_ascii=False) ) adjust_results.append({ "orderItemId": order_item_id, "adjustType": "delete", "success": True }) elif adjust_type == "substitute": # 替代 if not new_product_id or not new_quantity: return error_response(400, "替代商品ID和数量不能为空") # 获取新商品信息 new_product = product_crud.get(db, new_product_id) if not new_product: return error_response(404, "替代商品不存在") # 需要获取新商品的价格 from app.crud.price_list import price_list_item_crud price_item = price_list_item_crud.get_by_product_and_price_list( db, new_product_id, order.price_list_id ) if not price_item: return error_response(400, "替代商品不在价格清单中") new_unit_price = price_item.price new_amount = Decimal(str(new_quantity)) * new_unit_price order_item_crud.update_item_adjustment( db, order_item_id, new_quantity=Decimal(str(new_quantity)), new_amount=new_amount, new_status="substituted", note=f"{reason}，替代商品: {new_product.name}", updater_id=operator_id, updater_name=operator_name ) snapshot_items = order_item_crud.get_by_order(db, order_id) snapshot = [{ "orderItemId": it.id, "productId": it.product_id, "quantity": float(it.quantity) if it.quantity else 0, "status": it.status, } for it in snapshot_items if it.is_del == 0] # 创建调整记录 order_adjust_crud.create_adjust_record( db, order_id=order_id, order_item_id=order_item_id, adjust_type="substitute", operator_id=operator_id, operator_name=operator_name, before_quantity=before_quantity, after_quantity=Decimal(str(new_quantity)), before_amount=before_amount, after_amount=new_amount, difference_amount=new_amount - before_amount, reason=reason, substitute_product_id=new_product_id, snapshot_items_json=json.dumps(snapshot, ensure_ascii=False) ) adjust_results.append({ "orderItemId": order_item_id, "adjustType": "substitute", "newProductId": new_product_id, "success": True }) except Exception as e: adjust_results.append({ "orderItemId": order_item_id, "adjustType": adjust_type, "success": False, "error": str(e) }) if has_address_update: old_delivery_address = order.delivery_address or "" old_contact_person = order.contact_person or "" old_contact_phone = order.contact_phone or "" new_delivery_address = str(address_update.get("deliveryAddress") or old_delivery_address).strip() new_contact_person = str(address_update.get("contactPerson") or old_contact_person).strip() new_contact_phone = str(address_update.get("contactPhone") or old_contact_phone).strip() address_reason = str(address_update.get("reason") or "").strip() address_changed = ( new_delivery_address != old_delivery_address or new_contact_person != old_contact_person or new_contact_phone != old_contact_phone ) if address_changed: order.delivery_address = new_delivery_address order.contact_person = new_contact_person order.contact_phone = new_contact_phone active_task = db.query(BizDeliveryTask).filter( BizDeliveryTask.order_id == order_id, BizDeliveryTask.is_del == 0 ).first() if active_task: task_items = db.query(BizDeliveryTaskItem).filter( BizDeliveryTaskItem.task_id == active_task.id, BizDeliveryTaskItem.is_del == 0 ).all() for task_item in task_items: task_item.delivery_address = new_delivery_address task_item.contact_person = new_contact_person task_item.contact_phone = new_contact_phone task_item.update_at = datetime.now() snapshot_items = order_item_crud.get_by_order(db, order_id) snapshot = [{ "orderItemId": it.id, "productId": it.product_id, "quantity": float(it.quantity) if it.quantity else 0, "status": it.status, } for it in snapshot_items if it.is_del == 0] reason_parts = [] reason_parts.append(f"地址：{old_delivery_address or '-'} -> {new_delivery_address or '-'}") reason_parts.append(f"联系人：{old_contact_person or '-'} -> {new_contact_person or '-'}") reason_parts.append(f"电话：{old_contact_phone or '-'} -> {new_contact_phone or '-'}") if address_reason: reason_parts.append(f"备注：{address_reason}") address_log_reason = "；".join(reason_parts) order_adjust_crud.create_adjust_record( db, order_id=order_id, order_item_id=None, adjust_type="address", operator_id=operator_id, operator_name=operator_name, before_quantity=Decimal("0"), after_quantity=Decimal("0"), before_amount=Decimal("0"), after_amount=Decimal("0"), difference_amount=Decimal("0"), reason=address_log_reason, snapshot_items_json=json.dumps(snapshot, ensure_ascii=False) ) adjust_results.append({ "adjustType": "address", "success": True }) # 重新计算订单应收金额 all_items = order_item_crud.get_by_order(db, order_id) new_total = sum( float(item.amount) for item in all_items if item.status != "deleted" and item.is_del == 0 ) new_receivable = Decimal(str(new_total)) order.total_amount = new_receivable order.receivable_amount = new_receivable order.update_at = datetime.now() db.commit() db.refresh(order) return success_response({ "id": order.id, "totalAmount": float(order.total_amount), "receivableAmount": float(order.receivable_amount), "adjustments": adjust_results, "message": "调整成功" }) @router.get("/orders/{order_id}/adjusts", response_model=dict) def list_order_adjusts( order_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取订单调整记录""" import json from app.crud.order import order_crud, order_adjust_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") adjusts = order_adjust_crud.get_by_order(db, order_id) result = [] for adj in adjusts: result.append({ "id": adj.id, "orderId": adj.order_id, "orderItemId": adj.order_item_id, "adjustType": adj.adjust_type, "beforeQuantity": float(adj.before_quantity) if adj.before_quantity else 0, "afterQuantity": float(adj.after_quantity) if adj.after_quantity else 0, "beforeAmount": float(adj.before_amount) if adj.before_amount else 0, "afterAmount": float(adj.after_amount) if adj.after_amount else 0, "differenceAmount": float(adj.difference_amount) if adj.difference_amount else 0, "reason": adj.reason, "snapshotItems": json.loads(adj.snapshot_items_json) if getattr(adj, "snapshot_items_json", None) else [], "operatorName": adj.operator_name, "createAt": str(adj.create_at) if adj.create_at else None }) return success_response({ "list": result, "total": len(result) }) @router.post("/orders/{order_id}/force-complete", response_model=dict) def force_complete_order( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """运营强制完结订单""" from app.crud.order import order_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") reason = request.get("reason", "运营强制完结") try: order = order_crud.force_complete( db, order_id=order_id, operator_id=current_user.get("user_id"), operator_role="admin", reason=reason ) return success_response({ "id": order.id, "status": order.status, "orderStatus": getattr(order, "order_status", None) or "pending_confirm", "deliveryStatus": getattr(order, "delivery_status", None) or "pending_dispatch", "payStatus": getattr(order, "pay_status", None) or "pending_payment", "completedTime": str(order.completed_time) if order.completed_time else None, "message": "强制完结成功" }) except ValueError as e: return error_response(400, str(e)) @router.post("/orders/{order_id}/partial-delivered", response_model=dict) def mark_partial_delivered( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """已废弃：旧的部分送达流程不再支持。""" return error_response(410, "该接口已下线，请使用配送任务状态流转接口") @router.post("/orders/{order_id}/confirm-partial", response_model=dict) def confirm_partial_delivery( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """已废弃：旧的部分送达确认流程不再支持。""" return error_response(410, "该接口已下线，请使用订单确认收货流程") @router.post("/orders/{order_id}/create-delivery-task", response_model=dict) def create_delivery_task_from_order( order_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """从订单创建配送任务""" from app.crud.order import order_crud, order_item_crud from app.crud.delivery import delivery_task_crud, delivery_task_item_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") # 只有待分配配送状态可以创建配送任务 if getattr(order, "delivery_status", None) != "pending_dispatch": return error_response(400, "只有待分配配送状态的订单可以创建配送任务") # 检查是否已存在配送任务 existing_task = delivery_task_crud.get_by_order(db, order_id) if existing_task: return error_response(400, "该订单已存在配送任务") # 获取订单关联的价格清单，以获取供应商ID from app.crud.price_list import price_list_crud price_list = price_list_crud.get(db, order.price_list_id) if not price_list: return error_response(400, "订单关联的价格清单不存在") supplier_id = price_list.supplier_id # 获取订单项 order_items = order_item_crud.get_by_order(db, order_id) # 构建配送任务明细数据 task_items_data = [] for item in order_items: if item.status == "deleted" or item.is_del != 0: continue task_items_data.append({ "order_item_id": item.id, "product_id": item.product_id, "client_id": order.client_id, "delivery_address": order.delivery_address, "contact_person": order.contact_person, "contact_phone": order.contact_phone, "quantity": float(item.quantity) if item.quantity else 0 }) creator = { "user_id": current_user.get("user_id"), "real_name": current_user.get("real_name") } # 创建配送任务及明细 task, task_items = delivery_task_crud.create_delivery_task_from_order( db, order_id=order_id, order_items=task_items_data, supplier_id=supplier_id, region_id=order.region_id, creator=creator ) # 阶段D收尾：优先维护三字段，再回写旧 status from app.core.order_status_compat import derive_legacy_status_from_three order.order_status = "pending_confirm" order.delivery_status = "delivering" order.status = derive_legacy_status_from_three( order_status=order.order_status, delivery_status=order.delivery_status, pay_status=order.pay_status, ) order.delivery_assign_time = datetime.now() order.update_at = datetime.now() db.commit() return success_response({ "id": task.id, "taskNo": task.task_no, "status": task.status, "effectiveTime": str(task.effective_time) if task.effective_time else None, "message": "配送任务创建成功" }) @router.post("/orders/{order_id}/assign-deliverer", response_model=dict) def assign_deliverer_to_order( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """分配配送员并创建配送任务（一步完成）""" from app.crud.order import order_crud, order_item_crud from app.crud.delivery import delivery_task_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") # 检查订单是否被其他操作员锁定 operator_id = current_user.get("user_id") if order.lock_operator_id and order.lock_operator_id != operator_id: lock_time = order.lock_time.strftime("%Y-%m-%d %H:%M:%S") if order.lock_time else "未知" return error_response(400, f"订单已被其他操作员于 {lock_time} 锁定，请先解锁后再操作") deliver_id = request.get("deliverId") if not deliver_id: return error_response(400, "请选择配送员") deadline_raw = request.get("deliveryDeadlineAt") or request.get("delivery_deadline_at") if not deadline_raw: return error_response(400, "请选择配送时效（截止时间）") dispatch_remark_raw = request.get("dispatchRemark") or request.get("dispatch_remark") or "" dispatch_remark = str(dispatch_remark_raw).strip()[:500] if dispatch_remark_raw else None effective_deadline = None try: ds = str(deadline_raw).strip().replace("Z", "+00:00") effective_deadline = datetime.fromisoformat(ds) if effective_deadline.tzinfo is not None: effective_deadline = effective_deadline.replace(tzinfo=None) except (TypeError, ValueError): effective_deadline = None if not effective_deadline: return error_response(400, "配送时效格式不正确") # 验证配送员 deliver = user_crud.get(db, deliver_id) if not deliver: return error_response(404, "配送员不存在") if deliver.role != "deliver": return error_response(400, "该用户不是配送员") if not deliver.region_id or deliver.region_id != order.region_id: return error_response(400, "只能选择订单所属区域的配送员") prev_delivery_status = getattr(order, "delivery_status", None) or "pending_dispatch" operator_name = (current_user.get("real_name") or "").strip() or None updater = {"user_id": operator_id, "real_name": operator_name} # 检查是否已存在配送任务 existing_task = delivery_task_crud.get_by_order(db, order_id) old_deliver_id = existing_task.deliver_id if existing_task else None if existing_task: task = delivery_task_crud.update_with_history( db, existing_task.id, { "deliver_id": deliver_id, "status": "assigned", "effective_time": effective_deadline, "dispatch_remark": dispatch_remark, }, updater=updater, ) if not task: return error_response(400, "更新配送任务失败") else: # 获取供应商ID from app.crud.price_list import price_list_crud price_list = price_list_crud.get(db, order.price_list_id) if not price_list: return error_response(400, "订单关联的价格清单不存在") # 获取订单项 order_items = order_item_crud.get_by_order(db, order_id) # 构建配送任务明细数据 task_items_data = [] for item in order_items: if item.status == "deleted" or item.is_del != 0: continue task_items_data.append({ "order_item_id": item.id, "product_id": item.product_id, "client_id": order.client_id, "delivery_address": order.delivery_address, "contact_person": order.contact_person, "contact_phone": order.contact_phone, "quantity": float(item.quantity) if item.quantity else 0 }) creator = { "user_id": current_user.get("user_id"), "real_name": current_user.get("real_name") } # 创建配送任务 task, _ = delivery_task_crud.create_delivery_task_from_order( db, order_id=order_id, order_items=task_items_data, supplier_id=price_list.supplier_id, region_id=order.region_id, creator=creator ) # 分配配送员 task.deliver_id = deliver_id task.status = "assigned" task.effective_time = effective_deadline task.dispatch_remark = dispatch_remark task.update_at = datetime.now() # 分配配送员后同步双状态；首次分配时供应商从待备货起算，更换配送员不重置备货进度 if task and task.deliver_id: task.deliverer_status = "pending_pickup" if not old_deliver_id: task.supplier_prepare_status = "pending_prepare" delivery_task_crud._sync_legacy_status_from_dual(task) # 阶段D收尾：优先维护三字段，再回写旧 status from app.core.order_status_compat import derive_legacy_status_from_three order.order_status = "pending_confirm" if not order.delivery_status or order.delivery_status == "pending_dispatch": order.delivery_status = "delivering" order.status = derive_legacy_status_from_three( order_status=order.order_status, delivery_status=order.delivery_status, pay_status=order.pay_status, ) order.delivery_assign_time = datetime.now() order.update_at = datetime.now() def _deliver_display_name(uid: Optional[str]) -> Optional[str]: if not uid: return None u = user_crud.get(db, uid) return (u.real_name if u else None) or uid if not existing_task: order_crud._create_status_log( db, order_id, "", "delivery.assign", operator_id, "admin", reason=json.dumps( { "fromDeliverId": None, "toDeliverId": deliver_id, "fromDeliverName": None, "toDeliverName": deliver.real_name, }, ensure_ascii=False, ), create_name=operator_name, ) elif old_deliver_id and old_deliver_id != deliver_id: order_crud._create_status_log( db, order_id, "", "delivery.reassign", operator_id, "admin", reason=json.dumps( { "fromDeliverId": old_deliver_id, "toDeliverId": deliver_id, "fromDeliverName": _deliver_display_name(old_deliver_id), "toDeliverName": deliver.real_name, }, ensure_ascii=False, ), create_name=operator_name, ) elif not old_deliver_id and deliver_id: order_crud._create_status_log( db, order_id, "", "delivery.assign", operator_id, "admin", reason=json.dumps( { "fromDeliverId": None, "toDeliverId": deliver_id, "fromDeliverName": None, "toDeliverName": deliver.real_name, }, ensure_ascii=False, ), create_name=operator_name, ) if prev_delivery_status != order.delivery_status: order_crud._create_status_log( db, order_id, f"ds_{prev_delivery_status}", f"ds_{order.delivery_status}", operator_id, "admin", create_name=operator_name, ) db.commit() db.refresh(task) return success_response({ "id": task.id, "taskNo": task.task_no, "deliverId": task.deliver_id, "deliverName": deliver.real_name, "status": task.status, "effectiveTime": str(task.effective_time) if task.effective_time else None, "message": "配送员分配成功" }) @router.post("/orders/{order_id}/assign-delivery", response_model=dict) def assign_delivery_alias( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """兼容旧路径：与 assign-deliverer 相同""" return assign_deliverer_to_order(order_id, request, db, current_user) # ========================================= # 结算管理 API (Web端) # ========================================= def _build_settlement_no(prefix: str) -> str: return f"{prefix}{datetime.now().strftime('%Y%m%d%H%M%S')}{str(uuid.uuid4().hex)[:6].upper()}" def _parse_datetime_range(start_date: Optional[str], end_date: Optional[str]): dt_start = None dt_end = None if start_date: try: dt_start = datetime.fromisoformat(start_date.replace("Z", "+00:00")) except Exception: dt_start = None if end_date: try: dt_end = datetime.fromisoformat(end_date.replace("Z", "+00:00")) except Exception: dt_end = None return dt_start, dt_end def _supplier_unit_yuan_from_pli(pli: Optional[BizPriceListItem]) -> Optional[float]: """价格清单行 price 为面向客户的最终单价；pricing_coefficient 为调价系数(%)，供货单价 = price / (1 + 系数/100)。""" if pli is None or getattr(pli, "price", None) is None: return None final_price = Decimal(str(pli.price)) raw_coeff = getattr(pli, "pricing_coefficient", None) if raw_coeff is None: coeff_pct = Decimal("0") else: try: coeff_pct = Decimal(str(raw_coeff)) except Exception: coeff_pct = Decimal("0") factor = Decimal("1") + coeff_pct / Decimal("100") if factor <= Decimal("0"): factor = Decimal("1") supplier_unit = (final_price / factor).quantize(Decimal("0.01")) return float(supplier_unit) def _price_list_item_map(db: Session, price_list_ids: List[str]) -> dict: """(price_list_id, product_id, unit_id) -> BizPriceListItem""" if not price_list_ids: return {} rows = ( db.query(BizPriceListItem) .filter(BizPriceListItem.price_list_id.in_(price_list_ids), BizPriceListItem.is_del == 0) .all() ) return {(r.price_list_id, r.product_id, r.unit_id): r for r in rows} def _batch_supplier_totals_for_orders(db: Session, order_ids: List[str]) -> dict: """订单维度供货价合计（元），用于供应商结算。""" if not order_ids: return {} orders = db.query(BizOrder).filter(BizOrder.id.in_(order_ids), BizOrder.is_del == 0).all() om = {o.id: o for o in orders} lines = ( db.query(BizOrderItem) .filter(BizOrderItem.order_id.in_(order_ids), BizOrderItem.is_del == 0) .all() ) by_oid: dict = {} for ln in lines: by_oid.setdefault(ln.order_id, []).append(ln) pl_ids = list({om[oid].price_list_id for oid in order_ids if oid in om and om[oid].price_list_id}) pli_map = _price_list_item_map(db, pl_ids) out: dict = {} for oid in order_ids: o = om.get(oid) if not o: out[oid] = 0.0 continue tot = Decimal("0") for ln in by_oid.get(oid, []): pli = pli_map.get((o.price_list_id, ln.product_id, ln.unit_id)) sup_u = _supplier_unit_yuan_from_pli(pli) if sup_u is None: sup_u = float(ln.unit_price or 0) qty = Decimal(str(ln.quantity or 0)) tot += qty * Decimal(str(sup_u)) out[oid] = float(tot.quantize(Decimal("0.01"))) return out def _build_order_line_children_for_settlement_list( db: Session, order: BizOrder, lines: List[BizOrderItem], pli_map: dict, product_map: dict, unit_map: dict, ) -> List[dict]: children: List[dict] = [] for ln in lines: p = product_map.get(ln.product_id) u = unit_map.get(ln.unit_id) pli = pli_map.get((order.price_list_id, ln.product_id, ln.unit_id)) sup_u = _supplier_unit_yuan_from_pli(pli) if sup_u is None: sup_u = float(ln.unit_price or 0) sell_u = float(ln.unit_price or 0) qty = float(ln.quantity or 0) sell_amt = float(ln.amount or 0) if ln.amount is not None else round(qty * sell_u, 2) sup_amt = round(qty * sup_u, 2) img = None if p and getattr(p, "image_url", None): img = str(p.image_url).strip() or None mv = None if p is not None and p.measure_value is not None: try: mv = float(p.measure_value) except (TypeError, ValueError): mv = None children.append({ "rowType": "line", "id": f"line-{ln.id}", "lineId": ln.id, "productId": ln.product_id, "productName": p.name if p else None, "skuCode": p.sku_code if p else None, "imageUrl": img, "unitName": u.name if u else None, "measureValue": mv, "quantity": qty, "sellingUnitPrice": sell_u, "sellingAmount": sell_amt, "supplierUnitPrice": sup_u, "supplierAmount": sup_amt, }) return children def _build_client_pending_settlement_tree_rows( db: Session, orders: List[BizOrder], client_map: dict, ) -> List[dict]: if not orders: return [] # 避免模块级循环依赖：在需要处局部导入 from app.crud.order import order_crud order_ids = [o.id for o in orders] region_ids = list({o.region_id for o in orders if o.region_id}) region_map: dict = {} for rid in region_ids: region = region_crud.get(db, rid) region_map[rid] = region.name if region else None lines = ( db.query(BizOrderItem) .filter(BizOrderItem.order_id.in_(order_ids), BizOrderItem.is_del == 0) .order_by(BizOrderItem.order_id.asc(), BizOrderItem.create_at.asc()) .all() ) by_oid: dict = {} for ln in lines: by_oid.setdefault(ln.order_id, []).append(ln) product_ids = list({ln.product_id for ln in lines}) unit_ids = list({ln.unit_id for ln in lines}) products = ( db.query(BasProduct).filter(BasProduct.id.in_(product_ids), BasProduct.is_del == 0).all() ) if product_ids else [] product_map = {p.id: p for p in products} units = ( db.query(BasProductUnit).filter(BasProductUnit.id.in_(unit_ids), BasProductUnit.is_del == 0).all() ) if unit_ids else [] unit_map = {u.id: u for u in units} pl_ids = list({o.price_list_id for o in orders if o.price_list_id}) pli_map = _price_list_item_map(db, pl_ids) price_lists = ( db.query(BizPriceList).filter(BizPriceList.id.in_(pl_ids), BizPriceList.is_del == 0).all() ) if pl_ids else [] supplier_id_by_price_list = {p.id: p.supplier_id for p in price_lists} supplier_ids = list({sid for sid in supplier_id_by_price_list.values() if sid}) suppliers = ( db.query(SysUser).filter(SysUser.id.in_(supplier_ids), SysUser.is_del == 0).all() ) if supplier_ids else [] supplier_name_map = {u.id: (u.real_name or u.username) for u in suppliers} out: List[dict] = [] for o in orders: ch = _build_order_line_children_for_settlement_list( db, o, by_oid.get(o.id, []), pli_map, product_map, unit_map ) sup_ref = sum(c["supplierAmount"] for c in ch) sel = o.order_status == "completed" out.append({ "rowType": "order", "id": o.id, "orderNo": o.order_no, "clientId": o.client_id, "clientName": client_map.get(o.client_id), "regionId": o.region_id, "regionName": region_map.get(o.region_id), "supplierId": supplier_id_by_price_list.get(o.price_list_id), "supplierName": supplier_name_map.get(supplier_id_by_price_list.get(o.price_list_id)), "totalAmount": float(o.total_amount or 0), "settlementAmount": float(o.total_amount or 0), "supplierReferenceTotal": round(sup_ref, 2), "orderStatus": o.order_status, "settlementSelectable": sel, "paymentMethod": order_crud.normalize_payment_method(o.payment_method), "clientSettlementStatus": o.client_settlement_status, "supplierSettlementStatus": o.supplier_settlement_status, "createAt": str(o.create_at) if o.create_at else None, "children": ch, }) return out def _build_supplier_pending_settlement_tree_rows( db: Session, order_supplier_pairs: List[Tuple[BizOrder, Optional[str]]], supplier_map: dict, ) -> List[dict]: if not order_supplier_pairs: return [] # 避免模块级循环依赖：在需要处局部导入 from app.crud.order import order_crud orders = [p[0] for p in order_supplier_pairs] sup_by_oid = {p[0].id: p[1] for p in order_supplier_pairs} order_ids = [o.id for o in orders] lines = ( db.query(BizOrderItem) .filter(BizOrderItem.order_id.in_(order_ids), BizOrderItem.is_del == 0) .order_by(BizOrderItem.order_id.asc(), BizOrderItem.create_at.asc()) .all() ) by_oid: dict = {} for ln in lines: by_oid.setdefault(ln.order_id, []).append(ln) product_ids = list({ln.product_id for ln in lines}) unit_ids = list({ln.unit_id for ln in lines}) products = ( db.query(BasProduct).filter(BasProduct.id.in_(product_ids), BasProduct.is_del == 0).all() ) if product_ids else [] product_map = {p.id: p for p in products} units = ( db.query(BasProductUnit).filter(BasProductUnit.id.in_(unit_ids), BasProductUnit.is_del == 0).all() ) if unit_ids else [] unit_map = {u.id: u for u in units} pl_ids = list({o.price_list_id for o in orders if o.price_list_id}) pli_map = _price_list_item_map(db, pl_ids) client_ids = list({o.client_id for o in orders if o.client_id}) client_map: dict = {} if client_ids: clients = db.query(SysUser).filter(SysUser.id.in_(client_ids), SysUser.is_del == 0).all() client_map = {u.id: (u.real_name or u.username) for u in clients} region_ids = list({o.region_id for o in orders if o.region_id}) region_map: dict = {} for rid in region_ids: region = region_crud.get(db, rid) region_map[rid] = region.name if region else None out: List[dict] = [] for o in orders: ch = _build_order_line_children_for_settlement_list( db, o, by_oid.get(o.id, []), pli_map, product_map, unit_map ) sup_order_total = sum(c["supplierAmount"] for c in ch) sel = o.order_status == "completed" sid = sup_by_oid.get(o.id) out.append({ "rowType": "order", "id": o.id, "orderNo": o.order_no, "clientId": o.client_id, "clientName": client_map.get(o.client_id), "regionId": o.region_id, "regionName": region_map.get(o.region_id), "supplierId": sid, "supplierName": supplier_map.get(sid) if sid else None, "settlementAmount": round(sup_order_total, 2), "totalAmount": float(o.total_amount or 0), "clientSellingTotal": float(o.total_amount or 0), "orderStatus": o.order_status, "settlementSelectable": sel, "supplierSettlementStatus": o.supplier_settlement_status, "clientSettlementStatus": o.client_settlement_status, "paymentMethod": order_crud.normalize_payment_method(o.payment_method), "createAt": str(o.create_at) if o.create_at else None, "children": ch, }) return out @router.get("/client-settlements/orders", response_model=dict) def list_client_settlement_orders( page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100), clientId: Optional[str] = Query(None), keyword: Optional[str] = Query(None, description="订单号"), startDate: Optional[str] = Query(None), endDate: Optional[str] = Query(None), regionId: Optional[str] = Query(None, description="区域（Tab）；超级管理员可选，区域管理员忽略"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): admin_region_id = get_admin_region_filter(db, current_user) eff_region_id = _effective_settlement_region_id(admin_region_id, regionId) dt_start, dt_end = _parse_datetime_range(startDate, endDate) q = db.query(BizOrder).filter( BizOrder.is_del == 0, BizOrder.order_status != "cancelled", BizOrder.payment_method == "offline", BizOrder.client_settlement_status == "pending", ) if eff_region_id: q = q.filter(BizOrder.region_id == eff_region_id) if clientId: q = q.filter(BizOrder.client_id == clientId) if keyword: q = q.filter(BizOrder.order_no.like(f"%{keyword}%")) if dt_start: q = q.filter(BizOrder.create_at >= dt_start) if dt_end: q = q.filter(BizOrder.create_at <= dt_end) total = q.count() settleable_order_count = q.filter(BizOrder.order_status == "completed").count() q_done = q.filter(BizOrder.order_status == "completed") pending_amount = float( q_done.with_entities(func.coalesce(func.sum(BizOrder.total_amount), 0)).scalar() or 0 ) rows = q.order_by(BizOrder.create_at.desc()).offset((page - 1) * pageSize).limit(pageSize).all() client_ids = list({r.client_id for r in rows if r.client_id}) client_map = {} if client_ids: for u in db.query(SysUser).filter(SysUser.id.in_(client_ids)).all(): client_map[u.id] = u.real_name or u.username data = _build_client_pending_settlement_tree_rows(db, rows, client_map) current_page_amount = float(sum((o.total_amount or 0) for o in rows if o.order_status == "completed")) return success_response({ "list": data, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize, "summary": { "pendingAmount": pending_amount, "currentPageAmount": current_page_amount, "settleableOrderCount": settleable_order_count, }, }) @router.get("/supplier-settlements/orders", response_model=dict) def list_supplier_settlement_orders( page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100), supplierId: Optional[str] = Query(None), keyword: Optional[str] = Query(None, description="订单号"), startDate: Optional[str] = Query(None), endDate: Optional[str] = Query(None), regionId: Optional[str] = Query(None, description="区域（Tab）；超级管理员可选，区域管理员忽略"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): admin_region_id = get_admin_region_filter(db, current_user) eff_region_id = _effective_settlement_region_id(admin_region_id, regionId) dt_start, dt_end = _parse_datetime_range(startDate, endDate) q = db.query(BizOrder, BizPriceList.supplier_id).join( BizPriceList, BizPriceList.id == BizOrder.price_list_id ).filter( BizOrder.is_del == 0, BizOrder.order_status != "cancelled", BizOrder.supplier_settlement_status == "pending", ) if eff_region_id: q = q.filter(BizOrder.region_id == eff_region_id) if supplierId: q = q.filter(BizPriceList.supplier_id == supplierId) if keyword: q = q.filter(BizOrder.order_no.like(f"%{keyword}%")) if dt_start: q = q.filter(BizOrder.create_at >= dt_start) if dt_end: q = q.filter(BizOrder.create_at <= dt_end) total = q.count() settleable_order_count = q.filter(BizOrder.order_status == "completed").count() q_done = q.filter(BizOrder.order_status == "completed") done_pairs = q_done.with_entities(BizOrder.id).all() done_oids = [x[0] for x in done_pairs] totals_map = _batch_supplier_totals_for_orders(db, done_oids) pending_amount = float(sum(totals_map.values())) rows = q.order_by(BizOrder.create_at.desc()).offset((page - 1) * pageSize).limit(pageSize).all() supplier_ids = list({s_id for _, s_id in rows if s_id}) supplier_map = {} if supplier_ids: for u in db.query(SysUser).filter(SysUser.id.in_(supplier_ids)).all(): supplier_map[u.id] = u.real_name or u.username data = _build_supplier_pending_settlement_tree_rows(db, rows, supplier_map) page_done_ids = [o.id for o, _ in rows if o.order_status == "completed"] page_totals = _batch_supplier_totals_for_orders(db, page_done_ids) current_page_amount = float(sum(page_totals.values())) return success_response({ "list": data, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize, "summary": { "pendingAmount": pending_amount, "currentPageAmount": current_page_amount, "settleableOrderCount": settleable_order_count, }, }) def _resolve_client_order_ids(db: Session, body: dict, admin_region_id: Optional[str]) -> List[str]: selection_mode = (body.get("selectionMode") or "ids").strip() filters = body.get("filters") or {} eff_region = _effective_settlement_region_id(admin_region_id, filters.get("regionId")) if selection_mode == "allFiltered": dt_start, dt_end = _parse_datetime_range(filters.get("startDate"), filters.get("endDate")) q = db.query(BizOrder.id).filter( BizOrder.is_del == 0, BizOrder.order_status != "cancelled", BizOrder.order_status == "completed", BizOrder.payment_method == "offline", BizOrder.client_settlement_status == "pending", ) if eff_region: q = q.filter(BizOrder.region_id == eff_region) if filters.get("clientId"): q = q.filter(BizOrder.client_id == filters["clientId"]) if filters.get("keyword"): q = q.filter(BizOrder.order_no.like(f"%{filters['keyword']}%")) if dt_start: q = q.filter(BizOrder.create_at >= dt_start) if dt_end: q = q.filter(BizOrder.create_at <= dt_end) order_ids = [x[0] for x in q.all()] excluded = set(body.get("excludedOrderIds") or []) return [oid for oid in order_ids if oid not in excluded] return list(set(body.get("orderIds") or [])) def _resolve_supplier_order_ids(db: Session, body: dict, admin_region_id: Optional[str]) -> List[str]: selection_mode = (body.get("selectionMode") or "ids").strip() filters = body.get("filters") or {} eff_region = _effective_settlement_region_id(admin_region_id, filters.get("regionId")) if selection_mode == "allFiltered": dt_start, dt_end = _parse_datetime_range(filters.get("startDate"), filters.get("endDate")) q = db.query(BizOrder.id).join( BizPriceList, BizPriceList.id == BizOrder.price_list_id ).filter( BizOrder.is_del == 0, BizOrder.order_status != "cancelled", BizOrder.order_status == "completed", BizOrder.supplier_settlement_status == "pending", ) if eff_region: q = q.filter(BizOrder.region_id == eff_region) if filters.get("supplierId"): q = q.filter(BizPriceList.supplier_id == filters["supplierId"]) if filters.get("keyword"): q = q.filter(BizOrder.order_no.like(f"%{filters['keyword']}%")) if dt_start: q = q.filter(BizOrder.create_at >= dt_start) if dt_end: q = q.filter(BizOrder.create_at <= dt_end) order_ids = [x[0] for x in q.all()] excluded = set(body.get("excludedOrderIds") or []) return [oid for oid in order_ids if oid not in excluded] return list(set(body.get("orderIds") or [])) @router.post("/client-settlements", response_model=dict) def create_client_settlement( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): admin_region_id = get_admin_region_filter(db, current_user) body = dict(request or {}) voucher_urls = body.get("voucherUrls") or [] order_ids = _resolve_client_order_ids(db, body, admin_region_id) if not order_ids: return error_response(400, "未选择可结算订单") orders = db.query(BizOrder).filter( BizOrder.id.in_(order_ids), BizOrder.is_del == 0, BizOrder.order_status == "completed", BizOrder.payment_method == "offline", BizOrder.client_settlement_status == "pending", ).all() if not orders: return error_response(400, "无可结算订单") eff_region = _effective_settlement_region_id( admin_region_id, (body.get("filters") or {}).get("regionId") ) if eff_region and any((o.region_id or "") != eff_region for o in orders): return error_response(400, "存在不属于当前区域的订单") total_amount = sum((o.total_amount or 0) for o in orders) client_id = orders[0].client_id if any(o.client_id != client_id for o in orders): return error_response(400, "客户结算必须选择同一客户") client = user_crud.get(db, client_id) now = datetime.now() creator = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} settlement = OrdClientSettlement( id=generate_id(), settlement_no=_build_settlement_no("CST"), client_id=client_id, client_name=(client.real_name if client else None), order_count=len(orders), total_amount=total_amount, voucher_urls=json.dumps(voucher_urls, ensure_ascii=False), remark=body.get("remark"), settlement_at=now, status="done", create_by=creator["user_id"], create_name=creator["real_name"], update_by=creator["user_id"], ) db.add(settlement) db.flush() for order in orders: db.add(OrdClientSettlementItem( id=generate_id(), settlement_id=settlement.id, order_id=order.id, order_no=order.order_no, client_id=order.client_id, order_amount=order.total_amount or 0, create_by=creator["user_id"], create_name=creator["real_name"], update_by=creator["user_id"], )) order.client_settlement_status = "settled" order.client_settlement_at = now order.update_by = creator["user_id"] order.update_at = now db.commit() return success_response({ "id": settlement.id, "settlementNo": settlement.settlement_no, "orderCount": settlement.order_count, "totalAmount": float(settlement.total_amount or 0), }) @router.post("/supplier-settlements", response_model=dict) def create_supplier_settlement( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): admin_region_id = get_admin_region_filter(db, current_user) body = dict(request or {}) voucher_urls = body.get("voucherUrls") or [] order_ids = _resolve_supplier_order_ids(db, body, admin_region_id) if not order_ids: return error_response(400, "未选择可结算订单") rows = db.query(BizOrder, BizPriceList.supplier_id).join( BizPriceList, BizPriceList.id == BizOrder.price_list_id ).filter( BizOrder.id.in_(order_ids), BizOrder.is_del == 0, BizOrder.order_status == "completed", BizOrder.supplier_settlement_status == "pending", ).all() if not rows: return error_response(400, "无可结算订单") eff_region = _effective_settlement_region_id( admin_region_id, (body.get("filters") or {}).get("regionId") ) if eff_region and any((order.region_id or "") != eff_region for order, _ in rows): return error_response(400, "存在不属于当前区域的订单") supplier_id = rows[0][1] if any(s_id != supplier_id for _, s_id in rows): return error_response(400, "商家结算必须选择同一供应商") orders = [r[0] for r in rows] supplier_totals = _batch_supplier_totals_for_orders(db, [o.id for o in orders]) total_amount = sum(supplier_totals.get(o.id, 0) for o in orders) supplier = user_crud.get(db, supplier_id) if supplier_id else None now = datetime.now() creator = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} settlement = OrdSupplierSettlement( id=generate_id(), settlement_no=_build_settlement_no("SST"), supplier_id=supplier_id, supplier_name=(supplier.real_name if supplier else None), order_count=len(orders), total_amount=total_amount, voucher_urls=json.dumps(voucher_urls, ensure_ascii=False), remark=body.get("remark"), settlement_at=now, status="done", create_by=creator["user_id"], create_name=creator["real_name"], update_by=creator["user_id"], ) db.add(settlement) db.flush() for order in orders: db.add(OrdSupplierSettlementItem( id=generate_id(), settlement_id=settlement.id, order_id=order.id, order_no=order.order_no, supplier_id=supplier_id, order_amount=supplier_totals.get(order.id, 0), create_by=creator["user_id"], create_name=creator["real_name"], update_by=creator["user_id"], )) order.supplier_settlement_status = "settled" order.supplier_settlement_at = now order.update_by = creator["user_id"] order.update_at = now db.commit() return success_response({ "id": settlement.id, "settlementNo": settlement.settlement_no, "orderCount": settlement.order_count, "totalAmount": float(settlement.total_amount or 0), }) def _sheet_voucher_urls_list(raw) -> list: if not raw: return [] if isinstance(raw, list): return [str(x) for x in raw if x] s = str(raw).strip() if not s: return [] try: data = json.loads(s) if isinstance(data, list): return [str(x) for x in data if x] except Exception: pass return [s] def _build_settlement_detail_item_tree(db: Session, settlement_items: List[Any]) -> List[dict]: """结算单明细：每个订单一行父节点，子行为订单商品行（与前端树表对应）。""" from app.crud.order import order_crud if not settlement_items: return [] order_ids = list({it.order_id for it in settlement_items if getattr(it, "order_id", None)}) if not order_ids: return [] orders = ( db.query(BizOrder) .filter(BizOrder.id.in_(order_ids), BizOrder.is_del == 0) .all() ) order_map = {o.id: o for o in orders} client_ids = list({o.client_id for o in orders if o.client_id}) client_map: dict = {} if client_ids: clients = db.query(SysUser).filter(SysUser.id.in_(client_ids), SysUser.is_del == 0).all() client_map = {u.id: (u.real_name or u.username) for u in clients} region_ids = list({o.region_id for o in orders if o.region_id}) region_map: dict = {} for rid in region_ids: r = region_crud.get(db, rid) region_map[rid] = r.name if r else None lines = ( db.query(BizOrderItem) .filter(BizOrderItem.order_id.in_(order_ids), BizOrderItem.is_del == 0) .order_by(BizOrderItem.order_id.asc(), BizOrderItem.create_at.asc()) .all() ) by_order: dict = {} for ln in lines: by_order.setdefault(ln.order_id, []).append(ln) product_ids = list({ln.product_id for ln in lines}) unit_ids = list({ln.unit_id for ln in lines}) products = ( db.query(BasProduct) .filter(BasProduct.id.in_(product_ids), BasProduct.is_del == 0) .all() ) if product_ids else [] product_map = {p.id: p for p in products} category_ids = list({p.category_id for p in products if p.category_id}) category_map: dict = {} for cid in category_ids: c = category_crud.get(db, cid) category_map[cid] = c.name if c else None units = ( db.query(BasProductUnit) .filter(BasProductUnit.id.in_(unit_ids), BasProductUnit.is_del == 0) .all() ) if unit_ids else [] unit_map = {u.id: u for u in units} pl_ids = list({o.price_list_id for o in order_map.values() if o and o.price_list_id}) pli_map = _price_list_item_map(db, pl_ids) out: List[dict] = [] for it in settlement_items: order = order_map.get(it.order_id) rid = order.region_id if order else None children: List[dict] = [] for ln in by_order.get(it.order_id, []): p = product_map.get(ln.product_id) u = unit_map.get(ln.unit_id) img = None if p and getattr(p, "image_url", None): img = str(p.image_url).strip() or None mv = None if p is not None and p.measure_value is not None: try: mv = float(p.measure_value) except (TypeError, ValueError): mv = None pli = pli_map.get((order.price_list_id, ln.product_id, ln.unit_id)) if order else None sup_u = _supplier_unit_yuan_from_pli(pli) if sup_u is None: sup_u = float(ln.unit_price or 0) sell_u = float(ln.unit_price or 0) qty = float(ln.quantity or 0) sell_amt = float(ln.amount or 0) if ln.amount is not None else round(qty * sell_u, 2) sup_amt = round(qty * sup_u, 2) children.append({ "rowType": "line", "id": f"line-{ln.id}", "lineId": ln.id, "productId": ln.product_id, "productName": p.name if p else None, "skuCode": p.sku_code if p else None, "imageUrl": img, "categoryName": category_map.get(p.category_id) if p and p.category_id else None, "unitName": u.name if u else None, "measureValue": mv, "quantity": qty, "unitPrice": sell_u, "amount": sell_amt, "sellingUnitPrice": sell_u, "sellingAmount": sell_amt, "supplierUnitPrice": sup_u, "supplierAmount": sup_amt, "deliveredQuantity": float(ln.delivered_quantity or 0), "note": ln.note, }) out.append({ "rowType": "order", "id": f"order-{it.id}", "settlementItemId": it.id, "orderId": it.order_id, "orderNo": it.order_no, "orderAmount": float(it.order_amount or 0), "totalAmount": float(it.order_amount or 0), "clientId": order.client_id if order else None, "clientName": client_map.get(order.client_id) if order else None, "regionId": rid, "regionName": region_map.get(rid), "orderCreateAt": str(order.create_at) if order and order.create_at else None, "contactPerson": order.contact_person if order else None, "contactPhone": order.contact_phone if order else None, "deliveryAddress": order.delivery_address if order else None, "orderStatus": order.order_status if order else None, "paymentMethod": order_crud.normalize_payment_method(order.payment_method) if order else None, "createAt": str(it.create_at) if it.create_at else None, "children": children, }) return out @router.get("/client-settlements/sheets", response_model=dict) def list_client_settlement_sheets( page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100), clientId: Optional[str] = Query(None), settlementNo: Optional[str] = Query(None), startDate: Optional[str] = Query(None), endDate: Optional[str] = Query(None), regionId: Optional[str] = Query(None, description="区域（Tab）；超级管理员可选"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): """客户结算单（批次）列表""" from app.crud.ord_settlement import ord_client_settlement_crud admin_region_id = get_admin_region_filter(db, current_user) eff_region_id = _effective_settlement_region_id(admin_region_id, regionId) dt_start, dt_end = _parse_datetime_range(startDate, endDate) rows, total = ord_client_settlement_crud.get_multi_with_filters( db, client_id=clientId, settlement_no=settlementNo, start_date=dt_start, end_date=dt_end, region_id=eff_region_id, skip=(page - 1) * pageSize, limit=pageSize, ) out = [] for s in rows: out.append({ "id": s.id, "settlementNo": s.settlement_no, "clientId": s.client_id, "clientName": s.client_name, "orderCount": s.order_count, "totalAmount": float(s.total_amount or 0), "settlementAt": str(s.settlement_at) if s.settlement_at else None, "status": s.status, "remark": s.remark, "createAt": str(s.create_at) if s.create_at else None, "createName": s.create_name, }) return success_response({ "list": out, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize, }) @router.get("/client-settlements/sheets/{sheet_id}", response_model=dict) def get_client_settlement_sheet_detail( sheet_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): """客户结算单详情（含明细订单）""" from app.crud.ord_settlement import ord_client_settlement_crud admin_region_id = get_admin_region_filter(db, current_user) main, items = ord_client_settlement_crud.get_with_items(db, sheet_id) if not main: return error_response(404, "结算单不存在") if admin_region_id and not ord_client_settlement_crud.belongs_to_region(db, sheet_id, admin_region_id): return error_response(403, "无权限查看该结算单") item_out = _build_settlement_detail_item_tree(db, items) return success_response({ "id": main.id, "settlementNo": main.settlement_no, "clientId": main.client_id, "clientName": main.client_name, "orderCount": main.order_count, "totalAmount": float(main.total_amount or 0), "voucherUrls": _sheet_voucher_urls_list(main.voucher_urls), "remark": main.remark, "settlementAt": str(main.settlement_at) if main.settlement_at else None, "status": main.status, "createAt": str(main.create_at) if main.create_at else None, "createName": main.create_name, "items": item_out, }) @router.put("/client-settlements/sheets/{sheet_id}", response_model=dict) def update_client_settlement_sheet( sheet_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): """更新客户结算单（备注/凭证）；UPDATE 前写入 ord_client_settlement_his""" from app.crud.ord_settlement import ord_client_settlement_crud admin_region_id = get_admin_region_filter(db, current_user) if admin_region_id and not ord_client_settlement_crud.belongs_to_region(db, sheet_id, admin_region_id): return error_response(403, "无权限操作该结算单") body = dict(request or {}) update_data = {} if "remark" in body: update_data["remark"] = body.get("remark") if "voucherUrls" in body: urls = body.get("voucherUrls") or [] if not isinstance(urls, list): return error_response(400, "voucherUrls 须为数组") update_data["voucher_urls"] = json.dumps(urls, ensure_ascii=False) if not update_data: return error_response(400, "无可更新字段") updater = {"user_id": current_user.get("user_id")} row = ord_client_settlement_crud.update_with_history(db, sheet_id, update_data, updater) if not row: return error_response(404, "结算单不存在") return success_response({ "id": row.id, "settlementNo": row.settlement_no, "remark": row.remark, "voucherUrls": _sheet_voucher_urls_list(row.voucher_urls), }) @router.get("/supplier-settlements/sheets", response_model=dict) def list_supplier_settlement_sheets( page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100), supplierId: Optional[str] = Query(None), settlementNo: Optional[str] = Query(None), startDate: Optional[str] = Query(None), endDate: Optional[str] = Query(None), regionId: Optional[str] = Query(None, description="区域（Tab）；超级管理员可选"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): """供应商结算单（批次）列表""" from app.crud.ord_settlement import ord_supplier_settlement_crud admin_region_id = get_admin_region_filter(db, current_user) eff_region_id = _effective_settlement_region_id(admin_region_id, regionId) dt_start, dt_end = _parse_datetime_range(startDate, endDate) rows, total = ord_supplier_settlement_crud.get_multi_with_filters( db, supplier_id=supplierId, settlement_no=settlementNo, start_date=dt_start, end_date=dt_end, region_id=eff_region_id, skip=(page - 1) * pageSize, limit=pageSize, ) out = [] for s in rows: out.append({ "id": s.id, "settlementNo": s.settlement_no, "supplierId": s.supplier_id, "supplierName": s.supplier_name, "orderCount": s.order_count, "totalAmount": float(s.total_amount or 0), "settlementAt": str(s.settlement_at) if s.settlement_at else None, "status": s.status, "remark": s.remark, "createAt": str(s.create_at) if s.create_at else None, "createName": s.create_name, }) return success_response({ "list": out, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize, }) @router.get("/supplier-settlements/sheets/{sheet_id}", response_model=dict) def get_supplier_settlement_sheet_detail( sheet_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): """供应商结算单详情（含明细订单）""" from app.crud.ord_settlement import ord_supplier_settlement_crud admin_region_id = get_admin_region_filter(db, current_user) main, items = ord_supplier_settlement_crud.get_with_items(db, sheet_id) if not main: return error_response(404, "结算单不存在") if admin_region_id and not ord_supplier_settlement_crud.belongs_to_region(db, sheet_id, admin_region_id): return error_response(403, "无权限查看该结算单") item_out = _build_settlement_detail_item_tree(db, items) return success_response({ "id": main.id, "settlementNo": main.settlement_no, "supplierId": main.supplier_id, "supplierName": main.supplier_name, "orderCount": main.order_count, "totalAmount": float(main.total_amount or 0), "voucherUrls": _sheet_voucher_urls_list(main.voucher_urls), "remark": main.remark, "settlementAt": str(main.settlement_at) if main.settlement_at else None, "status": main.status, "createAt": str(main.create_at) if main.create_at else None, "createName": main.create_name, "items": item_out, }) @router.put("/supplier-settlements/sheets/{sheet_id}", response_model=dict) def update_supplier_settlement_sheet( sheet_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin), ): """更新供应商结算单（备注/凭证）；UPDATE 前写入 ord_supplier_settlement_his""" from app.crud.ord_settlement import ord_supplier_settlement_crud admin_region_id = get_admin_region_filter(db, current_user) if admin_region_id and not ord_supplier_settlement_crud.belongs_to_region(db, sheet_id, admin_region_id): return error_response(403, "无权限操作该结算单") body = dict(request or {}) update_data = {} if "remark" in body: update_data["remark"] = body.get("remark") if "voucherUrls" in body: urls = body.get("voucherUrls") or [] if not isinstance(urls, list): return error_response(400, "voucherUrls 须为数组") update_data["voucher_urls"] = json.dumps(urls, ensure_ascii=False) if not update_data: return error_response(400, "无可更新字段") updater = {"user_id": current_user.get("user_id")} row = ord_supplier_settlement_crud.update_with_history(db, sheet_id, update_data, updater) if not row: return error_response(404, "结算单不存在") return success_response({ "id": row.id, "settlementNo": row.settlement_no, "remark": row.remark, "voucherUrls": _sheet_voucher_urls_list(row.voucher_urls), }) @router.get("/settlements", response_model=dict) def list_settlements( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), regionId: Optional[str] = Query(None, description="区域ID"), clientId: Optional[str] = Query(None, description="客户ID"), status: Optional[str] = Query(None, description="状态"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取线下结算订单列表""" from app.crud.settlement import settlement_crud settlements, total = settlement_crud.get_multi_with_filters( db, region_id=regionId, client_id=clientId, status=status, start_date=startDate, end_date=endDate, skip=(page - 1) * pageSize, limit=pageSize ) region_map = {} client_map = {} result = [] for settlement in settlements: if settlement.region_id and settlement.region_id not in region_map: region = region_crud.get(db, settlement.region_id) region_map[settlement.region_id] = region.name if region else None if settlement.client_id and settlement.client_id not in client_map: client = user_crud.get(db, settlement.client_id) client_map[settlement.client_id] = client.real_name if client else None result.append({ "id": settlement.id, "orderId": settlement.order_id, "clientId": settlement.client_id, "clientName": client_map.get(settlement.client_id), "regionId": settlement.region_id, "regionName": region_map.get(settlement.region_id), "amount": float(settlement.amount) if settlement.amount else 0, "settlementMethod": settlement.settlement_method, "settlementTime": str(settlement.settlement_time) if settlement.settlement_time else None, "settlementNote": settlement.settlement_note, "status": settlement.status, "reviewedBy": settlement.reviewed_by, "reviewedTime": str(settlement.reviewed_time) if settlement.reviewed_time else None, "createAt": str(settlement.create_at) if settlement.create_at else None }) return success_response({ "list": result, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/settlements/client-summary", response_model=dict) def get_client_summary( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), regionId: Optional[str] = Query(None, description="区域ID"), clientId: Optional[str] = Query(None, description="客户ID"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取客户待结算汇总""" from app.crud.settlement import settlement_crud summaries, total = settlement_crud.get_client_summary( db, region_id=regionId, client_id=clientId, skip=(page - 1) * pageSize, limit=pageSize ) result = [] for summary in summaries: client = user_crud.get(db, summary["client_id"]) result.append({ "clientId": summary["client_id"], "clientName": client.real_name if client else None, "orderCount": summary["order_count"], "totalUnpaid": float(summary["total_unpaid"]) if summary["total_unpaid"] else 0 }) return success_response({ "list": result, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.post("/settlements", response_model=dict) def create_settlement( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """确认线下结算""" from app.crud.settlement import settlement_crud from datetime import datetime from decimal import Decimal order_id = request.get("orderId") settlement_method = request.get("settlementMethod") settlement_time_str = request.get("settlementTime") note = request.get("settlementNote") attachment_url = request.get("attachmentUrl") if not order_id: return error_response(400, "请选择订单") if not settlement_method: return error_response(400, "请选择结算方式") # 获取订单信息 from app.models.models import BizOrder order = db.query(BizOrder).filter(BizOrder.id == order_id).first() if not order: return error_response(404, "订单不存在") # 计算未付金额 unpaid_amount = (order.receivable_amount or Decimal("0")) - (order.paid_amount or Decimal("0")) if unpaid_amount <= 0: return error_response(400, "该订单已无欠款，无需结算") # 解析结算时间 settlement_time = None if settlement_time_str: try: settlement_time = datetime.fromisoformat(settlement_time_str.replace("Z", "+00:00")) except: settlement_time = datetime.now() creator = { "user_id": current_user.get("user_id"), "real_name": current_user.get("real_name") } settlement = settlement_crud.create_settlement( db, order_id=order_id, client_id=order.client_id, region_id=order.region_id, amount=unpaid_amount, settlement_method=settlement_method, settlement_time=settlement_time, attachment_url=attachment_url, note=note, creator=creator ) return success_response({ "id": settlement.id, "message": "结算成功" }) @router.post("/settlements/batch", response_model=dict) def batch_create_settlements( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """批量确认线下结算""" from app.crud.settlement import settlement_crud from datetime import datetime order_ids = request.get("orderIds", []) settlement_method = request.get("settlementMethod") settlement_time_str = request.get("settlementTime") note = request.get("settlementNote") attachment_url = request.get("attachmentUrl") if not order_ids: return error_response(400, "请选择要结算的订单") if not settlement_method: return error_response(400, "请选择结算方式") if len(order_ids) > 50: return error_response(400, "单次最多结算50个订单") # 解析结算时间 settlement_time = None if settlement_time_str: try: settlement_time = datetime.fromisoformat(settlement_time_str.replace("Z", "+00:00")) except: settlement_time = datetime.now() creator = { "user_id": current_user.get("user_id"), "real_name": current_user.get("real_name") } settlements = settlement_crud.batch_create_settlements( db, order_ids=order_ids, settlement_method=settlement_method, settlement_time=settlement_time, attachment_url=attachment_url, note=note, creator=creator ) return success_response({ "count": len(settlements), "message": f"成功结算{len(settlements)}个订单" }) @router.get("/settlements/records", response_model=dict) def list_settlement_records( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), regionId: Optional[str] = Query(None, description="区域ID"), clientId: Optional[str] = Query(None, description="客户ID"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取结算记录列表""" from app.crud.settlement import settlement_crud settlements, total = settlement_crud.get_multi_with_filters( db, region_id=regionId, client_id=clientId, status="confirmed", start_date=startDate, end_date=endDate, skip=(page - 1) * pageSize, limit=pageSize ) region_map = {} client_map = {} reviewer_map = {} result = [] for settlement in settlements: if settlement.region_id and settlement.region_id not in region_map: region = region_crud.get(db, settlement.region_id) region_map[settlement.region_id] = region.name if region else None if settlement.client_id and settlement.client_id not in client_map: client = user_crud.get(db, settlement.client_id) client_map[settlement.client_id] = client.real_name if client else None if settlement.reviewed_by and settlement.reviewed_by not in reviewer_map: reviewer = user_crud.get(db, settlement.reviewed_by) reviewer_map[settlement.reviewed_by] = reviewer.real_name if reviewer else None result.append({ "id": settlement.id, "orderId": settlement.order_id, "clientId": settlement.client_id, "clientName": client_map.get(settlement.client_id), "regionId": settlement.region_id, "regionName": region_map.get(settlement.region_id), "amount": float(settlement.amount) if settlement.amount else 0, "settlementMethod": settlement.settlement_method, "settlementTime": str(settlement.settlement_time) if settlement.settlement_time else None, "settlementNote": settlement.settlement_note, "settlementFileUrl": settlement.settlement_file_url, "reviewedBy": settlement.reviewed_by, "reviewedByName": reviewer_map.get(settlement.reviewed_by), "reviewedTime": str(settlement.reviewed_time) if settlement.reviewed_time else None, "createAt": str(settlement.create_at) if settlement.create_at else None }) return success_response({ "list": result, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/settlements/statistics", response_model=dict) def get_settlement_statistics( regionId: Optional[str] = Query(None, description="区域ID"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取结算统计数据""" from app.crud.settlement import settlement_crud # 区域管理员只能查看自己区域的统计数据 admin_region_id = get_admin_region_filter(db, current_user) if admin_region_id and not regionId: regionId = admin_region_id elif admin_region_id and regionId != admin_region_id: return success_response({ "pendingOrderCount": 0, "pendingAmount": 0, "settledCount": 0, "settledAmount": 0, "refundPendingCount": 0 }) stats = settlement_crud.get_statistics( db, region_id=regionId, start_date=startDate, end_date=endDate ) return success_response(stats) # ========================================= # 退款管理 API (Web端) # ========================================= @router.get("/refunds", response_model=dict) def list_refunds( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), regionId: Optional[str] = Query(None, description="区域ID"), status: Optional[str] = Query(None, description="状态"), orderNo: Optional[str] = Query(None, description="订单号"), refundMethod: Optional[str] = Query(None, description="退款方式"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取退款记录列表""" from app.crud.settlement import refund_crud from app.models.models import BizOrder refunds, total = refund_crud.get_multi_with_filters( db, region_id=regionId, order_no=orderNo, status=status, refund_method=refundMethod, start_date=startDate, end_date=endDate, skip=(page - 1) * pageSize, limit=pageSize ) order_map = {} result = [] for refund in refunds: if refund.order_id and refund.order_id not in order_map: order = db.query(BizOrder).filter(BizOrder.id == refund.order_id).first() if order: client = user_crud.get(db, order.client_id) order_map[refund.order_id] = { "orderNo": order.order_no, "clientId": order.client_id, "clientName": client.real_name if client else None, "regionId": order.region_id, "totalAmount": float(order.total_amount) if order.total_amount else 0, "paidAmount": float(order.paid_amount) if order.paid_amount else 0 } order_info = order_map.get(refund.order_id, {}) result.append({ "id": refund.id, "orderId": refund.order_id, "orderNo": order_info.get("orderNo"), "clientId": order_info.get("clientId"), "clientName": order_info.get("clientName"), "regionId": order_info.get("regionId"), "totalAmount": order_info.get("totalAmount"), "paidAmount": order_info.get("paidAmount"), "refundNo": refund.refund_no, "refundAmount": float(refund.refund_amount) if refund.refund_amount else 0, "refundMethod": refund.refund_method, "refundResult": refund.refund_result, "refundTime": str(refund.refund_time) if refund.refund_time else None, "refundNoWechat": refund.refund_no_wechat, "refundFailReason": refund.refund_fail_reason, "status": refund.status, "createAt": str(refund.create_at) if refund.create_at else None }) return success_response({ "list": result, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/refunds/{refund_id}", response_model=dict) def get_refund( refund_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取退款详情""" from app.crud.settlement import refund_crud from app.models.models import BizOrder refund = refund_crud.get(db, refund_id) if not refund: return error_response(404, "退款记录不存在") order = db.query(BizOrder).filter(BizOrder.id == refund.order_id).first() client = user_crud.get(db, order.client_id) if order else None return success_response({ "id": refund.id, "orderId": refund.order_id, "orderNo": order.order_no if order else None, "clientId": order.client_id if order else None, "clientName": client.real_name if client else None, "totalAmount": float(order.total_amount) if order and order.total_amount else 0, "paidAmount": float(order.paid_amount) if order and order.paid_amount else 0, "refundNo": refund.refund_no, "refundAmount": float(refund.refund_amount) if refund.refund_amount else 0, "refundMethod": refund.refund_method, "refundResult": refund.refund_result, "refundTime": str(refund.refund_time) if refund.refund_time else None, "refundNoWechat": refund.refund_no_wechat, "refundFailReason": refund.refund_fail_reason, "status": refund.status, "createAt": str(refund.create_at) if refund.create_at else None, "createName": refund.create_name }) @router.post("/refunds/wechat-refund/{order_id}", response_model=dict) def create_wechat_refund( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """微信退款""" from app.crud.settlement import refund_crud from datetime import datetime from decimal import Decimal refund_amount = request.get("refundAmount") reason = request.get("reason") if not refund_amount or refund_amount <= 0: return error_response(400, "退款金额必须大于0") # 获取订单信息 from app.models.models import BizOrder order = db.query(BizOrder).filter(BizOrder.id == order_id).first() if not order: return error_response(404, "订单不存在") # 检查已支付金额 paid_amount = float(order.paid_amount) if order.paid_amount else 0 if refund_amount > paid_amount: return error_response(400, "退款金额不能超过已支付金额") # 检查是否有处理中的退款 existing_refunds = refund_crud.get_by_order(db, order_id) processing_refund = any(r.status in ["pending", "processing"] for r in existing_refunds) if processing_refund: return error_response(400, "该订单有退款正在处理中，请勿重复提交") creator = { "user_id": current_user.get("user_id"), "real_name": current_user.get("real_name") } refund = refund_crud.create_refund( db, order_id=order_id, refund_amount=Decimal(str(refund_amount)), refund_method="wechat", reason=reason, creator=creator ) # 模拟微信退款处理（实际项目中需要调用微信支付API） refund_crud.process_refund( db, refund_id=refund.id, refund_result="success", refund_no_wechat=f"WXREF{datetime.now().strftime('%Y%m%d%H%M%S')}" ) return success_response({ "refundId": refund.id, "refundNo": refund.refund_no, "refundAmount": float(refund.refund_amount), "refundResult": "success", "refundTime": str(refund.refund_time) }) @router.get("/refunds/{refund_id}/status", response_model=dict) def get_refund_status( refund_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """查询退款状态""" from app.crud.settlement import refund_crud refund = refund_crud.get(db, refund_id) if not refund: return error_response(404, "退款记录不存在") return success_response({ "refundId": refund.id, "refundNo": refund.refund_no, "refundAmount": float(refund.refund_amount) if refund.refund_amount else 0, "refundResult": refund.refund_result, "refundTime": str(refund.refund_time) if refund.refund_time else None, "refundNoWechat": refund.refund_no_wechat }) @router.post("/refunds/{refund_id}/process", response_model=dict) def process_refund( refund_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """处理退款（成功/失败）""" from app.crud.settlement import refund_crud refund_result = request.get("refundResult") refund_no_wechat = request.get("refundNoWechat") if not refund_result or refund_result not in ["success", "failed"]: return error_response(400, "请提供正确的退款结果") refund = refund_crud.process_refund( db, refund_id=refund_id, refund_result=refund_result, refund_no_wechat=refund_no_wechat ) if not refund: return error_response(404, "退款记录不存在") return success_response({ "refundId": refund.id, "refundResult": refund.refund_result, "message": "处理成功" }) @router.post("/refunds/{refund_id}/handle-failure", response_model=dict) def handle_refund_failure( refund_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """处理退款失败""" from app.crud.settlement import refund_crud fail_reason = request.get("failReason") if not fail_reason: return error_response(400, "请提供失败原因") refund = refund_crud.handle_refund_failure( db, refund_id=refund_id, fail_reason=fail_reason ) if not refund: return error_response(404, "退款记录不存在") return success_response({ "refundId": refund.id, "refundResult": refund.refund_result, "message": "处理成功" }) @router.get("/refunds/statistics", response_model=dict) def get_refund_statistics( startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取退款统计数据""" from app.crud.settlement import refund_crud stats = refund_crud.get_refund_statistics( db, start_date=startDate, end_date=endDate ) return success_response(stats) # ========================================= # 配送时效预警 API # ========================================= @router.get("/delivery-warnings", response_model=dict) def list_delivery_warnings( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), regionId: Optional[str] = Query(None, description="区域ID"), urgencyLevel: Optional[str] = Query(None, description="紧急程度: critical/urgent/normal"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取配送时效预警列表""" from app.crud.delivery import delivery_task_crud from datetime import datetime, timedelta # 计算时间阈值 now = datetime.now() critical_time = now + timedelta(minutes=30) # 30分钟内 urgent_time = now + timedelta(minutes=60) # 1小时内 normal_time = now + timedelta(hours=2) # 2小时内 # 获取所有待配送任务 query = db.query(delivery_task_crud.model).filter( and_( delivery_task_crud.model.is_del == 0, delivery_task_crud.model.status.in_(["pending", "assigned", "picking", "picked"]), delivery_task_crud.model.effective_time != None, delivery_task_crud.model.effective_time >= now ) ) if regionId: query = query.filter(delivery_task_crud.model.region_id == regionId) all_tasks = query.order_by(delivery_task_crud.model.effective_time.asc()).all() # 计算紧急程度 result = [] for task in all_tasks: time_diff = (task.effective_time - now).total_seconds() / 60 # 分钟 if time_diff <= 30: urgency = "critical" remaining_minutes = int(time_diff) elif time_diff <= 60: urgency = "urgent" remaining_minutes = int(time_diff) else: urgency = "normal" remaining_minutes = int(time_diff) # 按紧急程度过滤 if urgencyLevel and urgency != urgencyLevel: continue # 获取关联信息 client_name = None if task.order_id: from app.crud.order import order_crud order = order_crud.get(db, task.order_id) if order: client = user_crud.get(db, order.client_id) client_name = client.real_name if client else None region_name = None if task.region_id: region = region_crud.get(db, task.region_id) region_name = region.name if region else None deliver_name = None if task.deliver_id: deliver = user_crud.get(db, task.deliver_id) deliver_name = deliver.real_name if deliver else None supplier_name = None if task.supplier_id: supplier = user_crud.get(db, task.supplier_id) supplier_name = supplier.real_name if supplier else None result.append({ "id": task.id, "taskNo": task.task_no, "orderId": task.order_id, "regionId": task.region_id, "regionName": region_name, "clientName": client_name, "supplierId": task.supplier_id, "supplierName": supplier_name, "deliverId": task.deliver_id, "deliverName": deliver_name, "status": task.status, "effectiveTime": str(task.effective_time) if task.effective_time else None, "remainingMinutes": remaining_minutes, "urgencyLevel": urgency, "isAssigned": task.deliver_id is not None, "createAt": str(task.create_at) if task.create_at else None }) # 手动分页 total = len(result) start_idx = (page - 1) * pageSize end_idx = start_idx + pageSize paginated_result = result[start_idx:end_idx] return success_response({ "list": paginated_result, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/delivery-warnings/statistics", response_model=dict) def get_delivery_warning_statistics( regionId: Optional[str] = Query(None, description="区域ID"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取配送时效预警统计""" from app.crud.delivery import delivery_task_crud from datetime import datetime, timedelta now = datetime.now() # 基础查询 base_query = db.query(delivery_task_crud.model).filter( and_( delivery_task_crud.model.is_del == 0, delivery_task_crud.model.status.in_(["pending", "assigned", "picking", "picked"]), delivery_task_crud.model.effective_time != None, delivery_task_crud.model.effective_time >= now ) ) if regionId: base_query = base_query.filter(delivery_task_crud.model.region_id == regionId) all_tasks = base_query.all() critical_count = 0 urgent_count = 0 normal_count = 0 unassigned_count = 0 total_pending = len(all_tasks) for task in all_tasks: time_diff = (task.effective_time - now).total_seconds() / 60 if time_diff <= 30: critical_count += 1 elif time_diff <= 60: urgent_count += 1 else: normal_count += 1 if not task.deliver_id: unassigned_count += 1 return success_response({ "totalPending": total_pending, "criticalCount": critical_count, "urgentCount": urgent_count, "normalCount": normal_count, "unassignedCount": unassigned_count }) @router.post("/delivery-warnings/{task_id}/reassign", response_model=dict) def reassign_delivery_task( task_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """紧急重新分配配送任务""" from app.crud.delivery import delivery_task_crud new_deliver_id = request.get("deliverId") reason = request.get("reason") if not new_deliver_id: return error_response(400, "请选择新的配送员") task = delivery_task_crud.get(db, task_id) if not task: return error_response(404, "任务不存在") if task.status in ["completed", "cancelled"]: return error_response(400, "已完成或已取消的任务无法重新分配") # 验证新配送员 new_deliver = user_crud.get(db, new_deliver_id) if not new_deliver: return error_response(404, "配送员不存在") if new_deliver.role != "deliver": return error_response(400, "该用户不是配送员") old_deliver_id = task.deliver_id old_deliver_name = None if old_deliver_id: old_deliver = user_crud.get(db, old_deliver_id) old_deliver_name = old_deliver.real_name if old_deliver else None # 更新任务（换配送员：只重置配送员侧待取货，不重置供应商备货进度） task.deliver_id = new_deliver_id task.update_at = datetime.now() task.deliverer_status = "pending_pickup" delivery_task_crud._sync_legacy_status_from_dual(task) # 如果之前没有分配配送员，设置新的截止时间 if not old_deliver_id: task.effective_time = datetime.now() + timedelta(hours=4) db.commit() db.refresh(task) return success_response({ "id": task.id, "taskNo": task.task_no, "newDeliverId": new_deliver_id, "newDeliverName": new_deliver.real_name, "oldDeliverId": old_deliver_id, "oldDeliverName": old_deliver_name, "message": "任务重新分配成功" }) def get_client_ip(request: Request) -> str: """获取客户端IP地址""" forwarded = request.headers.get("X-Forwarded-For") if forwarded: return forwarded.split(",")[0].strip() return request.client.host if request.client else "unknown" # ========================================= # API 请求日志（中间件落库，此处仅查询/删除） # ========================================= @router.get("/api-request-logs", response_model=dict) def list_api_request_logs( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), httpMethod: Optional[str] = Query(None, description="HTTP 方法"), statusCode: Optional[int] = Query(None, description="HTTP 状态码"), pathPrefix: Optional[str] = Query(None, description="路径包含"), keyword: Optional[str] = Query(None, description="关键词"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """API 请求日志列表""" from app.crud.api_request_log import api_request_log_crud logs, total = api_request_log_crud.get_multi_with_filters( db, http_method=httpMethod, status_code=statusCode, path_prefix=pathPrefix, keyword=keyword, start_date=startDate, end_date=endDate, skip=(page - 1) * pageSize, limit=pageSize, ) return success_response({ "list": [{ "id": log.id, "httpMethod": log.http_method, "path": log.path, "queryString": log.query_string, "statusCode": log.status_code, "durationMs": log.duration_ms, "userId": log.user_id, "username": log.username, "role": log.role, "ipAddress": log.ip_address, "userAgent": log.user_agent, "requestBody": log.request_body, "responseBody": log.response_body, "createAt": str(log.create_at) if log.create_at else None, } for log in logs], "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/api-request-logs/{log_id}", response_model=dict) def get_api_request_log( log_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """API 请求日志详情""" from app.crud.api_request_log import api_request_log_crud log = api_request_log_crud.get(db, log_id) if not log: return error_response(404, "日志不存在") return success_response({ "id": log.id, "httpMethod": log.http_method, "path": log.path, "queryString": log.query_string, "statusCode": log.status_code, "durationMs": log.duration_ms, "userId": log.user_id, "username": log.username, "role": log.role, "ipAddress": log.ip_address, "userAgent": log.user_agent, "requestBody": log.request_body, "responseBody": log.response_body, "createAt": str(log.create_at) if log.create_at else None, }) @router.delete("/api-request-logs/{log_id}", response_model=dict) def delete_api_request_log( log_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """删除 API 请求日志（软删除）""" from app.crud.api_request_log import api_request_log_crud if not api_request_log_crud.delete(db, log_id): return error_response(404, "日志不存在") return success_response({"id": log_id}) # ========================================= # 数据统计 API # ========================================= @router.get("/statistics/dashboard", response_model=dict) def get_dashboard_statistics( regionId: Optional[str] = Query(None, description="区域ID"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取仪表盘统计数据""" from app.crud.statistics import statistics_crud stats = statistics_crud.get_dashboard_summary(db, region_id=regionId) return success_response(stats) @router.get("/statistics/orders", response_model=dict) def get_order_statistics( regionId: Optional[str] = Query(None, description="区域ID"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取订单统计数据""" from app.crud.statistics import statistics_crud stats = statistics_crud.get_order_statistics( db, region_id=regionId, start_date=startDate, end_date=endDate ) return success_response(stats) @router.get("/statistics/orders/trend", response_model=dict) def get_order_trend( regionId: Optional[str] = Query(None, description="区域ID"), days: int = Query(7, ge=1, le=30, description="统计天数"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取订单趋势数据""" from app.crud.statistics import statistics_crud trend = statistics_crud.get_order_trend( db, region_id=regionId, days=days ) return success_response(trend) @router.get("/statistics/financial", response_model=dict) def get_financial_statistics( regionId: Optional[str] = Query(None, description="区域ID"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取财务统计数据""" from app.crud.statistics import statistics_crud stats = statistics_crud.get_financial_statistics( db, region_id=regionId, start_date=startDate, end_date=endDate ) return success_response(stats) @router.get("/statistics/financial/trend", response_model=dict) def get_financial_trend( regionId: Optional[str] = Query(None, description="区域ID"), days: int = Query(7, ge=1, le=30, description="统计天数"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取财务趋势数据""" from app.crud.statistics import statistics_crud trend = statistics_crud.get_financial_trend( db, region_id=regionId, days=days ) return success_response(trend) @router.get("/statistics/delivery", response_model=dict) def get_delivery_statistics( regionId: Optional[str] = Query(None, description="区域ID"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取配送统计数据""" from app.crud.statistics import statistics_crud stats = statistics_crud.get_delivery_statistics( db, region_id=regionId, start_date=startDate, end_date=endDate ) return success_response(stats) @router.get("/statistics/products", response_model=dict) def get_product_statistics( regionId: Optional[str] = Query(None, description="区域ID"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), limit: int = Query(10, ge=1, le=50, description="返回数量"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取商品销售排行""" from app.crud.statistics import statistics_crud stats = statistics_crud.get_product_statistics( db, region_id=regionId, start_date=startDate, end_date=endDate, limit=limit ) return success_response(stats) @router.get("/statistics/customers", response_model=dict) def get_customer_statistics( regionId: Optional[str] = Query(None, description="区域ID"), startDate: Optional[str] = Query(None, description="开始日期"), endDate: Optional[str] = Query(None, description="结束日期"), limit: int = Query(10, ge=1, le=50, description="返回数量"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取客户订单排行""" from app.crud.statistics import statistics_crud stats = statistics_crud.get_customer_statistics( db, region_id=regionId, start_date=startDate, end_date=endDate, limit=limit ) return success_response(stats) # ========================================= # 缺失的订单相关 API # ========================================= @router.post("/orders/{order_id}/complete", response_model=dict) def force_complete_order( order_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """完结订单""" from app.crud.order import order_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") # 新状态模型：待收货阶段可由运营强制完结 if getattr(order, "order_status", None) != "pending_confirm": return error_response(400, "只有待收货订单才能完结") try: completed_order = order_crud.force_complete(db, order_id, current_user.get("user_id")) return success_response({ "id": completed_order.id, "status": completed_order.status, "message": "订单已完结" }) except Exception as e: return error_response(400, str(e)) @router.post("/orders/{order_id}/cancel", response_model=dict) def cancel_order_admin( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """运营端取消订单""" from app.crud.order import order_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") # 新状态模型：仅未完成且未取消订单允许运营取消 if getattr(order, "order_status", None) in ["completed", "cancelled"]: return error_response(400, "已完成或已取消订单不能再次取消") reason = request.get("reason", "") note = request.get("note", "") evidence_image_urls = request.get("evidenceImageUrls", []) or [] if not reason: return error_response(400, "请填写取消原因") if not isinstance(evidence_image_urls, list): return error_response(400, "evidenceImageUrls 格式不正确") try: cancelled_order = order_crud.admin_cancel( db, order_id=order_id, operator_id=current_user.get("user_id"), operator_name=current_user.get("real_name"), reason=reason, note=note, evidence_image_urls=evidence_image_urls, ) if evidence_image_urls: from app.models.models import BizAttachment from app.core.utils import generate_id for url in evidence_image_urls: if not isinstance(url, str) or not url.strip(): continue db.add(BizAttachment( id=generate_id(), business_type="order", business_id=order_id, file_name=url.split("/")[-1][:200], file_url=url, file_type="image/*", upload_by=current_user.get("user_id"), upload_by_name=current_user.get("real_name"), create_by=current_user.get("user_id"), create_name=current_user.get("real_name"), )) db.commit() return success_response({ "id": cancelled_order.id, "status": cancelled_order.status, "message": "订单已取消" }) except Exception as e: return error_response(400, str(e)) @router.get("/orders/adjustable", response_model=dict) def list_adjustable_orders( keyword: Optional[str] = Query(None, description="关键词搜索"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取可调整的订单列表""" from app.crud.order import order_crud, order_item_crud # 可调整的订单状态：待支付、待收货（线下单） adjustable_statuses = ["pending_payment", "pending_confirm"] orders, total = order_crud.get_multi_with_filters( db, status=None, # 不按状态过滤，我们手动过滤 order_no=keyword, skip=0, limit=100 # 默认获取100条 ) # 过滤出可调整的订单 adjustable_orders = [] for order in orders: if getattr(order, "order_status", None) in adjustable_statuses and order_crud.normalize_payment_method(order.payment_method) == "offline": item_count = order_item_crud.get_count_by_order(db, order.id) adjustable_orders.append({ "id": order.id, "orderNo": order.order_no, "status": order.status, "orderStatus": getattr(order, "order_status", None) or "pending_confirm", "deliveryStatus": getattr(order, "delivery_status", None) or "pending_dispatch", "payStatus": getattr(order, "pay_status", None) or "pending_payment", "statusText": ORDER_STATUS_MAP.get(order.status, order.status), "totalAmount": float(order.total_amount) if order.total_amount else 0, "itemCount": item_count, "createAt": str(order.create_at) if order.create_at else None }) return success_response({ "list": adjustable_orders, "total": len(adjustable_orders) }) @router.get("/orders/{order_id}/items", response_model=dict) def get_order_items( order_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取订单商品列表""" from app.crud.order import order_crud, order_item_crud order = order_crud.get(db, order_id) if not order: return error_response(404, "订单不存在") items = order_item_crud.get_by_order(db, order_id) from app.crud.price_list import price_list_crud price_list = price_list_crud.get(db, order.price_list_id) if order.price_list_id else None supplier_id_from_price_list = getattr(price_list, "supplier_id", None) if price_list else None supplier_name_from_price_list = None if supplier_id_from_price_list: _sup_u = user_crud.get(db, supplier_id_from_price_list) if _sup_u: supplier_name_from_price_list = _sup_u.real_name or _sup_u.username pl_ids_for_order = [order.price_list_id] if order.price_list_id else [] pli_map_for_order = _price_list_item_map(db, pl_ids_for_order) from app.crud.delivery import delivery_task_item_crud as _merge_task_delivered_items _line_delivered_from_task = _merge_task_delivered_items.get_delivered_quantity_by_order_item(db, order_id) result = [] for item in items: product = product_crud.get(db, item.product_id) unit_name = None if item.unit_id: u = unit_crud.get(db, item.unit_id) unit_name = u.name if u else None category_name = None if product and product.category_id: cat = category_crud.get(db, product.category_id) category_name = cat.name if cat else None _oi_dq = float(item.delivered_quantity) if item.delivered_quantity else 0.0 _ti_dq = float(_line_delivered_from_task.get(item.id, 0.0)) _merged_dq = max(_oi_dq, _ti_dq) qty = float(item.quantity) if item.quantity else 0 selling_u = float(item.unit_price) if item.unit_price else 0 selling_amt = float(item.amount) if item.amount else round(qty * selling_u, 2) pli = pli_map_for_order.get((order.price_list_id, item.product_id, item.unit_id)) supplier_u = _supplier_unit_yuan_from_pli(pli) if supplier_u is None: supplier_u = selling_u supplier_amt = round(qty * supplier_u, 2) result.append({ "id": item.id, "productId": item.product_id, "productName": product.name if product else None, "imageUrl": getattr(product, "image_url", None) if product else None, "categoryName": category_name, "unitId": item.unit_id, "unitName": unit_name, "measureValue": _numeric_to_float_optional(getattr(product, "measure_value", None)) if product else None, "supplierId": supplier_id_from_price_list, "supplierName": supplier_name_from_price_list, "quantity": qty, "unitPrice": selling_u, "amount": selling_amt, "sellingUnitPrice": selling_u, "sellingAmount": selling_amt, "supplierUnitPrice": supplier_u, "supplierAmount": supplier_amt, "deliveredQuantity": _merged_dq, "status": item.status }) return success_response({ "list": result, "total": len(result) }) # ========================================= # 商品简单列表 API # ========================================= @router.get("/products/simple", response_model=dict) def list_products_simple( keyword: Optional[str] = Query(None, description="关键词搜索"), categoryId: Optional[str] = Query(None, description="分类ID"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取商品简单列表（用于订单调整选择替代商品）""" from app.models.models import BizProduct query = db.query(BizProduct).filter(BizProduct.is_del == 0, BizProduct.is_active == 1) if keyword: query = query.filter(BizProduct.name.ilike(f"%{keyword}%")) if categoryId: query = query.filter(BizProduct.category_id == categoryId) products = query.order_by(BizProduct.name).limit(100).all() result = [] for p in products: category_name = None if p.category_id: cat = category_crud.get(db, p.category_id) category_name = cat.name if cat else None unit_name = None if p.unit_id: u = unit_crud.get(db, p.unit_id) unit_name = u.name if u else None result.append({ "id": p.id, "name": p.name, "categoryId": p.category_id, "categoryName": category_name, "unitId": p.unit_id, "unitName": unit_name, "imageUrl": p.image_url }) return success_response({ "list": result, "total": len(result) }) # ========================================= # 询价单简单列表 API # ========================================= @router.get("/inquiries/simple", response_model=dict) def list_inquiries_simple( status: Optional[str] = Query(None, description="状态"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取询价单简单列表（用于报价对比选择）""" from app.models.models import BizInquiry query = db.query(BizInquiry).filter(BizInquiry.is_del == 0) if status: query = query.filter(BizInquiry.status == status) inquiries = query.order_by(BizInquiry.create_at.desc()).limit(50).all() result = [] for inquiry in inquiries: region_name = None if inquiry.region_id: region = region_crud.get(db, inquiry.region_id) region_name = region.name if region else None result.append({ "id": inquiry.id, "title": inquiry.title, "status": inquiry.status, "regionId": inquiry.region_id, "regionName": region_name, "createAt": str(inquiry.create_at) if inquiry.create_at else None }) return success_response({ "list": result, "total": len(result) }) # ========================================= # 报价确认供应商 API # ========================================= @router.post("/quotations/confirm-supplier", response_model=dict) def confirm_quotation_supplier( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """确认选择供应商报价，并生成区域生效价格清单（active） 约束： - 同一询价单，仅允许 1 个报价单为 selected - 报价单状态仅使用：submitted / selected - 不改变询价单状态（仍为 published 等业务状态） """ from app.models.models import BizInquiry, BizQuotation, BizPriceList, BizPriceListItem from app.crud.quotation import quotation_crud, quotation_item_crud from app.crud.inquiry import inquiry_item_crud from app.crud.product import product_crud from app.core.utils import generate_id from sqlalchemy import and_ from decimal import Decimal inquiry_id = request.get("inquiryId") quotation_id = request.get("quotationId") final_price_items = request.get("finalPriceItems") or [] if not inquiry_id or not quotation_id: return error_response(400, "询价单ID和报价ID不能为空") inquiry = db.query(BizInquiry).filter(and_(BizInquiry.id == inquiry_id, BizInquiry.is_del == 0)).first() if not inquiry: return error_response(404, "询价单不存在") quotation = quotation_crud.get(db, quotation_id) if not quotation: return error_response(404, "报价不存在") if quotation.inquiry_id != inquiry_id: return error_response(400, "报价不属于该询价单") # 询价单必须处于发布中，才允许选择供应商生成生效价格 if inquiry.status != "published": return error_response(400, "只能对已发布状态的询价单选择供应商报价") # 同一询价仅允许一个 selected：先将其它报价恢复为 submitted db.query(BizQuotation).filter( and_( BizQuotation.inquiry_id == inquiry_id, BizQuotation.id != quotation_id, BizQuotation.is_del == 0 ) ).update({"status": "submitted"}, synchronize_session=False) # 当前报价置为 selected quotation.status = "selected" # 生成价格清单：范围仅询价单的商品 inquiry_items = inquiry_item_crud.get_by_inquiry(db, inquiry_id) inquiry_product_ids = {i.product_id for i in inquiry_items if i and i.product_id} if not inquiry_product_ids: return error_response(400, "询价单必须包含至少一个商品") quotation_items = quotation_item_crud.get_by_quotation(db, quotation_id) quotation_item_by_pid: dict[str, any] = {} for qi in quotation_items or []: pid = str(getattr(qi, "product_id", "") or "") if not pid: continue if pid in inquiry_product_ids: quotation_item_by_pid[pid] = qi # 允许供应商某些商品未报价：只要运营端传入了最终售价(finalPriceItems)即可生成价格清单 if not quotation_item_by_pid and not final_price_items: return error_response(400, "该报价单未包含本询价单商品报价，且未传入最终售价，无法生成价格清单") # 计算版本号（区域维度自增） max_version_row = db.query(BizPriceList).filter( and_(BizPriceList.region_id == inquiry.region_id, BizPriceList.is_del == 0) ).order_by(BizPriceList.version.desc()).first() next_version = (max_version_row.version + 1) if max_version_row and max_version_row.version else 1 # 将该区域其他 active 价格清单置为 inactive db.query(BizPriceList).filter( and_( BizPriceList.region_id == inquiry.region_id, BizPriceList.status == "active", BizPriceList.is_del == 0 ) ).update({"status": "inactive"}, synchronize_session=False) # 创建新价格清单（直接生效） price_list_id = generate_id() now = datetime.now() price_list = BizPriceList( id=price_list_id, region_id=inquiry.region_id, supplier_id=quotation.supplier_id, inquiry_id=inquiry_id, quotation_id=quotation_id, source_type="quotation_select", version=next_version, status="active", effective_time=now, published_by=current_user.get("user_id"), published_time=now, remark=f"由询价单选择供应商报价生成（询价={inquiry_id}, 报价={quotation_id}）", create_by=current_user.get("user_id"), create_name=current_user.get("real_name") ) db.add(price_list) db.flush() # 可选：前端传入最终售卖价 + 本次调价系数（用于回显与二次发布） # key 统一使用 str(product_id)，避免类型不一致导致匹配失败 final_price_map: dict[str, Decimal] = {} pricing_coeff_map: dict[str, Decimal] = {} if final_price_items: if not isinstance(final_price_items, list): return error_response(400, "finalPriceItems 参数格式错误") for it in final_price_items: if not isinstance(it, dict): return error_response(400, "finalPriceItems 参数格式错误") pid = it.get("productId") price = it.get("price") pricing_coeff = it.get("pricingCoefficient") if not pid: return error_response(400, "finalPriceItems.productId 不能为空") pid_key = str(pid) try: price_decimal = Decimal(str(price)) except Exception: return error_response(400, f"商品 {pid} 的价格格式不正确") if price_decimal <= Decimal("0"): return error_response(400, f"商品 {pid} 的价格必须大于0") final_price_map[pid_key] = price_decimal.quantize(Decimal("0.01")) if pricing_coeff is not None and pricing_coeff != "": try: coeff_decimal = Decimal(str(pricing_coeff)) except Exception: return error_response(400, f"商品 {pid} 的调价系数格式不正确") pricing_coeff_map[pid_key] = coeff_decimal.quantize(Decimal("0.01")) # 写入价格清单明细：范围=询价单的商品（不要求供应商都报价） for ii in inquiry_items: pid_key = str(getattr(ii, "product_id", "") or "") if not pid_key: continue qi = quotation_item_by_pid.get(pid_key) final_price = final_price_map.get(pid_key) pricing_coeff = pricing_coeff_map.get(pid_key) # 最终售价优先取前端传入；否则回退为供应商报价 if final_price is None: if qi is not None and getattr(qi, "price", None) is not None: final_price = Decimal(str(qi.price)).quantize(Decimal("0.01")) else: return error_response(400, f"商品 {pid_key} 缺少供应商报价，且未传入最终售价，无法生成价格清单") # 单位ID：优先报价单项；否则从商品表取 unit_id = getattr(qi, "unit_id", None) if qi is not None else None if not unit_id: product_for_unit = product_crud.get(db, pid_key) unit_id = getattr(product_for_unit, "unit_id", None) if product_for_unit else None if not unit_id: return error_response(400, f"商品 {pid_key} 缺少单位信息，无法生成价格清单") # 若前端未传调价系数：尝试通过(最终价, 供应商报价)反推；再兜底商品默认系数 if pricing_coeff is None: if qi is not None and getattr(qi, "price", None) is not None: try: supplier_price_decimal = Decimal(str(qi.price)) except Exception: supplier_price_decimal = Decimal("0") if supplier_price_decimal > Decimal("0"): try: pricing_coeff = ((final_price / supplier_price_decimal) - Decimal("1")) * Decimal("100") pricing_coeff = pricing_coeff.quantize(Decimal("0.01")) except Exception: pricing_coeff = None if pricing_coeff is None: product = product_crud.get(db, pid_key) raw_default = getattr(product, "price_coefficient", None) if product else None try: pricing_coeff = Decimal(str(raw_default)).quantize(Decimal("0.01")) if raw_default is not None else Decimal("0.00") except Exception: pricing_coeff = Decimal("0.00") pli = BizPriceListItem( id=generate_id(), price_list_id=price_list_id, product_id=pid_key, unit_id=unit_id, price=final_price, pricing_coefficient=pricing_coeff, create_by=current_user.get("user_id"), create_name=current_user.get("real_name") ) db.add(pli) db.commit() return success_response({ "message": "已选择供应商报价并生成生效价格", "inquiryId": inquiry_id, "quotationId": quotation_id, "priceListId": price_list_id, "priceVersion": next_version }) # ========================================= # 价格版本管理 API # ========================================= @router.get("/price-versions", response_model=dict) def list_price_versions( page: int = Query(1, ge=1, description="页码"), pageSize: int = Query(10, ge=1, le=100, description="每页条数"), regionId: Optional[str] = Query(None, description="区域ID"), supplierId: Optional[str] = Query(None, description="供应商ID"), status: Optional[str] = Query(None, description="状态"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取价格版本列表""" from app.models.models import BizPriceList from sqlalchemy import or_ query = db.query(BizPriceList).filter(BizPriceList.is_del == 0) if regionId: query = query.filter(BizPriceList.region_id == regionId) if supplierId: query = query.filter(BizPriceList.supplier_id == supplierId) if status: query = query.filter(BizPriceList.status == status) total = query.count() versions = query.order_by(BizPriceList.create_at.desc()).offset((page - 1) * pageSize).limit(pageSize).all() result = [] for v in versions: region_name = None if v.region_id: region = region_crud.get(db, v.region_id) region_name = region.name if region else None supplier_name = None if v.supplier_id: supplier = user_crud.get(db, v.supplier_id) supplier_name = supplier.real_name if supplier else None result.append({ "id": v.id, "versionNo": getattr(v, "version_no", None) or getattr(v, "version", None), "regionId": v.region_id, "regionName": region_name, "supplierId": v.supplier_id, "supplierName": supplier_name, "status": v.status, "effectiveTime": str(v.effective_time) if v.effective_time else None, "remark": v.remark, "createAt": str(v.create_at) if v.create_at else None }) return success_response({ "list": result, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize }) @router.get("/price-versions/{version_id}", response_model=dict) def get_price_version_detail( version_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """获取价格版本详情（含供应商报价与最终售价拆解，便于订单侧只读查看）。""" from app.models.models import BizPriceList, BizPriceListItem, BizQuotationItem from app.crud.inquiry import inquiry_crud version = db.query(BizPriceList).filter(BizPriceList.id == version_id, BizPriceList.is_del == 0).first() if not version: return error_response(404, "价格清单不存在") region_name = None if version.region_id: region = region_crud.get(db, version.region_id) region_name = region.name if region else None supplier_name = None if version.supplier_id: supplier = user_crud.get(db, version.supplier_id) supplier_name = supplier.real_name if supplier else None inquiry_title = None if version.inquiry_id: iq = inquiry_crud.get(db, version.inquiry_id) inquiry_title = iq.title if iq else None # 来源报价单：用于还原供应商侧单价 q_item_by_pid: dict[str, object] = {} if version.quotation_id: q_rows = db.query(BizQuotationItem).filter( and_( BizQuotationItem.quotation_id == version.quotation_id, BizQuotationItem.is_del == 0 ) ).all() for qi in q_rows: pid = str(getattr(qi, "product_id", "") or "") if pid and pid not in q_item_by_pid: q_item_by_pid[pid] = qi # 获取版本明细 items = db.query(BizPriceListItem).filter( BizPriceListItem.price_list_id == version_id, BizPriceListItem.is_del == 0 ).all() item_details = [] for item in items: product = product_crud.get(db, item.product_id) unit_name = None if item.unit_id: u = unit_crud.get(db, item.unit_id) unit_name = u.name if u else None pid = str(item.product_id or "") q_item = q_item_by_pid.get(pid) supplier_price = float(getattr(q_item, "price", 0) or 0) if q_item else 0.0 coeff = None if getattr(item, "pricing_coefficient", None) is not None: coeff = float(item.pricing_coefficient) elif product and getattr(product, "price_coefficient", None) is not None: coeff = float(product.price_coefficient) else: coeff = 0.0 final_price = float(item.price) if item.price else 0.0 category_id = getattr(product, "category_id", None) if product else None category_path = _category_path_display(db, str(category_id)) if category_id else "" item_details.append({ "id": item.id, "productId": item.product_id, "productName": product.name if product else None, "imageUrl": getattr(product, "image_url", None) if product else None, "categoryId": str(category_id) if category_id else None, "categoryName": category_path or None, "measureValue": _numeric_to_float_optional(getattr(product, "measure_value", None)) if product else None, "unitId": item.unit_id, "unitName": unit_name, "supplierPrice": supplier_price, "pricingCoefficient": coeff, "finalPrice": final_price, "price": final_price }) return success_response({ "id": version.id, "versionNo": getattr(version, "version_no", None) or getattr(version, "version", None), "version": getattr(version, "version", None), "regionId": version.region_id, "regionName": region_name, "supplierId": version.supplier_id, "supplierName": supplier_name, "inquiryId": version.inquiry_id, "inquiryTitle": inquiry_title, "quotationId": version.quotation_id, "status": version.status, "effectiveTime": str(version.effective_time) if version.effective_time else None, "remark": version.remark, "createAt": str(version.create_at) if version.create_at else None, "items": item_details }) @router.post("/price-versions", response_model=dict) def create_price_version( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """创建价格版本""" from app.models.models import BizPriceList from app.core.utils import generate_uuid region_id = request.get("regionId") supplier_id = request.get("supplierId") effective_time = request.get("effectiveTime") remark = request.get("remark") if not region_id or not supplier_id: return error_response(400, "区域ID和供应商ID不能为空") # 生成版本号 today = datetime.now().strftime("%Y%m%d") count = db.query(BizPriceList).filter( BizPriceList.region_id == region_id, BizPriceList.supplier_id == supplier_id ).count() version_no = f"PL{today}{count + 1:04d}" # 解析生效时间 effective_dt = None if effective_time: try: effective_dt = datetime.strptime(effective_time, "%Y-%m-%d %H:%M:%S") except: try: effective_dt = datetime.strptime(effective_time, "%Y-%m-%d") except: pass new_version = BizPriceList( id=generate_uuid(), region_id=region_id, supplier_id=supplier_id, version_no=version_no, status="draft", effective_time=effective_dt, remark=remark, create_at=datetime.now(), create_by=current_user.get("user_id"), create_name=current_user.get("real_name"), is_del=0 ) db.add(new_version) db.commit() return success_response({ "id": new_version.id, "versionNo": version_no, "message": "价格版本创建成功" }) @router.post("/price-versions/{version_id}/publish", response_model=dict) def publish_price_version( version_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """发布价格版本""" from app.models.models import BizPriceList version = db.query(BizPriceList).filter(BizPriceList.id == version_id, BizPriceList.is_del == 0).first() if not version: return error_response(404, "价格版本不存在") if version.status != "draft": return error_response(400, "只有草稿状态的价格版本才能发布") version.status = "published" version.publish_time = datetime.now() version.publisher_id = current_user.get("user_id") db.commit() return success_response({ "id": version.id, "status": version.status, "message": "价格版本发布成功" }) @router.post("/price-versions/{version_id}/disable", response_model=dict) def disable_price_version( version_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """禁用价格版本""" from app.models.models import BizPriceList version = db.query(BizPriceList).filter(BizPriceList.id == version_id, BizPriceList.is_del == 0).first() if not version: return error_response(404, "价格版本不存在") if version.status == "invalid": return error_response(400, "该价格版本已经失效") version.status = "invalid" version.invalidate_time = datetime.now() version.invalidate_operator_id = current_user.get("user_id") db.commit() return success_response({ "id": version.id, "status": version.status, "message": "价格版本已禁用" }) @router.get("/price-versions/compare", response_model=dict) def compare_price_versions( versionIds: str = Query(..., description="版本ID列表，逗号分隔"), db: Session = Depends(get_db), current_user: dict = Depends(require_admin) ): """版本价格对比""" from app.models.models import BizPriceList, BizPriceListItem version_id_list = [v.strip() for v in versionIds.split(",") if v.strip()] if len(version_id_list) < 2: return error_response(400, "至少需要选择两个版本进行对比") versions = db.query(BizPriceList).filter( BizPriceList.id.in_(version_id_list), BizPriceList.is_del == 0 ).all() if len(versions) < 2: return error_response(400, "选择的版本中有不存在的") # 获取所有版本的价格明细 version_items = {} all_products = set() for version in versions: items = db.query(BizPriceListItem).filter( BizPriceListItem.price_list_id == version.id, BizPriceListItem.is_del == 0 ).all() version_items[version.id] = {item.product_id: float(item.price) if item.price else 0 for item in items} all_products.update([item.product_id for item in items]) # 构建对比数据 compare_data = [] for product_id in all_products: product = product_crud.get(db, product_id) if not product: continue row = { "productId": product_id, "productName": product.name, "prices": {} } for version in versions: row["prices"][version.id] = version_items.get(version.id, {}).get(product_id, None) if row["prices"][version.id] is not None: if not hasattr(row, "lowestPrice") or row["lowestPrice"] > row["prices"][version.id]: row["lowestPrice"] = row["prices"][version.id] row["lowestVersionId"] = version.id compare_data.append(row) # 版本基本信息 version_info = [{ "id": v.id, "versionNo": v.version_no, "status": v.status } for v in versions] return success_response({ "versions": version_info, "products": compare_data })