#!/usr/bin/env python3 """Collect current domestic chain listings from known Korean supplier sites. The output is evidence data. Matching is intentionally conservative: only product titles containing a candidate name/alias are linked automatically. Internal TN sources are not handled here and never count as public sellers. """ from __future__ import annotations import json import re import time from dataclasses import dataclass, asdict from pathlib import Path from urllib.parse import parse_qs, quote, urlencode, urljoin, urlsplit, urlunsplit import requests from bs4 import BeautifulSoup ROOT = Path(__file__).resolve().parent INPUT = ROOT / "candidates-v0.json" OUTPUT = ROOT / "domestic-public-listings.json" HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; TN-Silver-catalog-research/1.0)"} @dataclass(frozen=True) class Product: seller: str title: str url: str source: str def clean(text: str) -> str: return " ".join(text.replace("상품명 :", "").split()) def normalized(text: str) -> str: value = clean(text).lower() value = value.replace("커팅", "컷팅").replace("비형", "b형") return re.sub(r"[^0-9a-z가-힣]+", "", value) def clean_url(url: str) -> str: parts = urlsplit(url) query = parse_qs(parts.query) kept = [] for key in ("it_id", "goodsNo", "product_no", "branduid"): if key in query and query[key]: kept.append((key, query[key][0])) return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(kept), "")) def get(session: requests.Session, url: str) -> BeautifulSoup: response = session.get(url, timeout=30) response.raise_for_status() time.sleep(0.12) return BeautifulSoup(response.content, "html.parser") def add_product(products: dict[tuple[str, str], Product], product: Product) -> None: if not product.title or not product.url: return key = (product.seller, clean_url(product.url)) products[key] = Product(product.seller, clean(product.title), clean_url(product.url), product.source) def collect_kjf(session: requests.Session, products: dict[tuple[str, str], Product]) -> None: for url in ( "https://kjfmall.com/shop/list-10204010", "https://kjfmall.com/shop/list-102040?page=2", ): soup = get(session, url) for item in soup.select("li.sct_li"): title_node = item.select_one(".sct_txt__title") link = item.select_one('a[href*="it_id="]') if title_node and link: add_product(products, Product("KJF한국주얼리부속", title_node.get_text(" ", strip=True), urljoin(url, link["href"]), url)) def collect_daesung(session: requests.Session, products: dict[tuple[str, str], Product]) -> None: for page in range(1, 5): url = f"https://www.e-deasung.co.kr/goods/goods_search.php?page={page}&keyword={quote('체인')}" soup = get(session, url) found = 0 for title_node in soup.select(".item_name"): link = title_node.find_parent("a", href=True) if link and "goodsNo=" in link["href"] and not link["href"].endswith("goodsNo="): found += 1 add_product(products, Product("대성재료상사", title_node.get_text(" ", strip=True), urljoin(url, link["href"]), url)) if not found: break def collect_gmtool(session: requests.Session, products: dict[tuple[str, str], Product]) -> None: for page in range(1, 4): url = f"https://www.gmtool.co.kr/goods/goods_list.php?page={page}&cateCd=020001003" soup = get(session, url) found = 0 for title_node in soup.select(".item_name"): link = title_node.find_parent("a", href=True) if link and "goodsNo=" in link["href"] and not link["href"].endswith("goodsNo="): found += 1 add_product(products, Product("금속공예공구(GMTOOL)", title_node.get_text(" ", strip=True), urljoin(url, link["href"]), url)) if not found: break def collect_cafe24_category( session: requests.Session, products: dict[tuple[str, str], Product], seller: str, base_url: str, category_url: str, item_selector: str, title_selector: str, max_pages: int, ) -> None: for page in range(1, max_pages + 1): join = "&" if "?" in category_url else "?" url = f"{category_url}{join}page={page}" soup = get(session, url) found = 0 for item in soup.select(item_selector): title_node = item.select_one(title_selector) links = item.select('a[href*="/product/"]') link = next((a for a in links if "/list.html" not in a.get("href", "")), None) if title_node and link: found += 1 add_product(products, Product(seller, title_node.get_text(" ", strip=True), urljoin(base_url, link["href"]), url)) if not found: break def collect_known_catalogs(session: requests.Session) -> list[Product]: products: dict[tuple[str, str], Product] = {} collect_kjf(session, products) collect_daesung(session, products) collect_gmtool(session, products) collect_cafe24_category(session, products, "신안제작소", "https://sinan-house.com", "https://sinan-house.com/category/%EC%B2%B4%EC%9D%B8/82/", "li.xans-record-", ".name", 6) collect_cafe24_category(session, products, "신안제작소", "https://sinan-house.com", "https://sinan-house.com/category/%EC%8B%A4%EB%B2%84/188/", "li.xans-record-", ".name", 8) collect_cafe24_category(session, products, "Jewco", "https://kr.jewco.kr", "https://kr.jewco.kr/category/%EC%B2%B4%EC%9D%B8/140/", "li.df-prl-item", ".df-prl-name", 8) return sorted(products.values(), key=lambda p: (p.seller, p.title, p.url)) def terms_for(item: dict) -> list[str]: terms = [item["name"], *item.get("aliases", [])] extras = { "B형각줄체인": ["각B체인", "각B형체인"], "B형고방체인": ["B형 고방체인", "비형 고방체인"], "B형고방다이스체인": ["B형 고방 다이스체인", "비형 고방 다이스체인"], "B형커브체인": ["B형 커브체인", "비형 커브체인"], "A형모줄체인": ["A형 모줄체인"], "모줄B형컷팅체인": ["모줄 B형 컷팅체인", "모줄 비형 컷팅체인"], "스네이크팔각컷팅체인": ["팔각뱀줄컷팅체인", "팔각뱀줄커팅체인"], "팔각체인": ["팔각줄체인"], } terms.extend(extras.get(item["name"], [])) if item["name"] == "각줄체인": terms = [term for term in terms if term != "각체인"] return list(dict.fromkeys(term for term in terms if len(normalized(term)) >= 3)) def match_products(item: dict, products: list[Product]) -> list[Product]: terms = [normalized(term) for term in terms_for(item)] matches = [] for product in products: title = normalized(product.title) if item["name"] == "각줄체인" and ("팔각줄" in title or "직사각" in title): continue if item["name"] == "사각 링크체인" and "직사각" in title: continue if any(term in title for term in terms): matches.append(product) return matches def main() -> None: data = json.loads(INPUT.read_text(encoding="utf-8")) items = [item for group in data["candidate_groups"] for item in group["items"]] with requests.Session() as session: session.headers.update(HEADERS) products = collect_known_catalogs(session) rows = [] for item in items: matches = match_products(item, products) rows.append({ "name": item["name"], "aliases": item.get("aliases", []), "searched_terms": terms_for(item), "public_listings": [asdict(match) for match in matches], }) OUTPUT.write_text(json.dumps({ "checked_on": "2026-08-21", "method": "Known domestic supplier category crawl plus conservative exact-name/alias matching", "catalog_product_count": len(products), "seller_counts": {seller: sum(1 for product in products if product.seller == seller) for seller in sorted({p.seller for p in products})}, "rows": rows, }, ensure_ascii=False, indent=2), encoding="utf-8") print(f"wrote {OUTPUT} ({len(products)} catalog products, {len(rows)} candidates)") if __name__ == "__main__": main()