From 1c3c0d40b622a9d05b818013ab8492b56414397b Mon Sep 17 00:00:00 2001 From: wuzilong <704541015@qq.com> Date: Wed, 23 Sep 2026 12:38:51 +0800 Subject: [PATCH] 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 --- ...3-pl-plan-archive-attachment-zip-export.md | 659 ++++++++++++++++++ 1 file changed, 659 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-23-pl-plan-archive-attachment-zip-export.md diff --git a/docs/superpowers/plans/2026-09-23-pl-plan-archive-attachment-zip-export.md b/docs/superpowers/plans/2026-09-23-pl-plan-archive-attachment-zip-export.md new file mode 100644 index 000000000..a4a7dac8c --- /dev/null +++ b/docs/superpowers/plans/2026-09-23-pl-plan-archive-attachment-zip-export.md @@ -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 +}> +``` + +模板在原有两项后： + +```vue + +``` + +`handleCommand`：若命中 `additionalMenuItems`，用 `withIds` + `payloadExtra` 调用 `runExport`。 + +将 `runExport` 改为接受可选 `payloadExtra`： + +```ts +async function runExport(withIds: boolean, payloadExtra?: Record) { + ... + 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 + +``` + +- [ ] **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`。