shanxia2.0/docs/superpowers/plans/2026-09-23-pl-plan-archive-attachment-zip-export.md

运输计划归档 · 数据及附件 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: 写失败测试

# 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: 跑测确认失败

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 返回值中增加：

"planId": plan.id or "",

（勿加入 PLAN_ARCHIVE_EXPORT_COLUMNS，xlsx 不会出现该列。）

pl_plan_archive_zip_export.py 最小实现：

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: 跑测确认通过

cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py -v

Expected: PASS

• Step 5: Commit

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: 写失败测试

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: 跑测确认失败

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: 实现收集逻辑

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）：

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: 跑测确认通过

cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py -v

Expected: PASS

• Step 5: Commit

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: 写失败测试

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: 跑测确认失败

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 默认：

def _default_download(url: str) -> bytes:
    from app.services.bx.storage import download_file_bytes
    return download_file_bytes(url)

签名建议：

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: 跑测确认通过

cd backend && source venv/bin/activate && pytest tests/test_pl_plan_archive_zip_export.py -v

Expected: PASS

• Step 5: Commit

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 共享查计划逻辑）：

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: 跑相关单测

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

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

additionalMenuItems?: Array<{
  command: string
  label: string
  withIds?: boolean
  disabled?: boolean
  payloadExtra?: Record<string, any>
}>

模板在原有两项后：

<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：

async function runExport(withIds: boolean, payloadExtra?: Record<string, any>) {
  ...
  const payload = { ...buildPayload(withIds), ...(payloadExtra || {}) }
  ...
}

默认 filename 仍可为 .xlsx；ZIP 响应的 Content-Disposition 会覆盖（已有 filenameFromContentDisposition）。

• Step 2: 归档页传入附加项

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）。

<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

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 执行：

./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。