feat(pl): collect hazard certs for archive zip export.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/pl_plan_archive_zip_export.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
 import hashlib
+import json
 import re
-from typing import Any, Mapping, Sequence
+from typing import Any, Mapping, Optional, Sequence
 from urllib.parse import unquote, urlparse
 
 ARCHIVE_ZIP_ATTACH_COL_MAX = 20
@@ -55,3 +56,70 @@ def build_hyperlink_formula(rel_path: str, display: str) -> str:
     path = str(rel_path).replace('"', '""')
     text = sanitize_zip_segment(display).replace('"', '""')
     return f'=HYPERLINK("{path}","{text}")'
+
+
+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
+
+
+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

backend/tests/test_pl_plan_archive_zip_export.py

@@ -3,6 +3,9 @@ from app.services.pl_plan_archive_zip_export import (
     build_attachment_filename,
     first_row_no_by_plan_id,
     build_hyperlink_formula,
+    parse_url_list,
+    collect_hazard_cert_items,
+    merge_plan_attachment_items,
 )
 
 
@@ -30,3 +33,30 @@ def test_build_hyperlink_formula():
     f = build_hyperlink_formula("附件/01_司机_ab12.jpg", "司机")
     assert f.startswith('=HYPERLINK(')
     assert "附件/01_司机_ab12.jpg" in f
+
+
+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"] == "司机"