4. 数据模型设计：合同表、发票表、付款计划表的 ER 图与 SQLAlchemy 模型

数据模型是工程合同发票匹配与付款风险控制系统的核心骨架。本章将详细定义三张核心业务表——合同表（contracts）、发票表（invoices）、付款计划表（payment_schedules）——的实体关系（ER）及其对应的 SQLAlchemy ORM 模型。清晰的数据模型是后续接口调用与风险规则引擎的基础。

4.1 核心实体关系概述

三张表构成典型的“一对多”与“主从”关系：

• 合同表作为主实体，记录工程合同的核心信息。
• 付款计划表依附于合同，定义合同约定的分期付款里程碑。
• 发票表记录供应商/分包商提交的发票，通过外键关联到合同，并可选关联到具体的付款计划项。

4.2 字段设计明细

以下表格详细列出三张表的字段名称、类型、约束与业务说明。

4.2.1 合同表（contracts）

4.2.2 付款计划表（payment_schedules）

4.2.3 发票表（invoices）

4.3 SQLAlchemy ORM 模型实现

以下代码使用 SQLAlchemy 2.0 风格（Declarative Base + Mapped 注解）实现三张表的模型定义。代码中包含了关系定义与业务约束。

from datetime import date, datetime
from decimal import Decimal
from typing import Optional, List
from sqlalchemy import (
    String, Integer, Numeric, Date, DateTime,
    ForeignKey, UniqueConstraint, Text
)
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship
)

class Base(DeclarativeBase):
    pass

class Contract(Base):
    __tablename__ = "contracts"

    contract_id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    contract_code: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    contract_name: Mapped[str] = mapped_column(String(200), nullable=False)
    party_a: Mapped[str] = mapped_column(String(100), nullable=False)
    party_b: Mapped[str] = mapped_column(String(100), nullable=False)
    total_amount: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False)
    signed_date: Mapped[date] = mapped_column(Date, nullable=False)
    status: Mapped[str] = mapped_column(String(20), default="active")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # 关系定义
    payment_schedules: Mapped[List["PaymentSchedule"]] = relationship(
        back_populates="contract", cascade="all, delete-orphan"
    )
    invoices: Mapped[List["Invoice"]] = relationship(
        back_populates="contract", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"<Contract(id={self.contract_id}, code={self.contract_code})>"

class PaymentSchedule(Base):
    __tablename__ = "payment_schedules"

    schedule_id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    contract_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("contracts.contract_id"), nullable=False
    )
    milestone_name: Mapped[str] = mapped_column(String(100), nullable=False)
    planned_amount: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False)
    planned_date: Mapped[date] = mapped_column(Date, nullable=False)
    payment_ratio: Mapped[Decimal] = mapped_column(Numeric(5, 2), default=Decimal("0.00"))
    status: Mapped[str] = mapped_column(String(20), default="pending")

    # 关系定义
    contract: Mapped["Contract"] = relationship(back_populates="payment_schedules")
    invoices: Mapped[List["Invoice"]] = relationship(back_populates="payment_schedule")

    def __repr__(self) -> str:
        return f"<PaymentSchedule(id={self.schedule_id}, milestone={self.milestone_name})>"

class Invoice(Base):
    __tablename__ = "invoices"

    invoice_id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    contract_id: Mapped[int] = mapped_column(
        Integer, ForeignKey("contracts.contract_id"), nullable=False
    )
    schedule_id: Mapped[Optional[int]] = mapped_column(
        Integer, ForeignKey("payment_schedules.schedule_id"), nullable=True
    )
    invoice_code: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    invoice_amount: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False)
    tax_amount: Mapped[Decimal] = mapped_column(Numeric(18, 2), default=Decimal("0.00"))
    invoice_date: Mapped[date] = mapped_column(Date, nullable=False)
    verify_status: Mapped[str] = mapped_column(String(20), default="unverified")
    match_status: Mapped[str] = mapped_column(String(20), default="unmatched")

    # 关系定义
    contract: Mapped["Contract"] = relationship(back_populates="invoices")
    payment_schedule: Mapped[Optional["PaymentSchedule"]] = relationship(
        back_populates="invoices"
    )

    def __repr__(self) -> str:
        return f"<Invoice(id={self.invoice_id}, code={self.invoice_code})>"

4.4 模型关键设计决策

• 级联删除（cascade="all, delete-orphan"）： 当合同被删除时，其关联的付款计划和发票自动删除，保证数据一致性。
• 可选外键（schedule_id nullable=True）： 发票可以独立于付款计划存在，适应工程结算中“先开票后匹配”或“无计划开票”的灵活场景。
• 状态字段枚举： 使用字符串字段而非布尔值，便于扩展（如 verify_status 可扩展为 verified / mismatch / expired）。
• 金额字段使用 Numeric(18,2)： 避免浮点数精度问题，满足财务计算要求。

4.5 数据模型与匹配流程的映射

下图展示了数据模型如何支撑“发票-付款计划-合同”的匹配与风险控制流程。

4.6 模型验证与迁移

定义好模型后，可通过 Alembic 生成迁移脚本，或直接使用 Base.metadata.create_all() 在开发环境快速建表。生产环境建议使用 Alembic 管理版本迁移。

# 开发环境快速建表示例
from sqlalchemy import create_engine
from your_models import Base

engine = create_engine("postgresql://user:pass@localhost:5432/engineering_db")
Base.metadata.create_all(engine)
print("所有表已创建成功。")