"""
管理员 API 蓝图：用户管理、生成历史、版本管理（后台）
"""
import json
import os
import re
import secrets
import tempfile
import threading
import zipfile
from datetime import datetime
from pathlib import Path
from urllib.parse import quote
import traceback

import requests
from requests.adapters import HTTPAdapter
from flask import (
    Blueprint,
    request,
    jsonify,
    session,
    current_app,
    g,
    Response,
    send_file,
    has_request_context,
)

import pymysql
from werkzeug.security import generate_password_hash

from utils.db import get_db
from utils.auth import admin_required, login_required, get_current_admin_role

from ali_oss import upload_file as oss_upload_file

try:
    from config import bucket_path, backend_java_base_url
except ImportError:
    bucket_path = os.environ.get('BUCKET_PATH', 'nanri-image/')
    backend_java_base_url = os.environ.get('BACKEND_JAVA_BASE_URL', 'http://api.aishufu.top:18080/').rstrip('/')

admin_api = Blueprint('admin_api', __name__, url_prefix='/api/admin')
_backend_java_session_local = threading.local()
_internal_token_lock = threading.Lock()
IMAGE_VIDEO_DATA_PERMISSION_KEY = 'admin_image_video_task_data'

ADMIN_MENU_ACCESS_CONFIG = {
    'dedupe-total-data': {
        'column_key': 'admin_dedupe_total_data',
        'route_path': 'dedupe-total-data',
        'error': '无权访问数据去重汇总数据',
    },
    'shop-manage': {
        'column_key': 'admin_shop_manage',
        'route_path': 'shop-manage',
        'error': '无权访问店铺管理模块',
    },
    'skip-price-asin': {
        'column_key': 'admin_skip_price_asin',
        'route_path': 'skip-price-asin',
        'error': '无权访问跳过跟价 ASIN 模块',
    },
    'query-asin': {
        'column_key': 'admin_query_asin',
        'route_path': 'query-asin',
        'error': '无权访问查询 ASIN 模块',
    },
    'product-categories': {
        'column_key': 'admin_product_categories',
        'route_path': 'product-categories',
        'error': '无权访问商品类目模块',
    },
    'invalid-asin-data': {
        'column_key': 'admin_invalid_asin_data',
        'route_path': 'invalid-asin-data',
        'error': '无权访问不符合ASIN数据模块',
    },
    'image-video-tasks': {
        'column_key': 'admin_image_video_tasks',
        'route_path': 'image-video-tasks',
        'error': '无权访问视频任务管理模块',
    },
}

ADMIN_MENU_ACCESS_CONFIG.update({
    'users': {
        'column_key': 'admin_users',
        'route_path': 'users',
        'error': '无权访问用户管理模块',
    },
    'columns': {
        'column_key': 'admin_columns',
        'route_path': 'columns',
        'error': '无权访问栏目权限配置模块',
    },
    'shop-keys': {
        'column_key': 'admin_shop_keys',
        'route_path': 'shop-keys',
        'error': '无权访问店铺密钥管理模块',
    },
    'history': {
        'column_key': 'admin_history',
        'route_path': 'history',
        'error': '无权访问查看生成记录模块',
    },
    'version': {
        'column_key': 'admin_version',
        'route_path': 'version',
        'error': '无权访问版本管理模块',
    },
    'digital-human-version': {
        'column_key': 'digital_human_version',
        'route_path': 'digital-human-version',
        'error': '无权访问数字人版本管理模块',
    },
})


def _safe_version_key(version):
    """将版本号转换为安全的 OSS 对象名片段"""
    s = (version or '').strip()
    s = re.sub(r'[^\w.\-]', '_', s)
    return s or 'unknown'


def _get_backend_java_session():
    http_session = getattr(_backend_java_session_local, 'session', None)
    if http_session is None:
        http_session = requests.Session()
        adapter = HTTPAdapter(pool_connections=8, pool_maxsize=32, max_retries=0)
        http_session.mount('http://', adapter)
        http_session.mount('https://', adapter)
        _backend_java_session_local.session = http_session
    return http_session


def _backend_java_forward_headers():
    """Forward the caller's Java auth material through the Flask compatibility proxy."""
    if not has_request_context():
        return {}
    headers = {}
    for name in ('Authorization', 'Cookie', 'X-Device-Id'):
        value = request.headers.get(name)
        if value:
            headers[name] = value
    return headers


def _proxy_backend_java(
    method,
    path,
    *,
    params=None,
    json_data=None,
    files=None,
    data=None,
    timeout=10,
    headers=None,
):
    url = f"{backend_java_base_url}{path}"
    forwarded_headers = _backend_java_forward_headers()
    if headers:
        forwarded_headers.update(headers)
    try:
        requester = requests if files else _get_backend_java_session()
        resp = requester.request(
            method=method,
            url=url,
            params=params,
            json=json_data,
            files=files,
            data=data,
            headers=forwarded_headers or None,
            timeout=timeout,
        )
    except requests.RequestException:
        return None, jsonify({'success': False, 'error': 'backend-java 服务不可用'}), 502
    try:
        data = resp.json()
    except ValueError:
        data = None
    if resp.status_code >= 400:
        if isinstance(data, dict):
            return data, jsonify({'success': False, 'error': data.get('message') or data.get('error') or 'backend-java 请求失败'}), resp.status_code
        return None, jsonify({'success': False, 'error': 'backend-java 请求失败'}), resp.status_code
    if not isinstance(data, dict):
        return None, jsonify({'success': False, 'error': 'backend-java 返回格式错误'}), 502
    if not data.get('success'):
        raw_code = data.get('code')
        try:
            error_status = int(raw_code) if not isinstance(raw_code, bool) else None
        except (TypeError, ValueError):
            error_status = None
        status = (
            error_status
            if resp.status_code == 200 and error_status is not None and 400 <= error_status <= 599
            else 200
        )
        return data, jsonify({'success': False, 'error': data.get('message') or '操作失败'}), status
    return data, None, 200


def _format_permission_item(item):
    if not isinstance(item, dict):
        return {}
    created_at = item.get('createdAt')
    if created_at in (None, ''):
        created_at = item.get('created_at')
    if hasattr(created_at, 'strftime'):
        created_at_text = created_at.strftime('%Y-%m-%d %H:%M')
    else:
        created_at_text = (str(created_at).replace('T', ' ')[:16]) if created_at else ''
    sort_order = item.get('sortOrder')
    if sort_order is None:
        sort_order = item.get('sort_order')
    parent_id = item.get('parentId')
    if parent_id is None:
        parent_id = item.get('parent_id')
    root_column_key = item.get('rootColumnKey') or item.get('root_column_key') or ''
    formatted = {
        'id': item.get('id'),
        'name': item.get('name') or '',
        'column_key': item.get('columnKey') or item.get('column_key') or '',
        'menu_type': item.get('menuType') or item.get('menu_type') or 'app',
        'route_path': item.get('routePath') or item.get('route_path') or '',
        'sort_order': sort_order if sort_order is not None else 0,
        'created_at': created_at_text,
        # Keep both spellings while old Flask pages migrate to the Java schema.
        'parent_id': parent_id,
        'parentId': parent_id,
        'root_column_key': root_column_key,
        'rootColumnKey': root_column_key,
    }
    children = item.get('children')
    if isinstance(children, list):
        formatted['children'] = [_format_permission_item(child) for child in children]
    return formatted


def _iter_permission_items(items):
    """Yield a flat view of a possibly recursive Java permission response."""
    for item in items or []:
        if not isinstance(item, dict):
            continue
        yield item
        children = item.get('children')
        if isinstance(children, list):
            yield from _iter_permission_items(children)


def _permission_response_items(result):
    """Read menu lists from both ApiResponse and legacy Flask-shaped payloads."""
    if not isinstance(result, dict):
        return []
    payload = result.get('data')
    if isinstance(payload, list):
        return payload
    if isinstance(payload, dict):
        for key in ('items', 'menus', 'columns', 'permissions'):
            value = payload.get(key)
            if isinstance(value, list):
                return value
    for key in ('items', 'menus', 'columns', 'permissions'):
        value = result.get(key)
        if isinstance(value, list):
            return value
    return []


def _permission_response_column_ids(result):
    """Read direct IDs from Java's `data.columnIds` or old snake_case aliases."""
    if not isinstance(result, dict):
        return []
    payload = result.get('data')
    candidates = [payload, result]
    if isinstance(payload, dict):
        candidates.extend(payload.get(key) for key in ('data', 'items') if isinstance(payload.get(key), dict))
    for candidate in candidates:
        if not isinstance(candidate, dict):
            continue
        for key in ('columnIds', 'column_ids', 'ids'):
            value = candidate.get(key)
            if isinstance(value, list):
                normalized = []
                for raw in value:
                    try:
                        value_int = int(raw)
                    except (TypeError, ValueError):
                        continue
                    if value_int > 0 and value_int not in normalized:
                        normalized.append(value_int)
                return normalized
    return []


class _PermissionProxyError(Exception):
    """Carry a Flask response when a Java permission call cannot be completed."""

    def __init__(self, response, status):
        super().__init__('backend-java permission request failed')
        self.response = response
        self.status = status


def _proxy_permission_java(method, path, *, params=None, json_data=None, current_row=None):
    """Call Java permission APIs using either forwarded JWT or trusted Flask identity."""
    proxy_params = {}
    request_row = getattr(g, '_current_user_row', None) if has_request_context() else None
    current_row = current_row or request_row or {}
    operator_id = current_row.get('id')
    if not operator_id and has_request_context():
        operator_id = session.get('user_id')
    if operator_id:
        proxy_params['operatorId'] = operator_id
    if params:
        proxy_params.update(params)
    internal_token = _resolve_internal_token()
    headers = {'X-Internal-Token': internal_token} if internal_token else None
    result, error_response, status = _proxy_backend_java(
        method,
        path,
        params=proxy_params or None,
        json_data=json_data,
        headers=headers,
    )
    return result, error_response, status


def _resolve_internal_token():
    configured_token = (os.environ.get('AIIMAGE_INTERNAL_TOKEN') or '').strip()
    if configured_token:
        return configured_token
    configured_path = (os.environ.get('AIIMAGE_INTERNAL_TOKEN_FILE') or '').strip()
    token_path = Path(configured_path).expanduser() if configured_path else Path.home() / '.aiimage' / 'internal-token'
    if configured_path and not token_path.is_absolute():
        return ''
    with _internal_token_lock:
        try:
            existing_token = token_path.read_text(encoding='utf-8').strip()
            if existing_token:
                return existing_token
        except FileNotFoundError:
            pass
        except OSError:
            return ''

        try:
            token_path.parent.mkdir(parents=True, exist_ok=True)
            generated_token = secrets.token_urlsafe(48)
            file_descriptor = os.open(str(token_path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(file_descriptor, 'w', encoding='utf-8') as token_file:
                token_file.write(generated_token)
                token_file.write('\n')
            return generated_token
        except FileExistsError:
            try:
                return token_path.read_text(encoding='utf-8').strip()
            except OSError:
                return ''
        except OSError:
            return ''


def _java_permission_items(*, menu_type=None, user_id=None, effective=False, current_row=None, role=None):
    """Fetch permission menus from Java and normalize its response for Flask callers."""
    params = {}
    if menu_type:
        params['menuType'] = menu_type
    if effective:
        if user_id is None:
            user_id = _get_current_admin_id(current_row)
        if user_id is None:
            return [], None, 200
        path = f'/api/admin/permission-users/{int(user_id)}/column-permissions'
    else:
        path = '/api/admin/permission-menus'
    result, error_response, status = _proxy_permission_java(
        'GET',
        path,
        params=params,
        current_row=current_row,
    )
    if error_response is not None:
        return [], error_response, status
    raw_items = _permission_response_items(result)
    return [_format_permission_item(item) for item in raw_items], None, status


def _java_user_direct_column_ids(user_id, *, menu_type=None, current_row=None, role=None):
    params = {}
    if menu_type:
        params['menuType'] = menu_type
    result, error_response, status = _proxy_permission_java(
        'GET',
        f'/api/admin/permission-users/{int(user_id)}/columns',
        params=params,
        current_row=current_row,
    )
    if error_response is not None:
        return [], error_response, status
    return _permission_response_column_ids(result), None, status


def _java_update_user_direct_column_ids(
    user_id,
    column_ids,
    *,
    menu_type=None,
    current_row=None,
    role=None,
):
    normalized = _normalize_column_ids(column_ids)
    params = {'menuType': menu_type} if menu_type else None
    result, error_response, status = _proxy_permission_java(
        'PUT',
        f'/api/admin/permission-users/{int(user_id)}/columns',
        params=params,
        json_data={'columnIds': normalized},
        current_row=current_row,
    )
    return result, error_response, status


def _normalize_column_ids(raw_ids):
    if raw_ids is None:
        return []
    if not isinstance(raw_ids, (list, tuple, set)):
        raise ValueError('column_ids 必须是数组')
    normalized = []
    for raw in raw_ids:
        if isinstance(raw, bool):
            raise ValueError('column_ids 包含无效 ID')
        try:
            value = int(raw)
        except (TypeError, ValueError) as exc:
            raise ValueError('column_ids 包含无效 ID') from exc
        if value <= 0:
            raise ValueError('column_ids 包含无效 ID')
        if value not in normalized:
            normalized.append(value)
    return normalized


def _effective_permission_sets(user_id, *, menu_type='admin', current_row=None, role=None):
    items, error_response, status = _java_permission_items(
        menu_type=menu_type,
        user_id=user_id,
        effective=True,
        current_row=current_row,
        role=role,
    )
    if error_response is not None:
        raise _PermissionProxyError(error_response, status)
    key_set = set()
    route_set = set()
    id_set = set()
    for item in _iter_permission_items(items):
        try:
            item_id = int(item.get('id'))
        except (TypeError, ValueError):
            item_id = None
        if item_id and item_id > 0:
            id_set.add(item_id)
        column_key = str(item.get('column_key') or '').strip()
        route_path = str(item.get('route_path') or '').strip()
        if column_key:
            key_set.add(column_key)
        if route_path:
            route_set.add(route_path)
    return id_set, key_set, route_set


def _parse_optional_int(value):
    if value is None:
        return None
    text = str(value).strip()
    if text == '':
        return None
    return int(text)


_IMAGE_VIDEO_SECRET_KEYS = {
    'api_key', 'apikey', 'token', 'coze_token', 'cozetoken', 't8_key', 't8key',
    't8_video_key', 't8videokey', 't8star_key', 't8starkey',
    'ai_conductor_key', 'aiconductorkey', 'copy_api_key', 'copyapikey',
    'voice_api_key', 'voiceapikey',
}


def _parse_json_value(value, fallback=None):
    if value in (None, ''):
        return fallback
    if isinstance(value, (dict, list)):
        return value
    try:
        return json.loads(value)
    except (TypeError, ValueError):
        return value


def _mask_image_video_secrets(value):
    if isinstance(value, dict):
        masked = {}
        for key, item in value.items():
            canonical = re.sub(r'[^a-z0-9]', '', str(key).lower())
            configured_keys = {re.sub(r'[^a-z0-9]', '', item) for item in _IMAGE_VIDEO_SECRET_KEYS}
            masked[key] = '******' if canonical in configured_keys and item not in (None, '') else _mask_image_video_secrets(item)
        return masked
    if isinstance(value, list):
        return [_mask_image_video_secrets(item) for item in value]
    return value


def _image_video_mode(request_value):
    payload = _parse_json_value(request_value, {})
    try:
        mode = str(payload.get('parameters', {}).get('video_info', {}).get('mode', '')).strip()
    except AttributeError:
        mode = ''
    if mode == '1':
        return '图生视频'
    if mode == '2':
        return '视频复刻'
    return '未知'


def _image_video_access_sql(role, current_row, alias='t'):
    if role == 'super_admin':
        return '1=1', []
    admin_id = int(current_row['id'])
    return (
        f"({alias}.user_id = %s OR {alias}.user_id IN "
        "(SELECT id FROM users WHERE role = 'normal' AND created_by_id = %s))",
        [admin_id, admin_id],
    )


def _format_admin_datetime(value):
    if hasattr(value, 'strftime'):
        return value.strftime('%Y-%m-%d %H:%M:%S')
    return str(value).replace('T', ' ')[:19] if value else ''


def _image_video_urls(row):
    originals = _parse_json_value(row.get('video_urls_json'), [])
    originals = originals if isinstance(originals, list) else []
    archived = _parse_json_value(row.get('archived_videos_json'), [])
    archived = archived if isinstance(archived, list) else []
    archived_by_source = {
        str(item.get('sourceUrl') or ''): item
        for item in archived if isinstance(item, dict)
    }
    items = []
    for source_url in originals:
        source_url = str(source_url or '')
        stored = archived_by_source.get(source_url) or {}
        items.append({
            'source_url': source_url,
            'archived_url': stored.get('url') or '',
            'object_key': stored.get('objectKey') or '',
            'archive_status': stored.get('status') or '',
            'archive_error': stored.get('error') or '',
            'display_url': stored.get('url') or source_url,
        })
    return items


def _image_video_admin_item(row, include_json=False):
    videos = _image_video_urls(row)
    item = {
        'task_id': row.get('id'),
        'user_id': row.get('user_id'),
        'username': row.get('username') or '',
        'group_name': row.get('group_name') or '',
        'mode': _image_video_mode(row.get('request_json')),
        'status': row.get('status') or '',
        'coze_status': row.get('coze_status') or '',
        'coze_execute_id': row.get('coze_execute_id') or '',
        'videos': videos,
        'video_url': videos[0]['display_url'] if videos else '',
        'debug_url': row.get('debug_url') or '',
        'archive_status': row.get('archive_status') or '',
        'archive_error': row.get('archive_error') or '',
        'archive_attempt_count': row.get('archive_attempt_count') or 0,
        'submitted_at': _format_admin_datetime(row.get('submitted_at')),
        'completed_at': _format_admin_datetime(row.get('completed_at')),
    }
    if include_json:
        item.update({
            'error_message': row.get('error_message') or '',
            'archived_at': _format_admin_datetime(row.get('archived_at')),
            'request': _mask_image_video_secrets(_parse_json_value(row.get('request_json'), {})),
            'submit_response': _parse_json_value(row.get('submit_response_json'), {}),
            'result': _parse_json_value(row.get('result_json'), {}),
        })
    return item


PRODUCT_CATEGORY_DEFAULT_TREE = [
    {
        'name': '护肤品',
        'key': 'skincare',
        'children': [
            ('面霜', 'skincare_cream'),
            ('精华', 'skincare_essence'),
            ('面膜', 'skincare_mask'),
            ('爽肤水', 'skincare_toner'),
            ('身体乳', 'skincare_body_lotion'),
        ],
    },
    {
        'name': '彩妆',
        'key': 'makeup',
        'children': [
            ('口红', 'makeup_lipstick'),
            ('粉底', 'makeup_foundation'),
            ('眼影', 'makeup_eyeshadow'),
            ('睫毛', 'makeup_mascara'),
        ],
    },
    {
        'name': '洗护',
        'key': 'haircare',
        'children': [
            ('洗发水', 'haircare_shampoo'),
            ('护发素', 'haircare_conditioner'),
            ('染发剂', 'haircare_hair_dye'),
            ('烫发膏', 'haircare_perm_cream'),
        ],
    },
    {
        'name': '功效类',
        'key': 'functional',
        'children': [
            ('防晒', 'functional_sunscreen'),
            ('祛斑', 'functional_spot_removal'),
            ('美白', 'functional_whitening'),
            ('祛痘产品', 'functional_acne_care'),
        ],
    },
    {
        'name': '香氛',
        'key': 'fragrance',
        'children': [
            ('香水', 'fragrance_perfume'),
            ('香薰精油(高酒精属于危险品)', 'fragrance_essential_oil_hazardous'),
        ],
    },
]


def _get_current_admin_id(current_row=None):
    if current_row and current_row.get('id'):
        return current_row.get('id')
    return session.get('user_id')


def _ensure_admin_menu_access(*menu_names):
    role, current_row = get_current_admin_role()
    if role == 'super_admin':
        return role, current_row, None
    if role != 'admin' or not current_row:
        return role, current_row, (jsonify({'success': False, 'error': '需要管理员权限'}), 403)
    try:
        _, key_set, route_set = _effective_permission_sets(
            _get_current_admin_id(current_row),
            current_row=current_row,
            role=role,
        )
    except _PermissionProxyError as exc:
        return role, current_row, (exc.response, exc.status)
    for menu_name in menu_names:
        config = ADMIN_MENU_ACCESS_CONFIG.get(menu_name) or {}
        column_key = (config.get('column_key') or '').strip()
        route_path = (config.get('route_path') or '').strip()
        if (column_key and column_key in key_set) or (route_path and route_path in route_set):
            return role, current_row, None
    fallback_menu = menu_names[0] if menu_names else ''
    fallback_config = ADMIN_MENU_ACCESS_CONFIG.get(fallback_menu) or {}
    error_message = fallback_config.get('error') or '无权访问当前模块'
    return role, current_row, (jsonify({'success': False, 'error': error_message}), 403)


def _ensure_dedupe_total_data_access():
    return _ensure_backend_menu_access('dedupe-total-data')


def _load_current_backend_menu_items():
    role, current_row = get_current_admin_role()
    if not role or not current_row:
        return None, None, None, (jsonify({'success': False, 'error': '需要登录'}), 403)
    user_id = _get_current_admin_id(current_row)
    if not user_id:
        return role, current_row, None, (jsonify({'success': False, 'error': '当前登录用户缺少有效 ID'}), 400)
    items, error_response, status = _java_permission_items(
        menu_type='admin',
        user_id=user_id,
        effective=True,
        current_row=current_row,
        role=role,
    )
    if error_response is not None:
        return role, current_row, None, (error_response, status)
    return role, current_row, items, None


def _ensure_backend_menu_access(*menu_names):
    role, current_row = get_current_admin_role()
    if role == 'super_admin':
        return role, current_row, None
    if not role or not current_row:
        return role, current_row, (jsonify({'success': False, 'error': '需要登录'}), 403)
    try:
        _, key_set, route_set = _effective_permission_sets(
            _get_current_admin_id(current_row),
            current_row=current_row,
            role=role,
        )
    except _PermissionProxyError as exc:
        return role, current_row, (exc.response, exc.status)
    for menu_name in menu_names:
        config = ADMIN_MENU_ACCESS_CONFIG.get(menu_name) or {}
        column_key = (config.get('column_key') or '').strip()
        route_path = (config.get('route_path') or '').strip()
        if (column_key and column_key in key_set) or (route_path and route_path in route_set):
            return role, current_row, None
    fallback_menu = menu_names[0] if menu_names else ''
    fallback_config = ADMIN_MENU_ACCESS_CONFIG.get(fallback_menu) or {}
    error_message = fallback_config.get('error') or '无权访问当前模块'
    return role, current_row, (jsonify({'success': False, 'error': error_message}), 403)


def _ensure_image_video_data_access():
    role, current_row = get_current_admin_role()
    if role == 'super_admin':
        return role, current_row, None
    if not role or not current_row:
        return role, current_row, (jsonify({'success': False, 'error': '需要登录'}), 403)
    try:
        _, key_set, route_set = _effective_permission_sets(
            _get_current_admin_id(current_row),
            menu_type=None,
            current_row=current_row,
            role=role,
        )
    except _PermissionProxyError as exc:
        return role, current_row, (exc.response, exc.status)
    if IMAGE_VIDEO_DATA_PERMISSION_KEY in key_set or 'image-video-task-data' in route_set:
        return role, current_row, None
    return role, current_row, (jsonify({'success': False, 'error': '无权查看视频任务数据'}), 403)


def _ensure_product_category_access():
    role, current_row, items, denied = _load_current_backend_menu_items()
    if denied:
        return role, current_row, denied
    if role == 'super_admin':
        return role, current_row, None
    for item in items or []:
        if (item.get('column_key') or '').strip() == 'admin_product_categories':
            return role, current_row, None
        if (item.get('route_path') or '').strip() == 'product-categories':
            return role, current_row, None
    return role, current_row, (jsonify({'success': False, 'error': '无权访问商品类目模块'}), 403)


def _get_column_ids_by_route_paths(route_paths, menu_type='admin'):
    normalized_paths = [(path or '').strip() for path in (route_paths or []) if (path or '').strip()]
    if not normalized_paths:
        return []
    items, error_response, _ = _java_permission_items(menu_type=menu_type)
    if error_response is not None:
        return []
    wanted = set(normalized_paths)
    return [
        int(item['id'])
        for item in _iter_permission_items(items)
        if item.get('id') is not None and (item.get('route_path') or '').strip() in wanted
    ]


def _grant_backend_menu_permissions(user_ids, route_paths):
    normalized_user_ids = [int(uid) for uid in (user_ids or []) if str(uid).isdigit() and int(uid) > 0]
    column_ids = _get_column_ids_by_route_paths(route_paths, 'admin')
    if not normalized_user_ids or not column_ids:
        return
    role, current_row = get_current_admin_role()
    for user_id in normalized_user_ids:
        try:
            direct_ids, error_response, _ = _java_user_direct_column_ids(
                user_id,
                current_row=current_row,
                role=role,
            )
            if error_response is not None:
                continue
            _, error_response, _ = _java_update_user_direct_column_ids(
                user_id,
                sorted(set(direct_ids).union(column_ids)),
                current_row=current_row,
                role=role,
            )
            if error_response is not None:
                continue
        except (TypeError, ValueError):
            continue


@admin_api.route('/current-user')
@login_required
def get_admin_current_user():
    role, current_row = get_current_admin_role()
    if not role or not current_row:
        return jsonify({'success': False, 'error': '需要管理员权限'}), 403
    return jsonify({
        'success': True,
        'item': {
            'id': current_row.get('id'),
            'username': current_row.get('username') or session.get('username') or '',
            'role': role,
        }
    })


@admin_api.route('/current-user/menus')
@login_required
def get_admin_current_user_menus():
    _, _, items, denied = _load_current_backend_menu_items()
    if denied:
        return denied
    return jsonify({'success': True, 'items': items})


@admin_api.route('/logout', methods=['POST'])
@login_required
def admin_logout():
    session.clear()
    response = jsonify({'success': True, 'msg': '退出成功', 'redirect': '/login?logout=1'})
    response.delete_cookie(current_app.config.get('SESSION_COOKIE_NAME', 'session'))
    response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0'
    return response


# ---------- 用户管理 ----------

@admin_api.route('/users')
@login_required
def list_users():
    """分页获取用户列表，支持用户名模糊搜索和按管理员归属筛选普通用户。"""
    role, current_row = get_current_admin_role()
    if not role:
        return jsonify({'success': False, 'error': '需要登录'}), 403
    page = max(1, int(request.args.get('page', 1)))
    _, _, denied = _ensure_backend_menu_access(
        'users', 'history', 'shop-manage', 'skip-price-asin', 'query-asin', 'dedupe-total-data'
    )
    if denied:
        return denied
    page_size = min(999, max(5, int(request.args.get('page_size', 15))))
    offset = (page - 1) * page_size
    search_username = (request.args.get('username') or request.args.get('search') or '').strip()
    created_by_id_arg = request.args.get('created_by_id') or request.args.get('admin_id')
    created_by_id = int(created_by_id_arg) if created_by_id_arg and str(created_by_id_arg).isdigit() else None
    # 用户范围只由角色决定，用户名不能绕过权限模型。
    can_view_all_users = role == 'super_admin'
    can_view_child_users = role == 'admin' and not can_view_all_users
    if not can_view_all_users:
        created_by_id = None
    try:
        conn = get_db()
        with conn.cursor() as cur:
            if can_view_all_users:
                where_parts = ["1=1"]
                params = []
                if search_username:
                    where_parts.append("u.username LIKE %s")
                    params.append("%" + search_username + "%")
                if created_by_id is not None:
                    where_parts.append("u.created_by_id = %s")
                    params.append(created_by_id)
                where_sql = " AND ".join(where_parts)
                cur.execute(
                    """SELECT u.id, u.username, u.is_admin, u.role, u.created_at, u.created_by_id,
                          creator.username AS creator_username
                   FROM users u
                   LEFT JOIN users creator ON creator.id = u.created_by_id
                   WHERE """ + where_sql + """ ORDER BY u.id LIMIT %s OFFSET %s""",
                    tuple(params) + (page_size, offset),
                )
                rows = cur.fetchall()
                cur.execute("SELECT COUNT(*) as total FROM users u WHERE " + where_sql, tuple(params))
                total = cur.fetchone()['total']
                cur.execute("SELECT id, username FROM users WHERE role = 'admin' ORDER BY id")
                admins = [{'id': r['id'], 'username': r['username']} for r in cur.fetchall()]
            elif can_view_child_users:
                admin_id = current_row['id']
                where_parts = ["(u.id = %s OR (u.role = 'normal' AND u.created_by_id = %s))"]
                params = [admin_id, admin_id]
                if search_username:
                    where_parts.append("u.username LIKE %s")
                    params.append("%" + search_username + "%")
                where_sql = " AND ".join(where_parts)
                cur.execute(
                    """SELECT u.id, u.username, u.is_admin, u.role, u.created_at, u.created_by_id,
                          creator.username AS creator_username
                   FROM users u
                   LEFT JOIN users creator ON creator.id = u.created_by_id
                   WHERE """ + where_sql + """ ORDER BY u.id LIMIT %s OFFSET %s""",
                    tuple(params) + (page_size, offset),
                )
                rows = cur.fetchall()
                cur.execute(
                    "SELECT COUNT(*) as total FROM users u WHERE " + where_sql,
                    tuple(params),
                )
                total = cur.fetchone()['total']
                admins = []
            else:
                where_parts = ["u.id = %s"]
                params = [current_row['id']]
                if search_username:
                    where_parts.append("u.username LIKE %s")
                    params.append("%" + search_username + "%")
                where_sql = " AND ".join(where_parts)
                cur.execute(
                    """SELECT u.id, u.username, u.is_admin, u.role, u.created_at, u.created_by_id,
                          creator.username AS creator_username
                   FROM users u
                   LEFT JOIN users creator ON creator.id = u.created_by_id
                   WHERE """ + where_sql + """ ORDER BY u.id LIMIT %s OFFSET %s""",
                    tuple(params) + (page_size, offset),
                )
                rows = cur.fetchall()
                cur.execute(
                    "SELECT COUNT(*) as total FROM users u WHERE " + where_sql,
                    tuple(params),
                )
                total = cur.fetchone()['total']
                admins = []
            items = [
                {
                    'id': r['id'],
                    'username': r['username'],
                    'is_admin': bool(r.get('is_admin')),
                    'role': r.get('role') or 'normal',
                    'created_by_id': r.get('created_by_id'),
                    'creator_username': r.get('creator_username') or '',
                    'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r.get('created_at') else '',
                }
                for r in rows
            ]
        conn.close()
        return jsonify({
            'success': True,
            'items': items,
            'total': total,
            'page': page,
            'page_size': page_size,
            'current_user_id': session.get('user_id'),
            'current_user_role': role,
            'current_user_username': current_row.get('username') or '',
            'admins': admins,
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})

@admin_api.route('/user', methods=['POST'])
@admin_required
def create_user():
    _, _, denied = _ensure_admin_menu_access('users')
    if denied:
        return denied
    data = request.get_json() or {}
    username = (data.get('username') or '').strip()
    password = data.get('password') or ''
    role, current_row = get_current_admin_role()
    if not role:
        return jsonify({'success': False, 'error': '需要管理员权限'}), 403
    want_role = (data.get('role') or 'normal').strip() or 'normal'
    if want_role not in ('admin', 'normal'):
        want_role = 'normal'
    if role == 'admin' and want_role == 'admin':
        return jsonify({'success': False, 'error': '仅超级管理员可创建管理员'})
    if want_role == 'admin':
        want_created_by = current_row['id']
    elif role == 'super_admin':
        want_created_by = data.get('created_by_id')
    else:
        want_created_by = current_row['id']
    if not username or not password:
        return jsonify({'success': False, 'error': '用户名和密码不能为空'})
    if len(username) < 2:
        return jsonify({'success': False, 'error': '用户名至少2个字符'})
    if len(password) < 6:
        return jsonify({'success': False, 'error': '密码至少6个字符'})
    if want_role == 'normal' and role == 'super_admin' and want_created_by is None:
        try:
            conn = get_db()
            with conn.cursor() as cur:
                cur.execute("SELECT id FROM users WHERE role = 'admin' ORDER BY id LIMIT 1")
                r = cur.fetchone()
            conn.close()
            want_created_by = r['id'] if r else current_row['id']
        except Exception:
            want_created_by = current_row['id']
    if want_role == 'normal' and want_created_by is None:
        want_created_by = current_row['id']
    is_admin = 1 if want_role in ('super_admin', 'admin') else 0
    pwd_hash = generate_password_hash(password, method='pbkdf2:sha256')
    raw_column_ids = data.get('column_ids') if 'column_ids' in data else data.get('columnIds')
    try:
        column_ids = _normalize_column_ids(raw_column_ids)
    except ValueError as exc:
        return jsonify({'success': False, 'error': str(exc)}), 400
    try:
        conn = get_db()
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO users (username, password_hash, is_admin, role, created_by_id) VALUES (%s, %s, %s, %s, %s)",
                (username, pwd_hash, is_admin, want_role, want_created_by),
            )
            new_uid = cur.lastrowid
            _validate_admin_granted_columns(role, current_row, column_ids)
            # Commit the user row before Java validates and stores direct grants.
            conn.commit()
            _set_user_column_permissions_via_java(
                new_uid,
                column_ids,
                role=role,
                current_row=current_row,
            )
        conn.close()
        return jsonify({'success': True, 'msg': '用户创建成功'})
    except _PermissionProxyError as exc:
        try:
            conn.close()
        except Exception:
            pass
        return exc.response, exc.status
    except ValueError as exc:
        try:
            conn.close()
        except Exception:
            pass
        return jsonify({'success': False, 'error': str(exc)}), 400
    except PermissionError as exc:
        try:
            conn.rollback()
            conn.close()
        except Exception:
            pass
        return jsonify({'success': False, 'error': str(exc)}), 403
    except pymysql.IntegrityError:
        return jsonify({'success': False, 'error': '用户名已存在'})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@admin_api.route('/user/<int:uid>', methods=['PUT'])
@admin_required
def update_user(uid):
    """更新用户（密码、角色、栏目权限）"""
    data = request.get_json() or {}
    _, _, denied = _ensure_admin_menu_access('users')
    if denied:
        return denied
    password = data.get('password')
    want_role = (data.get('role') or '').strip() or data.get('role')
    role, current_row = get_current_admin_role()
    if not role:
        return jsonify({'success': False, 'error': '需要管理员权限'}), 403
    if want_role is None and not password and 'column_ids' not in data and 'columnIds' not in data:
        return jsonify({'success': False, 'error': '请提供要修改的内容'})
    try:
        conn = get_db()
        with conn.cursor() as cur:
            cur.execute("SELECT id, role, created_by_id FROM users WHERE id = %s", (uid,))
            target = cur.fetchone()
            if not target:
                conn.close()
                return jsonify({'success': False, 'error': '用户不存在'})
            if role == 'admin':
                if target['role'] != 'normal' or target.get('created_by_id') != current_row['id']:
                    conn.close()
                    return jsonify({'success': False, 'error': '只能编辑自己创建的普通用户'}), 403
                want_role = None
            else:
                if target.get('role') == 'super_admin':
                    conn.close()
                    return jsonify({'success': False, 'error': '不能修改超级管理员'})
                if want_role == 'super_admin':
                    return jsonify({'success': False, 'error': '不能将用户设为超级管理员'})
                if want_role not in ('admin', 'normal', None, ''):
                    want_role = None
            if password:
                if len(password) < 6:
                    conn.close()
                    return jsonify({'success': False, 'error': '密码至少6个字符'})
                pwd_hash = generate_password_hash(password, method='pbkdf2:sha256')
                cur.execute("UPDATE users SET password_hash = %s WHERE id = %s", (pwd_hash, uid))
            if want_role is not None and want_role != '':
                is_admin = 1 if want_role == 'admin' else 0
                cur.execute(
                    "UPDATE users SET is_admin = %s, role = %s WHERE id = %s",
                    (is_admin, want_role, uid),
                )
            if 'column_ids' in data or 'columnIds' in data:
                try:
                    raw_column_ids = data.get('column_ids') if 'column_ids' in data else data.get('columnIds')
                    normalized_column_ids = _normalize_column_ids(raw_column_ids)
                except ValueError as exc:
                    raise exc
                _validate_admin_granted_columns(role, current_row, normalized_column_ids)
                conn.commit()
                _set_user_column_permissions_via_java(
                    uid,
                    normalized_column_ids,
                    role=role,
                    current_row=current_row,
                )
        conn.commit()
        conn.close()
        return jsonify({'success': True, 'msg': '更新成功'})
    except _PermissionProxyError as exc:
        try:
            conn.close()
        except Exception:
            pass
        return exc.response, exc.status
    except ValueError as exc:
        try:
            conn.close()
        except Exception:
            pass
        return jsonify({'success': False, 'error': str(exc)}), 400
    except PermissionError as exc:
        try:
            conn.rollback()
            conn.close()
        except Exception:
            pass
        return jsonify({'success': False, 'error': str(exc)}), 403
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@admin_api.route('/user/<int:uid>', methods=['DELETE'])
@admin_required
def delete_user(uid):
    """删除用户"""
    _, _, denied = _ensure_admin_menu_access('users')
    if denied:
        return denied
    if session.get('user_id') == uid:
        return jsonify({'success': False, 'error': '不能删除当前登录账号'})
    role, current_row = get_current_admin_role()
    if not role:
        return jsonify({'success': False, 'error': '需要管理员权限'}), 403
    try:
        conn = get_db()
        with conn.cursor() as cur:
            cur.execute("SELECT id, role, created_by_id FROM users WHERE id = %s", (uid,))
            target = cur.fetchone()
            if not target:
                conn.close()
                return jsonify({'success': False, 'error': '用户不存在'})
            if target.get('role') == 'super_admin':
                conn.close()
                return jsonify({'success': False, 'error': '不能删除超级管理员'})
            if role == 'admin':
                if target.get('role') != 'normal' or target.get('created_by_id') != current_row['id']:
                    conn.close()
                    return jsonify({'success': False, 'error': '只能删除自己创建的普通用户'}), 403
            cur.execute("DELETE FROM users WHERE id = %s", (uid,))
            affected = cur.rowcount
        conn.commit()
        conn.close()
        if affected == 0:
            return jsonify({'success': False, 'error': '用户不存在'})
        return jsonify({'success': True, 'msg': '删除成功'})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


# ---------- 生成历史 ----------

@admin_api.route('/history')
@admin_required
def history():
    """管理员分页获取所有生成记录"""
    _, _, denied = _ensure_admin_menu_access('history')
    if denied:
        return denied
    page = max(1, int(request.args.get('page', 1)))
    page_size = min(50, max(10, int(request.args.get('page_size', 15))))
    offset = (page - 1) * page_size
    user_id = request.args.get('user_id', type=int)
    time_start = (request.args.get('time_start') or '').strip()
    time_end = (request.args.get('time_end') or '').strip()
    conditions, params = [], []
    if user_id:
        conditions.append("h.user_id = %s")
        params.append(user_id)
    if time_start:
        conditions.append("h.created_at >= %s")
        params.append(time_start)
    if time_end:
        conditions.append("h.created_at <= %s")
        params.append(time_end + ' 23:59:59' if len(time_end) <= 10 else time_end)
    where_clause = " AND ".join(conditions) if conditions else "1=1"
    params_count = params[:]
    params.extend([page_size, offset])
    try:
        conn = get_db()
        with conn.cursor() as cur:
            cur.execute(
                """SELECT h.id, h.user_id, h.created_at, h.panel_type, h.original_urls, h.params, h.result_urls,
                          h.long_image_url, u.username
                   FROM image_history h
                   LEFT JOIN users u ON h.user_id = u.id
                   WHERE """ + where_clause + """ ORDER BY h.created_at DESC LIMIT %s OFFSET %s""",
                params,
            )
            rows = cur.fetchall()
            cur.execute("SELECT COUNT(*) as total FROM image_history h WHERE " + where_clause, params_count)
            total = cur.fetchone()['total']
        conn.close()

        def _parse_json(val, default=None):
            if val is None:
                return default if default is not None else []
            if isinstance(val, (list, dict)):
                return val
            try:
                return json.loads(val)
            except Exception:
                return default if default is not None else []

        items = []
        for r in rows:
            items.append({
                'id': r['id'],
                'user_id': r['user_id'],
                'username': r.get('username') or '-',
                'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r['created_at'] else '',
                'panel_type': r['panel_type'] or '',
                'original_urls': _parse_json(r['original_urls'], []),
                'params': _parse_json(r['params'], {}),
                'result_urls': _parse_json(r['result_urls'], []),
                'long_image_url': (r.get('long_image_url') or '').strip() or None,
            })
        return jsonify({'success': True, 'items': items, 'total': total, 'page': page, 'page_size': page_size})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


# ---------- 栏目权限配置 ----------

_IMAGE_VIDEO_ADMIN_COLUMNS = """
    t.id, t.user_id, t.status, t.request_json, t.submit_response_json, t.result_json,
    t.video_urls_json, t.debug_url, t.archived_videos_json, t.archive_status,
    t.archive_error, t.archive_attempt_count, t.archived_at, t.error_message,
    t.coze_execute_id, t.coze_status, t.submitted_at, t.completed_at, u.username,
    COALESCE((
        SELECT GROUP_CONCAT(DISTINCT g.group_name ORDER BY g.id SEPARATOR '、')
        FROM biz_shop_manage_group g
        LEFT JOIN biz_shop_manage_group_member gm ON gm.group_id = g.id
        WHERE g.created_by_id = t.user_id
           OR g.user_id = t.user_id
           OR gm.user_id = t.user_id
    ), '') AS group_name
"""


def _parse_admin_datetime_arg(name):
    value = (request.args.get(name) or '').strip()
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace('Z', '+00:00')).replace(tzinfo=None)
    except ValueError as exc:
        raise ValueError(f'{name} 时间格式无效') from exc


@admin_api.route('/image-video-task-permissions', methods=['GET', 'PUT'])
@login_required
def manage_image_video_task_permissions():
    json_data = None
    if request.method == 'PUT':
        data = request.get_json(silent=True) or {}
        raw_user_ids = data.get('user_ids') if 'user_ids' in data else data.get('userIds')
        json_data = {'userIds': raw_user_ids}

    result, error_response, status = _proxy_permission_java(
        request.method,
        '/api/admin/image-video-task-permissions',
        json_data=json_data,
    )
    if error_response is not None:
        return error_response, status
    if request.method == 'GET':
        return jsonify({'success': True, 'items': _permission_response_items(result)})
    return jsonify({
        'success': True,
        'granted_count': result.get('data'),
        'msg': result.get('message') or '视频任务权限已更新',
    })


@admin_api.route('/image-video-tasks/download-zip', methods=['POST'])
@login_required
def download_image_video_tasks_zip():
    _, _, denied = _ensure_backend_menu_access('image-video-tasks')
    if not denied:
        _, _, denied = _ensure_image_video_data_access()
    if denied:
        return denied

    data = request.get_json(silent=True) or {}
    raw_items = data.get('items')
    if not isinstance(raw_items, list) or not raw_items:
        return jsonify({'success': False, 'error': '请至少选择一个视频'}), 400
    if len(raw_items) > 100:
        return jsonify({'success': False, 'error': '单次最多打包 100 个视频'}), 400

    selections = []
    seen = set()
    try:
        for item in raw_items:
            if not isinstance(item, dict):
                raise ValueError
            task_id = int(item.get('task_id'))
            video_index = int(item.get('video_index'))
            if task_id <= 0 or video_index < 0:
                raise ValueError
            key = (task_id, video_index)
            if key not in seen:
                seen.add(key)
                selections.append(key)
    except (TypeError, ValueError):
        return jsonify({'success': False, 'error': '视频选择参数无效'}), 400

    task_ids = sorted({task_id for task_id, _ in selections})
    conn = get_db()
    try:
        with conn.cursor() as cur:
            placeholders = ','.join(['%s'] * len(task_ids))
            cur.execute(
                'SELECT ' + _IMAGE_VIDEO_ADMIN_COLUMNS +
                ' FROM biz_image_video_async_task t LEFT JOIN users u ON u.id = t.user_id '
                f"WHERE t.task_type = 'IMAGE_VIDEO_WORKFLOW' "
                f"AND t.submitted_at >= DATE_SUB(NOW(), INTERVAL 3 DAY) "
                f"AND t.id IN ({placeholders})",
                tuple(task_ids),
            )
            rows = cur.fetchall()
    finally:
        conn.close()
    tasks = {int(row['id']): row for row in rows}

    archive = tempfile.SpooledTemporaryFile(max_size=64 * 1024 * 1024, mode='w+b')
    errors = []
    file_count = 0
    try:
        with zipfile.ZipFile(archive, mode='w', compression=zipfile.ZIP_STORED, allowZip64=True) as output_zip:
            for task_id, video_index in selections:
                row = tasks.get(task_id)
                videos = _image_video_urls(row) if row else []
                if video_index >= len(videos) or not videos[video_index].get('display_url'):
                    errors.append(f'task-{task_id}-video-{video_index + 1}: 视频地址不存在')
                    continue
                url = videos[video_index]['display_url']
                extension_match = re.search(r'\.([a-z0-9]{2,5})(?:[?#]|$)', url, re.IGNORECASE)
                extension = extension_match.group(1).lower() if extension_match else 'mp4'
                filename = f'task-{task_id}-video-{video_index + 1}.{extension}'
                remote_response = None
                try:
                    remote_response = requests.get(url, stream=True, timeout=(10, 120))
                    remote_response.raise_for_status()
                    with output_zip.open(filename, mode='w', force_zip64=True) as target:
                        for chunk in remote_response.iter_content(chunk_size=1024 * 1024):
                            if chunk:
                                target.write(chunk)
                    file_count += 1
                except requests.RequestException as exc:
                    errors.append(f'{filename}: 下载失败 ({exc})')
                finally:
                    if remote_response is not None:
                        remote_response.close()
            if errors:
                output_zip.writestr('download-errors.txt', '\n'.join(errors).encode('utf-8'))
        archive.seek(0)
        response = send_file(
            archive,
            mimetype='application/zip',
            as_attachment=True,
            download_name=f"video-tasks-{datetime.now().strftime('%Y%m%d-%H%M%S')}.zip",
            max_age=0,
        )
        response.headers['X-Archive-File-Count'] = str(file_count)
        response.headers['X-Archive-Error-Count'] = str(len(errors))
        response.call_on_close(archive.close)
        return response
    except Exception:
        archive.close()
        raise


@admin_api.route('/image-video-tasks')
@login_required
def list_image_video_tasks():
    _, _, denied = _ensure_backend_menu_access('image-video-tasks')
    if not denied:
        _, _, denied = _ensure_image_video_data_access()
    if denied:
        return denied
    try:
        page = max(1, int(request.args.get('page', 1)))
        page_size = min(100, max(10, int(request.args.get('page_size', 20))))
        user_id = request.args.get('user_id', type=int)
        username = (request.args.get('username') or '').strip()
        status = (request.args.get('status') or '').strip().upper()
        execute_id = (request.args.get('coze_execute_id') or '').strip()
        submitted_from = _parse_admin_datetime_arg('submitted_from')
        submitted_to = _parse_admin_datetime_arg('submitted_to')

        params = []
        conditions = [
            "t.task_type = 'IMAGE_VIDEO_WORKFLOW'",
            "t.submitted_at >= DATE_SUB(NOW(), INTERVAL 3 DAY)",
        ]
        if user_id:
            conditions.append('t.user_id = %s')
            params.append(user_id)
        if username:
            conditions.append('u.username LIKE %s')
            params.append('%' + username + '%')
        if status:
            conditions.append('t.status = %s')
            params.append(status)
        if execute_id:
            conditions.append('t.coze_execute_id LIKE %s')
            params.append('%' + execute_id + '%')
        if submitted_from:
            conditions.append('t.submitted_at >= %s')
            params.append(submitted_from)
        if submitted_to:
            conditions.append('t.submitted_at <= %s')
            params.append(submitted_to)
        where_sql = ' AND '.join(conditions)
        offset = (page - 1) * page_size

        conn = get_db()
        try:
            with conn.cursor() as cur:
                cur.execute(
                    'SELECT ' + _IMAGE_VIDEO_ADMIN_COLUMNS +
                    ' FROM biz_image_video_async_task t LEFT JOIN users u ON u.id = t.user_id WHERE ' +
                    where_sql + ' ORDER BY t.submitted_at DESC, t.id DESC LIMIT %s OFFSET %s',
                    tuple(params + [page_size, offset]),
                )
                rows = cur.fetchall()
                cur.execute(
                    'SELECT COUNT(*) AS total FROM biz_image_video_async_task t '
                    'LEFT JOIN users u ON u.id = t.user_id WHERE ' + where_sql,
                    tuple(params),
                )
                total = int((cur.fetchone() or {}).get('total') or 0)
        finally:
            conn.close()
        return jsonify({
            'success': True,
            'items': [_image_video_admin_item(row) for row in rows],
            'total': total,
            'page': page,
            'page_size': page_size,
        })
    except ValueError as exc:
        return jsonify({'success': False, 'error': str(exc)}), 400
    except Exception as exc:
        return jsonify({'success': False, 'error': str(exc)}), 500


@admin_api.route('/image-video-tasks/<int:task_id>')
@login_required
def get_image_video_task(task_id):
    _, _, denied = _ensure_backend_menu_access('image-video-tasks')
    if not denied:
        _, _, denied = _ensure_image_video_data_access()
    if denied:
        return denied
    conditions = [
        "t.task_type = 'IMAGE_VIDEO_WORKFLOW'",
        "t.submitted_at >= DATE_SUB(NOW(), INTERVAL 3 DAY)",
        't.id = %s',
    ]
    params = [task_id]
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                'SELECT ' + _IMAGE_VIDEO_ADMIN_COLUMNS +
                ' FROM biz_image_video_async_task t LEFT JOIN users u ON u.id = t.user_id WHERE ' +
                ' AND '.join(conditions) + ' LIMIT 1',
                tuple(params),
            )
            row = cur.fetchone()
    finally:
        conn.close()
    if not row:
        return jsonify({'success': False, 'error': '视频任务不存在或无权访问'}), 404
    return jsonify({'success': True, 'item': _image_video_admin_item(row, include_json=True)})


@admin_api.route('/columns')
@admin_required
def list_columns():
    _, _, denied = _ensure_admin_menu_access('columns', 'users')
    if denied:
        return denied
    menu_type = (request.args.get('menu_type') or request.args.get('menuType') or '').strip()
    items, error_response, status = _java_permission_items(menu_type=menu_type or None)
    if error_response is not None:
        return error_response, status
    items = [
        item for item in items
        if (item.get('column_key') or '').strip() != IMAGE_VIDEO_DATA_PERMISSION_KEY
    ]
    # Keep the legacy `items` field; some Java-aware callers use `data`/`columns`.
    return jsonify({'success': True, 'items': items, 'columns': items, 'data': items})


@admin_api.route('/column', methods=['POST'])
@admin_required
def create_column():
    _, _, denied = _ensure_admin_menu_access('columns')
    if denied:
        return denied
    data = request.get_json() or {}
    name = (data.get('name') or '').strip()
    column_key = (data.get('column_key') or data.get('columnKey') or '').strip()
    menu_type = (data.get('menu_type') or data.get('menuType') or '').strip() or 'admin'
    route_path = (data.get('route_path') or data.get('routePath') or '').strip()
    try:
        sort_order = _parse_optional_int(
            data.get('sort_order') if 'sort_order' in data else data.get('sortOrder')
        )
    except (TypeError, ValueError):
        return jsonify({'success': False, 'error': '排序必须是数字'})
    if not name:
        return jsonify({'success': False, 'error': '栏目名不能为空'})
    if not column_key:
        return jsonify({'success': False, 'error': '栏目标识不能为空'})
    if not route_path:
        return jsonify({'success': False, 'error': '菜单路由不能为空'})
    result, error_response, status = _proxy_permission_java(
        'POST',
        '/api/admin/permission-menus',
        json_data={
            'name': name,
            'columnKey': column_key,
            'menuType': menu_type,
            'routePath': route_path,
            'sortOrder': sort_order,
            'parentId': data.get('parent_id') if data.get('parent_id') not in ('', None) else data.get('parentId'),
        },
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '创建成功',
        'id': item.get('id'),
        'item': _format_permission_item(item),
    })


@admin_api.route('/column/<int:cid>', methods=['PUT'])
@admin_required
def update_column(cid):
    _, _, denied = _ensure_admin_menu_access('columns')
    if denied:
        return denied
    data = request.get_json() or {}
    name = (data.get('name') or '').strip()
    column_key = (data.get('column_key') or data.get('columnKey') or '').strip()
    menu_type = (data.get('menu_type') or data.get('menuType') or '').strip() or 'admin'
    route_path = (data.get('route_path') or data.get('routePath') or '').strip()
    try:
        sort_order = _parse_optional_int(
            data.get('sort_order') if 'sort_order' in data else data.get('sortOrder')
        )
    except (TypeError, ValueError):
        return jsonify({'success': False, 'error': '排序必须是数字'})
    if not name:
        return jsonify({'success': False, 'error': '栏目名不能为空'})
    if not column_key:
        return jsonify({'success': False, 'error': '栏目标识不能为空'})
    if not route_path:
        return jsonify({'success': False, 'error': '菜单路由不能为空'})
    result, error_response, status = _proxy_permission_java(
        'PUT',
        f'/api/admin/permission-menus/{cid}',
        json_data={
            'name': name,
            'columnKey': column_key,
            'menuType': menu_type,
            'routePath': route_path,
            'sortOrder': sort_order,
            'parentId': data.get('parent_id') if data.get('parent_id') not in ('', None) else data.get('parentId'),
        },
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '更新成功',
        'item': _format_permission_item(item),
    })


@admin_api.route('/column/reorder', methods=['POST'])
@admin_required
def reorder_column():
    _, _, denied = _ensure_admin_menu_access('columns')
    if denied:
        return denied
    data = request.get_json() or {}
    try:
        column_id = int(data.get('column_id'))
        target_id = int(data.get('target_id'))
    except (TypeError, ValueError):
        return jsonify({'success': False, 'error': '菜单重排参数无效'}), 400
    if column_id == target_id:
        return jsonify({'success': True, 'msg': '排序未变化'})
    menu_type = (data.get('menu_type') or data.get('menuType') or 'admin').strip() or 'admin'
    items, error_response, status = _java_permission_items(menu_type=menu_type)
    if error_response is not None:
        return error_response, status
    by_id = {}
    for item in _iter_permission_items(items):
        try:
            item_id = int(item.get('id'))
        except (TypeError, ValueError):
            continue
        by_id[item_id] = item
    left = by_id.get(column_id)
    right = by_id.get(target_id)
    if not left or not right:
        return jsonify({'success': False, 'error': '栏目不存在'}), 400

    def _reorder_payload(item, sort_order):
        return {
            'name': item.get('name') or '',
            'columnKey': item.get('column_key') or '',
            'menuType': item.get('menu_type') or menu_type,
            'routePath': item.get('route_path') or '',
            'sortOrder': int(sort_order if sort_order is not None else 0),
            'parentId': item.get('parent_id'),
        }

    left_order = left.get('sort_order') or column_id
    right_order = right.get('sort_order') or target_id
    for item_id, item, sort_order in (
        (column_id, left, right_order),
        (target_id, right, left_order),
    ):
        _, error_response, status = _proxy_permission_java(
            'PUT',
            f'/api/admin/permission-menus/{item_id}',
            json_data=_reorder_payload(item, sort_order),
        )
        if error_response is not None:
            return error_response, status
    return jsonify({'success': True, 'msg': '排序更新成功'})


@admin_api.route('/column/<int:cid>', methods=['DELETE'])
@admin_required
def delete_column(cid):
    _, _, denied = _ensure_admin_menu_access('columns')
    if denied:
        return denied
    result, error_response, status = _proxy_permission_java(
        'DELETE',
        f'/api/admin/permission-menus/{cid}',
    )
    if error_response is not None:
        return error_response, status
    return jsonify({'success': True, 'msg': result.get('message') or '删除成功'})


@admin_api.route('/user/<int:uid>/columns', methods=['GET', 'PUT'])
@admin_required
def get_user_columns(uid):
    _, _, denied = _ensure_admin_menu_access('users')
    if denied:
        return denied
    menu_type = (request.args.get('menu_type') or request.args.get('menuType') or '').strip()
    if request.method == 'PUT':
        data = request.get_json(silent=True) or {}
        raw_ids = data.get('column_ids') if 'column_ids' in data else data.get('columnIds')
        role, current_row = get_current_admin_role()
        try:
            normalized_ids = _normalize_column_ids(raw_ids)
            if role == 'admin':
                allowed_ids, _, _ = _effective_permission_sets(
                    _get_current_admin_id(current_row),
                    menu_type=None,
                    current_row=current_row,
                    role=role,
                )
                if not set(normalized_ids).issubset(allowed_ids):
                    return jsonify({'success': False, 'error': '普通管理员只能分配自己的有效菜单权限'}), 403
            _, error_response, status = _java_update_user_direct_column_ids(
                uid,
                normalized_ids,
                menu_type=menu_type or None,
                current_row=current_row,
                role=role,
            )
            if error_response is not None:
                return error_response, status
        except _PermissionProxyError as exc:
            return exc.response, exc.status
        except ValueError as exc:
            return jsonify({'success': False, 'error': str(exc)}), 400
        return jsonify({'success': True, 'column_ids': normalized_ids, 'columnIds': normalized_ids})
    column_ids, error_response, status = _java_user_direct_column_ids(
        uid,
        menu_type=menu_type or None,
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'column_ids': column_ids,
        'columnIds': column_ids,
        'data': {'columnIds': column_ids, 'column_ids': column_ids},
    })


@admin_api.route('/user/<int:uid>/column-permissions')
@admin_required
def get_user_column_permissions(uid):
    _, _, denied = _ensure_admin_menu_access('users')
    if denied:
        return denied
    menu_type = (request.args.get('menu_type') or request.args.get('menuType') or '').strip()
    items, error_response, status = _java_permission_items(
        menu_type=menu_type or None,
        user_id=uid,
        effective=True,
    )
    if error_response is not None:
        return error_response, status
    return jsonify({'success': True, 'items': items, 'data': items})


def _validate_admin_granted_columns(role, current_row, column_ids):
    """普通管理员只能把自己已拥有的栏目授予其直接管理的普通用户。"""
    if role != 'admin' or not current_row or not column_ids:
        return
    requested_ids = set(_normalize_column_ids(column_ids))
    allowed_ids, _, _ = _effective_permission_sets(
        _get_current_admin_id(current_row),
        menu_type=None,
        current_row=current_row,
        role=role,
    )
    denied_ids = requested_ids - allowed_ids
    if denied_ids:
        raise PermissionError('普通管理员只能分配自己的有效菜单权限')


# ---------- 版本管理（web_config） ----------

# ---------- 商品类目 ----------


def _set_user_column_permissions_via_java(user_id, column_ids, *, role=None, current_row=None):
    """Persist direct grants in Java; never mirror them in Flask's legacy tables."""
    _, error_response, status = _java_update_user_direct_column_ids(
        user_id,
        column_ids,
        current_row=current_row,
        role=role,
    )
    if error_response is not None:
        raise _PermissionProxyError(error_response, status)


def _format_product_category_item(item, include_children=True):
    formatted = {
        'id': item.get('id'),
        'parent_id': item.get('parentId'),
        'name': item.get('name') or '',
        'category_key': item.get('categoryKey') or '',
        'sort_order': item.get('sortOrder') if item.get('sortOrder') is not None else 0,
        'description': item.get('description') or '',
        'is_builtin': bool(item.get('isBuiltin')),
        'child_count': item.get('childCount') if item.get('childCount') is not None else 0,
        'level': item.get('level') if item.get('level') is not None else 0,
        'path': item.get('path') or '',
        'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
        'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
    }
    if include_children:
        formatted['children'] = [_format_product_category_item(child) for child in (item.get('children') or [])]
    return formatted


def _build_product_category_payload_for_java(data):
    sort_order = data.get('sort_order') if 'sort_order' in data else data.get('sortOrder')
    parent_id = data.get('parent_id') if 'parent_id' in data else data.get('parentId')
    if parent_id == '':
        parent_id = None
    if sort_order == '':
        sort_order = None
    return {
        'parentId': parent_id,
        'name': (data.get('name') or '').strip(),
        'sortOrder': sort_order,
        'description': (data.get('description') or '').strip(),
    }


@admin_api.route('/product-categories')
@login_required
def list_product_categories():
    _, _, denied = _ensure_product_category_access()
    if denied:
        return denied
    keyword = (request.args.get('keyword') or '').strip()
    query = f'?keyword={quote(keyword)}' if keyword else ''
    result, error_response, status = _proxy_backend_java('GET', f'/api/admin/product-categories{query}')
    if error_response is not None:
        return error_response, status
    payload = result.get('data') or {}
    return jsonify({
        'success': True,
        'tree': [_format_product_category_item(item) for item in (payload.get('tree') or [])],
        'items': [_format_product_category_item(item, include_children=False) for item in (payload.get('items') or [])],
        'total': payload.get('total') or 0,
        'page': payload.get('page') or 1,
        'page_size': payload.get('pageSize') or 0,
        'has_more': bool(payload.get('hasMore')),
    })


@admin_api.route('/product-categories/children')
@login_required
def list_product_category_children():
    _, _, denied = _ensure_product_category_access()
    if denied:
        return denied
    parent_id_raw = (request.args.get('parent_id') or request.args.get('parentId') or '').strip()
    page = max(1, int(request.args.get('page', 1)))
    page_size = min(100, max(1, int(request.args.get('page_size', request.args.get('pageSize', 20)))))
    params = {
        'page': page,
        'pageSize': page_size,
    }
    if parent_id_raw:
        params['parentId'] = parent_id_raw
    result, error_response, status = _proxy_backend_java(
        'GET',
        '/api/admin/product-categories/children',
        params=params,
    )
    if error_response is not None:
        return error_response, status
    payload = result.get('data') or {}
    return jsonify({
        'success': True,
        'items': [_format_product_category_item(item, include_children=False) for item in (payload.get('items') or [])],
        'total': payload.get('total') or 0,
        'page': payload.get('page') or page,
        'page_size': payload.get('pageSize') or page_size,
        'has_more': bool(payload.get('hasMore')),
    })


@admin_api.route('/product-categories/search')
@login_required
def search_product_categories():
    _, _, denied = _ensure_product_category_access()
    if denied:
        return denied
    keyword = (request.args.get('keyword') or '').strip()
    page = max(1, int(request.args.get('page', 1)))
    page_size = min(100, max(1, int(request.args.get('page_size', request.args.get('pageSize', 20)))))
    result, error_response, status = _proxy_backend_java(
        'GET',
        '/api/admin/product-categories/search',
        params={
            'keyword': keyword,
            'page': page,
            'pageSize': page_size,
        },
    )
    if error_response is not None:
        return error_response, status
    payload = result.get('data') or {}
    return jsonify({
        'success': True,
        'items': [_format_product_category_item(item, include_children=False) for item in (payload.get('items') or [])],
        'total': payload.get('total') or 0,
        'page': payload.get('page') or page,
        'page_size': payload.get('pageSize') or page_size,
        'has_more': bool(payload.get('hasMore')),
    })


@admin_api.route('/product-categories/export')
@login_required
def export_product_categories():
    _, _, denied = _ensure_product_category_access()
    if denied:
        return denied
    keyword = (request.args.get('keyword') or '').strip()
    params = {}
    if keyword:
        params['keyword'] = keyword

    url = f"{backend_java_base_url}/api/admin/product-categories/export"
    try:
        resp = _get_backend_java_session().get(url, params=params, timeout=60)
    except requests.RequestException:
        return jsonify({'success': False, 'error': 'backend-java 服务不可用'}), 502
    if resp.status_code >= 400:
        try:
            data = resp.json()
            error = data.get('message') or data.get('error') or '导出失败'
        except ValueError:
            error = '导出失败'
        return jsonify({'success': False, 'error': error}), resp.status_code

    headers = {}
    disposition = resp.headers.get('Content-Disposition')
    if disposition:
        headers['Content-Disposition'] = disposition
    return Response(
        resp.content,
        status=resp.status_code,
        headers=headers,
        content_type=resp.headers.get(
            'Content-Type',
            'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        ),
    )


@admin_api.route('/product-category', methods=['POST'])
@login_required
def create_product_category():
    _, _, denied = _ensure_product_category_access()
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/product-category',
        json_data=_build_product_category_payload_for_java(request.get_json() or {}),
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'msg': result.get('message') or '创建成功',
        'item': _format_product_category_item(result.get('data') or {}, include_children=False),
    })


@admin_api.route('/product-category/<int:item_id>', methods=['PUT'])
@login_required
def update_product_category(item_id):
    _, _, denied = _ensure_product_category_access()
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'PUT',
        f'/api/admin/product-category/{item_id}',
        json_data=_build_product_category_payload_for_java(request.get_json() or {}),
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'msg': result.get('message') or '保存成功',
        'item': _format_product_category_item(result.get('data') or {}, include_children=False),
    })


@admin_api.route('/product-category/<int:item_id>', methods=['DELETE'])
@login_required
def delete_product_category(item_id):
    _, _, denied = _ensure_product_category_access()
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java('DELETE', f'/api/admin/product-category/{item_id}')
    if error_response is not None:
        return error_response, status
    return jsonify({'success': True, 'msg': result.get('message') or '删除成功'})


@admin_api.route('/versions')
@admin_required
def list_versions():
    _, _, denied = _ensure_admin_menu_access('version')
    if denied:
        return denied
    """获取版本列表"""
    try:
        conn = get_db()
        with conn.cursor() as cur:
            cur.execute(
                "SELECT id, version, file_url, created_at FROM web_config ORDER BY created_at DESC"
            )
            rows = cur.fetchall()
        conn.close()
        items = [
            {
                'id': r['id'],
                'version': r['version'] or '',
                'file_url': r['file_url'] or '',
                'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r.get('created_at') else '',
            }
            for r in rows
        ]
        return jsonify({'success': True, 'items': items})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


# ========== 数字人版本管理（代理到 Java 后端）==========

@admin_api.route('/digital-human-versions')
@admin_required
def list_digital_human_versions():
    """代理：查询数字人版本列表"""
    _, _, denied = _ensure_admin_menu_access('digital-human-version')
    if denied:
        return denied
    page = request.args.get('page', '1')
    page_size = request.args.get('pageSize', '20')
    status = request.args.get('status', '')
    params = {'page': page, 'pageSize': page_size}
    if status:
        params['status'] = status
    result, error_response, status_code = _proxy_backend_java(
        'GET',
        '/api/digital-human/versions',
        params=params,
        timeout=30
    )
    if error_response:
        return error_response, status_code
    return jsonify(result)


@admin_api.route('/digital-human-versions/latest')
@admin_required
def get_latest_digital_human_version():
    """代理：获取最新数字人版本"""
    _, _, denied = _ensure_admin_menu_access('digital-human-version')
    if denied:
        return denied
    result, error_response, status_code = _proxy_backend_java(
        'GET',
        '/api/digital-human/versions/latest',
        timeout=10
    )
    if error_response:
        return error_response, status_code
    return jsonify(result)


@admin_api.route('/digital-human-versions/<version>/download-url')
@admin_required
def get_digital_human_version_download_url(version):
    """代理：获取数字人版本下载链接"""
    _, _, denied = _ensure_admin_menu_access('digital-human-version')
    if denied:
        return denied
    safe_version = quote((version or '').strip(), safe='')
    if not safe_version:
        return jsonify({'success': False, 'message': '请指定版本号'}), 400
    result, error_response, status_code = _proxy_backend_java(
        'GET',
        f'/api/digital-human/versions/{safe_version}/download-url',
        timeout=10
    )
    if error_response:
        return error_response, status_code
    return jsonify(result)


@admin_api.route('/digital-human-versions/upload', methods=['POST'])
@admin_required
def upload_digital_human_version():
    """代理：上传数字人版本"""
    _, _, denied = _ensure_admin_menu_access('digital-human-version')
    if denied:
        return denied

    version = request.form.get('version', '').strip()
    changelog = request.form.get('changelog', '').strip()
    min_client_version = request.form.get('minClientVersion', '').strip()
    file_storage = request.files.get('file')

    if not version:
        return jsonify({'success': False, 'message': '请填写版本号'}), 400
    if not file_storage:
        return jsonify({'success': False, 'message': '请选择文件'}), 400

    files = {'file': (file_storage.filename, file_storage.stream, file_storage.content_type)}
    data = {'version': version}
    if changelog:
        data['changelog'] = changelog
    if min_client_version:
        data['minClientVersion'] = min_client_version

    result, error_response, status_code = _proxy_backend_java(
        'POST',
        '/api/digital-human/versions/upload',
        files=files,
        data=data,
        timeout=1800
    )
    if error_response:
        return error_response, status_code
    return jsonify(result)


@admin_api.route('/digital-human-versions/<version>/release', methods=['POST'])
@admin_required
def release_digital_human_version(version):
    """代理：发布数字人版本"""
    _, _, denied = _ensure_admin_menu_access('digital-human-version')
    if denied:
        return denied
    result, error_response, status_code = _proxy_backend_java(
        'POST',
        f'/api/digital-human/versions/{version}/release',
        timeout=10
    )
    if error_response:
        return error_response, status_code
    return jsonify(result)


@admin_api.route('/digital-human-versions/<version>/set-latest', methods=['POST'])
@admin_required
def set_latest_digital_human_version(version):
    """代理：设为最新数字人版本"""
    _, _, denied = _ensure_admin_menu_access('digital-human-version')
    if denied:
        return denied
    result, error_response, status_code = _proxy_backend_java(
        'POST',
        f'/api/digital-human/versions/{version}/set-latest',
        timeout=10
    )
    if error_response:
        return error_response, status_code
    return jsonify(result)


@admin_api.route('/digital-human-versions/<version>', methods=['DELETE'])
@admin_required
def delete_digital_human_version(version):
    """代理：删除数字人版本"""
    _, _, denied = _ensure_admin_menu_access('digital-human-version')
    if denied:
        return denied
    result, error_response, status_code = _proxy_backend_java(
        'DELETE',
        f'/api/digital-human/versions/{version}',
        timeout=30
    )
    if error_response:
        return error_response, status_code
    return jsonify(result)


@admin_api.route('/version', methods=['POST'])
@admin_required
def upload_version():
    """接收版本号 + zip 文件，上传到 OSS，写入 web_config"""
    _, _, denied = _ensure_admin_menu_access('version')
    if denied:
        return denied
    version = (request.form.get('version') or '').strip()
    if not version:
        return jsonify({'success': False, 'error': '请填写版本号'})
    file_storage = request.files.get('file')
    if not file_storage or file_storage.filename == '':
        return jsonify({'success': False, 'error': '请选择要上传的 zip 压缩包'})
    if not (file_storage.filename or '').lower().endswith('.zip'):
        return jsonify({'success': False, 'error': '仅支持 .zip 格式'})
    try:
        file_content = file_storage.read()
        if not file_content:
            return jsonify({'success': False, 'error': '文件为空'})
        safe_key = _safe_version_key(version)
        key = f"{bucket_path}versions/{safe_key}.zip"
        file_url = oss_upload_file(file_content, key)
        conn = get_db()
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO web_config (version, file_url) VALUES (%s, %s)",
                (version, file_url)
            )
        conn.commit()
        conn.close()
        return jsonify({
            'success': True,
            'version': version,
            'file_url': file_url,
            'msg': '上传成功',
        })
    except Exception as e:
        traceback.print_exc()
        return jsonify({'success': False, 'error': str(e)})


# ---------- 店铺密钥管理 ----------

@admin_api.route('/shop-keys')
@admin_required
def list_shop_keys():
    _, _, denied = _ensure_admin_menu_access('shop-keys')
    if denied:
        return denied
    page = max(1, int(request.args.get('page', 1)))
    page_size = min(100, max(1, int(request.args.get('page_size', 15))))
    result, error_response, status = _proxy_backend_java(
        'GET',
        '/api/admin/shop-keys',
        params={'page': page, 'pageSize': page_size},
    )
    if error_response is not None:
        return error_response, status
    payload = result.get('data') or {}
    items = [
        {
            'id': item.get('id'),
            'remark_name': item.get('remarkName') or '',
            'ziniao_account_name': item.get('ziniaoAccountName') or '',
            'ziniao_token': item.get('ziniaoToken') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        }
        for item in (payload.get('items') or [])
    ]
    return jsonify({
        'success': True,
        'items': items,
        'total': payload.get('total') or 0,
        'page': payload.get('page') or page,
        'page_size': payload.get('pageSize') or page_size,
    })


@admin_api.route('/shop-key', methods=['POST'])
@admin_required
def create_shop_key():
    _, _, denied = _ensure_admin_menu_access('shop-keys')
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {
        'remarkName': (data.get('remark_name') or '').strip(),
        'ziniaoAccountName': (data.get('ziniao_account_name') or '').strip(),
        'ziniaoToken': (data.get('ziniao_token') or '').strip(),
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/shop-keys',
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '创建成功',
        'item': {
            'id': item.get('id'),
            'remark_name': item.get('remarkName') or '',
            'ziniao_account_name': item.get('ziniaoAccountName') or '',
            'ziniao_token': item.get('ziniaoToken') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/shop-key/<int:item_id>', methods=['PUT'])
@admin_required
def update_shop_key(item_id):
    _, _, denied = _ensure_admin_menu_access('shop-keys')
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {
        'remarkName': (data.get('remark_name') or '').strip(),
        'ziniaoAccountName': (data.get('ziniao_account_name') or '').strip(),
        'ziniaoToken': (data.get('ziniao_token') or '').strip(),
    }
    result, error_response, status = _proxy_backend_java(
        'PUT',
        f'/api/admin/shop-keys/{item_id}',
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '更新成功',
        'item': {
            'id': item.get('id'),
            'remark_name': item.get('remarkName') or '',
            'ziniao_account_name': item.get('ziniaoAccountName') or '',
            'ziniao_token': item.get('ziniaoToken') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/shop-key/<int:item_id>', methods=['DELETE'])
@admin_required
def delete_shop_key(item_id):
    _, _, denied = _ensure_admin_menu_access('shop-keys')
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'DELETE',
        f'/api/admin/shop-keys/{item_id}',
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'msg': result.get('message') or '删除成功',
    })


# ---------- 数据去重总数据 ----------

@admin_api.route('/dedupe-total-data')
@login_required
def list_dedupe_total_data():
    _, current_row, denied = _ensure_dedupe_total_data_access()
    if denied:
        return denied
    page = max(1, int(request.args.get('page', 1)))
    page_size = min(100, max(1, int(request.args.get('page_size', 15))))
    keyword = (request.args.get('keyword') or '').strip()
    username = (request.args.get('username') or '').strip()
    data, error_response, status = _proxy_backend_java(
        'GET',
        '/api/admin/dedupe-total-data',
        params={
            'page': page,
            'pageSize': page_size,
            'keyword': keyword,
            'username': username,
            'operatorId': current_row.get('id'),
        },
    )
    if error_response is not None:
        return error_response, status
    payload = data.get('data') or {}
    items = [
        {
            'id': item.get('id'),
            'data_value': item.get('dataValue') or '',
            'uploader_user_id': item.get('uploaderUserId'),
            'username': item.get('username') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
        }
        for item in (payload.get('items') or [])
    ]
    return jsonify({
        'success': True,
        'items': items,
        'total': payload.get('total') or 0,
        'page': payload.get('page') or page,
        'page_size': payload.get('pageSize') or page_size,
    })


@admin_api.route('/dedupe-total-data/export')
@login_required
def export_dedupe_total_data():
    _, current_row, denied = _ensure_dedupe_total_data_access()
    if denied:
        return denied
    params = {'operatorId': current_row.get('id')}
    username = (request.args.get('username') or '').strip()
    start_date = (request.args.get('start_date') or request.args.get('startDate') or '').strip()
    end_date = (request.args.get('end_date') or request.args.get('endDate') or '').strip()
    if username:
        params['username'] = username
    if start_date:
        params['startDate'] = start_date
    if end_date:
        params['endDate'] = end_date

    url = f"{backend_java_base_url}/api/admin/dedupe-total-data/export"
    try:
        resp = _get_backend_java_session().get(url, params=params, timeout=60)
    except requests.RequestException:
        return jsonify({'success': False, 'error': 'backend-java 服务不可用'}), 502
    if resp.status_code >= 400:
        try:
            data = resp.json()
            error = data.get('message') or data.get('error') or '导出失败'
        except ValueError:
            error = '导出失败'
        return jsonify({'success': False, 'error': error}), resp.status_code

    headers = {}
    disposition = resp.headers.get('Content-Disposition')
    if disposition:
        headers['Content-Disposition'] = disposition
    return Response(
        resp.content,
        status=resp.status_code,
        headers=headers,
        content_type=resp.headers.get(
            'Content-Type',
            'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        ),
    )


@admin_api.route('/dedupe-total-data/import/<import_id>')
@login_required
def dedupe_total_data_import_progress(import_id):
    _, current_row, denied = _ensure_dedupe_total_data_access()
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'GET',
        f'/api/admin/dedupe-total-data/import/{import_id}',
        params={'operatorId': current_row.get('id')},
    )
    if error_response is not None:
        return error_response, status
    progress = result.get('data') or {}
    return jsonify({
        'success': True,
        'progress': {
            'status': progress.get('status') or 'pending',
            'total_rows': progress.get('totalRows') or 0,
            'processed_rows': progress.get('processedRows') or 0,
            'asin_count': progress.get('asinCount') or 0,
            'inserted_count': progress.get('insertedCount') or 0,
            'skipped_count': progress.get('skippedCount') or 0,
            'error_message': progress.get('errorMessage') or '',
        },
    })


@admin_api.route('/dedupe-total-data/import', methods=['POST'])
@login_required
def import_dedupe_total_data():
    _, current_row, denied = _ensure_dedupe_total_data_access()
    if denied:
        return denied
    file_storage = request.files.get('file')
    if not file_storage or file_storage.filename == '':
        return jsonify({'success': False, 'error': '请选择 Excel 文件'})
    files = {
        'file': (file_storage.filename, file_storage.stream, file_storage.mimetype or 'application/octet-stream')
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/dedupe-total-data/import',
        files=files,
        data={'operatorId': current_row.get('id')},
    )
    if error_response is not None:
        return error_response, status
    summary = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '开始导入',
        'import_id': summary.get('importId') or '',
    })


@admin_api.route('/dedupe-total-data/delete-import/<import_id>')
@login_required
def dedupe_total_data_delete_import_progress(import_id):
    _, current_row, denied = _ensure_dedupe_total_data_access()
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'GET',
        f'/api/admin/dedupe-total-data/delete-import/{import_id}',
        params={'operatorId': current_row.get('id')},
    )
    if error_response is not None:
        return error_response, status
    progress = result.get('data') or {}
    return jsonify({
        'success': True,
        'progress': {
            'status': progress.get('status') or 'pending',
            'total_rows': progress.get('totalRows') or 0,
            'processed_rows': progress.get('processedRows') or 0,
            'asin_count': progress.get('asinCount') or 0,
            'deleted_count': progress.get('insertedCount') or 0,
            'skipped_count': progress.get('skippedCount') or 0,
            'error_message': progress.get('errorMessage') or '',
        },
    })


@admin_api.route('/dedupe-total-data/delete-import', methods=['POST'])
@login_required
def delete_import_dedupe_total_data():
    _, current_row, denied = _ensure_dedupe_total_data_access()
    if denied:
        return denied
    file_storage = request.files.get('file')
    if not file_storage or file_storage.filename == '':
        return jsonify({'success': False, 'error': '请选择 Excel 文件'})
    files = {
        'file': (file_storage.filename, file_storage.stream, file_storage.mimetype or 'application/octet-stream')
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/dedupe-total-data/delete-import',
        files=files,
        data={'operatorId': current_row.get('id')},
    )
    if error_response is not None:
        return error_response, status
    summary = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '开始删除',
        'import_id': summary.get('importId') or '',
    })


@admin_api.route('/dedupe-total-data', methods=['POST'])
@login_required
def create_dedupe_total_data():
    _, current_row, denied = _ensure_dedupe_total_data_access()
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {'dataValue': (data.get('data_value') or '').strip()}
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/dedupe-total-data',
        params={'operatorId': current_row.get('id')},
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '创建成功',
        'item': {
            'id': item.get('id'),
            'data_value': item.get('dataValue') or '',
            'uploader_user_id': item.get('uploaderUserId'),
            'username': item.get('username') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/dedupe-total-data/<int:item_id>', methods=['PUT'])
@login_required
def update_dedupe_total_data(item_id):
    _, current_row, denied = _ensure_dedupe_total_data_access()
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {'dataValue': (data.get('data_value') or '').strip()}
    result, error_response, status = _proxy_backend_java(
        'PUT',
        f'/api/admin/dedupe-total-data/{item_id}',
        params={'operatorId': current_row.get('id')},
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '更新成功',
        'item': {
            'id': item.get('id'),
            'data_value': item.get('dataValue') or '',
            'uploader_user_id': item.get('uploaderUserId'),
            'username': item.get('username') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/dedupe-total-data/<int:item_id>', methods=['DELETE'])
@login_required
def delete_dedupe_total_data(item_id):
    _, current_row, denied = _ensure_dedupe_total_data_access()
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'DELETE',
        f'/api/admin/dedupe-total-data/{item_id}',
        params={'operatorId': current_row.get('id')},
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'msg': result.get('message') or '删除成功',
    })


# ---------- 不符合 ASIN 数据 ----------


def _format_invalid_asin_data_item(item):
    return {
        'id': item.get('id'),
        'data_value': item.get('dataValue') or '',
        'brand': item.get('brand') or '',
        'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
    }


@admin_api.route('/invalid-asin-data')
@admin_required
def list_invalid_asin_data():
    _, _, denied = _ensure_admin_menu_access('invalid-asin-data')
    if denied:
        return denied
    page = max(1, int(request.args.get('page', 1)))
    page_size = min(100, max(1, int(request.args.get('page_size', 15))))
    keyword = (request.args.get('keyword') or '').strip()
    data, error_response, status = _proxy_backend_java(
        'GET',
        '/api/admin/invalid-asin-data',
        params={'page': page, 'pageSize': page_size, 'keyword': keyword},
    )
    if error_response is not None:
        return error_response, status
    payload = data.get('data') or {}
    items = [_format_invalid_asin_data_item(item) for item in (payload.get('items') or [])]
    return jsonify({
        'success': True,
        'items': items,
        'total': payload.get('total') or 0,
        'page': payload.get('page') or page,
        'page_size': payload.get('pageSize') or page_size,
    })


@admin_api.route('/invalid-asin-data', methods=['POST'])
@admin_required
def create_invalid_asin_data():
    _, _, denied = _ensure_admin_menu_access('invalid-asin-data')
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {
        'dataValue': (data.get('data_value') or '').strip(),
        'brand': (data.get('brand') or '').strip(),
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/invalid-asin-data',
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '创建成功',
        'item': _format_invalid_asin_data_item(item),
    })


@admin_api.route('/invalid-asin-data/<int:item_id>', methods=['PUT'])
@admin_required
def update_invalid_asin_data(item_id):
    _, _, denied = _ensure_admin_menu_access('invalid-asin-data')
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {
        'dataValue': (data.get('data_value') or '').strip(),
        'brand': (data.get('brand') or '').strip(),
    }
    result, error_response, status = _proxy_backend_java(
        'PUT',
        f'/api/admin/invalid-asin-data/{item_id}',
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '更新成功',
        'item': _format_invalid_asin_data_item(item),
    })


@admin_api.route('/invalid-asin-data/<int:item_id>', methods=['DELETE'])
@admin_required
def delete_invalid_asin_data(item_id):
    _, _, denied = _ensure_admin_menu_access('invalid-asin-data')
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'DELETE',
        f'/api/admin/invalid-asin-data/{item_id}',
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'msg': result.get('message') or '删除成功',
    })


# ---------- 店铺管理 ----------

def _format_shop_manage_item(item):
    return {
        'id': item.get('id'),
        'group_id': item.get('groupId'),
        'group_name': item.get('groupName') or '',
        'shop_name': item.get('shopName') or '',
        'mall_name': item.get('mallName') or '',
        'account': item.get('account') or '',
        'password': item.get('passwordMasked') or '',
        'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
        'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
    }


def _format_shop_manage_group_item(item):
    return {
        'id': item.get('id'),
        'group_name': item.get('groupName') or '',
        'leader_user_id': item.get('leaderUserId'),
        'leader_username': item.get('leaderUsername') or '',
        'member_user_ids': item.get('memberUserIds') or [],
        'member_usernames': item.get('memberUsernames') or [],
        'member_count': item.get('memberCount') or 0,
        'user_id': item.get('userId'),
        'username': item.get('username') or '',
        'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
        'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
    }


def _load_expanded_shop_manage_groups(role, current_row):
    params = {}
    if current_row and current_row.get('id'):
        params['operatorId'] = current_row.get('id')
    params['superAdmin'] = 'true' if role == 'super_admin' else 'false'
    result, error_response, status = _proxy_backend_java(
        'GET',
        '/api/admin/shop-manages/groups',
        params=params,
    )
    if error_response is not None:
        return None, error_response, status

    return result.get('data') or [], None, 200


@admin_api.route('/shop-manages')
@login_required
def list_shop_manages():
    role, current_row, denied = _ensure_backend_menu_access('shop-manage', 'skip-price-asin', 'query-asin')
    if denied:
        return denied
    page = max(1, int(request.args.get('page', 1)))
    page_size = min(100, max(1, int(request.args.get('page_size', 15))))
    group_id_raw = (request.args.get('group_id') or '').strip()
    shop_name = (request.args.get('shop_name') or '').strip()

    params = {'page': page, 'pageSize': page_size}
    if current_row and current_row.get('id'):
        params['operatorId'] = current_row.get('id')
    params['superAdmin'] = 'true' if role == 'super_admin' else 'false'
    if group_id_raw:
        try:
            group_id = int(group_id_raw)
            if group_id > 0:
                params['groupId'] = group_id
        except (TypeError, ValueError):
            pass
    if shop_name:
        params['shopName'] = shop_name

    result, error_response, status = _proxy_backend_java(
        'GET',
        '/api/admin/shop-manages',
        params=params,
    )
    if error_response is not None:
        return error_response, status
    payload = result.get('data') or {}
    return jsonify({
        'success': True,
        'items': [_format_shop_manage_item(item) for item in (payload.get('items') or [])],
        'total': payload.get('total') or 0,
        'page': payload.get('page') or page,
        'page_size': payload.get('pageSize') or page_size,
    })


@admin_api.route('/shop-manage', methods=['POST'])
@login_required
def create_shop_manage():
    role, current_row, denied = _ensure_backend_menu_access('shop-manage')
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {
        'groupId': data.get('group_id'),
        'shopName': (data.get('shop_name') or '').strip(),
        'mallName': (data.get('mall_name') or '').strip(),
        'account': (data.get('account') or '').strip(),
        'password': (data.get('password') or '').strip(),
        'createdById': current_row.get('id') if current_row else None,
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/shop-manages',
        params={
            'operatorId': current_row.get('id') if current_row else None,
            'superAdmin': 'true' if role == 'super_admin' else 'false',
        },
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '创建成功',
        'item': {
            'id': item.get('id'),
            'group_id': item.get('groupId'),
            'group_name': item.get('groupName') or '',
            'shop_name': item.get('shopName') or '',
            'mall_name': item.get('mallName') or '',
            'account': item.get('account') or '',
            'password': item.get('password') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/shop-manage/<int:item_id>', methods=['PUT'])
@login_required
def update_shop_manage(item_id):
    role, current_row, denied = _ensure_backend_menu_access('shop-manage')
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {
        'groupId': data.get('group_id'),
        'shopName': (data.get('shop_name') or '').strip(),
        'mallName': (data.get('mall_name') or '').strip(),
        'account': (data.get('account') or '').strip(),
        'password': (data.get('password') or '').strip(),
    }
    result, error_response, status = _proxy_backend_java(
        'PUT',
        f'/api/admin/shop-manages/{item_id}',
        params={
            'operatorId': current_row.get('id') if current_row else None,
            'superAdmin': 'true' if role == 'super_admin' else 'false',
        },
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '更新成功',
        'item': {
            'id': item.get('id'),
            'group_name': item.get('groupName') or '',
            'shop_name': item.get('shopName') or '',
            'mall_name': item.get('mallName') or '',
            'account': item.get('account') or '',
            'password': item.get('password') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/shop-manage/<int:item_id>', methods=['DELETE'])
@login_required
def delete_shop_manage(item_id):
    role, current_row, denied = _ensure_backend_menu_access('shop-manage')
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'DELETE',
        f'/api/admin/shop-manages/{item_id}',
        params={
            'operatorId': current_row.get('id') if current_row else None,
            'superAdmin': 'true' if role == 'super_admin' else 'false',
        },
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'msg': result.get('message') or '删除成功',
    })


@admin_api.route('/shop-manage-groups')
@login_required
def list_shop_manage_groups():
    role, current_row, denied = _ensure_backend_menu_access(
        'shop-manage', 'skip-price-asin', 'query-asin', 'dedupe-total-data'
    )
    if denied:
        return denied
    groups, error_response, status = _load_expanded_shop_manage_groups(role, current_row)
    if error_response is not None:
        return error_response, status
    return jsonify({'success': True, 'items': [_format_shop_manage_group_item(item) for item in (groups or [])]})


@admin_api.route('/shop-manage-group', methods=['POST'])
@login_required
def create_shop_manage_group():
    role, current_row, denied = _ensure_backend_menu_access(
        'shop-manage', 'skip-price-asin', 'query-asin', 'dedupe-total-data'
    )
    if denied:
        return denied
    data = request.get_json() or {}
    grant_menu_routes = data.get('grant_menu_routes') or []
    payload = {
        'groupName': (data.get('group_name') or '').strip(),
        'memberUserIds': data.get('member_user_ids') or [],
        'createdById': current_row.get('id') if current_row else None,
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/shop-manages/groups',
        params={
            'operatorId': current_row.get('id') if current_row else None,
            'superAdmin': 'true' if role == 'super_admin' else 'false',
        },
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    _grant_backend_menu_permissions(data.get('member_user_ids') or [], grant_menu_routes)
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '创建成功',
        'item': {
            'id': item.get('id'),
            'group_name': item.get('groupName') or '',
            'leader_user_id': item.get('leaderUserId'),
            'leader_username': item.get('leaderUsername') or '',
            'member_user_ids': item.get('memberUserIds') or [],
            'member_usernames': item.get('memberUsernames') or [],
            'member_count': item.get('memberCount') or 0,
            'user_id': item.get('userId'),
            'username': item.get('username') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/shop-manage-group/<int:item_id>', methods=['PUT'])
@login_required
def update_shop_manage_group(item_id):
    role, current_row, denied = _ensure_backend_menu_access(
        'shop-manage', 'skip-price-asin', 'query-asin', 'dedupe-total-data'
    )
    if denied:
        return denied
    data = request.get_json() or {}
    grant_menu_routes = data.get('grant_menu_routes') or []
    payload = {
        'groupName': (data.get('group_name') or '').strip(),
        'memberUserIds': data.get('member_user_ids') or [],
    }
    result, error_response, status = _proxy_backend_java(
        'PUT',
        f'/api/admin/shop-manages/groups/{item_id}',
        params={
            'operatorId': current_row.get('id') if current_row else None,
            'superAdmin': 'true' if role == 'super_admin' else 'false',
        },
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    _grant_backend_menu_permissions(data.get('member_user_ids') or [], grant_menu_routes)
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '更新成功',
        'item': {
            'id': item.get('id'),
            'group_name': item.get('groupName') or '',
            'leader_user_id': item.get('leaderUserId'),
            'leader_username': item.get('leaderUsername') or '',
            'member_user_ids': item.get('memberUserIds') or [],
            'member_usernames': item.get('memberUsernames') or [],
            'member_count': item.get('memberCount') or 0,
            'user_id': item.get('userId'),
            'username': item.get('username') or '',
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/skip-price-asins')
@login_required
def list_skip_price_asins():
    role, current_row, denied = _ensure_backend_menu_access('skip-price-asin')
    if denied:
        return denied
    page = max(1, int(request.args.get('page', 1)))
    page_size = min(100, max(1, int(request.args.get('page_size', 15))))
    group_id_raw = (request.args.get('group_id') or '').strip()
    shop_name = (request.args.get('shop_name') or '').strip()
    asin = (request.args.get('asin') or '').strip()
    params = {'page': page, 'page_size': page_size, 'pageSize': page_size}
    params.update(_skip_price_asin_auth_params(role, current_row))
    if group_id_raw:
        try:
            group_id = int(group_id_raw)
            if group_id > 0:
                params['group_id'] = group_id
                params['groupId'] = group_id
        except (TypeError, ValueError):
            pass
    if shop_name:
        params['shop_name'] = shop_name
        params['shopName'] = shop_name
    if asin:
        params['asin'] = asin
    result, error_response, status = _proxy_backend_java(
        'GET',
        '/api/admin/skip-price-asins',
        params=params,
    )
    if error_response is not None:
        return error_response, status
    payload = result.get('data') or {}
    items = [
        {
            'id': item.get('id'),
            'group_id': item.get('groupId'),
            'group_name': item.get('groupName') or '',
            'shop_name': item.get('shopName') or '',
            'asin_de': item.get('asinDe') or '',
            'minimum_price_de': item.get('minimumPriceDe'),
            'asin_uk': item.get('asinUk') or '',
            'minimum_price_uk': item.get('minimumPriceUk'),
            'asin_fr': item.get('asinFr') or '',
            'minimum_price_fr': item.get('minimumPriceFr'),
            'asin_it': item.get('asinIt') or '',
            'minimum_price_it': item.get('minimumPriceIt'),
            'asin_es': item.get('asinEs') or '',
            'minimum_price_es': item.get('minimumPriceEs'),
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        }
        for item in (payload.get('items') or [])
    ]
    return jsonify({
        'success': True,
        'items': items,
        'total': payload.get('total') or 0,
        'page': payload.get('page') or page,
        'page_size': payload.get('pageSize') or page_size,
    })


def _skip_price_asin_auth_params(role, current_row):
    operator_id = current_row.get('id') if current_row else None
    super_admin = 'true' if role == 'super_admin' else 'false'
    return {
        'operator_id': operator_id,
        'operatorId': operator_id,
        'super_admin': super_admin,
        'superAdmin': super_admin,
    }


@admin_api.route('/skip-price-asins/export')
@login_required
def export_skip_price_asins():
    role, current_row, denied = _ensure_backend_menu_access('skip-price-asin')
    if denied:
        return denied
    params = _skip_price_asin_auth_params(role, current_row)
    group_id_raw = (request.args.get('group_id') or request.args.get('groupId') or '').strip()
    shop_name = (request.args.get('shop_name') or request.args.get('shopName') or '').strip()
    asin = (request.args.get('asin') or '').strip()
    if group_id_raw:
        params['group_id'] = group_id_raw
        params['groupId'] = group_id_raw
    if shop_name:
        params['shop_name'] = shop_name
        params['shopName'] = shop_name
    if asin:
        params['asin'] = asin

    url = f"{backend_java_base_url}/api/admin/skip-price-asins/export"
    try:
        resp = _get_backend_java_session().get(url, params=params, timeout=60)
    except requests.RequestException:
        return jsonify({'success': False, 'error': 'backend-java 服务不可用'}), 502
    if resp.status_code >= 400:
        try:
            data = resp.json()
            error = data.get('message') or data.get('error') or '导出失败'
        except ValueError:
            error = '导出失败'
        return jsonify({'success': False, 'error': error}), resp.status_code

    headers = {}
    disposition = resp.headers.get('Content-Disposition')
    if disposition:
        headers['Content-Disposition'] = disposition
    return Response(
        resp.content,
        status=resp.status_code,
        headers=headers,
        content_type=resp.headers.get(
            'Content-Type',
            'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        ),
    )


@admin_api.route('/skip-price-asin', methods=['POST'])
@login_required
def create_skip_price_asin():
    role, current_row, denied = _ensure_backend_menu_access('skip-price-asin')
    if denied:
        return denied
    data = request.get_json() or {}
    asin_mappings = data.get('asin_mappings') or {}
    minimum_price_mappings = data.get('minimum_price_mappings') or {}
    fallback_asin = (data.get('asin') or '').strip()
    if not fallback_asin and isinstance(asin_mappings, dict):
        for value in asin_mappings.values():
            fallback_asin = (value or '').strip()
            if fallback_asin:
                break
    fallback_minimum_price = data.get('minimum_price')
    if fallback_minimum_price in ('', None) and isinstance(minimum_price_mappings, dict):
        for value in minimum_price_mappings.values():
            if value not in ('', None):
                fallback_minimum_price = value
                break
    payload = {
        'groupId': data.get('group_id'),
        'shopName': (data.get('shop_name') or '').strip(),
        'countries': data.get('countries') or [],
        'asin': fallback_asin,
        'asinMappings': asin_mappings,
        'minimumPrice': fallback_minimum_price,
        'minimumPriceMappings': minimum_price_mappings,
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/skip-price-asins',
        params=_skip_price_asin_auth_params(role, current_row),
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '保存成功',
        'item': {
            'id': item.get('id'),
            'group_id': item.get('groupId'),
            'group_name': item.get('groupName') or '',
            'shop_name': item.get('shopName') or '',
            'asin_de': item.get('asinDe') or '',
            'minimum_price_de': item.get('minimumPriceDe'),
            'asin_uk': item.get('asinUk') or '',
            'minimum_price_uk': item.get('minimumPriceUk'),
            'asin_fr': item.get('asinFr') or '',
            'minimum_price_fr': item.get('minimumPriceFr'),
            'asin_it': item.get('asinIt') or '',
            'minimum_price_it': item.get('minimumPriceIt'),
            'asin_es': item.get('asinEs') or '',
            'minimum_price_es': item.get('minimumPriceEs'),
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/skip-price-asin/<int:item_id>/country/<country>', methods=['DELETE'])
@login_required
def delete_skip_price_asin_country(item_id, country):
    role, current_row, denied = _ensure_backend_menu_access('skip-price-asin')
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'DELETE',
        f'/api/admin/skip-price-asins/{item_id}/countries/{country}',
        params=_skip_price_asin_auth_params(role, current_row),
    )
    if error_response is not None:
        return error_response, status
    return jsonify({'success': True, 'msg': result.get('message') or '删除成功'})


@admin_api.route('/shop-manage-group/<int:item_id>', methods=['DELETE'])
@login_required
def delete_shop_manage_group(item_id):
    role, current_row, denied = _ensure_backend_menu_access(
        'shop-manage', 'skip-price-asin', 'query-asin', 'dedupe-total-data'
    )
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'DELETE',
        f'/api/admin/shop-manages/groups/{item_id}',
        params={
            'operatorId': current_row.get('id') if current_row else None,
            'superAdmin': 'true' if role == 'super_admin' else 'false',
        },
    )
    if error_response is not None:
        return error_response, status
    return jsonify({'success': True, 'msg': result.get('message') or '删除成功'})


@admin_api.route('/skip-price-asin/<int:item_id>/country/<country>', methods=['PUT'])
@login_required
def update_skip_price_asin_country(item_id, country):
    role, current_row, denied = _ensure_backend_menu_access('skip-price-asin')
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {
        'asin': (data.get('asin') or '').strip(),
        'minimumPrice': data.get('minimum_price'),
    }
    result, error_response, status = _proxy_backend_java(
        'PUT',
        f'/api/admin/skip-price-asins/{item_id}/countries/{country}',
        params=_skip_price_asin_auth_params(role, current_row),
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    item = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '保存成功',
        'item': {
            'id': item.get('id'),
            'group_id': item.get('groupId'),
            'group_name': item.get('groupName') or '',
            'shop_name': item.get('shopName') or '',
            'asin_de': item.get('asinDe') or '',
            'minimum_price_de': item.get('minimumPriceDe'),
            'asin_uk': item.get('asinUk') or '',
            'minimum_price_uk': item.get('minimumPriceUk'),
            'asin_fr': item.get('asinFr') or '',
            'minimum_price_fr': item.get('minimumPriceFr'),
            'asin_it': item.get('asinIt') or '',
            'minimum_price_it': item.get('minimumPriceIt'),
            'asin_es': item.get('asinEs') or '',
            'minimum_price_es': item.get('minimumPriceEs'),
            'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
            'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
        },
    })


@admin_api.route('/skip-price-asins/import/<import_id>')
@login_required
def skip_price_asin_import_progress(import_id):
    role, current_row, denied = _ensure_backend_menu_access('skip-price-asin')
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'GET',
        f'/api/admin/skip-price-asins/import/{import_id}',
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'progress': _format_query_asin_import_progress(result.get('data') or {}),
    })


@admin_api.route('/skip-price-asins/import', methods=['POST'])
@login_required
def import_skip_price_asins():
    role, current_row, denied = _ensure_backend_menu_access('skip-price-asin')
    if denied:
        return denied
    file_storage = request.files.get('file')
    if not file_storage or file_storage.filename == '':
        return jsonify({'success': False, 'error': '请选择 Excel 文件'})
    group_id_raw = (request.form.get('group_id') or '').strip()
    if not group_id_raw:
        return jsonify({'success': False, 'error': '请先选择分组'})
    params = _skip_price_asin_auth_params(role, current_row)
    params['group_id'] = group_id_raw
    params['groupId'] = group_id_raw
    files = {
        'file': (file_storage.filename, file_storage.stream, file_storage.mimetype or 'application/octet-stream')
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/skip-price-asins/import',
        params=params,
        files=files,
        timeout=60,
    )
    if error_response is not None:
        return error_response, status
    summary = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '开始导入',
        'import_id': summary.get('importId') or '',
    })


@admin_api.route('/skip-price-asins/delete-import/<import_id>')
@login_required
def skip_price_asin_delete_import_progress(import_id):
    role, current_row, denied = _ensure_backend_menu_access('skip-price-asin')
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'GET',
        f'/api/admin/skip-price-asins/delete-import/{import_id}',
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'progress': _format_query_asin_import_progress(result.get('data') or {}),
    })


@admin_api.route('/skip-price-asins/delete-import', methods=['POST'])
@login_required
def delete_import_skip_price_asins():
    role, current_row, denied = _ensure_backend_menu_access('skip-price-asin')
    if denied:
        return denied
    file_storage = request.files.get('file')
    if not file_storage or file_storage.filename == '':
        return jsonify({'success': False, 'error': '请选择 Excel 文件'})
    group_id_raw = (request.form.get('group_id') or '').strip()
    if not group_id_raw:
        return jsonify({'success': False, 'error': '请先选择分组'})
    params = _skip_price_asin_auth_params(role, current_row)
    params['group_id'] = group_id_raw
    params['groupId'] = group_id_raw
    files = {
        'file': (file_storage.filename, file_storage.stream, file_storage.mimetype or 'application/octet-stream')
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/skip-price-asins/delete-import',
        params=params,
        files=files,
        timeout=60,
    )
    if error_response is not None:
        return error_response, status
    summary = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '开始删除',
        'import_id': summary.get('importId') or '',
    })


def _format_query_asin_item(item):
    return {
        'id': item.get('id'),
        'group_id': item.get('groupId'),
        'group_name': item.get('groupName') or '',
        'shop_name': item.get('shopName') or '',
        'asin_de': item.get('asinDe') or '',
        'asin_uk': item.get('asinUk') or '',
        'asin_fr': item.get('asinFr') or '',
        'asin_it': item.get('asinIt') or '',
        'asin_es': item.get('asinEs') or '',
        'created_at': (item.get('createdAt') or '').replace('T', ' ')[:16],
        'updated_at': (item.get('updatedAt') or '').replace('T', ' ')[:16],
    }


@admin_api.route('/query-asins')
@login_required
def list_query_asins():
    role, current_row, denied = _ensure_backend_menu_access('query-asin')
    if denied:
        return denied
    page = max(1, int(request.args.get('page', 1)))
    page_size = min(100, max(1, int(request.args.get('page_size', 15))))
    group_id_raw = (request.args.get('group_id') or '').strip()
    shop_name = (request.args.get('shop_name') or '').strip()
    asin = (request.args.get('asin') or '').strip()
    params = {'page': page, 'pageSize': page_size}
    if current_row and current_row.get('id'):
        params['operatorId'] = current_row.get('id')
    params['superAdmin'] = 'true' if role == 'super_admin' else 'false'
    if group_id_raw:
        try:
            group_id = int(group_id_raw)
            if group_id > 0:
                params['groupId'] = group_id
        except (TypeError, ValueError):
            pass
    if shop_name:
        params['shopName'] = shop_name
    if asin:
        params['asin'] = asin
    result, error_response, status = _proxy_backend_java(
        'GET',
        '/api/admin/query-asins',
        params=params,
    )
    if error_response is not None:
        return error_response, status
    payload = result.get('data') or {}
    return jsonify({
        'success': True,
        'items': [_format_query_asin_item(item) for item in (payload.get('items') or [])],
        'total': payload.get('total') or 0,
        'page': payload.get('page') or page,
        'page_size': payload.get('pageSize') or page_size,
    })


@admin_api.route('/query-asins/export')
@login_required
def export_query_asins():
    role, current_row, denied = _ensure_backend_menu_access('query-asin')
    if denied:
        return denied
    params = {
        'operatorId': current_row.get('id') if current_row else None,
        'superAdmin': 'true' if role == 'super_admin' else 'false',
    }
    group_id_raw = (request.args.get('group_id') or request.args.get('groupId') or '').strip()
    shop_name = (request.args.get('shop_name') or request.args.get('shopName') or '').strip()
    asin = (request.args.get('asin') or '').strip()
    if group_id_raw:
        params['groupId'] = group_id_raw
    if shop_name:
        params['shopName'] = shop_name
    if asin:
        params['asin'] = asin

    url = f"{backend_java_base_url}/api/admin/query-asins/export"
    try:
        resp = _get_backend_java_session().get(url, params=params, timeout=60)
    except requests.RequestException:
        return jsonify({'success': False, 'error': 'backend-java 服务不可用'}), 502
    if resp.status_code >= 400:
        try:
            data = resp.json()
            error = data.get('message') or data.get('error') or '导出失败'
        except ValueError:
            error = '导出失败'
        return jsonify({'success': False, 'error': error}), resp.status_code

    headers = {}
    disposition = resp.headers.get('Content-Disposition')
    if disposition:
        headers['Content-Disposition'] = disposition
    return Response(
        resp.content,
        status=resp.status_code,
        headers=headers,
        content_type=resp.headers.get(
            'Content-Type',
            'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        ),
    )


def _format_query_asin_import_progress(progress):
    return {
        'status': progress.get('status') or 'pending',
        'total_rows': progress.get('totalRows') or 0,
        'processed_rows': progress.get('processedRows') or 0,
        'asin_count': progress.get('asinCount') or 0,
        'inserted_count': progress.get('insertedCount') or 0,
        'deleted_count': progress.get('deletedCount') or 0,
        'skipped_count': progress.get('skippedCount') or 0,
        'error_message': progress.get('errorMessage') or '',
    }


@admin_api.route('/query-asins/import/<import_id>')
@login_required
def query_asin_import_progress(import_id):
    role, current_row, denied = _ensure_backend_menu_access('query-asin')
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'GET',
        f'/api/admin/query-asins/import/{import_id}',
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'progress': _format_query_asin_import_progress(result.get('data') or {}),
    })


@admin_api.route('/query-asins/import', methods=['POST'])
@login_required
def import_query_asins():
    role, current_row, denied = _ensure_backend_menu_access('query-asin')
    if denied:
        return denied
    file_storage = request.files.get('file')
    if not file_storage or file_storage.filename == '':
        return jsonify({'success': False, 'error': '请选择 Excel 文件'})
    group_id_raw = (request.form.get('group_id') or '').strip()
    params = {
        'operatorId': current_row.get('id') if current_row else None,
        'superAdmin': 'true' if role == 'super_admin' else 'false',
    }
    if group_id_raw:
        params['groupId'] = group_id_raw
    files = {
        'file': (file_storage.filename, file_storage.stream, file_storage.mimetype or 'application/octet-stream')
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/query-asins/import',
        params=params,
        files=files,
        timeout=60,
    )
    if error_response is not None:
        return error_response, status
    summary = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '开始导入',
        'import_id': summary.get('importId') or '',
    })


@admin_api.route('/query-asins/delete-import/<import_id>')
@login_required
def query_asin_delete_import_progress(import_id):
    role, current_row, denied = _ensure_backend_menu_access('query-asin')
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'GET',
        f'/api/admin/query-asins/delete-import/{import_id}',
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'progress': _format_query_asin_import_progress(result.get('data') or {}),
    })


@admin_api.route('/query-asins/delete-import', methods=['POST'])
@login_required
def delete_import_query_asins():
    role, current_row, denied = _ensure_backend_menu_access('query-asin')
    if denied:
        return denied
    file_storage = request.files.get('file')
    if not file_storage or file_storage.filename == '':
        return jsonify({'success': False, 'error': '请选择 Excel 文件'})
    group_id_raw = (request.form.get('group_id') or '').strip()
    params = {
        'operatorId': current_row.get('id') if current_row else None,
        'superAdmin': 'true' if role == 'super_admin' else 'false',
    }
    if group_id_raw:
        params['groupId'] = group_id_raw
    files = {
        'file': (file_storage.filename, file_storage.stream, file_storage.mimetype or 'application/octet-stream')
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/query-asins/delete-import',
        params=params,
        files=files,
        timeout=60,
    )
    if error_response is not None:
        return error_response, status
    summary = result.get('data') or {}
    return jsonify({
        'success': True,
        'msg': result.get('message') or '开始删除',
        'import_id': summary.get('importId') or '',
    })


@admin_api.route('/query-asin', methods=['POST'])
@login_required
def create_query_asin():
    role, current_row, denied = _ensure_backend_menu_access('query-asin')
    if denied:
        return denied
    data = request.get_json() or {}
    asin_mappings = data.get('asin_mappings') or {}
    fallback_asin = (data.get('asin') or '').strip()
    if not fallback_asin and isinstance(asin_mappings, dict):
        for value in asin_mappings.values():
            fallback_asin = (value or '').strip()
            if fallback_asin:
                break
    payload = {
        'groupId': data.get('group_id'),
        'shopName': (data.get('shop_name') or '').strip(),
        'countries': data.get('countries') or [],
        'asin': fallback_asin,
        'asinMappings': asin_mappings,
    }
    result, error_response, status = _proxy_backend_java(
        'POST',
        '/api/admin/query-asins',
        params={
            'operatorId': current_row.get('id') if current_row else None,
            'superAdmin': 'true' if role == 'super_admin' else 'false',
        },
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'msg': result.get('message') or '保存成功',
        'item': _format_query_asin_item(result.get('data') or {}),
    })


@admin_api.route('/query-asin/<int:item_id>/country/<country>', methods=['DELETE'])
@login_required
def delete_query_asin_country(item_id, country):
    role, current_row, denied = _ensure_backend_menu_access('query-asin')
    if denied:
        return denied
    result, error_response, status = _proxy_backend_java(
        'DELETE',
        f'/api/admin/query-asins/{item_id}/countries/{country}',
        params={
            'operatorId': current_row.get('id') if current_row else None,
            'superAdmin': 'true' if role == 'super_admin' else 'false',
        },
    )
    if error_response is not None:
        return error_response, status
    return jsonify({'success': True, 'msg': result.get('message') or '删除成功'})


@admin_api.route('/query-asin/<int:item_id>/country/<country>', methods=['PUT'])
@login_required
def update_query_asin_country(item_id, country):
    role, current_row, denied = _ensure_backend_menu_access('query-asin')
    if denied:
        return denied
    data = request.get_json() or {}
    payload = {
        'asin': (data.get('asin') or '').strip(),
    }
    result, error_response, status = _proxy_backend_java(
        'PUT',
        f'/api/admin/query-asins/{item_id}/countries/{country}',
        params={
            'operatorId': current_row.get('id') if current_row else None,
            'superAdmin': 'true' if role == 'super_admin' else 'false',
        },
        json_data=payload,
    )
    if error_response is not None:
        return error_response, status
    return jsonify({
        'success': True,
        'msg': result.get('message') or '保存成功',
        'item': _format_query_asin_item(result.get('data') or {}),
    })