首页 发现 帮助
注册 登录
wuzilong
/
zkxw
1
0
派生 0
文件 工单管理 0 合并请求 0 Wiki
目录树: 66e442e1e8
分支列表 标签列表
zkxw
/
docs
/
superpowers
/
plans
/
2026-06-17-protocol-invoice-request.md

2026-06-17-protocol-invoice-request.md 42 KB
文件历史 原始文件

采集协议单开票申请 Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 在满月待办中支持多期合并开票申请（主表+子表），财务在缴退费财务确认新增「开票申请」Tab 上传发票附件并完成处理。

Architecture: 新增 biz_protocol_invoice_request 主表与 biz_protocol_invoice_request_item 子表；CRUD 落在 collection_protocol.py（与缴退费确认同模块）；API 挂在现有 collection_protocol 路由；前端扩展 FullmoonTodoTab.vue 统一弹窗 + 新建 InvoiceRequestTab.vue 对齐 RefundConfirmTab.vue。

Tech Stack: MySQL 5.7 · FastAPI + SQLAlchemy · Vue 3 + Element Plus · pnpm

Design Doc: docs/superpowers/specs/2026-06-17-protocol-invoice-request-design.md

Reference Pattern Files:

  • 缴退费财务 CRUD/API：backend/app/crud/collection_protocol.py（list_receipt_confirm_details L1509+）
  • 缴退费 API：backend/app/api/collection_protocol.py（receipt-confirm L565+）
  • 满月待办前端：web-admin/src/views/cell-storage/protocol-audit/FullmoonTodoTab.vue
  • 财务退费 Tab：web-admin/src/views/cell-storage/payment-confirm/RefundConfirmTab.vue
  • 财务附件：web-admin/src/components/business/protocol-finance-confirm-fields/

File Structure Map

New Files

File Responsibility
docs/sql/protocol_invoice_request.sql 主表 + 子表 DDL
backend/tests/test_protocol_invoice_request.py 开票申请 CRUD 单元测试
web-admin/src/views/cell-storage/payment-confirm/InvoiceRequestTab.vue 财务开票 Tab

Modified Files

File Change
backend/app/models/models.py +2 Models
backend/app/crud/collection_protocol.py +开票申请 CRUD 函数
backend/app/api/collection_protocol.py +6 端点 + Pydantic Body
web-admin/src/api/collection-protocol.ts +6 fetch*
web-admin/src/views/cell-storage/protocol-audit/FullmoonTodoTab.vue +开票申请统一弹窗
web-admin/src/views/cell-storage/payment-confirm/index.vue +第三 Tab + 角标

Task 1: 数据库 DDL

Files:

  • Create: docs/sql/protocol_invoice_request.sql

  • [ ] Step 1: 写入 DDL（与设计文档 §5 一致）

-- docs/sql/protocol_invoice_request.sql
DROP TABLE IF EXISTS `biz_protocol_invoice_request_item`;
DROP TABLE IF EXISTS `biz_protocol_invoice_request`;

CREATE TABLE `biz_protocol_invoice_request` (
  `id` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '主键ID',
  `collection_protocol_id` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '采集协议单ID',
  `invoice_amount` decimal(18,2) NOT NULL COMMENT '开票合计金额',
  `invoice_title` varchar(200) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '发票抬头',
  `tax_no` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '税号',
  `company_address` varchar(512) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '单位地址',
  `company_phone` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '电话',
  `bank_name` varchar(200) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '开户银行',
  `bank_account` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '银行账号',
  `apply_remark` varchar(512) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '申请备注',
  `invoice_attachments` text COLLATE utf8mb4_unicode_ci COMMENT '财务发票附件逗号分隔URL',
  `finance_remark` varchar(512) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '财务备注',
  `status` varchar(8) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT 'dkp' COMMENT 'dkp待开票 ykp已开票',
  `finance_handle_time` datetime DEFAULT NULL COMMENT '财务处理时间',
  `tenant_no` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT 'zkxw' COMMENT '租户编号',
  `order_num` int(11) DEFAULT NULL COMMENT '排序号',
  `remark` varchar(512) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '备注',
  `create_by` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '创建人id',
  `create_dept_id` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '' COMMENT '创建人部门ID',
  `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_by` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记',
  PRIMARY KEY (`id`),
  KEY `idx_inv_req_protocol` (`collection_protocol_id`,`del_flag`),
  KEY `idx_inv_req_status` (`tenant_no`,`status`,`del_flag`,`create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='采集协议单开票申请主表';

CREATE TABLE `biz_protocol_invoice_request_item` (
  `id` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '主键ID',
  `invoice_request_id` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '开票申请主表ID',
  `payment_id` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '缴费主表ID',
  `payment_detail_id` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '缴费明细ID',
  `installment_no` int(11) DEFAULT NULL COMMENT '期数快照',
  `line_amount` decimal(18,2) NOT NULL COMMENT '该行开票金额',
  `tenant_no` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT 'zkxw' COMMENT '租户编号',
  `order_num` int(11) DEFAULT NULL COMMENT '排序号',
  `remark` varchar(512) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '备注',
  `create_by` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '创建人id',
  `create_dept_id` varchar(64) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '' COMMENT '创建人部门ID',
  `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_by` varchar(64) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记',
  PRIMARY KEY (`id`),
  KEY `idx_inv_item_req` (`invoice_request_id`,`del_flag`),
  KEY `idx_inv_item_detail` (`payment_detail_id`,`del_flag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='采集协议单开票申请明细';
  • Step 2: 本地库执行迁移（开发者环境）
mysql -u root -p zkxw < docs/sql/protocol_invoice_request.sql

Expected: 两张表创建成功，无 ERROR。

Task 2: SQLAlchemy Models

Files:

  • Modify: backend/app/models/models.py（在 BizProtocolPaymentDetail 之后插入）

  • [ ] Step 1: 新增两个 Model 类

class BizProtocolInvoiceRequest(Base):
    """采集协议单开票申请主表 biz_protocol_invoice_request"""

    __tablename__ = "biz_protocol_invoice_request"

    id = Column(String(64), primary_key=True, comment="主键ID")
    collection_protocol_id = Column(String(64), nullable=False, comment="采集协议单ID")
    invoice_amount = Column(Numeric(18, 2), nullable=False, comment="开票合计金额")
    invoice_title = Column(String(200), nullable=True, comment="发票抬头")
    tax_no = Column(String(64), nullable=True, comment="税号")
    company_address = Column(String(512), nullable=True, comment="单位地址")
    company_phone = Column(String(64), nullable=True, comment="电话")
    bank_name = Column(String(200), nullable=True, comment="开户银行")
    bank_account = Column(String(64), nullable=True, comment="银行账号")
    apply_remark = Column(String(512), nullable=True, comment="申请备注")
    invoice_attachments = Column(Text, nullable=True, comment="财务发票附件")
    finance_remark = Column(String(512), nullable=True, comment="财务备注")
    status = Column(String(8), nullable=False, default="dkp", comment="dkp/ykp")
    finance_handle_time = Column(DateTime, nullable=True, comment="财务处理时间")
    tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
    order_num = Column(Integer, nullable=True, comment="排序号")
    remark = Column(String(512), nullable=True, comment="备注")
    create_by = Column(String(64), nullable=True, comment="创建人id")
    create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
    create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
    update_by = Column(String(64), nullable=True, comment="更新人id")
    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
    del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")


class BizProtocolInvoiceRequestItem(Base):
    """采集协议单开票申请明细 biz_protocol_invoice_request_item"""

    __tablename__ = "biz_protocol_invoice_request_item"

    id = Column(String(64), primary_key=True, comment="主键ID")
    invoice_request_id = Column(String(64), nullable=False, comment="开票申请主表ID")
    payment_id = Column(String(64), nullable=False, comment="缴费主表ID")
    payment_detail_id = Column(String(64), nullable=False, comment="缴费明细ID")
    installment_no = Column(Integer, nullable=True, comment="期数快照")
    line_amount = Column(Numeric(18, 2), nullable=False, comment="该行开票金额")
    tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
    order_num = Column(Integer, nullable=True, comment="排序号")
    remark = Column(String(512), nullable=True, comment="备注")
    create_by = Column(String(64), nullable=True, comment="创建人id")
    create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
    create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
    update_by = Column(String(64), nullable=True, comment="更新人id")
    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
    del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")
  • Step 2: 验证 import
cd backend && source venv/bin/activate && python -c "from app.models.models import BizProtocolInvoiceRequest, BizProtocolInvoiceRequestItem; print('ok')"

Expected: ok

Task 3: CRUD 层（含单元测试）

Files:

  • Modify: backend/app/crud/collection_protocol.py
  • Create: backend/tests/test_protocol_invoice_request.py

在文件末尾（upsert_fullmoon_todo 之后）新增以下函数。

3.1 辅助函数

  • Step 1: 写失败测试 — installmentSummary
# backend/tests/test_protocol_invoice_request.py
import unittest
from decimal import Decimal
from unittest.mock import MagicMock, patch

class TestInstallmentSummary(unittest.TestCase):
    def test_build_installment_summary_single(self):
        from app.crud.collection_protocol import build_installment_summary
        self.assertEqual(build_installment_summary([1]), "第1期")

    def test_build_installment_summary_multi(self):
        from app.crud.collection_protocol import build_installment_summary
        self.assertEqual(build_installment_summary([1, 2, 3]), "第1、2、3期")

    def test_build_installment_summary_lump_sum(self):
        from app.crud.collection_protocol import build_installment_summary
        self.assertEqual(build_installment_summary([None]), "趸交")
  • Step 2: 运行测试确认 FAIL
cd backend && source venv/bin/activate && pytest tests/test_protocol_invoice_request.py::TestInstallmentSummary -v

Expected: FAIL ImportError or cannot import build_installment_summary

  • Step 3: 实现辅助函数
def build_installment_summary(installment_nos: List[Optional[int]]) -> str:
    nums = sorted({n for n in installment_nos if n is not None})
    if not nums:
        return "趸交"
    if len(nums) == 1:
        return f"第{nums[0]}期"
    return "第" + "、".join(str(n) for n in nums) + "期"


def _invoice_request_item_to_dict(item: BizProtocolInvoiceRequestItem) -> Dict[str, Any]:
    return {
        "id": item.id,
        "invoiceRequestId": item.invoice_request_id,
        "paymentId": item.payment_id,
        "paymentDetailId": item.payment_detail_id,
        "installmentNo": item.installment_no,
        "lineAmount": float(item.line_amount) if item.line_amount is not None else None,
        "orderNum": item.order_num,
        "remark": item.remark or "",
        "createTime": item.create_time.strftime("%Y-%m-%d %H:%M:%S") if item.create_time else "",
    }


def _invoice_request_to_dict(
    req: BizProtocolInvoiceRequest,
    *,
    items: Optional[List[Dict[str, Any]]] = None,
    installment_summary: Optional[str] = None,
    protocol_ctx: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    out = {
        "id": req.id,
        "collectionProtocolId": req.collection_protocol_id,
        "invoiceAmount": float(req.invoice_amount) if req.invoice_amount is not None else None,
        "invoiceTitle": req.invoice_title or "",
        "taxNo": req.tax_no or "",
        "companyAddress": req.company_address or "",
        "companyPhone": req.company_phone or "",
        "bankName": req.bank_name or "",
        "bankAccount": req.bank_account or "",
        "applyRemark": req.apply_remark or "",
        "invoiceAttachments": req.invoice_attachments or "",
        "financeRemark": req.finance_remark or "",
        "status": req.status,
        "financeHandleTime": req.finance_handle_time.strftime("%Y-%m-%d %H:%M:%S") if req.finance_handle_time else "",
        "tenantNo": req.tenant_no,
        "orderNum": req.order_num,
        "remark": req.remark or "",
        "createTime": req.create_time.strftime("%Y-%m-%d %H:%M:%S") if req.create_time else "",
        "updateTime": req.update_time.strftime("%Y-%m-%d %H:%M:%S") if req.update_time else "",
    }
    if items is not None:
        out["items"] = items
        out["itemCount"] = len(items)
    if installment_summary is not None:
        out["installmentSummary"] = installment_summary
    if protocol_ctx:
        out.update(protocol_ctx)
    return out


def _active_invoiced_payment_detail_ids(db: Session, tenant_no: str) -> set:
    rows = (
        db.query(BizProtocolInvoiceRequestItem.payment_detail_id)
        .join(
            BizProtocolInvoiceRequest,
            BizProtocolInvoiceRequest.id == BizProtocolInvoiceRequestItem.invoice_request_id,
        )
        .filter(
            BizProtocolInvoiceRequest.tenant_no == tenant_no,
            BizProtocolInvoiceRequest.del_flag == 0,
            BizProtocolInvoiceRequest.status.in_(("dkp", "ykp")),
            BizProtocolInvoiceRequestItem.del_flag == 0,
        )
        .all()
    )
    return {r[0] for r in rows}

需在文件顶部 import 区增加：

from app.models.models import (
    ...
    BizProtocolInvoiceRequest,
    BizProtocolInvoiceRequestItem,
)
  • Step 4: 运行 TestInstallmentSummary 通过
pytest tests/test_protocol_invoice_request.py::TestInstallmentSummary -v

Expected: 3 passed

3.2 可开票明细 & 创建申请

  • Step 5: 写失败测试 — 重复 payment_detail_id 拒绝
class TestCreateInvoiceRequestValidation(unittest.TestCase):
    @patch("app.crud.collection_protocol.generate_uuid", return_value="req-1")
    def test_rejects_duplicate_payment_detail_in_active_request(self, _mock_uuid):
        from app.crud.collection_protocol import create_invoice_request, ValueError as CrudValueError
        mock_db = MagicMock()
        # 简化：mock 查询返回已占用 detail_id
        with patch("app.crud.collection_protocol._active_invoiced_payment_detail_ids", return_value={"pd-1"}):
            with self.assertRaises(ValueError):
                create_invoice_request(
                    mock_db,
                    collection_protocol_id="proto-1",
                    payment_detail_ids=["pd-1"],
                    tenant_no="t1",
                    user_info={"user_id": "u1", "dept_id": ""},
                )
  • Step 6: 实现 list_invoice_eligible_payment_details

逻辑要点：

  1. 查协议单 collection_protocol_id 下所有 BizProtocolPayment → BizProtocolPaymentDetail
  2. 过滤 pay_status == "yjf"
  3. 排除 _active_invoiced_payment_detail_ids 中的 id
  4. 复用 _rows_to_payment_detail_dicts 或轻量 dict（含 installmentNo, transferAmount, transferTime, transferType, installmentDueDate）
def list_invoice_eligible_payment_details(
    db: Session,
    *,
    collection_protocol_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    if not tenant_no or not collection_protocol_id:
        return []
    invoiced_ids = _active_invoiced_payment_detail_ids(db, tenant_no)
    payment = (
        db.query(BizProtocolPayment)
        .filter(
            BizProtocolPayment.collection_protocol_id == collection_protocol_id,
            BizProtocolPayment.tenant_no == tenant_no,
            BizProtocolPayment.del_flag == 0,
        )
        .first()
    )
    if not payment:
        return []
    details = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            BizProtocolPaymentDetail.payment_id == payment.id,
            BizProtocolPaymentDetail.tenant_no == tenant_no,
            BizProtocolPaymentDetail.del_flag == 0,
            BizProtocolPaymentDetail.pay_status == "yjf",
        )
        .order_by(BizProtocolPaymentDetail.order_num.asc(), BizProtocolPaymentDetail.create_time.desc())
        .all()
    )
    out = []
    for d in details:
        if d.id in invoiced_ids:
            continue
        out.append({
            "id": d.id,
            "paymentId": payment.id,
            "installmentNo": d.installment_no,
            "dueAmount": float(d.due_amount) if d.due_amount is not None else None,
            "transferAmount": float(d.transfer_amount) if d.transfer_amount is not None else None,
            "transferTime": d.transfer_time.strftime("%Y-%m-%d %H:%M:%S") if d.transfer_time else "",
            "transferType": d.transfer_type or "",
            "installmentDueDate": d.installment_due_date or "",
        })
    return out
  • Step 7: 实现 create_invoice_request（同事务 1 主表 + N 子表）
def create_invoice_request(
    db: Session,
    *,
    collection_protocol_id: str,
    payment_detail_ids: List[str],
    tenant_no: str,
    user_info: Optional[dict] = None,
    invoice_title: Optional[str] = None,
    tax_no: Optional[str] = None,
    company_address: Optional[str] = None,
    company_phone: Optional[str] = None,
    bank_name: Optional[str] = None,
    bank_account: Optional[str] = None,
    apply_remark: Optional[str] = None,
) -> Dict[str, Any]:
    ids = list(dict.fromkeys([str(i).strip() for i in (payment_detail_ids or []) if str(i).strip()]))
    if not ids:
        raise ValueError("请至少选择一条缴费明细")

    invoiced_ids = _active_invoiced_payment_detail_ids(db, tenant_no)
    overlap = set(ids) & invoiced_ids
    if overlap:
        raise ValueError("所选缴费明细中存在已申请开票的记录")

    payment = (
        db.query(BizProtocolPayment)
        .filter(
            BizProtocolPayment.collection_protocol_id == collection_protocol_id,
            BizProtocolPayment.tenant_no == tenant_no,
            BizProtocolPayment.del_flag == 0,
        )
        .first()
    )
    if not payment:
        raise ValueError("未找到缴费信息")

    details = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            BizProtocolPaymentDetail.payment_id == payment.id,
            BizProtocolPaymentDetail.id.in_(ids),
            BizProtocolPaymentDetail.tenant_no == tenant_no,
            BizProtocolPaymentDetail.del_flag == 0,
            BizProtocolPaymentDetail.pay_status == "yjf",
        )
        .all()
    )
    if len(details) != len(ids):
        raise ValueError("部分缴费明细不可开票")

    total = Decimal("0")
    detail_map = {d.id: d for d in details}
    ordered_details = [detail_map[i] for i in ids]

    uid = (user_info or {}).get("user_id") or ""
    dept_id = (user_info or {}).get("dept_id") or ""

    req = BizProtocolInvoiceRequest(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        invoice_amount=Decimal("0"),
        invoice_title=(invoice_title or "").strip() or None,
        tax_no=(tax_no or "").strip() or None,
        company_address=(company_address or "").strip() or None,
        company_phone=(company_phone or "").strip() or None,
        bank_name=(bank_name or "").strip() or None,
        bank_account=(bank_account or "").strip() or None,
        apply_remark=(apply_remark or "").strip() or None,
        status="dkp",
        tenant_no=tenant_no,
        create_by=uid,
        create_dept_id=dept_id,
        update_by=uid,
    )
    db.add(req)
    db.flush()

    item_dicts = []
    for idx, d in enumerate(ordered_details, start=1):
        line_amt = Decimal(str(d.transfer_amount or 0))
        total += line_amt
        item = BizProtocolInvoiceRequestItem(
            id=generate_uuid(),
            invoice_request_id=req.id,
            payment_id=payment.id,
            payment_detail_id=d.id,
            installment_no=d.installment_no,
            line_amount=line_amt,
            tenant_no=tenant_no,
            order_num=idx,
            create_by=uid,
            create_dept_id=dept_id,
            update_by=uid,
        )
        db.add(item)
        item_dicts.append(_invoice_request_item_to_dict(item))

    req.invoice_amount = total
    db.commit()
    db.refresh(req)

    summary = build_installment_summary([d.installment_no for d in ordered_details])
    return _invoice_request_to_dict(req, items=item_dicts, installment_summary=summary)
  • Step 8: 实现列表/完成/计数函数
def list_invoice_requests_by_protocol(
    db: Session,
    *,
    collection_protocol_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizProtocolInvoiceRequest)
        .filter(
            BizProtocolInvoiceRequest.collection_protocol_id == collection_protocol_id,
            BizProtocolInvoiceRequest.tenant_no == tenant_no,
            BizProtocolInvoiceRequest.del_flag == 0,
        )
        .order_by(BizProtocolInvoiceRequest.create_time.desc())
        .all()
    )
    return [_pack_invoice_request_with_items(db, r) for r in rows]


def _pack_invoice_request_with_items(db: Session, req: BizProtocolInvoiceRequest) -> Dict[str, Any]:
    items_orm = (
        db.query(BizProtocolInvoiceRequestItem)
        .filter(
            BizProtocolInvoiceRequestItem.invoice_request_id == req.id,
            BizProtocolInvoiceRequestItem.del_flag == 0,
        )
        .order_by(BizProtocolInvoiceRequestItem.order_num.asc())
        .all()
    )
    item_dicts = [_invoice_request_item_to_dict(i) for i in items_orm]
    summary = build_installment_summary([i.installment_no for i in items_orm])
    return _invoice_request_to_dict(req, items=item_dicts, installment_summary=summary)


def list_protocol_invoice_requests(
    db: Session,
    *,
    status: str,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    # 参照 list_receipt_confirm_details：join 协议单 + party_customer 做 keyword
    # 主查询 BizProtocolInvoiceRequest.status == status
    # 排序 create_time DESC（待开票）/ finance_handle_time DESC（已开票）
    # 每条 _pack_invoice_request_with_items + 附加 motherName/protocolNo 等
    ...


def count_pending_protocol_invoice_requests(db: Session, *, tenant_no: str, user_info=None) -> int:
    ...


def complete_invoice_request(
    db: Session,
    *,
    request_id: str,
    tenant_no: str,
    invoice_attachments: str,
    finance_remark: Optional[str] = None,
    user_info: Optional[dict] = None,
) -> Dict[str, Any]:
    if not (invoice_attachments or "").strip():
        raise ValueError("请上传发票附件")
    req = (
        db.query(BizProtocolInvoiceRequest)
        .filter(
            BizProtocolInvoiceRequest.id == request_id,
            BizProtocolInvoiceRequest.tenant_no == tenant_no,
            BizProtocolInvoiceRequest.del_flag == 0,
        )
        .first()
    )
    if not req:
        raise ValueError("开票申请不存在")
    if req.status != "dkp":
        raise ValueError("当前状态不可完成开票")
    uid = (user_info or {}).get("user_id") or ""
    req.status = "ykp"
    req.invoice_attachments = invoice_attachments.strip()
    req.finance_remark = (finance_remark or "").strip() or None
    req.finance_handle_time = datetime.now()
    req.update_by = uid
    db.commit()
    db.refresh(req)
    return _pack_invoice_request_with_items(db, req)

list_protocol_invoice_requests 的 keyword 筛选复用 _apply_payment_detail_list_filters 的 party_customer 子查询模式，或复制 RefundConfirmTab 对应 CRUD 中的 join 写法（BizProtocolPartyCustomer.mother_name 等 ilike）。

  • Step 9: 运行全部开票测试
pytest tests/test_protocol_invoice_request.py -v

Expected: all passed

Task 4: API 路由

Files:

  • Modify: backend/app/api/collection_protocol.py

  • [ ] Step 1: 新增 Pydantic Body（文件内，与现有 Body 风格一致）

class InvoiceRequestCreateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    payment_detail_ids: List[str] = Field(..., alias="paymentDetailIds", min_length=1)
    invoice_title: Optional[str] = Field(None, alias="invoiceTitle")
    tax_no: Optional[str] = Field(None, alias="taxNo")
    company_address: Optional[str] = Field(None, alias="companyAddress")
    company_phone: Optional[str] = Field(None, alias="companyPhone")
    bank_name: Optional[str] = Field(None, alias="bankName")
    bank_account: Optional[str] = Field(None, alias="bankAccount")
    apply_remark: Optional[str] = Field(None, alias="applyRemark")


class InvoiceRequestCompleteBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    invoice_attachments: str = Field(..., alias="invoiceAttachments")
    finance_remark: Optional[str] = Field(None, alias="financeRemark")
  • Step 2: 注册 6 个端点
@router.get("/collection-protocols/{protocol_id}/invoice-eligible-payments")
async def get_invoice_eligible_payments(protocol_id: str, db: Session = Depends(get_db), user=Depends(require_admin_or_super_admin)):
    tenant_no = _get_tenant_no(db, user)
    items = cp_svc.list_invoice_eligible_payment_details(db, collection_protocol_id=protocol_id, tenant_no=tenant_no)
    return success_response({"list": items, "total": len(items)})


@router.get("/collection-protocols/{protocol_id}/invoice-requests")
async def get_protocol_invoice_requests(protocol_id: str, db: Session = Depends(get_db), user=Depends(require_admin_or_super_admin)):
    tenant_no = _get_tenant_no(db, user)
    items = cp_svc.list_invoice_requests_by_protocol(db, collection_protocol_id=protocol_id, tenant_no=tenant_no)
    return success_response({"list": items, "total": len(items)})


@router.post("/collection-protocols/{protocol_id}/invoice-requests")
async def post_protocol_invoice_request(protocol_id: str, body: InvoiceRequestCreateBody, db: Session = Depends(get_db), user=Depends(require_admin_or_super_admin)):
    tenant_no = _get_tenant_no(db, user)
    try:
        row = cp_svc.create_invoice_request(
            db,
            collection_protocol_id=protocol_id,
            payment_detail_ids=body.payment_detail_ids,
            tenant_no=tenant_no,
            user_info=user,
            invoice_title=body.invoice_title,
            tax_no=body.tax_no,
            company_address=body.company_address,
            company_phone=body.company_phone,
            bank_name=body.bank_name,
            bank_account=body.bank_account,
            apply_remark=body.apply_remark,
        )
    except ValueError as e:
        return error_response(str(e))
    return success_response(row, message="提交成功")


@router.get("/protocol-invoice-requests")
async def list_protocol_invoice_requests_api(
    status: str = Query(..., description="dkp|ykp"),
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
    keyword: Optional[str] = Query(None),
    db: Session = Depends(get_db),
    user=Depends(require_admin_or_super_admin),
):
    tenant_no = _get_tenant_no(db, user)
    items, total = cp_svc.list_protocol_invoice_requests(
        db, status=status, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user
    )
    return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize})


@router.get("/protocol-invoice-requests/pending-count")
async def protocol_invoice_pending_count(db: Session = Depends(get_db), user=Depends(require_admin_or_super_admin)):
    tenant_no = _get_tenant_no(db, user)
    count = cp_svc.count_pending_protocol_invoice_requests(db, tenant_no=tenant_no, user_info=user)
    return success_response({"count": count})


@router.put("/protocol-invoice-requests/{request_id}/complete")
async def complete_protocol_invoice_request(
    request_id: str,
    body: InvoiceRequestCompleteBody,
    db: Session = Depends(get_db),
    user=Depends(require_admin_or_super_admin),
):
    tenant_no = _get_tenant_no(db, user)
    try:
        row = cp_svc.complete_invoice_request(
            db,
            request_id=request_id,
            tenant_no=tenant_no,
            invoice_attachments=body.invoice_attachments,
            finance_remark=body.finance_remark,
            user_info=user,
        )
    except ValueError as e:
        return error_response(str(e))
    return success_response(row, message="已完成开票")
  • Step 3: 冒烟（需本地 DB 已迁移且后端运行）
cd backend && source venv/bin/activate && pytest tests/test_protocol_invoice_request.py tests/test_collection_protocol_conventions.py -q

Expected: all passed

Task 5: 前端 API

Files:

  • Modify: web-admin/src/api/collection-protocol.ts

  • [ ] Step 1: 追加 6 个函数

export function fetchInvoiceEligiblePayments(protocolId: string) {
  return request.get<{ list: Record<string, unknown>[]; total: number }>({
    url: `/web/admin/collection-protocols/${protocolId}/invoice-eligible-payments`
  })
}

export function fetchProtocolInvoiceRequests(protocolId: string) {
  return request.get<{ list: Record<string, unknown>[]; total: number }>({
    url: `/web/admin/collection-protocols/${protocolId}/invoice-requests`
  })
}

export function fetchCreateProtocolInvoiceRequest(
  protocolId: string,
  data: {
    paymentDetailIds: string[]
    invoiceTitle?: string
    taxNo?: string
    companyAddress?: string
    companyPhone?: string
    bankName?: string
    bankAccount?: string
    applyRemark?: string
  }
) {
  return request.post({
    url: `/web/admin/collection-protocols/${protocolId}/invoice-requests`,
    data,
    showSuccessMessage: true
  })
}

export function fetchProtocolInvoiceRequestList(params: {
  status: 'dkp' | 'ykp'
  page?: number
  pageSize?: number
  keyword?: string
}) {
  return request.get<PageList<Record<string, unknown>>>({
    url: '/web/admin/protocol-invoice-requests',
    params
  })
}

export function fetchProtocolInvoicePendingCount() {
  return request.get<{ count: number }>({
    url: '/web/admin/protocol-invoice-requests/pending-count'
  })
}

export function fetchCompleteProtocolInvoiceRequest(
  requestId: string,
  data: { invoiceAttachments: string; financeRemark?: string }
) {
  return request.put({
    url: `/web/admin/protocol-invoice-requests/${requestId}/complete`,
    data,
    showSuccessMessage: true
  })
}

Task 6: 满月待办 — 开票申请统一弹窗

Files:

  • Modify: web-admin/src/views/cell-storage/protocol-audit/FullmoonTodoTab.vue

  • [ ] Step 1: 操作列增加「开票申请」按钮

在「办理保险」后增加：

<el-button class="btn-op-item" @click="openInvoiceDialog(row)">
  <el-icon><Tickets /></el-icon>开票申请
</el-button>

import { Tickets } from '@element-plus/icons-vue'；操作列 min-width 调至约 320。

  • Step 2: 新增弹窗 state 与加载函数
const invoiceDialogVisible = ref(false)
const invoiceProtocolId = ref('')
const invoiceSummaryRow = ref<any>(null)
const invoiceHistory = ref<any[]>([])
const invoiceEligible = ref<any[]>([])
const invoiceSelectedIds = ref<string[]>([])
const invoiceForm = reactive({
  invoiceTitle: '',
  taxNo: '',
  companyAddress: '',
  companyPhone: '',
  bankName: '',
  bankAccount: '',
  applyRemark: ''
})
const invoiceSubmitLoading = ref(false)

const invoiceSelectedTotal = computed(() =>
  invoiceEligible.value
    .filter((r) => invoiceSelectedIds.value.includes(String(r.id)))
    .reduce((sum, r) => sum + Number(r.transferAmount || 0), 0)
)

async function loadInvoiceDialogData() {
  const pid = invoiceProtocolId.value
  if (!pid) return
  const [histRes, eligRes] = await Promise.all([
    fetchProtocolInvoiceRequests(pid),
    fetchInvoiceEligiblePayments(pid)
  ])
  invoiceHistory.value = (histRes as any)?.list || []
  invoiceEligible.value = (eligRes as any)?.list || []
  invoiceSelectedIds.value = []
}

function openInvoiceDialog(row: any) {
  invoiceProtocolId.value = String(row.protocolId || row.id || '')
  invoiceSummaryRow.value = row
  invoiceDialogVisible.value = true
  loadInvoiceDialogData()
}
  • Step 3: 弹窗模板（三块结构）
<el-dialog
  v-model="invoiceDialogVisible"
  title="开票申请"
  width="880px"
  class="dialog-form-layout"
  align-center
  append-to-body
  modal-class="protocol-dialog-mask"
  destroy-on-close
>
  <!-- 摘要 -->
  <div class="invoice-dialog-summary">
    <span>{{ invoiceSummaryRow?.motherName || '—' }}</span>
    <span>协议号 {{ invoiceSummaryRow?.protocolNo || '—' }}</span>
  </div>

  <!-- 申请记录 -->
  <h4 class="detail-section-title">申请记录</h4>
  <el-table :data="invoiceHistory" stripe class="list-table" empty-text="暂无开票申请">
    <el-table-column type="expand">
      <template #default="{ row }">
        <el-table :data="row.items || []" size="small">
          <el-table-column prop="installmentNo" label="期数" min-width="80">
            <template #default="{ row: item }">{{ item.installmentNo ?? '趸交' }}</template>
          </el-table-column>
          <el-table-column prop="lineAmount" label="金额(元)" min-width="100" />
        </el-table>
      </template>
    </el-table-column>
    <el-table-column prop="installmentSummary" label="包含期数" min-width="120" />
    <el-table-column prop="invoiceAmount" label="合计金额(元)" min-width="120" />
    <el-table-column prop="status" label="状态" min-width="100">
      <template #default="{ row }">
        <el-tag :type="row.status === 'ykp' ? 'success' : 'warning'" size="small">
          {{ row.status === 'ykp' ? '已开票' : '待开票' }}
        </el-tag>
      </template>
    </el-table-column>
    <el-table-column prop="createTime" label="申请时间" min-width="160" />
    <el-table-column prop="financeHandleTime" label="财务处理时间" min-width="160" />
    <!-- ykp 行附件：解析 invoiceAttachments 逗号分隔，下载按钮 -->
  </el-table>

  <!-- 新增申请 -->
  <h4 class="detail-section-title">新增申请</h4>
  <template v-if="invoiceEligible.length">
    <div class="invoice-selected-summary">
      已选 {{ invoiceSelectedIds.length }} 期 · 合计 ¥{{ formatMoneyDisplay(invoiceSelectedTotal) }}
    </div>
    <el-table
      :data="invoiceEligible"
      @selection-change="(rows) => (invoiceSelectedIds = rows.map((r) => String(r.id)))"
    >
      <el-table-column type="selection" min-width="48" />
      <el-table-column prop="installmentNo" label="期数" min-width="80">
        <template #default="{ row }">{{ row.installmentNo ?? '趸交' }}</template>
      </el-table-column>
      <el-table-column prop="transferAmount" label="实缴金额(元)" min-width="120" />
      <el-table-column prop="transferTime" label="转账时间" min-width="160" />
    </el-table>
    <el-form label-position="top" class="form-grid-2 biz-dialog-form">
      <!-- 6 个选填开票字段 + applyRemark form-col-span-2 -->
    </el-form>
  </template>
  <div v-else class="finance-empty">暂无可开票缴费记录</div>

  <template #footer>
    <el-button class="btn-ghost" @click="invoiceDialogVisible = false">
      <el-icon><Close /></el-icon>取消
    </el-button>
    <el-button
      class="btn-main"
      type="primary"
      :loading="invoiceSubmitLoading"
      :disabled="!invoiceSelectedIds.length"
      @click="submitInvoiceRequest"
    >
      <el-icon><Check /></el-icon>提交申请
    </el-button>
  </template>
</el-dialog>
  • Step 4: 提交函数
async function submitInvoiceRequest() {
  if (!invoiceSelectedIds.value.length) return
  invoiceSubmitLoading.value = true
  try {
    await fetchCreateProtocolInvoiceRequest(invoiceProtocolId.value, {
      paymentDetailIds: invoiceSelectedIds.value,
      invoiceTitle: invoiceForm.invoiceTitle || undefined,
      taxNo: invoiceForm.taxNo || undefined,
      companyAddress: invoiceForm.companyAddress || undefined,
      companyPhone: invoiceForm.companyPhone || undefined,
      bankName: invoiceForm.bankName || undefined,
      bankAccount: invoiceForm.bankAccount || undefined,
      applyRemark: invoiceForm.applyRemark || undefined
    })
    Object.assign(invoiceForm, {
      invoiceTitle: '', taxNo: '', companyAddress: '', companyPhone: '',
      bankName: '', bankAccount: '', applyRemark: ''
    })
    await loadInvoiceDialogData()
  } finally {
    invoiceSubmitLoading.value = false
  }
}

引入 formatMoneyDisplay from @/utils/money-format。

Task 7: 财务开票 Tab

Files:

  • Create: web-admin/src/views/cell-storage/payment-confirm/InvoiceRequestTab.vue

  • Modify: web-admin/src/views/cell-storage/payment-confirm/index.vue

  • [ ] Step 1: 创建 InvoiceRequestTab.vue

从 RefundConfirmTab.vue 复制骨架，替换为：

区块 API status
待开票卡片 fetchProtocolInvoiceRequestList({ status: 'dkp' }) dkp
已开票表格 fetchProtocolInvoiceRequestList({ status: 'ykp' }) ykp

卡片展示：motherName、installmentSummary、invoiceAmount（合计）、createTime；按钮「处理开票」。

处理弹窗：

  • 只读子表 items（期数、lineAmount）
  • 只读开票信息字段（空显示「—」）
  • ProtocolFinanceConfirmFields：upload-folder="protocol-invoice-finance"
  • 提交调用 fetchCompleteProtocolInvoiceRequest(id, { invoiceAttachments, financeRemark })

已开票「查看」弹窗：只读 + 附件下载（downloadAttachmentFile）。

defineEmits(['action-done'])；完成/提交后 emit('action-done')。

  • Step 2: 修改 payment-confirm/index.vue
  1. Tab 按钮在退费后增加：
<button
  type="button"
  role="tab"
  class="panel-pill-tabs__btn"
  :class="{ 'is-active': mainTab === 'invoice' }"
  @click="mainTab = 'invoice'"
>
  开票申请
  <span v-if="invoicePendingBadge > 0" class="panel-pill-tabs__badge">
    {{ formatTabBadgeCount(invoicePendingBadge) }}
  </span>
</button>
  1. import InvoiceRequestTab from './InvoiceRequestTab.vue'
  2. const invoicePendingBadge = ref(0)
  3. loadTabPendingCounts 内增加：
const invRes = await fetchProtocolInvoicePendingCount()
invoicePendingBadge.value = Number((invRes as any)?.count ?? 0)

  1. 模板底部：<InvoiceRequestTab v-if="mainTab === 'invoice'" @action-done="loadTabPendingCounts" />

  2. onMounted / route.query.tab：支持 tab=invoice 直达（与现有 payment/refund 一致）。

Task 8: 联调与收尾

  • Step 1: 字段一致性检查
cd backend && source venv/bin/activate && python scripts/check_field_consistency.py

Expected: 无开票相关字段缺失告警（若脚本未覆盖新实体可跳过并记录）。

  • Step 2: 前端 lint
cd web-admin && pnpm lint

Expected: 无新增 error

  • Step 3: 更新知识图谱
./scripts/update-graph.sh all
  • Step 4: 手动验收清单（设计文档 §9）
步骤 操作 预期
1 满月待办 → 开票申请 → 多选 2 期提交 1 条 dkp，合计金额正确
2 弹窗申请记录展开 2 行子表明细
3 再次打开，已选期不可勾选 仅剩余期可选
4 财务 Tab 待开票 → 处理 → 上传附件 状态变 ykp
5 业务弹窗查看 可见附件
6 尝试重复含已开票期 后端拒绝
  • Step 5: 更新设计文档状态

将 docs/superpowers/specs/2026-06-17-protocol-invoice-request-design.md 首行 状态 改为 已实现。

Spec Coverage Self-Review

设计章节 对应 Task
§4.1 可开票条件 Task 3 list_invoice_eligible_payment_details
§4.2 合计金额 Task 3 create_invoice_request
§4.3 多期合并一张票 Task 3 主表+子表
§4.4 选填开票信息 Task 6 表单
§4.5 财务附件 Task 7 处理弹窗
§6 API 六端点 Task 4
§7.2 统一弹窗 Task 6
§7.3 财务 Tab Task 7
§9 测试要点 Task 8 Step 4

无 TBD / 占位步骤。

执行方式

Plan complete and saved to docs/superpowers/plans/2026-06-17-protocol-invoice-request.md.

两种执行选项：

  1. Subagent-Driven（推荐） — 按 Task 1→8 逐任务派发子代理，任务间人工/助手复核
  2. Inline Execution — 本会话内按 Task 顺序直接实现，每完成 2～3 个 Task 汇报一次

请选择执行方式；若无偏好，默认 Inline Execution 从 Task 1 开始。