結論：採「多服務（同機）」架構，對應你給的 FR/NFR 與資料模型，

提供可直接啟動的專案骨架、API、DB schema、Docker Compose、systemd 與最小可用範例程式。

一、架構（對應規格）

服務：capture（拍照1/2）、qrcode（USB HID/Serial）、ocr、ingest-api（建檔/查詢/匯出）、web-ui（覆核/查詢）、sync（選配，多站同步）、backup。
資料層：SQLite 單站；要集中改用 PostgreSQL。影像按日期樹狀存，計算 SHA-256。
目標流程：輸入出貨單號→拍照1→OCR→覆核→拍照2→掃 QR→建檔→可查詢/匯出。
命名與路徑：/data/shipments/{YYYY}/{MM}/{DD}/SHP123_image1_20250115T103045Z.jpg；縮圖保存於 thumbs/。
權限與稽核：角色、登入、AuditLogs。
KPI：≥6–8 筆/分、OCR 成功門檻與人工覆核機制。

二、專案目錄（Mono-repo）


shipcap/
  apps/
    ingest_api/           # FastAPI (DB/檔案/匯出)
    web_ui/               # 覆核/查詢頁（FastAPI+Jinja2 或 Qt 選一）
    capture/              # OpenCV 拍照1/2 + 縮圖 + SHA256
    ocr_worker/           # Tesseract/PaddleOCR + 版面模板
    qrcode_listener/      # HID/Serial 讀碼並關聯當前作業
    sync_agent/           # 選配：上傳中央 DB/NAS
  libs/
    db/                   # SQLAlchemy models, DAL
    core/                 # 共用：config、paths、hash、schemas
  deploy/
    docker-compose.yml
    systemd/
      capture.service
      ocr.service
      ingest-api.service
      web-ui.service
  migrations/
  tests/
  .env.example
  config.yaml
  README.md

三、資料庫 Schema（SQLite / PostgreSQL）


-- Shipments
CREATE TABLE IF NOT EXISTS shipments(
  id INTEGER PRIMARY KEY,
  ship_no TEXT UNIQUE NOT NULL,
  customer TEXT, ship_date DATE,
  operator TEXT,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Images
CREATE TABLE IF NOT EXISTS images(
  id INTEGER PRIMARY KEY,
  shipment_id INTEGER NOT NULL REFERENCES shipments(id),
  kind TEXT CHECK(kind IN ('image1','image2')) NOT NULL,
  path TEXT NOT NULL, width INT, height INT,
  size_bytes INT, sha256 TEXT, taken_at TIMESTAMP
);

-- QRCodes
CREATE TABLE IF NOT EXISTS qrcodes(
  id INTEGER PRIMARY KEY,
  shipment_id INTEGER NOT NULL REFERENCES shipments(id),
  payload TEXT NOT NULL, symbology TEXT,
  raw_text TEXT, scanned_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- OCR tables
CREATE TABLE IF NOT EXISTS ocr_tables(
  id INTEGER PRIMARY KEY,
  shipment_id INTEGER NOT NULL REFERENCES shipments(id),
  json_data TEXT NOT NULL, version TEXT,
  ocr_engine TEXT, confidence_avg REAL,
  reviewed_by TEXT, reviewed_at TIMESTAMP
);

-- AuditLogs
CREATE TABLE IF NOT EXISTS audit_logs(
  id INTEGER PRIMARY KEY,
  user TEXT, action TEXT, subject TEXT,
  payload_json TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

（欄位對應規格之資料模型與稽核）

四、API（FastAPI 最小集合）


# apps/ingest_api/main.py
from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from fastapi.responses import StreamingResponse
from pathlib import Path
import hashlib, time, io, zipfile, sqlite3, json, os

app = FastAPI(title="shipcap-ingest")
DB = os.getenv("DB_PATH", "/data/shipcap.db")
ROOT = Path(os.getenv("DATA_ROOT", "/data/shipments"))

def db():
    conn = sqlite3.connect(DB); conn.row_factory = sqlite3.Row
    return conn

@app.post("/api/shipments")
def create_shipment(ship_no: str = Form(...), customer: str = Form(""), ship_date: str = Form(""), operator: str = Form("")):
    with db() as conn:
        conn.execute("INSERT OR IGNORE INTO shipments(ship_no,customer,ship_date,operator) VALUES(?,?,?,?)",
                     (ship_no, customer, ship_date, operator))
        s = conn.execute("SELECT * FROM shipments WHERE ship_no=?", (ship_no,)).fetchone()
    return dict(s)

@app.post("/api/shipments/{ship_no}/images")
def upload_image(ship_no: str, kind: str = Form(...), file: UploadFile = File(...)):
    assert kind in ("image1","image2")
    ts = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())
    subdir = ROOT / time.strftime("%Y/%m/%d")
    subdir.mkdir(parents=True, exist_ok=True)
    fn = f"{ship_no}_{kind}_{ts}.jpg"
    data = file.file.read()
    p = subdir / fn
    p.write_bytes(data)
    sha256 = hashlib.sha256(data).hexdigest()
    with db() as conn:
        s = conn.execute("SELECT id FROM shipments WHERE ship_no=?", (ship_no,)).fetchone()
        if not s: raise HTTPException(404, "shipment not found")
        conn.execute("""INSERT INTO images(shipment_id,kind,path,size_bytes,sha256,taken_at)
                        VALUES(?,?,?,?,?,datetime('now'))""",
                     (s["id"], kind, str(p), len(data), sha256))
    return {"path": str(p), "sha256": sha256}

@app.post("/api/shipments/{ship_no}/ocr")
def post_ocr(ship_no: str, json_data: str = Form(...), engine: str = Form("tesseract"), conf: float = Form(0.0)):
    with db() as conn:
        s = conn.execute("SELECT id FROM shipments WHERE ship_no=?", (ship_no,)).fetchone()
        if not s: raise HTTPException(404, "shipment not found")
        conn.execute("""INSERT INTO ocr_tables(shipment_id,json_data,ocr_engine,confidence_avg)
                        VALUES(?,?,?,?)""", (s["id"], json_data, engine, conf))
    return {"ok": True}

@app.post("/api/shipments/{ship_no}/qrcode")
def add_qr(ship_no: str, payload: str = Form(...), symbology: str = Form("QR"), raw_text: str = Form("")):
    with db() as conn:
        s = conn.execute("SELECT id FROM shipments WHERE ship_no=?", (ship_no,)).fetchone()
        if not s: raise HTTPException(404, "shipment not found")
        conn.execute("""INSERT INTO qrcodes(shipment_id,payload,symbology,raw_text)
                        VALUES(?,?,?,?)""", (s["id"], payload, symbology, raw_text))
    return {"ok": True}

@app.get("/api/shipments")
def list_shipments(q: str = "", ship_no: str = "", start: str = "", end: str = ""):
    sql = "SELECT * FROM shipments WHERE 1=1"; args=[]
    if ship_no: sql+=" AND ship_no=?"; args.append(ship_no)
    if q: sql+=" AND (customer LIKE ? OR operator LIKE ?)"; args += [f"%{q}%", f"%{q}%"]
    if start: sql+=" AND date(created_at)>=date(?)"; args.append(start)
    if end: sql+=" AND date(created_at)<=date(?)"; args.append(end)
    with db() as conn:
        rows = [dict(r) for r in conn.execute(sql, args).fetchall()]
    return rows

@app.get("/api/shipments/{ship_no}/export")
def export_zip(ship_no: str):
    with db() as conn:
        s = conn.execute("SELECT id FROM shipments WHERE ship_no=?", (ship_no,)).fetchone()
        if not s: raise HTTPException(404, "shipment not found")
        imgs = conn.execute("SELECT path FROM images WHERE shipment_id=?", (s["id"],)).fetchall()
        ocr = conn.execute("SELECT json_data FROM ocr_tables WHERE shipment_id=? ORDER BY id DESC LIMIT 1", (s["id"],)).fetchone()
        qrs = conn.execute("SELECT payload FROM qrcodes WHERE shipment_id=?", (s["id"],)).fetchall()
    zbuf = io.BytesIO()
    with zipfile.ZipFile(zbuf, "w", zipfile.ZIP_DEFLATED) as z:
        for r in imgs: z.write(r["path"], Path(r["path"]).name)
        z.writestr("ocr.json", ocr["json_data"] if ocr else "{}")
        z.writestr("qrcodes.txt", "\n".join([r["payload"] for r in qrs]))
    zbuf.seek(0)
    return StreamingResponse(zbuf, media_type="application/zip",
                             headers={"Content-Disposition": f'attachment; filename="{ship_no}.zip"'})

（端點對應規格第 16 節 API 草案與第 5–9 節功能）

五、拍照、OCR、掃碼最小服務

capture：


# apps/capture/capture.py
import cv2, time, hashlib
from pathlib import Path
from datetime import datetime
import requests, os, sys

API = os.getenv("API","http://127.0.0.1:8000")
DATA_ROOT = Path(os.getenv("DATA_ROOT","/data/shipments"))
CAM_INDEX = int(os.getenv("CAM_INDEX","0"))

def shot(ship_no: str, kind: str):
    cap = cv2.VideoCapture(CAM_INDEX, cv2.CAP_V4L2)
    ok, frame = cap.read(); cap.release()
    if not ok: raise SystemExit("camera read fail")
    ts = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
    outdir = DATA_ROOT / datetime.utcnow().strftime("%Y/%m/%d")
    outdir.mkdir(parents=True, exist_ok=True)
    path = outdir / f"{ship_no}_{kind}_{ts}.jpg"
    cv2.imwrite(str(path), frame, [cv2.IMWRITE_JPEG_QUALITY, 92])
    # 回報 API 建檔
    with open(path, "rb") as f:
        requests.post(f"{API}/api/shipments/{ship_no}/images",
                      data={"kind": kind}, files={"file": ("img.jpg", f, "image/jpeg")})
if __name__ == "__main__":
    ship_no, kind = sys.argv[1], sys.argv[2]   # kind=image1|image2
    shot(ship_no, kind)

ocr_worker：


# apps/ocr_worker/worker.py
import pytesseract, json, cv2, sys, os, requests
API = os.getenv("API","http://127.0.0.1:8000")
def run(ship_no: str, img_path: str):
    img = cv2.imread(img_path)
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
    txt = pytesseract.image_to_string(gray, lang="chi_tra+eng")
    # 簡單解析：示意（實務用模板/規則）
    result = {"ship_no": ship_no, "raw": txt}
    requests.post(f"{API}/api/shipments/{ship_no}/ocr",
                  data={"json_data": json.dumps(result, ensure_ascii=False),
                        "engine": "tesseract", "conf": 0.90})
if __name__=="__main__":
    run(sys.argv[1], sys.argv[2])

qrcode_listener：


# apps/qrcode_listener/listener.py
import sys, os, requests, time
API = os.getenv("API","http://127.0.0.1:8000")
CUR_SHIP = None

def set_current(ship_no):  # 可由 web-ui 設定或輸入
    global CUR_SHIP; CUR_SHIP = ship_no
    requests.post(f"{API}/api/shipments", data={"ship_no": ship_no})

def on_scan(payload: str):
    if not CUR_SHIP: return
    requests.post(f"{API}/api/shipments/{CUR_SHIP}/qrcode", data={"payload": payload, "symbology":"QR"})

if __name__=="__main__":
    set_current(sys.argv[1])         # e.g. python listener.py SHP0001
    for line in sys.stdin:           # 示意：從 HID/Serial 驅動換成標準輸入
        on_scan(line.strip())

（拍照、OCR、掃碼職責對應 FR-01/02/03）

六、Web 覆核/查詢最小頁（Jinja2）


# apps/web_ui/main.py
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
import sqlite3, os
app = FastAPI()
DB = os.getenv("DB_PATH","/data/shipcap.db")
tmpl = Jinja2Templates("templates")

@app.get("/")
def index(r: Request, q: str=""):
    conn = sqlite3.connect(DB); conn.row_factory=sqlite3.Row
    rows = conn.execute("SELECT ship_no, customer, created_at FROM shipments ORDER BY id DESC LIMIT 100").fetchall()
    return tmpl.TemplateResponse("index.html", {"request": r, "rows": rows})

（頁面包含查詢、預覽、覆核欄位，對應 7.1）

七、設定、部署與維運

config.yaml（關鍵節點）


data_root: /data/shipments
db_path: /data/shipcap.db
camera:
  index: 0
  resolution: [3264, 2448]
ocr:
  engine: tesseract   # or paddleocr
security:
  jwt_secret: "change_me"
backup:
  daily: "02:00"

（對應設定檔與備份需求）

docker-compose（同機部署）


version: "3.8"
services:
  ingest:
    build: ./apps/ingest_api
    environment:
      - DATA_ROOT=/data/shipments
      - DB_PATH=/data/shipcap.db
    volumes:
      - ./data:/data
    ports: ["8000:8000"]
    command: uvicorn main:app --host 0.0.0.0 --port 8000

  webui:
    build: ./apps/web_ui
    environment:
      - DB_PATH=/data/shipcap.db
    volumes: [ "./data:/data", "./apps/web_ui/templates:/app/templates" ]
    ports: ["8080:8080"]
    command: uvicorn main:app --host 0.0.0.0 --port 8080

  ocr:
    build: ./apps/ocr_worker
    environment: [ "API=http://ingest:8000" ]
    volumes: [ "./data:/data" ]

  capture:
    build: ./apps/capture
    privileged: true
    devices: [ "/dev/video0:/dev/video0" ]
    environment:
      - API=http://ingest:8000
      - DATA_ROOT=/data/shipments
    volumes: [ "./data:/data" ]

（容器化屬「可選」；亦可用 systemd）

systemd 範例


# deploy/systemd/ingest-api.service
[Unit]
Description=shipcap ingest api
After=network.target

[Service]
WorkingDirectory=/opt/shipcap/apps/ingest_api
ExecStart=/usr/bin/uvicorn main:app --host 0.0.0.0 --port 8000
Environment=DATA_ROOT=/data/shipments
Environment=DB_PATH=/data/shipcap.db
Restart=always

[Install]
WantedBy=multi-user.target

（符合規格之 systemd、logrotate、備份排程）

八、最小驗收腳本（單筆 E2E）


# 1) 建單
curl -F ship_no=SHP0001 -F customer=Acme -F ship_date=2025-09-09 -F operator=op1 http://127.0.0.1:8000/api/shipments
# 2) 拍照1/2
python apps/capture/capture.py SHP0001 image1
python apps/capture/capture.py SHP0001 image2
# 3) OCR（示意用剛才的 image1 路徑）
python apps/ocr_worker/worker.py SHP0001 /data/shipments/2025/09/09/SHP0001_image1_*.jpg
# 4) 掃一個 QR payload
curl -F payload="QR:ABC123" http://127.0.0.1:8000/api/shipments/SHP0001/qrcode
# 5) 匯出
curl -L -o SHP0001.zip http://127.0.0.1:8000/api/shipments/SHP0001/export

（對應功能/匯出與驗收）

九、風險與強化點

單據版式多樣：先以模板規則與欄位信心值門檻，低於門檻強制覆核；後續導入 PP-Structure。
反光/陰影：硬體治具與固定光源；拍照前自動曝光/對焦檢查。
多站：本機先存，sync_agent 排程上傳中央 DB/NAS。

需要我把上述骨架打包成 zip 或加上 PaddleOCR 版 worker 與 HID/Serial 讀碼驅動，直接可跑

Handel Class Note AI and ML , ROS

2025年9月8日星期一

出軟體建制的架構跟範例

一、架構（對應規格）

二、專案目錄（Mono-repo）

三、資料庫 Schema（SQLite / PostgreSQL）

四、API（FastAPI 最小集合）

五、拍照、OCR、掃碼最小服務

六、Web 覆核/查詢最小頁（Jinja2）

七、設定、部署與維運

八、最小驗收腳本（單筆 E2E）

九、風險與強化點

沒有留言:

張貼留言

2025年9月8日 星期一

出軟體建制的架構 跟 範例

一、架構（對應規格）

二、專案目錄（Mono-repo）

三、資料庫 Schema（SQLite / PostgreSQL）

四、API（FastAPI 最小集合）

五、拍照、OCR、掃碼最小服務

六、Web 覆核/查詢最小頁（Jinja2）

七、設定、部署與維運

八、最小驗收腳本（單筆 E2E）

九、風險與強化點

沒有留言:

張貼留言

2025年9月8日星期一

出軟體建制的架構跟範例