196 lines
10 KiB
Python
196 lines
10 KiB
Python
"""Two bounded negative imports; preserve evidence and never retry unknown writes."""
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import re
|
|
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')
|
|
|
|
|
|
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]
|
|
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)
|
|
report['checks'].append({'field': title, 'passed': True, 'foreignOptionSha256': hashlib.sha256(foreign[0].encode()).hexdigest()})
|
|
save()
|
|
report['success'] = True
|
|
report['limitation'] = '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)
|