chore: fork from IRIS track 供 LICA 部门独立运行

- 复制来源: /home/yueli/track @ 192c8ee (feature/ai-audit-update)
- 组织隔离目标: LICA
- 端口规划: 前端 8030 / 后端 8031 / 数据库 8032
- 已排除 deploy.sh、deploy_full.sh、docker-compose.prod.yml(IRIS 生产发布脚本)
- 已排除工作区未提交改动,取干净的 192c8ee 状态
This commit is contained in:
2026-09-21 15:56:52 +08:00
commit 3286a11bc7
212 changed files with 44060 additions and 0 deletions

View File

@ -0,0 +1,99 @@
"""标签打印 API — 预览 / 执行 / 打印机配置"""
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from app.services.label_service import generate_preview_image, send_to_printer
from app.services.print_config import PrintConfigManager
from app.services.auth_service import get_current_user
router = APIRouter(prefix="/print", tags=["标签打印"])
# ============================================================
# 请求体
# ============================================================
class LabelPreviewRequest(BaseModel):
serial_number: str = Field(..., min_length=1, description="16位HEX系统ID(二维码内容)")
material_name: str = Field("", description="物料名称")
spec_model: str = Field("", description="规格型号")
order_no: str = Field("", description="订单号(条件渲染)")
class PrintExecuteRequest(LabelPreviewRequest):
copies: int = Field(1, ge=1, le=100, description="打印份数")
printer_ip: str | None = Field(None, description="覆盖配置的打印机 IP")
printer_port: int | None = Field(None, description="覆盖配置的打印机端口")
class PrinterConfigUpdate(BaseModel):
ip: str = Field(..., description="打印机 IP 地址")
port: int = Field(9100, ge=1, le=65535, description="打印机端口")
enabled: bool = Field(True, description="是否启用")
# ============================================================
# 端点
# ============================================================
@router.post("/preview")
def print_preview(data: LabelPreviewRequest) -> dict:
"""生成标签预览图(Base64 JPEG)"""
try:
data_url = generate_preview_image(**data.model_dump())
return {"data_url": data_url}
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"生成预览失败: {str(e)}",
)
@router.post("/execute")
def print_execute(
data: PrintExecuteRequest,
current_user: dict = Depends(get_current_user),
) -> dict:
"""发送打印指令到物理打标机"""
payload = data.model_dump()
copies = payload.pop("copies", 1)
printer_ip = payload.pop("printer_ip", None)
printer_port = payload.pop("printer_port", None)
result = send_to_printer(
copies=copies,
printer_ip=printer_ip,
printer_port=printer_port,
**payload,
)
if not result["success"]:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=result["message"],
)
return result
@router.get("/config")
def get_printer_config() -> dict:
"""获取打印机当前配置"""
return PrintConfigManager.get_config()
@router.post("/config")
def update_printer_config(
data: PrinterConfigUpdate,
current_user: dict = Depends(get_current_user),
) -> dict:
"""更新打印机配置(IP/端口)"""
current = PrintConfigManager.get_config()
current["label_printer"] = {
"ip": data.ip,
"port": data.port,
"enabled": data.enabled,
}
PrintConfigManager.save_config(current)
return {
"message": "打印机配置已更新",
"config": current["label_printer"],
}