Stabilize patrol delete automation against delayed Amazon UI

The patrol-delete flow now waits for country and listing-status controls before acting, uses the dedicated PatrolDeleteTask entry point, and includes the browser helper scripts required by the bulk-delete path. Tests cover delayed dropdown readiness, payload rejection/retry behavior, complete-draft normalization, filtered bulk deletion, and asset fallback lookup.

Constraint: Amazon listing UI exposes status controls through dynamic KAT components and delayed option rendering

Rejected: Keep row-by-row delete flow | bulk selection is the current implemented path and is covered by the added helper scripts

Confidence: high

Scope-risk: moderate

Directive: Do not remove the patrol_delete JS helper files without checking app/amazon/patrol_delete.py script loading

Tested: uv run --group dev pytest tests\\test_amazon_base.py tests\\test_patrol_delete.py

Tested: uv run python -m py_compile amazon\\base.py amazon\\main.py blueprints\\main.py

Not-tested: Real Amazon Seller Central browser session
This commit is contained in:
koko
2026-04-28 20:46:50 +08:00
parent 4ff4d8cfdc
commit 1be28f10e9
14 changed files with 1062 additions and 170 deletions

View File

@@ -26,9 +26,15 @@ main_bp = Blueprint('main', __name__)
def _resolve_asset_filename(filename):
"""Resolve hashed Vite assets without maintaining hardcoded alias tables."""
safe_name = os.path.basename(filename)
exact_path = os.path.join(ASSETS_DIR, safe_name)
if os.path.isfile(exact_path):
return safe_name
asset_dirs = [
ASSETS_DIR,
os.path.abspath(os.path.join(BASE_DIR, 'new_web_source', 'assets')),
os.path.abspath(os.path.join(BASE_DIR, '..', 'new_web_source', 'assets')),
]
for asset_dir in asset_dirs:
exact_path = os.path.join(asset_dir, safe_name)
if os.path.isfile(exact_path):
return safe_name
base_name, ext = os.path.splitext(safe_name)
if not base_name or not ext:
@@ -38,9 +44,13 @@ def _resolve_asset_filename(filename):
if len(parts) < 2:
return safe_name
try:
asset_names = os.listdir(ASSETS_DIR)
except OSError:
asset_names = []
for asset_dir in asset_dirs:
try:
asset_names.extend(os.listdir(asset_dir))
except OSError:
continue
if not asset_names:
return safe_name
for prefix_length in range(len(parts) - 1, 0, -1):
@@ -82,25 +92,30 @@ def wb():
return _render_html('index.html')
@main_bp.route('/brand')
@login_required
def brand_page():
repo_brand_path = os.path.abspath(os.path.join(BASE_DIR, '..', 'web_source', 'brand.html'))
if os.path.isfile(repo_brand_path):
try:
with open(repo_brand_path, 'rb') as f:
raw = f.read()
try:
from html_crypto import decrypt
content = decrypt(raw).decode('utf-8')
except Exception:
if raw[:7] == b'gAAAAAB':
raise
content = raw.decode('utf-8', errors='replace')
return render_template_string(content, user_id=session.get('user_id'))
except Exception:
pass
return _render_html('brand.html', user_id=session.get('user_id'))
@main_bp.route('/brand')
@login_required
def brand_page():
candidate_paths = [
os.path.join(BASE_DIR, 'web_source', 'brand.html'),
os.path.abspath(os.path.join(BASE_DIR, '..', 'web_source', 'brand.html')),
]
for candidate_path in candidate_paths:
if not os.path.isfile(candidate_path):
continue
try:
with open(candidate_path, 'rb') as f:
raw = f.read()
try:
from html_crypto import decrypt
content = decrypt(raw).decode('utf-8')
except Exception:
if raw[:7] == b'gAAAAAB':
raise
content = raw.decode('utf-8', errors='replace')
return render_template_string(content, user_id=session.get('user_id'))
except Exception:
continue
return _render_html('brand.html', user_id=session.get('user_id'))
@main_bp.route('/brand/legacy')
@@ -137,12 +152,19 @@ def serve_static(filename):
def serve_assets(filename):
"""提供 static 目录及子目录下的静态文件访问。"""
filename = _resolve_asset_filename(filename)
filepath = os.path.normpath(os.path.join(ASSETS_DIR, filename))
static_abs = os.path.abspath(ASSETS_DIR)
file_abs = os.path.abspath(filepath)
if not file_abs.startswith(static_abs) or not os.path.isfile(file_abs):
return '', 404
return send_file(file_abs, as_attachment=False)
candidate_dirs = [
os.path.abspath(ASSETS_DIR),
os.path.abspath(os.path.join(BASE_DIR, 'new_web_source', 'assets')),
os.path.abspath(os.path.join(BASE_DIR, '..', 'new_web_source', 'assets')),
]
for static_abs in candidate_dirs:
filepath = os.path.normpath(os.path.join(static_abs, filename))
file_abs = os.path.abspath(filepath)
if not file_abs.startswith(static_abs):
continue
if os.path.isfile(file_abs):
return send_file(file_abs, as_attachment=False)
return '', 404
@main_bp.route('/new_web_source/<path:filename>')