#!/usr/bin/env python3 """Merge public seller evidence, POM supplier evidence, and chainbook provenance.""" from __future__ import annotations import json import re from collections import defaultdict from pathlib import Path HERE = Path(__file__).resolve().parent SOURCE = json.loads((HERE / "candidates-v0.json").read_text(encoding="utf-8")) PUBLIC = json.loads((HERE / "domestic-public-listings.json").read_text(encoding="utf-8")) POM = json.loads((HERE / "pom-supplier-evidence.json").read_text(encoding="utf-8")) OUTPUT = HERE / "chain-audit-2026-08-21.json" INTERNAL_ORGS = {"VTN", "본사", "(VTN재고)", "(본사재고)"} SUPPLEMENTARY = { "스네이크라운드체인": [ { "seller": "H SILVER", "title": "실버925 라운드 스네이크 2mm 뱀줄 팔찌 SBF102", "url": "https://hsilverstore.com/?idx=1212", "listing_type": "finished", "source": "exact public web search", }, { "seller": "클로첼(지그재그 입점)", "title": "실버925 납작 뱀줄 목걸이 스네이크 라운드", "url": "https://zigzag.kr/catalog/products/145686439", "listing_type": "finished", "source": "exact public web search", }, ], "미라볼체인": [ { "seller": "젬비즈", "title": "완성목걸이 (미라볼체인) 3mm | 실버92.5", "url": "https://www.gembeads.co.kr/shop/shopbrand.html?mcode=007&xcode=014", "listing_type": "finished", "source": "exact public web search; category listing", }, { "seller": "코리아쥬얼리", "title": "3mm 미라볼컷팅 체인 14k 팔찌", "url": "https://www.koreajewelry.co.kr/shop/shopdetail.html?branduid=718577&special=3", "listing_type": "finished", "source": "exact public web search", }, { "seller": "DODOJEWELRY(G마켓)", "title": "14k/18k Chain Collection — 16_미라볼체인 선택", "url": "https://mg.gmarket.co.kr/Item?goodsCode=777694330", "listing_type": "finished", "source": "exact public web search", }, ], } def normalize(value: str) -> str: value = value.lower().replace("커팅", "컷팅").replace("비형", "b형") return re.sub(r"[^0-9a-z가-힣]+", "", value) def listing_type(title: str, seller: str) -> str: product_title = title.replace("목걸이 재료", "").replace("팔찌 재료", "").replace("발찌 재료", "") if re.search(r"목걸이|팔찌|발찌|완성|프리메이드", product_title): return "finished" if seller in {"KJF한국주얼리부속", "금속공예공구(GMTOOL)"}: return "raw" if re.search(r"1m|1미터|50cm|미터체인|단위 판매|체인 \(", title, re.I): return "raw" return "raw" def match_score(name: str, aliases: list[str], listing: dict) -> tuple[int, int, int, str]: title = normalize(listing["title"]) terms = [normalize(name), *[normalize(alias) for alias in aliases]] exact = max((len(term) for term in terms if term and term in title), default=0) finished_penalty = 1 if listing_type(listing["title"], listing["seller"]) == "finished" else 0 return (finished_penalty, -exact, len(listing["title"]), listing["title"]) def representative_listings(name: str, aliases: list[str], listings: list[dict]) -> list[dict]: grouped = defaultdict(list) for item in listings: grouped[item["seller"]].append(item) result = [] for seller, seller_listings in sorted(grouped.items()): best = sorted(seller_listings, key=lambda item: match_score(name, aliases, item))[0].copy() best["listing_type"] = listing_type(best["title"], best["seller"]) best["matched_product_count"] = len(seller_listings) result.append(best) return result def status_for(raw_sellers: list[str], finished_sellers: list[str], internal_suppliers: list[str]) -> tuple[str, str]: if len(raw_sellers) >= 2: return "복수 국내 부자재 판매처 확인", "포함 우선" if len(raw_sellers) == 1 and internal_suppliers: return "국내 부자재 1곳 + 사내 공급처 확인", "포함 검토" if len(raw_sellers) == 1: return "국내 부자재 판매처 1곳 확인", "단일 판매처 명칭 검토" if internal_suppliers: return "사내 공급처만 확인", "현재 주문 가능 여부 재확인" if finished_sellers: return "완제품 판매명만 확인", "부자재 조달 확인 전 보류" return "현재 조달 근거 미확인", "보류" def main() -> None: source_items = [item for group in SOURCE["candidate_groups"] for item in group["items"]] public_by_name = {row["name"]: row for row in PUBLIC["rows"]} pom_by_name = {row["name"]: row for row in POM["rows"]} rows = [] for index, item in enumerate(source_items, 1): name = item["name"] aliases = item.get("aliases", []) public = representative_listings(name, aliases, public_by_name[name]["public_listings"]) public.extend(SUPPLEMENTARY.get(name, [])) internal_suppliers = sorted({ supplier for supplier in pom_by_name[name]["internal_supplier_names"] if supplier not in INTERNAL_ORGS }) raw_sellers = sorted({entry["seller"] for entry in public if entry["listing_type"] == "raw"}) finished_sellers = sorted({entry["seller"] for entry in public if entry["listing_type"] == "finished"}) status, recommendation = status_for(raw_sellers, finished_sellers, internal_suppliers) rows.append({ "index": index, "name": name, "aliases": aliases, "searched_terms": public_by_name[name]["searched_terms"], "public_raw_sellers": raw_sellers, "public_finished_sellers": finished_sellers, "representative_listings": public, "internal_supplier_names": internal_suppliers, "internal_supplier_record_count": len(pom_by_name[name]["matched_records"]), "chainbook_evidence": "chainbook_6" in item.get("sources", []), "status": status, "recommendation": recommendation, "publication_asset_rule": "판매처 사진·설명·상품코드 비복제; 독자 촬영 또는 별도 생성 이미지만 사용", }) counts = defaultdict(int) for row in rows: counts[row["status"]] += 1 result = { "title": "TN 체인 카탈로그 국내 조달 전수조사", "checked_on": "2026-08-21", "decision_question": "77개 후보 중 국내에서 실제 조달 근거가 있고 공개 카탈로그 후보로 검토할 수 있는 체인은 무엇인가?", "method": { "public": "국내 5개 부자재 판매처의 체인 카테고리 1,388개 상품명·직접 URL 대조; 미확인 24개는 이름·별칭 exact web search 추가", "internal": "POM 운영 DB를 read-only로 조회하여 POM 자체가 아닌 실제 supplier organization 명칭 추출", "chainbook": "TN 체인북은 이름·형태 참고 근거로만 사용하며 판매처로 계산하지 않음", "seller_count": "동일 업체의 복수 도메인·입점 채널은 한 판매처로 계산", "limits": "로그인 뒤 재고, 전화 주문 가능 여부, 비공개 도매 거래는 공개 웹만으로 확정할 수 없음", }, "catalog_scope": PUBLIC["seller_counts"], "status_counts": dict(counts), "rows": rows, } OUTPUT.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(f"wrote {OUTPUT} ({len(rows)} rows)") print(json.dumps(result["status_counts"], ensure_ascii=False)) if __name__ == "__main__": main()