# 采集协议单开票申请 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 一致）** ```sql -- 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: 本地库执行迁移（开发者环境）** ```bash 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 类** ```python 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** ```bash 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** ```python # 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** ```bash 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: 实现辅助函数** ```python 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 区增加： ```python from app.models.models import ( ... BizProtocolInvoiceRequest, BizProtocolInvoiceRequestItem, ) ``` - [ ] **Step 4: 运行 TestInstallmentSummary 通过** ```bash pytest tests/test_protocol_invoice_request.py::TestInstallmentSummary -v ``` Expected: 3 passed ### 3.2 可开票明细 & 创建申请 - [ ] **Step 5: 写失败测试 — 重复 payment_detail_id 拒绝** ```python 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`） ```python 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 子表）** ```python 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: 实现列表/完成/计数函数** ```python 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: 运行全部开票测试** ```bash 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 风格一致）** ```python 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 个端点** ```python @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 已迁移且后端运行）** ```bash 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 个函数** ```typescript export function fetchInvoiceEligiblePayments(protocolId: string) { return request.get<{ list: Record[]; total: number }>({ url: `/web/admin/collection-protocols/${protocolId}/invoice-eligible-payments` }) } export function fetchProtocolInvoiceRequests(protocolId: string) { return request.get<{ list: Record[]; 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>>({ 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: 操作列增加「开票申请」按钮** 在「办理保险」后增加： ```vue 开票申请 ``` `import { Tickets } from '@element-plus/icons-vue'`；操作列 `min-width` 调至约 `320`。 - [ ] **Step 2: 新增弹窗 state 与加载函数** ```typescript const invoiceDialogVisible = ref(false) const invoiceProtocolId = ref('') const invoiceSummaryRow = ref(null) const invoiceHistory = ref([]) const invoiceEligible = ref([]) const invoiceSelectedIds = ref([]) 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: 弹窗模板（三块结构）** ```vue
{{ invoiceSummaryRow?.motherName || '—' }} 协议号 {{ invoiceSummaryRow?.protocolNo || '—' }}

申请记录

新增申请

暂无可开票缴费记录
 ``` - [ ] **Step 4: 提交函数** ```typescript 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 按钮在退费后增加： ```vue ``` 2. `import InvoiceRequestTab from './InvoiceRequestTab.vue'` 3. `const invoicePendingBadge = ref(0)` 4. `loadTabPendingCounts` 内增加： ```typescript const invRes = await fetchProtocolInvoicePendingCount() invoicePendingBadge.value = Number((invRes as any)?.count ?? 0) ``` 5. 模板底部：`` 6. `onMounted` / `route.query.tab`：支持 `tab=invoice` 直达（与现有 `payment`/`refund` 一致）。 --- ## Task 8: 联调与收尾 - [ ] **Step 1: 字段一致性检查** ```bash cd backend && source venv/bin/activate && python scripts/check_field_consistency.py ``` Expected: 无开票相关字段缺失告警（若脚本未覆盖新实体可跳过并记录）。 - [ ] **Step 2: 前端 lint** ```bash cd web-admin && pnpm lint ``` Expected: 无新增 error - [ ] **Step 3: 更新知识图谱** ```bash ./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 开始。