docs: add implementation plan for plan-archive attachment ZIP export.

Break work into TDD tasks for zip helpers, hazard cert collection, API wiring, and ListExportDropdown menu items.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
wuzilongandCursor committed 2026-09-23 12:38:51 +08:00
1 parent f5b76ef711
commit 1c3c0d40b6
1 file changed
+659
@@ -0,0 +1,659 @@
# 运输计划归档 · 数据及附件 ZIP 导出 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:** 在运输计划归档页导出下拉中新增「数据及附件 ZIP」两项,后端一站式打包归档 Excel + 单据附件 + 危化品证,目录对齐参考 ZIP,Excel 用相对路径超链接。
**Architecture:** 扩展 `POST /transport/plan/export`(`exportFormat=archive_zip`)。服务层收集附件、下载存储对象、生成带 `HYPERLINK` 的 xlsx、打 zip。前端扩展 `ListExportDropdown` 菜单项;纯 Excel 路径不变。
**Tech Stack:** FastAPI、openpyxl、zipfile、`download_file_bytes`(`app.services.bx.storage`)、Vue 3、Element Plus、pytest
**Spec:** `docs/superpowers/specs/2026-09-23-pl-plan-archive-attachment-zip-export-design.md`
---
## 文件结构
| 操作 | 文件 | 职责 |
|------|------|------|
| Create | `backend/app/services/pl_plan_archive_zip_export.py` | 命名/序号映射、收集危化证、下载、写 xlsx 超链接、组 ZIP、返回 StreamingResponse |
| Create | `backend/tests/test_pl_plan_archive_zip_export.py` | 纯函数与打包逻辑单测(mock 下载) |
| Modify | `backend/app/services/pl_plan_archive_export.py` | `plan_common` 增加内部字段 `planId`(不进导出列定义) |
| Modify | `backend/app/api/pl/plan.py` | `export_plans` 增加 `archive_zip` 分支 |
| Modify | `web-admin/src/components/business/list-export-dropdown/index.vue` | 可选附加菜单项(ZIP) |
| Modify | `web-admin/src/views/pl/plan-archive/index.vue` | 传入 ZIP 两项菜单 |
**复用(不改行为):**
- `build_plan_archive_export_rows`、`PLAN_ARCHIVE_EXPORT_COLUMNS`、`load_plan_document_attachments_by_plan_ids`
- `export_rows_to_xlsx_response` / `DictLabelResolver` / `ExportColumn`(ZIP 内 xlsx 可复用 resolver + 自建 workbook)
- `download_file_bytes` / `normalize_storage_path`
**常量(实现时写入 zip 服务模块):**
- `ARCHIVE_ZIP_ATTACH_COL_MAX = 20`
- `ARCHIVE_ZIP_MAX_FILES = 2000`
- `ARCHIVE_ZIP_MAX_BYTES = 500 * 1024 * 1024`
---
### Task 1: 导出行带上 planId + 纯函数(命名 / 序号映射)
**Files:**
- Modify: `backend/app/services/pl_plan_archive_export.py`(`_plan_common_export_fields`)
- Create: `backend/app/services/pl_plan_archive_zip_export.py`(先只放纯函数)
- Create: `backend/tests/test_pl_plan_archive_zip_export.py`
- [ ] **Step 1: 写失败测试**
```python
# backend/tests/test_pl_plan_archive_zip_export.py
from app.services.pl_plan_archive_zip_export import (
sanitize_zip_segment,
short_hash,
build_attachment_filename,
first_row_no_by_plan_id,
build_hyperlink_formula,
)
def test_sanitize_zip_segment():
assert sanitize_zip_segment("装货·司机/A") == "装货·司机_A"
assert sanitize_zip_segment(" a ") == "a"
def test_build_attachment_filename():
name = build_attachment_filename(1, "装货·司机", "https://cdn.example/path/a.jpg")
assert name.startswith("01_装货·司机_")
assert name.endswith(".jpg")
def test_first_row_no_by_plan_id():
rows = [
{"planId": "p1", "rowNo": 1},
{"planId": "p1", "rowNo": 2},
{"planId": "p2", "rowNo": 3},
]
assert first_row_no_by_plan_id(rows) == {"p1": 1, "p2": 3}
def test_build_hyperlink_formula():
f = build_hyperlink_formula("附件/01_司机_ab12.jpg", "司机")
assert f.startswith('=HYPERLINK(')
assert "附件/01_司机_ab12.jpg" in f
```
- [ ] **Step 2: 跑测确认失败**
```bash
cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py -v
```
Expected: FAIL(模块/符号不存在)
- [ ] **Step 3: 实现纯函数 + planId**
在 `_plan_common_export_fields` 返回值中增加:
```python
"planId": plan.id or "",
```
(勿加入 `PLAN_ARCHIVE_EXPORT_COLUMNS`,xlsx 不会出现该列。)
`pl_plan_archive_zip_export.py` 最小实现:
```python
from __future__ import annotations
import hashlib
import re
from typing import Any, Mapping, Optional, Sequence
from urllib.parse import unquote, urlparse
ARCHIVE_ZIP_ATTACH_COL_MAX = 20
ARCHIVE_ZIP_MAX_FILES = 2000
ARCHIVE_ZIP_MAX_BYTES = 500 * 1024 * 1024
def sanitize_zip_segment(raw: str) -> str:
value = re.sub(r'[\\/:*?"<>|]+', "_", str(raw or "").strip())
value = re.sub(r"\s+", " ", value).strip()
return value or "未命名"
def short_hash(text: str, *, n: int = 6) -> str:
return hashlib.sha1(str(text or "").encode("utf-8")).hexdigest()[:n]
def guess_ext_from_url(url: str) -> str:
path = unquote(urlparse(str(url or "").split("?", 1)[0]).path)
name = path.rsplit("/", 1)[-1]
if "." not in name:
return "bin"
ext = name.rsplit(".", 1)[-1].lower()
if not ext or len(ext) > 8 or "/" in ext:
return "bin"
return ext
def build_attachment_filename(row_no: int, label: str, url: str) -> str:
return (
f"{int(row_no):02d}_{sanitize_zip_segment(label)}_"
f"{short_hash(url)}.{guess_ext_from_url(url)}"
)
def first_row_no_by_plan_id(rows: Sequence[Mapping[str, Any]]) -> dict[str, int]:
out: dict[str, int] = {}
for row in rows:
plan_id = str(row.get("planId") or "").strip()
if not plan_id or plan_id in out:
continue
try:
out[plan_id] = int(row.get("rowNo") or 0)
except (TypeError, ValueError):
out[plan_id] = 0
return out
def build_hyperlink_formula(rel_path: str, display: str) -> str:
# Excel 公式内双引号转义
path = str(rel_path).replace('"', '""')
text = sanitize_zip_segment(display).replace('"', '""')
return f'=HYPERLINK("{path}","{text}")'
```
- [ ] **Step 4: 跑测确认通过**
```bash
cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py -v
```
Expected: PASS
- [ ] **Step 5: Commit**
```bash
git add backend/app/services/pl_plan_archive_export.py \
backend/app/services/pl_plan_archive_zip_export.py \
backend/tests/test_pl_plan_archive_zip_export.py
git commit -m "feat(pl): add archive zip naming helpers and planId on export rows."
```
---
### Task 2: 收集危化证 + 合并单据附件条目
**Files:**
- Modify: `backend/app/services/pl_plan_archive_zip_export.py`
- Modify: `backend/tests/test_pl_plan_archive_zip_export.py`
- [ ] **Step 1: 写失败测试**
```python
from app.services.pl_plan_archive_zip_export import (
parse_url_list,
collect_hazard_cert_items,
merge_plan_attachment_items,
)
def test_parse_url_list_csv_and_json():
assert parse_url_list("a.jpg,b.jpg") == ["a.jpg", "b.jpg"]
assert parse_url_list('["x.png","y.png"]') == ["x.png", "y.png"]
def test_collect_hazard_cert_items_labels():
class P:
product_name = "甲醇"
purchase_cert_url = "https://x/p.pdf"
transport_cert_url = None
legal_use_cert_url = "https://x/l1.pdf,https://x/l2.pdf"
items = collect_hazard_cert_items([P()])
labels = [i["label"] for i in items]
assert "购买证" in labels[0]
assert any("合法使用证明" in x for x in labels)
assert len(items) == 3
def test_merge_dedupes_url():
doc = [{"url": "https://a/1.jpg", "label": "司机"}]
haz = [{"url": "https://a/1.jpg", "label": "购买证"}]
merged = merge_plan_attachment_items(doc, haz)
assert len(merged) == 1
assert merged[0]["label"] == "司机"
```
- [ ] **Step 2: 跑测确认失败**
```bash
cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py::test_parse_url_list_csv_and_json tests/test_pl_plan_archive_zip_export.py::test_collect_hazard_cert_items_labels tests/test_pl_plan_archive_zip_export.py::test_merge_dedupes_url -v
```
Expected: FAIL
- [ ] **Step 3: 实现收集逻辑**
```python
import json
from typing import Any, Optional
def parse_url_list(raw: Optional[str]) -> list[str]:
if not raw:
return []
text = str(raw).strip()
if not text:
return []
try:
parsed = json.loads(text)
if isinstance(parsed, list):
return [str(x).strip() for x in parsed if str(x).strip()]
except json.JSONDecodeError:
pass
return [p.strip() for p in text.split(",") if p.strip()]
def collect_hazard_cert_items(products: Sequence[Any]) -> list[dict[str, str]]:
out: list[dict[str, str]] = []
for product in products or []:
pname = sanitize_zip_segment(getattr(product, "product_name", None) or "")
specs = (
("购买证", getattr(product, "purchase_cert_url", None)),
("运输证", getattr(product, "transport_cert_url", None)),
("合法使用证明", getattr(product, "legal_use_cert_url", None)),
)
for base_label, raw in specs:
for url in parse_url_list(raw):
label = base_label if not pname or pname == "未命名" else f"{base_label}·{pname}"
out.append({"url": url, "label": label})
return out
def merge_plan_attachment_items(
document_items: Sequence[Mapping[str, str]],
hazard_items: Sequence[Mapping[str, str]],
) -> list[dict[str, str]]:
seen: set[str] = set()
out: list[dict[str, str]] = []
for item in list(document_items or []) + list(hazard_items or []):
url = str(item.get("url") or "").strip()
if not url or url in seen:
continue
seen.add(url)
out.append({"url": url, "label": str(item.get("label") or "附件")})
return out
```
另增 DB 加载函数(可同文件,单测可后接集成或 mock Session):
```python
def load_plan_products_by_plan_ids(db, *, tenant_no: str, plan_ids: list[str]):
from app.models.pl_models import PlPlanProduct
if not plan_ids:
return {}
rows = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.plan_id.in_(plan_ids),
PlPlanProduct.tenant_no == tenant_no,
PlPlanProduct.del_flag == 0,
)
.order_by(PlPlanProduct.order_num, PlPlanProduct.create_time)
.all()
)
grouped: dict[str, list] = {pid: [] for pid in plan_ids}
for row in rows:
grouped.setdefault(row.plan_id, []).append(row)
return grouped
```
- [ ] **Step 4: 跑测确认通过**
```bash
cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py -v
```
Expected: PASS
- [ ] **Step 5: Commit**
```bash
git add backend/app/services/pl_plan_archive_zip_export.py backend/tests/test_pl_plan_archive_zip_export.py
git commit -m "feat(pl): collect hazard certs for archive zip export."
```
---
### Task 3: 构建带超链接的 workbook + ZIP 字节
**Files:**
- Modify: `backend/app/services/pl_plan_archive_zip_export.py`
- Modify: `backend/tests/test_pl_plan_archive_zip_export.py`
- [ ] **Step 1: 写失败测试**
```python
import io
import zipfile
from unittest.mock import patch
from openpyxl import load_workbook
from app.utils.excel_export import ExportColumn
from app.services.pl_plan_archive_zip_export import build_archive_zip_bytes
def test_build_archive_zip_bytes_structure_and_hyperlink():
columns = [
ExportColumn(title="序号", field="rowNo"),
ExportColumn(title="计划编号", field="planNo"),
]
rows = [
{"planId": "p1", "rowNo": 1, "planNo": "PL001"},
{"planId": "p1", "rowNo": 2, "planNo": "PL001"},
]
attachments_by_plan = {
"p1": [{"url": "https://cdn.example/a.jpg", "label": "司机"}],
}
def fake_download(url: str) -> bytes:
return b"fake-image-bytes"
stamp = "202609231230"
data, failures = build_archive_zip_bytes(
db=None, # resolver 无 dict 列时可允许 None;若实现要求 Session,传入 MagicMock
tenant_no="t1",
columns=columns,
rows=rows,
attachments_by_plan=attachments_by_plan,
stamp=stamp,
download_fn=fake_download,
)
assert failures == []
with zipfile.ZipFile(io.BytesIO(data)) as zf:
names = zf.namelist()
assert f"计划归档({stamp})/(总表)计划归档.xlsx" in names
att = [n for n in names if n.startswith(f"计划归档({stamp})/附件/")]
assert len(att) == 1
assert "01_司机_" in att[0]
xlsx_bytes = zf.read(f"计划归档({stamp})/(总表)计划归档.xlsx")
wb = load_workbook(io.BytesIO(xlsx_bytes))
ws = wb.active
# 表头含 附件_1;首行有公式;第二行空
headers = [c.value for c in ws[1]]
assert "附件_1" in headers
attach_col = headers.index("附件_1") + 1
assert str(ws.cell(2, attach_col).value).startswith("=HYPERLINK(")
assert ws.cell(3, attach_col).value in (None, "")
```
- [ ] **Step 2: 跑测确认失败**
```bash
cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py::test_build_archive_zip_bytes_structure_and_hyperlink -v
```
Expected: FAIL
- [ ] **Step 3: 实现 `build_archive_zip_bytes` 与响应封装**
要点:
1. `first_row_no_by_plan_id(rows)` 得序号。
2. 遍历计划附件,`download_fn(url)`;失败记入 `failures`;成功写入 `files: list[tuple[zip_path, bytes]]`。
3. 校验 `ARCHIVE_ZIP_MAX_FILES` / `ARCHIVE_ZIP_MAX_BYTES`,超限抛 `ExportValidationError`。
4. `K = min(ARCHIVE_ZIP_ATTACH_COL_MAX, max(len(atts) for atts in ...), default 0)`;为每行填充 `attach1`… 字段(仅首行)。
5. 用 `DictLabelResolver` + openpyxl 写表;附件列写入公式字符串(openpyxl 会当公式)。
6. `zipfile.ZipFile` 写入 root 下 xlsx + 附件;若 `failures` 非空写 `导出说明.txt`。
7. `build_archive_zip_response(...)` → `StreamingResponse`,`media_type=application/zip`,`filename*=UTF-8''计划归档({stamp}).zip`。
`download_fn` 默认:
```python
def _default_download(url: str) -> bytes:
from app.services.bx.storage import download_file_bytes
return download_file_bytes(url)
```
签名建议:
```python
def build_archive_zip_bytes(
*,
db: Session,
tenant_no: str,
columns: Sequence[ExportColumn],
rows: Sequence[Mapping[str, Any]],
attachments_by_plan: Mapping[str, Sequence[Mapping[str, str]]],
stamp: str,
download_fn=None,
) -> tuple[bytes, list[str]]:
...
def build_archive_zip_streaming_response(...) -> StreamingResponse:
...
```
若 `db=None` 不便,测试传入 `MagicMock()`,且 columns 无 `dict_type`。
- [ ] **Step 4: 跑测确认通过**
```bash
cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py -v
```
Expected: PASS
- [ ] **Step 5: Commit**
```bash
git add backend/app/services/pl_plan_archive_zip_export.py backend/tests/test_pl_plan_archive_zip_export.py
git commit -m "feat(pl): build archive export zip with relative hyperlinks."
```
---
### Task 4: 接入 `export_plans` 的 `archive_zip` 分支
**Files:**
- Modify: `backend/app/api/pl/plan.py`(`export_plans`)
- [ ] **Step 1: 在 `is_archive_export` 旁增加 zip 分支**
伪代码(插入现有 archive 分支逻辑之后、else 之前;或与 archive 共享查计划逻辑):
```python
is_archive_zip = (body.export_format or "").strip().lower() == "archive_zip"
# ... 现有 query / plans 加载不变 ...
if len(plans) > EXPORT_MAX_ROWS:
raise ExportValidationError(f"导出超过 {EXPORT_MAX_ROWS} 条,请缩小筛选范围")
if not plans:
raise ExportValidationError("没有可导出的数据")
if is_archive_zip or is_archive_export:
transport_ids = [plan.transport_id for plan in plans if plan.transport_id]
transports_by_id = _load_transports_by_id(db, tenant_no, transport_ids)
items = build_plan_archive_export_rows(...)
cols = filter_export_columns(PLAN_ARCHIVE_EXPORT_COLUMNS, body.columns)
if not user_can_view_archive_prices(db, user):
cols = [c for c in cols if c.field not in ARCHIVE_PRICE_EXPORT_FIELDS]
if is_archive_zip:
from datetime import datetime
from app.services.pl_plan_list_attachments import load_plan_document_attachments_by_plan_ids
from app.services.pl_plan_archive_zip_export import (
build_archive_zip_streaming_response,
collect_hazard_cert_items,
load_plan_products_by_plan_ids,
merge_plan_attachment_items,
)
plan_ids = [p.id for p in plans]
docs = load_plan_document_attachments_by_plan_ids(db, tenant_no=tenant_no, plan_ids=plan_ids)
products_by_plan = load_plan_products_by_plan_ids(db, tenant_no=tenant_no, plan_ids=plan_ids)
attachments_by_plan = {}
for pid in plan_ids:
attachments_by_plan[pid] = merge_plan_attachment_items(
docs.get(pid, []),
collect_hazard_cert_items(products_by_plan.get(pid, [])),
)
stamp = datetime.now().strftime("%Y%m%d%H%M")
return build_archive_zip_streaming_response(
db=db,
tenant_no=tenant_no,
columns=cols,
rows=items,
attachments_by_plan=attachments_by_plan,
stamp=stamp,
)
# 原 archive xlsx return ...
```
同步更新 `PlanExportBody.export_format` 的 description:`archive | archive_zip`。
- [ ] **Step 2: 跑相关单测**
```bash
cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py tests/test_pl_plan_archive_export.py -v
```
Expected: PASS
- [ ] **Step 3: Commit**
```bash
git add backend/app/api/pl/plan.py
git commit -m "feat(pl): wire archive_zip export format on plan export API."
```
---
### Task 5: 前端下拉增加 ZIP 两项
**Files:**
- Modify: `web-admin/src/components/business/list-export-dropdown/index.vue`
- Modify: `web-admin/src/views/pl/plan-archive/index.vue`
- [ ] **Step 1: 扩展 `ListExportDropdown` props**
```ts
additionalMenuItems?: Array<{
command: string
label: string
withIds?: boolean
disabled?: boolean
payloadExtra?: Record<string, any>
}>
```
模板在原有两项后:
```vue
<template v-for="item in additionalMenuItems" :key="item.command">
<el-dropdown-item
:command="item.command"
:disabled="item.disabled || (item.withIds && !selectedCount)"
>
{{ item.label }}
</el-dropdown-item>
</template>
```
`handleCommand`:若命中 `additionalMenuItems`,用 `withIds` + `payloadExtra` 调用 `runExport`。
将 `runExport` 改为接受可选 `payloadExtra`:
```ts
async function runExport(withIds: boolean, payloadExtra?: Record<string, any>) {
...
const payload = { ...buildPayload(withIds), ...(payloadExtra || {}) }
...
}
```
默认 `filename` 仍可为 `.xlsx`;ZIP 响应的 `Content-Disposition` 会覆盖(已有 `filenameFromContentDisposition`)。
- [ ] **Step 2: 归档页传入附加项**
```ts
const archiveZipMenuItems = [
{
command: 'filtered-zip',
label: '导出筛选结果(数据及附件 ZIP)',
withIds: false,
payloadExtra: { exportFormat: 'archive_zip' }
},
{
command: 'selected-zip',
label: '导出已选(数据及附件 ZIP)',
withIds: true,
payloadExtra: { exportFormat: 'archive_zip' }
}
]
```
`getPlanArchiveExportFilterParams` 对 Excel 仍带 `exportFormat: 'archive'`;ZIP 项用 `payloadExtra` **覆盖**为 `archive_zip`(确保 `buildPayload` 先展开 filter,再展开 extra)。
```vue
<ListExportDropdown
:export-fn="planApi.export"
:get-filter-params="getPlanArchiveExportFilterParams"
:has-selection="true"
:selected-ids="selectedRows.map((r) => r.id)"
:additional-menu-items="archiveZipMenuItems"
filename="计划归档.xlsx"
/>
```
- [ ] **Step 3: 本地手测清单(实现者勾选)**
1. Excel 两项仍下 xlsx。
2. ZIP 两项下 zip;解压后打开 xlsx,点击附件列能打开文件。
3. 有危化证的计划包内可见购买证/运输证/合法使用证明。
4. 无附件计划仍能下 zip。
- [ ] **Step 4: Commit**
```bash
git add web-admin/src/components/business/list-export-dropdown/index.vue \
web-admin/src/views/pl/plan-archive/index.vue
git commit -m "feat(web-admin): add archive data-and-attachments ZIP export menu."
```
---
### Task 6: 图谱更新(若仓库要求)
- [ ] **Step 1:** 若本会话改了 backend/web-admin 代码,按 AGENTS.md 执行:
```bash
./scripts/update-graph.sh backend
# 若改了 web-admin:
./scripts/update-graph.sh frontend
```
或根目录 `graphify update .`(以仓库当前脚本为准)。
- [ ] **Step 2:** 如有 graph 产物变更且需入库,单独 commit:`chore: update graphify after archive zip export.`
---
## Spec 覆盖自检
| Spec 要求 | Task |
|-----------|------|
| 四项菜单并存 | Task 5 |
| ZIP=xlsx+附件 | Task 3–4 |
| 筛选/已选 | Task 5 |
| 参考包目录结构 | Task 3 |
| HYPERLINK 路径引用 | Task 3 |
| 单据 6 类 | Task 4(复用 list attachments) |
| 危化三证 | Task 2–4 |
| 首行序号挂载 / 附件_1..K / 上限 20 | Task 1 + 3 |
| 失败写说明 / 软上限 | Task 3 |
| 纯 Excel 不变 | Task 4 分支隔离 |
| 后端一站式 | Task 4 |
占位符扫描:无 TBD/TODO。命名在各 Task 一致:`build_archive_zip_bytes`、`exportFormat=archive_zip`、`additionalMenuItems`。