Files
track/backend/app/services/label_service.py
duxingchen 1fed716829 feat(services): 核心业务逻辑 — 接收/返工/裂变/记录/标签打印
task_service:
- receive_task() — PENDING→WIP, 记录 received_at
- reject_task() — →REJECTED + 自动返工闭环
- transfer_task() — →COMPLETED + 多路裂变 + virtual_warehouse入库
- add_task_record() — 图文记录追加
- 全链路 selectinload(Task.records)
- 全部 now 改用 get_beijing_time()

product_service:
- create_product() — PG Sequence 自动生成16位HEX
- update_overall_status() — 宏观状态校验+更新
- 扫码返回完整 task_tree (含 records)
- order_no 自由文本→自动创建 ProductionOrder
- update_product() 支持 order_no 编辑

label_service:
- 480×360 工业级排版 (QR+名/规/单+底部通栏码)
- simhei.ttf 34px/28px + stroke_width 描边
- TSPL Socket 发送到打标机
2026-08-05 14:00:55 +08:00

209 lines
6.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""标签打印服务 — 工业级精确定位标签 (480×360) + TSPL 发送"""
import base64
import socket
from io import BytesIO
from pathlib import Path
from typing import Optional
import qrcode
from PIL import Image, ImageDraw, ImageFont
from app.services.print_config import PrintConfigManager
# ============================================================
# 字体 — 项目内 simhei.ttf
# ============================================================
_FONT_PATH = str(Path(__file__).resolve().parent.parent.parent / "simhei.ttf")
FONT_NORMAL = ImageFont.truetype(_FONT_PATH, 28) # 右侧:名/规/单
FONT_LARGE = ImageFont.truetype(_FONT_PATH, 34) # 底部:序列号
# ============================================================
# 画布 & 坐标常量
# ============================================================
WIDTH, HEIGHT = 480, 360
QR_X, QR_Y = 24, 60
QR_SIZE = 180
TEXT_X = 228
TEXT_MAX_W = WIDTH - TEXT_X - 12 # ~240px
BOTTOM_Y = 260 # 序列号 y
LINE_H = 44 # 28px 行高
# ============================================================
# QR 码
# ============================================================
def _generate_qr(content: str) -> Image.Image:
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_M,
box_size=10,
border=2,
)
qr.add_data(content)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
return img.resize((QR_SIZE, QR_SIZE), Image.NEAREST)
# ============================================================
# 文字绘制 — 带描边 + 自动换行
# ============================================================
def _draw(
draw: ImageDraw.Draw,
text: str,
x: int,
y: int,
font: ImageFont.FreeTypeFont,
max_width: int,
stroke_width: int = 1,
) -> int:
"""
绘制文字,超出 max_width 自动折行。
返回下一行可用的 y 坐标。
"""
if not text:
return y + LINE_H
# 单行不超宽 → 直接画(带描边)
bbox = draw.textbbox((0, 0), text, font=font)
if bbox[2] - bbox[0] <= max_width:
draw.text((x, y), text, font=font, fill="black",
stroke_width=stroke_width, stroke_fill="black")
return y + LINE_H
# 逐字符折行
lines: list[str] = []
current = ""
for char in text:
test = current + char
w = draw.textbbox((0, 0), test, font=font)[2]
if w <= max_width:
current = test
else:
lines.append(current)
current = char
if current:
lines.append(current)
cy = y
for line in lines:
draw.text((x, cy), line, font=font, fill="black",
stroke_width=stroke_width, stroke_fill="black")
cy += LINE_H
return cy
# ============================================================
# 标签绑制 — 工业级精确坐标
# ============================================================
def _create_label(data: dict) -> Image.Image:
"""
480×360 工业级排版:
┌───────────────┬──────────────────────────────┐
│ │ 名: 样品升降台V1J y=60 │ ← 28px sw=1
│ [QR Code] │ 规: PH-B4V1J/类A y=104 │
│ 180×180 │ 单: ORD-2024-001 y=148 │
│ (24, 60) │ │
│ │ │
│ 码: 0000000000000001 y=260 │ ← 34px sw=2
└───────────────┴──────────────────────────────┘
"""
img = Image.new("RGB", (WIDTH, HEIGHT), color="white")
draw = ImageDraw.Draw(img)
serial = data.get("serial_number", "")
# ── QR 码 (24, 60) ──
if serial:
qr_img = _generate_qr(serial)
img.paste(qr_img, (QR_X, QR_Y))
# ── 右侧文字 x=228 — 28px, stroke_width=1 ──
y = 60
name = data.get("material_name", "") or "未命名"
y = _draw(draw, f"名: {name}", TEXT_X, y, FONT_NORMAL, TEXT_MAX_W, stroke_width=1)
spec = data.get("spec_model", "") or "-"
y = _draw(draw, f"规: {spec}", TEXT_X, y, FONT_NORMAL, TEXT_MAX_W, stroke_width=1)
order_no = data.get("order_no", "")
if order_no and str(order_no).strip():
y = _draw(draw, f"单: {str(order_no).strip()}", TEXT_X, y, FONT_NORMAL, TEXT_MAX_W, stroke_width=1)
# ── 底部通栏 (24, 260) — 34px, stroke_width=2 ──
code = serial or "-"
draw.text((24, BOTTOM_Y), f"码: {code}", font=FONT_LARGE, fill="black",
stroke_width=2, stroke_fill="black")
return img
# ============================================================
# 公开 API
# ============================================================
def generate_preview_image(**data) -> str:
"""返回 Base64 JPEG data URL"""
img = _create_label(data)
buf = BytesIO()
img.save(buf, format="JPEG", quality=92)
b64 = base64.b64encode(buf.getvalue()).decode()
return f"data:image/jpeg;base64,{b64}"
def send_to_printer(
copies: int = 1,
printer_ip: Optional[str] = None,
printer_port: Optional[int] = None,
**data,
) -> dict:
"""二值化 → TSPL → Socket 发送"""
printer = PrintConfigManager.get_printer("label_printer")
ip = printer_ip or printer.get("ip", "192.168.9.221")
port = printer_port or printer.get("port", 9100)
img_rgb = _create_label(data)
img_gray = img_rgb.convert("L")
img_bw = img_gray.point(lambda px: 0 if px < 128 else 255, "1")
width_bytes = (img_bw.width + 7) // 8
height_dots = img_bw.height
tspl = (
"SIZE 40 mm, 30 mm\r\n"
"GAP 2 mm, 0 mm\r\n"
"CLS\r\n"
"DIRECTION 1\r\n"
).encode("gbk", errors="replace")
bitmap_cmd = f"BITMAP 0,0,{width_bytes},{height_dots},0,".encode("gbk", errors="replace")
bitmap_data = img_bw.tobytes()
footer = f"\r\nPRINT 1,{copies}\r\n".encode("gbk", errors="replace")
payload = tspl + bitmap_cmd + bitmap_data + footer
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(5)
try:
s.connect((ip, port))
s.sendall(payload)
s.close()
return {
"success": True,
"message": f"打印指令已发送 → {ip}:{port},份数: {copies}",
"printer": f"{ip}:{port}",
}
except Exception as e:
return {
"success": False,
"message": f"打印机连接失败 ({ip}:{port}): {str(e)}",
"printer": f"{ip}:{port}",
}