"""小程序客户端扩展接口：购物车、商品分类、仪表盘、发票、下单、支付""" from typing import Optional, List, Tuple, Dict from decimal import Decimal from fastapi import APIRouter, Depends, Query from sqlalchemy.orm import Session from sqlalchemy import and_ from app.core.database import get_db from app.core.auth import get_current_user from app.core.response import success_response, error_response from app.core.config import settings from app.core.utils import generate_id from datetime import datetime import random from app.models.models import BizInvoice, BizWechatPay from app.crud.user import user_crud from app.crud.address import address_crud from app.crud.client_cart import client_cart_crud from app.crud.price_list import price_list_crud, price_list_item_crud from app.crud.product import product_crud from app.models.models import BasProduct, BasProductCategory from app.crud.order import order_crud from app.crud.category import category_crud from app.crud.unit import unit_crud from app.crud.product_banner import product_banner from app.crud.product_detail_media import product_detail_media from app.crud.product_parameter_value import product_parameter_value from app.api.mini_client import require_client router = APIRouter(prefix="/api/mini/client", tags=["小程序端-客户端扩展"]) def _upsert_wechat_pay_pending( db: Session, *, order_id: str, out_trade_no: str, total_fee: int, operator_id: Optional[str], operator_name: Optional[str], ) -> BizWechatPay: """创建支付单时落一条待支付微信流水（按 out_trade_no 幂等）。""" row = ( db.query(BizWechatPay) .filter(BizWechatPay.out_trade_no == out_trade_no, BizWechatPay.is_del == 0) .first() ) if row: row.order_id = order_id row.total_fee = total_fee row.trade_state = row.trade_state or "NOTPAY" row.trade_state_desc = row.trade_state_desc or "统一下单成功，待支付" row.update_at = datetime.now() row.update_by = operator_id return row row = BizWechatPay( id=generate_id(), order_id=order_id, out_trade_no=out_trade_no, total_fee=total_fee, trade_state="NOTPAY", trade_state_desc="统一下单成功，待支付", create_by=operator_id, create_name=operator_name, update_by=operator_id, ) db.add(row) return row def _cart_item_to_dict( row, product: Optional[BasProduct] = None, root_category: Optional[BasProductCategory] = None, ) -> dict: return { "id": row.id, "productId": row.product_id, "productName": row.product_name or "", "categoryId": product.category_id if product and product.category_id else None, "firstCategoryId": root_category.id if root_category else None, "firstCategoryName": (root_category.name if root_category else None) or "", "categoryName": row.category_name or "", "unitName": row.unit_name or "", "measureValue": float(product.measure_value) if (product and product.measure_value is not None) else None, "price": float(row.unit_price) if row.unit_price is not None else 0, "quantity": float(row.quantity) if row.quantity is not None else 0, "amount": float(row.amount) if row.amount is not None else 0, "imageUrl": row.image_url, } def _root_and_leaf_category( db: Session, category_id: Optional[str], cache: Dict[str, Tuple[Optional[BasProductCategory], Optional[BasProductCategory]]] ) -> Tuple[Optional[BasProductCategory], Optional[BasProductCategory]]: """从叶子分类 id 向上解析（含缓存）：返回 (根分类, 叶子分类)。""" if not category_id: return None, None if category_id in cache: return cache[category_id] leaf = category_crud.get(db, category_id) if not leaf or getattr(leaf, "is_del", 0): cache[category_id] = (None, None) return None, None cur = leaf guard = 0 while cur.parent_id and guard < 10: parent = category_crud.get(db, cur.parent_id) if not parent or getattr(parent, "is_del", 0): break cur = parent guard += 1 root = cur cache[category_id] = (root, leaf) return root, leaf def _cart_resolve_root_category(db: Session, product: Optional[BasProduct]) -> Optional[BasProductCategory]: """商品对应的一级（根）分类，供购物车行展示与序列化。""" if not product or not product.category_id: return None cache: Dict[str, Tuple[Optional[BasProductCategory], Optional[BasProductCategory]]] = {} root, _ = _root_and_leaf_category(db, product.category_id, cache) return root def _cart_row_sort_key( db: Session, row, pmap: dict, cat_cache: Dict[str, Tuple[Optional[BasProductCategory], Optional[BasProductCategory]]], ) -> tuple: """购物车：先一级分类 sort_order，再商品名、商品 id（稳定）。""" product = pmap.get(row.product_id) pname = (row.product_name or (product.name if product else "") or "").lower() pid = row.product_id or "" if not product or not product.category_id: return (999999, pname, pid) root, _leaf = _root_and_leaf_category(db, product.category_id, cat_cache) fo = int(root.sort_order) if root else 999999 return (fo, pname, pid) def _cart_rows_after_sync(db: Session, uid: str, region_id: Optional[str], op: dict) -> List: """当前用户有效购物车行（含区域价目表同步、过滤已删）。""" rows = client_cart_crud.list_by_client(db, uid) if region_id: client_cart_crud.sync_to_active_price_list(db, uid, region_id, op, rows=rows) rows = [r for r in rows if getattr(r, "is_del", 0) == 0] return rows def _cart_list_dicts(db: Session, rows: List) -> List[dict]: """将购物车 ORM 行序列化为与 GET /cart 一致的 list 项（仅未删除行）。""" rows = [r for r in rows if getattr(r, "is_del", 0) == 0] product_ids = list({r.product_id for r in rows if r and getattr(r, "product_id", None)}) products = ( db.query(BasProduct) .filter(and_(BasProduct.id.in_(product_ids), BasProduct.is_del == 0)) .all() if product_ids else [] ) pmap = {p.id: p for p in products} cat_cache: Dict[str, Tuple[Optional[BasProductCategory], Optional[BasProductCategory]]] = {} rows_sorted = sorted(rows, key=lambda r: _cart_row_sort_key(db, r, pmap, cat_cache)) out: List[dict] = [] for r in rows_sorted: p = pmap.get(r.product_id) root: Optional[BasProductCategory] = None if p and p.category_id: root, _ = _root_and_leaf_category(db, p.category_id, cat_cache) out.append(_cart_item_to_dict(r, p, root)) return out @router.get("/cart/summary", response_model=dict) def cart_summary( db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): """购物车汇总（件数、行数、金额），供 Tab 角标等轻量刷新。""" uid = current_user.get("user_id") region_id = current_user.get("region_id") op = {"user_id": uid, "real_name": current_user.get("real_name")} rows = _cart_rows_after_sync(db, uid, region_id, op) tq = sum(float(r.quantity or 0) for r in rows) ta = sum(float(r.amount or 0) for r in rows) return success_response({ "lineCount": len(rows), "totalQuantity": tq, "totalAmount": ta, }) @router.get("/cart", response_model=dict) def list_cart( db: Session = Depends(get_db), current_user: dict = Depends(require_client), page: Optional[int] = Query(None, ge=1, description="传则分页；不传则返回全量 list（兼容确认订单等）"), pageSize: Optional[int] = Query(None, ge=1, le=100, description="分页时每页条数，默认 20"), ): uid = current_user.get("user_id") region_id = current_user.get("region_id") op = {"user_id": uid, "real_name": current_user.get("real_name")} rows = _cart_rows_after_sync(db, uid, region_id, op) full_list = _cart_list_dicts(db, rows) if page is None: return success_response({"list": full_list}) ps = pageSize if pageSize is not None else 20 total = len(full_list) start = (page - 1) * ps chunk = full_list[start : start + ps] total_pages = (total + ps - 1) // ps if total else 0 return success_response({ "list": chunk, "total": total, "page": page, "pageSize": ps, "totalPages": total_pages, }) @router.post("/cart", response_model=dict) def add_cart( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): product_id = request.get("productId") quantity = Decimal(str(request.get("quantity", 1))) if not product_id or quantity <= 0: return error_response(400, "商品与数量无效") region_id = current_user.get("region_id") if not region_id: return error_response(400, "用户未绑定区域") pl = price_list_crud.get_active(db, region_id) if not pl: return error_response(404, "当前区域无生效价格清单") # 只查询该商品对应的价目表行，避免拉取整张价目表明细导致加购变慢 pi = price_list_item_crud.get_by_product_and_price_list(db, product_id, pl.id) if not pi: return error_response(400, "商品不在当前价格清单中") # 用一次 JOIN 拉齐商品 + 单位 + 分类，避免 3 次单查 from app.models.models import BasProductUnit, BasProductCategory q = ( db.query(BasProduct, BasProductUnit.name, BasProductCategory.name) .select_from(BasProduct) # 明确从 BasProduct 出发，避免 join 路径歧义 .outerjoin(BasProductUnit, BasProductUnit.id == pi.unit_id) .outerjoin(BasProductCategory, BasProductCategory.id == BasProduct.category_id) .filter(and_(BasProduct.id == product_id, BasProduct.is_del == 0)) ) r = q.first() product = r[0] if r else None unit_name = r[1] if r else "" cat_name = r[2] if r else "" op = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} row = client_cart_crud.add_or_update( db, client_id=current_user.get("user_id"), price_list_id=pl.id, product_id=product_id, unit_id=pi.unit_id, quantity=quantity, unit_price=Decimal(str(pi.price)), product_name=product.name if product else "", unit_name=unit_name or "", category_name=cat_name, image_url=product.image_url if product else "", operator=op, ) root_cat = _cart_resolve_root_category(db, product) return success_response(_cart_item_to_dict(row, product, root_cat)) @router.put("/cart/{cart_id}", response_model=dict) def put_cart( cart_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): qty = Decimal(str(request.get("quantity", 0))) if qty <= 0: return error_response(400, "数量无效") uid = current_user.get("user_id") region_id = current_user.get("region_id") op = {"user_id": uid, "real_name": current_user.get("real_name")} # 先拿到该行，只有当其 price_list_id 不是当前生效清单时才触发“全车同步” cur = client_cart_crud.get(db, cart_id) if not cur or cur.client_id != uid: return error_response(404, "购物车项不存在") need_full_cart = False if region_id: pl = price_list_crud.get_active(db, region_id) if pl and getattr(cur, "price_list_id", None) != pl.id: client_cart_crud.sync_to_active_price_list(db, uid, region_id, op) # 同步可能合并/软删多行，前端需拉全量列表 need_full_cart = True row = client_cart_crud.update_quantity(db, cart_id, uid, qty, op, current_row=cur) if not row: return error_response(404, "购物车项不存在") # 行更新接口优先返回轻量字段，避免每次点加减都额外查询商品/分类。 line = _cart_item_to_dict(row, None, None) if need_full_cart: # 与 GET /cart 同结构，避免前端再发一次全量请求 rows_after = _cart_rows_after_sync(db, uid, region_id, op) return success_response({"list": _cart_list_dicts(db, rows_after), "item": line}) return success_response(line) @router.delete("/cart", response_model=dict) def delete_cart_clear( db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): op = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} n = client_cart_crud.clear(db, current_user.get("user_id"), op) return success_response({"cleared": n}) @router.delete("/cart/batch", response_model=dict) def delete_cart_batch( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): ids = request.get("cartIds") or [] if not isinstance(ids, list): return error_response(400, "cartIds 格式错误") op = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} n = client_cart_crud.batch_delete(db, current_user.get("user_id"), ids, op) return success_response({"removed": n}) @router.delete("/cart/{cart_id}", response_model=dict) def delete_cart_one( cart_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): op = {"user_id": current_user.get("user_id"), "real_name": current_user.get("real_name")} ok = client_cart_crud.soft_delete(db, cart_id, current_user.get("user_id"), op) if not ok: return error_response(404, "购物车项不存在") return success_response({"ok": True}) @router.get("/price-list", response_model=dict) def get_price_list_alias( db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): """与 /price-list/current 相同，便于小程序路径统一""" user = user_crud.get(db, current_user.get("user_id")) if not user or not user.region_id: return error_response(400, "用户未绑定区域") price_list = price_list_crud.get_active(db, user.region_id) if not price_list: return error_response(404, "暂无生效价格清单") items = price_list_item_crud.get_by_price_list(db, price_list.id) item_details = [] for item in items: product = product_crud.get(db, item.product_id) unit = unit_crud.get(db, item.unit_id) if item.unit_id else None item_details.append({ "productId": item.product_id, "productName": product.name if product else None, "unitId": item.unit_id, "unitName": unit.name if unit else None, "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, "version": price_list.version, "items": item_details, }) def _sort_category_dicts_locally(cats: List[dict]) -> List[dict]: def _key(c: dict): return ( int(c.get("sortOrder") or 0), (c.get("name") or "").lower(), c.get("id") or "", ) return sorted(cats, key=_key) def _client_aggregated_categories(db: Session, current_user: dict) -> Tuple[Optional[List[dict]], Optional[str]]: """返回 (分类 dict 列表, 错误信息)。无清单时返回 ([], None)。""" user = user_crud.get(db, current_user.get("user_id")) if not user or not user.region_id: return None, "用户未绑定区域" pl = price_list_crud.get_active(db, user.region_id) if not pl: return [], None items = price_list_item_crud.get_by_price_list(db, pl.id) pmap: Dict[str, dict] = {} for it in items: product = product_crud.get(db, it.product_id) if not product or not product.category_id: continue cid = product.category_id if cid not in pmap: cat = category_crud.get(db, cid) pmap[cid] = { "id": cid, "name": cat.name if cat else "", "parentId": cat.parent_id if cat else None, "sortOrder": cat.sort_order if cat else 0, "isActive": True, "imageUrl": cat.image_url if cat else None, } return list(pmap.values()), None @router.get("/product-categories", response_model=dict) def list_product_categories( db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): """基于当前生效价格清单中的商品，聚合分类（全量，兼容旧调用）""" cats, err = _client_aggregated_categories(db, current_user) if err: return error_response(400, err) return success_response(cats) @router.get("/product-categories/first-level", response_model=dict) def list_product_categories_first_level( db: Session = Depends(get_db), current_user: dict = Depends(require_client), page: int = Query(1, ge=1), pageSize: int = Query(8, ge=1, le=50), ): """首页一级分类（swiper），服务端分页""" cats, err = _client_aggregated_categories(db, current_user) if err: return error_response(400, err) sorted_all = _sort_category_dicts_locally(cats or []) roots = [c for c in sorted_all if not c.get("parentId")] total = len(roots) start = (page - 1) * pageSize slice_roots = roots[start : start + pageSize] out = [ { "id": c["id"], "name": c["name"], "imageUrl": c.get("imageUrl") or "", } for c in slice_roots ] total_pages = (total + pageSize - 1) // pageSize if total else 0 return success_response({ "list": out, "total": total, "page": page, "pageSize": pageSize, "totalPages": total_pages, }) @router.get("/product-categories/sidebar", response_model=dict) def list_product_categories_sidebar( db: Session = Depends(get_db), current_user: dict = Depends(require_client), firstCategoryId: str = Query("all", description="一级分类 id，all 表示顶部选「全部」"), page: int = Query(1, ge=1), pageSize: int = Query(30, ge=1, le=100), ): """首页左侧二级分类栏选项，服务端分页（含「全部」行）""" cats, err = _client_aggregated_categories(db, current_user) if err: return error_response(400, err) sorted_all = _sort_category_dicts_locally(cats or []) roots = [c for c in sorted_all if not c.get("parentId")] rows: List[dict] = [{"id": "all", "name": "全部"}] fid = (firstCategoryId or "all").strip() if fid == "all": rows.extend({"id": c["id"], "name": c["name"]} for c in roots) else: children = [c for c in sorted_all if c.get("parentId") == fid] rows.extend({"id": c["id"], "name": c["name"]} for c in children) total = len(rows) start = (page - 1) * pageSize chunk = rows[start : start + pageSize] total_pages = (total + pageSize - 1) // pageSize if total else 0 return success_response({ "list": chunk, "total": total, "page": page, "pageSize": pageSize, "totalPages": total_pages, }) def _root_category_id_for_leaf(db: Session, category_id: Optional[str], memo: Dict[str, Optional[str]]) -> Optional[str]: """商品分类叶子节点向上解析一级（根）分类 id，与小程序端 getFirstCategoryId 一致。""" if not category_id: return None if category_id in memo: return memo[category_id] cur: Optional[str] = category_id chain: List[str] = [] while cur and len(chain) < 32: chain.append(cur) cat = category_crud.get(db, cur) if not cat: root = chain[-1] for cid in chain: memo[cid] = root return root if not cat.parent_id: root = cat.id for cid in chain: memo[cid] = root return root cur = cat.parent_id root = chain[-1] if chain else None for cid in chain: memo[cid] = root return root @router.get("/products", response_model=dict) def list_products( db: Session = Depends(get_db), current_user: dict = Depends(require_client), page: int = Query(1, ge=1), pageSize: int = Query(20, ge=1, le=100), categoryId: Optional[str] = None, rootCategoryId: Optional[str] = None, keyword: Optional[str] = None, ): """分页查询当前价目表中的商品（与 price-list/products 一致逻辑）""" user = user_crud.get(db, current_user.get("user_id")) if not user or not user.region_id: return error_response(400, "用户未绑定区域") pl = price_list_crud.get_active(db, user.region_id) if not pl: return success_response({"list": [], "total": 0, "page": page, "pageSize": pageSize, "totalPages": 0}) items = price_list_item_crud.get_by_price_list(db, pl.id) root_memo: Dict[str, Optional[str]] = {} rows = [] for item in items: product = product_crud.get(db, item.product_id) if not product: continue if categoryId and product.category_id != categoryId: continue if rootCategoryId: rid = _root_category_id_for_leaf(db, product.category_id, root_memo) if rid != rootCategoryId: continue if keyword and keyword.lower() not in (product.name or "").lower(): continue unit = unit_crud.get(db, item.unit_id) if item.unit_id else None cat_name = "" if product.category_id: cat = category_crud.get(db, product.category_id) cat_name = cat.name if cat else "" rows.append({ "id": product.id, "name": product.name, "categoryId": product.category_id, "categoryName": cat_name, "unitId": item.unit_id, "unitName": unit.name if unit else "", "measureValue": float(product.measure_value) if getattr(product, "measure_value", None) is not None else None, "description": product.description, "imageUrl": product.image_url, "price": float(item.price) if item.price else 0, "isActive": True, "sortOrder": int(getattr(product, "sort_order", None) or 0), }) cat_cache: Dict[str, Optional[BasProductCategory]] = {} def _ensure_category_chain_cached(cid: Optional[str]) -> None: cur = cid for _ in range(32): if not cur: break if cur in cat_cache: c = cat_cache[cur] cur = c.parent_id if c else None continue c = category_crud.get(db, cur) cat_cache[cur] = c cur = c.parent_id if c else None for row in rows: _ensure_category_chain_cached(row.get("categoryId")) def _sort_tuple(r: dict) -> Tuple: cid = r.get("categoryId") first_ord = 0 second_ord = 0 if cid: chain: List[BasProductCategory] = [] cur: Optional[str] = cid for _ in range(32): c = cat_cache.get(cur) if cur else None if not c: break chain.append(c) if not c.parent_id: break cur = c.parent_id if chain: second_ord = int(chain[0].sort_order or 0) first_ord = int(chain[-1].sort_order or 0) name = (r.get("name") or "").lower() sid = r.get("id") or "" return (first_ord, second_ord, int(r.get("sortOrder") or 0), name, sid) rows.sort(key=_sort_tuple) total = len(rows) start = (page - 1) * pageSize paginated = rows[start : start + pageSize] total_pages = (total + pageSize - 1) // pageSize if total else 0 return success_response({ "list": paginated, "total": total, "page": page, "pageSize": pageSize, "totalPages": total_pages, }) @router.get("/products/{product_id}", response_model=dict) def get_client_product_detail( product_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): """客户端商品详情：价目表内商品 + 轮播图 + 参数值 + 富文本 + 详情媒体（图/视频）""" user = user_crud.get(db, current_user.get("user_id")) if not user or not user.region_id: return error_response(400, "用户未绑定区域") pl = price_list_crud.get_active(db, user.region_id) if not pl: return error_response(404, "暂无生效价格清单") items = price_list_item_crud.get_by_price_list(db, pl.id) line = next((it for it in items if it.product_id == product_id), None) if not line: return error_response(404, "商品不存在或不在当前价目表") product = product_crud.get(db, product_id) if not product or getattr(product, "is_del", 0) == 1: return error_response(404, "商品不存在") unit = unit_crud.get(db, line.unit_id) if line.unit_id else None cat_name = "" if product.category_id: cat = category_crud.get(db, product.category_id) cat_name = cat.name if cat else "" banners_raw = product_banner.get_by_product(db, product_id) banners = [{ "mediaType": b.media_type, "fileUrl": b.file_url, "thumbnailUrl": b.thumbnail_url, "sortOrder": b.sort_order, } for b in banners_raw] pv_rows = product_parameter_value.get_by_product_with_detail(db, product_id) parameter_values = [] for row in pv_rows: val = row.get("param_value") if val is None or str(val).strip() == "": continue parameter_values.append({ "parameterId": row.get("parameter_id"), "parameterName": row.get("parameter_name") or "", "paramValue": str(val), }) media_raw = product_detail_media.get_by_product(db, product_id) detail_media = [] for m in media_raw: if m.content_type not in ("image", "video"): continue if not m.file_url: continue detail_media.append({ "id": m.id, "contentType": m.content_type, "fileUrl": m.file_url, "thumbnailUrl": m.thumbnail_url, "description": m.description, "sortOrder": m.sort_order, }) product_payload = { "id": product.id, "name": product.name, "skuCode": product.sku_code, "categoryId": product.category_id, "categoryName": cat_name, "unitId": line.unit_id, "unitName": unit.name if unit else None, "measureValue": float(product.measure_value) if getattr(product, "measure_value", None) is not None else None, "price": float(line.price) if line.price else 0, "imageUrl": product.image_url, "description": product.description or "", "detailContent": product.detail_content or "", "isActive": bool(getattr(product, "is_active", 1)), } return success_response({ "product": product_payload, "banners": banners, "parameterValues": parameter_values, "detailMedia": detail_media, }) @router.get("/products-ro/{product_id}", response_model=dict) def get_client_product_detail_readonly( product_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): """客户端只读商品详情：用于订单历史等场景，不依赖区域价目表。""" product = product_crud.get(db, product_id) if not product or getattr(product, "is_del", 0) == 1: return error_response(404, "商品不存在") unit = unit_crud.get(db, product.unit_id) if getattr(product, "unit_id", None) else None cat_name = "" if getattr(product, "category_id", None): cat = category_crud.get(db, product.category_id) cat_name = cat.name if cat else "" banners_raw = product_banner.get_by_product(db, product_id) banners = [{ "mediaType": b.media_type, "fileUrl": b.file_url, "thumbnailUrl": b.thumbnail_url, "sortOrder": b.sort_order, } for b in banners_raw] pv_rows = product_parameter_value.get_by_product_with_detail(db, product_id) parameter_values = [] for row in pv_rows: val = row.get("param_value") if val is None or str(val).strip() == "": continue parameter_values.append({ "parameterId": row.get("parameter_id"), "parameterName": row.get("parameter_name") or "", "paramValue": str(val), }) media_raw = product_detail_media.get_by_product(db, product_id) detail_media = [] for m in media_raw: if m.content_type not in ("image", "video"): continue if not m.file_url: continue detail_media.append({ "id": m.id, "contentType": m.content_type, "fileUrl": m.file_url, "thumbnailUrl": m.thumbnail_url, "description": m.description, "sortOrder": m.sort_order, }) product_payload = { "id": product.id, "name": product.name, "skuCode": product.sku_code, "categoryId": product.category_id, "categoryName": cat_name, "unitId": getattr(product, "unit_id", None), "unitName": unit.name if unit else None, "measureValue": float(product.measure_value) if getattr(product, "measure_value", None) is not None else None, # 价格由前端在只读订单场景使用历史成交价（fallbackPrice）展示，这里不做价目表绑定 "price": 0, "imageUrl": product.image_url, "description": product.description or "", "detailContent": product.detail_content or "", "isActive": bool(getattr(product, "is_active", 1)), } return success_response({ "product": product_payload, "banners": banners, "parameterValues": parameter_values, "detailMedia": detail_media, }) @router.post("/orders/from-cart", response_model=dict) def create_order_from_cart( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): address_id = request.get("addressId") payment_method = order_crud.normalize_payment_method(request.get("paymentMethod", "online")) remark = request.get("remark") if not address_id: return error_response(400, "请选择配送地址") uid = current_user.get("user_id") user = user_crud.get(db, uid) if not user or not user.region_id: return error_response(400, "用户未绑定区域") creator = {"user_id": uid, "real_name": current_user.get("real_name")} client_cart_crud.sync_to_active_price_list(db, uid, user.region_id, creator) pl = price_list_crud.get_active(db, user.region_id) if not pl: return error_response(404, "当前区域无生效价格清单") lines = client_cart_crud.list_by_client(db, uid) if not lines: return error_response(400, "购物车为空") price_items = price_list_item_crud.get_by_price_list(db, pl.id) pmap = {i.product_id: i for i in price_items} address = address_crud.get(db, address_id) if not address or address.user_id != uid: return error_response(404, "地址不存在") order_items = [] for ln in lines: if ln.product_id not in pmap: return error_response( 400, "购物车中含有已不在当前区域价格清单的商品，请返回购物车刷新后移除无效商品再下单", ) if ln.price_list_id != pl.id: return error_response(400, "购物车价格已更新，请返回购物车刷新后再下单") order_items.append({ "productId": ln.product_id, "unitId": ln.unit_id, "quantity": float(ln.quantity), "unitPrice": float(ln.unit_price), }) price_list_id = pl.id full_address = f"{address.province or ''}{address.city or ''}{address.district or ''}{address.detail_address or ''}" order = order_crud.create_order( db, client_id=current_user.get("user_id"), region_id=user.region_id, price_list_id=price_list_id, price_version=pl.version, items=order_items, address_id=address_id, address_info={ "contactPerson": address.contact_person, "contactPhone": address.contact_phone, "fullAddress": full_address, }, payment_method=payment_method, remark=remark, creator=creator, ) client_cart_crud.clear(db, current_user.get("user_id"), creator) return success_response({ "orderId": order.id, "orderNo": order.order_no, "totalAmount": float(order.total_amount), "status": order.status, "paymentMethod": order.payment_method, }) @router.get("/dashboard/statistics", response_model=dict) def client_dashboard_stats( db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): cid = current_user.get("user_id") today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) orders, _ = order_crud.get_multi_with_filters(db, client_id=cid, skip=0, limit=5000) pending_pay = sum(1 for o in orders if o.status == "pending_payment") delivering = sum(1 for o in orders if getattr(o, "order_status", None) == "pending_confirm") completed = sum(1 for o in orders if o.status == "completed") total_amt = sum(float(o.total_amount or 0) for o in orders if o.create_at and o.create_at >= today) return success_response({ "pendingPayment": pending_pay, "delivering": delivering, "completed": completed, "todayOrderAmount": total_amt, }) @router.get("/invoices", response_model=dict) def list_client_invoices( db: Session = Depends(get_db), current_user: dict = Depends(require_client), page: int = Query(1, ge=1), pageSize: int = Query(10, ge=1, le=100), status: Optional[str] = None, ): q = db.query(BizInvoice).filter( and_(BizInvoice.client_id == current_user.get("user_id"), BizInvoice.is_del == 0) ) if status: q = q.filter(BizInvoice.status == status) total = q.count() rows = q.order_by(BizInvoice.create_at.desc()).offset((page - 1) * pageSize).limit(pageSize).all() lst = [{ "id": r.id, "invoiceNo": r.invoice_no, "orderId": r.order_id, "title": r.title, "invoiceAmount": float(r.invoice_amount) if r.invoice_amount else 0, "status": r.status, "createAt": str(r.create_at) if r.create_at else None, } for r in rows] return success_response({ "list": lst, "total": total, "page": page, "pageSize": pageSize, "totalPages": (total + pageSize - 1) // pageSize if total else 0, }) @router.get("/invoices/summary", response_model=dict) def invoice_summary( db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): rows = db.query(BizInvoice).filter( and_(BizInvoice.client_id == current_user.get("user_id"), BizInvoice.is_del == 0) ).all() issued = [r for r in rows if r.status == "issued"] total_issued = sum(float(r.invoice_amount or 0) for r in issued) return success_response({ "totalInvoicedAmount": total_issued, "invoiceCount": len(rows), }) @router.get("/invoices/available-orders", response_model=dict) def invoices_available_orders( db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): """可申请开票的已完成订单（简化：已完成且尚无发票）""" orders, _ = order_crud.get_multi_with_filters( db, client_id=current_user.get("user_id"), status="completed", skip=0, limit=200 ) out = [] for o in orders: inv = db.query(BizInvoice).filter( and_(BizInvoice.order_id == o.id, BizInvoice.is_del == 0) ).first() if inv: continue out.append({"orderId": o.id, "orderNo": o.order_no, "totalAmount": float(o.total_amount or 0)}) return success_response({"list": out}) @router.get("/invoices/{invoice_id}", response_model=dict) def get_client_invoice( invoice_id: str, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): r = db.query(BizInvoice).filter( and_(BizInvoice.id == invoice_id, BizInvoice.client_id == current_user.get("user_id"), BizInvoice.is_del == 0) ).first() if not r: return error_response(404, "发票不存在") return success_response({ "id": r.id, "invoiceNo": r.invoice_no, "orderId": r.order_id, "title": r.title, "invoiceAmount": float(r.invoice_amount) if r.invoice_amount else 0, "status": r.status, "invoiceUrl": r.invoice_url, "createAt": str(r.create_at) if r.create_at else None, }) @router.post("/invoices", response_model=dict) def apply_invoice( request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): """申请开票（简化）""" order_id = request.get("orderId") title = request.get("title") or "" invoice_type = request.get("invoiceType") or "normal" if not order_id: return error_response(400, "缺少 orderId") order = order_crud.get(db, order_id) if not order or order.client_id != current_user.get("user_id"): return error_response(404, "订单不存在") if order.status != "completed": return error_response(400, "仅已完成订单可申请开票") inv = BizInvoice( id=generate_id(), invoice_no=f"INV{datetime.now().strftime('%Y%m%d%H%M%S')}{random.randint(1000, 9999)}", order_id=order.id, order_no=order.order_no, client_id=order.client_id, client_name=current_user.get("real_name") or "", invoice_type=invoice_type, title=title or "个人", invoice_amount=order.total_amount, tax_rate=Decimal("0"), tax_amount=Decimal("0"), status="pending", create_by=current_user.get("user_id"), create_name=current_user.get("real_name"), ) db.add(inv) db.commit() db.refresh(inv) return success_response({"id": inv.id, "invoiceNo": inv.invoice_no, "status": inv.status}) @router.get("/payment-config", response_model=dict) def client_payment_config(): """小程序支付配置（不含密钥）""" return success_response({ "appId": settings.WXPAY_MINI_APPID, "enabled": bool(settings.WXPAY_MCHID and settings.WXPAY_MINI_APPID), }) @router.post("/orders/{order_id}/wechat-pay", response_model=dict) async def mini_order_wechat_pay( order_id: str, request: dict, db: Session = Depends(get_db), current_user: dict = Depends(require_client), ): """调起微信支付参数（需 openid）""" from app.core.wechat_pay import get_wechat_pay_v2 openid = request.get("openid") if not openid: return error_response(400, "缺少 openid") order = order_crud.get(db, order_id) if not order or order.client_id != current_user.get("user_id"): return error_response(404, "订单不存在") if getattr(order, "payment_method", None) != "online": return error_response(400, "线下支付订单无需微信支付") if order.status != "pending_payment": return error_response(400, "订单状态不允许支付") total_amount = int(float(order.total_amount) * 100) if total_amount <= 0: return error_response(400, "订单金额必须大于0") pay_client = get_wechat_pay_v2() result = pay_client.unified_order( out_trade_no=order.order_no, total_fee=total_amount, body="陇耘鲜达-订单支付", openid=openid, trade_type="JSAPI", ) if result.get("success"): pp = result.get("pay_params") or {} _upsert_wechat_pay_pending( db, order_id=str(order.id), out_trade_no=order.order_no, total_fee=total_amount, operator_id=current_user.get("user_id"), operator_name=current_user.get("real_name"), ) db.commit() return success_response({ "orderId": order.id, "orderNo": order.order_no, "prepayId": result.get("prepay_id"), "appId": pp.get("appId"), "timeStamp": pp.get("timeStamp"), "nonceStr": pp.get("nonceStr"), "package": pp.get("package"), "signType": pp.get("signType"), "paySign": pp.get("paySign"), }) return error_response(500, result.get("error", "创建支付失败"))