后台店铺数据记录页新增 Tab 切换与重复 ASIN 分析;task-169/170 HTTP 客户端配置接入;dedupe/brand 相关改造
Build Backend JAR / build (push) Has been cancelled

- 管理后台:店铺数据记录改为列表分页展示(去任务号/文件列/累计标题,'最新'改'更新时间'),新增'重复 ASIN' Tab 按跨店聚合展示 ASIN/店铺/分组/国家/日期/价格;修复失败/进行中任务被过滤不显示的问题
- task-169: BrandCheckHttpConfigResolver 从 aiimage.http-client.* 读取品牌检查超时/重试(默认同现状、钳制),附 8 条测试
- task-170: surefire 内存调整为 1536m
- dedupe: DedupeRunService 重构(事务边界/进度)、新增 DedupeRunProgressVo、Controller 与结果 VO 调整,PermissionMenuService 相应适配
- brand/dedupe 前端:BrandDedupeTab/BrandPatrolDeleteTab 改造、新增 CountrySelector 组件与 country-options、类型与接口端对齐、测试更新
- 移除无引用文件:backend/static/logo.jpg、prompts/
This commit is contained in:
2026-09-03 01:57:41 +08:00
parent 152ea6eec0
commit 417a2bf831
20 changed files with 1883 additions and 475 deletions
@@ -0,0 +1,178 @@
"""重复 ASIN 分析接口单元测试:模拟数据库行与结果文件,验证跨店铺重复聚合逻辑。"""
import sys
import unittest
from datetime import datetime
from io import BytesIO
from pathlib import Path
from unittest.mock import patch
from openpyxl import Workbook
from flask import Flask
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from blueprints import admin_api
def _make_workbook(rows_by_sheet):
"""构造结果 Workbookrows_by_sheet = {sheet名: [(日期, ASIN, 价格, 品牌), ...]}"""
wb = Workbook()
wb.remove(wb.active)
for sheet_name, rows in rows_by_sheet.items():
ws = wb.create_sheet(sheet_name)
ws.append(['日期', 'ASIN', '商品图片', '库存销量', '销售排名',
'页面浏览量', '售出件数', '价格', '推荐报价', '品牌'])
for date, asin, price, brand in rows:
row = [date, asin, '', '', '', '', '', price, '', brand]
ws.append(row)
return wb
class ShopDataDuplicateAsinTest(unittest.TestCase):
def setUp(self):
self.app = Flask(__name__)
self.group_rows = [
{'shop_name': 'Shop A', 'latest_created_at': datetime(2026, 8, 31, 1, 15, 1)},
{'shop_name': 'Shop B', 'latest_created_at': datetime(2026, 8, 31, 1, 15, 2)},
{'shop_name': 'Shop C', 'latest_created_at': datetime(2026, 8, 31, 1, 15, 3)},
]
# 每家店一个结果文件:Shop A 与 Shop B 共享 ASIN1Shop C 单独 ASIN3
self.shop_files = {
'Shop A': _make_workbook({
'英国': [('2026-08-30', 'B0ABC111', 'GBP 9.99', 'BrandA'),
('2026-08-30', 'B0UNIQUE1', 'GBP 5.00', 'BrandA')],
'德国': [('2026-08-30', 'B0ABC111', 'EUR 10.99', 'BrandA')],
}),
'Shop B': _make_workbook({
'英国': [('2026-08-31', 'B0ABC111', 'GBP 8.50', 'BrandA'),
('2026-08-31', 'B0ABC222', 'GBP 12.00', 'BrandB')],
}),
'Shop C': _make_workbook({
'法国': [('2026-08-29', 'B0ABC333', 'EUR 7.50', 'BrandC')],
}),
}
def _result_row(self, result_id, shop_name, country_codes_json=None):
return {
'result_id': result_id,
'task_id': result_id + 100,
'user_id': 7,
'shop_name': shop_name,
'shop_id': shop_name.lower(),
'task_no': f'task-{result_id}',
'task_status': 'SUCCESS',
'result_success': 1,
'result_error': None,
'task_error': None,
'file_error': None,
'result_file_url': f'object-{result_id}',
'result_filename': f'result-{result_id}.xlsx',
'result_file_size': 10,
'row_count': 2,
'request_json': '{}',
'country_codes_json': country_codes_json,
'created_at': '2026-08-31T01:15:00',
'updated_at': '2026-08-31T01:15:00',
'finished_at': '2026-08-31T01:15:00',
'latest_file_updated_at': '2026-08-31T01:15:00',
'file_job_id': None,
'file_status': 'SUCCESS',
'username': 'operator',
}
class _FakeCursor:
def __init__(self, group_rows, result_rows):
self.group_rows = group_rows
self.result_rows = result_rows
self.kind = None
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def execute(self, sql, params=()):
if 'COUNT(*) AS total' in sql:
self.kind = 'count'
elif 'AS latest_created_at' in sql and 'GROUP BY' in sql:
self.kind = 'groups'
elif 'GROUP_CONCAT' in sql:
self.kind = 'group_names'
else:
self.kind = 'results'
def fetchone(self):
return {'total': len(self.group_rows)}
def fetchall(self):
if self.kind == 'groups':
return self.group_rows
if self.kind == 'group_names':
return [{'shop_name': row['shop_name'], 'group_name': 'Group-' + row['shop_name']}
for row in self.group_rows]
return self.result_rows
class _FakeConnection:
def __init__(self, cursor):
self.cursor_value = cursor
def cursor(self):
return self.cursor_value
def close(self):
pass
def _make_workbook_bytes(self, shop_name):
stream = BytesIO()
self.shop_files[shop_name].save(stream)
return stream.getvalue()
def _run_request(self, query=''):
cursor = self._FakeCursor(
self.group_rows,
[self._result_row(i + 1, row['shop_name'], '["UK","DE"]') for i, row in enumerate(self.group_rows)],
)
connection = self._FakeConnection(cursor)
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-asins?' + query):
with patch.object(admin_api, 'get_db', return_value=connection), \
patch.object(admin_api, '_ensure_backend_menu_access', return_value=(None, None, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=(None, None, None)), \
patch.object(admin_api, '_shop_data_crawl_fetch_result_bytes',
side_effect=lambda row: self._make_workbook_bytes(row['shop_name'])):
return admin_api.shop_data_crawl_duplicate_asins.__wrapped__()
def test_detects_duplicate_asins_across_shops(self):
response = self._run_request('page=1&page_size=10')
self.assertEqual(response.status_code, 200)
body = response.get_json()
self.assertTrue(body['success'])
self.assertEqual(body['total'], 1) # 只有 B0ABC111 跨店重复
self.assertEqual(body['analyzed_shop_count'], 3)
item = body['items'][0]
self.assertEqual(item['asin'], 'B0ABC111')
self.assertEqual(item['shop_count'], 2)
shops = {occ['shop_name'] for occ in item['occurrences']}
self.assertEqual(shops, {'Shop A', 'Shop B'})
# 国家与日期从行/表头正确映射
shop_a = next(occ for occ in item['occurrences'] if occ['shop_name'] == 'Shop A')
self.assertIn('UK', shop_a['country_codes'])
self.assertEqual(shop_a['date'], '2026-08-30')
def test_pagination_when_page_out_of_range(self):
response = self._run_request('page=2&page_size=10')
body = response.get_json()
self.assertEqual(body['total'], 1)
self.assertEqual(body['items'], [])
def test_parse_workbook_skips_unknown_sheets(self):
wb = _make_workbook({'英国': [('2026-08-30', 'B0TEST01', 'GBP 1.00', '')]})
# 手工追加一个无标准表头的 sheet,模拟未知表
ws = wb.create_sheet('未知表')
ws.append(['随便', '某列'])
ws.append(['2026-08-30', 'B0NOHEADER'])
rows = admin_api._shop_data_crawl_parse_workbook(wb)
self.assertEqual([row['asin'] for row in rows], ['B0TEST01'])
if __name__ == '__main__':
unittest.main()