From ada00da88b2d09c3511f9f3b7d6296e6a5473bdc Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Mon, 28 Sep 2026 13:41:43 +0900 Subject: [PATCH] fix: require explicit safe shelf enrollment paths --- recognization/code/build_store_from_shelf.py | 303 +++++++++++++++---- 1 file changed, 245 insertions(+), 58 deletions(-) diff --git a/recognization/code/build_store_from_shelf.py b/recognization/code/build_store_from_shelf.py index 589328d..f65ac02 100644 --- a/recognization/code/build_store_from_shelf.py +++ b/recognization/code/build_store_from_shelf.py @@ -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 ".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())