fix: require explicit safe shelf enrollment paths
This commit is contained in:
kiyo9w
parent
745a29d979
commit
ada00da88b
1 changed files with 245 additions and 58 deletions
|
|
@ -1,66 +1,253 @@
|
|||
"""
|
||||
Enroll a shelf: the products read off its crops become the store's catalog.
|
||||
Crop a labelled shelf photo into a stock directory.
|
||||
|
||||
This is the onboarding flow a real store would go through once - detect the facings,
|
||||
name the products, and keep their crops as the gallery that recognition matches
|
||||
against. The two demo shelves were built with it; the product lists live in this file
|
||||
because reading a contact sheet is a human step.
|
||||
This is a one-shot offline helper. SKU names and prices are supplied by a human
|
||||
who read a contact sheet; this script does not enroll a phone recording and
|
||||
does not talk to the recognition server.
|
||||
|
||||
python code/build_store_from_shelf.py
|
||||
The Meidiya dairy labels that built stock-mediya are kept as --example mediya.
|
||||
Point --image, --boxes, and --output at real paths. There is no default home
|
||||
directory or /tmp path.
|
||||
|
||||
python code/build_store_from_shelf.py \\
|
||||
--image shelf.jpg --boxes boxes.json --output /path/to/new-stock \\
|
||||
--labels labels.json
|
||||
|
||||
boxes.json is {"boxes": [[x1, y1, x2, y2], ...]} in the image's own pixels;
|
||||
this script does not rescale the photo. labels.json is
|
||||
{"products": [{"sku", "name", "price", "boxes": [index, ...]}]}.
|
||||
An existing output directory is left untouched unless --replace is passed,
|
||||
and even then the old directory is renamed to "<output>.previous", not deleted.
|
||||
"""
|
||||
import json, os, shutil, sys
|
||||
sys.path.insert(0, "code")
|
||||
import cv2
|
||||
from __future__ import annotations
|
||||
|
||||
ROOT = os.path.expanduser("~/dorayaki-data/shelf-images")
|
||||
manifest = json.load(open(os.path.join(ROOT, "manifest.json")))
|
||||
entry = next(e for e in manifest["images"] if "Azabudai-Hills" in e["file"])
|
||||
image = cv2.imread(os.path.join(ROOT, entry["file"]))
|
||||
height, width = image.shape[:2]
|
||||
scale = min(1.0, 1600.0 / max(height, width))
|
||||
image = cv2.resize(image, (int(width*scale), int(height*scale)), interpolation=cv2.INTER_AREA)
|
||||
boxes = json.load(open("/tmp/dorayaki-verify/mediya-boxes.json"))["boxes"]
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
# read off the numbered contact sheet; prices come from the shelf's own tags
|
||||
PRODUCTS = [
|
||||
("sku_m01", "小岩井 低脂肪牛乳", [9, 23, 31]),
|
||||
("sku_m02", "荻山酪農 カフェ・オ・レ", [10]),
|
||||
("sku_m03", "丹那 3.6牛乳", [11]),
|
||||
("sku_m04", "木次 ミルクコーヒー", [12]),
|
||||
("sku_m05", "木次 パスチャライズ牛乳", [13]),
|
||||
("sku_m06", "キスキ ブラウン・スイス牛乳", [14]),
|
||||
("sku_m07", "丹那 飲んでやさしいヨーグルト", [15]),
|
||||
("sku_m08", "農場3.7牛乳 特選", [24]),
|
||||
("sku_m09", "MINOR FIGURES OAT", [26]),
|
||||
("sku_m10", "北海道 低脂肪乳", [27]),
|
||||
("sku_m11", "タカナシ 低温殺菌牛乳", [28]),
|
||||
("sku_m12", "RUDE HEALTH Oat Organic", [29]),
|
||||
("sku_m13", "小岩井 無脂肪牛乳", [30]),
|
||||
("sku_m14", "きすき ノンホモ牛乳", [8]),
|
||||
# ponytail: demo counts/dates are the Meidiya sheet's bookkeeping, not a catalog API.
|
||||
EXAMPLE_RECEIPT = "2026-09-25"
|
||||
EXAMPLE_EXPIRATION = "2026-10-20"
|
||||
EXAMPLE_COUNTS = (24, 8, 16)
|
||||
|
||||
# Named example only. These are the labels read off the Meidiya contact sheet,
|
||||
# not an enrollment of a phone walk.
|
||||
MEDIYA_PRODUCTS = [
|
||||
{"sku": "sku_m01", "name": "小岩井 低脂肪牛乳", "price": 186, "boxes": [9, 23, 31]},
|
||||
{"sku": "sku_m02", "name": "荻山酪農 カフェ・オ・レ", "price": 397, "boxes": [10]},
|
||||
{"sku": "sku_m03", "name": "丹那 3.6牛乳", "price": 214, "boxes": [11]},
|
||||
{"sku": "sku_m04", "name": "木次 ミルクコーヒー", "price": 195, "boxes": [12]},
|
||||
{"sku": "sku_m05", "name": "木次 パスチャライズ牛乳", "price": 240, "boxes": [13]},
|
||||
{"sku": "sku_m06", "name": "キスキ ブラウン・スイス牛乳", "price": 254, "boxes": [14]},
|
||||
{"sku": "sku_m07", "name": "丹那 飲んでやさしいヨーグルト", "price": 225, "boxes": [15]},
|
||||
{"sku": "sku_m08", "name": "農場3.7牛乳 特選", "price": 216, "boxes": [24]},
|
||||
{"sku": "sku_m09", "name": "MINOR FIGURES OAT", "price": 300, "boxes": [26]},
|
||||
{"sku": "sku_m10", "name": "北海道 低脂肪乳", "price": 214, "boxes": [27]},
|
||||
{"sku": "sku_m11", "name": "タカナシ 低温殺菌牛乳", "price": 396, "boxes": [28]},
|
||||
{"sku": "sku_m12", "name": "RUDE HEALTH Oat Organic", "price": 400, "boxes": [29]},
|
||||
{"sku": "sku_m13", "name": "小岩井 無脂肪牛乳", "price": 178, "boxes": [30]},
|
||||
{"sku": "sku_m14", "name": "きすき ノンホモ牛乳", "price": 198, "boxes": [8]},
|
||||
]
|
||||
PRICES = {"sku_m01": 186, "sku_m02": 397, "sku_m03": 214, "sku_m04": 195, "sku_m05": 240,
|
||||
"sku_m06": 254, "sku_m07": 225, "sku_m08": 216, "sku_m09": 300, "sku_m10": 214,
|
||||
"sku_m11": 396, "sku_m12": 400, "sku_m13": 178, "sku_m14": 198}
|
||||
|
||||
store = "stock-mediya"
|
||||
shutil.rmtree(store, ignore_errors=True)
|
||||
os.makedirs(store, exist_ok=True)
|
||||
rows = ["SKU_ID,product_name,receipt_date,expiration_date,number_in,number_out,number_on,image_link,price"]
|
||||
written = []
|
||||
for sku, name, indices in PRODUCTS:
|
||||
saved = 0
|
||||
for position, index in enumerate(indices):
|
||||
box = boxes[index]
|
||||
x1, y1, x2, y2 = [int(v) for v in box]
|
||||
crop = image[max(0, y1):y2, max(0, x1):x2]
|
||||
if crop.size == 0:
|
||||
continue
|
||||
suffix = "" if saved == 0 else f"_{saved}"
|
||||
cv2.imwrite(os.path.join(store, f"{sku}{suffix}.jpeg"), crop)
|
||||
saved += 1
|
||||
if saved == 0:
|
||||
raise SystemExit(f"{sku}: no readable crop; refusing a catalog row with no view")
|
||||
written.append(sku)
|
||||
rows.append(f"{sku},{name},2026-09-25,2026-10-20,24,8,16,'',{PRICES[sku]}")
|
||||
open(os.path.join(store, "stock.csv"), "w").write("\n".join(rows) + "\n")
|
||||
print(f"{store}: {len(written)} products, csv written")
|
||||
|
||||
class BuildStoreError(Exception):
|
||||
"""Refused input. The destination, if it already existed, is unchanged."""
|
||||
|
||||
|
||||
def _is_within(path: str, root: str) -> bool:
|
||||
path = os.path.realpath(path)
|
||||
root = os.path.realpath(root)
|
||||
return path == root or path.startswith(root + os.sep)
|
||||
|
||||
|
||||
def _load_json(path: str):
|
||||
try:
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
return json.load(handle)
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
raise BuildStoreError(f"{path}: {exc}") from exc
|
||||
|
||||
|
||||
def load_boxes(path: str) -> list:
|
||||
payload = _load_json(path)
|
||||
boxes = payload.get("boxes") if isinstance(payload, dict) else None
|
||||
if not isinstance(boxes, list) or not boxes:
|
||||
raise BuildStoreError(f"{path}: expected a non-empty {{\"boxes\": [...]}} list")
|
||||
parsed = []
|
||||
for index, box in enumerate(boxes):
|
||||
if not isinstance(box, (list, tuple)) or len(box) != 4:
|
||||
raise BuildStoreError(f"{path}: box {index} is not [x1, y1, x2, y2]")
|
||||
try:
|
||||
coords = [int(value) for value in box]
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise BuildStoreError(f"{path}: box {index} is not four numbers") from exc
|
||||
x1, y1, x2, y2 = coords
|
||||
if x2 <= x1 or y2 <= y1:
|
||||
raise BuildStoreError(f"{path}: box {index} has no area")
|
||||
parsed.append(coords)
|
||||
return parsed
|
||||
|
||||
|
||||
def load_labels(path: str | None, example: str | None) -> list:
|
||||
if example and path:
|
||||
raise BuildStoreError("pass either --labels or --example, not both")
|
||||
if example == "mediya":
|
||||
return [dict(item) for item in MEDIYA_PRODUCTS]
|
||||
if example:
|
||||
raise BuildStoreError(f"unknown example {example!r}; the only named example is mediya")
|
||||
if not path:
|
||||
raise BuildStoreError("pass --labels, or --example mediya for the demo sheet")
|
||||
payload = _load_json(path)
|
||||
products = payload.get("products") if isinstance(payload, dict) else None
|
||||
if not isinstance(products, list) or not products:
|
||||
raise BuildStoreError(f"{path}: expected a non-empty {{\"products\": [...]}} list")
|
||||
return products
|
||||
|
||||
|
||||
def _sku_name(value) -> str:
|
||||
text = str(value).strip()
|
||||
if not text or text in {".", ".."} or "/" in text or "\\" in text or "\x00" in text:
|
||||
raise BuildStoreError(f"refusing sku {value!r}: it is not a single file name")
|
||||
return text
|
||||
|
||||
|
||||
def _products(labels: list) -> list:
|
||||
seen = set()
|
||||
products = []
|
||||
for item in labels:
|
||||
if not isinstance(item, dict):
|
||||
raise BuildStoreError("each product must be an object with sku, name, price, boxes")
|
||||
sku = _sku_name(item.get("sku", ""))
|
||||
if sku in seen:
|
||||
raise BuildStoreError(f"duplicate sku {sku}")
|
||||
seen.add(sku)
|
||||
name = str(item.get("name", "")).strip()
|
||||
if not name or any(char in name for char in ",\n\r"):
|
||||
raise BuildStoreError(f"{sku}: name must be one CSV field")
|
||||
try:
|
||||
price = int(item["price"])
|
||||
indices = [int(index) for index in item["boxes"]]
|
||||
except (KeyError, TypeError, ValueError) as exc:
|
||||
raise BuildStoreError(f"{sku}: price and boxes must be integers") from exc
|
||||
if price < 0 or not indices:
|
||||
raise BuildStoreError(f"{sku}: price and boxes must be present")
|
||||
products.append({"sku": sku, "name": name, "price": price, "boxes": indices})
|
||||
return products
|
||||
|
||||
|
||||
def check_destination(output: str, image: str, boxes_path: str, labels_path: str | None, replace: bool) -> str:
|
||||
output = os.path.abspath(output)
|
||||
real_output = os.path.realpath(output)
|
||||
if real_output == os.path.realpath(os.sep) or real_output == os.path.realpath(os.getcwd()):
|
||||
raise BuildStoreError("refusing to use the filesystem root or the current directory as the store")
|
||||
parent = os.path.dirname(output)
|
||||
if not parent or not os.path.isdir(parent):
|
||||
raise BuildStoreError(f"{output}: parent directory does not exist")
|
||||
for source in (image, boxes_path, labels_path):
|
||||
if source and (_is_within(source, real_output) or _is_within(real_output, os.path.realpath(source))):
|
||||
raise BuildStoreError(f"refusing to write {output}: it aliases input {source}")
|
||||
if os.path.lexists(output):
|
||||
if not replace:
|
||||
raise BuildStoreError(f"{output}: already exists; pass --replace to move it aside")
|
||||
if os.path.islink(output) or not os.path.isdir(output):
|
||||
raise BuildStoreError(f"{output}: --replace only moves aside a real directory")
|
||||
previous = output + ".previous"
|
||||
if os.path.lexists(previous):
|
||||
raise BuildStoreError(f"{previous}: already exists; move it away before --replace")
|
||||
return output
|
||||
|
||||
|
||||
def build_store(image_path: str, boxes: list, products: list, output: str, replace: bool,
|
||||
boxes_path: str, labels_path: str | None) -> int:
|
||||
output = check_destination(output, image_path, boxes_path, labels_path, replace)
|
||||
import cv2
|
||||
|
||||
image = cv2.imread(image_path)
|
||||
if image is None or getattr(image, "size", 0) == 0:
|
||||
raise BuildStoreError(f"{image_path}: not a readable image")
|
||||
height, width = image.shape[:2]
|
||||
|
||||
parent = os.path.dirname(output)
|
||||
staging = tempfile.mkdtemp(prefix=".build-store-", dir=parent)
|
||||
try:
|
||||
rows = [[
|
||||
"SKU_ID", "product_name", "receipt_date", "expiration_date",
|
||||
"number_in", "number_out", "number_on", "image_link", "price",
|
||||
]]
|
||||
written = []
|
||||
for product in products:
|
||||
saved = 0
|
||||
for index in product["boxes"]:
|
||||
if index < 0 or index >= len(boxes):
|
||||
raise BuildStoreError(f"{product['sku']}: box {index} is outside the boxes file")
|
||||
x1, y1, x2, y2 = boxes[index]
|
||||
x1, y1 = max(0, x1), max(0, y1)
|
||||
x2, y2 = min(width, x2), min(height, y2)
|
||||
crop = image[y1:y2, x1:x2] if x2 > x1 and y2 > y1 else image[0:0, 0:0]
|
||||
if crop.size == 0:
|
||||
raise BuildStoreError(f"{product['sku']}: box {index} does not crop the image")
|
||||
suffix = "" if saved == 0 else f"_{saved}"
|
||||
target = os.path.join(staging, f"{product['sku']}{suffix}.jpeg")
|
||||
if not cv2.imwrite(target, crop):
|
||||
raise BuildStoreError(f"{product['sku']}: failed to write crop {index}")
|
||||
saved += 1
|
||||
written.append(product["sku"])
|
||||
number_in, number_out, number_on = EXAMPLE_COUNTS
|
||||
rows.append([
|
||||
product["sku"], product["name"], EXAMPLE_RECEIPT, EXAMPLE_EXPIRATION,
|
||||
number_in, number_out, number_on, "", product["price"],
|
||||
])
|
||||
with open(os.path.join(staging, "stock.csv"), "w", encoding="utf-8", newline="") as handle:
|
||||
csv.writer(handle).writerows(rows)
|
||||
if not os.path.lexists(output):
|
||||
os.rename(staging, output)
|
||||
staging = None
|
||||
else:
|
||||
previous = output + ".previous"
|
||||
os.rename(output, previous)
|
||||
try:
|
||||
os.rename(staging, output)
|
||||
except OSError:
|
||||
os.rename(previous, output)
|
||||
raise
|
||||
staging = None
|
||||
print(f"previous store kept at {previous}", file=sys.stderr)
|
||||
return len(written)
|
||||
finally:
|
||||
if staging and os.path.isdir(staging):
|
||||
shutil.rmtree(staging)
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
parser.add_argument("--image", required=True, help="shelf photo whose pixels match the box coordinates")
|
||||
parser.add_argument("--boxes", required=True, help='JSON file {"boxes": [[x1, y1, x2, y2], ...]}')
|
||||
parser.add_argument("--output", required=True, help="new stock directory to create")
|
||||
parser.add_argument("--labels", help='JSON file {"products": [{"sku", "name", "price", "boxes"}]}')
|
||||
parser.add_argument("--example", choices=["mediya"], help="use the named demo contact-sheet labels")
|
||||
parser.add_argument("--replace", action="store_true", help="replace an existing output directory")
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parse_args(argv)
|
||||
try:
|
||||
products = _products(load_labels(args.labels, args.example))
|
||||
count = build_store(
|
||||
args.image, load_boxes(args.boxes), products, args.output, args.replace,
|
||||
args.boxes, args.labels,
|
||||
)
|
||||
except BuildStoreError as exc:
|
||||
print(f"build_store_from_shelf: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
print(f"{os.path.abspath(args.output)}: {count} products, csv written")
|
||||
print("phone recordings are not enrolled by this script; labels stay a manual contact sheet")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue