feat: yakkok.com 제품 이미지 크롤러 + 어드민 페이지

크롤러 (utils/yakkok_crawler.py):
- yakkok.com에서 제품 검색 및 이미지 추출
- MSSQL 오늘 판매 품목 자동 조회
- base64 변환 후 SQLite 저장
- CLI 지원 (--today, --product)

DB (product_images.db):
- 바코드, 제품명, 이미지(base64), 상태 저장
- 크롤링 로그 테이블

어드민 페이지 (/admin/product-images):
- 이미지 목록/검색/필터
- 통계 (성공/실패/대기)
- 상세 보기/삭제
- 오늘 판매 제품 일괄 크롤링

API:
- GET /api/admin/product-images
- GET /api/admin/product-images/<barcode>
- POST /api/admin/product-images/crawl-today
- DELETE /api/admin/product-images/<barcode>
This commit is contained in:
thug0bin
2026-03-02 23:19:52 +09:00
parent 4713395557
commit 29648e3a7d
6 changed files with 1648 additions and 0 deletions

View File

@@ -5649,6 +5649,169 @@ def api_search_mssql_drug():
return jsonify({'success': False, 'error': str(e)}), 500
# ============================================================
# 제품 이미지 관리 (yakkok 크롤러)
# ============================================================
@app.route('/admin/product-images')
def admin_product_images():
"""제품 이미지 관리 어드민 페이지"""
return render_template('admin_product_images.html')
@app.route('/api/admin/product-images')
def api_product_images_list():
"""제품 이미지 목록 조회"""
import sqlite3
try:
db_path = os.path.join(os.path.dirname(__file__), 'db', 'product_images.db')
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
status_filter = request.args.get('status', '')
search = request.args.get('search', '')
limit = int(request.args.get('limit', 50))
offset = int(request.args.get('offset', 0))
where_clauses = []
params = []
if status_filter:
where_clauses.append("status = ?")
params.append(status_filter)
if search:
where_clauses.append("(product_name LIKE ? OR barcode LIKE ?)")
params.extend([f'%{search}%', f'%{search}%'])
where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
# 총 개수
cursor.execute(f"SELECT COUNT(*) FROM product_images {where_sql}", params)
total = cursor.fetchone()[0]
# 목록 조회
cursor.execute(f"""
SELECT id, barcode, drug_code, product_name, thumbnail_base64,
image_url, status, created_at, error_message
FROM product_images
{where_sql}
ORDER BY created_at DESC
LIMIT ? OFFSET ?
""", params + [limit, offset])
items = [dict(row) for row in cursor.fetchall()]
conn.close()
return jsonify({
'success': True,
'total': total,
'items': items
})
except Exception as e:
logging.error(f"제품 이미지 목록 조회 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/admin/product-images/<barcode>')
def api_product_image_detail(barcode):
"""제품 이미지 상세 조회 (원본 base64 포함)"""
import sqlite3
try:
db_path = os.path.join(os.path.dirname(__file__), 'db', 'product_images.db')
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM product_images WHERE barcode = ?", (barcode,))
row = cursor.fetchone()
conn.close()
if row:
return jsonify({'success': True, 'image': dict(row)})
else:
return jsonify({'success': False, 'error': '이미지 없음'}), 404
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/admin/product-images/crawl-today', methods=['POST'])
def api_crawl_today():
"""오늘 판매 제품 크롤링"""
try:
from utils.yakkok_crawler import crawl_today_sales
result = crawl_today_sales(headless=True)
return jsonify({'success': True, 'result': result})
except Exception as e:
logging.error(f"크롤링 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/admin/product-images/crawl', methods=['POST'])
def api_crawl_products():
"""특정 제품 크롤링"""
try:
from utils.yakkok_crawler import crawl_products
data = request.get_json()
products = data.get('products', []) # [(barcode, drug_code, product_name), ...]
if not products:
return jsonify({'success': False, 'error': '제품 목록 필요'}), 400
result = crawl_products(products, headless=True)
return jsonify({'success': True, 'result': result})
except Exception as e:
logging.error(f"크롤링 오류: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/admin/product-images/<barcode>', methods=['DELETE'])
def api_delete_product_image(barcode):
"""제품 이미지 삭제"""
import sqlite3
try:
db_path = os.path.join(os.path.dirname(__file__), 'db', 'product_images.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("DELETE FROM product_images WHERE barcode = ?", (barcode,))
conn.commit()
conn.close()
return jsonify({'success': True})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/admin/product-images/stats')
def api_product_images_stats():
"""이미지 통계"""
import sqlite3
try:
db_path = os.path.join(os.path.dirname(__file__), 'db', 'product_images.db')
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT status, COUNT(*) as count
FROM product_images
GROUP BY status
""")
stats = {row[0]: row[1] for row in cursor.fetchall()}
cursor.execute("SELECT COUNT(*) FROM product_images")
total = cursor.fetchone()[0]
conn.close()
return jsonify({
'success': True,
'total': total,
'stats': stats
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
if __name__ == '__main__':
import os