chore: DB 분석 스크립트 추가 (APC/바코드 조사용)
This commit is contained in:
168
backend/scripts/query_aniparm.py
Normal file
168
backend/scripts/query_aniparm.py
Normal file
@@ -0,0 +1,168 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
애니팜 PostgreSQL 조회 스크립트
|
||||
Usage:
|
||||
python scripts/query_aniparm.py schema # 테이블 구조 확인
|
||||
python scripts/query_aniparm.py search <제품명> # 제품 검색
|
||||
python scripts/query_aniparm.py barcode <바코드> # 바코드로 검색
|
||||
python scripts/query_aniparm.py sample # 샘플 데이터
|
||||
python scripts/query_aniparm.py stats # 통계
|
||||
"""
|
||||
import sys
|
||||
import io
|
||||
import json
|
||||
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
# 인코딩 설정 (Windows CP949 문제 방지)
|
||||
# ═══════════════════════════════════════════════════════════
|
||||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
|
||||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
|
||||
from sqlalchemy import create_engine, text
|
||||
|
||||
# PostgreSQL 연결
|
||||
DATABASE_URI = 'postgresql://admin:trajet6640@192.168.0.87:5432/apdb_master'
|
||||
|
||||
def get_connection():
|
||||
engine = create_engine(DATABASE_URI)
|
||||
return engine.connect()
|
||||
|
||||
def cmd_schema():
|
||||
"""apc 테이블 구조 확인"""
|
||||
conn = get_connection()
|
||||
result = conn.execute(text("""
|
||||
SELECT column_name, data_type, character_maximum_length
|
||||
FROM information_schema.columns
|
||||
WHERE table_name = 'apc'
|
||||
ORDER BY ordinal_position
|
||||
"""))
|
||||
|
||||
print('=== apc 테이블 컬럼 ===')
|
||||
for row in result:
|
||||
length = f'({row.character_maximum_length})' if row.character_maximum_length else ''
|
||||
print(f' {row.column_name}: {row.data_type}{length}')
|
||||
|
||||
conn.close()
|
||||
|
||||
def cmd_search(keyword):
|
||||
"""제품명으로 검색"""
|
||||
conn = get_connection()
|
||||
result = conn.execute(text("""
|
||||
SELECT idx, apc, product_name, company_name,
|
||||
image_url1, godoimage_url_f, for_pets
|
||||
FROM apc
|
||||
WHERE product_name ILIKE :keyword
|
||||
LIMIT 20
|
||||
"""), {'keyword': f'%{keyword}%'})
|
||||
|
||||
print(f'=== "{keyword}" 검색 결과 ===')
|
||||
count = 0
|
||||
for row in result:
|
||||
count += 1
|
||||
print(f'\n[{count}] {row.product_name}')
|
||||
print(f' APC: {row.apc}')
|
||||
print(f' 제조사: {row.company_name}')
|
||||
print(f' 동물용: {row.for_pets}')
|
||||
if row.image_url1:
|
||||
print(f' 이미지1: {row.image_url1[:50]}...')
|
||||
if row.godoimage_url_f:
|
||||
print(f' 고도몰F: {row.godoimage_url_f[:50]}...')
|
||||
|
||||
if count == 0:
|
||||
print('(결과 없음)')
|
||||
|
||||
conn.close()
|
||||
|
||||
def cmd_barcode(barcode):
|
||||
"""바코드로 검색 - 바코드 컬럼이 있는지 먼저 확인"""
|
||||
conn = get_connection()
|
||||
|
||||
# 바코드 관련 컬럼 찾기
|
||||
result = conn.execute(text("""
|
||||
SELECT column_name FROM information_schema.columns
|
||||
WHERE table_name = 'apc'
|
||||
AND column_name ILIKE '%barcode%'
|
||||
"""))
|
||||
|
||||
barcode_cols = [row.column_name for row in result]
|
||||
|
||||
if not barcode_cols:
|
||||
print('apc 테이블에 barcode 관련 컬럼이 없습니다.')
|
||||
print('다른 컬럼으로 검색해야 합니다.')
|
||||
else:
|
||||
print(f'바코드 컬럼 발견: {barcode_cols}')
|
||||
# TODO: 바코드로 검색 구현
|
||||
|
||||
conn.close()
|
||||
|
||||
def cmd_sample():
|
||||
"""샘플 데이터 (동물용 제품)"""
|
||||
conn = get_connection()
|
||||
result = conn.execute(text("""
|
||||
SELECT idx, apc, product_name, company_name,
|
||||
image_url1, godoimage_url_f, for_pets
|
||||
FROM apc
|
||||
WHERE for_pets = true
|
||||
LIMIT 10
|
||||
"""))
|
||||
|
||||
print('=== 동물용 제품 샘플 ===')
|
||||
count = 0
|
||||
for row in result:
|
||||
count += 1
|
||||
print(f'\n[{count}] {row.product_name}')
|
||||
print(f' APC: {row.apc}')
|
||||
print(f' 제조사: {row.company_name}')
|
||||
img = row.image_url1 or row.godoimage_url_f or '(없음)'
|
||||
if len(img) > 50:
|
||||
img = img[:50] + '...'
|
||||
print(f' 이미지: {img}')
|
||||
|
||||
if count == 0:
|
||||
print('(동물용 제품 없음 - for_pets 필터 확인 필요)')
|
||||
|
||||
conn.close()
|
||||
|
||||
def cmd_stats():
|
||||
"""통계"""
|
||||
conn = get_connection()
|
||||
|
||||
result = conn.execute(text("""
|
||||
SELECT
|
||||
COUNT(*) as total,
|
||||
SUM(CASE WHEN for_pets = true THEN 1 ELSE 0 END) as pet_count,
|
||||
SUM(CASE WHEN image_url1 IS NOT NULL AND image_url1 != '' THEN 1 ELSE 0 END) as has_img1,
|
||||
SUM(CASE WHEN godoimage_url_f IS NOT NULL AND godoimage_url_f != '' THEN 1 ELSE 0 END) as has_godo_f
|
||||
FROM apc
|
||||
"""))
|
||||
|
||||
row = result.fetchone()
|
||||
print('=== apc 테이블 통계 ===')
|
||||
print(f'전체 제품: {row.total:,}개')
|
||||
print(f'동물용(for_pets=true): {row.pet_count:,}개')
|
||||
print(f'image_url1 있음: {row.has_img1:,}개')
|
||||
print(f'godoimage_url_f 있음: {row.has_godo_f:,}개')
|
||||
|
||||
conn.close()
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print(__doc__)
|
||||
return
|
||||
|
||||
cmd = sys.argv[1]
|
||||
|
||||
if cmd == 'schema':
|
||||
cmd_schema()
|
||||
elif cmd == 'search' and len(sys.argv) > 2:
|
||||
cmd_search(sys.argv[2])
|
||||
elif cmd == 'barcode' and len(sys.argv) > 2:
|
||||
cmd_barcode(sys.argv[2])
|
||||
elif cmd == 'sample':
|
||||
cmd_sample()
|
||||
elif cmd == 'stats':
|
||||
cmd_stats()
|
||||
else:
|
||||
print(__doc__)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user