feat: finish the enrollment protocol's reporting checks

Three items from DECISION §8's enrollment protocol were unimplemented and none
of them needed Tier-2 data:

- Eval reports now record provenance: catalog hash, label-file hash, image hash
  and the code revision, so a number can be traced to the inputs that produced
  it. The catalog hash covers the CSV bytes and the view names, not image bytes:
  a re-encoded view of the same product is the same catalog.
- rebuild_catalog flags cross-SKU near-duplicates as text-discriminated
  siblings. These are the near-unknown case no threshold rejects - the geometry
  is real because the sibling shares the brand's artwork - so they are reported
  rather than refused.
- --shelf-tags turns the completeness rule into a live-ready verdict. Without
  the shelf's SKU list the verdict is 'unknown', not a guess.

Adds test_rebuild_catalog.py (11 tests; the tool had no test file). Suite 261
-> 272. Also corrects score_crop_lanes' docstring, which still described the
live path as averaging local scores when the local scale decides on the newest
crop.
This commit is contained in:
tngo 2026-09-28 22:44:07 +09:00
parent 7fff0cf724
commit e53c1ef59b
4 changed files with 301 additions and 3 deletions

View file

@ -21,6 +21,12 @@ images by default), whose image entries use ``file`` plus
``capture_session_id`` or ``session_id``. Pixel equality is not a leak test:
re-cut crops pass it, so leakage is refused by capture session instead.
Two checks report rather than refuse, because both are properties of the
catalog as a whole: cross-SKU near-duplicates are flagged as
text-discriminated siblings (the near-unknown case only package text
separates), and ``--shelf-tags`` turns the completeness rule into a live-ready
verdict. Without that list the verdict is ``unknown``, not a guess.
python code/rebuild_catalog.py --source /tmp/lopia-clean \
--output stock-lopia-clean --csv stock-lopia/stock.csv
python code/rebuild_catalog.py --source /tmp/lopia-clean \
@ -278,6 +284,95 @@ def audit(
return kept, refused
def text_discriminated_siblings(
kept: Dict[str, List[str]], source: str, names: Optional[Dict[str, str]] = None
) -> List[Dict[str, Any]]:
"""Cross-SKU near-duplicates: the same artwork on two different products.
These are the near-unknown case no threshold rejects. The geometry is real
because the sibling shares the brand's artwork, so only the package text
separates them. They are kept - they are legitimately different products -
but flagged, because a catalog that enrolls one and not the other will
accept the missing one as the enrolled one.
"""
views: List[Tuple[str, str, np.ndarray]] = []
for sku, filenames in sorted(kept.items()):
for filename in filenames:
image = cv2.imread(os.path.join(source, filename))
if image is not None:
views.append((sku, filename, image))
flagged: List[Dict[str, Any]] = []
for index, (sku, filename, image) in enumerate(views):
for other_sku, other_name, other in views[index + 1:]:
if other_sku == sku or image.shape != other.shape:
continue
similarity = cosine(image, other)
if similarity < DUPLICATE_COSINE:
continue
flagged.append({
"sku": sku,
"view": filename,
"sibling_sku": other_sku,
"sibling_view": other_name,
"cosine": round(similarity, 4),
"name_similarity": round(
name_ocr_similarity(
(names or {}).get(sku, ""), (names or {}).get(other_sku, "")
), 3,
),
})
return flagged
def live_ready_verdict(
kept: Dict[str, List[str]], names: Optional[Dict[str, str]], shelf_skus: Optional[Set[str]]
) -> Dict[str, Any]:
"""The enrollment protocol's completeness rule, as a verdict.
A bay is live-ready when every SKU on its shelf tags is enrolled or
explicitly listed as text-only. Without a shelf-tag list the rule cannot be
evaluated, and the verdict says so rather than guessing.
"""
catalog = set(names or {})
enrolled = set(kept)
text_only = catalog - enrolled
if shelf_skus is None:
return {
"live_ready": None,
"reason": "no --shelf-tags given; the completeness rule needs the shelf's SKU list",
"enrolled": len(enrolled),
"text_only": sorted(text_only),
}
missing = sorted(shelf_skus - enrolled - text_only)
return {
"live_ready": not missing,
"reason": (
"every shelf SKU is enrolled or text-only"
if not missing
else "on the shelf but absent from the catalog: " + ", ".join(missing)
),
"enrolled": len(enrolled),
"text_only": sorted(text_only),
"shelf_skus": len(shelf_skus),
"missing_from_catalog": missing,
}
def load_shelf_skus(path: Optional[str]) -> Optional[Set[str]]:
"""SKUs on the shelf's tags: one per line, or a JSON list."""
if not path:
return None
with open(path, encoding="utf-8") as handle:
text = handle.read()
try:
payload = json.loads(text)
except json.JSONDecodeError:
return {line.strip() for line in text.splitlines() if line.strip()}
if isinstance(payload, list):
return {str(item).strip() for item in payload if str(item).strip()}
raise RebuildError(f"{path}: expected a JSON list or one SKU per line")
def write_catalog(kept: Dict[str, List[str]], source: str, output: str, csv_path: Optional[str]) -> None:
"""Copy the kept views and the whole CSV into a staging dir, then swap it in."""
staging = tempfile.mkdtemp(prefix=".catalog-", dir=os.path.dirname(os.path.abspath(output)) or ".")
@ -329,6 +424,9 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
help="image capture metadata (default: SOURCE/manifest.json when present)")
parser.add_argument("--eval-manifest",
help="manifest of evaluation sessions; images captured in them are refused")
parser.add_argument("--shelf-tags",
help="the shelf's SKU list (JSON list or one per line) for the "
"completeness rule; without it the verdict is 'unknown'")
parser.add_argument("--json", help="write the audit here")
args = parser.parse_args(argv)
@ -378,6 +476,8 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
# A row with no kept image is a text-only entry: the text lane still names
# and vetoes it from the CSV alone, so it is a coverage gap, not a refusal.
text_only = sorted(set(names) - set(kept))
siblings = text_discriminated_siblings(kept, args.source, names)
verdict = live_ready_verdict(kept, names, load_shelf_skus(args.shelf_tags))
report = {
"source": args.source,
@ -389,6 +489,8 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
"min_name_ocr": MIN_NAME_OCR,
"ocr": engine.name if ocr_engine is not None else "skipped",
"text_only_skus": text_only,
"text_discriminated_siblings": siblings,
"completeness": verdict,
"skus": {sku: views for sku, views in sorted(kept.items())},
"refused": refused,
}
@ -401,6 +503,17 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
print(f" {sku:<10} {len(views)} view(s): {', '.join(views)}")
if text_only:
print(f"\ntext-only CSV row(s), no enrolled image ({len(text_only)}): {', '.join(text_only)}")
if siblings:
print(f"\ntext-discriminated sibling(s) ({len(siblings)}):")
for row in siblings:
print(f" {row['sku']:<10} {row['view']:<20} ~ {row['sibling_sku']} "
f"({row['sibling_view']}) cosine {row['cosine']}")
if verdict["live_ready"] is None:
print(f"\ncompleteness: unknown - {verdict['reason']}")
elif verdict["live_ready"]:
print(f"\ncompleteness: live-ready ({verdict['reason']})")
else:
print(f"\ncompleteness: NOT live-ready - {verdict['reason']}")
if refused:
print(f"\nrefused {len(refused)} view(s):")
for row in refused: