501 lines
17 KiB
Python
501 lines
17 KiB
Python
"""
|
|
PChome vs MOMO 比價 API 路由
|
|
"""
|
|
|
|
from flask import Blueprint, request, jsonify, render_template
|
|
import logging
|
|
|
|
from auth import login_required
|
|
from utils.momo_url_utils import extract_momo_i_code, is_probable_momo_icode
|
|
from services.price_comparison import (
|
|
compare_brand_prices,
|
|
BRAND_ALIASES,
|
|
BRAND_NORMALIZE_MAP
|
|
)
|
|
from services.momo_crawler import search_momo_products, search_momo_products_for_pchome_products
|
|
from services.pchome_crawler import search_pchome_products
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
price_comparison_bp = Blueprint('price_comparison', __name__)
|
|
|
|
|
|
def _candidate_auto_compare_type(item: dict) -> str:
|
|
"""Normalize old and new targeted MOMO candidate flags."""
|
|
explicit_type = item.get("auto_compare_type")
|
|
if explicit_type:
|
|
return explicit_type
|
|
if item.get("can_auto_compare"):
|
|
return "total_price"
|
|
return "manual_review"
|
|
|
|
|
|
def _build_pchome_product_url(product_id: str) -> str:
|
|
product_id = str(product_id or "").strip()
|
|
return f"https://24h.pchome.com.tw/prod/{product_id}" if product_id else ""
|
|
|
|
|
|
def _humanize_targeted_review_reasons(candidate: dict) -> list[str]:
|
|
labels = []
|
|
score = candidate.get("target_match_score")
|
|
try:
|
|
score_pct = round(float(score or 0) * 100)
|
|
except (TypeError, ValueError):
|
|
score_pct = 0
|
|
if score_pct > 0:
|
|
labels.append(f"可信度 {score_pct}%")
|
|
|
|
reason_map = {
|
|
"variant_selection_review": "需確認色號",
|
|
"strong_exact_spec_match": "規格接近",
|
|
"strong_product_line_match": "系列接近",
|
|
"count_conflict": "組合數量需確認",
|
|
"unit_comparable": "可用單位價判斷",
|
|
"makeup_catalog_selection_gap": "款式需確認",
|
|
}
|
|
for reason in candidate.get("target_match_reasons") or []:
|
|
label = reason_map.get(str(reason or "").strip())
|
|
if label and label not in labels:
|
|
labels.append(label)
|
|
return labels or ["需人工確認同款"]
|
|
|
|
|
|
def _present_momo_review_candidate(candidate: dict) -> dict:
|
|
"""Return only user-facing review candidate fields for the UI."""
|
|
pchome_product_id = str(candidate.get("target_pchome_product_id") or "").strip()
|
|
return {
|
|
"product_id": candidate.get("product_id") or candidate.get("goodsCode") or candidate.get("id"),
|
|
"name": candidate.get("name") or candidate.get("title") or "",
|
|
"price": candidate.get("price"),
|
|
"product_url": candidate.get("product_url") or candidate.get("url") or "",
|
|
"image_url": candidate.get("image_url") or "",
|
|
"target_pchome_product_id": pchome_product_id,
|
|
"target_pchome_name": candidate.get("target_pchome_name") or pchome_product_id,
|
|
"target_pchome_price": candidate.get("target_pchome_price"),
|
|
"target_pchome_url": candidate.get("target_pchome_url") or _build_pchome_product_url(pchome_product_id),
|
|
"target_gap_pct": candidate.get("target_gap_pct"),
|
|
"target_match_score": candidate.get("target_match_score"),
|
|
"target_match_reason_labels": _humanize_targeted_review_reasons(candidate),
|
|
}
|
|
|
|
|
|
def _sync_targeted_candidates_to_external_offers(candidates: list[dict]) -> dict:
|
|
from database.manager import DatabaseManager
|
|
from services.external_market_offer_service import sync_targeted_momo_candidates_to_external_offers
|
|
|
|
db = DatabaseManager()
|
|
return sync_targeted_momo_candidates_to_external_offers(
|
|
db.engine,
|
|
candidates,
|
|
dry_run=False,
|
|
)
|
|
|
|
|
|
# ============================================
|
|
# 頁面路由
|
|
# ============================================
|
|
|
|
@price_comparison_bp.route('/price_comparison')
|
|
@login_required
|
|
def price_comparison_page():
|
|
"""比價管理頁面"""
|
|
return render_template('price_comparison.html', active_page='price_comparison')
|
|
|
|
|
|
# ============================================
|
|
# API 路由
|
|
# ============================================
|
|
|
|
@price_comparison_bp.route('/api/price_comparison/brands', methods=['GET'])
|
|
@login_required
|
|
def get_supported_brands():
|
|
"""取得支援的品牌列表"""
|
|
brands = []
|
|
for canonical, aliases in BRAND_ALIASES.items():
|
|
# 找中文名稱
|
|
chinese_name = next((a for a in aliases if any('\u4e00' <= c <= '\u9fff' for c in a)), canonical)
|
|
brands.append({
|
|
'code': canonical,
|
|
'name': chinese_name,
|
|
'aliases': aliases
|
|
})
|
|
|
|
# 按中文名稱排序
|
|
brands.sort(key=lambda x: x['name'])
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': {
|
|
'brands': brands,
|
|
'count': len(brands)
|
|
}
|
|
})
|
|
|
|
|
|
@price_comparison_bp.route('/api/price_comparison/compare', methods=['POST'])
|
|
@login_required
|
|
def compare_prices():
|
|
"""
|
|
執行比價
|
|
|
|
Request Body:
|
|
{
|
|
"brand": "理膚寶水",
|
|
"pchome_products": [...], // 可選，如果沒有會自動爬取
|
|
"momo_products": [...] // 可選，如果沒有會自動爬取
|
|
}
|
|
"""
|
|
try:
|
|
data = request.get_json()
|
|
brand = data.get('brand', '').strip()
|
|
|
|
if not brand:
|
|
return jsonify({'success': False, 'message': '請提供品牌名稱'}), 400
|
|
|
|
# 取得 PChome 商品
|
|
pchome_products = data.get('pchome_products')
|
|
if not pchome_products:
|
|
# 自動爬取
|
|
logger.info(f"自動搜尋 PChome: {brand}")
|
|
success, msg, pchome_products = search_pchome_products(brand, limit=100)
|
|
if not success:
|
|
logger.warning(f"PChome 搜尋失敗: {msg}")
|
|
pchome_products = []
|
|
|
|
targeted_momo_summary = None
|
|
|
|
# 取得 MOMO 商品
|
|
momo_products = data.get('momo_products')
|
|
if not momo_products:
|
|
if pchome_products:
|
|
logger.info("[PriceComparison] 以 PChome 商品精準搜尋 MOMO 候選: brand=%s pchome_count=%s", brand, len(pchome_products))
|
|
success, msg, targeted_momo_products = search_momo_products_for_pchome_products(
|
|
pchome_products,
|
|
max_products=30,
|
|
limit_per_product=8,
|
|
)
|
|
exact_products = [
|
|
item for item in (targeted_momo_products or [])
|
|
if _candidate_auto_compare_type(item) == "total_price"
|
|
]
|
|
unit_compare_candidates = [
|
|
item for item in (targeted_momo_products or [])
|
|
if _candidate_auto_compare_type(item) == "unit_price"
|
|
]
|
|
review_candidates = [
|
|
item for item in (targeted_momo_products or [])
|
|
if _candidate_auto_compare_type(item) not in {"total_price", "unit_price"}
|
|
]
|
|
targeted_momo_summary = {
|
|
"message": msg,
|
|
"candidate_count": len(targeted_momo_products or []),
|
|
"auto_compare_count": len(exact_products) + len(unit_compare_candidates),
|
|
"exact_compare_count": len(exact_products),
|
|
"unit_compare_count": len(unit_compare_candidates),
|
|
"review_count": len(review_candidates),
|
|
}
|
|
momo_products = exact_products
|
|
else:
|
|
logger.info(f"自動搜尋 MOMO: {brand}")
|
|
success, msg, momo_products = search_momo_products(brand, limit=100)
|
|
if not success:
|
|
logger.warning(f"MOMO 搜尋失敗: {msg}")
|
|
momo_products = []
|
|
|
|
# 執行比價
|
|
result = compare_brand_prices(brand, pchome_products, momo_products)
|
|
if targeted_momo_summary:
|
|
result["momo_targeted_search"] = targeted_momo_summary
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': result
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"比價失敗: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'message': f'比價失敗: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
@price_comparison_bp.route('/api/price_comparison/fetch_pchome', methods=['POST'])
|
|
@login_required
|
|
def fetch_pchome_products():
|
|
"""
|
|
爬取 PChome 商品
|
|
|
|
Request Body:
|
|
{
|
|
"keyword": "理膚寶水",
|
|
"limit": 50
|
|
}
|
|
"""
|
|
try:
|
|
data = request.get_json()
|
|
keyword = data.get('keyword', '').strip()
|
|
limit = data.get('limit', 50)
|
|
|
|
if not keyword:
|
|
return jsonify({'success': False, 'message': '請提供搜尋關鍵字'}), 400
|
|
|
|
success, msg, products = search_pchome_products(keyword, limit=limit)
|
|
|
|
return jsonify({
|
|
'success': success,
|
|
'message': msg,
|
|
'data': {
|
|
'products': products,
|
|
'count': len(products)
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"爬取 PChome 失敗: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'message': f'爬取失敗: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
@price_comparison_bp.route('/api/price_comparison/fetch_momo_for_pchome', methods=['POST'])
|
|
@login_required
|
|
def fetch_momo_for_pchome_products():
|
|
"""用 PChome 商品清單反查 MOMO 候選；可選擇同步安全候選到外部報價層。"""
|
|
try:
|
|
data = request.get_json() or {}
|
|
pchome_products = data.get('pchome_products') or []
|
|
should_sync_external_offers = str(
|
|
data.get('sync_external_offers') or ''
|
|
).strip().lower() in {'1', 'true', 'yes'}
|
|
if not pchome_products:
|
|
return jsonify({
|
|
'success': False,
|
|
'message': '請先取得 PChome 商品，再搜尋 MOMO 候選'
|
|
}), 400
|
|
|
|
success, message, products = search_momo_products_for_pchome_products(
|
|
pchome_products,
|
|
max_products=30,
|
|
limit_per_product=8,
|
|
)
|
|
exact_products = [
|
|
item for item in products
|
|
if _candidate_auto_compare_type(item) == "total_price"
|
|
]
|
|
unit_compare_candidates = [
|
|
item for item in products
|
|
if _candidate_auto_compare_type(item) == "unit_price"
|
|
]
|
|
raw_review_candidates = [
|
|
item for item in products
|
|
if _candidate_auto_compare_type(item) not in {"total_price", "unit_price"}
|
|
]
|
|
review_candidates = [
|
|
_present_momo_review_candidate(item)
|
|
for item in raw_review_candidates
|
|
]
|
|
external_offer_sync = {
|
|
"success": True,
|
|
"status": "not_requested",
|
|
"written_count": 0,
|
|
"message": "未要求同步外部價格參考。",
|
|
}
|
|
if should_sync_external_offers and (exact_products or unit_compare_candidates):
|
|
try:
|
|
external_offer_sync = _sync_targeted_candidates_to_external_offers(
|
|
[*exact_products, *unit_compare_candidates],
|
|
)
|
|
except Exception as sync_exc:
|
|
logger.warning("[PriceComparison] 外部價格參考同步失敗: %s", sync_exc, exc_info=True)
|
|
external_offer_sync = {
|
|
"success": False,
|
|
"status": "failed",
|
|
"written_count": 0,
|
|
"message": "MOMO 候選已找到，但暫時無法同步到外部價格參考。",
|
|
}
|
|
|
|
return jsonify({
|
|
'success': success,
|
|
'message': message,
|
|
'data': {
|
|
'products': exact_products,
|
|
'unit_compare_candidates': unit_compare_candidates,
|
|
'review_candidates': review_candidates,
|
|
'count': len(exact_products),
|
|
'exact_compare_count': len(exact_products),
|
|
'unit_compare_count': len(unit_compare_candidates),
|
|
'auto_compare_count': len(exact_products) + len(unit_compare_candidates),
|
|
'review_count': len(review_candidates),
|
|
'candidate_count': len(products),
|
|
'external_offer_sync': external_offer_sync,
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"搜尋 MOMO 候選失敗: {e}", exc_info=True)
|
|
return jsonify({
|
|
'success': False,
|
|
'message': f'搜尋 MOMO 候選失敗: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
@price_comparison_bp.route('/api/price_comparison/parse_momo_excel', methods=['POST'])
|
|
@login_required
|
|
def parse_momo_excel():
|
|
"""
|
|
解析 MOMO 商品 Excel
|
|
|
|
上傳 MOMO 匯出的商品清單 Excel
|
|
"""
|
|
try:
|
|
if 'file' not in request.files:
|
|
return jsonify({'success': False, 'message': '請上傳檔案'}), 400
|
|
|
|
file = request.files['file']
|
|
if not file.filename:
|
|
return jsonify({'success': False, 'message': '請選擇檔案'}), 400
|
|
|
|
# 讀取 Excel
|
|
import pandas as pd
|
|
df = pd.read_excel(file)
|
|
|
|
# 嘗試識別欄位
|
|
# 常見欄位名: 商品名稱, 售價, 商品編號, 連結 等
|
|
name_col = None
|
|
price_col = None
|
|
id_col = None
|
|
url_col = None
|
|
|
|
for col in df.columns:
|
|
col_lower = str(col).lower()
|
|
if any(k in col_lower for k in ['商品名', '名稱', 'name', 'product']):
|
|
name_col = col
|
|
elif any(k in col_lower for k in ['售價', '價格', 'price', '金額']):
|
|
price_col = col
|
|
elif any(k in col_lower for k in ['編號', 'id', 'code', 'sku']):
|
|
id_col = col
|
|
elif any(k in col_lower for k in ['連結', 'url', 'link']):
|
|
url_col = col
|
|
|
|
if not name_col or not price_col:
|
|
return jsonify({
|
|
'success': False,
|
|
'message': f'無法識別必要欄位 (商品名稱/售價)。找到的欄位: {list(df.columns)}'
|
|
}), 400
|
|
|
|
# 轉換為商品列表
|
|
products = []
|
|
for idx, row in df.iterrows():
|
|
try:
|
|
raw_id = ''
|
|
if id_col:
|
|
raw_id = str(row[id_col]).strip()
|
|
if raw_id in {'nan', 'None', 'NoneType', 'nan.'}:
|
|
raw_id = ''
|
|
raw_url = ''
|
|
if url_col:
|
|
raw_url = str(row[url_col]).strip()
|
|
if raw_url in {'nan', 'None', 'NoneType'}:
|
|
raw_url = ''
|
|
|
|
product_id = raw_id if is_probable_momo_icode(raw_id) else extract_momo_i_code(raw_url) or ''
|
|
if not is_probable_momo_icode(product_id):
|
|
product_id = ''
|
|
|
|
product = {
|
|
'name': str(row[name_col]),
|
|
'price': int(float(row[price_col])),
|
|
'product_id': product_id,
|
|
'url': raw_url
|
|
}
|
|
products.append(product)
|
|
except (ValueError, TypeError):
|
|
continue
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'成功解析 {len(products)} 筆商品',
|
|
'data': {
|
|
'products': products,
|
|
'count': len(products),
|
|
'columns_detected': {
|
|
'name': name_col,
|
|
'price': price_col,
|
|
'id': id_col,
|
|
'url': url_col
|
|
}
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"解析 Excel 失敗: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'message': f'解析失敗: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
@price_comparison_bp.route('/api/price_comparison/quick_compare', methods=['POST'])
|
|
@login_required
|
|
def quick_compare():
|
|
"""
|
|
快速比價 (手動輸入)
|
|
|
|
Request Body:
|
|
{
|
|
"pchome_products": [
|
|
{"name": "商品A", "price": 100},
|
|
...
|
|
],
|
|
"momo_products": [
|
|
{"name": "商品A", "price": 95},
|
|
...
|
|
]
|
|
}
|
|
"""
|
|
try:
|
|
data = request.get_json()
|
|
pchome_products = data.get('pchome_products', [])
|
|
momo_products = data.get('momo_products', [])
|
|
|
|
if not pchome_products or not momo_products:
|
|
return jsonify({
|
|
'success': False,
|
|
'message': '請提供兩平台的商品資料'
|
|
}), 400
|
|
|
|
# 補充缺失欄位
|
|
def _normalize_product_id(products_list, source_prefix, id_field, url_field):
|
|
for i, p in enumerate(products_list):
|
|
if id_field not in p:
|
|
p[id_field] = ''
|
|
product_id = str(p.get(id_field) or '').strip()
|
|
if not is_probable_momo_icode(product_id):
|
|
product_id = extract_momo_i_code(p.get(url_field)) or ''
|
|
if not is_probable_momo_icode(product_id):
|
|
p[id_field] = f'{source_prefix}_{i}'
|
|
else:
|
|
p[id_field] = product_id
|
|
if url_field not in p or p.get(url_field) is None:
|
|
p[url_field] = ''
|
|
|
|
_normalize_product_id(pchome_products, 'pchome', 'product_id', 'product_url')
|
|
_normalize_product_id(momo_products, 'momo', 'product_id', 'url')
|
|
|
|
# 執行比價 (不限定品牌)
|
|
result = compare_brand_prices('', pchome_products, momo_products)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': result
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"快速比價失敗: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'message': f'比價失敗: {str(e)}'
|
|
}), 500
|