baiyue-firstleg/scripts/client-xlsx-address-probe.py

"""Two bounded negative imports; preserve evidence and never retry unknown writes."""
import hashlib
import io
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.request
import uuid
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from xml.sax.saxutils import escape
from openpyxl import load_workbook


def require(ok, message):
    if not ok:
        raise AssertionError(message)


def options(content, title):
    book = load_workbook(io.BytesIO(content))
    try:
        sheet = book['订单导入']
        headers = {str(c.value).removesuffix(' *'): c.column for c in sheet[2] if c.value}
        require(len(headers) == 31 and title in headers, 'Unexpected template headers')
        col = headers[title]
        validations = [v for v in sheet.data_validations.dataValidation
                       if any(r.min_col <= col <= r.max_col and r.min_row <= 3 <= r.max_row for r in v.sqref.ranges)]
        if not validations:
            return []
        require(len(validations) == 1, 'Ambiguous option range')
        defined = book.defined_names[validations[0].formula1]
        return [c.value for name, span in defined.destinations for row in book[name][span] for c in row]
    finally:
        book.close()


def workbook(content, row):
    require(len(row) == 31, 'Expected 31 cells')
    output = io.BytesIO()
    with zipfile.ZipFile(io.BytesIO(content)) as source, zipfile.ZipFile(output, 'w', zipfile.ZIP_DEFLATED) as target:
        for item in source.infolist():
            data = source.read(item.filename)
            if item.filename == 'xl/worksheets/sheet1.xml':
                xml = data.decode()
                headers = re.findall(r'<row\b[^>]*\br="[12]"[^>]*>[\s\S]*?</row>', xml)
                require(len(headers) == 2, 'Both original header rows required')
                cells = []
                for index, value in enumerate(row):
                    col = chr(65 + index) if index < 26 else 'A' + chr(65 + index - 26)
                    cells.append(f'<c r="{col}3" t="inlineStr"><is><t xml:space="preserve">{escape(str(value))}</t></is></c>')
                xml, count = re.subn(r'<sheetData\b[^>]*>[\s\S]*?</sheetData>',
                                    '<sheetData>' + ''.join(headers) + '<row r="3">' + ''.join(cells) + '</row></sheetData>', xml)
                require(count == 1, 'One data sheet required')
                data = xml.encode()
            target.writestr(item, data)
    return output.getvalue()


def assert_rejection(body, title):
    require(str(body.get('code')) == '200', 'Import failed before row-level validation')
    data = body.get('data') or {}
    result = data.get('result') or {}
    require(data.get('mode') == 'PARTIAL' and data.get('sourceFileId') and data.get('errorReportFileId'), 'Missing import file evidence')
    require(result.get('total') == 1 and result.get('succeeded') == 0 and result.get('failed') == 1, 'Unexpected creation result')
    items = result.get('items') or []
    require(len(items) == 1, 'Expected exactly one error row')
    row = items[0]
    require(row.get('success') is False and row.get('clientRowId') == 'excel-row-3'
            and all(row.get(k) is None for k in ['orderId', 'orderNo', 'inboundCode'])
            and row.get('errorCode') == 'A0422'
            and row.get('errorMessage') == title + '不存在、已停用或不属于当前客户', 'Not the precise customer-address rejection')


SNAPSHOT_TABLES = ['ord_order', 'ord_order_cargo', 'ord_order_party', 'ord_order_internal_note',
                   'ord_order_status_history', 'ord_idempotency_record', 'ord_outbox_event']


def snapshot_sql(columns, identities):
    require(len(identities) == 2 and len(set(identities)) == 2
            and all(re.fullmatch(r'[1-9][0-9]{0,19}', x) for x in identities), 'Invalid customer scope')
    queries = ['SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;',
               'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;']
    for table in SNAPSHOT_TABLES:
        names = columns.get(table, [])
        require(names and len(names) == len(set(names)) and 'id' in names and 'tenant_id' in names
                and all(re.fullmatch(r'[a-z][a-z0-9_]*', n) for n in names), 'Incomplete snapshot schema')
        fields = ','.join('CAST(`' + n + '` AS CHAR)' for n in names)
        where = 'tenant_id IN (' + ','.join(identities) + ')'
        queries.append(f"SELECT JSON_OBJECT('table','{table}','hash',SHA2(CAST(JSON_ARRAY({fields}) AS CHAR),256)) FROM `{table}` WHERE {where} ORDER BY id;")
        queries.append(f"SELECT JSON_OBJECT('table','{table}','count',COUNT(*)) FROM `{table}` WHERE {where};")
    return '\n'.join(queries + ['COMMIT;'])


def snapshot_summary(rows):
    hashes = {t: [] for t in SNAPSHOT_TABLES}
    counts = {}
    for row in rows:
        table = row.get('table')
        require(table in hashes, 'Unexpected snapshot table')
        if 'count' in row:
            require(table not in counts and isinstance(row['count'], int) and row['count'] >= 0, 'Invalid snapshot count')
            counts[table] = row['count']
        else:
            require(re.fullmatch(r'[0-9a-f]{64}', row.get('hash', '')), 'Invalid row digest')
            hashes[table].append(row['hash'])
    require(len(counts) == len(hashes) and all(len(hashes[t]) == counts[t] for t in hashes), 'Truncated snapshot')
    return {t: {'count': counts[t], 'sha256': hashlib.sha256('\n'.join(hashes[t]).encode()).hexdigest()}
            for t in SNAPSHOT_TABLES}


def database_snapshot(config, identities):
    def query(sql):
        # SQL is sent via stdin; credentials are inherited only by this child process.
        result = subprocess.run(['mysql', '--protocol=TCP', '-h', config['host'], '-P', config['port'],
                                 '-u', config['username'], '--batch', '--raw', '--skip-column-names',
                                 '--default-character-set=utf8mb4', '--database=baiyue_order'],
                                input=sql, text=True, encoding='utf-8', capture_output=True, timeout=30,
                                env={**os.environ, 'MYSQL_PWD': config['password']},
                                creationflags=subprocess.CREATE_NO_WINDOW if os.name == 'nt' else 0)
        require(result.returncode == 0, 'Read-only TEST database query failed')
        return [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
    selected = ','.join("'" + t + "'" for t in SNAPSHOT_TABLES)
    metadata = query("SELECT JSON_OBJECT('table',TABLE_NAME,'column',COLUMN_NAME) FROM information_schema.COLUMNS "
                     "WHERE TABLE_SCHEMA='baiyue_order' AND TABLE_NAME IN (" + selected + ") ORDER BY TABLE_NAME,ORDINAL_POSITION;")
    columns = {}
    for row in metadata:
        columns.setdefault(row['table'], []).append(row['column'])
    return snapshot_summary(query(snapshot_sql(columns, identities)))


def main(config):
    require(config.get('environment') == 'TEST' and re.fullmatch(r'[A-Za-z0-9_-]{1,60}', config.get('runId', '')), 'Explicit TEST run required')
    directory = Path(__file__).resolve().parent.parent / 'reports' / ('client-xlsx-address-' + config['runId'])
    directory.mkdir()
    report = {'startedAt': datetime.now(timezone.utc).isoformat(), 'success': False, 'requests': [], 'checks': [], 'sessionsClosed': []}
    sessions = {}

    def save():
        (directory / 'report.json').write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8')

    class NoRedirect(urllib.request.HTTPRedirectHandler):
        def redirect_request(self, *args, **kwargs):
            return None
    opener = urllib.request.build_opener(NoRedirect)

    def request(name, method, path, body=None, binary=False, key=None):
        require(method == 'GET' or path in ['/auth/login', '/auth/logout', '/merchant/v1/orders/first-leg/batch-imports?mode=PARTIAL'], 'Unexpected write')
        entry = {'session': name, 'method': method, 'path': path, 'outcome': 'PENDING'}
        headers = {}
        if sessions.get(name, {}).get('accessToken'):
            headers['Authorization'] = 'Bearer ' + sessions[name]['accessToken']
        if isinstance(body, bytes):
            boundary = 'probe-' + uuid.uuid4().hex
            entry['workbookSha256'] = hashlib.sha256(body).hexdigest()
            body = (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="address-probe.xlsx"\r\nContent-Type: application/vnd.openxmlformats-officedocument.spreadsheetml.sheet\r\n\r\n'.encode()
                    + body + f'\r\n--{boundary}--\r\n'.encode())
            headers['Content-Type'] = 'multipart/form-data; boundary=' + boundary
            headers['Idempotency-Key'] = key
            entry['key'] = key
        elif body is not None:
            body = json.dumps(body).encode()
            headers['Content-Type'] = 'application/json'
        report['requests'].append(entry)
        save()
        try:
            response = opener.open(urllib.request.Request(config['base'] + path, body, headers, method=method), timeout=30)
            with response:
                data = response.read()
                entry.update(outcome='RECEIVED', http=response.status, requestId=response.headers.get('x-request-id'))
                if binary:
                    require('spreadsheetml' in response.headers.get('content-type', ''), 'Expected workbook content type')
                    save()
                    return data
                result = json.loads(data)
                entry['code'] = result.get('code')
                if key:
                    imported = result.get('data') or {}
                    rows = (imported.get('result') or {}).get('items') or []
                    entry.update(batchNo=imported.get('batchNo'), sourceFileId=imported.get('sourceFileId'),
                                 errorReportFileId=imported.get('errorReportFileId'),
                                 returnedOrderIds=[r['orderId'] for r in rows if r.get('orderId') is not None],
                                 rowErrorCodes=[r.get('errorCode') for r in rows])
                save()
                return result
        except Exception:
            entry['outcome'] = 'UNKNOWN_STOP' if entry['outcome'] == 'PENDING' else 'RECEIVED_INVALID'
            save()
            raise

    def success(name, method, path, body=None):
        result = request(name, method, path, body)
        require(str(result.get('code')) == '200', 'Setup/cleanup failed')
        return result['data']

    try:
        identities, templates = [], {}
        for name in ['source', 'control']:
            sessions[name] = success(name, 'POST', '/auth/login', config[name])
            require(sessions[name].get('accessToken') and sessions[name].get('refreshToken'), 'Incomplete session')
            profile = success(name, 'GET', '/merchant/v1/iam/tenant/profile')
            require(profile.get('id') is not None, 'Missing customer identity')
            identities.append(str(profile['id']))
            templates[name] = request(name, 'GET', '/merchant/v1/orders/first-leg/batch-imports/template', binary=True)
        require(identities[0] != identities[1], 'Distinct customers required')
        report['customerIdentitySha256'] = [hashlib.sha256(x.encode()).hexdigest() for x in identities]
        source, target = templates['source'], templates['control']
        common = {}
        for title in ['国内收货仓', '运输渠道', 'FBA目的仓']:
            shared = [x for x in options(source, title) if x in options(target, title)]
            require(shared, 'No shared valid master-data option')
            common[title] = shared[0]
        if config.get('database'):
            report['databaseBefore'] = database_snapshot(config['database'], identities)
            save()
        for title in ['取货地址', '私人目的仓地址']:
            foreign = [x for x in options(source, title) if x not in options(target, title)]
            require(foreign, 'No distinct foreign address option')
            row = ['海运', '', '自行快递', common['国内收货仓'], common['运输渠道'], '亚马逊FBA', common['FBA目的仓'], '', '', '',
                   '地址归属测试', '纸箱', '1', '1', '0.001', '塑料', '否', '否', 'ADDRESS-PROBE', '', '', '',
                   '买单报关', '包税', '散货拼箱', '否', '', '', '普通头程', '', config['runId']]
            if title == '取货地址':
                row[2], row[9], row[29] = '平台上门提货', '2026-10-10 10:00:00', foreign[0]
            else:
                row[5], row[6], row[8] = '私人地址', '', foreign[0]
            # Use the actual source template, including its foreign hidden option directory.
            payload = workbook(source, row)
            result = request('control', 'POST', '/merchant/v1/orders/first-leg/batch-imports?mode=PARTIAL', payload,
                             key='XADDR-' + config['runId'] + '-' + str(len(report['checks'])))
            assert_rejection(result, title)
            if config.get('database'):
                after = database_snapshot(config['database'], identities)
                report.setdefault('databaseAfterEachImport', []).append({'field': title, 'tables': after})
                save()
                require(after == report['databaseBefore'], 'Order-domain database changed after rejected XLSX import')
            report['checks'].append({'field': title, 'passed': True, 'foreignOptionSha256': hashlib.sha256(foreign[0].encode()).hexdigest()})
            save()
        report['success'] = True
        report['databaseVerified'] = bool(config.get('database'))
        report['limitation'] = ('Complete rows in seven Order tables for both customers, including logically deleted rows; '
                               'file/audit/auth writes are outside this unchanged-order-domain assertion; no browser claim'
                               if config.get('database') else 'XLSX parser rejects foreign customer template choices; source/error files retained; no SQL zero-drift or browser claim')
    except Exception as error:
        report['failure'] = str(error) if isinstance(error, AssertionError) else 'Transport/setup/format failure; no retry'
    finally:
        for name, session in sessions.items():
            if session.get('accessToken'):
                try:
                    success(name, 'POST', '/auth/logout', {'refreshToken': session['refreshToken']})
                    report['sessionsClosed'].append(name)
                except Exception:
                    report['success'] = False
        report['success'] = report['success'] and len(report['sessionsClosed']) == 2
        report['completedAt'] = datetime.now(timezone.utc).isoformat()
        save()
    print(json.dumps({'success': report['success'], 'checks': len(report['checks'])}))
    return report['success']


if __name__ == '__main__':
    sys.exit(0 if main(json.load(sys.stdin)) else 1)