283 lines
10 KiB
Python
283 lines
10 KiB
Python
"""
|
||
按海运询价单标签补写协同字段。
|
||
|
||
主账填表只认英文码，货好时间 / 海关编码 / 含油电磁常空着。
|
||
本文件只改 Excel 字节，不打主账、不占回调线程。
|
||
|
||
注意：openpyxl 保存会弄坏海运顶部 Form Control（drawing/VML/sheet1 锚点）。
|
||
因此只用来「读」标签位置；真正写入走 zip 内定点改格，不动控件骨架。
|
||
无待写字段、或目标格已是目标值时原样返回。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any, Mapping
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 长的「含油,电,磁」必须排在「是否含油」前面，避免只写到油。
|
||
_COLLAB_LABEL_MARKERS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
||
("是否含油,电,磁", ("oil_elec_mag", "是否含油,电,磁")),
|
||
("OIL,ELEC,MAG", ("oil_elec_mag", "是否含油,电,磁")),
|
||
("贸易条款", ("贸易条款", "trade_terms")),
|
||
("货好时间", ("货好时间", "cargo_ready_date")),
|
||
("商品海关编码", ("商品海关编码", "HS编码", "海关编码", "hs_code")),
|
||
("HS Code", ("商品海关编码", "HS编码", "海关编码", "hs_code")),
|
||
("是否含油", ("是否含油", "has_oil")),
|
||
("是否含电", ("是否含电", "has_battery")),
|
||
("是否含磁", ("是否含磁", "has_magnet")),
|
||
("客户名称", ("客户名称", "customer_name")),
|
||
("货值", ("货值", "cargo_value")),
|
||
("是否为危险品", ("是否为危险品", "is_dangerous")),
|
||
("包装方式", ("包装方式", "包装类型")),
|
||
)
|
||
|
||
# 旧路径兜底：若仍误走 openpyxl 保存，至少盖回这些部件
|
||
_PRESERVE_ZIP_PARTS = (
|
||
"xl/drawings/drawing1.xml",
|
||
"xl/drawings/vmlDrawing1.vml",
|
||
"xl/styles.xml",
|
||
"xl/ctrlProps/ctrlProp1.xml",
|
||
"xl/ctrlProps/ctrlProp2.xml",
|
||
"xl/worksheets/sheet1.xml",
|
||
"xl/worksheets/_rels/sheet1.xml.rels",
|
||
"[Content_Types].xml",
|
||
)
|
||
|
||
|
||
def _fact_for_markers(facts: Mapping[str, Any], keys: tuple[str, ...]) -> str:
|
||
for key in keys:
|
||
val = str((facts or {}).get(key) or "").strip()
|
||
if val and val != "-":
|
||
return val
|
||
return ""
|
||
|
||
|
||
def _restore_sea_draw_parts(original: bytes, rewritten: bytes) -> bytes:
|
||
"""把原件里的 drawing/VML/ctrl/styles/sheet1 控件骨架盖回 openpyxl 重写后的 xlsx。"""
|
||
import io
|
||
import zipfile
|
||
|
||
try:
|
||
with zipfile.ZipFile(io.BytesIO(original)) as zin:
|
||
keep = {
|
||
name: zin.read(name)
|
||
for name in _PRESERVE_ZIP_PARTS
|
||
if name in zin.namelist()
|
||
}
|
||
if not keep:
|
||
return rewritten
|
||
out = io.BytesIO()
|
||
with zipfile.ZipFile(io.BytesIO(rewritten)) as zin, zipfile.ZipFile(
|
||
out, "w", compression=zipfile.ZIP_DEFLATED
|
||
) as zout:
|
||
for info in zin.infolist():
|
||
data = keep.get(info.filename) or zin.read(info.filename)
|
||
zout.writestr(info, data)
|
||
for name, data in keep.items():
|
||
if name not in zin.namelist():
|
||
zout.writestr(name, data)
|
||
return out.getvalue()
|
||
except Exception:
|
||
logger.exception("overlay 盖回海运 drawing 失败，退回 openpyxl 结果")
|
||
return rewritten
|
||
|
||
|
||
def _col_letter(col: int) -> str:
|
||
"""1-based 列号 → Excel 列字母。"""
|
||
n = int(col)
|
||
letters: list[str] = []
|
||
while n > 0:
|
||
n, rem = divmod(n - 1, 26)
|
||
letters.append(chr(65 + rem))
|
||
return "".join(reversed(letters)) or "A"
|
||
|
||
|
||
def _xml_escape(text: str) -> str:
|
||
return (
|
||
str(text)
|
||
.replace("&", "&")
|
||
.replace("<", "<")
|
||
.replace(">", ">")
|
||
.replace('"', """)
|
||
)
|
||
|
||
|
||
def _patch_cells_keep_structure(data: bytes, cells: list[tuple[int, int, str]]) -> bytes:
|
||
"""
|
||
只改 sheet1 里指定格的字符串，保留 AlternateContent / 控件关系。
|
||
|
||
禁止整表经 openpyxl 重存：否则拉群协同 Excel 顶部 Port/其它单选会空白，
|
||
而「直接生成」不经本函数所以看起来正常（WO031 vs WO033）。
|
||
"""
|
||
import io
|
||
import zipfile
|
||
|
||
if not cells:
|
||
return data
|
||
with zipfile.ZipFile(io.BytesIO(data)) as zin:
|
||
if "xl/worksheets/sheet1.xml" not in zin.namelist():
|
||
return data
|
||
sheet = zin.read("xl/worksheets/sheet1.xml").decode("utf-8", "replace")
|
||
others = {name: zin.read(name) for name in zin.namelist() if name != "xl/worksheets/sheet1.xml"}
|
||
|
||
for row, col, val in cells:
|
||
ref = f"{_col_letter(col)}{int(row)}"
|
||
cell_xml = (
|
||
f'<c r="{ref}" t="inlineStr"><is><t>{_xml_escape(val)}</t></is></c>'
|
||
)
|
||
pat = re.compile(rf'<c\b[^>]*\br="{re.escape(ref)}"[^>]*(?:/>|>[\s\S]*?</c>)')
|
||
if pat.search(sheet):
|
||
sheet = pat.sub(cell_xml, sheet, count=1)
|
||
continue
|
||
row_pat = re.compile(rf'(<row\b[^>]*\br="{int(row)}"[^>]*>)([\s\S]*?)(</row>)')
|
||
m = row_pat.search(sheet)
|
||
if m:
|
||
sheet = sheet[: m.start(3)] + cell_xml + sheet[m.start(3) :]
|
||
continue
|
||
sheet = sheet.replace(
|
||
"</sheetData>",
|
||
f'<row r="{int(row)}">{cell_xml}</row></sheetData>',
|
||
1,
|
||
)
|
||
|
||
out = io.BytesIO()
|
||
with zipfile.ZipFile(out, "w", compression=zipfile.ZIP_DEFLATED) as zout:
|
||
for name, blob in others.items():
|
||
zout.writestr(name, blob)
|
||
zout.writestr("xl/worksheets/sheet1.xml", sheet.encode("utf-8"))
|
||
return out.getvalue()
|
||
|
||
|
||
def overlay_labeled_facts(
|
||
*,
|
||
facts: Mapping[str, Any],
|
||
filepath: str = "",
|
||
file_b64: str = "",
|
||
) -> dict[str, str]:
|
||
"""
|
||
按模板标签再写协同格。失败则原样交回。
|
||
|
||
副作用：可能写临时 xlsx。无待写字段、或格子已是目标值时不改字节，
|
||
避免毁掉海运顶部运输分类单选。
|
||
"""
|
||
import base64
|
||
import tempfile
|
||
|
||
raw = b""
|
||
if filepath:
|
||
try:
|
||
raw = Path(filepath).read_bytes()
|
||
except OSError as exc:
|
||
logger.warning("overlay 读文件失败 err=%s", exc)
|
||
if not raw and file_b64 and len(file_b64) > 64:
|
||
try:
|
||
raw = base64.b64decode(file_b64)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.warning("overlay 解码失败 err=%s", exc)
|
||
if not raw:
|
||
return {"filepath": filepath, "file_b64": file_b64}
|
||
try:
|
||
from openpyxl import load_workbook
|
||
except ImportError:
|
||
return {"filepath": filepath, "file_b64": file_b64}
|
||
|
||
tmp = Path(tempfile.mkdtemp(prefix="quote-overlay-")) / "filled.xlsx"
|
||
tmp.write_bytes(raw)
|
||
written = 0
|
||
try:
|
||
wb = load_workbook(tmp)
|
||
ws = wb.worksheets[0]
|
||
pending: list[tuple[int, int, str]] = []
|
||
used_rows: set[int] = set()
|
||
for row in ws.iter_rows(min_row=1, max_row=ws.max_row or 1, max_col=min(ws.max_column or 1, 16)):
|
||
for cell in row:
|
||
text = str(cell.value or "").strip()
|
||
if not text or cell.row in used_rows:
|
||
continue
|
||
keys = _match_collab_label(text)
|
||
if not keys:
|
||
continue
|
||
val = _fact_for_markers(facts, keys)
|
||
if not val:
|
||
continue
|
||
dest = _label_value_cell(ws, cell)
|
||
# 主账已填过同值（如贸易条款）：再写会经 openpyxl 毁掉单选
|
||
if str(dest.value or "").strip() == val:
|
||
used_rows.add(cell.row)
|
||
break
|
||
pending.append((dest.row, dest.column, val))
|
||
used_rows.add(cell.row)
|
||
break
|
||
wb.close()
|
||
if not pending:
|
||
# 无增量可写：原样返回，禁止重存
|
||
logger.info("overlay 协同字段 written=0 keep_original")
|
||
return {
|
||
"filepath": filepath or str(tmp),
|
||
"file_b64": file_b64 or base64.b64encode(raw).decode("ascii"),
|
||
}
|
||
patched = _patch_cells_keep_structure(raw, pending)
|
||
tmp.write_bytes(patched)
|
||
written = len(pending)
|
||
except Exception:
|
||
logger.exception("overlay 协同字段失败")
|
||
return {"filepath": filepath, "file_b64": file_b64}
|
||
|
||
out_b64 = base64.b64encode(tmp.read_bytes()).decode("ascii")
|
||
logger.info("overlay 协同字段 written=%s keep_structure=1", written)
|
||
return {"filepath": str(tmp), "file_b64": out_b64}
|
||
|
||
|
||
def _band_value_goes_below(ws, label_cell) -> bool:
|
||
"""同一行凑齐业务员、客户名称、始发地，就是集拼询价条带。"""
|
||
text = str(label_cell.value or "").replace(" ", "")
|
||
if not any(key in text for key in ("客户名称", "业务员", "始发地", "货物类型")):
|
||
return False
|
||
hits = 0
|
||
for cell in ws[label_cell.row]:
|
||
compact = str(cell.value or "").replace(" ", "").replace("：", "").replace(":", "")
|
||
if compact in {"业务员", "客户名称", "始发地", "货物类型", "条款", "发货日期"}:
|
||
hits += 1
|
||
return hits >= 3
|
||
|
||
|
||
def _match_collab_label(text: str) -> tuple[str, ...] | None:
|
||
raw = (text or "").replace("：", ":").strip()
|
||
for marker, keys in _COLLAB_LABEL_MARKERS:
|
||
if marker in raw:
|
||
return keys
|
||
return None
|
||
|
||
|
||
def _label_value_cell(ws, label_cell):
|
||
"""标签右侧格子；合并标签则从合并区右边开始。
|
||
|
||
跨境集拼蓝条是「业务员 | 客户名称 | 始发地」，值在下一行。
|
||
写到右侧会把客户名称填进「始发地」表头。
|
||
"""
|
||
if _band_value_goes_below(ws, label_cell):
|
||
col = label_cell.column
|
||
merged = getattr(ws, "merged_cells", None)
|
||
if merged is not None:
|
||
for rng in merged.ranges:
|
||
if (
|
||
rng.min_row <= label_cell.row <= rng.max_row
|
||
and rng.min_col <= label_cell.column <= rng.max_col
|
||
):
|
||
col = rng.min_col
|
||
break
|
||
return ws.cell(row=label_cell.row + 1, column=col)
|
||
row_idx = label_cell.row
|
||
start_col = label_cell.column + 1
|
||
merged = getattr(ws, "merged_cells", None)
|
||
if merged is not None:
|
||
for rng in merged.ranges:
|
||
if rng.min_row <= row_idx <= rng.max_row and rng.min_col <= label_cell.column <= rng.max_col:
|
||
start_col = rng.max_col + 1
|
||
break
|
||
return ws.cell(row=row_idx, column=start_col)
|