duxingchen/track
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(services): 核心业务逻辑 — 接收/返工/裂变/记录/标签打印

Browse Source 
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 发送到打标机
This commit is contained in:
duxingchen
2026-08-05 14:00:55 +08:00
parent 79583cf8f4
commit 1fed716829
4 changed files with 415 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
backend/app/services/__init__.py
View File

@ -4,6 +4,7 @@ from app.services.product_service import (
get_product,
create_product,
update_product,
update_overall_status,
get_all_products,
)
from app.services.task_service import (
@ -16,6 +17,7 @@ from app.services.task_service import (
reject_task,
transfer_task,
create_subtask,
add_task_record,
get_all_tasks,
)
@ -25,6 +27,7 @@ __all__ = [
"get_product",
"create_product",
"update_product",
"update_overall_status",
"get_all_products",
# Task
"get_task",
@ -36,5 +39,6 @@ __all__ = [
"reject_task",
"transfer_task",
"create_subtask",
"add_task_record",
"get_all_tasks",
]

208
backend/app/services/label_service.py Normal file
View File

@ -0,0 +1,208 @@
"""标签打印服务 — 工业级精确定位标签 (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}",
}

157
backend/app/services/product_service.py
View File

@ -10,7 +10,7 @@ from app.models.product import Product
from app.models.production_order import ProductionOrder
from app.models.task import Task
from app.schemas.product import ProductCreate, ProductUpdate, ProductResponse, ProductScanResponse
from app.schemas.task import TaskSummaryResponse, TaskResponse
from app.schemas.task import TaskSummaryResponse, TaskResponse, TaskRecordResponse
def _task_to_response(task: Task) -> TaskResponse:
@ -18,6 +18,8 @@ def _task_to_response(task: Task) -> TaskResponse:
return TaskResponse(
id=task.id,
product_id=task.product_id,
product_sn=task.product.serial_number if task.product else "",
product_material=task.product.material_name or task.product.material_id or "" if task.product else "",
parent_task_id=task.parent_task_id,
task_name=task.task_name,
assignee_id=task.assignee_id,
@ -29,6 +31,7 @@ def _task_to_response(task: Task) -> TaskResponse:
completed_at=task.completed_at,
created_at=task.created_at,
child_tasks=[_task_to_response(c) for c in task.child_tasks],
records=[TaskRecordResponse.model_validate(r) for r in (task.records or [])],
)
@ -37,7 +40,7 @@ async def _load_task_tree(db: AsyncSession, product_id: uuid.UUID) -> list[TaskR
# 先取顶层任务
result = await db.execute(
select(Task)
.options(selectinload(Task.child_tasks))
.options(selectinload(Task.child_tasks), selectinload(Task.records))
.where(
Task.product_id == product_id,
Task.parent_task_id.is_(None),
@ -51,7 +54,7 @@ async def _load_task_tree(db: AsyncSession, product_id: uuid.UUID) -> list[TaskR
for child in t.child_tasks:
child_result = await db.execute(
select(Task)
.options(selectinload(Task.child_tasks))
.options(selectinload(Task.child_tasks), selectinload(Task.records))
.where(Task.id == child.id)
)
refreshed = child_result.scalar_one()
@ -98,11 +101,17 @@ async def get_product_by_serial(db: AsyncSession, serial_number: str) -> Product
return ProductScanResponse(
id=product.id,
serial_number=product.serial_number,
external_serial=product.external_serial,
order_id=product.order_id,
order_no=product.order.order_no if product.order else "",
material_id=product.material_id,
material_name=product.material_name,
spec_model=product.spec_model,
category=product.category,
material_type=product.material_type,
parent_product_id=product.parent_product_id,
current_location_id=product.current_location_id,
overall_status=product.overall_status,
status=product.status,
created_at=product.created_at,
top_level_tasks=[
@ -129,23 +138,134 @@ async def get_product(db: AsyncSession, product_id: uuid.UUID) -> Product:
async def create_product(db: AsyncSession, data: ProductCreate) -> ProductResponse:
"""创建产品"""
product = Product(**data.model_dump())
"""创建产品 — 自动生成 16 位 HEX 序列号"""
from app.services.counter_service import ensure_sequence, next_hex_id
from app.models.production_order import ProductionOrder
await ensure_sequence(db)
hex_id = await next_hex_id(db)
# 处理订单: 如果传了 order_no 但没传 order_id，查找或创建
order_id = data.order_id
if not order_id and data.order_no:
result = await db.execute(
select(ProductionOrder).where(ProductionOrder.order_no == data.order_no.strip())
)
existing = result.scalar_one_or_none()
if existing:
order_id = existing.id
else:
new_order = ProductionOrder(order_no=data.order_no.strip())
db.add(new_order)
await db.flush()
order_id = new_order.id
product = Product(
serial_number=hex_id,
order_id=order_id,
material_id=data.material_id,
material_name=data.material_name or None,
spec_model=data.spec_model or None,
category=data.category or None,
material_type=data.material_type or None,
external_serial=data.external_serial,
parent_product_id=data.parent_product_id,
)
db.add(product)
await db.commit()
await db.refresh(product)
return ProductResponse.model_validate(product)
await db.refresh(product, ["order"])
return ProductResponse(
id=product.id,
serial_number=product.serial_number,
external_serial=product.external_serial,
order_id=product.order_id,
order_no=product.order.order_no if product.order else (data.order_no or ""),
material_id=product.material_id,
material_name=product.material_name,
spec_model=product.spec_model,
category=product.category,
material_type=product.material_type,
parent_product_id=product.parent_product_id,
current_location_id=product.current_location_id,
overall_status=product.overall_status,
status=product.status,
created_at=product.created_at,
)
async def update_product(db: AsyncSession, product_id: uuid.UUID, data: ProductUpdate) -> ProductResponse:
"""更新产品"""
from app.models.production_order import ProductionOrder
product = await get_product(db, product_id)
update_data = data.model_dump(exclude_unset=True)
# 处理 order_no → order_id 映射
if "order_no" in update_data:
order_no_val = update_data.pop("order_no")
if order_no_val and order_no_val.strip():
result = await db.execute(
select(ProductionOrder).where(ProductionOrder.order_no == order_no_val.strip())
)
existing = result.scalar_one_or_none()
if existing:
product.order_id = existing.id
else:
new_order = ProductionOrder(order_no=order_no_val.strip())
db.add(new_order)
await db.flush()
product.order_id = new_order.id
else:
product.order_id = None
for field, value in update_data.items():
setattr(product, field, value)
await db.commit()
await db.refresh(product, ["order"])
return ProductResponse(
id=product.id,
serial_number=product.serial_number,
external_serial=product.external_serial,
order_id=product.order_id,
order_no=product.order.order_no if product.order else "",
material_id=product.material_id,
material_name=product.material_name,
spec_model=product.spec_model,
category=product.category,
material_type=product.material_type,
parent_product_id=product.parent_product_id,
current_location_id=product.current_location_id,
overall_status=product.overall_status,
status=product.status,
created_at=product.created_at,
)
VALID_OVERALL_STATUS = {"备货", "生产", "测试", "维修", "在库"}
async def update_overall_status(db: AsyncSession, serial_number: str, status_value: str) -> ProductScanResponse:
"""更新产品宏观状态"""
if status_value not in VALID_OVERALL_STATUS:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"无效状态: {status_value}，合法值: {', '.join(sorted(VALID_OVERALL_STATUS))}",
)
result = await db.execute(
select(Product)
.options(selectinload(Product.order))
.where(Product.serial_number == serial_number)
)
product = result.scalar_one_or_none()
if not product:
raise HTTPException(status_code=404, detail=f"未找到序列号 {serial_number} 的产品")
product.overall_status = status_value
await db.commit()
await db.refresh(product)
return ProductResponse.model_validate(product)
return await get_product_by_serial(db, serial_number)
async def get_all_products(db: AsyncSession, skip: int = 0, limit: int = 50) -> list[ProductResponse]:
@ -158,4 +278,23 @@ async def get_all_products(db: AsyncSession, skip: int = 0, limit: int = 50) ->
.order_by(Product.created_at.desc())
)
products = result.scalars().all()
return [ProductResponse.model_validate(p) for p in products]
return [
ProductResponse(
id=p.id,
serial_number=p.serial_number,
external_serial=p.external_serial,
order_id=p.order_id,
order_no=p.order.order_no if p.order else "",
material_id=p.material_id,
material_name=p.material_name,
spec_model=p.spec_model,
category=p.category,
material_type=p.material_type,
parent_product_id=p.parent_product_id,
current_location_id=p.current_location_id,
overall_status=p.overall_status,
status=p.status,
created_at=p.created_at,
)
for p in products
]

66
backend/app/services/task_service.py
View File

@ -1,13 +1,13 @@
"""任务服务 — 核心业务逻辑：接收、驳回返工、裂变转交、无限嵌套子任务"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from fastapi import HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.models.task import Task, TASK_STATUS_PENDING, TASK_STATUS_WIP, TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED
from app.models.task import Task, TaskRecord, TASK_STATUS_PENDING, TASK_STATUS_WIP, TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED
from app.core.time_utils import get_beijing_time
from app.models.product import Product
from app.models.task_log import TaskLog
from app.schemas.task import (
@ -17,6 +17,8 @@ from app.schemas.task import (
TaskRejectRequest,
TaskTransferRequest,
SubtaskCreate,
TaskRecordCreate,
TaskRecordResponse,
TaskResponse,
TaskCompleteResponse,
TaskTransferResponse,
@ -40,6 +42,7 @@ async def _get_task_or_404(db: AsyncSession, task_id: uuid.UUID) -> Task:
selectinload(Task.child_tasks),
selectinload(Task.parent_task),
selectinload(Task.product),
selectinload(Task.records),
)
.where(Task.id == task_id)
)
@ -56,7 +59,10 @@ async def _get_task_with_children_recursive(db: AsyncSession, task_id: uuid.UUID
"""递归加载任务及其所有子孙任务"""
result = await db.execute(
select(Task)
.options(selectinload(Task.child_tasks))
.options(
selectinload(Task.child_tasks),
selectinload(Task.records),
)
.where(Task.id == task_id)
)
task = result.scalar_one_or_none()
@ -71,7 +77,10 @@ async def _get_task_with_children_recursive(db: AsyncSession, task_id: uuid.UUID
for child in t.child_tasks:
child_result = await db.execute(
select(Task)
.options(selectinload(Task.child_tasks))
.options(
selectinload(Task.child_tasks),
selectinload(Task.records),
)
.where(Task.id == child.id)
)
refreshed_child = child_result.scalar_one()
@ -87,6 +96,8 @@ def _to_response(task: Task) -> TaskResponse:
return TaskResponse(
id=task.id,
product_id=task.product_id,
product_sn=task.product.serial_number if task.product else "",
product_material=task.product.material_name or task.product.material_id or "" if task.product else "",
parent_task_id=task.parent_task_id,
task_name=task.task_name,
assignee_id=task.assignee_id,
@ -98,6 +109,7 @@ def _to_response(task: Task) -> TaskResponse:
completed_at=task.completed_at,
created_at=task.created_at,
child_tasks=[_to_response(c) for c in task.child_tasks],
records=[TaskRecordResponse.model_validate(r) for r in (task.records or [])],
)
@ -187,12 +199,19 @@ async def update_task(db: AsyncSession, task_id: uuid.UUID, data: TaskUpdate) ->
async def get_all_tasks(
db: AsyncSession, product_id: uuid.UUID | None = None, skip: int = 0, limit: int = 50
db: AsyncSession, product_id: uuid.UUID | None = None,
assignee_id: str | None = None, skip: int = 0, limit: int = 50
) -> TaskListResponse:
"""获取任务列表，可按产品筛选"""
stmt = select(Task).options(selectinload(Task.child_tasks))
"""获取任务列表，可按产品/负责人筛选"""
stmt = select(Task).options(
selectinload(Task.child_tasks),
selectinload(Task.records),
selectinload(Task.product),
)
if product_id:
stmt = stmt.where(Task.product_id == product_id)
if assignee_id:
stmt = stmt.where(Task.assignee_id == assignee_id)
stmt = stmt.offset(skip).limit(limit).order_by(Task.created_at.desc())
result = await db.execute(stmt)
@ -225,7 +244,7 @@ async def receive_task(
detail=f"只有待接收(PENDING)状态的任务可接收，当前状态: {task.status}",
)
now = datetime.now(timezone.utc)
now = get_beijing_time()
task.status = TASK_STATUS_WIP
task.received_at = now
@ -269,7 +288,7 @@ async def reject_task(
detail=f"任务状态为 {task.status}，无法驳回",
)
now = datetime.now(timezone.utc)
now = get_beijing_time()
# --- 1. 标记当前任务为已驳回 ---
task.status = TASK_STATUS_REJECTED
@ -364,7 +383,7 @@ async def transfer_task(
detail=f"请等待相关子任务完成：{', '.join(incomplete_names)}",
)
now = datetime.now(timezone.utc)
now = get_beijing_time()
# --- 动作 1：闭环当前节点 ---
task.status = TASK_STATUS_COMPLETED
@ -501,7 +520,7 @@ async def complete_task(
detail=f"请等待相关子任务完成：{', '.join(incomplete_names)}",
)
now = datetime.now(timezone.utc)
now = get_beijing_time()
# --- 3. 标记当前任务为已完成 ---
task.status = TASK_STATUS_COMPLETED
@ -591,3 +610,28 @@ async def create_subtask(
await db.commit()
return _to_response(subtask)
# ============================================================
# 任务进度记录 — 随时备注/传图
# ============================================================
async def add_task_record(
db: AsyncSession, task_id: uuid.UUID, data: TaskRecordCreate
) -> TaskResponse:
"""追加进度记录（备注+图片），不改变任务状态"""
import json
task = await _get_task_with_children_recursive(db, task_id)
record = TaskRecord(
task_id=task_id,
remark=data.remark or None,
images=json.dumps(data.images) if data.images else None,
)
db.add(record)
await db.commit()
await db.refresh(record)
# 重新加载 task 带上新 record
return await get_task(db, task_id)