Python
유통상품 표준DB 조회
T-Blog-000
2025. 8. 11. 11:23
유통상품 표준DB 조회
개요
유통상품 표준DB는 국내에서 유통되는 상품의 정보를 표준화하여 제공하는 국가 주도 데이터베이스입니다. 대한상공회의소와 한국상품정보관리원이 공동으로 운영하며, 상품의 바코드, 제품명, 이미지, 카테고리 등 다양한 정보를 포함하고 있습니다.
특징
- 표준화된 상품 정보: 일관된 형식으로 바코드, 제품명, 카테고리 등 제공
- 방대한 데이터베이스: 국내 유통되는 대부분의 상품 정보 포함
- 이미지 정보: 제품 이미지 URL 제공 (일부 제한적 접근성)
- 계층적 카테고리: 상품 분류를 계층 구조로 제공
사용법
제공된 Python 코드는 대한민국 유통상품 표준DB에서 상품 정보를 검색하고 추출하는 스크래퍼입니다. 다음은 이 도구의 사용 방법입니다.
Google Colab 소스코드 파일
AllProductKoreaScraper.ipynb
0.04MB
요구 사항
- Python 3.6 이상
- 필요 라이브러리:
requests,beautifulsoup4,lxml,urllib3
설치 방법
- 필요 라이브러리 설치:
pip install requests beautifulsoup4 lxml urllib3- 스크래퍼 코드 저장:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import json
import urllib.parse
import urllib3
import requests
import argparse
import re
from bs4 import BeautifulSoup
from datetime import datetime
# SSL 경고 무시
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
SEARCH_URL = "https://www.allproductkorea.or.kr/products/search"
class AllProductKoreaScraper:
def __init__(self):
self.session = requests.Session()
self.session.verify = False # SSL 검증 비활성화
self.headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
"Accept-Encoding": "gzip, deflate",
"Cache-Control": "no-cache",
"Connection": "keep-alive"
}
def convert_to_original_image(self, image_url: str) -> str:
"""작은 이미지 URL을 원본 이미지 URL로 변환"""
if not image_url or not image_url.strip():
return image_url
# Firebase Storage URL인 경우만 변환
if "firebasestorage.googleapis.com" not in image_url:
return image_url
try:
# 1단계: _숫자_w 패턴 제거
step1 = re.sub(r'_\d+_w(?=\.[a-zA-Z]{2,5})', '', image_url)
# 2단계: 토큰 파라미터 제거
step2 = re.sub(r'&token=[a-f0-9\-]+', '', step1)
return step2
except Exception as e:
print(f"이미지 URL 변환 오류: {e} (URL: {image_url})", file=sys.stderr)
return image_url
def test_image_accessibility(self, image_url: str) -> dict:
"""이미지 URL이 실제로 로드되는지 테스트"""
test_result = {
"url": image_url,
"accessible": False,
"status_code": None,
"content_type": None,
"content_length": None,
"error": None
}
if not image_url or not image_url.strip():
test_result["error"] = "URL이 비어있음"
return test_result
try:
response = self.session.head(image_url, timeout=10)
test_result["status_code"] = response.status_code
test_result["accessible"] = response.status_code == 200
test_result["content_type"] = response.headers.get("content-type", "")
test_result["content_length"] = response.headers.get("content-length")
if test_result["content_length"]:
test_result["content_length"] = int(test_result["content_length"])
except Exception as e:
test_result["error"] = str(e)
return test_result
def build_search_url(self, main_keyword: str, sub_keyword: str = "", page: int = 1, size: int = 10) -> str:
"""검색 URL 생성"""
q = json.dumps({
"mainKeyword": main_keyword,
"subKeyword": sub_keyword
}, ensure_ascii=False)
params = {
"q": q,
"page": page,
"size": size
}
return f"{SEARCH_URL}?{urllib.parse.urlencode(params)}"
def fetch_html(self, url: str) -> str:
"""HTML 페이지 가져오기"""
response = self.session.get(url, headers=self.headers, timeout=30)
response.raise_for_status()
response.encoding = 'utf-8'
return response.text
def extract_pagination_info(self, soup: BeautifulSoup) -> dict:
"""페이지네이션 정보 추출"""
pagination_info = {
"current_page": 1,
"total_pages": 1,
"page_size": 10,
"total_count": 0,
"result_count": 0,
"has_next": False,
"has_prev": False
}
try:
# 총 결과 수 추출
total_element = soup.select_one(".pl_total span")
if total_element:
total_text = total_element.get_text(strip=True)
# "총 : 30건" 형태에서 숫자 추출
total_match = re.search(r'총\s*:\s*(\d+)건', total_text)
if total_match:
pagination_info["total_count"] = int(total_match.group(1))
# 페이지 정보 추출
page_info = soup.select_one(".per_pageinfo")
if page_info:
page_text = page_info.get_text(strip=True)
# "총 3페이지" 형태에서 숫자 추출
page_match = re.search(r'총\s*(\d+)페이지', page_text)
if page_match:
pagination_info["total_pages"] = int(page_match.group(1))
# 현재 페이지 추출 (스크립트에서)
script_tags = soup.find_all('script')
for script in script_tags:
if script.string and 'pagination' in script.string:
# JavaScript에서 페이지 정보 추출 시도
page_match = re.search(r'"page"\s*:\s*(\d+)', script.string)
if page_match:
pagination_info["current_page"] = int(page_match.group(1))
size_match = re.search(r'"size"\s*:\s*(\d+)', script.string)
if size_match:
pagination_info["page_size"] = int(size_match.group(1))
count_match = re.search(r'"resultCount"\s*:\s*(\d+)', script.string)
if count_match:
pagination_info["result_count"] = int(count_match.group(1))
# 이전/다음 페이지 존재 여부
pagination_info["has_prev"] = pagination_info["current_page"] > 1
pagination_info["has_next"] = pagination_info["current_page"] < pagination_info["total_pages"]
except Exception as e:
print(f"페이지네이션 정보 추출 오류: {e}", file=sys.stderr)
return pagination_info
def extract_search_metadata(self, soup: BeautifulSoup) -> dict:
"""검색 메타데이터 추출"""
metadata = {
"title": "",
"breadcrumb": [],
"search_input_placeholder": "",
"page_description": ""
}
try:
# 페이지 제목
title_element = soup.select_one("title")
if title_element:
metadata["title"] = title_element.get_text(strip=True)
# 브레드크럼
breadcrumb_elements = soup.select(".loc_wrap li")
for element in breadcrumb_elements:
text = element.get_text(strip=True)
if text and text != "홈":
metadata["breadcrumb"].append(text)
# 검색 입력창 플레이스홀더
search_input = soup.select_one("#searchText")
if search_input:
metadata["search_input_placeholder"] = search_input.get("placeholder", "")
# 페이지 설명
h3_element = soup.select_one(".h3_wrap h3")
if h3_element:
metadata["page_description"] = h3_element.get_text(strip=True)
except Exception as e:
print(f"메타데이터 추출 오류: {e}", file=sys.stderr)
return metadata
def extract_products(self, soup: BeautifulSoup) -> list:
"""제품 리스트 추출"""
products = []
for li in soup.select(".spl_list li[data-prd-no]"):
try:
product = {
"prdNo": li.get("data-prd-no", "").strip(),
"barcode": "",
"name": "",
"image": "",
"categoryPath": "",
"categoryHierarchy": [],
"link": ""
}
# 이미지 URL (원본 이미지로 변환)
img = li.select_one(".spl_img img")
if img:
original_image_url = img.get("src", "")
# 작은 이미지를 원본 이미지로 변환
converted_image_url = self.convert_to_original_image(original_image_url)
product["image"] = converted_image_url
# 원본과 변환된 URL이 다르면 둘 다 저장
if original_image_url != converted_image_url:
product["original_small_image"] = original_image_url
product["image_conversion_applied"] = True
else:
product["image_conversion_applied"] = False
# 바코드와 제품명
pt = li.select_one(".spl_pt")
if pt:
barcode_elem = pt.select_one("em")
if barcode_elem:
product["barcode"] = barcode_elem.get_text(strip=True)
name_elem = pt.select_one("strong")
if name_elem:
product["name"] = name_elem.get_text(strip=True)
# 카테고리 경로
pm = li.select_one(".spl_pm")
if pm:
category_path = pm.get_text(strip=True)
product["categoryPath"] = category_path
# 카테고리를 배열로도 저장
if category_path:
product["categoryHierarchy"] = [cat.strip() for cat in category_path.split(">")]
# 링크 (추정)
if product["prdNo"]:
product["link"] = f"/products/info/korcham/{product['prdNo']}"
products.append(product)
except Exception as e:
print(f"제품 정보 추출 오류: {e}", file=sys.stderr)
continue
return products
def scrape(self, main_keyword: str, sub_keyword: str = "", page: int = 1, size: int = 10) -> dict:
"""메인 스크래핑 함수"""
url = self.build_search_url(main_keyword, sub_keyword, page, size)
print(f"검색 파라미터:", file=sys.stderr)
print(f" - 주요 키워드: {main_keyword}", file=sys.stderr)
print(f" - 보조 키워드: {sub_keyword}", file=sys.stderr)
print(f" - 페이지: {page}", file=sys.stderr)
print(f" - 사이즈: {size}", file=sys.stderr)
print(f"요청 URL: {url}", file=sys.stderr)
html = self.fetch_html(url)
print(f"HTML 가져오기 완료 ({len(html):,} bytes)", file=sys.stderr)
soup = BeautifulSoup(html, "lxml")
# 모든 데이터 추출
products = self.extract_products(soup)
pagination = self.extract_pagination_info(soup)
metadata = self.extract_search_metadata(soup)
print(f"스크래핑 결과: {len(products)}개 제품", file=sys.stderr)
# 이미지 URL 변환 및 접근성 테스트
image_test_results = []
conversion_count = 0
accessible_count = 0
for i, product in enumerate(products[:3]): # 처음 3개 제품만 테스트
if product.get("image"):
test_result = self.test_image_accessibility(product["image"])
test_result["product_index"] = i
test_result["product_name"] = product.get("name", "이름 없음")
test_result["conversion_applied"] = product.get("image_conversion_applied", False)
if test_result["conversion_applied"]:
conversion_count += 1
if test_result["accessible"]:
accessible_count += 1
image_test_results.append(test_result)
content_length = test_result.get('content_length') or 0
status = "OK" if test_result['accessible'] else "FAIL"
print(f"이미지 테스트 [{i+1}] {product.get('name', '이름없음')[:20]}: {status} ({content_length:,} bytes)", file=sys.stderr)
if image_test_results:
print(f"URL 변환: {conversion_count}/{len(image_test_results)}개", file=sys.stderr)
print(f"접근 가능: {accessible_count}/{len(image_test_results)}개", file=sys.stderr)
# 결과 구성
result = {
"timestamp": datetime.now().isoformat(),
"request": {
"url": url,
"parameters": {
"mainKeyword": main_keyword,
"subKeyword": sub_keyword,
"page": page,
"size": size
}
},
"metadata": metadata,
"pagination": pagination,
"products": {
"count": len(products),
"items": products
},
"image_tests": {
"tested_count": len(image_test_results),
"conversion_count": conversion_count,
"accessible_count": accessible_count,
"results": image_test_results
},
"summary": {
"scraped_count": len(products),
"total_available": pagination.get("total_count", 0),
"current_page": pagination.get("current_page", page),
"total_pages": pagination.get("total_pages", 1),
"has_more_pages": pagination.get("has_next", False)
}
}
return result
def main():
parser = argparse.ArgumentParser(
description="AllProductKorea.or.kr 제품 검색 스크래퍼",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
사용 예시:
python allproductkorea_scraper.py "빙그레 더단백"
python allproductkorea_scraper.py "삼양라면" --page 2 --size 20
python allproductkorea_scraper.py "커피" --sub-keyword "원두" --page 1 --size 50
python allproductkorea_scraper.py --url "https://www.allproductkorea.or.kr/products/search?q=..."
"""
)
# 파라미터 그룹 1: 검색어 방식
search_group = parser.add_argument_group('검색어 파라미터')
search_group.add_argument('main_keyword', nargs='?', help='주요 검색어')
search_group.add_argument('--sub-keyword', '-s', default='', help='보조 검색어 (기본값: 빈 문자열)')
search_group.add_argument('--page', '-p', type=int, default=1, help='페이지 번호 (기본값: 1)')
search_group.add_argument('--size', '-z', type=int, default=10, help='페이지당 결과 수 (기본값: 10)')
# 파라미터 그룹 2: URL 직접 입력
url_group = parser.add_argument_group('URL 직접 입력')
url_group.add_argument('--url', '-u', help='완성된 검색 URL 직접 입력')
# 출력 옵션
output_group = parser.add_argument_group('출력 옵션')
output_group.add_argument('--pretty', action='store_true', help='JSON 출력을 예쁘게 포맷')
output_group.add_argument('--output', '-o', help='결과를 파일로 저장')
args = parser.parse_args()
# 파라미터 검증
if not args.url and not args.main_keyword:
print("오류: 검색어 또는 URL을 제공해야 합니다.", file=sys.stderr)
parser.print_help()
sys.exit(1)
try:
scraper = AllProductKoreaScraper()
if args.url:
# URL에서 파라미터 추출
parsed_url = urllib.parse.urlparse(args.url)
query_params = urllib.parse.parse_qs(parsed_url.query)
if 'q' in query_params:
q_data = json.loads(query_params['q'][0])
main_keyword = q_data.get('mainKeyword', '')
sub_keyword = q_data.get('subKeyword', '')
else:
main_keyword = ''
sub_keyword = ''
page = int(query_params.get('page', [1])[0])
size = int(query_params.get('size', [10])[0])
else:
main_keyword = args.main_keyword
sub_keyword = args.sub_keyword
page = args.page
size = args.size
# 스크래핑 실행
result = scraper.scrape(main_keyword, sub_keyword, page, size)
# 결과 출력
if args.pretty:
json_output = json.dumps(result, ensure_ascii=False, indent=2)
else:
json_output = json.dumps(result, ensure_ascii=False)
if args.output:
with open(args.output, 'w', encoding='utf-8') as f:
f.write(json_output)
print(f"결과를 {args.output}에 저장했습니다.", file=sys.stderr)
else:
print(json_output)
except Exception as e:
print(f"오류 발생: {e}", file=sys.stderr)
import traceback
traceback.print_exc(file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()위의 코드를 allproductkorea_scraper.py 파일로 저장합니다.
기본 사용법
명령줄에서 다음과 같이 실행합니다:
python allproductkorea_scraper.py "검색어"고급 옵션
다양한 매개변수를 사용하여 검색을 정교화할 수 있습니다:
# 기본 검색
python allproductkorea_scraper.py "빙그레 더단백"
# 페이지 및 결과 수 지정
python allproductkorea_scraper.py "삼양라면" --page 2 --size 20
# 주요 키워드와 보조 키워드 사용
python allproductkorea_scraper.py "커피" --sub-keyword "원두" --page 1 --size 50
# 직접 URL 사용
python allproductkorea_scraper.py --url "https://www.allproductkorea.or.kr/products/search?q=..."
# 결과를 JSON 파일로 저장
python allproductkorea_scraper.py "라면" --output results.json --pretty출력 결과 해석
python allproductkorea_scraper.py --pretty "더단백"
검색 파라미터:
- 주요 키워드: 더단백
- 보조 키워드:
- 페이지: 1
- 사이즈: 10
요청 URL: https://www.allproductkorea.or.kr/products/search?q=%7B%22mainKeyword%22%3A+%22%EB%8D%94%EB%8B%A8%EB%B0%B1%22%2C+%22subKeyword%22%3A+%22%22%7D&page=1&size=10
HTML 가져오기 완료 (21,994 bytes)
스크래핑 결과: 10개 제품
이미지 테스트 [1] 자연주의 한번 쪄내어 더 부드러운 고: OK (2,336,020 bytes)
이미지 테스트 [2] 빙그레 건강tft 더단백 워터 프로틴: OK (2,558,638 bytes)
이미지 테스트 [3] 빙그레 건강 tft 더:단백 워터 프: OK (2,011,492 bytes)
URL 변환: 3/3개
접근 가능: 3/3개
{
"timestamp": "2025-08-11T11:14:59.911933",
"request": {
"url": "https://www.allproductkorea.or.kr/products/search?q=%7B%22mainKeyword%22%3A+%22%EB%8D%94%EB%8B%A8%EB%B0%B1%22%2C+%22subKeyword%22%3A+%22%22%7D&page=1&size=10",
"parameters": {
"mainKeyword": "더단백",
"subKeyword": "",
"page": 1,
"size": 10
}
},
"metadata": {
"title": "유통상품 표준DB",
"breadcrumb": ["상품 검색"],
"search_input_placeholder": "상품명 / 바코드/ 회사명을 입력해 주세요.",
"page_description": "상품정보 검색"
},
"pagination": {
"current_page": 1,
"total_pages": 4,
"page_size": 10,
"total_count": 34,
"result_count": 10,
"has_next": true,
"has_prev": false
},
"products": {
"count": 10,
"items": [
{
"prdNo": "102171669",
"barcode": "8809020767885",
"name": "자연주의 한번 쪄내어 더 부드러운 고단백 유기농 서리태볶음 400g",
"categoryHierarchy": ["가공식품", "농산가공식품", "기타농산가공식품", "기타농산가공식품"]
},
{
"prdNo": "102184523",
"barcode": "8801104949231",
"name": "빙그레 건강tft 더단백 워터 프로틴 청사과 400mL",
"categoryHierarchy": ["가공식품", "음료류", "기능성/건강음료", "기타기능성/건강음료"]
},
/* 이하 8개 제품 정보 생략 */
]
},
"image_tests": {
"tested_count": 3,
"conversion_count": 3,
"accessible_count": 3
/* 테스트 세부 결과 생략 */
},
"summary": {
"scraped_count": 10,
"total_available": 34,
"current_page": 1,
"total_pages": 4,
"has_more_pages": true
}
}
스크래퍼는 JSON 형식으로 결과를 출력합니다. 주요 구성 요소:
- timestamp: 검색 시점의 타임스탬프
- request: 요청 정보 (URL, 검색 매개변수)
- metadata: 검색 메타데이터
- pagination: 페이지네이션 정보 (총 페이지, 현재 페이지 등)
- products: 검색된 상품 목록
- count: 검색된 상품 수
- items: 상품 상세 정보 배열
- image_tests: 이미지 접근성 테스트 결과
- summary: 검색 결과 요약
주의사항
- 교육 및 연구 목적으로만 사용: 이 도구는 교육 및 연구 목적으로만 사용해야 합니다.
- 불법적 활용 금지: 상업적 목적이나 불법적인 활동에 이 도구를 사용하는 것은 엄격히 금지됩니다.