super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(admin): A1 管理后台收敛 Java 单后台完整实现并修复 guard 误拦内部令牌
Build Backend JAR / build (push) Has been cancelled
Details

Browse Source 
- Java 补齐后台全部迁移差集:shopduplicatecheck 店铺数据重复检查模块(V108 扫描表+查询/扫描服务)、
  PinyinAbbrUtil 拼音缩写、ImageHistory 接口调整为内部可用、AdminUser 支持内部令牌操作并放宽列表上限
- Flask 后台 admin_api.py 路由收敛转发 Java、admin.html/admin.js 适配新后台形态
- AdminApiGuardFilter 对可信 X-Internal-Token 放行(controller 自校验兜底),修复客户端仅凭
  内部令牌调用 /api/admin/shop-manages/credential 被误拦 401
- 测试:AdminApiGuardFilterTest 补可信/假令牌用例;AdminUserServiceTest 补菜单权限 mock;
  shopduplicatecheck 新增查询/聚合/CSV 单测
This commit is contained in:
黄自达
2026-09-04 11:30:09 +08:00
parent 3420c72c1d
commit 8241cd704e
43 changed files with 3071 additions and 1318 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_admin_shop_data_duplicate_asins.py
-307
View File
@@ -1,307 +0,0 @@
"""重复 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):
"""构造结果 Workbook：rows_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.app.config['SECRET_KEY'] = 'test-secret'
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 共享 ASIN1；Shop C 单独 ASIN3
# 国家码按店铺实际 sheet：A=UK+DE、B=仅UK、C=FR
self.shop_country = {
'Shop A': '["UK","DE"]',
'Shop B': '["UK"]',
'Shop C': '["FR"]',
}
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 _build_cache(self):
"""构造与旧扫描一致的缓存：3 家店铺、跨店重复 B0ABC111。"""
return {
'scanned_at': '2026-09-03 03:10:00',
'summary': {'shop_count': 3, 'total': 1},
'items': [{
'asin': 'B0ABC111',
'shop_count': 2,
'record_count': 3,
'occurrences': [
{'asin': 'B0ABC111', 'date': '2026-08-30', 'price': 'GBP 9.99', 'brand': 'BrandA',
'shop_name': 'Shop A', 'group_name': 'Group-Shop A', 'country_codes': ['UK', 'DE']},
{'asin': 'B0ABC111', 'date': '2026-08-30', 'price': 'EUR 10.99', 'brand': 'BrandA',
'shop_name': 'Shop A', 'group_name': 'Group-Shop A', 'country_codes': ['DE']},
{'asin': 'B0ABC111', 'date': '2026-08-31', 'price': 'GBP 8.50', 'brand': 'BrandA',
'shop_name': 'Shop B', 'group_name': 'Group-Shop B', 'country_codes': ['UK']},
],
}],
}
def _run_request(self, query='', cache=None):
# 界面默认读缓存；force 走实时扫描（需 mock 扫描核心）
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-asins?' + query):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
patch.object(admin_api, '_latest_duplicate_scan', return_value=cache), \
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)):
return admin_api.shop_data_crawl_duplicate_asins()
def test_detects_duplicate_asins_across_shops(self):
response = self._run_request('page=1&page_size=10', cache=self._build_cache())
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)
self.assertEqual(item['record_count'], 3) # ShopA 英国/德国 + ShopB 英国 共 3 条
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')
self.assertEqual(body['scanned_at'], '2026-09-03 03:10:00')
def test_no_cache_returns_pending_empty(self):
# 无缓存时（定时任务尚未执行）：返回空 + pending 提示
response = self._run_request('page=1&page_size=10', cache=None)
body = response.get_json()
self.assertTrue(body['success'])
self.assertEqual(body['total'], 0)
self.assertEqual(body['items'], [])
self.assertTrue(body['pending'])
def test_cache_filters_match_live_semantics(self):
cache = self._build_cache()
# 店铺过滤 Shop B：命中 → 完整记录
response = self._run_request('page=1&page_size=10&shop_name=Shop+B', cache=cache)
body = response.get_json()
self.assertEqual(body['total'], 1)
self.assertEqual(body['items'][0]['asin'], 'B0ABC111')
# 日期范围 08-31 命中
response = self._run_request('page=1&page_size=10&date_from=2026-08-31&date_to=2026-08-31', cache=cache)
body = response.get_json()
self.assertEqual(body['total'], 1)
# 国家 FR 无命中
response = self._run_request('page=1&page_size=10&country=FR', cache=cache)
body = response.get_json()
self.assertEqual(body['total'], 0)
def test_force_runs_live_scan(self):
# force=1 触发实时扫描并保存缓存后返回
cursor = self._FakeCursor(
self.group_rows,
[self._result_row(i + 1, row['shop_name'], self.shop_country[row['shop_name']])
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?page=1&page_size=10&force=1'):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
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, '_save_duplicate_scan', return_value=1), \
patch.object(admin_api, '_latest_duplicate_scan',
return_value=self._build_cache()), \
patch.object(admin_api, '_shop_data_crawl_fetch_result_bytes',
side_effect=lambda row: self._make_workbook_bytes(row['shop_name'])):
response = admin_api.shop_data_crawl_duplicate_asins()
self.assertIsNotNone(response)
body = response.get_json()
self.assertTrue(body['success'])
self.assertEqual(body['total'], 1)
self.assertEqual(body['items'][0]['asin'], 'B0ABC111')
def test_force_conflict_when_lock_held(self):
# 锁被占（定时任务/他请求在扫）时 force 返回 409
cursor = self._FakeCursor(self.group_rows[:0], [])
connection = self._FakeConnection(cursor)
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-asins?page=1&page_size=10&force=1'):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
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, '_duplicate_scan_lock'):
admin_api._duplicate_scan_lock.acquire.return_value = False
response = admin_api.shop_data_crawl_duplicate_asins()
# 409 返回 (jsonify, status) tuple
self.assertEqual(response[1], 409)
def test_latest_duplicate_scan_deserializes_json_columns(self):
"""真实读库：summary_json / payload_json 为 JSON 字符串，需反序列化为 dict/list。"""
class _ScanCursor(self._FakeCursor):
def __init__(self):
self.kind = None
def execute(self, sql, params=()):
self.kind = 'scan'
def fetchone(self):
return {
'id': 9,
'summary_json': '{"shop_count": 3, "total": 1, "source": "import"}',
'payload_json': '[{"asin": "B0ABC111", "shop_count": 2}]',
'created_at': datetime(2026, 9, 3, 13, 35, 25),
}
class _ScanConnection(self._FakeConnection):
def cursor(self):
return _ScanCursor()
with patch.object(admin_api, 'get_db', return_value=_ScanConnection(None)):
cache = admin_api._latest_duplicate_scan()
self.assertIsNotNone(cache)
self.assertEqual(cache['summary']['shop_count'], 3)
self.assertEqual(cache['summary']['source'], 'import')
self.assertEqual(cache['items'][0]['asin'], 'B0ABC111')
self.assertEqual(cache['scanned_at'], '2026-09-03 13:35:25')
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'])
def test_date_key_normalizes_chinese_datetime(self):
# 中文日期归一化为 ISO 供范围比较
cases = {
'2026年8月18日 上午4:34': '2026-08-18',
'2026年8月19日 05:48': '2026-08-19',
'2026-08-19': '2026-08-19',
'2026.8.19': '2026-08-19',
'2026/8/19': '2026-08-19',
'': '',
'abc': 'abc', # 无法识别时原样返回，范围比较自然失败
}
for raw, expected in cases.items():
self.assertEqual(admin_api._shop_data_date_key(raw), expected)
if __name__ == '__main__':
unittest.main()
backend/tests/test_admin_shop_data_duplicate_check.py
+273 -172
View File
@@ -1,4 +1,11 @@
"""店铺数据重复检查接口单元测试：全量缓存格式、矩阵分页、主管/超管数据范围过滤。"""
"""店铺数据重复检查接口（撞款 duplicate-check）转发契约测试。
四个端点已迁移到 Java（/api/admin/shop-data-crawl/duplicate-check-{overview,items,detail,export}），
Flask 侧仅做：本地菜单/数据权限预检 → 带 operatorId + X-Internal-Token 转发 Java →
把 Java ApiResponse.data 原样透传并加 success 包装。本文件验证透传形状、参数名映射与错误码映射；
筛选/裁剪/排序/统计语义由 backend-java 模块的 Java 单测覆盖。
"""
import io
import sys
import unittest
from pathlib import Path
@@ -10,215 +17,309 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from blueprints import admin_api
def _shop(name, group, country_codes=None):
return {
'shop_name': name,
'group_name': group,
'country_codes': country_codes or [],
'rows': [],
}
def _java_ok(data):
"""模拟 Java ApiResponse 成功体：{'success': True, 'data': {...}, 'message': ...}。"""
return {'success': True, 'data': data, 'message': '操作成功'}, None, 200
class DuplicateCheckApiTest(unittest.TestCase):
"""直接对接口函数做单元测试：缓存通过 _latest_duplicate_scan mock 注入。"""
def _java_fail_json(code, message):
"""构建转发失败态：error_response 为 Flask jsonify 对象（需在请求上下文内调用）。"""
return ({'success': False, 'message': message, 'code': code},
admin_api.jsonify({'success': False, 'error': message}),
code)
class DuplicateCheckProxyTest(unittest.TestCase):
def setUp(self):
self.app = Flask(__name__)
self.app.config['SECRET_KEY'] = 'test-secret'
# 4 家店 / 3 个组；ASIN 分布：
# A1: ShopA(UK) + ShopB(UK) —— 跨店重复
# B1: ShopA(DE) 唯一
# C1: ShopC(FR) 唯一
# D1: ShopD(UK) 唯一
self.cache = {
self.shops = [
{'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['UK', 'DE'], 'asin_count': 3, 'record_count': 3},
{'shop_name': 'ShopB', 'group_name': 'GroupA', 'country_codes': ['UK'], 'asin_count': 3, 'record_count': 3},
{'shop_name': 'ShopC', 'group_name': 'GroupC', 'country_codes': ['FR'], 'asin_count': 2, 'record_count': 2},
{'shop_name': 'ShopD', 'group_name': 'GroupD', 'country_codes': ['UK'], 'asin_count': 2, 'record_count': 2},
]
self.items_all = [
{'asin': 'E0000001', 'shop_count': 3, 'record_count': 3, 'occurrences': [
{'asin': 'E0000001', 'date': '2026年8月29日 上午4:34', 'price': 'GBP 12.00', 'brand': 'BrandE',
'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['UK'], 'country': 'UK'},
{'asin': 'E0000001', 'date': '2026-08-30', 'price': 'GBP 12.00', 'brand': 'BrandE',
'shop_name': 'ShopB', 'group_name': 'GroupA', 'country_codes': ['UK'], 'country': 'UK'},
{'asin': 'E0000001', 'date': '2026-08-31', 'price': 'EUR 12.00', 'brand': 'BrandE',
'shop_name': 'ShopC', 'group_name': 'GroupC', 'country_codes': ['FR'], 'country': 'FR'},
]},
{'asin': 'A0000001', 'shop_count': 2, 'record_count': 2, 'occurrences': [
{'asin': 'A0000001', 'date': '2026-08-30', 'price': 'GBP 9.99', 'brand': 'BrandA',
'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['UK', 'DE'], 'country': 'UK'},
{'asin': 'A0000001', 'date': '2026-08-30 08:30:00', 'price': 'GBP 8.50', 'brand': 'BrandA',
'shop_name': 'ShopB', 'group_name': 'GroupA', 'country_codes': ['UK'], 'country': 'UK'},
]},
{'asin': 'F0000001', 'shop_count': 2, 'record_count': 2, 'occurrences': [
{'asin': 'F0000001', 'date': '2026-08-28', 'price': 'GBP 6.00', 'brand': 'BrandF',
'shop_name': 'ShopB', 'group_name': 'GroupA', 'country_codes': ['UK'], 'country': 'UK'},
{'asin': 'F0000001', 'date': '2026-08-30', 'price': 'GBP 6.00', 'brand': 'BrandF',
'shop_name': 'ShopD', 'group_name': 'GroupD', 'country_codes': ['UK'], 'country': 'UK'},
]},
{'asin': 'B0000001', 'shop_count': 1, 'record_count': 1, 'occurrences': [
{'asin': 'B0000001', 'date': '2026-08-30', 'price': 'EUR 10.99', 'brand': 'BrandA',
'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['DE'], 'country': 'DE'},
]},
{'asin': 'C0000001', 'shop_count': 1, 'record_count': 1, 'occurrences': [
{'asin': 'C0000001', 'date': '2026-08-29', 'price': 'EUR 7.50', 'brand': 'BrandC',
'shop_name': 'ShopC', 'group_name': 'GroupC', 'country_codes': ['FR'], 'country': 'FR'},
]},
{'asin': 'D0000001', 'shop_count': 1, 'record_count': 1, 'occurrences': [
{'asin': 'D0000001', 'date': '2026-08-28', 'price': 'GBP 5.00', 'brand': 'BrandD',
'shop_name': 'ShopD', 'group_name': 'GroupD', 'country_codes': ['UK'], 'country': 'UK'},
]},
]
self.overview_all = {
'pending': False,
'scanned_at': '2026-09-04 03:10:00',
'summary': {
'shop_count': 4, 'asin_total': 4, 'record_total': 5,
'duplicate_asin_total': 1, 'duplicate_shop_count': 2,
'site_count': 3, 'asin_per_shop': 1.2, 'source': 'job',
'shop_count': 4, 'asin_total': 6, 'record_total': 10,
'duplicate_asin_total': 3, 'duplicate_shop_count': 4,
'site_count': 3, 'asin_per_shop': 2.5, 'source': 'job',
},
'shops': [
{'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['UK', 'DE'], 'asin_count': 2, 'record_count': 3},
{'shop_name': 'ShopB', 'group_name': 'GroupA', 'country_codes': ['UK'], 'asin_count': 1, 'record_count': 1},
{'shop_name': 'ShopC', 'group_name': 'GroupC', 'country_codes': ['FR'], 'asin_count': 1, 'record_count': 1},
{'shop_name': 'ShopD', 'group_name': 'GroupD', 'country_codes': ['UK'], 'asin_count': 1, 'record_count': 1},
],
'items': [
{'asin': 'A0000001', 'shop_count': 2, 'record_count': 2, 'occurrences': [
{'asin': 'A0000001', 'date': '2026-08-30', 'price': 'GBP 9.99', 'brand': 'BrandA',
'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['UK', 'DE'], 'country': 'UK'},
{'asin': 'A0000001', 'date': '2026-08-31', 'price': 'GBP 8.50', 'brand': 'BrandA',
'shop_name': 'ShopB', 'group_name': 'GroupA', 'country_codes': ['UK'], 'country': 'UK'},
]},
{'asin': 'B0000001', 'shop_count': 1, 'record_count': 1, 'occurrences': [
{'asin': 'B0000001', 'date': '2026-08-30', 'price': 'EUR 10.99', 'brand': 'BrandA',
'shop_name': 'ShopA', 'group_name': 'GroupA', 'country_codes': ['DE'], 'country': 'DE'},
]},
{'asin': 'C0000001', 'shop_count': 1, 'record_count': 1, 'occurrences': [
{'asin': 'C0000001', 'date': '2026-08-29', 'price': 'EUR 7.50', 'brand': 'BrandC',
'shop_name': 'ShopC', 'group_name': 'GroupC', 'country_codes': ['FR'], 'country': 'FR'},
]},
{'asin': 'D0000001', 'shop_count': 1, 'record_count': 1, 'occurrences': [
{'asin': 'D0000001', 'date': '2026-08-28', 'price': 'GBP 5.00', 'brand': 'BrandD',
'shop_name': 'ShopD', 'group_name': 'GroupD', 'country_codes': ['UK'], 'country': 'UK'},
]},
],
'shops': self.shops,
}
def _access_patches(self, role='super_admin', current_row=None):
return [
patch.object(admin_api, '_ensure_backend_menu_access',
return_value=(role, current_row or {'id': 1}, None)),
patch.object(admin_api, '_ensure_shop_data_crawl_data_access',
return_value=(role, current_row or {'id': 1}, None)),
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache),
]
def _patched(self, data):
return patch.object(admin_api, '_proxy_backend_java', side_effect=lambda *a, **k: _java_ok(data))
def _call(self, url, role='super_admin', current_row=None):
def _call(self, view_name, url, data, role='super_admin', current_row=None):
with self.app.test_request_context(url):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True):
for p in self._access_patches(role, current_row):
p.start()
try:
return admin_api.shop_data_crawl_duplicate_check_items()
finally:
for p in reversed(self._access_patches(role, current_row)):
p.stop()
def test_overview_super_admin_sees_all(self):
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-overview'):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
patch.object(admin_api, '_ensure_backend_menu_access', return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache):
response = admin_api.shop_data_crawl_duplicate_check_overview()
patch.object(admin_api, '_ensure_backend_menu_access',
return_value=(role, current_row or {'id': 1}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access',
return_value=(role, current_row or {'id': 1}, None)), \
self._patched(data):
return getattr(admin_api, view_name)()
def test_overview_super_admin_passthrough(self):
response = self._call('shop_data_crawl_duplicate_check_overview',
'/api/admin/shop-data-crawl/duplicate-check-overview',
self.overview_all)
body = response.get_json()
self.assertTrue(body['success'])
self.assertFalse(body['pending'])
self.assertEqual(body['summary']['asin_total'], 4)
self.assertEqual(body['summary']['record_total'], 5)
self.assertEqual(body['summary']['duplicate_asin_total'], 1)
self.assertEqual(body['summary']['asin_total'], 6)
self.assertEqual(len(body['shops']), 4)
def test_items_matix_columns_are_shops(self):
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=monitor')
def test_overview_pending_empty_summary(self):
response = self._call('shop_data_crawl_duplicate_check_overview',
'/api/admin/shop-data-crawl/duplicate-check-overview',
{'pending': True, 'scanned_at': '', 'summary': {}, 'shops': []})
body = response.get_json()
self.assertTrue(body['success'])
self.assertEqual(body['total'], 1) # monitor 只保留跨店重复
self.assertEqual(body['items'][0]['asin'], 'A0000001')
self.assertTrue(body['pending'])
self.assertEqual(body['summary'], {})
def test_overview_force_passed_through(self):
captured = {}
data = dict(self.overview_all)
def side_effect(*args, **kwargs):
captured['params'] = kwargs.get('params')
captured['timeout'] = kwargs.get('timeout')
return _java_ok(data)
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-overview?force=1'):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
patch.object(admin_api, '_ensure_backend_menu_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_proxy_backend_java', side_effect=side_effect):
response = admin_api.shop_data_crawl_duplicate_check_overview()
body = response.get_json()
self.assertTrue(body['success'])
self.assertEqual((captured['params'] or {}).get('force'), '1')
# force 同步扫描需要长超时
self.assertEqual(captured['timeout'], (10, 1800))
def test_overview_scan_conflict_409(self):
def side_effect(*args, **kwargs):
return _java_fail_json(409, '扫描进行中，请稍后刷新')
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-overview?force=1'):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
patch.object(admin_api, '_ensure_backend_menu_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_proxy_backend_java', side_effect=side_effect):
response = admin_api.shop_data_crawl_duplicate_check_overview()
self.assertEqual(response[1], 409)
self.assertFalse(response[0].get_json()['success'])
self.assertEqual(response[0].get_json()['error'], '扫描进行中，请稍后刷新')
def test_items_matrix_monitor_total_and_columns(self):
data = {'pending': False, 'items': self.items_all[:3], 'shops': self.shops,
'total': 3, 'page': 1, 'page_size': 20, 'scanned_at': '2026-09-04 03:10:00'}
response = self._call('shop_data_crawl_duplicate_check_items',
'/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=monitor',
data)
body = response.get_json()
self.assertTrue(body['success'])
self.assertEqual(body['total'], 3)
self.assertEqual([shop['shop_name'] for shop in body['shops']],
['ShopA', 'ShopB', 'ShopC', 'ShopD'])
def test_items_all_view_includes_unique_asins(self):
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=all')
body = response.get_json()
self.assertEqual(body['total'], 4)
def test_items_all_view_passthrough(self):
data = {'pending': False, 'items': self.items_all, 'shops': self.shops,
'total': 6, 'page': 1, 'page_size': 20, 'scanned_at': '2026-09-04 03:10:00'}
response = self._call('shop_data_crawl_duplicate_check_items',
'/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=all',
data)
self.assertEqual(response.get_json()['total'], 6)
def test_items_filter_by_asin_and_site(self):
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&asin=A0000001&site=UK')
body = response.get_json()
self.assertEqual(body['total'], 1)
self.assertEqual(body['items'][0]['asin'], 'A0000001')
def test_items_shop_name_alias_merged_and_camel_params(self):
captured = {}
def test_leader_sees_only_own_group_shops(self):
# 主管 id=653 只管理 GroupA（含 ShopA/ShopB）
with patch.object(admin_api, '_shop_data_managed_shop_names',
return_value={'shopa', 'shopb'}):
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=all',
role='admin', current_row={'id': 653})
body = response.get_json()
self.assertEqual([shop['shop_name'] for shop in body['shops']], ['ShopA', 'ShopB'])
self.assertEqual(body['total'], 2) # A0000001（跨店）+ B0000001（唯一）
# 跨店 ASIN 在主管范围内仍是 2 家店
item = next(i for i in body['items'] if i['asin'] == 'A0000001')
self.assertEqual(item['shop_count'], 2)
def side_effect(*args, **kwargs):
captured['params'] = kwargs.get('params') or {}
return _java_ok({'pending': False, 'items': [], 'shops': [],
'total': 0, 'page': 1, 'page_size': 20, 'scanned_at': ''})
def test_leader_overview_stats_recomputed_after_filter(self):
# 主管见 2 家店：唯一ASIN 2、上架记录 3、重复ASIN 1、重复店铺 2
with patch.object(admin_api, '_shop_data_managed_shop_names',
return_value={'shopa', 'shopb'}):
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-overview'):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
patch.object(admin_api, '_ensure_backend_menu_access', return_value=('admin', {'id': 653}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=('admin', {'id': 653}, None)), \
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache):
response = admin_api.shop_data_crawl_duplicate_check_overview()
body = response.get_json()
self.assertEqual(body['summary']['shop_count'], 2)
self.assertEqual(body['summary']['asin_total'], 2)
self.assertEqual(body['summary']['record_total'], 3)
self.assertEqual(body['summary']['duplicate_asin_total'], 1)
def test_leader_sees_nothing_when_no_group(self):
with patch.object(admin_api, '_shop_data_managed_shop_names', return_value=set()):
response = self._call('/api/admin/shop-data-crawl/duplicate-check-items?page=1&page_size=20&view=all',
role='admin', current_row={'id': 999999})
body = response.get_json()
self.assertEqual(body['shops'], [])
self.assertEqual(body['total'], 0)
def test_export_csv_contains_filtered_rows(self):
"""导出 CSV：行=上架记录，含 BOM，按筛选裁剪。"""
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-export?'):
with self.app.test_request_context(
'/api/admin/shop-data-crawl/duplicate-check-items?page=2&page_size=50&view=all&shop=ShopA&asin=abc'):
# 内部代理 operatorId 从 flask session 取当前登录管理员
admin_api.session['user_id'] = 1
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
patch.object(admin_api, '_ensure_backend_menu_access', return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache):
response = admin_api.shop_data_crawl_duplicate_check_export()
self.assertEqual(response.status_code, 200)
text = response.get_data(as_text=True)
self.assertTrue(text.startswith(''))
lines = text.lstrip('').strip().splitlines()
self.assertEqual(lines[0], 'ASIN,店铺数,店铺,分组,站点,上架时间,价格,品牌')
self.assertEqual(len(lines), 3) # header + 两条上架记录（A0000001 两店各一条）
patch.object(admin_api, '_ensure_backend_menu_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_proxy_backend_java', side_effect=side_effect):
admin_api.shop_data_crawl_duplicate_check_items()
params = captured['params']
self.assertEqual(params['page'], '2')
self.assertEqual(params['pageSize'], '50')
self.assertEqual(params['shopName'], 'ShopA') # shop 别名合并进 shop_name→shopName
self.assertEqual(params['asin'], 'abc')
self.assertIn('operatorId', params)
def test_items_denied_403_message_preserved(self):
def denied_menu(*args, **kwargs):
return ('admin', {'id': 5}, (
admin_api.jsonify({'success': False, 'error': '无权访问店铺数据记录模块'}), 403))
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-items'):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
patch.object(admin_api, '_ensure_backend_menu_access', side_effect=denied_menu), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access',
return_value=('admin', {'id': 5}, None)), \
patch.object(admin_api, '_proxy_backend_java',
side_effect=lambda *a, **k: _java_ok({})):
response = admin_api.shop_data_crawl_duplicate_check_items()
self.assertEqual(response[1], 403)
body = response[0].get_json()
self.assertEqual(body['error'], '无权访问店铺数据记录模块')
def test_detail_sorted_passthrough(self):
detail_items = [
{'asin': 'E0000001', 'shop_count': 3, 'record_count': 3, 'shop_names': ['ShopA', 'ShopB', 'ShopC'],
'brand': 'BrandE', 'first_date': '2026-08-29 04:34:00', 'occurrences': self.items_all[0]['occurrences']},
{'asin': 'A0000001', 'shop_count': 2, 'record_count': 2, 'shop_names': ['ShopA', 'ShopB'],
'brand': 'BrandA', 'first_date': '2026-08-30 00:00:00', 'occurrences': self.items_all[1]['occurrences']},
{'asin': 'F0000001', 'shop_count': 2, 'record_count': 2, 'shop_names': ['ShopB', 'ShopD'],
'brand': 'BrandF', 'first_date': '2026-08-28 00:00:00', 'occurrences': self.items_all[2]['occurrences']},
]
data = {'pending': False, 'items': detail_items, 'total': 3,
'page': 1, 'page_size': 6, 'scanned_at': '2026-09-04 03:10:00'}
response = self._call('shop_data_crawl_duplicate_check_detail',
'/api/admin/shop-data-crawl/duplicate-check-detail?page=1&page_size=6',
data)
body = response.get_json()
self.assertEqual(body['total'], 3)
self.assertEqual(body['items'][0]['asin'], 'E0000001')
self.assertEqual(body['items'][0]['first_date'], '2026-08-29 04:34:00')
def test_detail_pending(self):
response = self._call('shop_data_crawl_duplicate_check_detail',
'/api/admin/shop-data-crawl/duplicate-check-detail',
{'pending': True, 'items': [], 'total': 0, 'page': 1, 'page_size': 6, 'scanned_at': ''})
body = response.get_json()
self.assertTrue(body['pending'])
self.assertEqual(body['items'], [])
def test_export_streams_java_csv(self):
csv_bytes = ('' + 'ASIN,店铺数,店铺,分组,站点,上架时间,价格,品牌\r\n'
'E0000001,3,ShopA,GroupA,UK,2026年8月29日 上午4:34,GBP 12.00,BrandE\r\n').encode('utf-8')
class FakeResp:
status_code = 200
headers = {'Content-Disposition': 'attachment; filename="shop-data-duplicate-check.csv"',
'Content-Type': 'text/csv; charset=utf-8'}
def iter_content(self, chunk_size=1):
yield csv_bytes
def close(self):
pass
def json(self):
return {}
class FakeSession:
def get(self, *args, **kwargs):
return FakeResp()
def test_export_monitor_view_excludes_unique_asins(self):
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-export?view=monitor'):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
patch.object(admin_api, '_ensure_backend_menu_access', return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access', return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_latest_duplicate_scan', return_value=self.cache):
patch.object(admin_api, 'get_current_admin_role',
return_value=('super_admin', {'id': 1})), \
patch.object(admin_api, '_ensure_backend_menu_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_get_backend_java_session', return_value=FakeSession()):
response = admin_api.shop_data_crawl_duplicate_check_export()
lines = response.get_data(as_text=True).lstrip('').strip().splitlines()
self.assertEqual(len(lines), 3)
self.assertIn('A0000001', lines[1])
body = b''.join(response.response)
self.assertTrue(body.startswith(b'\xef\xbb\xbf'))
self.assertIn('ASIN,店铺数'.encode('utf-8'), body)
self.assertIn('E0000001'.encode('utf-8'), body)
def test_internal_request_falls_back_to_system_operator(self):
"""无请求上下文（定时扫描线程）时，内部请求用系统级超管作为 operatorId。"""
with patch('utils.auth.session', {'user_id': 1}): # 仅用于模拟无异常环境
# 无请求上下文：has_request_context() 为 False，直接走 _resolve_system_operator_id
with patch.object(admin_api, '_resolve_internal_token', return_value='test-token'), \
patch.object(admin_api, '_resolve_system_operator_id', return_value=7):
headers, params = admin_api._backend_java_internal_request()
self.assertEqual(headers.get('X-Internal-Token'), 'test-token')
self.assertEqual(params, {'operatorId': 7})
def test_export_no_scan_400(self):
class FakeResp:
status_code = 400
headers = {}
def test_internal_request_system_operator_missing_fails(self):
"""无请求上下文且系统中没有任何管理员时，直接报错而不是传 0。"""
with patch.object(admin_api, '_resolve_internal_token', return_value='test-token'), \
patch.object(admin_api, '_resolve_system_operator_id', return_value=None):
with self.assertRaises(ValueError):
admin_api._backend_java_internal_request()
def json(self):
return {'success': False, 'message': '暂无扫描结果，请先点击「重新分析」'}
def test_fetch_result_bytes_rejects_json_error_body(self):
"""Java 内部端点返回 JSON 错误体（如鉴权失败）时，抛出业务错误而不是 BadZipFile。"""
fake_response = Mock()
fake_response.raise_for_status = Mock()
fake_response.iter_content = Mock(return_value=[
'{"success":false,"message":"用户不存在","data":null,"code":401}'.encode('utf-8')])
fake_response.close = Mock()
with patch.object(admin_api, '_backend_java_internal_request',
return_value=({'X-Internal-Token': 'test-token'}, {'operatorId': 7})), \
patch.object(admin_api, '_get_backend_java_session') as fake_session:
fake_session.return_value.get = Mock(return_value=fake_response)
with self.assertRaises(ValueError) as ctx:
admin_api._shop_data_crawl_fetch_result_bytes({'result_id': 123})
self.assertIn('用户不存在', str(ctx.exception))
def close(self):
pass
def iter_content(self, chunk_size=1):
return iter(())
class FakeSession:
def get(self, *args, **kwargs):
return FakeResp()
with self.app.test_request_context('/api/admin/shop-data-crawl/duplicate-check-export'):
with patch('utils.auth.session', {'user_id': 1}), \
patch('utils.auth.is_session_user_valid', return_value=True), \
patch.object(admin_api, 'get_current_admin_role',
return_value=('super_admin', {'id': 1})), \
patch.object(admin_api, '_ensure_backend_menu_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_ensure_shop_data_crawl_data_access',
return_value=('super_admin', {'id': 1}, None)), \
patch.object(admin_api, '_get_backend_java_session', return_value=FakeSession()):
response = admin_api.shop_data_crawl_duplicate_check_export()
self.assertEqual(response[1], 400)
self.assertEqual(response[0].get_json()['error'], '暂无扫描结果，请先点击「重新分析」')
if __name__ == '__main__':