ソースを参照

adding the {"itemId":itemId, "stockInLineId": stockInLineId} as qr code in Bag1

reset-do-picking-order
コミット
38481fee23
4個のファイルの変更133行の追加717行の削除
  1. +0
    -690
      python/Bag0.py
  2. +118
    -26
      python/Bag1.py
  3. +10
    -1
      src/main/java/com/ffii/fpsms/py/PyController.kt
  4. +5
    -0
      src/main/java/com/ffii/fpsms/py/PyJobOrderListItem.kt

+ 0
- 690
python/Bag0.py ファイルの表示

@@ -1,690 +0,0 @@
#!/usr/bin/env python3
"""
Bag1 – GUI to show FPSMS job orders by plan date.
Uses the public API GET /py/job-orders (no login required).
UI tuned for aged users: larger font, Traditional Chinese labels, prev/next date.

Run: python Bag1.py
"""

import json
import os
import socket
import sys
import tkinter as tk
from datetime import date, timedelta
from tkinter import messagebox, ttk
from typing import Optional

import requests

try:
import serial
except ImportError:
serial = None # type: ignore

DEFAULT_BASE_URL = os.environ.get("FPSMS_BASE_URL", "http://localhost:8090/api")
# When run as PyInstaller exe, save settings next to the exe; otherwise next to script
if getattr(sys, "frozen", False):
_SETTINGS_DIR = os.path.dirname(sys.executable)
else:
_SETTINGS_DIR = os.path.dirname(os.path.abspath(__file__))
SETTINGS_FILE = os.path.join(_SETTINGS_DIR, "bag1_settings.json")

DEFAULT_SETTINGS = {
"api_ip": "localhost",
"api_port": "8090",
"dabag_ip": "",
"dabag_port": "3008",
"laser_ip": "",
"laser_port": "9100",
"label_com": "COM3",
}


def load_settings() -> dict:
"""Load settings from JSON file; return defaults if missing or invalid."""
try:
if os.path.isfile(SETTINGS_FILE):
with open(SETTINGS_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
return {**DEFAULT_SETTINGS, **data}
except Exception:
pass
return dict(DEFAULT_SETTINGS)


def save_settings(settings: dict) -> None:
"""Save settings to JSON file."""
with open(SETTINGS_FILE, "w", encoding="utf-8") as f:
json.dump(settings, f, indent=2, ensure_ascii=False)


def build_base_url(api_ip: str, api_port: str) -> str:
ip = (api_ip or "localhost").strip()
port = (api_port or "8090").strip()
return f"http://{ip}:{port}/api"


def try_printer_connection(printer_name: str, sett: dict) -> bool:
"""Try to connect to the selected printer (TCP IP:port or COM). Returns True if OK."""
if printer_name == "打袋機 DataFlex":
ip = (sett.get("dabag_ip") or "").strip()
port_str = (sett.get("dabag_port") or "9100").strip()
if not ip:
return False
try:
port = int(port_str)
s = socket.create_connection((ip, port), timeout=PRINTER_SOCKET_TIMEOUT)
s.close()
return True
except (socket.error, ValueError, OSError):
return False
if printer_name == "激光機":
ip = (sett.get("laser_ip") or "").strip()
port_str = (sett.get("laser_port") or "9100").strip()
if not ip:
return False
try:
port = int(port_str)
s = socket.create_connection((ip, port), timeout=PRINTER_SOCKET_TIMEOUT)
s.close()
return True
except (socket.error, ValueError, OSError):
return False
if printer_name == "標簽機":
if serial is None:
return False
com = (sett.get("label_com") or "").strip()
if not com:
return False
try:
ser = serial.Serial(com, timeout=1)
ser.close()
return True
except (serial.SerialException, OSError):
return False
return False

# Larger font for aged users (point size)
FONT_SIZE = 16
FONT_SIZE_BUTTONS = 15
FONT_SIZE_QTY = 12 # smaller for 數量 under batch no.
FONT_SIZE_ITEM = 20 # item code and item name (larger for readability)
FONT_FAMILY = "Microsoft JhengHei UI" # Traditional Chinese; fallback to TkDefaultFont
# Column widths: item code own column; item name at least double, wraps in its column
ITEM_CODE_WRAP = 140 # item code column width (long codes wrap under code only)
ITEM_NAME_WRAP = 640 # item name column (double width), wraps under name only

# Light blue theme (softer than pure grey)
BG_TOP = "#E8F4FC"
BG_LIST = "#D4E8F7"
BG_ROOT = "#E1F0FF"
BG_ROW = "#C5E1F5"
BG_ROW_SELECTED = "#6BB5FF" # highlighted when selected (for printing)
# Connection status bar
BG_STATUS_ERROR = "#FFCCCB" # red background when disconnected
FG_STATUS_ERROR = "#B22222" # red text
BG_STATUS_OK = "#90EE90" # light green when connected
FG_STATUS_OK = "#006400" # green text
RETRY_MS = 30 * 1000 # 30 seconds reconnect
REFRESH_MS = 60 * 1000 # 60 seconds refresh when connected
PRINTER_CHECK_MS = 60 * 1000 # 1 minute when printer OK
PRINTER_RETRY_MS = 30 * 1000 # 30 seconds when printer failed
PRINTER_SOCKET_TIMEOUT = 3
DATAFLEX_SEND_TIMEOUT = 10 # seconds when sending ZPL to DataFlex


def _zpl_escape(s: str) -> str:
"""Escape text for ZPL ^FD...^FS (backslash and caret)."""
return s.replace("\\", "\\\\").replace("^", "\\^")


def generate_zpl_dataflex(
batch_no: str,
item_code: str,
item_name: str,
font_regular: str = "E:STXihei.ttf",
font_bold: str = "E:STXihei.ttf",
) -> str:
"""
Generate ZPL for DataFlex label (53 mm media, 90° rotated).
Uses UTF-8 (^CI28) and configurable .TTF fonts for Chinese (e.g. E:STXihei.ttf).
"""
desc = _zpl_escape((item_name or "—").strip())
code = _zpl_escape((item_code or "—").strip())
batch = _zpl_escape(batch_no.strip())
return f"""^XA
^PW420
^LL780
^PO N
^CI28
^FO70,70
^A@R,60,60,{font_regular}^FD{desc}^FS
^FO220,70
^A@R,50,50,{font_bold}^FD{code}^FS
^FO310,70
^A@R,45,45,{font_bold}^FD批次: {batch}^FS
^FO150,420
^BQN,2,6^FDQA,{batch_no}^FS
^XZ"""


def send_zpl_to_dataflex(ip: str, port: int, zpl: str) -> None:
"""Send ZPL to DataFlex printer via TCP. Raises on connection/send error."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(DATAFLEX_SEND_TIMEOUT)
try:
sock.connect((ip, port))
sock.sendall(zpl.encode("utf-8"))
finally:
sock.close()


def format_qty(val) -> str:
"""Format quantity: integer without .0, with thousand separator."""
if val is None:
return "—"
try:
n = float(val)
if n == int(n):
return f"{int(n):,}"
return f"{n:,.2f}".rstrip("0").rstrip(".")
except (TypeError, ValueError):
return str(val)


def batch_no(year: int, job_order_id: int) -> str:
"""Batch no.: B + 4-digit year + jobOrderId zero-padded to 6 digits."""
return f"B{year}{job_order_id:06d}"


def get_font(size: int = FONT_SIZE, bold: bool = False) -> tuple:
try:
return (FONT_FAMILY, size, "bold" if bold else "normal")
except Exception:
return ("TkDefaultFont", size, "bold" if bold else "normal")


def fetch_job_orders(base_url: str, plan_start: date) -> list:
"""Call GET /py/job-orders and return the JSON list."""
url = f"{base_url.rstrip('/')}/py/job-orders"
params = {"planStart": plan_start.isoformat()}
resp = requests.get(url, params=params, timeout=30)
resp.raise_for_status()
return resp.json()


def set_row_highlight(row_frame: tk.Frame, selected: bool) -> None:
"""Set row and all its child widgets to selected or normal background."""
bg = BG_ROW_SELECTED if selected else BG_ROW
row_frame.configure(bg=bg)
for w in row_frame.winfo_children():
if isinstance(w, (tk.Frame, tk.Label)):
w.configure(bg=bg)
for c in w.winfo_children():
if isinstance(c, tk.Label):
c.configure(bg=bg)


def on_job_order_click(jo: dict, batch: str) -> None:
"""Show message and highlight row (keeps printing to selected printer)."""
item_code = jo.get("itemCode") or "—"
item_name = jo.get("itemName") or "—"
messagebox.showinfo(
"工單",
f'已點選:批次 {batch}\n品號 {item_code} {item_name}',
)


def ask_label_count(parent: tk.Tk) -> Optional[str]:
"""
When printer is 標簽機, ask how many labels to print.
Returns "1", "10", "50", "100", "C" (continuous), or None if cancelled.
"""
result = [None] # mutable so inner callback can set it

win = tk.Toplevel(parent)
win.title("標簽列印數量")
win.geometry("360x180")
win.transient(parent)
win.grab_set()
win.configure(bg=BG_TOP)
ttk.Label(win, text="列印多少張標簽?", font=get_font(FONT_SIZE)).pack(pady=(16, 12))
btn_frame = tk.Frame(win, bg=BG_TOP)
btn_frame.pack(pady=8)
for label, value in [("1", "1"), ("10", "10"), ("50", "50"), ("100", "100"), ("連續 (C)", "C")]:
def make_cmd(v):
def cmd():
result[0] = v
win.destroy()
return cmd
ttk.Button(btn_frame, text=label, command=make_cmd(value), width=10).pack(side=tk.LEFT, padx=4)
win.protocol("WM_DELETE_WINDOW", win.destroy)
win.wait_window()
return result[0]


def main() -> None:
settings = load_settings()
base_url_ref = [build_base_url(settings["api_ip"], settings["api_port"])]

root = tk.Tk()
root.title("FP-MTMS Bag v1.1 打袋機")
root.geometry("1120x960")
root.minsize(480, 360)
root.configure(bg=BG_ROOT)

# Style: larger font for aged users; light blue theme
style = ttk.Style()
try:
style.configure(".", font=get_font(FONT_SIZE), background=BG_TOP)
style.configure("TButton", font=get_font(FONT_SIZE_BUTTONS), background=BG_TOP)
style.configure("TLabel", font=get_font(FONT_SIZE), background=BG_TOP)
style.configure("TEntry", font=get_font(FONT_SIZE))
style.configure("TFrame", background=BG_TOP)
except tk.TclError:
pass

# Status bar at top: connection state (no popup on error)
status_frame = tk.Frame(root, bg=BG_STATUS_ERROR, padx=12, pady=6)
status_frame.pack(fill=tk.X)
status_lbl = tk.Label(
status_frame,
text="連接不到服務器",
font=get_font(FONT_SIZE_BUTTONS),
bg=BG_STATUS_ERROR,
fg=FG_STATUS_ERROR,
anchor=tk.CENTER,
)
status_lbl.pack(fill=tk.X)

def set_status_ok():
status_frame.configure(bg=BG_STATUS_OK)
status_lbl.configure(text="連接正常", bg=BG_STATUS_OK, fg=FG_STATUS_OK)

def set_status_error():
status_frame.configure(bg=BG_STATUS_ERROR)
status_lbl.configure(text="連接不到服務器", bg=BG_STATUS_ERROR, fg=FG_STATUS_ERROR)

# Top: left [前一天] [date] [後一天] | right [printer dropdown]
top = tk.Frame(root, padx=12, pady=12, bg=BG_TOP)
top.pack(fill=tk.X)

date_var = tk.StringVar(value=date.today().isoformat())
printer_options = ["打袋機 DataFlex", "標簽機", "激光機"]
printer_var = tk.StringVar(value=printer_options[0])

def go_prev_day() -> None:
try:
d = date.fromisoformat(date_var.get().strip())
date_var.set((d - timedelta(days=1)).isoformat())
load_job_orders(from_user_date_change=True)
except ValueError:
date_var.set(date.today().isoformat())
load_job_orders(from_user_date_change=True)

def go_next_day() -> None:
try:
d = date.fromisoformat(date_var.get().strip())
date_var.set((d + timedelta(days=1)).isoformat())
load_job_orders(from_user_date_change=True)
except ValueError:
date_var.set(date.today().isoformat())
load_job_orders(from_user_date_change=True)

# 前一天 (previous day) with left arrow icon
btn_prev = ttk.Button(top, text="◀ 前一天", command=go_prev_day)
btn_prev.pack(side=tk.LEFT, padx=(0, 8))

# Date field (no "日期:" label); shorter width
date_entry = tk.Entry(
top,
textvariable=date_var,
font=get_font(FONT_SIZE),
width=10,
bg="white",
)
date_entry.pack(side=tk.LEFT, padx=(0, 8), ipady=4)

# 後一天 (next day) with right arrow icon
btn_next = ttk.Button(top, text="後一天 ▶", command=go_next_day)
btn_next.pack(side=tk.LEFT, padx=(0, 8))

# Top right: Setup button + printer selection
right_frame = tk.Frame(top, bg=BG_TOP)
right_frame.pack(side=tk.RIGHT)
ttk.Button(right_frame, text="設定", command=lambda: open_setup_window(root, settings, base_url_ref)).pack(
side=tk.LEFT, padx=(0, 12)
)
# 列印機 label: green when printer connected, red when not (checked periodically)
printer_status_lbl = tk.Label(
right_frame,
text="列印機:",
font=get_font(FONT_SIZE),
bg=BG_STATUS_ERROR,
fg="black",
padx=6,
pady=2,
)
printer_status_lbl.pack(side=tk.LEFT, padx=(0, 4))
printer_combo = ttk.Combobox(
right_frame,
textvariable=printer_var,
values=printer_options,
state="readonly",
width=14,
font=get_font(FONT_SIZE),
)
printer_combo.pack(side=tk.LEFT)

printer_after_ref = [None]

def set_printer_status_ok():
printer_status_lbl.configure(bg=BG_STATUS_OK, fg=FG_STATUS_OK)

def set_printer_status_error():
printer_status_lbl.configure(bg=BG_STATUS_ERROR, fg=FG_STATUS_ERROR)

def check_printer() -> None:
if printer_after_ref[0] is not None:
root.after_cancel(printer_after_ref[0])
printer_after_ref[0] = None
ok = try_printer_connection(printer_var.get(), settings)
if ok:
set_printer_status_ok()
printer_after_ref[0] = root.after(PRINTER_CHECK_MS, check_printer)
else:
set_printer_status_error()
printer_after_ref[0] = root.after(PRINTER_RETRY_MS, check_printer)

def on_printer_selection_changed(*args) -> None:
check_printer()

printer_var.trace_add("write", on_printer_selection_changed)

def open_setup_window(parent_win: tk.Tk, sett: dict, base_url_ref_list: list) -> None:
"""Modal setup: API IP/port, 打袋機/激光機 IP+port, 標簽機 COM port."""
d = tk.Toplevel(parent_win)
d.title("設定")
d.geometry("440x520")
d.transient(parent_win)
d.grab_set()
d.configure(bg=BG_TOP)
f = tk.Frame(d, padx=16, pady=16, bg=BG_TOP)
f.pack(fill=tk.BOTH, expand=True)
grid_row = [0] # use list so inner function can update

def _ensure_dot_in_entry(entry: tk.Entry) -> None:
"""Allow typing dot (.) in Entry when IME or layout blocks it (e.g. 192.168.17.27)."""
def on_key(event):
if event.keysym in ("period", "decimal"):
pos = entry.index(tk.INSERT)
entry.insert(tk.INSERT, ".")
return "break"
entry.bind("<KeyPress>", on_key)

def add_section(label_text: str, key_ip: str | None, key_port: str | None, key_single: str | None):
out = []
ttk.Label(f, text=label_text, font=get_font(FONT_SIZE_BUTTONS)).grid(
row=grid_row[0], column=0, columnspan=2, sticky=tk.W, pady=(8, 2)
)
grid_row[0] += 1
if key_single:
ttk.Label(f, text="COM:").grid(row=grid_row[0], column=0, sticky=tk.W, pady=2)
var = tk.StringVar(value=sett.get(key_single, ""))
e = tk.Entry(f, textvariable=var, width=14, font=get_font(FONT_SIZE), bg="white")
e.grid(row=grid_row[0], column=1, sticky=tk.W, pady=2)
_ensure_dot_in_entry(e)
grid_row[0] += 1
return [(key_single, var)]
if key_ip:
ttk.Label(f, text="IP:").grid(row=grid_row[0], column=0, sticky=tk.W, pady=2)
var_ip = tk.StringVar(value=sett.get(key_ip, ""))
e_ip = tk.Entry(f, textvariable=var_ip, width=22, font=get_font(FONT_SIZE), bg="white")
e_ip.grid(row=grid_row[0], column=1, sticky=tk.W, pady=2)
_ensure_dot_in_entry(e_ip)
grid_row[0] += 1
out.append((key_ip, var_ip))
if key_port:
ttk.Label(f, text="Port:").grid(row=grid_row[0], column=0, sticky=tk.W, pady=2)
var_port = tk.StringVar(value=sett.get(key_port, ""))
e_port = tk.Entry(f, textvariable=var_port, width=12, font=get_font(FONT_SIZE), bg="white")
e_port.grid(row=grid_row[0], column=1, sticky=tk.W, pady=2)
_ensure_dot_in_entry(e_port)
grid_row[0] += 1
out.append((key_port, var_port))
return out

all_vars = []
all_vars.extend(add_section("API 伺服器", "api_ip", "api_port", None))
all_vars.extend(add_section("打袋機 DataFlex", "dabag_ip", "dabag_port", None))
all_vars.extend(add_section("激光機", "laser_ip", "laser_port", None))
all_vars.extend(add_section("標簽機 COM 埠", None, None, "label_com"))

def on_save():
for key, var in all_vars:
sett[key] = var.get().strip()
save_settings(sett)
base_url_ref_list[0] = build_base_url(sett["api_ip"], sett["api_port"])
d.destroy()

btn_f = tk.Frame(d, bg=BG_TOP)
btn_f.pack(pady=12)
ttk.Button(btn_f, text="儲存", command=on_save).pack(side=tk.LEFT, padx=4)
ttk.Button(btn_f, text="取消", command=d.destroy).pack(side=tk.LEFT, padx=4)
d.wait_window()

job_orders_frame = tk.Frame(root, bg=BG_LIST)
job_orders_frame.pack(fill=tk.BOTH, expand=True, padx=12, pady=12)

# Scrollable area for buttons
canvas = tk.Canvas(job_orders_frame, highlightthickness=0, bg=BG_LIST)
scrollbar = ttk.Scrollbar(job_orders_frame, orient=tk.VERTICAL, command=canvas.yview)
inner = tk.Frame(canvas, bg=BG_LIST)

win_id = canvas.create_window((0, 0), window=inner, anchor=tk.NW)
canvas.configure(yscrollcommand=scrollbar.set)

def _on_inner_configure(event):
canvas.configure(scrollregion=canvas.bbox("all"))

def _on_canvas_configure(event):
canvas.itemconfig(win_id, width=event.width)

inner.bind("<Configure>", _on_inner_configure)
canvas.bind("<Configure>", _on_canvas_configure)

# Mouse wheel: make scroll work when hovering over canvas or the list (inner/buttons)
def _on_mousewheel(event):
if getattr(event, "delta", None) is not None:
canvas.yview_scroll(int(-1 * (event.delta / 120)), "units")
elif event.num == 5:
canvas.yview_scroll(1, "units")
elif event.num == 4:
canvas.yview_scroll(-1, "units")

canvas.bind("<MouseWheel>", _on_mousewheel)
inner.bind("<MouseWheel>", _on_mousewheel)
canvas.bind("<Button-4>", _on_mousewheel)
canvas.bind("<Button-5>", _on_mousewheel)
inner.bind("<Button-4>", _on_mousewheel)
inner.bind("<Button-5>", _on_mousewheel)

scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)

# Track which row is highlighted (selected for printing) and which job id
selected_row_holder = [None] # [tk.Frame | None]
selected_jo_id_ref = [None] # [int | None] job order id for selection preservation
last_data_ref = [None] # [list | None] last successful fetch for current date
after_id_ref = [None] # [str | None] root.after id to cancel retry/refresh

def _data_equal(a: Optional[list], b: Optional[list]) -> bool:
if a is None or b is None:
return a is b
if len(a) != len(b):
return False
ids_a = [x.get("id") for x in a]
ids_b = [x.get("id") for x in b]
return ids_a == ids_b

def _build_list_from_data(data: list, plan_start: date, preserve_selection: bool) -> None:
selected_row_holder[0] = None
year = plan_start.year
selected_id = selected_jo_id_ref[0] if preserve_selection else None
found_row = None
for jo in data:
jo_id = jo.get("id")
batch = batch_no(year, jo_id) if jo_id is not None else "—"
item_code = jo.get("itemCode") or "—"
item_name = jo.get("itemName") or "—"
req_qty = jo.get("reqQty")
qty_str = format_qty(req_qty)
# Three columns: batch+數量 | item code (own column) | item name (≥2× width, wraps in column)
row = tk.Frame(inner, bg=BG_ROW, relief=tk.RAISED, bd=2, cursor="hand2", padx=12, pady=10)
row.pack(fill=tk.X, pady=4)

left = tk.Frame(row, bg=BG_ROW)
left.pack(side=tk.LEFT, anchor=tk.NW)
batch_lbl = tk.Label(
left,
text=batch,
font=get_font(FONT_SIZE_BUTTONS),
bg=BG_ROW,
fg="black",
)
batch_lbl.pack(anchor=tk.W)
qty_lbl = None
if qty_str != "—":
qty_lbl = tk.Label(
left,
text=f"數量:{qty_str}",
font=get_font(FONT_SIZE_QTY),
bg=BG_ROW,
fg="black",
)
qty_lbl.pack(anchor=tk.W)

# Column 2: item code only, bigger font, wraps in its own column
code_lbl = tk.Label(
row,
text=item_code,
font=get_font(FONT_SIZE_ITEM),
bg=BG_ROW,
fg="black",
wraplength=ITEM_CODE_WRAP,
justify=tk.LEFT,
anchor=tk.NW,
)
code_lbl.pack(side=tk.LEFT, anchor=tk.NW, padx=(12, 8))

# Column 3: item name only, same bigger font, at least double width, wraps under its own column
name_lbl = tk.Label(
row,
text=item_name or "—",
font=get_font(FONT_SIZE_ITEM),
bg=BG_ROW,
fg="black",
wraplength=ITEM_NAME_WRAP,
justify=tk.LEFT,
anchor=tk.NW,
)
name_lbl.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, anchor=tk.NW)

def _on_click(e, j=jo, b=batch, r=row):
if selected_row_holder[0] is not None:
set_row_highlight(selected_row_holder[0], False)
set_row_highlight(r, True)
selected_row_holder[0] = r
selected_jo_id_ref[0] = j.get("id")
if printer_var.get() == "打袋機 DataFlex":
ip = (settings.get("dabag_ip") or "").strip()
port_str = (settings.get("dabag_port") or "3008").strip()
try:
port = int(port_str)
except ValueError:
port = 3008
if not ip:
messagebox.showerror("打袋機", "請在設定中填寫打袋機 DataFlex 的 IP。")
else:
item_code = j.get("itemCode") or "—"
item_name = j.get("itemName") or "—"
zpl = generate_zpl_dataflex(b, item_code, item_name)
try:
send_zpl_to_dataflex(ip, port, zpl)
messagebox.showinfo("打袋機", f"已送出列印:批次 {b}")
except ConnectionRefusedError:
messagebox.showerror("打袋機", f"無法連線至 {ip}:{port},請確認印表機已開機且 IP 正確。")
except socket.timeout:
messagebox.showerror("打袋機", f"連線逾時 ({ip}:{port}),請檢查網路與連接埠。")
except OSError as err:
messagebox.showerror("打袋機", f"列印失敗:{err}")
elif printer_var.get() == "標簽機":
count = ask_label_count(root)
if count is not None:
if count == "C":
msg = "已選擇連續列印標簽"
else:
msg = f"將列印 {count} 張標簽"
messagebox.showinfo("標簽機", msg)
on_job_order_click(j, b)

for w in (row, left, batch_lbl, code_lbl, name_lbl):
w.bind("<Button-1>", _on_click)
w.bind("<MouseWheel>", _on_mousewheel)
w.bind("<Button-4>", _on_mousewheel)
w.bind("<Button-5>", _on_mousewheel)
if qty_lbl is not None:
qty_lbl.bind("<Button-1>", _on_click)
qty_lbl.bind("<MouseWheel>", _on_mousewheel)
qty_lbl.bind("<Button-4>", _on_mousewheel)
qty_lbl.bind("<Button-5>", _on_mousewheel)
if preserve_selection and selected_id is not None and jo.get("id") == selected_id:
found_row = row
if found_row is not None:
set_row_highlight(found_row, True)
selected_row_holder[0] = found_row

def load_job_orders(from_user_date_change: bool = False) -> None:
if after_id_ref[0] is not None:
root.after_cancel(after_id_ref[0])
after_id_ref[0] = None
date_str = date_var.get().strip()
try:
plan_start = date.fromisoformat(date_str)
except ValueError:
messagebox.showerror("日期錯誤", f"請使用 yyyy-MM-dd 格式。目前:{date_str}")
return
if from_user_date_change:
selected_row_holder[0] = None
selected_jo_id_ref[0] = None
try:
data = fetch_job_orders(base_url_ref[0], plan_start)
except requests.RequestException:
set_status_error()
after_id_ref[0] = root.after(RETRY_MS, lambda: load_job_orders(from_user_date_change=False))
return
set_status_ok()
old_data = last_data_ref[0]
last_data_ref[0] = data
data_changed = not _data_equal(old_data, data)
if data_changed or from_user_date_change:
# Rebuild list: clear and rebuild from current data (last_data_ref already updated)
for w in inner.winfo_children():
w.destroy()
preserve = not from_user_date_change
_build_list_from_data(data, plan_start, preserve_selection=preserve)
if from_user_date_change:
canvas.yview_moveto(0)
after_id_ref[0] = root.after(REFRESH_MS, lambda: load_job_orders(from_user_date_change=False))

# Load default (today) on start; then start printer connection check
root.after(100, lambda: load_job_orders(from_user_date_change=True))
root.after(300, check_printer)

root.mainloop()


if __name__ == "__main__":
main()

+ 118
- 26
python/Bag1.py ファイルの表示

@@ -148,16 +148,25 @@ def generate_zpl_dataflex(
batch_no: str,
item_code: str,
item_name: str,
item_id: Optional[int] = None,
stock_in_line_id: Optional[int] = None,
lot_no: Optional[str] = None,
font_regular: str = "E:STXihei.ttf",
font_bold: str = "E:STXihei.ttf",
) -> str:
"""
Generate ZPL for DataFlex label (53 mm media, 90° rotated).
Uses UTF-8 (^CI28) and configurable .TTF fonts for Chinese (e.g. E:STXihei.ttf).
QR contains {"itemId": xxx, "stockInLineId": xxx} when both present; else QA,{batch_no}.
Label shows lotNo when present, else batch_no.
"""
desc = _zpl_escape((item_name or "—").strip())
code = _zpl_escape((item_code or "—").strip())
batch = _zpl_escape(batch_no.strip())
label_line2 = (lot_no or batch_no or "—").strip()
batch = _zpl_escape(label_line2)
if item_id is not None and stock_in_line_id is not None:
qr_data = _zpl_escape(json.dumps({"itemId": item_id, "stockInLineId": stock_in_line_id}))
else:
qr_data = f"QA,{batch_no}"
return f"""^XA
^PW420
^LL780
@@ -170,7 +179,45 @@ def generate_zpl_dataflex(
^FO310,70
^A@R,45,45,{font_bold}^FD批次: {batch}^FS
^FO150,420
^BQN,2,6^FDQA,{batch_no}^FS
^BQN,2,6^FD{qr_data}^FS
^XZ"""


def generate_zpl_label_small(
batch_no: str,
item_code: str,
item_name: str,
item_id: Optional[int] = None,
stock_in_line_id: Optional[int] = None,
lot_no: Optional[str] = None,
font: str = "ARIALR.TTF",
) -> str:
"""
ZPL for 標簽機. Row 1: item name. Row 2: QR left | item code + lot no (or batch) right.
QR contains {"itemId": xxx, "stockInLineId": xxx} when both present; else batch_no.
Font: ARIALR.TTF.
"""
desc = _zpl_escape((item_name or "—").strip())
code = _zpl_escape((item_code or "—").strip())
label_line2 = (lot_no or batch_no or "—").strip()
label_line2_esc = _zpl_escape(label_line2)
if item_id is not None and stock_in_line_id is not None:
qr_data = _zpl_escape(json.dumps({"itemId": item_id, "stockInLineId": stock_in_line_id}))
else:
qr_data = f"QA,{batch_no}"
return f"""^XA
^CI28
^PW500
^LL500
^FO10,15
^FB480,3,0,L,0
^A@N,38,38,{font}^FD{desc}^FS
^FO10,110
^BQN,2,6^FD{qr_data}^FS
^FO150,110
^A@N,48,48,{font}^FD{code}^FS
^FO150,175
^A@N,40,40,{font}^FD{label_line2_esc}^FS
^XZ"""


@@ -185,6 +232,17 @@ def send_zpl_to_dataflex(ip: str, port: int, zpl: str) -> None:
sock.close()


def send_zpl_to_label_printer(com_port: str, zpl: str) -> None:
"""Send ZPL to 標簽機 via serial COM port. Raises on error."""
if serial is None:
raise RuntimeError("pyserial not installed. Run: pip install pyserial")
ser = serial.Serial(com_port, timeout=5)
try:
ser.write(zpl.encode("utf-8"))
finally:
ser.close()


def load_laser_last_count() -> tuple[int, Optional[str]]:
"""Load last batch count and date from laser counter file. Returns (count, date_str)."""
if not os.path.exists(LASER_COUNTER_FILE):
@@ -278,20 +336,23 @@ def send_job_to_laser(
conn_ref: list,
ip: str,
port: int,
job_order_id: int,
item_id: Optional[int],
stock_in_line_id: Optional[int],
item_code: str,
item_name: str,
) -> tuple[bool, str]:
"""
Send job order number; item code; item name to laser. Keeps connection open for next send.
Send to laser. Standard format: {"itemId": xxx, "stockInLineId": xxx}.
conn_ref: [socket or None] - reused across calls; closed only when switching printer.
Format: {job_order_id};{item_code};{item_name};;
When both item_id and stock_in_line_id present, sends JSON; else fallback: 0;item_code;item_name;;
Returns (success, message).
"""
job_id_str = str(job_order_id) if job_order_id is not None else ""
code_str = (item_code or "").strip().replace(";", ",")
name_str = (item_name or "").strip().replace(";", ",")
reply = f"{job_id_str};{code_str};{name_str};;"
if item_id is not None and stock_in_line_id is not None:
reply = json.dumps({"itemId": item_id, "stockInLineId": stock_in_line_id})
else:
code_str = (item_code or "").strip().replace(";", ",")
name_str = (item_name or "").strip().replace(";", ",")
reply = f"0;{code_str};{name_str};;"
conn = conn_ref[0]
try:
if conn is None:
@@ -307,10 +368,10 @@ def send_job_to_laser(
if data:
ack = data.decode("utf-8", errors="ignore").strip().lower()
if "receive" in ack and "invalid" not in ack:
return True, f"已送出激光機:{job_id_str};{code_str};{name_str}(已確認)"
return True, f"已送出激光機:{reply}(已確認)"
except socket.timeout:
pass
return True, f"已送出激光機:{job_id_str};{code_str};{name_str}"
return True, f"已送出激光機:{reply}"
except (ConnectionRefusedError, socket.timeout, OSError) as e:
if conn_ref[0] is not None:
try:
@@ -329,15 +390,16 @@ def send_job_to_laser_with_retry(
conn_ref: list,
ip: str,
port: int,
job_order_id: int,
item_id: Optional[int],
stock_in_line_id: Optional[int],
item_code: str,
item_name: str,
) -> tuple[bool, str]:
"""Send job to laser; on failure, retry once. Returns (success, message)."""
ok, msg = send_job_to_laser(conn_ref, ip, port, job_order_id, item_code, item_name)
ok, msg = send_job_to_laser(conn_ref, ip, port, item_id, stock_in_line_id, item_code, item_name)
if ok:
return True, msg
ok2, msg2 = send_job_to_laser(conn_ref, ip, port, job_order_id, item_code, item_name)
ok2, msg2 = send_job_to_laser(conn_ref, ip, port, item_id, stock_in_line_id, item_code, item_name)
return ok2, msg2


@@ -776,7 +838,8 @@ def main() -> None:
found_row = None
for jo in data:
jo_id = jo.get("id")
batch = batch_no(year, jo_id) if jo_id is not None else "—"
lot_no_val = jo.get("lotNo")
batch = (lot_no_val or "—").strip() if lot_no_val else "—"
item_code = jo.get("itemCode") or "—"
item_name = jo.get("itemName") or "—"
req_qty = jo.get("reqQty")
@@ -850,7 +913,14 @@ def main() -> None:
else:
item_code = j.get("itemCode") or "—"
item_name = j.get("itemName") or "—"
zpl = generate_zpl_dataflex(b, item_code, item_name)
item_id = j.get("itemId")
stock_in_line_id = j.get("stockInLineId")
lot_no = j.get("lotNo")
zpl = generate_zpl_dataflex(
b, item_code, item_name,
item_id=item_id, stock_in_line_id=stock_in_line_id,
lot_no=lot_no,
)
try:
send_zpl_to_dataflex(ip, port, zpl)
messagebox.showinfo("打袋機", f"已送出列印:批次 {b}")
@@ -861,13 +931,32 @@ def main() -> None:
except OSError as err:
messagebox.showerror("打袋機", f"列印失敗:{err}")
elif printer_var.get() == "標簽機":
count = ask_label_count(root)
if count is not None:
if count == "C":
msg = "已選擇連續列印標簽"
else:
msg = f"將列印 {count} 張標簽"
messagebox.showinfo("標簽機", msg)
com = (settings.get("label_com") or "").strip()
if not com:
messagebox.showerror("標簽機", "請在設定中填寫標簽機 COM 埠。")
else:
count = ask_label_count(root)
if count is not None:
item_code = j.get("itemCode") or "—"
item_name = j.get("itemName") or "—"
item_id = j.get("itemId")
stock_in_line_id = j.get("stockInLineId")
lot_no = j.get("lotNo")
zpl = generate_zpl_label_small(
b, item_code, item_name,
item_id=item_id, stock_in_line_id=stock_in_line_id,
lot_no=lot_no,
)
n = 100 if count == "C" else int(count)
try:
for i in range(n):
send_zpl_to_label_printer(com, zpl)
if i < n - 1:
time.sleep(0.5)
msg = f"已送出列印:{n} 張標簽" if count != "C" else f"已送出列印:{n} 張標簽 (連續)"
messagebox.showinfo("標簽機", msg)
except Exception as err:
messagebox.showerror("標簽機", f"列印失敗:{err}")
elif printer_var.get() == "激光機":
ip = (settings.get("laser_ip") or "").strip()
port_str = (settings.get("laser_port") or "45678").strip()
@@ -880,14 +969,17 @@ def main() -> None:
else:
count = ask_laser_count(root)
if count is not None:
jo_id = j.get("id")
item_id = j.get("itemId")
stock_in_line_id = j.get("stockInLineId")
item_code_val = j.get("itemCode") or ""
item_name_val = j.get("itemName") or ""
n = 100 if count == -1 else count
sent = 0
for i in range(n):
ok, msg = send_job_to_laser_with_retry(
laser_conn_ref, ip, port, jo_id, item_code_val, item_name_val
laser_conn_ref, ip, port,
item_id, stock_in_line_id,
item_code_val, item_name_val,
)
if ok:
sent += 1


+ 10
- 1
src/main/java/com/ffii/fpsms/py/PyController.kt ファイルの表示

@@ -2,6 +2,7 @@ package com.ffii.fpsms.py

import com.ffii.fpsms.modules.jobOrder.entity.JobOrder
import com.ffii.fpsms.modules.jobOrder.entity.JobOrderRepository
import com.ffii.fpsms.modules.stock.entity.StockInLineRepository
import org.springframework.format.annotation.DateTimeFormat
import org.springframework.http.ResponseEntity
import org.springframework.web.bind.annotation.GetMapping
@@ -19,13 +20,14 @@ import java.time.LocalDateTime
@RequestMapping("/py")
open class PyController(
private val jobOrderRepository: JobOrderRepository,
private val stockInLineRepository: StockInLineRepository,
) {

/**
* List job orders by planStart date.
* GET /py/job-orders?planStart=yyyy-MM-dd
* @param planStart Date to filter by (default: today). Format: yyyy-MM-dd
* @return List of job orders with id, code, planStart, itemCode, itemName, reqQty
* @return List of job orders with id, code, planStart, itemCode, itemName, reqQty, stockInLineId, itemId
*/
@GetMapping("/job-orders")
open fun listJobOrders(
@@ -42,6 +44,10 @@ open class PyController(
private fun toListItem(jo: JobOrder): PyJobOrderListItem {
val itemCode = jo.bom?.item?.code ?: jo.bom?.code
val itemName = jo.bom?.name ?: jo.bom?.item?.name
val itemId = jo.bom?.item?.id
val stockInLine = jo.id?.let { stockInLineRepository.findFirstByJobOrder_IdAndDeletedFalse(it) }
val stockInLineId = stockInLine?.id
val lotNo = stockInLine?.lotNo
return PyJobOrderListItem(
id = jo.id!!,
code = jo.code,
@@ -49,6 +55,9 @@ open class PyController(
itemCode = itemCode,
itemName = itemName,
reqQty = jo.reqQty,
stockInLineId = stockInLineId,
itemId = itemId,
lotNo = lotNo,
)
}
}

+ 5
- 0
src/main/java/com/ffii/fpsms/py/PyJobOrderListItem.kt ファイルの表示

@@ -6,6 +6,8 @@ import java.time.LocalDateTime
/**
* Job order list item for Python API (/py/job-orders).
* No login required.
* stockInLineId and itemId are for QR code: {"itemId": xxx, "stockInLineId": xxx}
* lotNo replaces job order no. on the label display.
*/
data class PyJobOrderListItem(
val id: Long,
@@ -14,4 +16,7 @@ data class PyJobOrderListItem(
val itemCode: String?,
val itemName: String?,
val reqQty: BigDecimal?,
val stockInLineId: Long?,
val itemId: Long?,
val lotNo: String?,
)

読み込み中…
キャンセル
保存