From e9ffa5d04240c336a73de2461d7815b765c4eb04 Mon Sep 17 00:00:00 2001
From: wuzilong <704541015@qq.com>
Date: Wed, 23 Sep 2026 12:40:04 +0800
Subject: [PATCH] feat(pl): add archive zip naming helpers and planId on export
 rows.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../app/services/pl_plan_archive_export.py    |  1 +
 .../services/pl_plan_archive_zip_export.py    | 57 +++++++++++++++++++
 .../tests/test_pl_plan_archive_zip_export.py  | 33 +++++++++++
 3 files changed, 91 insertions(+)
 create mode 100644 backend/app/services/pl_plan_archive_zip_export.py
 create mode 100644 backend/tests/test_pl_plan_archive_zip_export.py

diff --git a/backend/app/services/pl_plan_archive_export.py b/backend/app/services/pl_plan_archive_export.py
index 5fdc54377..705ecdf09 100644
--- a/backend/app/services/pl_plan_archive_export.py
+++ b/backend/app/services/pl_plan_archive_export.py
@@ -131,6 +131,7 @@ def _plan_common_export_fields(
         freight_amount = transport.freight_amount
     plan_date = _format_date_value(plan.plan_date)
     return {
+        "planId": plan.id or "",
         "planNo": plan.plan_no or "",
         "planDate": plan_date,
         "archiveStatus": ARCHIVE_STATUS_LABELS.get(plan.archive_status or "", plan.archive_status or ""),
diff --git a/backend/app/services/pl_plan_archive_zip_export.py b/backend/app/services/pl_plan_archive_zip_export.py
new file mode 100644
index 000000000..bc70b326b
--- /dev/null
+++ b/backend/app/services/pl_plan_archive_zip_export.py
@@ -0,0 +1,57 @@
+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:
+    path = str(rel_path).replace('"', '""')
+    text = sanitize_zip_segment(display).replace('"', '""')
+    return f'=HYPERLINK("{path}","{text}")'
diff --git a/backend/tests/test_pl_plan_archive_zip_export.py b/backend/tests/test_pl_plan_archive_zip_export.py
new file mode 100644
index 000000000..100affd49
--- /dev/null
+++ b/backend/tests/test_pl_plan_archive_zip_export.py
@@ -0,0 +1,33 @@
+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