zhouchengcheng/ctms
1
1
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

整合合同费用并移除旧财务合同

Browse Source
This commit is contained in:
Cheng Zhou
2026-05-28 10:49:08 +08:00
parent 2c85742040
commit 8da7fc715c
20 changed files with 400 additions and 948 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/alembic/versions/20260527_02_add_contract_fee_basic_fields.py
+51
View File
@@ -0,0 +1,51 @@
"""add contract fee basic fields
Revision ID: 20260527_02
Revises: 20260527_01
Create Date: 2026-05-27 10:15:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "20260527_02"
down_revision: Union[str, None] = "20260527_01"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def _column_exists(inspector: sa.Inspector, table_name: str, column_name: str) -> bool:
return column_name in {column["name"] for column in inspector.get_columns(table_name)}
def upgrade() -> None:
bind = op.get_bind()
inspector = sa.inspect(bind)
if "contract_fees" not in inspector.get_table_names():
return
if not _column_exists(inspector, "contract_fees", "contract_no"):
op.add_column("contract_fees", sa.Column("contract_no", sa.String(length=100), nullable=True))
if not _column_exists(inspector, "contract_fees", "signed_date"):
op.add_column("contract_fees", sa.Column("signed_date", sa.Date(), nullable=True))
if not _column_exists(inspector, "contract_fees", "currency"):
op.add_column("contract_fees", sa.Column("currency", sa.String(length=10), server_default="CNY", nullable=False))
if not _column_exists(inspector, "contract_fees", "remark"):
op.add_column("contract_fees", sa.Column("remark", sa.Text(), nullable=True))
def downgrade() -> None:
bind = op.get_bind()
inspector = sa.inspect(bind)
if "contract_fees" not in inspector.get_table_names():
return
for column_name in ("remark", "currency", "signed_date", "contract_no"):
if _column_exists(inspector, "contract_fees", column_name):
op.drop_column("contract_fees", column_name)
backend/alembic/versions/20260527_03_remove_legacy_finance_contracts.py
+124
View File
@@ -0,0 +1,124 @@
"""remove legacy finance contracts
Revision ID: 20260527_03
Revises: 20260527_02
Create Date: 2026-05-27 10:25:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision: str = "20260527_03"
down_revision: Union[str, None] = "20260527_02"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
LEGACY_KEYS = (
"finance_contracts:create",
"finance_contracts:list",
"finance_contracts:read",
"finance_contracts:update",
"finance_contracts:delete",
"fees_payments:create",
"fees_payments:update",
"fees_payments:delete",
"fees_attachments:create",
"fees_attachments:read",
"fees_attachments:delete",
)
def _table_exists(inspector: sa.Inspector, table_name: str) -> bool:
return table_name in inspector.get_table_names()
def upgrade() -> None:
bind = op.get_bind()
inspector = sa.inspect(bind)
if _table_exists(inspector, "finance_contracts") and _table_exists(inspector, "contract_fees"):
op.execute(
"""
UPDATE contract_fees cf
SET
contract_no = COALESCE(cf.contract_no, legacy.contract_no),
signed_date = COALESCE(cf.signed_date, legacy.signed_date),
currency = COALESCE(NULLIF(cf.currency, ''), legacy.currency, 'CNY'),
remark = COALESCE(cf.remark, legacy.remark)
FROM (
SELECT DISTINCT ON (fc.study_id, s.id)
fc.study_id,
s.id AS center_id,
fc.contract_no,
fc.signed_date,
fc.currency,
fc.remark
FROM finance_contracts fc
JOIN sites s
ON s.study_id = fc.study_id
AND s.name = fc.site_name
ORDER BY fc.study_id, s.id, fc.updated_at DESC, fc.created_at DESC
) legacy
WHERE cf.project_id = legacy.study_id
AND cf.center_id = legacy.center_id
"""
)
if _table_exists(inspector, "api_endpoint_permissions"):
quoted_keys = ", ".join(f"'{key}'" for key in LEGACY_KEYS)
op.execute(f"DELETE FROM api_endpoint_permissions WHERE endpoint_key IN ({quoted_keys})")
if _table_exists(inspector, "permission_templates"):
for key in LEGACY_KEYS:
op.execute(f"UPDATE permission_templates SET permissions = permissions::jsonb #- '{{PM,{key}}}'")
op.execute(f"UPDATE permission_templates SET permissions = permissions::jsonb #- '{{CRA,{key}}}'")
op.execute(f"UPDATE permission_templates SET permissions = permissions::jsonb #- '{{PV,{key}}}'")
op.execute(f"UPDATE permission_templates SET permissions = permissions::jsonb #- '{{QA,{key}}}'")
op.execute(f"UPDATE permission_templates SET permissions = permissions::jsonb #- '{{CTA,{key}}}'")
if _table_exists(inspector, "permission_template_versions"):
for key in LEGACY_KEYS:
op.execute(f"UPDATE permission_template_versions SET permissions = permissions::jsonb #- '{{PM,{key}}}'")
op.execute(f"UPDATE permission_template_versions SET permissions = permissions::jsonb #- '{{CRA,{key}}}'")
op.execute(f"UPDATE permission_template_versions SET permissions = permissions::jsonb #- '{{PV,{key}}}'")
op.execute(f"UPDATE permission_template_versions SET permissions = permissions::jsonb #- '{{QA,{key}}}'")
op.execute(f"UPDATE permission_template_versions SET permissions = permissions::jsonb #- '{{CTA,{key}}}'")
if _table_exists(inspector, "attachments"):
op.execute("DELETE FROM attachments WHERE entity_type = 'finance_contract'")
if _table_exists(inspector, "audit_logs"):
op.execute("DELETE FROM audit_logs WHERE entity_type = 'finance_contract'")
if _table_exists(inspector, "finance_contracts"):
op.drop_table("finance_contracts")
def downgrade() -> None:
bind = op.get_bind()
inspector = sa.inspect(bind)
if not _table_exists(inspector, "finance_contracts"):
op.create_table(
"finance_contracts",
sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False),
sa.Column("study_id", postgresql.UUID(as_uuid=True), nullable=False),
sa.Column("site_name", sa.String(length=255), nullable=False),
sa.Column("contract_no", sa.String(length=100), nullable=False),
sa.Column("signed_date", sa.Date(), nullable=True),
sa.Column("amount", sa.Numeric(12, 2), nullable=False),
sa.Column("currency", sa.String(length=10), nullable=False),
sa.Column("remark", sa.Text(), nullable=True),
sa.Column("created_by", postgresql.UUID(as_uuid=True), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.ForeignKeyConstraint(["created_by"], ["users.id"]),
sa.ForeignKeyConstraint(["study_id"], ["studies.id"]),
sa.PrimaryKeyConstraint("id"),
)
op.create_index("ix_finance_contracts_study_id", "finance_contracts", ["study_id"])
backend/app/api/v1/fees_contracts.py
+11 -3
View File
@@ -109,7 +109,11 @@ async def list_contract_fees(
id=contract.id,
project_id=contract.project_id,
center_id=contract.center_id,
contract_no=contract.contract_no,
signed_date=contract.signed_date,
contract_amount=contract_amount_decimal,
currency=contract.currency,
remark=contract.remark,
contract_cases=contract.contract_cases,
actual_cases=contract.actual_cases,
settlement_amount=_to_decimal(contract.settlement_amount) if contract.settlement_amount else None,
@@ -191,7 +195,11 @@ async def get_contract_fee(
id=contract.id,
project_id=contract.project_id,
center_id=contract.center_id,
contract_no=contract.contract_no,
signed_date=contract.signed_date,
contract_amount=_to_decimal(contract.contract_amount),
currency=contract.currency,
remark=contract.remark,
contract_cases=contract.contract_cases,
actual_cases=contract.actual_cases,
settlement_amount=_to_decimal(contract.settlement_amount) if contract.settlement_amount else None,
@@ -314,7 +322,7 @@ async def create_contract_payment(
contract = await contract_fee_crud.get_contract_fee(db, contract_id)
if not contract:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="合同费用不存在")
await _ensure_project_access(db, contract.project_id, current_user, "fees_payments:create")
await _ensure_project_access(db, contract.project_id, current_user, "fees_contracts:update")
_validate_payment_rules(payment_in)
payment = await payment_crud.create_payment(db, contract_id, payment_in)
await audit_crud.log_action(
@@ -347,7 +355,7 @@ async def update_contract_payment(
contract = await contract_fee_crud.get_contract_fee(db, payment.contract_fee_id)
if not contract:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="合同费用不存在")
await _ensure_project_access(db, contract.project_id, current_user, "fees_payments:update")
await _ensure_project_access(db, contract.project_id, current_user, "fees_contracts:update")
merged_payment = ContractFeePaymentCreate(
amount=payment_in.amount if payment_in.amount is not None else payment.amount,
paid_date=payment_in.paid_date if payment_in.paid_date is not None else payment.paid_date,
@@ -387,7 +395,7 @@ async def delete_contract_payment(
contract = await contract_fee_crud.get_contract_fee(db, payment.contract_fee_id)
if not contract:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="合同费用不存在")
await _ensure_project_access(db, contract.project_id, current_user, "fees_payments:delete")
await _ensure_project_access(db, contract.project_id, current_user, "fees_contracts:update")
await payment_crud.delete_payment(db, payment)
await payment_crud.resequence_payments(db, contract.id)
await audit_crud.log_action(
backend/app/api/v1/finance_contracts.py
-177
View File
@@ -1,177 +0,0 @@
import uuid
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.deps import get_cra_site_scope, get_current_user, get_db_session, require_study_not_locked, require_api_permission
from app.crud import audit as audit_crud
from app.crud import finance_contract as contract_crud
from app.crud import site as site_crud
from app.crud import study as study_crud
from app.schemas.finance_contract import FinanceContractCreate, FinanceContractRead, FinanceContractUpdate
router = APIRouter()
async def _ensure_study_exists(db: AsyncSession, study_id: uuid.UUID):
study = await study_crud.get(db, study_id)
if not study:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="项目不存在")
return study
async def _ensure_site_name_active(db: AsyncSession, study_id: uuid.UUID, site_name: str | None):
if not site_name:
return
active_names = await site_crud.list_active_names(db, study_id)
if site_name not in active_names:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="中心已停用")
@router.post(
"/contracts",
response_model=FinanceContractRead,
status_code=status.HTTP_201_CREATED,
dependencies=[Depends(require_api_permission("finance_contracts:create")), Depends(require_study_not_locked())],
)
async def create_contract(
study_id: uuid.UUID,
contract_in: FinanceContractCreate,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> FinanceContractRead:
await _ensure_study_exists(db, study_id)
cra_scope = await get_cra_site_scope(db, study_id, current_user)
if cra_scope and contract_in.site_name not in cra_scope[1]:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
await _ensure_site_name_active(db, study_id, contract_in.site_name)
contract = await contract_crud.create_contract(db, study_id, contract_in, created_by=current_user.id)
await audit_crud.log_action(
db,
study_id=study_id,
entity_type="finance_contract",
entity_id=contract.id,
action="CREATE_FINANCE_CONTRACT",
detail=f"合同费用 {contract.contract_no} 已创建",
operator_id=current_user.id,
operator_role=current_user.role,
)
return FinanceContractRead.model_validate(contract)
@router.get(
"/contracts",
response_model=list[FinanceContractRead],
dependencies=[Depends(require_api_permission("finance_contracts:read"))],
)
async def list_contracts(
study_id: uuid.UUID,
site_name: str | None = None,
contract_no: str | None = None,
skip: int = 0,
limit: int = 100,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> list[FinanceContractRead]:
await _ensure_study_exists(db, study_id)
cra_scope = await get_cra_site_scope(db, study_id, current_user)
site_names = cra_scope[1] if cra_scope else None
if site_name and site_names is not None and site_name not in site_names:
return []
items = await contract_crud.list_contracts(
db,
study_id,
site_name=site_name,
site_names=site_names,
contract_no=contract_no,
skip=skip,
limit=limit,
)
return [FinanceContractRead.model_validate(item) for item in items]
@router.get(
"/contracts/{contract_id}",
response_model=FinanceContractRead,
dependencies=[Depends(require_api_permission("finance_contracts:read"))],
)
async def get_contract(
study_id: uuid.UUID,
contract_id: uuid.UUID,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> FinanceContractRead:
await _ensure_study_exists(db, study_id)
contract = await contract_crud.get_contract(db, contract_id)
if not contract or contract.study_id != study_id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="合同不存在")
cra_scope = await get_cra_site_scope(db, study_id, current_user)
if cra_scope and contract.site_name not in cra_scope[1]:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
return FinanceContractRead.model_validate(contract)
@router.patch(
"/contracts/{contract_id}",
response_model=FinanceContractRead,
dependencies=[Depends(require_api_permission("finance_contracts:update")), Depends(require_study_not_locked())],
)
async def update_contract(
study_id: uuid.UUID,
contract_id: uuid.UUID,
contract_in: FinanceContractUpdate,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> FinanceContractRead:
await _ensure_study_exists(db, study_id)
contract = await contract_crud.get_contract(db, contract_id)
if not contract or contract.study_id != study_id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="合同不存在")
cra_scope = await get_cra_site_scope(db, study_id, current_user)
if cra_scope and contract.site_name not in cra_scope[1]:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
await _ensure_site_name_active(db, study_id, contract.site_name)
contract = await contract_crud.update_contract(db, contract, contract_in)
await audit_crud.log_action(
db,
study_id=study_id,
entity_type="finance_contract",
entity_id=contract_id,
action="UPDATE_FINANCE_CONTRACT",
detail=f"合同费用 {contract_id} 已更新",
operator_id=current_user.id,
operator_role=current_user.role,
)
return FinanceContractRead.model_validate(contract)
@router.delete(
"/contracts/{contract_id}",
status_code=status.HTTP_204_NO_CONTENT,
dependencies=[Depends(require_api_permission("finance_contracts:delete")), Depends(require_study_not_locked())],
)
async def delete_contract(
study_id: uuid.UUID,
contract_id: uuid.UUID,
db: AsyncSession = Depends(get_db_session),
current_user=Depends(get_current_user),
) -> None:
await _ensure_study_exists(db, study_id)
contract = await contract_crud.get_contract(db, contract_id)
if not contract or contract.study_id != study_id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="合同不存在")
cra_scope = await get_cra_site_scope(db, study_id, current_user)
if cra_scope and contract.site_name not in cra_scope[1]:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
await _ensure_site_name_active(db, study_id, contract.site_name)
await contract_crud.delete_contract(db, contract)
await audit_crud.log_action(
db,
study_id=study_id,
entity_type="finance_contract",
entity_id=contract_id,
action="DELETE_FINANCE_CONTRACT",
detail=f"合同费用 {contract_id} 已删除",
operator_id=current_user.id,
operator_role=current_user.role,
)
backend/app/crud/contract_fee.py
+4
View File
@@ -18,7 +18,11 @@ async def create_contract_fee(
contract = ContractFee(
project_id=contract_in.project_id,
center_id=contract_in.center_id,
contract_no=contract_in.contract_no,
signed_date=contract_in.signed_date,
contract_amount=contract_in.contract_amount,
currency=contract_in.currency,
remark=contract_in.remark,
contract_cases=contract_in.contract_cases,
actual_cases=contract_in.actual_cases,
settlement_amount=contract_in.settlement_amount,
backend/app/crud/finance_contract.py
-77
View File
@@ -1,77 +0,0 @@
import uuid
from typing import Sequence
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.finance_contract import FinanceContract
from app.schemas.finance_contract import FinanceContractCreate, FinanceContractUpdate
async def create_contract(
db: AsyncSession,
study_id: uuid.UUID,
contract_in: FinanceContractCreate,
created_by: uuid.UUID | None,
) -> FinanceContract:
contract = FinanceContract(
study_id=study_id,
site_name=contract_in.site_name,
contract_no=contract_in.contract_no,
signed_date=contract_in.signed_date,
amount=contract_in.amount,
currency=contract_in.currency,
remark=contract_in.remark,
created_by=created_by,
)
db.add(contract)
await db.commit()
await db.refresh(contract)
return contract
async def get_contract(db: AsyncSession, contract_id: uuid.UUID) -> FinanceContract | None:
result = await db.execute(select(FinanceContract).where(FinanceContract.id == contract_id))
return result.scalar_one_or_none()
async def list_contracts(
db: AsyncSession,
study_id: uuid.UUID,
site_name: str | None = None,
site_names: set[str] | None = None,
contract_no: str | None = None,
skip: int = 0,
limit: int = 100,
) -> Sequence[FinanceContract]:
stmt = (
select(FinanceContract)
.where(FinanceContract.study_id == study_id)
)
if site_names is not None:
if not site_names:
return []
stmt = stmt.where(FinanceContract.site_name.in_(site_names))
if site_name:
stmt = stmt.where(FinanceContract.site_name.ilike(f"%{site_name}%"))
if contract_no:
stmt = stmt.where(FinanceContract.contract_no.ilike(f"%{contract_no}%"))
stmt = stmt.order_by(FinanceContract.created_at.desc()).offset(skip).limit(limit)
result = await db.execute(stmt)
return result.scalars().all()
async def update_contract(
db: AsyncSession, contract: FinanceContract, contract_in: FinanceContractUpdate
) -> FinanceContract:
update_data = contract_in.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(contract, key, value)
await db.commit()
await db.refresh(contract)
return contract
async def delete_contract(db: AsyncSession, contract: FinanceContract) -> None:
await db.delete(contract)
await db.commit()
backend/app/models/contract_fee.py
+8 -4
View File
@@ -2,10 +2,10 @@ from __future__ import annotations
from typing import Optional
import uuid
from datetime import datetime
from datetime import date, datetime
from decimal import Decimal
from sqlalchemy import DateTime, ForeignKey, Integer, Numeric, UniqueConstraint, func
from sqlalchemy import Date, DateTime, ForeignKey, Integer, Numeric, String, Text, UniqueConstraint, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
@@ -21,11 +21,15 @@ class ContractFee(Base):
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
project_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("studies.id"), index=True, nullable=False)
center_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("sites.id"), index=True, nullable=False)
contract_no: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
signed_date: Mapped[Optional[date]] = mapped_column(Date, nullable=True)
contract_amount: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
currency: Mapped[str] = mapped_column(String(10), nullable=False, default="CNY")
remark: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
contract_cases: Mapped[int] = mapped_column(Integer, nullable=False)
actual_cases: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
settlement_amount: Mapped[Decimal | None] = mapped_column(Numeric(12, 2), nullable=True)
final_payment_amount: Mapped[Decimal | None] = mapped_column(Numeric(12, 2), nullable=True)
settlement_amount: Mapped[Optional[Decimal]] = mapped_column(Numeric(12, 2), nullable=True)
final_payment_amount: Mapped[Optional[Decimal]] = mapped_column(Numeric(12, 2), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
backend/app/models/finance_contract.py
-29
View File
@@ -1,29 +0,0 @@
from __future__ import annotations
from typing import Optional
import uuid
from datetime import date, datetime
from sqlalchemy import Date, DateTime, ForeignKey, Numeric, String, Text, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base_class import Base
class FinanceContract(Base):
__tablename__ = "finance_contracts"
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
study_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), ForeignKey("studies.id"), index=True, nullable=False)
site_name: Mapped[str] = mapped_column(String(255), nullable=False)
contract_no: Mapped[str] = mapped_column(String(100), nullable=False)
signed_date: Mapped[Optional[date]] = mapped_column(Date, nullable=True)
amount: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
currency: Mapped[str] = mapped_column(String(10), nullable=False, default="CNY")
remark: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
created_by: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
)
backend/app/schemas/contract_fee.py
+12
View File
@@ -12,7 +12,11 @@ from app.schemas.fee_attachment import FeeAttachmentRead
class ContractFeeCreate(BaseModel):
project_id: uuid.UUID
center_id: uuid.UUID
contract_no: Optional[str] = None
signed_date: Optional[date] = None
contract_amount: Decimal = Field(ge=0)
currency: str = "CNY"
remark: Optional[str] = None
contract_cases: int = Field(ge=0)
actual_cases: Optional[int] = Field(default=None, ge=0)
settlement_amount: Optional[Decimal] = Field(default=None, ge=0)
@@ -20,7 +24,11 @@ class ContractFeeCreate(BaseModel):
class ContractFeeUpdate(BaseModel):
contract_no: Optional[str] = None
signed_date: Optional[date] = None
contract_amount: Optional[Decimal] = Field(default=None, ge=0)
currency: Optional[str] = None
remark: Optional[str] = None
contract_cases: Optional[int] = Field(default=None, ge=0)
actual_cases: Optional[int] = Field(default=None, ge=0)
settlement_amount: Optional[Decimal] = Field(default=None, ge=0)
@@ -31,7 +39,11 @@ class ContractFeeRead(BaseModel):
id: uuid.UUID
project_id: uuid.UUID
center_id: uuid.UUID
contract_no: Optional[str]
signed_date: Optional[date]
contract_amount: Decimal
currency: str
remark: Optional[str]
contract_cases: int
actual_cases: Optional[int]
settlement_amount: Optional[Decimal]
backend/app/schemas/finance_contract.py
-40
View File
@@ -1,40 +0,0 @@
import uuid
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
class FinanceContractCreate(BaseModel):
site_name: str
contract_no: str
signed_date: Optional[date] = None
amount: Decimal = Field(gt=0)
currency: str = "CNY"
remark: Optional[str] = None
class FinanceContractUpdate(BaseModel):
site_name: Optional[str] = None
contract_no: Optional[str] = None
signed_date: Optional[date] = None
amount: Optional[Decimal] = Field(default=None, gt=0)
currency: Optional[str] = None
remark: Optional[str] = None
class FinanceContractRead(BaseModel):
id: uuid.UUID
study_id: uuid.UUID
site_name: str
contract_no: str
signed_date: Optional[date]
amount: Decimal
currency: str
remark: Optional[str]
created_by: Optional[uuid.UUID]
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
backend/tests/test_contract_fee_basic_fields.py
+50
View File
@@ -0,0 +1,50 @@
"""合同费用基础字段测试"""
from datetime import date
from decimal import Decimal
import uuid
from app.models.contract_fee import ContractFee
from app.schemas.contract_fee import ContractFeeCreate, ContractFeeRead, ContractFeeUpdate
def test_contract_fee_schema_includes_contract_basic_fields():
"""合同费用应包含合同基础信息字段。"""
project_id = uuid.uuid4()
center_id = uuid.uuid4()
payload = ContractFeeCreate(
project_id=project_id,
center_id=center_id,
contract_no="CT-001",
signed_date=date(2026, 5, 27),
contract_amount=Decimal("120000.00"),
currency="CNY",
remark="首版合同",
contract_cases=12,
)
assert payload.contract_no == "CT-001"
assert payload.signed_date == date(2026, 5, 27)
assert payload.currency == "CNY"
assert payload.remark == "首版合同"
update_payload = ContractFeeUpdate(contract_no="CT-002", currency="USD", remark="变更合同信息")
assert update_payload.model_dump(exclude_unset=True) == {
"contract_no": "CT-002",
"currency": "USD",
"remark": "变更合同信息",
}
for field in ("contract_no", "signed_date", "currency", "remark"):
assert field in ContractFeeRead.model_fields
def test_contract_fee_model_includes_contract_basic_columns():
"""合同费用模型应持久化合同基础信息。"""
columns = ContractFee.__table__.columns
assert "contract_no" in columns
assert "signed_date" in columns
assert "currency" in columns
assert "remark" in columns
backend/tests/test_remove_legacy_finance_contracts.py
+25
View File
@@ -0,0 +1,25 @@
"""旧合同基础信息模块移除测试"""
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_api_router_no_longer_registers_legacy_finance_contracts():
router_source = (ROOT / "app" / "api" / "v1" / "router.py").read_text()
assert "finance_contracts" not in router_source
assert "finance-contracts" not in router_source
def test_active_backend_code_no_longer_imports_finance_contract_model():
checked_paths = [
ROOT / "app" / "db" / "base.py",
ROOT / "app" / "crud" / "study.py",
ROOT / "app" / "crud" / "site.py",
ROOT / "app" / "crud" / "overview.py",
]
for path in checked_paths:
assert "FinanceContract" not in path.read_text()