feat(pl): add archive zip naming helpers and planId on export rows.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
wuzilongandCursor committed 2026-09-23 12:40:04 +08:00
1 parent 1c3c0d40b6
commit e9ffa5d042
3 files changed
+91

No files matched your search

@@ -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 ""),
@@ -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}")'
@@ -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