126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
"""
|
||
按海运询价单标签补写协同字段。
|
||
|
||
主账填表只认英文码，货好时间 / 海关编码 / 含油电磁常空着。
|
||
本文件只改 Excel 字节，不打主账、不占回调线程。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
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")),
|
||
)
|
||
|
||
|
||
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 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)
|
||
pending.append((dest.row, dest.column, val))
|
||
used_rows.add(cell.row)
|
||
break
|
||
for r, c, val in pending:
|
||
ws.cell(row=r, column=c, value=val)
|
||
written += 1
|
||
wb.save(tmp)
|
||
wb.close()
|
||
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", written)
|
||
return {"filepath": str(tmp), "file_b64": out_b64}
|
||
|
||
|
||
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):
|
||
"""标签右侧格子；合并标签则从合并区右边开始。"""
|
||
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)
|