Add sign-off receipt report and refine packing list: single-line info bar, real-glyph row height, auto-paginated detail table
- Packing list: info bar changed to a single-line three-column layout (schedule no. | order no. | box no.); packing date moved to bottom-right on the same row as the total; date separator changed to slash - Packing list: range column widened 20->30mm; quantity/position-no./remark start shifted right 10mm; remark column narrowed to compensate; total width unchanged - Packing list: detail cells now use line_spacing=1.0 + valign=top; separators changed to thin solid lines at row bottom (+1mm line spacing); placeholder empty rows no longer step y - Packing list: row height estimation now uses fpdf2's same simhei glyph measured width (replacing rough character-count estimate); _clean adds control-character cleanup - New sign-off receipt report reports/sign_receipt: master+detail linked query, ReportBro auto-paginated table, PyMuPDF stamped footer (document no. / page no. / sign-off column) - run.py: required params now read from each report's config.yaml; default output path auto-generated; added transform.stamp_footer post-processing hook - .gitignore: ignore .workbuddy/
This commit is contained in:
319
reports/sign_receipt/_build_template.py
Normal file
319
reports/sign_receipt/_build_template.py
Normal file
@@ -0,0 +1,319 @@
|
||||
"""签收单模板:样式定义 + 文档元素布局(方案 A:ReportBro 表格自动换页)。
|
||||
|
||||
设计理念:
|
||||
明细改用 ReportBro **Table 元素**绑定数组参数 `items`,由 ReportBro 自动按内容高度
|
||||
换页、每页重复列头、末页渲染合计。完整的「多页分页」由 ReportBro 原生支持,
|
||||
不再像早期方案那样在 Python 端逐行算高度 + 绝对定位(那套在内容超一页时会报错/截断)。
|
||||
|
||||
- 标题 / 收发信息 / 运输信息:仍是 content band 内的绝对定位文本,仅第 1 页出现。
|
||||
- 明细:Table(header 行 repeatHeader + body 行绑定 items + footer 行放合计)。
|
||||
- 页脚家具(单据号、第 X/共 Y 页、签收栏、备注):由 run.py 在生成后用 PyMuPDF
|
||||
逐页盖印——因为 reportbro-lib 的文档页脚 band 不渲染自由文本,且 page_number/
|
||||
page_count 未注册为可解析参数,无法在模板内直接输出页码。
|
||||
|
||||
坐标系:mm。A4 横向 297 x 210。docElements 的 x/y/width/height 均经 mm() 换算为 pt。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
PT = 2.834645669 # 1mm = 2.834645669pt(72/25.4)
|
||||
|
||||
|
||||
def mm(v: float) -> int:
|
||||
"""mm → pt(取整)。所有坐标/尺寸输出前必须经过此换算。"""
|
||||
return round(v * PT)
|
||||
|
||||
|
||||
# ---- 页面与边距(mm,设计值)----
|
||||
PAGE_W = 297 # A4 横向:宽
|
||||
PAGE_H = 210 # A4 横向:高
|
||||
MARGIN_T = 10
|
||||
MARGIN_B = 16 # 底部留白较大:容纳盖印的签收栏 + 页码
|
||||
MARGIN_L = 12
|
||||
MARGIN_R = 12
|
||||
CONTENT_W = PAGE_W - MARGIN_L - MARGIN_R # 273mm
|
||||
CONTENT_H = PAGE_H - MARGIN_T - MARGIN_B # 184mm
|
||||
|
||||
CJK = "simhei"
|
||||
INK = "#1a1a1a"
|
||||
MUTE = "#8a8a8a"
|
||||
FAINT = "#b5b5b5"
|
||||
RULE = "#1a1a1a"
|
||||
|
||||
# ---- 明细列定义:(字段后缀, 表头, x偏移mm, 宽度mm, 对齐) ----
|
||||
# 宽度合计 = 272mm(A4 横向可用 273mm,留 1mm 余量避免越界)。所有内容居中。
|
||||
COLS = [
|
||||
("seq", "序号", 0, 14, "center"),
|
||||
# 产品名称列宽收窄:省出的 3.4mm 加到箱号列(箱号按 1.2× = 20.4mm),总宽仍 272mm
|
||||
("product_name", "产品名称", 14, 46.6, "center"),
|
||||
("model", "产品型号", 60.6, 79, "center"),
|
||||
("range_", "量程", 139.6, 40, "center"),
|
||||
("qty", "数量", 179.6, 24, "center"),
|
||||
("paichan_order", "排产号/订单号", 203.6, 48, "center"),
|
||||
("box_no", "箱号", 251.6, 20.4, "center"),
|
||||
]
|
||||
assert sum(c[3] for c in COLS) == 272, sum(c[3] for c in COLS)
|
||||
|
||||
# 表格行高(mm)
|
||||
HDR_H = 7.0
|
||||
BODY_H = 8.0
|
||||
FOOT_H = 8.0
|
||||
|
||||
|
||||
# ---------- 样式 ----------
|
||||
def _debug_border() -> bool:
|
||||
try:
|
||||
from core.settings import settings
|
||||
return settings.report.debug_border
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _text_style(id_, *, size=10, bold=False, color=INK, halign="left",
|
||||
valign="middle", name=None, pad_l=0, pad_r=0, font=CJK,
|
||||
debug_border=None, line_spacing=1.25):
|
||||
"""文本样式。padding 默认 0 让坐标精确可控;可经 pad_l/pad_r 加左右内边距(mm)。
|
||||
|
||||
debug_border 由配置文件 report.debug_border 决定(True 时给元素加边框便于核对占位)。
|
||||
"""
|
||||
db = _debug_border() if debug_border is None else debug_border
|
||||
return {
|
||||
"id": id_, "type": "text", "name": name or f"s{id_}",
|
||||
"font": font, "fontSize": size, "bold": bold, "italic": False,
|
||||
"underline": False, "strikethrough": False,
|
||||
"horizontalAlignment": halign, "verticalAlignment": valign,
|
||||
"textColor": color, "backgroundColor": "",
|
||||
"lineSpacing": line_spacing,
|
||||
"paddingLeft": pad_l, "paddingTop": 0, "paddingRight": pad_r, "paddingBottom": 0,
|
||||
"borderColor": FAINT,
|
||||
"borderWidth": 0.3 if db else 0,
|
||||
"borderRadius": 0,
|
||||
"borderAll": db,
|
||||
"borderLeft": db, "borderTop": db,
|
||||
"borderRight": db, "borderBottom": db,
|
||||
}
|
||||
|
||||
|
||||
def styles() -> list[dict]:
|
||||
return [
|
||||
_text_style(101, size=18, bold=True, halign="center", name="title"),
|
||||
_text_style(103, size=8, color=INK, name="info_lbl"),
|
||||
_text_style(105, size=9.5, bold=True, color=INK, halign="center", name="colhdr"),
|
||||
_text_style(106, size=9, color=INK, name="cell_l", line_spacing=1.0, valign="middle"),
|
||||
_text_style(107, size=9, color=INK, halign="center", name="cell_c", line_spacing=1.0, valign="middle"),
|
||||
_text_style(109, size=11.5, bold=True, color=INK, halign="left", name="total"),
|
||||
_text_style(118, size=13.5, bold=True, color=INK, halign="left", name="big_val_l"),
|
||||
_text_style(119, size=9, bold=True, color=INK, name="info_val_bold", line_spacing=1.0, valign="middle"),
|
||||
# 收发信息非加粗值(与标签垂直居中)
|
||||
_text_style(123, size=9, color=INK, name="info_val", line_spacing=1.0, valign="middle"),
|
||||
# 表格单元格
|
||||
_text_style(127, size=9, bold=True, color=MUTE, halign="center", name="tbl_hdr", valign="middle"),
|
||||
_text_style(128, size=9, color=INK, halign="center", name="tbl_cell", valign="middle"),
|
||||
_text_style(129, size=9, bold=True, color=INK, halign="center", name="tbl_ftr", valign="middle"),
|
||||
# 表格 + 表头带样式
|
||||
# 边框风格沿用项目既有约定(与 packing_list 一致):无竖向网格、无左右外框,
|
||||
# 仅横向细线分隔;线条用主色 RULE(#1a1a1a)而非浅灰,表头不加灰底纹。
|
||||
{"id": 130, "type": "table", "name": "tbl", "border": "row",
|
||||
"borderColor": RULE, "borderWidth": 0.3},
|
||||
{"id": 131, "type": "tableBand", "name": "tbl_hdr_band",
|
||||
"backgroundColor": "", "alternateBackgroundColor": ""},
|
||||
# 线段样式
|
||||
{"id": 201, "type": "line", "name": "rule_faint", "color": FAINT, "borderWidth": 0.3},
|
||||
{"id": 202, "type": "line", "name": "rule_strong", "color": RULE, "borderWidth": 2.5},
|
||||
]
|
||||
|
||||
|
||||
def document_properties() -> dict:
|
||||
return {
|
||||
"pageFormat": "A4", "orientation": "landscape",
|
||||
"marginLeft": mm(MARGIN_L), "marginRight": mm(MARGIN_R),
|
||||
"marginTop": mm(MARGIN_T), "marginBottom": mm(MARGIN_B),
|
||||
"headerDisplay": "never", "headerSize": 0,
|
||||
"footerDisplay": "never", "footerSize": 0,
|
||||
"patternLocale": "zh", "patternCurrencySymbol": "",
|
||||
"patternNumberGroupSymbol": "",
|
||||
}
|
||||
|
||||
|
||||
def parameters() -> list[dict]:
|
||||
"""模板参数:单头标量 + 明细数组 items(子字段)。"""
|
||||
scalar = [
|
||||
("receipt_no", "string"), ("seq", "number"), ("receipt_date", "string"),
|
||||
("shipper_name", "string"), ("shipper_address", "string"), ("shipper_phone", "string"),
|
||||
("receiver_name", "string"), ("receiver_address", "string"), ("receiver_phone", "string"),
|
||||
("customer_order_no", "string"), ("customer_total_no", "string"),
|
||||
("transport_mode", "string"), ("ship_date", "string"),
|
||||
("sign_date", "string"), ("signed_by", "string"), ("maker", "string"),
|
||||
("receipt_remark", "string"), ("total_qty", "number"), ("total_boxes", "number"),
|
||||
]
|
||||
params = [{"id": i + 1, "name": n, "type": t, "nullable": True}
|
||||
for i, (n, t) in enumerate(scalar)]
|
||||
# 明细数组参数
|
||||
params.append({"id": 100, "name": "items", "type": "array", "nullable": True, "children": [
|
||||
{"id": 101, "name": "seq", "type": "string", "nullable": True},
|
||||
{"id": 102, "name": "product_name", "type": "string", "nullable": True},
|
||||
{"id": 103, "name": "model", "type": "string", "nullable": True},
|
||||
{"id": 104, "name": "range_", "type": "string", "nullable": True},
|
||||
{"id": 105, "name": "qty", "type": "string", "nullable": True},
|
||||
{"id": 106, "name": "paichan_order", "type": "string", "nullable": True},
|
||||
{"id": 107, "name": "box_no", "type": "string", "nullable": True},
|
||||
]})
|
||||
return params
|
||||
|
||||
|
||||
# ---------- 元素工厂 ----------
|
||||
def _text(id_, x, y, w, h, content, *, style_id, containerId="0_content"):
|
||||
return {
|
||||
"id": id_, "elementType": "text", "containerId": containerId,
|
||||
"x": x, "y": y, "width": w, "height": h,
|
||||
"content": content, "styleId": style_id, "eval": False,
|
||||
"printIf": "", "removeEmptyElement": False, "alwaysPrintOnSamePage": False,
|
||||
"link": "", "pattern": "", "cs_condition": "",
|
||||
"richText": False, "richTextHtml": "",
|
||||
"spreadsheet_hide": True, "spreadsheet_column": 0,
|
||||
"spreadsheet_colspan": 1, "spreadsheet_addEmptyRow": False,
|
||||
}
|
||||
|
||||
|
||||
def _line(id_, x, y, w, *, style_id, weight=0.3):
|
||||
return {
|
||||
"id": id_, "elementType": "line", "containerId": "0_content",
|
||||
"x": x, "y": y, "width": w, "height": mm(weight),
|
||||
"styleId": style_id, "printIf": "", "removeEmptyElement": False,
|
||||
"spreadsheet_hide": True, "spreadsheet_column": 0,
|
||||
"spreadsheet_addEmptyRow": False,
|
||||
}
|
||||
|
||||
|
||||
def _tcell(cid, width, content, style_id, colspan=1):
|
||||
"""表格单元格(TableTextElement)。x/y 由表格自动计算,只需提供 width/height。"""
|
||||
return {
|
||||
"id": cid, "elementType": "text", "x": 0, "y": 0, "width": width, "height": mm(BODY_H),
|
||||
"content": content, "styleId": style_id, "eval": False, "printIf": "",
|
||||
"removeEmptyElement": False, "colspan": colspan, "growWeight": 0,
|
||||
}
|
||||
|
||||
|
||||
def build_doc_elements(context: dict[str, Any]) -> list[dict]:
|
||||
"""按数据生成文档元素:第 1 页标题/收发信息/运输信息 + 自动换页的明细表格。
|
||||
|
||||
返回 content band 内的元素列表(不含页脚——页脚由 PyMuPDF 盖印)。
|
||||
"""
|
||||
L = 0
|
||||
CW = mm(CONTENT_W) - 2 # 满宽减 2pt 余量,避免误差越界
|
||||
els: list[dict] = []
|
||||
nid = [2000]
|
||||
|
||||
def nid_():
|
||||
nid[0] += 1
|
||||
return nid[0]
|
||||
|
||||
# ===== 标题区(仅第 1 页)=====
|
||||
y = 2.0
|
||||
els.append(_text(nid_(), L, mm(y), mm(70), mm(5), "客户总排行号", style_id=103))
|
||||
els.append(_text(nid_(), L, mm(y + 5), mm(70), mm(8), "${customer_total_no}", style_id=118))
|
||||
els.append(_text(nid_(), L, mm(y), CW, mm(13), "签 收 单", style_id=101))
|
||||
y = y + 16
|
||||
# 标题双线:粗线(与装箱单标题双线同款 weight=0.45)在上,细线(fpdf 默认最细)在下,
|
||||
# 间距 1.5mm,风格与装箱单保持一致。
|
||||
els.append(_line(nid_(), L, mm(y), CW, style_id=202, weight=0.45))
|
||||
els.append(_line(nid_(), L, mm(y + 1.5), CW, style_id=202, weight=0))
|
||||
y = y + 4
|
||||
|
||||
# ===== 收发信息区(左发货 / 右收货)=====
|
||||
# 运输方式 / 发货日期并入左列(与发货单位/地址/电话同列对齐),
|
||||
# 客户订单编号并入右列(与收货电话同列对齐)。左列因此比右列多一行(发货日期),
|
||||
# 用户确认「多出来一个没得问题」。
|
||||
info_left = [
|
||||
("发货单位", "${shipper_name}", True),
|
||||
("发货地址", "${shipper_address}", False),
|
||||
("发货电话", "${shipper_phone}", False),
|
||||
("运输方式", "${transport_mode}", False),
|
||||
("发货日期", "${ship_date}", False),
|
||||
]
|
||||
info_right = [
|
||||
("收货人", "${receiver_name}", True),
|
||||
("收货地址", "${receiver_address}", False),
|
||||
("收货电话", "${receiver_phone}", False),
|
||||
("客户订单编号", "${customer_order_no}", False),
|
||||
]
|
||||
col_w = 112
|
||||
label_w = 24
|
||||
x_l = L
|
||||
x_r = L + mm(137)
|
||||
|
||||
def draw_col(x: float, rows) -> float:
|
||||
yy = y
|
||||
for lbl, val, bold in rows:
|
||||
h = 12 if "地址" in lbl else 9
|
||||
els.append(_text(nid_(), x, mm(yy), mm(label_w), mm(h), lbl, style_id=103))
|
||||
els.append(_text(nid_(), x + mm(label_w), mm(yy), mm(col_w - label_w), mm(h),
|
||||
val, style_id=119 if bold else 123))
|
||||
yy += h
|
||||
return yy
|
||||
|
||||
yl = draw_col(x_l, info_left)
|
||||
yr = draw_col(x_r, info_right)
|
||||
# 去掉「签收单表头(收发信息区)下方」的额外分隔线,仅保留明细表头上方由表格
|
||||
# border:"row" 渲染的顶边线作为两者之间的唯一分隔线。y 步进保持原值(+5),
|
||||
# 使明细表格纵向位置不变。
|
||||
y = max(yl, yr) + 5
|
||||
|
||||
# ===== 明细表格(自动换页)=====
|
||||
col_w_pt = [mm(c[3]) for c in COLS]
|
||||
|
||||
def header_row():
|
||||
cells = []
|
||||
for i, (_f, label, _xoff, _w, _a) in enumerate(COLS):
|
||||
cells.append(_tcell(3000 + i, col_w_pt[i], label, 127))
|
||||
return {"id": 4000, "height": mm(HDR_H), "repeatHeader": True, "columnData": cells}
|
||||
|
||||
def body_row():
|
||||
cells = []
|
||||
for i, (f, _label, _xoff, _w, _a) in enumerate(COLS):
|
||||
cells.append(_tcell(5000 + i, col_w_pt[i], "${%s}" % f, 128))
|
||||
# alwaysPrintOnSamePage: 当前页剩余高度放不下整行(含折行后的真实高度)时,
|
||||
# 整行推到下一页,绝不在两页之间断开一条记录(尤其产品名称/型号折行时)。
|
||||
return {"id": 4100, "height": mm(BODY_H), "alwaysPrintOnSamePage": True,
|
||||
"columnData": cells}
|
||||
|
||||
def footer_row():
|
||||
# 合计数量(序号列)+ 数量列填 total_qty + 箱号列填共 X 箱(与箱号字段同列对齐,
|
||||
# 不再放在排产号/订单号列)。产品名称列宽已收窄、箱号列宽已加宽以容纳该内容。
|
||||
cells = [
|
||||
_tcell(6000, col_w_pt[0], "合计", 129),
|
||||
_tcell(6001, col_w_pt[1], "", 129),
|
||||
_tcell(6002, col_w_pt[2], "", 129),
|
||||
_tcell(6003, col_w_pt[3], "", 129),
|
||||
_tcell(6004, col_w_pt[4], "${total_qty}", 129),
|
||||
_tcell(6005, col_w_pt[5], "", 129),
|
||||
_tcell(6006, col_w_pt[6], "共 ${total_boxes} 箱", 129),
|
||||
]
|
||||
# footer 带高度回到 FOOT_H:表格底边线(rendering.py 始终画在表格最底)即落在
|
||||
# 「紧贴合计下方」,作为合计区与下方备注/签收栏的分隔线。备注/签收栏由 stamp_footer
|
||||
# 用 fitz 定位「合计」后画在正下方,落在表格底边线之下。
|
||||
return {"id": 4200, "height": mm(FOOT_H), "columnData": cells}
|
||||
|
||||
table = {
|
||||
"id": 7000, "elementType": "table", "containerId": "0_content",
|
||||
"x": 0, "y": mm(y), "width": sum(col_w_pt), "height": mm(HDR_H + BODY_H),
|
||||
"dataSource": "items", "columns": len(COLS), "styleId": 130,
|
||||
"header": True, "headerData": header_row(),
|
||||
"contentDataRows": [body_row()],
|
||||
"footer": True, "footerData": footer_row(),
|
||||
}
|
||||
els.append(table)
|
||||
|
||||
return els
|
||||
|
||||
|
||||
def build_report_definition(context: dict[str, Any]) -> dict:
|
||||
"""组装完整 report_definition(运行时调用)。"""
|
||||
return {
|
||||
"version": 6,
|
||||
"documentProperties": document_properties(),
|
||||
"parameters": parameters(),
|
||||
"styles": styles(),
|
||||
"docElements": build_doc_elements(context),
|
||||
}
|
||||
131
reports/sign_receipt/add_stress_rows.py
Normal file
131
reports/sign_receipt/add_stress_rows.py
Normal file
@@ -0,0 +1,131 @@
|
||||
"""签收单 20260803001 压力测试数据。
|
||||
|
||||
目的:观察当前「单页绝对定位」模板在【内容 > 1 页】时的表现:
|
||||
* 明细行 > MAX_ROWS(20) 时,超出部分被静默截断(合计仍按全量计)。
|
||||
* 即便在 20 行内,若累计高度超过内容区,行会越过页脚、与签收栏重叠。
|
||||
* 发货/收货地址为固定高度(12mm)文本,加长后会换行并压到下一行(电话)上。
|
||||
|
||||
可重复运行:STRESS_BOXES 为显式列表,运行前先 DELETE 这些箱,再 INSERT;
|
||||
地址直接 UPDATE(幂等)。
|
||||
|
||||
运行:
|
||||
.venv/Scripts/python.exe reports/sign_receipt/add_stress_rows.py
|
||||
|
||||
注意:本脚本只动 20260803001;重新跑 setup_db.py 会清空 202608030% 测试
|
||||
数据(含本脚本追加内容),恢复干净状态。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# 让脚本能 import core.*
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(PROJECT_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
from core.db import get_engine
|
||||
from sqlalchemy import text
|
||||
|
||||
RECEIPT_NO = "20260803001"
|
||||
|
||||
# 显式选取一批真实箱(来自 finished_goods_box,单箱明细行数较多,确保 > 20 行)。
|
||||
# 格式:(paichan_no, box_no)
|
||||
STRESS_BOXES = [
|
||||
("R07200", 1), ("R06994", 1), ("R06994", 2),
|
||||
("R06626", 1), ("R06687", 1), ("R06884", 1),
|
||||
("R05541", 1), ("R06006", 87),
|
||||
]
|
||||
|
||||
LONG_SHIPPER_ADDR = (
|
||||
"北京市朝阳区南三环成寿寺路甲135号院1号楼 北京布莱迪仪器仪表有限公司 "
|
||||
"仓储物流中心收发室(近地铁成寿寺站B口,工作日8:30-17:30可收货,"
|
||||
"节假日及夜间到货请提前与仓管预约,货车限高3.2米,卸货区在厂区西北角)"
|
||||
)
|
||||
LONG_RECEIVER_ADDR = (
|
||||
"上海市浦东新区张江高科技园区博云路2号三期厂房B栋5层收货平台 "
|
||||
"(收货联系人:王女士,电话:15230028484;请走北门货运通道,"
|
||||
"货车需登记后入厂,卸货位B1-07,到货前1小时请短信通知)"
|
||||
)
|
||||
|
||||
# 静态 VALUES 列表(STRESS_BOXES 为硬编码常量,无注入风险)
|
||||
VALUES_SQL = ", ".join(f"(:p{i}, :b{i})" for i in range(len(STRESS_BOXES)))
|
||||
PAIR_VALS = {f"p{i}": p for i, (p, _) in enumerate(STRESS_BOXES)} | \
|
||||
{f"b{i}": b for i, (_, b) in enumerate(STRESS_BOXES)}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
eng = get_engine()
|
||||
with eng.begin() as conn:
|
||||
# 1) 清掉上一轮追加的箱(VALUES + EXISTS,兼容 SQL Server)
|
||||
conn.execute(
|
||||
text(
|
||||
f"DELETE FROM warehouseOutbound.delivery_receipt_item "
|
||||
f"WHERE receipt_no = :r AND EXISTS ("
|
||||
f" SELECT 1 FROM (VALUES {VALUES_SQL}) v(pa, bo) "
|
||||
f" WHERE v.pa = paichan_no AND v.bo = box_no)"
|
||||
),
|
||||
{**PAIR_VALS, "r": RECEIPT_NO},
|
||||
)
|
||||
|
||||
# 2) 仅当箱真实存在才插入
|
||||
exist = {
|
||||
(p, b) for (p, b) in conn.execute(
|
||||
text(
|
||||
f"SELECT v.pa, v.bo FROM (VALUES {VALUES_SQL}) v(pa, bo) "
|
||||
f"WHERE EXISTS (SELECT 1 FROM CargoTrace.finished_goods_box b "
|
||||
f" WHERE b.paichan_no = v.pa AND b.box_no = v.bo)"
|
||||
),
|
||||
PAIR_VALS,
|
||||
).fetchall()
|
||||
}
|
||||
missing = [pb for pb in STRESS_BOXES if pb not in exist]
|
||||
if missing:
|
||||
print(" 警告:以下箱在 finished_goods_box 不存在,已跳过:", missing)
|
||||
|
||||
for (p, b) in STRESS_BOXES:
|
||||
if (p, b) in exist:
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO warehouseOutbound.delivery_receipt_item "
|
||||
"(receipt_no, paichan_no, box_no) VALUES (:r, :p, :b)"
|
||||
),
|
||||
{"r": RECEIPT_NO, "p": p, "b": b},
|
||||
)
|
||||
|
||||
# 3) 拉长地址(幂等 UPDATE)
|
||||
conn.execute(
|
||||
text(
|
||||
"UPDATE warehouseOutbound.delivery_receipt "
|
||||
"SET shipper_address = :sa, receiver_address = :ra "
|
||||
"WHERE receipt_no = :r"
|
||||
),
|
||||
{"sa": LONG_SHIPPER_ADDR, "ra": LONG_RECEIVER_ADDR, "r": RECEIPT_NO},
|
||||
)
|
||||
|
||||
# 校验
|
||||
with eng.connect() as conn:
|
||||
n = conn.execute(
|
||||
text("SELECT COUNT(*) FROM warehouseOutbound.delivery_receipt_item "
|
||||
"WHERE receipt_no = :r"), {"r": RECEIPT_NO}).scalar()
|
||||
rows = conn.execute(
|
||||
text(
|
||||
"SELECT COUNT(*) FROM ("
|
||||
"SELECT bi.id FROM warehouseOutbound.delivery_receipt_item di "
|
||||
"JOIN CargoTrace.finished_goods_box b ON b.paichan_no=di.paichan_no AND b.box_no=di.box_no "
|
||||
"JOIN CargoTrace.finished_goods_box_item bi ON bi.box_id=b.id "
|
||||
"WHERE di.receipt_no = :r) t"
|
||||
), {"r": RECEIPT_NO}).scalar()
|
||||
sa = conn.execute(
|
||||
text("SELECT LEN(shipper_address) FROM warehouseOutbound.delivery_receipt "
|
||||
"WHERE receipt_no=:r"), {"r": RECEIPT_NO}).scalar()
|
||||
ra = conn.execute(
|
||||
text("SELECT LEN(receiver_address) FROM warehouseOutbound.delivery_receipt "
|
||||
"WHERE receipt_no=:r"), {"r": RECEIPT_NO}).scalar()
|
||||
print(f" {RECEIPT_NO} 明细箱数={n},预计明细行数(产品粒度)={rows}")
|
||||
print(f" 发货地址长度={sa}字符,收货地址长度={ra}字符")
|
||||
print(f" MAX_ROWS=20 → 超出 {max(0, rows - 20)} 行将被截断(合计仍按 {rows} 行计)。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
38
reports/sign_receipt/config.yaml
Normal file
38
reports/sign_receipt/config.yaml
Normal file
@@ -0,0 +1,38 @@
|
||||
name: sign_receipt
|
||||
title: 签收单
|
||||
description: 按签收单序列号生成 PDF(一次发货的集合视图,含多排产号多箱明细 + 收发方签收信息)
|
||||
renderer: reportbro-lib
|
||||
params:
|
||||
receipt_no:
|
||||
description: 签收单序列号(11位,YYYYMMDD + 3位当日序号,如 20260803001)
|
||||
required: true
|
||||
files:
|
||||
query.sql: 参数化查询(:receipt_no),主表头 LEFT JOIN 明细链路,按 (箱号,产品) 聚合
|
||||
transform.py: build_context(rows) → 单头标量 + 明细预展开(r0_*..r19_*) + 行高估算
|
||||
_build_template.py: 样式定义 + 运行时按数据动态布局的文档元素生成(无边框列表式)
|
||||
fields:
|
||||
receipt_no: 签收单序列号(主键)
|
||||
seq: "当日顺序号;receipt_date: 序列号日期部分"
|
||||
shipper_name/address/phone: 发货单位/地址/电话
|
||||
receiver_name/address/phone: 收货人/地址/电话
|
||||
customer_order_no: 客户订单编号
|
||||
customer_total_no: 客户提供的外部总排行号(与内部排产号无关)
|
||||
transport_mode: 运输方式
|
||||
ship_date: "发货日期;sign_date/signed_by: 签收日期/签收人(回写)"
|
||||
maker: "制单人;receipt_remark: 备注"
|
||||
r{i}_product_name: 产品名称(合同表[客户名称])
|
||||
r{i}_model: 产品型号
|
||||
r{i}_range_: 量程
|
||||
r{i}_qty: 数量
|
||||
r{i}_contract_order_no: 合同号/订单号(合同表[订单号])
|
||||
r{i}_box_no: 箱号
|
||||
total_boxes: "合计箱数(按 paichan_no+box_no 去重);total_qty: 合计件数"
|
||||
notes:
|
||||
- 单位:一个序列号一份 PDF(一次发货的集合视图)
|
||||
- 纸张:A4 纵向(默认)。内容多时同 A4 纵向,本期统一 A4 纵向
|
||||
- 排版:无边框、单色、字体驱动的列表式(复用装箱单方案,ReportBro 表格行高不自适应故弃用)
|
||||
- 明细行预展开为标量参数 r0_*..r19_*;模板按 row_heights 累加 y 坐标定位
|
||||
- 产品信息按总排号双表 LEFT JOIN(压力表/温度计合同表)取 COALESCE
|
||||
- 合计箱数 = 去重 (paichan_no, box_no);合计件数 = SUM(qty)
|
||||
- 中文通过 additional_fonts 注册 simhei.ttf 渲染(见 core/fonts.py)
|
||||
- 预留最多 20 行明细;超过可在 _build_template.MAX_ROWS 与 transform.MAX_ROWS 同步调大
|
||||
72
reports/sign_receipt/query.sql
Normal file
72
reports/sign_receipt/query.sql
Normal file
@@ -0,0 +1,72 @@
|
||||
-- 签收单:按序列号 receipt_no 取主表头 + 聚合多排产号多箱的明细。
|
||||
-- 参数:receipt_no (str, 11 位序列号)
|
||||
-- 注:load_sql 会剥离整行 -- 注释,故注释里可自由书写中文。
|
||||
--
|
||||
-- 数据来源:
|
||||
-- warehouseOutbound.delivery_receipt 签收单主表(收发方/客户总排行号/运输/日期)
|
||||
-- warehouseOutbound.delivery_receipt_item 签收单明细(仅 receipt_no + paichan_no + box_no)
|
||||
-- CargoTrace.finished_goods_box 箱头(按 paichan_no + box_no 定位)
|
||||
-- CargoTrace.finished_goods_box_item 箱内明细(总排号 + 数量)
|
||||
-- productionContractData.26年压力表/温度计合同数据 产品信息
|
||||
--
|
||||
-- 设计要点:
|
||||
-- * 主表 LEFT JOIN 进明细链路,使每一行都冗余主表头字段;
|
||||
-- transform 取首行作为单头,其余行作为明细预展开。
|
||||
-- * 明细粒度:一个 (箱号, 产品) 一行;同一箱同一产品数量 SUM 聚合。
|
||||
-- * 排产号/订单号:由「箱内排产号 paichan_no + 合同表[订单号]」组合而成,
|
||||
-- 对应示例"排产号/订单号"列(不再单独取合同号)。
|
||||
-- * 客户总排行号 customer_total_no 已在主表,直接带出,与内部 paichan_no 无关。
|
||||
SELECT
|
||||
h.receipt_no,
|
||||
h.receipt_date,
|
||||
h.seq,
|
||||
h.shipper_name,
|
||||
h.shipper_address,
|
||||
h.shipper_phone,
|
||||
h.receiver_name,
|
||||
h.receiver_address,
|
||||
h.receiver_phone,
|
||||
h.customer_order_no,
|
||||
h.customer_total_no,
|
||||
h.transport_mode,
|
||||
h.ship_date,
|
||||
h.sign_date,
|
||||
h.signed_by,
|
||||
h.maker,
|
||||
h.remark AS receipt_remark,
|
||||
|
||||
b.box_no,
|
||||
di.paichan_no AS paichan_no,
|
||||
COALESCE(p.[客户名称], t.[客户名称]) AS product_name,
|
||||
COALESCE(p.[产品型号], t.[产品型号]) AS model,
|
||||
COALESCE(p.[量程], t.[量程]) AS range_,
|
||||
COALESCE(p.[位号], t.[位号]) AS weihao,
|
||||
SUM(bi.quantity) AS qty,
|
||||
COALESCE(p.[订单号], t.[订单号]) AS order_no
|
||||
FROM warehouseOutbound.delivery_receipt h
|
||||
JOIN warehouseOutbound.delivery_receipt_item di
|
||||
ON di.receipt_no = h.receipt_no
|
||||
JOIN CargoTrace.finished_goods_box b
|
||||
ON b.paichan_no = di.paichan_no AND b.box_no = di.box_no
|
||||
JOIN CargoTrace.finished_goods_box_item bi
|
||||
ON bi.box_id = b.id
|
||||
LEFT JOIN [productionContractData].[26年压力表合同数据] p
|
||||
ON p.[总排号] = bi.zongpai_no
|
||||
LEFT JOIN [productionContractData].[26年温度计合同数据] t
|
||||
ON t.[总排号] = bi.zongpai_no
|
||||
WHERE h.receipt_no = :receipt_no
|
||||
GROUP BY
|
||||
h.receipt_no, h.receipt_date, h.seq,
|
||||
h.shipper_name, h.shipper_address, h.shipper_phone,
|
||||
h.receiver_name, h.receiver_address, h.receiver_phone,
|
||||
h.customer_order_no, h.customer_total_no,
|
||||
h.transport_mode, h.ship_date, h.sign_date, h.signed_by,
|
||||
h.maker, h.remark,
|
||||
b.box_no,
|
||||
di.paichan_no,
|
||||
COALESCE(p.[客户名称], t.[客户名称]),
|
||||
COALESCE(p.[产品型号], t.[产品型号]),
|
||||
COALESCE(p.[量程], t.[量程]),
|
||||
COALESCE(p.[位号], t.[位号]),
|
||||
COALESCE(p.[订单号], t.[订单号])
|
||||
ORDER BY b.box_no;
|
||||
341
reports/sign_receipt/setup_db.py
Normal file
341
reports/sign_receipt/setup_db.py
Normal file
@@ -0,0 +1,341 @@
|
||||
"""签收单(delivery_receipt)建表 + 测试假数据。
|
||||
|
||||
用法:
|
||||
.venv/Scripts/python.exe reports/sign_receipt/setup_db.py
|
||||
|
||||
逻辑:
|
||||
1. 若 warehouseOutbound.delivery_receipt / delivery_receipt_item 不存在则建表
|
||||
(含 schema、主键、唯一约束、外键级联、按天索引、updated_at 触发器、列备注)。
|
||||
2. 清理旧的测试签收单(receipt_no 以 202608030 开头),再插入:
|
||||
- 1 张单排产号签收单
|
||||
- 1 张多排产号签收单
|
||||
明细引用真实存在的 CargoTrace.finished_goods_box (paichan_no, box_no)。
|
||||
3. 打印验证结果。
|
||||
|
||||
可重复运行:建表幂等(先判断存在),测试数据先删后插。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# 让脚本能 import core.*
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(PROJECT_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
from sqlalchemy import text # noqa: E402
|
||||
|
||||
from core.db import get_engine # noqa: E402
|
||||
|
||||
ENGINE = get_engine()
|
||||
|
||||
SCHEMA = "warehouseOutbound"
|
||||
|
||||
# ---------------------------------------------------------------- DDL
|
||||
DDL_STATEMENTS = [
|
||||
# 1) schema(CREATE SCHEMA 必须是批首语句,用动态 SQL 包裹以兼容 IF 判断)
|
||||
"IF NOT EXISTS (SELECT 1 FROM sys.schemas WHERE name = 'warehouseOutbound') "
|
||||
"EXEC('CREATE SCHEMA warehouseOutbound')",
|
||||
|
||||
# 2) 主表
|
||||
"""CREATE TABLE warehouseOutbound.delivery_receipt (
|
||||
receipt_no VARCHAR(11) NOT NULL,
|
||||
receipt_date DATE NOT NULL,
|
||||
seq INT NOT NULL,
|
||||
shipper_name NVARCHAR(200),
|
||||
shipper_address NVARCHAR(500),
|
||||
shipper_phone NVARCHAR(50),
|
||||
receiver_name NVARCHAR(100),
|
||||
receiver_address NVARCHAR(500),
|
||||
receiver_phone NVARCHAR(50),
|
||||
customer_order_no VARCHAR(100),
|
||||
customer_total_no VARCHAR(50),
|
||||
transport_mode NVARCHAR(50),
|
||||
ship_date DATE,
|
||||
sign_date DATE,
|
||||
signed_by NVARCHAR(100),
|
||||
maker NVARCHAR(100),
|
||||
remark NVARCHAR(1000),
|
||||
created_at DATETIME2 NOT NULL DEFAULT GETDATE(),
|
||||
updated_at DATETIME2 NOT NULL DEFAULT GETDATE(),
|
||||
CONSTRAINT PK_delivery_receipt PRIMARY KEY (receipt_no),
|
||||
CONSTRAINT UQ_delivery_receipt_date_seq UNIQUE (receipt_date, seq)
|
||||
)""",
|
||||
|
||||
# 3) 明细表(仅最小关联键)
|
||||
"""CREATE TABLE warehouseOutbound.delivery_receipt_item (
|
||||
receipt_no VARCHAR(11) NOT NULL,
|
||||
paichan_no VARCHAR(50) NOT NULL,
|
||||
box_no INT NOT NULL,
|
||||
CONSTRAINT PK_delivery_receipt_item PRIMARY KEY (receipt_no, paichan_no, box_no),
|
||||
CONSTRAINT FK_delivery_receipt_item_receipt FOREIGN KEY (receipt_no)
|
||||
REFERENCES warehouseOutbound.delivery_receipt (receipt_no) ON DELETE CASCADE
|
||||
)""",
|
||||
|
||||
# 4) 按天查询辅助索引
|
||||
"CREATE INDEX IX_delivery_receipt_date ON warehouseOutbound.delivery_receipt (receipt_date)",
|
||||
|
||||
# 5) updated_at 自动维护触发器
|
||||
"""CREATE TRIGGER warehouseOutbound.TR_delivery_receipt_updated
|
||||
ON warehouseOutbound.delivery_receipt AFTER UPDATE AS
|
||||
BEGIN
|
||||
UPDATE r SET updated_at = GETDATE()
|
||||
FROM warehouseOutbound.delivery_receipt r
|
||||
JOIN inserted i ON i.receipt_no = r.receipt_no;
|
||||
END""",
|
||||
]
|
||||
|
||||
# 主表列备注:(列名, 备注)
|
||||
RECEIPT_COL_COMMENTS = [
|
||||
("receipt_no", "签收单序列号(YYYYMMDD+3位当日序号),唯一主键"),
|
||||
("receipt_date", "序列号中的日期部分(冗余,便于按天查询)"),
|
||||
("seq", "当日顺序号:从1起、按天重置、步长1"),
|
||||
("shipper_name", "发货单位名称"),
|
||||
("shipper_address", "发货单位地址"),
|
||||
("shipper_phone", "发货单位联系电话"),
|
||||
("receiver_name", "收货人姓名"),
|
||||
("receiver_address", "收货人地址"),
|
||||
("receiver_phone", "收货人联系电话"),
|
||||
("customer_order_no", "客户订单编号"),
|
||||
("customer_total_no", "客户提供的外部总排行号(与内部排产号完全无关)"),
|
||||
("transport_mode", "运输方式"),
|
||||
("ship_date", "发货日期"),
|
||||
("sign_date", "签收日期(客户签收后回写)"),
|
||||
("signed_by", "签收人(客户签收后回写)"),
|
||||
("maker", "制单人"),
|
||||
("remark", "备注"),
|
||||
("created_at", "记录创建时间"),
|
||||
("updated_at", "记录最后更新时间(触发器自动维护)"),
|
||||
]
|
||||
ITEM_COL_COMMENTS = [
|
||||
("receipt_no", "签收单序列号(关联主表 delivery_receipt)"),
|
||||
("paichan_no", "内部排产号(用于定位箱子,与客户总排行号无关)"),
|
||||
("box_no", "箱号(在排产号下从1开始)"),
|
||||
]
|
||||
|
||||
|
||||
def _table_exists(conn, name: str) -> bool:
|
||||
return conn.execute(
|
||||
text(
|
||||
"SELECT 1 FROM sys.tables t "
|
||||
"JOIN sys.schemas s ON s.schema_id = t.schema_id "
|
||||
"WHERE s.name = :sch AND t.name = :tbl"
|
||||
),
|
||||
{"sch": SCHEMA, "tbl": name},
|
||||
).first() is not None
|
||||
|
||||
|
||||
def _add_col_comment(conn, table: str, col: str, desc: str) -> None:
|
||||
conn.execute(
|
||||
text(
|
||||
"EXEC sys.sp_addextendedproperty "
|
||||
"N'MS_Description', :desc, "
|
||||
"N'SCHEMA', N'warehouseOutbound', "
|
||||
"N'TABLE', :tbl, N'COLUMN', :col"
|
||||
),
|
||||
{"desc": desc, "tbl": table, "col": col},
|
||||
)
|
||||
|
||||
|
||||
def _build_schema(conn) -> None:
|
||||
print("[1/3] 建表(warehouseOutbound schema)...")
|
||||
for stmt in DDL_STATEMENTS:
|
||||
conn.execute(text(stmt))
|
||||
|
||||
if not _table_exists(conn, "delivery_receipt"):
|
||||
# 理论上上面已建,防御性检查
|
||||
pass
|
||||
for col, desc in RECEIPT_COL_COMMENTS:
|
||||
_add_col_comment(conn, "delivery_receipt", col, desc)
|
||||
for col, desc in ITEM_COL_COMMENTS:
|
||||
_add_col_comment(conn, "delivery_receipt_item", col, desc)
|
||||
print(" 表与列备注已就绪。")
|
||||
|
||||
|
||||
def _probe_boxes(conn):
|
||||
"""返回可用的 (paichan_no, [box_no,...]) 列表(最多取 3 个排产号)。"""
|
||||
rows = conn.execute(
|
||||
text(
|
||||
"SELECT DISTINCT TOP 3 paichan_no "
|
||||
"FROM CargoTrace.finished_goods_box ORDER BY paichan_no"
|
||||
)
|
||||
).fetchall()
|
||||
result = []
|
||||
for (p,) in rows:
|
||||
boxes = [
|
||||
r[0]
|
||||
for r in conn.execute(
|
||||
text(
|
||||
"SELECT box_no FROM CargoTrace.finished_goods_box "
|
||||
"WHERE paichan_no = :p ORDER BY box_no"
|
||||
),
|
||||
{"p": p},
|
||||
).fetchall()
|
||||
]
|
||||
if boxes:
|
||||
result.append((p, boxes))
|
||||
return result
|
||||
|
||||
|
||||
def _seed(conn, boxes_by_paichan) -> None:
|
||||
print("[2/3] 写入测试假数据...")
|
||||
# 清理旧测试数据(幂等)
|
||||
conn.execute(
|
||||
text(
|
||||
"DELETE FROM warehouseOutbound.delivery_receipt_item "
|
||||
"WHERE receipt_no LIKE '202608030%'"
|
||||
)
|
||||
)
|
||||
conn.execute(
|
||||
text(
|
||||
"DELETE FROM warehouseOutbound.delivery_receipt "
|
||||
"WHERE receipt_no LIKE '202608030%'"
|
||||
)
|
||||
)
|
||||
|
||||
# 收货/发货方示范信息(仿用户提供的签收单样式)
|
||||
shipper = {
|
||||
"shipper_name": "北京布莱迪仪器仪表有限公司",
|
||||
"shipper_address": "北京市朝阳区南三环成寿寺路甲135号",
|
||||
"shipper_phone": "010-67690053",
|
||||
"transport_mode": "汽运",
|
||||
"maker": "王制单",
|
||||
}
|
||||
|
||||
if not boxes_by_paichan:
|
||||
print(" !! 警告:CargoTrace.finished_goods_box 无数据,"
|
||||
"将使用合成 (paichan_no, box_no) 作为明细引用(报表 JOIN 可能无产品明细)。")
|
||||
boxes_by_paichan = [("R09999", [1, 2])]
|
||||
|
||||
# ---- 签收单 1:单排产号 ----
|
||||
p1, boxes1 = boxes_by_paichan[0]
|
||||
item_boxes_1 = boxes1[:3] # 最多 3 箱
|
||||
conn.execute(
|
||||
text(
|
||||
"""INSERT INTO warehouseOutbound.delivery_receipt
|
||||
(receipt_no, receipt_date, seq, shipper_name, shipper_address,
|
||||
shipper_phone, receiver_name, receiver_address, receiver_phone,
|
||||
customer_order_no, customer_total_no, transport_mode, ship_date,
|
||||
maker, remark)
|
||||
VALUES
|
||||
(:receipt_no, :receipt_date, :seq, :shipper_name, :shipper_address,
|
||||
:shipper_phone, :receiver_name, :receiver_address, :receiver_phone,
|
||||
:customer_order_no, :customer_total_no, :transport_mode, :ship_date,
|
||||
:maker, :remark)"""
|
||||
),
|
||||
{
|
||||
"receipt_no": "20260803001",
|
||||
"receipt_date": "2026-08-03",
|
||||
"seq": 1,
|
||||
**shipper,
|
||||
"receiver_name": "李家",
|
||||
"receiver_address": "北京市昌平区南口地区李流路三一产业园三一重能1号厂房D1",
|
||||
"receiver_phone": "15230028484",
|
||||
"customer_order_no": "123321",
|
||||
"customer_total_no": "202589472",
|
||||
"ship_date": "2025-12-16",
|
||||
"remark": "单排产号示例",
|
||||
},
|
||||
)
|
||||
for b in item_boxes_1:
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO warehouseOutbound.delivery_receipt_item "
|
||||
"(receipt_no, paichan_no, box_no) VALUES (:r, :p, :b)"
|
||||
),
|
||||
{"r": "20260803001", "p": p1, "b": b},
|
||||
)
|
||||
|
||||
# ---- 签收单 2:多排产号 ----
|
||||
if len(boxes_by_paichan) > 1:
|
||||
p2, boxes2 = boxes_by_paichan[1]
|
||||
else:
|
||||
p2, boxes2 = p1, boxes1
|
||||
box2 = boxes2[0]
|
||||
conn.execute(
|
||||
text(
|
||||
"""INSERT INTO warehouseOutbound.delivery_receipt
|
||||
(receipt_no, receipt_date, seq, shipper_name, shipper_address,
|
||||
shipper_phone, receiver_name, receiver_address, receiver_phone,
|
||||
customer_order_no, customer_total_no, transport_mode, ship_date,
|
||||
maker, remark)
|
||||
VALUES
|
||||
(:receipt_no, :receipt_date, :seq, :shipper_name, :shipper_address,
|
||||
:shipper_phone, :receiver_name, :receiver_address, :receiver_phone,
|
||||
:customer_order_no, :customer_total_no, :transport_mode, :ship_date,
|
||||
:maker, :remark)"""
|
||||
),
|
||||
{
|
||||
"receipt_no": "20260803002",
|
||||
"receipt_date": "2026-08-03",
|
||||
"seq": 2,
|
||||
**shipper,
|
||||
"receiver_name": "张收货",
|
||||
"receiver_address": "上海市浦东新区张江高科技园区博云路2号",
|
||||
"receiver_phone": "13800001111",
|
||||
"customer_order_no": "PO-2026-7788",
|
||||
"customer_total_no": "202589999",
|
||||
"ship_date": "2026-08-02",
|
||||
"remark": "多排产号示例(含两个排产号)",
|
||||
},
|
||||
)
|
||||
# 第一个排产号的第 1 箱
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO warehouseOutbound.delivery_receipt_item "
|
||||
"(receipt_no, paichan_no, box_no) VALUES (:r, :p, :b)"
|
||||
),
|
||||
{"r": "20260803002", "p": p1, "b": boxes1[0]},
|
||||
)
|
||||
# 第二个排产号的第 1 箱(若与第一个相同则跳过重复主键)
|
||||
if (p2, box2) != (p1, boxes1[0]):
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO warehouseOutbound.delivery_receipt_item "
|
||||
"(receipt_no, paichan_no, box_no) VALUES (:r, :p, :b)"
|
||||
),
|
||||
{"r": "20260803002", "p": p2, "b": box2},
|
||||
)
|
||||
|
||||
print(f" 已插入 2 张签收单;明细引用排产号: "
|
||||
f"{[p for p, _ in boxes_by_paichan][:2]}")
|
||||
|
||||
|
||||
def _verify(conn) -> None:
|
||||
print("[3/3] 验证...")
|
||||
heads = conn.execute(
|
||||
text(
|
||||
"SELECT receipt_no, seq, receiver_name, customer_total_no "
|
||||
"FROM warehouseOutbound.delivery_receipt ORDER BY receipt_no"
|
||||
)
|
||||
).fetchall()
|
||||
for r in heads:
|
||||
items = conn.execute(
|
||||
text(
|
||||
"SELECT paichan_no, box_no FROM warehouseOutbound.delivery_receipt_item "
|
||||
"WHERE receipt_no = :r ORDER BY paichan_no, box_no"
|
||||
),
|
||||
{"r": r[0]},
|
||||
).fetchall()
|
||||
print(f" {r[0]} | seq={r[1]} | 收货人={r[2]} | 客户总排行号={r[3]}"
|
||||
f" | 明细 {len(items)} 箱: {[(p, b) for p, b in items]}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
with ENGINE.begin() as conn:
|
||||
need_build = not _table_exists(conn, "delivery_receipt")
|
||||
if need_build:
|
||||
_build_schema(conn)
|
||||
else:
|
||||
print("[1/3] 表已存在,跳过建表。")
|
||||
|
||||
boxes_by_paichan = _probe_boxes(conn)
|
||||
_seed(conn, boxes_by_paichan)
|
||||
_verify(conn)
|
||||
print("完成。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
301
reports/sign_receipt/transform.py
Normal file
301
reports/sign_receipt/transform.py
Normal file
@@ -0,0 +1,301 @@
|
||||
"""签收单数据装配(纯逻辑,可单测)。
|
||||
|
||||
把 run_query 返回的明细行(list[dict])转换成 ReportBro 模板所需的数据结构。
|
||||
|
||||
方案 A(多页分页)下的分工:
|
||||
- 单头标量 + 明细数组 `items` 在此装配,交给 ReportBro 表格自动换页。
|
||||
- 明细数组每个元素是一行(seq/product_name/model/range_/qty/paichan_order/box_no)。
|
||||
- 页脚家具(单据号、第 X/共 Y 页、签收栏、备注)由 stamp_footer() 在生成后
|
||||
用 PyMuPDF 逐页盖印——reportbro-lib 文档页脚 band 不渲染自由文本,且其
|
||||
page_number/page_count 未注册为可解析参数,无法在模板内直接输出页码。
|
||||
|
||||
字段超长保护:表格行高固定 8mm(约容纳 2 行),超出 2 行的字段会被截断加「…」,
|
||||
避免 ReportBro 表格固定行高下文字被裁切丢失可读性。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
from datetime import date, datetime
|
||||
from typing import Any
|
||||
|
||||
# 列宽(mm) 与字号(pt);截断估算需与 _build_template 的列布局对齐。
|
||||
# 可用宽度 = 列宽 - 左右 padding(product_name 有 1.5+1.5 padding)。
|
||||
_COL_METRICS = {
|
||||
"product_name": (47.0, 9.0),
|
||||
"model": (79.0, 9.0),
|
||||
"range_": (40.0, 9.0),
|
||||
"paichan_order": (48.0, 9.0),
|
||||
}
|
||||
PT = 2.834645669
|
||||
_SLACK_MM = 1.0
|
||||
_MAX_LINES = 2 # 8mm 行高约容纳 2 行
|
||||
|
||||
# 调试边框:与模板侧 _text_style 共用同一套取值(FAINT 色 + 0.3mm 线宽),
|
||||
# 当 config report.debug_border=true 时,给 fitz 盖印区(单据号/页码/备注/签字)也画占位框,
|
||||
# 否则这些区域因非 ReportBro 元素而永远没有调试框。
|
||||
_FAINT_RGB = (0.7098, 0.7098, 0.7098)
|
||||
_DBW = 0.3 * PT # 调试边框线宽(mm→pt)
|
||||
|
||||
|
||||
def _debug_border() -> bool:
|
||||
"""读取 report.debug_border 开关,与模板侧 _text_style 共用同一配置。"""
|
||||
try:
|
||||
from core.settings import settings
|
||||
return settings.report.debug_border
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
_CJK_FONT_PATH = os.environ.get("REPORT_CJK_FONT", r"C:/Windows/Fonts/simhei.ttf")
|
||||
_FP = None
|
||||
|
||||
|
||||
def _fpdf_engine(font_size: float):
|
||||
"""惰性创建复用了 simhei 字体的 fpdf2 引擎(c_margin=0,换行宽度公式与 reportbro 对齐)。"""
|
||||
global _FP
|
||||
if _FP is None:
|
||||
try:
|
||||
from fpdf import FPDF
|
||||
pdf = FPDF(unit="pt", format="A4", orientation="P")
|
||||
pdf.add_font("simhei", fname=_CJK_FONT_PATH)
|
||||
pdf.c_margin = 0
|
||||
_FP = pdf
|
||||
except Exception:
|
||||
_FP = None
|
||||
return None
|
||||
_FP.set_font("simhei", size=font_size)
|
||||
return _FP
|
||||
|
||||
|
||||
def _text_width_pt(text: str, font_size: float) -> float:
|
||||
eng = _fpdf_engine(font_size)
|
||||
if eng is not None:
|
||||
return eng.get_string_width(text)
|
||||
cjk = sum(1 for c in text if ord(c) > 0x2E80)
|
||||
asc = len(text) - cjk
|
||||
return (cjk * 3.0 + asc * 1.6) / PT
|
||||
|
||||
|
||||
def _wrap_lines(text: str, col_w_mm: float, font_size: float) -> int:
|
||||
if not text:
|
||||
return 1
|
||||
max_pt = (col_w_mm + _SLACK_MM) * PT
|
||||
w = _text_width_pt(text, font_size)
|
||||
return max(1, math.ceil(w / max_pt))
|
||||
|
||||
|
||||
_CTRL_RE = re.compile(r"[\x00-\x08\x0a-\x1f\x7f]")
|
||||
|
||||
|
||||
def _clean(value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
s = str(value)
|
||||
s = s.replace("\r", " ").replace("\n", " ").replace("\t", " ")
|
||||
s = _CTRL_RE.sub("", s)
|
||||
return s.strip()
|
||||
|
||||
|
||||
def _to_date(value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, datetime):
|
||||
return value.strftime("%Y-%m-%d")
|
||||
if isinstance(value, date):
|
||||
return value.strftime("%Y-%m-%d")
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
return datetime.fromisoformat(value.replace("Z", "+00:00")).strftime("%Y-%m-%d")
|
||||
except ValueError:
|
||||
return value[:10]
|
||||
return str(value)
|
||||
|
||||
|
||||
def _fit(text: str, field: str) -> str:
|
||||
"""超长字段截断到约 2 行并加「…」,防止固定行高下被裁切。"""
|
||||
if not text or field not in _COL_METRICS:
|
||||
return text
|
||||
cw, fs = _COL_METRICS[field]
|
||||
if _wrap_lines(text, cw, fs) <= _MAX_LINES:
|
||||
return text
|
||||
# 二分/逐字符截断到 2 行容量
|
||||
max_pt = (_MAX_LINES * cw + _SLACK_MM) * PT
|
||||
lo, hi = 0, len(text)
|
||||
while lo < hi:
|
||||
mid = (lo + hi + 1) // 2
|
||||
if _text_width_pt(text[:mid], fs) <= max_pt:
|
||||
lo = mid
|
||||
else:
|
||||
hi = mid - 1
|
||||
return text[:max(1, lo - 1)].rstrip() + "…"
|
||||
|
||||
|
||||
class EmptyReceiptError(ValueError):
|
||||
"""查询结果为空 —— 该序列号不存在或无明细。"""
|
||||
|
||||
|
||||
def build_context(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
"""把查询行装配成 ReportBro 模板数据。
|
||||
|
||||
:param rows: query.sql 的结果(每行一个 (箱, 产品) 明细,冗余主表头字段)。
|
||||
:return: dict,含单头标量参数 + 明细数组 items + 合计箱数/件数。
|
||||
:raises EmptyReceiptError: rows 为空。
|
||||
"""
|
||||
if not rows:
|
||||
raise EmptyReceiptError("签收单不存在或无可打印明细,请核对序列号。")
|
||||
|
||||
first = rows[0]
|
||||
total_qty = sum(int(float(r.get("qty") or 0)) for r in rows)
|
||||
boxes = {(r.get("paichan_no"), r.get("box_no")) for r in rows}
|
||||
total_boxes = len(boxes)
|
||||
|
||||
ctx: dict[str, Any] = {
|
||||
"receipt_no": _clean(first.get("receipt_no")),
|
||||
"receipt_date": _to_date(first.get("receipt_date")).replace("-", "/"),
|
||||
"seq": int(first.get("seq") or 0),
|
||||
"shipper_name": _clean(first.get("shipper_name")),
|
||||
"shipper_address": _clean(first.get("shipper_address")),
|
||||
"shipper_phone": _clean(first.get("shipper_phone")),
|
||||
"receiver_name": _clean(first.get("receiver_name")),
|
||||
"receiver_address": _clean(first.get("receiver_address")),
|
||||
"receiver_phone": _clean(first.get("receiver_phone")),
|
||||
"customer_order_no": _clean(first.get("customer_order_no")),
|
||||
"customer_total_no": _clean(first.get("customer_total_no")),
|
||||
"transport_mode": _clean(first.get("transport_mode")),
|
||||
"ship_date": _to_date(first.get("ship_date")).replace("-", "/"),
|
||||
"sign_date": _to_date(first.get("sign_date")).replace("-", "/"),
|
||||
"signed_by": _clean(first.get("signed_by")),
|
||||
"maker": _clean(first.get("maker")),
|
||||
"receipt_remark": _clean(first.get("receipt_remark")),
|
||||
"total_qty": total_qty,
|
||||
"total_boxes": total_boxes,
|
||||
}
|
||||
|
||||
items = []
|
||||
for i, r in enumerate(rows):
|
||||
paichan_order = " ".join([_clean(r.get("paichan_no")), _clean(r.get("order_no"))]).strip()
|
||||
items.append({
|
||||
"seq": str(i + 1),
|
||||
"product_name": _fit(_clean(r.get("product_name")), "product_name"),
|
||||
"model": _fit(_clean(r.get("model")), "model"),
|
||||
"range_": _fit(_clean(r.get("range_")), "range_"),
|
||||
"qty": str(int(float(r.get("qty") or 0))),
|
||||
"paichan_order": _fit(paichan_order, "paichan_order"),
|
||||
"box_no": str(r.get("box_no")) if r.get("box_no") is not None else "",
|
||||
})
|
||||
ctx["items"] = items
|
||||
return ctx
|
||||
|
||||
|
||||
# ----------------------- 页脚盖印(PyMuPDF) -----------------------
|
||||
def stamp_footer(pdf_path: str, ctx: dict[str, Any]) -> None:
|
||||
"""在已生成的 PDF 上逐页盖印页脚家具:单据号(右) + 第 X/共 Y 页(居中)。
|
||||
|
||||
备注与签收栏**不再**盖在页脚 —— 它们由模板表格页脚带(合计行)下方预留的 18mm
|
||||
空间承载,本函数在末页定位「合计」文字后,把备注 + 签收栏精确画在合计正下方,
|
||||
实现「紧跟合计后面、不落入页码页脚区」的效果。
|
||||
|
||||
reportbro-lib 文档页脚 band 不渲染自由文本、page_number/page_count 未注册为可解析
|
||||
参数,故单据号/页码仍在此用 fitz 盖印。
|
||||
"""
|
||||
import fitz # 延迟导入,非签收单报表不依赖
|
||||
|
||||
font_path = os.environ.get("REPORT_CJK_FONT", r"C:/Windows/Fonts/simhei.ttf")
|
||||
if not os.path.exists(font_path):
|
||||
# 字体缺失则跳过盖印,仅告警(不阻断主流程)
|
||||
print(f"⚠️ 页脚盖印跳过:字体不存在 {font_path}")
|
||||
return
|
||||
|
||||
doc = fitz.open(pdf_path)
|
||||
page_count = doc.page_count
|
||||
if page_count == 0:
|
||||
doc.close()
|
||||
return
|
||||
|
||||
font = fitz.Font(fontfile=font_path)
|
||||
W, H = doc[0].rect.width, doc[0].rect.height # pt
|
||||
M = 12 * PT # 左右页边距(mm→pt)
|
||||
right_x = W - M
|
||||
y_pg = H - 6 * PT # 页码/单据号基线:距底 6mm
|
||||
db = _debug_border() # 与模板侧共用开关
|
||||
|
||||
def _db_rect(page, rect):
|
||||
sh = page.new_shape()
|
||||
sh.draw_rect(rect)
|
||||
sh.finish(width=_DBW, color=_FAINT_RGB)
|
||||
sh.commit()
|
||||
|
||||
def _db_vline(page, x, y0, y1):
|
||||
sh = page.new_shape()
|
||||
sh.draw_line(fitz.Point(x, y0), fitz.Point(x, y1))
|
||||
sh.finish(width=_DBW, color=_FAINT_RGB)
|
||||
sh.commit()
|
||||
|
||||
sign_fields = [
|
||||
"收货人签字:____________",
|
||||
"盖章:____________",
|
||||
"签收日期:____________",
|
||||
f"制单:{ctx.get('maker', '')}",
|
||||
]
|
||||
remark = ctx.get("receipt_remark", "")
|
||||
|
||||
for i, page in enumerate(doc):
|
||||
# 单据号(右对齐)
|
||||
docno = ctx.get("receipt_no", "")
|
||||
if docno:
|
||||
pw = font.text_length(docno, fontsize=8)
|
||||
page.insert_text((right_x - pw, y_pg), docno,
|
||||
fontname="simhei", fontfile=font_path, fontsize=8,
|
||||
color=(0.54, 0.54, 0.54))
|
||||
# 第 X 页 / 共 Y 页(居中)
|
||||
pgtext = f"第 {i + 1} 页 / 共 {page_count} 页"
|
||||
pgw = font.text_length(pgtext, fontsize=8)
|
||||
page.insert_text(((W - pgw) / 2, y_pg), pgtext,
|
||||
fontname="simhei", fontfile=font_path, fontsize=8,
|
||||
color=(0.54, 0.54, 0.54))
|
||||
# 调试边框:单据号 + 页码各外包一个 FAINT 框(与模板元素一致)
|
||||
if db:
|
||||
_db_rect(page, fitz.Rect(right_x - pw - 1, y_pg - 2, right_x + 1, y_pg + 6))
|
||||
_db_rect(page, fitz.Rect((W - pgw) / 2 - 1, y_pg - 2, (W + pgw) / 2 + 1, y_pg + 6))
|
||||
|
||||
# 末页:在「合计」正下方预留空间内画备注 + 签收栏(紧跟合计,不进页脚区)
|
||||
last = doc[page_count - 1]
|
||||
words = last.get_text("words")
|
||||
he = [w for w in words if w[4] == "合计"]
|
||||
if he:
|
||||
y_he_bottom = max(w[3] for w in he) # 「合计」文字底边 y(PDF 坐标,原点左下)
|
||||
# 合计下方分隔线由表格底边线承担(footer 带高度=FOOT_H,底边线紧贴合计下方),
|
||||
# 此处不再额外画线,避免与表格底边线重叠成双线。
|
||||
FS = 9 # 字号(pt)
|
||||
# 行高:原「备注 +6mm」与「签收 +11mm」的基线间距为 5mm,现改为两倍 = 10mm。
|
||||
# 备注行 / 签收行各占一个 10mm 行高,文本在各自行内垂直居中。
|
||||
ROW_H = 10 * PT
|
||||
top1 = y_he_bottom + 6 * PT # 备注行顶:距合计底 6mm(保持原视觉起点)
|
||||
y_remark = top1 + (ROW_H + FS) / 2 # 备注文本垂直居中于 [top1, top1+ROW_H]
|
||||
# 备注(若有)
|
||||
if remark:
|
||||
last.insert_text((M, y_remark), f"备注:{remark}",
|
||||
fontname="simhei", fontfile=font_path, fontsize=FS,
|
||||
color=(0.1, 0.1, 0.1))
|
||||
if db:
|
||||
_db_rect(last, fitz.Rect(M, top1, W - M, top1 + ROW_H))
|
||||
# 签收栏:紧随备注行之后,同样行高翻倍 + 垂直居中;4 段等分整行
|
||||
top2 = top1 + ROW_H
|
||||
y_sign = top2 + (ROW_H + FS) / 2
|
||||
step = (W - 2 * M) / 4
|
||||
for j, fld in enumerate(sign_fields):
|
||||
last.insert_text((M + j * step, y_sign), fld,
|
||||
fontname="simhei", fontfile=font_path, fontsize=FS,
|
||||
color=(0.1, 0.1, 0.1))
|
||||
if db:
|
||||
# 签收整行外包框 + 内部 3 条等分竖线(对齐 4 段文字)
|
||||
_db_rect(last, fitz.Rect(M, top2, W - M, top2 + ROW_H))
|
||||
for j in range(1, 4):
|
||||
_db_vline(last, M + j * step, top2, top2 + ROW_H)
|
||||
|
||||
import tempfile
|
||||
tmp_path = pdf_path + ".stamp.tmp"
|
||||
doc.save(tmp_path)
|
||||
doc.close()
|
||||
os.replace(tmp_path, pdf_path)
|
||||
98
reports/sign_receipt/verify_keep_together.py
Normal file
98
reports/sign_receipt/verify_keep_together.py
Normal file
@@ -0,0 +1,98 @@
|
||||
"""验证明细行「整行不跨页」:构造长名称明细,逐页检查每条记录是否完整落在同一页。
|
||||
|
||||
判定方法:每条记录的产品名称/型号中嵌入唯一 ASCII 标记——
|
||||
产品名称:开头 "S{nn}" + 长文本 + 结尾 "E{nn}"
|
||||
型号: 开头 "M{nn}" + 长文本 + 结尾 "N{nn}"
|
||||
同一记录的开头标记与结尾标记必然落在不同的折行上。
|
||||
若整行被跨页拆开,则开头标记在某页、结尾标记在另一页 → 被抓到。
|
||||
(ASCII 字形在 simhei 中存在,避免生僻字缺字形问题。)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
from reportbro import Report # noqa: E402
|
||||
from core.fonts import additional_fonts # noqa: E402
|
||||
import fitz # noqa: E402
|
||||
|
||||
import reports.sign_receipt._build_template as T # noqa: E402
|
||||
|
||||
LONG = "(长文本测试内容用于触发产品名称折行显示效果以验证分页时整行保持完整不被拆开到两页之间)"
|
||||
N = 28
|
||||
items = []
|
||||
for i in range(1, N + 1):
|
||||
nn = f"{i:02d}"
|
||||
pname = f"S{nn}产品名称{LONG}E{nn}"
|
||||
model = f"M{nn}型号标识{LONG}N{nn}"
|
||||
items.append({
|
||||
"seq": str(i),
|
||||
"product_name": pname,
|
||||
"model": model,
|
||||
"range_": "0~100MPa",
|
||||
"qty": "12",
|
||||
"paichan_order": f"R{nn}/ORD{nn}",
|
||||
"box_no": f"B{nn}",
|
||||
})
|
||||
|
||||
ctx = {
|
||||
"receipt_no": "TESTKEEPTOGETHER",
|
||||
"seq": 1,
|
||||
"receipt_date": "2026-08-03",
|
||||
"shipper_name": "发货单位甲",
|
||||
"shipper_address": "北京市朝阳区建国路88号",
|
||||
"shipper_phone": "010-88886666",
|
||||
"receiver_name": "收货人乙",
|
||||
"receiver_address": "上海市浦东新区张江路99号",
|
||||
"receiver_phone": "021-66668888",
|
||||
"customer_order_no": "CUST-001",
|
||||
"customer_total_no": "TOTAL-001",
|
||||
"transport_mode": "公路运输",
|
||||
"ship_date": "2026-08-03",
|
||||
"sign_date": "2026-08-03",
|
||||
"signed_by": "收货人",
|
||||
"maker": "王制单",
|
||||
"receipt_remark": "备注测试",
|
||||
"total_qty": N * 12,
|
||||
"total_boxes": N,
|
||||
"items": items,
|
||||
}
|
||||
|
||||
report_def = T.build_report_definition(ctx)
|
||||
report = Report(report_definition=report_def, data=ctx, additional_fonts=additional_fonts())
|
||||
if report.errors:
|
||||
print("❌ 模板渲染错误:", report.errors)
|
||||
sys.exit(1)
|
||||
|
||||
out = PROJECT_ROOT / "out" / "verify_keep_together.pdf"
|
||||
report.generate_pdf(str(out))
|
||||
print(f"✅ 已生成:{out}")
|
||||
|
||||
doc = fitz.open(str(out))
|
||||
pages_text = [" ".join(w[4] for w in pg.get_text("words")) for pg in doc]
|
||||
|
||||
problems = []
|
||||
for i in range(1, N + 1):
|
||||
nn = f"{i:02d}"
|
||||
s, e = f"S{nn}", f"E{nn}" # 产品名称 头/尾标记
|
||||
m, n = f"M{nn}", f"N{nn}" # 型号 头/尾标记
|
||||
for a, b, label in [(s, e, "产品名称"), (m, n, "型号")]:
|
||||
pa = [p + 1 for p, t in enumerate(pages_text) if a in t]
|
||||
pb = [p + 1 for p, t in enumerate(pages_text) if b in t]
|
||||
if set(pa) != set(pb):
|
||||
problems.append((i, nn, label, f"头{a}页={pa} 尾{b}页={pb}"))
|
||||
|
||||
print(f"总页数:{doc.page_count},明细 {N} 条")
|
||||
if problems:
|
||||
print("❌ 发现跨页拆行的记录:")
|
||||
for p in problems:
|
||||
print(" ", p)
|
||||
doc.close()
|
||||
sys.exit(1)
|
||||
else:
|
||||
print(f"✅ 所有 {N} 条记录的「头/尾标记」均共页,无跨页拆行。"
|
||||
f"(产品名称、型号的长文本折行均随整行一起移动)")
|
||||
doc.close()
|
||||
Reference in New Issue
Block a user