如何开发ERP（离散制造-MTO）系统中的采购管理板块（附架构图+流程图+代码参考）

+---------------------------+       +---------------------+
|   前端应用（React/PC）    | <-->  | API Gateway / BFF   |
+---------------------------+       +---------------------+
             |                                 |
             v                                 v
+--------------------------------------------------------------+
|                 采购管理微服务 / 模块 (Procurement)          |
|  - REST API                                                   |
|  - 审批引擎（可插拔）                                         |
|  - 统计与报表服务                                             |
|  - 看板（WebSocket / Push）                                   |
+--------------------------------------------------------------+
     |        |           |              |               |
     v        v           v              v               v
+--------+ +--------+ +--------+  +--------------+  +-------------+
| DB (订单) | | DB(库存)| | SRM/供应商| | MQ (异步流程)|  | Search/BI |
+--------+ +--------+ +--------+  +--------------+  +-------------+
     |           |              |              |
     v           v              v              v
  MES/WMS   财务系统        供应商门户      第三方物流/邮件/ERP

[需求触发] --> [创建 PR] --> [PR 审批?]
                               |yes
                               v
                         [合并/生成 PO] --> [下发供应商] --> [供应商发货]
                                                           |
                                                           v
                                                     [到货/验收] --合格--> [入库更新库存/财务]
                                                           |
                                                          不合格
                                                           v
                                                      [发起退货/索赔] --> [退货/补发/结算]

# procurement_app.py
from fastapi import FastAPI, HTTPException, WebSocket, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime, date
from sqlalchemy import (
    create_engine, Column, Integer, String, DateTime, ForeignKey,
    Float, Enum, Boolean, Date
)
from sqlalchemy.orm import sessionmaker, declarative_base, relationship, Session
from sqlalchemy.exc import IntegrityError
import enum
import uuid
# DB setup (SQLite for demo)
SQLALCHEMY_DATABASE_URL = "sqlite:///./procurement_demo.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()
# Enums
class PRStatus(str, enum.Enum):
    DRAFT = "DRAFT"
    SUBMITTED = "SUBMITTED"
    APPROVED = "APPROVED"
    REJECTED = "REJECTED"
class POStatus(str, enum.Enum):
    DRAFT = "DRAFT"
    SENT = "SENT"
    PART_RECEIVED = "PART_RECEIVED"
    RECEIVED = "RECEIVED"
    CANCELLED = "CANCELLED"
# Models
class Supplier(Base):
    __tablename__ = "suppliers"
    id = Column(Integer, primary_key=True, index=True)
    code = Column(String, unique=True, index=True)
    name = Column(String, index=True)
    contact = Column(String, nullable=True)
    lead_time_mean = Column(Integer, default=7)  # days
    rating = Column(Float, default=5.0)
class Material(Base):
    __tablename__ = "materials"
    id = Column(Integer, primary_key=True, index=True)
    sku = Column(String, unique=True)
    description = Column(String)
class PurchaseRequest(Base):
    __tablename__ = "purchase_requests"
    id = Column(Integer, primary_key=True, index=True)
    pr_no = Column(String, unique=True, default=lambda: "PR-" + uuid.uuid4().hex[:8])
    requester = Column(String)
    created_at = Column(DateTime, default=datetime.utcnow)
    status = Column(Enum(PRStatus), default=PRStatus.DRAFT)
    lines = relationship("PurchaseRequestLine", back_populates="pr")
class PurchaseRequestLine(Base):
    __tablename__ = "pr_lines"
    id = Column(Integer, primary_key=True, index=True)
    pr_id = Column(Integer, ForeignKey("purchase_requests.id"))
    material_id = Column(Integer, ForeignKey("materials.id"))
    qty = Column(Float)
    required_date = Column(Date)
    priority = Column(Integer, default=3)
    pr = relationship("PurchaseRequest", back_populates="lines")
    material = relationship("Material")
class PurchaseOrder(Base):
    __tablename__ = "purchase_orders"
    id = Column(Integer, primary_key=True, index=True)
    po_no = Column(String, unique=True, default=lambda: "PO-" + uuid.uuid4().hex[:8])
    supplier_id = Column(Integer, ForeignKey("suppliers.id"))
    created_by = Column(String)
    status = Column(Enum(POStatus), default=POStatus.DRAFT)
    created_at = Column(DateTime, default=datetime.utcnow)
    lines = relationship("PurchaseOrderLine", back_populates="po")
    supplier = relationship("Supplier")
class PurchaseOrderLine(Base):
    __tablename__ = "po_lines"
    id = Column(Integer, primary_key=True, index=True)
    po_id = Column(Integer, ForeignKey("purchase_orders.id"))
    material_id = Column(Integer, ForeignKey("materials.id"))
    qty_ordered = Column(Float)
    qty_received = Column(Float, default=0.0)
    unit_price = Column(Float, default=0.0)
    promised_date = Column(Date)
    po = relationship("PurchaseOrder", back_populates="lines")
    material = relationship("Material")
class Receipt(Base):
    __tablename__ = "receipts"
    id = Column(Integer, primary_key=True, index=True)
    receipt_no = Column(String, unique=True, default=lambda: "RC-" + uuid.uuid4().hex[:8])
    po_id = Column(Integer, ForeignKey("purchase_orders.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    warehouse = Column(String, default="MAIN")
    lines = Column(String)  # for demo: JSON string of received lines
Base.metadata.create_all(bind=engine)
# Pydantic schemas
class PRLineIn(BaseModel):
    material_sku: str
    qty: float
    required_date: date
    priority: int = 3
class PRIn(BaseModel):
    requester: str
    lines: List[PRLineIn]
class POCreateIn(BaseModel):
    supplier_code: str
    created_by: str
    items: List[dict]  # each: {material_sku, qty, unit_price, promised_date}
# FastAPI app
app = FastAPI(title="Procurement Demo API (MTO)")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
# Helper: get or create material
def get_or_create_material(db: Session, sku: str, description: Optional[str] = None):
    m = db.query(Material).filter(Material.sku == sku).first()
    if m:
        return m
    m = Material(sku=sku, description=description or sku)
    db.add(m)
    db.commit()
    db.refresh(m)
    return m
# API: create PR
@app.post("/pr", summary="创建采购申请 PR")
def create_pr(pr_in: PRIn, db: Session = Depends(get_db)):
    pr = PurchaseRequest(requester=pr_in.requester)
    db.add(pr)
    db.commit()
    db.refresh(pr)
    for l in pr_in.lines:
        mat = get_or_create_material(db, l.material_sku)
        pr_line = PurchaseRequestLine(pr_id=pr.id, material_id=mat.id, qty=l.qty,
                                      required_date=l.required_date, priority=l.priority)
        db.add(pr_line)
    db.commit()
    db.refresh(pr)
    return {"pr_no": pr.pr_no, "id": pr.id, "status": pr.status}
# API: submit PR (simple approval simulation)
@app.post("/pr/{pr_id}/submit", summary="提交 PR 审批")
def submit_pr(pr_id: int, db: Session = Depends(get_db)):
    pr = db.query(PurchaseRequest).filter(PurchaseRequest.id == pr_id).first()
    if not pr:
        raise HTTPException(status_code=404, detail="PR not found")
    pr.status = PRStatus.SUBMITTED
    db.commit()
    return {"pr_no": pr.pr_no, "status": pr.status}
# API: approve PR and create PO (very simplified logic: one PO per PR, using supplier heuristic)
@app.post("/pr/{pr_id}/approve", summary="审批 PR 并生成 PO（示例）")
def approve_pr_and_create_po(pr_id: int, supplier_code: str, db: Session = Depends(get_db)):
    pr = db.query(PurchaseRequest).filter(PurchaseRequest.id == pr_id).first()
    if not pr:
        raise HTTPException(status_code=404, detail="PR not found")
    supplier = db.query(Supplier).filter(Supplier.code == supplier_code).first()
    if not supplier:
        raise HTTPException(status_code=404, detail="Supplier not found")
    # mark PR approved
    pr.status = PRStatus.APPROVED
    db.commit()
    # create PO
    po = PurchaseOrder(supplier_id=supplier.id, created_by="system-batch", status=POStatus.SENT)
    db.add(po); db.commit(); db.refresh(po)
    for line in pr.lines:
        pol = PurchaseOrderLine(po_id=po.id, material_id=line.material_id,
                                qty_ordered=line.qty, qty_received=0.0,
                                unit_price=0.0, promised_date=line.required_date)
        db.add(pol)
    db.commit()
    return {"po_no": po.po_no, "po_id": po.id}
# API: manually create PO
@app.post("/po", summary="创建 PO（手工）")
def create_po(payload: POCreateIn, db: Session = Depends(get_db)):
    supplier = db.query(Supplier).filter(Supplier.code == payload.supplier_code).first()
    if not supplier:
        raise HTTPException(status_code=404, detail="Supplier not found")
    po = PurchaseOrder(supplier_id=supplier.id, created_by=payload.created_by, status=POStatus.SENT)
    db.add(po); db.commit(); db.refresh(po)
    for it in payload.items:
        mat = get_or_create_material(db, it["material_sku"])
        pol = PurchaseOrderLine(po_id=po.id, material_id=mat.id, qty_ordered=it["qty"],
                                unit_price=it.get("unit_price", 0.0), promised_date=it.get("promised_date"))
        db.add(pol)
    db.commit()
    return {"po_no": po.po_no, "po_id": po.id}
# API: receive goods (partial allowed)
@app.post("/po/{po_id}/receive", summary="入库/收货（简化）")
def receive_po(po_id: int, lines: List[dict], warehouse: str = "MAIN", db: Session = Depends(get_db)):
    po = db.query(PurchaseOrder).filter(PurchaseOrder.id == po_id).first()
    if not po:
        raise HTTPException(status_code=404, detail="PO not found")
    # update quantities
    for l in lines:
        pol = db.query(PurchaseOrderLine).filter(PurchaseOrderLine.po_id == po_id,
                                                 PurchaseOrderLine.material_id == db.query(Material).filter(Material.sku == l["material_sku"]).first().id).first()
        if not pol:
            raise HTTPException(status_code=400, detail=f"PO line for {l['material_sku']} not found")
        pol.qty_received = (pol.qty_received or 0) + l["qty"]
    # set status
    all_received = all([ln.qty_received >= ln.qty_ordered for ln in po.lines])
    po.status = POStatus.RECEIVED if all_received else POStatus.PART_RECEIVED
    db.add(po)
    # create receipt (simple)
    r = Receipt(po_id=po.id, warehouse=warehouse, lines=str(lines))
    db.add(r)
    db.commit()
    return {"po_no": po.po_no, "status": po.status}
# API: create supplier (demo)
@app.post("/supplier", summary="新增供应商")
def create_supplier(code: str, name: str, db: Session = Depends(get_db)):
    s = Supplier(code=code, name=name)
    db.add(s)
    try:
        db.commit()
    except IntegrityError:
        db.rollback(); raise HTTPException(status_code=400, detail="Supplier code already exists")
    db.refresh(s)
    return {"supplier_id": s.id, "code": s.code}
# API: stats: get pending/overdue POs
@app.get("/stats/po_summary", summary="采购统计")
def po_summary(db: Session = Depends(get_db)):
    total_po = db.query(PurchaseOrder).count()
    pending_po = db.query(PurchaseOrder).filter(PurchaseOrder.status != POStatus.RECEIVED).count()
    # overdue: promised_date < today and qty_received < qty_ordered
    from sqlalchemy import func
    sub = db.query(PurchaseOrderLine).filter(PurchaseOrderLine.qty_received < PurchaseOrderLine.qty_ordered,
                                             PurchaseOrderLine.promised_date < date.today()).count()
    return {"total_po": total_po, "pending_po": pending_po, "overdue_lines": sub}
# WebSocket for simple tracking board (push minimal updates)
connected = []
@app.websocket("/ws/board")
async def ws_board(ws: WebSocket):
    await ws.accept()
    connected.append(ws)
    try:
        while True:
            # ping/pong or wait for client messages; in demo simply echo
            data = await ws.receive_text()
            await ws.send_text(f"echo: {data}")
    except Exception:
        connected.remove(ws)