405 lines
16 KiB
Python
405 lines
16 KiB
Python
from sqlalchemy import create_engine, text
|
|
|
|
|
|
def _seed_growth_tables(engine):
|
|
with engine.begin() as conn:
|
|
conn.execute(text(
|
|
'CREATE TABLE daily_sales_snapshot ('
|
|
'"商品ID" TEXT, "商品名稱" TEXT, snapshot_date TEXT, "總業績" REAL, "數量" REAL, "商品館" TEXT)'
|
|
))
|
|
conn.execute(text(
|
|
"CREATE TABLE competitor_prices ("
|
|
"sku TEXT, source TEXT, competitor_product_id TEXT, competitor_product_name TEXT, "
|
|
"price REAL, match_score REAL, tags TEXT, crawled_at TEXT, expires_at TEXT)"
|
|
))
|
|
conn.execute(text(
|
|
"CREATE TABLE products ("
|
|
"id INTEGER PRIMARY KEY, i_code TEXT, name TEXT, status TEXT)"
|
|
))
|
|
conn.execute(text(
|
|
"CREATE TABLE price_records ("
|
|
"id INTEGER PRIMARY KEY, product_id INTEGER, price REAL, timestamp TEXT)"
|
|
))
|
|
conn.execute(text("""
|
|
INSERT INTO daily_sales_snapshot
|
|
("商品ID", "商品名稱", snapshot_date, "總業績", "數量", "商品館")
|
|
VALUES
|
|
('PCH-1', '高業績商品', '2026-06-14', 120000, 36, '保養'),
|
|
('PCH-1', '高業績商品', '2026-06-08', 180000, 50, '保養'),
|
|
('PCH-2', '待補對應商品', '2026-06-14', 90000, 18, '彩妝'),
|
|
('PCH-2', '待補對應商品', '2026-06-08', 30000, 8, '彩妝')
|
|
"""))
|
|
conn.execute(text(
|
|
"INSERT INTO products (id, i_code, name, status) "
|
|
"VALUES (1, 'MOMO-1', 'MOMO 參考商品', 'ACTIVE')"
|
|
))
|
|
conn.execute(text(
|
|
"INSERT INTO price_records (product_id, price, timestamp) "
|
|
"VALUES (1, 900, '2026-06-14 10:00:00')"
|
|
))
|
|
conn.execute(text("""
|
|
INSERT INTO competitor_prices
|
|
(sku, source, competitor_product_id, competitor_product_name,
|
|
price, match_score, tags, crawled_at, expires_at)
|
|
VALUES
|
|
('MOMO-1', 'pchome', 'PCH-1', '高業績商品',
|
|
1000, 0.91, '["identity_v2","match_type_exact"]', '2026-06-14 11:00:00', NULL)
|
|
"""))
|
|
|
|
|
|
def _seed_growth_external_offers(engine):
|
|
with engine.begin() as conn:
|
|
conn.execute(text("""
|
|
CREATE TABLE external_offers (
|
|
id INTEGER PRIMARY KEY,
|
|
source_code TEXT,
|
|
platform_code TEXT,
|
|
source_product_id TEXT,
|
|
source_offer_key TEXT,
|
|
title TEXT,
|
|
price REAL,
|
|
observed_at TEXT,
|
|
expires_at TEXT,
|
|
ingestion_method TEXT,
|
|
pchome_product_id TEXT,
|
|
momo_sku TEXT,
|
|
match_status TEXT,
|
|
quality_score REAL,
|
|
data_quality_status TEXT,
|
|
quality_notes_json TEXT,
|
|
raw_payload_json TEXT
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
INSERT INTO external_offers (
|
|
id, source_code, platform_code, source_product_id, source_offer_key,
|
|
title, price, observed_at, expires_at, ingestion_method,
|
|
pchome_product_id, momo_sku, match_status, quality_score,
|
|
data_quality_status, quality_notes_json, raw_payload_json
|
|
)
|
|
VALUES (
|
|
1, 'momo_reference', 'momo', 'MOMO-NEW', 'momo_reference:MOMO-NEW:PCH-1',
|
|
'MOMO 新資料層商品', 870, '2026-06-14 12:00:00', NULL, 'legacy_competitor_cache',
|
|
'PCH-1', 'MOMO-NEW', 'verified', 92,
|
|
'verified', '["自動同步"]',
|
|
'{"pchome_public_price": 1000, "pchome_public_name": "PChome 公開商品"}'
|
|
)
|
|
"""))
|
|
|
|
|
|
def _seed_growth_unit_price_external_offer(engine):
|
|
with engine.begin() as conn:
|
|
conn.execute(text("""
|
|
CREATE TABLE external_offers (
|
|
id INTEGER PRIMARY KEY,
|
|
source_code TEXT,
|
|
platform_code TEXT,
|
|
source_product_id TEXT,
|
|
source_offer_key TEXT,
|
|
title TEXT,
|
|
price REAL,
|
|
observed_at TEXT,
|
|
expires_at TEXT,
|
|
ingestion_method TEXT,
|
|
pchome_product_id TEXT,
|
|
momo_sku TEXT,
|
|
match_status TEXT,
|
|
quality_score REAL,
|
|
data_quality_status TEXT,
|
|
quality_notes_json TEXT,
|
|
raw_payload_json TEXT
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
INSERT INTO external_offers (
|
|
id, source_code, platform_code, source_product_id, source_offer_key,
|
|
title, price, observed_at, expires_at, ingestion_method,
|
|
pchome_product_id, momo_sku, match_status, quality_score,
|
|
data_quality_status, quality_notes_json, raw_payload_json
|
|
)
|
|
VALUES (
|
|
1, 'momo_reference', 'momo', 'MOMO-UNIT', 'momo_reference:MOMO-UNIT:PCH-1:unit_price',
|
|
'MOMO 單位價商品', 468, '2026-06-14 12:00:00', NULL, 'targeted_momo_search',
|
|
'PCH-1', 'MOMO-UNIT', 'verified', 82,
|
|
'verified', '["自動單位價比較"]',
|
|
'{"price_basis": "unit_price", "pchome_public_price": 920, "pchome_public_name": "PChome 公開商品", "tags": ["identity_v2", "price_basis_unit_price"], "unit_price_comparison": {"unit_label": "ml", "momo_unit_price": 11.7, "competitor_unit_price": 23.0, "momo_total_quantity": 40, "competitor_total_quantity": 40, "unit_gap_pct": -49.13}}'
|
|
)
|
|
"""))
|
|
|
|
|
|
def test_pchome_growth_opportunities_use_plain_language_and_pause_shopee_coupang():
|
|
from services.pchome_revenue_growth_service import build_pchome_growth_opportunities
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
_seed_growth_tables(engine)
|
|
|
|
payload = build_pchome_growth_opportunities(engine, limit=5)
|
|
|
|
assert payload["success"] is True
|
|
assert payload["system_name"] == "PChome 業績成長自動化作戰系統"
|
|
assert payload["source_scope"]["active_external_sources"] == ["MOMO 外部價格參考"]
|
|
assert payload["source_scope"]["paused_external_sources"] == ["蝦皮", "酷澎"]
|
|
readiness = payload["source_scope"]["source_readiness"]
|
|
sources = {source["code"]: source for source in readiness["sources"]}
|
|
assert sources["momo_reference"]["status_label"] == "正在使用"
|
|
assert sources["shopee"]["status_label"] == "先暫停"
|
|
assert sources["coupang"]["status_label"] == "先暫停"
|
|
assert payload["stats"]["candidate_count"] == 2
|
|
assert payload["stats"]["mapped_count"] == 1
|
|
assert payload["stats"]["needs_mapping_count"] == 1
|
|
|
|
actions = {item["pchome_product_id"]: item["recommended_action"]["label"] for item in payload["opportunities"]}
|
|
assert actions["PCH-1"] == "檢查售價與活動"
|
|
assert actions["PCH-2"] == "先補商品對應"
|
|
pchome_1 = next(item for item in payload["opportunities"] if item["pchome_product_id"] == "PCH-1")
|
|
assert pchome_1["pchome_price"] == 3488.37
|
|
assert pchome_1["external_price"]["data_source"] == "competitor_prices"
|
|
assert payload["stats"]["external_data_source_counts"] == {"舊比價快取": 1}
|
|
assert all("identity" not in " ".join(item["reason_lines"]).lower() for item in payload["opportunities"])
|
|
|
|
|
|
def test_pchome_growth_prefers_external_offers_over_legacy_competitor_cache():
|
|
from services.pchome_revenue_growth_service import build_pchome_growth_opportunities
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
_seed_growth_tables(engine)
|
|
_seed_growth_external_offers(engine)
|
|
|
|
payload = build_pchome_growth_opportunities(engine, limit=5)
|
|
|
|
assert payload["success"] is True
|
|
pchome_1 = next(item for item in payload["opportunities"] if item["pchome_product_id"] == "PCH-1")
|
|
external_price = pchome_1["external_price"]
|
|
assert external_price["data_source"] == "external_offers"
|
|
assert external_price["data_source_label"] == "自動同步資料層"
|
|
assert external_price["momo_sku"] == "MOMO-NEW"
|
|
assert external_price["momo_price"] == 870
|
|
assert external_price["pchome_price"] == 1000
|
|
assert external_price["gap_pct"] == -13.0
|
|
assert payload["stats"]["external_data_source_counts"] == {"自動同步資料層": 1}
|
|
|
|
|
|
def test_pchome_growth_understands_unit_price_external_offers():
|
|
from services.pchome_revenue_growth_service import build_pchome_growth_opportunities
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
_seed_growth_tables(engine)
|
|
_seed_growth_unit_price_external_offer(engine)
|
|
|
|
payload = build_pchome_growth_opportunities(engine, limit=5)
|
|
|
|
assert payload["success"] is True
|
|
pchome_1 = next(item for item in payload["opportunities"] if item["pchome_product_id"] == "PCH-1")
|
|
external_price = pchome_1["external_price"]
|
|
assert external_price["data_source"] == "external_offers"
|
|
assert external_price["price_basis"] == "unit_price"
|
|
assert external_price["price_basis_label"] == "單位價"
|
|
assert external_price["momo_price"] == 468
|
|
assert external_price["pchome_price"] == 920
|
|
assert external_price["momo_unit_price"] == 11.7
|
|
assert external_price["pchome_unit_price"] == 23.0
|
|
assert external_price["gap_pct"] == -49.1
|
|
assert pchome_1["recommended_action"]["label"] == "檢查售價與活動"
|
|
assert pchome_1["data_quality"]["label"] == "資料可用單位價判斷"
|
|
assert any("單位價" in line for line in pchome_1["reason_lines"])
|
|
|
|
|
|
def test_pchome_growth_route_cache_respects_shared_invalidation_epoch(monkeypatch):
|
|
from routes import ai_routes as routes
|
|
|
|
epoch = {"value": 1.0}
|
|
monkeypatch.setattr(routes, "get_pchome_growth_cache_epoch", lambda: epoch["value"])
|
|
routes._PCHOME_GROWTH_CACHE.update({
|
|
"expires_at": 0.0,
|
|
"epoch": 0.0,
|
|
"payload": None,
|
|
})
|
|
|
|
routes._set_pchome_growth_cache({"success": True, "stats": {"candidate_count": 1}})
|
|
|
|
cached = routes._get_cached_pchome_growth_payload()
|
|
assert cached["success"] is True
|
|
assert cached["stats"]["candidate_count"] == 1
|
|
|
|
epoch["value"] = 2.0
|
|
assert routes._get_cached_pchome_growth_payload() is None
|
|
|
|
|
|
def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items():
|
|
from services.pchome_growth_momo_backfill_service import run_pchome_growth_momo_backfill
|
|
|
|
captured = {}
|
|
|
|
class FakeEngine:
|
|
pass
|
|
|
|
before_payload = {
|
|
"success": True,
|
|
"stats": {"mapping_rate": 0, "candidate_count": 3, "mapped_count": 0},
|
|
"opportunities": [
|
|
{
|
|
"pchome_product_id": "PCH-NEEDS-1",
|
|
"product_name": "需要補對應商品一",
|
|
"pchome_price": 920,
|
|
"sales_7d": 120000,
|
|
"priority_score": 91,
|
|
"external_price": None,
|
|
"recommended_action": {"code": "map_external_product"},
|
|
},
|
|
{
|
|
"pchome_product_id": "PCH-MAPPED",
|
|
"product_name": "已有比價商品",
|
|
"pchome_price": 880,
|
|
"external_price": {"momo_sku": "MOMO-OLD"},
|
|
"recommended_action": {"code": "review_price_or_promo"},
|
|
},
|
|
{
|
|
"pchome_product_id": "PCH-NEEDS-2",
|
|
"product_name": "需要補對應商品二",
|
|
"pchome_price": 760,
|
|
"sales_7d": 90000,
|
|
"priority_score": 82,
|
|
"external_price": None,
|
|
"recommended_action": {"code": "map_external_product"},
|
|
},
|
|
],
|
|
}
|
|
after_payload = {
|
|
"success": True,
|
|
"stats": {"mapping_rate": 66.7, "candidate_count": 3, "mapped_count": 2},
|
|
"opportunities": [],
|
|
}
|
|
|
|
payload_calls = []
|
|
|
|
def fake_build_payload(engine, limit):
|
|
payload_calls.append(limit)
|
|
return before_payload if len(payload_calls) == 1 else after_payload
|
|
|
|
def fake_search(targets, limit):
|
|
captured["targets"] = targets
|
|
captured["search_limit"] = limit
|
|
return True, "找到候選", [
|
|
{
|
|
"product_id": "MOMO-AUTO",
|
|
"auto_compare_type": "total_price",
|
|
"can_auto_compare": True,
|
|
},
|
|
{
|
|
"product_id": "MOMO-UNIT",
|
|
"auto_compare_type": "unit_price",
|
|
"can_auto_compare": True,
|
|
},
|
|
{
|
|
"product_id": "MOMO-REVIEW",
|
|
"auto_compare_type": "manual_review",
|
|
"can_auto_compare": False,
|
|
},
|
|
]
|
|
|
|
def fake_sync(engine, candidates):
|
|
captured["sync_candidates"] = candidates
|
|
return {
|
|
"success": True,
|
|
"status": "synced",
|
|
"written_count": len(candidates),
|
|
"total_price_count": 1,
|
|
"unit_price_count": 1,
|
|
}
|
|
|
|
payload = run_pchome_growth_momo_backfill(
|
|
FakeEngine(),
|
|
limit=2,
|
|
build_payload_func=fake_build_payload,
|
|
search_func=fake_search,
|
|
sync_func=fake_sync,
|
|
)
|
|
|
|
assert payload["success"] is True
|
|
assert payload["data"]["scanned_products"] == 2
|
|
assert payload["data"]["candidate_count"] == 3
|
|
assert payload["data"]["auto_compare_count"] == 2
|
|
assert payload["data"]["review_count"] == 1
|
|
assert payload["data"]["external_offer_sync"]["written_count"] == 2
|
|
assert payload["data"]["after_stats"]["mapping_rate"] == 66.7
|
|
assert [item["product_id"] for item in captured["targets"]] == ["PCH-NEEDS-1", "PCH-NEEDS-2"]
|
|
assert [item["price"] for item in captured["targets"]] == [920, 760]
|
|
assert [item["product_id"] for item in captured["sync_candidates"]] == ["MOMO-AUTO", "MOMO-UNIT"]
|
|
assert captured["search_limit"] == 2
|
|
|
|
|
|
def test_pchome_growth_momo_backfill_route_calls_shared_service(monkeypatch):
|
|
from flask import Flask
|
|
from routes import ai_routes as routes
|
|
|
|
captured = {}
|
|
|
|
class FakeEngine:
|
|
def dispose(self):
|
|
captured["disposed"] = True
|
|
|
|
def fake_run(engine, limit):
|
|
captured["limit"] = limit
|
|
return {
|
|
"success": True,
|
|
"message": "已完成",
|
|
"data": {
|
|
"scanned_products": 2,
|
|
"candidate_count": 3,
|
|
"auto_compare_count": 2,
|
|
"review_count": 1,
|
|
"external_offer_sync": {"written_count": 2},
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", lambda database_path: FakeEngine())
|
|
monkeypatch.setattr(routes, "_run_pchome_growth_momo_backfill", fake_run)
|
|
|
|
app = Flask(__name__)
|
|
with app.test_request_context(
|
|
"/api/ai/pchome-growth/backfill-momo-candidates",
|
|
method="POST",
|
|
json={"limit": 2},
|
|
):
|
|
response = routes.api_pchome_growth_backfill_momo_candidates.__wrapped__()
|
|
|
|
payload = response.get_json()
|
|
assert payload["success"] is True
|
|
assert payload["data"]["external_offer_sync"]["written_count"] == 2
|
|
assert captured["limit"] == 2
|
|
assert captured["disposed"] is True
|
|
|
|
|
|
def test_ai_product_pick_sales_join_by_sku_disabled_by_default(monkeypatch):
|
|
from services.ai_product_pick_agent import _sales_join_by_momo_sku_enabled
|
|
|
|
monkeypatch.delenv("PCHOME_SALES_JOIN_BY_MOMO_SKU_ENABLED", raising=False)
|
|
|
|
assert _sales_join_by_momo_sku_enabled() is False
|
|
|
|
|
|
def test_ai_intelligence_template_uses_pchome_growth_name_and_endpoint():
|
|
from pathlib import Path
|
|
|
|
template = Path("templates/ai_intelligence.html").read_text(encoding="utf-8")
|
|
|
|
assert "PChome 業績成長自動化作戰系統" in template
|
|
assert "/api/ai/pchome-growth/opportunities" in template
|
|
assert "/api/ai/pchome-growth/external-offers/csv-dry-run" in template
|
|
assert "growthSourceReadiness" in template
|
|
assert "今日重點總覽" in template
|
|
assert "nextActionTitle" in template
|
|
assert "商品處理進度" in template
|
|
assert "價格風險分佈" in template
|
|
assert "growthActionHint" in template
|
|
assert "growthDataSourceSummary" in template
|
|
assert "external_data_source_counts" in template
|
|
assert "compSourceSummary" in template
|
|
assert "scrollToPanel('externalPricePanel')" in template
|
|
assert "備援資料檢查" in template
|
|
assert "外部報價預檢" not in template
|
|
assert "growth-ops-table" in template
|
|
assert "鎖定商品" in template
|
|
assert "無法比價" in template
|
|
assert "補齊比價資料" in template
|