[V10.337] 兼容 PChome 商品 API list 回應
All checks were successful
CD Pipeline / deploy (push) Successful in 1m6s

This commit is contained in:
OoO
2026-05-20 15:00:22 +08:00
parent 255140224b
commit 3f070a668b
3 changed files with 57 additions and 3 deletions

View File

@@ -320,7 +320,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
# ==========================================
# 系統版本與路徑
# ==========================================
SYSTEM_VERSION = "V10.336"
SYSTEM_VERSION = "V10.337"
LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
public_url = PUBLIC_URL # 用於模板顯示

View File

@@ -238,8 +238,28 @@ class PChomeCrawler:
data = response.json()
crawled_at = datetime.now()
if isinstance(data, dict):
product_entries = data.items()
product_count = len(data)
elif isinstance(data, list):
product_entries = [
((item or {}).get('Id') or f'index_{idx}', item)
for idx, item in enumerate(data)
if isinstance(item, dict)
]
product_count = len(product_entries)
failed_count += max(0, len(data) - product_count)
else:
logger.warning(
"PChome 商品 API 回傳格式異常 (批次 %s): %s",
i // batch_size + 1,
type(data).__name__,
)
failed_count += len(batch)
continue
# 解析商品資料
for prod_key, prod_data in data.items():
for prod_key, prod_data in product_entries:
try:
product = self._parse_product_data(prod_data, crawled_at)
if product:
@@ -248,7 +268,7 @@ class PChomeCrawler:
logger.warning(f"解析商品 {prod_key} 失敗: {e}")
failed_count += 1
logger.info(f"批次 {i // batch_size + 1}: 取得 {len(data)} 個商品資料")
logger.info(f"批次 {i // batch_size + 1}: 取得 {product_count} 個商品資料")
except requests.RequestException as e:
logger.error(f"API 請求失敗 (批次 {i // batch_size + 1}): {e}")

View File

@@ -92,6 +92,40 @@ def test_pchome_get_retries_transient_timeout():
assert len(calls) == 2
def test_pchome_fetch_product_details_accepts_list_payload():
from services.pchome_crawler import PChomeCrawler
crawler = PChomeCrawler(timeout=1, delay=0, max_retries=0)
calls = []
class FakeSession:
headers = {}
def get(self, url, params=None, timeout=None):
calls.append((url, params, timeout))
return _FakeResponse([
{
"Id": "DDABCD-12345678",
"Name": "測試商品 50ml",
"Price": {"P": 799, "M": 999},
"Pic": {"B": "/items/DDABCD12345678.jpg"},
"Qty": 8,
"Store": "24h",
"isOnSale": True,
}
])
crawler.session = FakeSession()
success, message, products = crawler.fetch_product_details(["DDABCD-12345678"])
assert success is True
assert message == "成功取得 1 個商品資料"
assert len(calls) == 1
assert [product.product_id for product in products] == ["DDABCD-12345678"]
assert products[0].price == 799
def test_feeder_search_cleanup_preserves_bracket_brand_and_specs():
from services.competitor_price_feeder import _clean_search_text