"""Extract only this workbook's local detail cells; never evaluate external links.""" import hashlib from pathlib import Path import posixpath import re from xml.etree import ElementTree as ET from zipfile import ZipFile from .quote_data import decimal_text NS = {'s':'http://schemas.openxmlformats.org/spreadsheetml/2006/main'} TRUCK_SHEETS = ('美森正班海卡','美森加班海卡','EXX海卡','合德海卡','纽约EXX海铁', '洛杉矶普船海卡','洛杉矶普船全卡','芝加哥普船海铁','休斯顿普船海卡','纽约普船海卡','萨凡纳普船海卡') def read_sheets(path): path = Path(path) if path.stat().st_size > 20*1024*1024: raise ValueError('Quote workbook is too large') with ZipFile(path) as archive: if sum(info.file_size for info in archive.infolist()) > 100*1024*1024: raise ValueError('Expanded quote workbook is too large') strings = [] if 'xl/sharedStrings.xml' in archive.namelist(): for entry in ET.fromstring(archive.read('xl/sharedStrings.xml')): strings.append(''.join(t.text or '' for t in entry.findall('.//s:t', NS))) workbook = ET.fromstring(archive.read('xl/workbook.xml')) rels = {r.attrib['Id']:r.attrib['Target'] for r in ET.fromstring(archive.read('xl/_rels/workbook.xml.rels'))} result = {} for sheet in workbook.find('s:sheets', NS): target = rels[sheet.attrib['{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id']] target = target.lstrip('/') if target.startswith('/') else posixpath.normpath('xl/'+target) rows = {} for cell in ET.fromstring(archive.read(target)).findall('s:sheetData/s:row/s:c', NS): ref = cell.attrib['r'] column = re.sub('[0-9]', '', ref) if len(column) != 1 or column > 'Q': continue raw = cell.find('s:v', NS) value = raw.text if raw is not None else '' if cell.attrib.get('t') == 's': value = strings[int(value)] elif cell.attrib.get('t') == 'inlineStr': value = ''.join(t.text or '' for t in cell.findall('.//s:t', NS)) formula = cell.find('s:f', NS) if formula is not None and formula.text and re.search(r'\[[0-9]+\]', formula.text): continue if value: rows.setdefault(int(re.sub('[^0-9]', '', ref)), {})[column] = value result[sheet.attrib['name']] = rows return result def export_snapshot(path): sheets = read_sheets(path) missing = set(TRUCK_SHEETS + ('海派','FBA分区表','亚马逊仓库代码表','UPS&FedEx偏远地区表')) - sheets.keys() if missing: raise ValueError('Quote workbook is missing required local detail sheets') result = {'schema':1, 'source_name':Path(path).name, 'source_sha256':hashlib.sha256(Path(path).read_bytes()).hexdigest(), 'rates':[], 'warehouses':[], 'postcodes':[], 'documents':[]} for name in TRUCK_SHEETS: rows = sheets[name] destination_col = 'A' if name in TRUCK_SHEETS[:4] else 'B' price_cols = ('B','C','D') if destination_col == 'A' else ('C','D','E') eta_col = 'E' if destination_col == 'A' else 'F' for number, row in rows.items(): if number < 6 or not re.fullmatch(r'\d+(?:\.\d+)?', row.get(price_cols[0], '')): continue destinations = [p.strip() for p in re.split(r'[/\n、]+', row.get(destination_col, '')) if p.strip()] if not destinations: raise ValueError('Price row is missing destinations') for column, origin, unit, tax in zip(price_cols, ('义乌','义乌','华南'), ('cbm','kg','kg'), ('excluded','included','included')): result['rates'].append({'service':name, 'kind':'truck', 'origin':origin, 'zone':'', 'unit':unit, 'tax':tax, 'min_weight':'0', 'price':decimal_text(row[column], positive=True), 'eta':row.get(eta_col, ''), 'destinations':destinations, 'source_sheet':name, 'source_cell':f'{column}{number}'}) parcel = sheets['海派'] for origin, header, first in (('义乌',4,9), ('华南',13,18)): for start_col in ('C','F','I','L'): for offset, tier in enumerate(('12','51','101')): column = chr(ord(start_col)+offset) for index, zone in enumerate(('美西','美中','美东')): number = first+index result['rates'].append({'service':parcel[header][start_col], 'kind':'parcel', 'origin':origin, 'zone':zone, 'unit':'kg', 'tax':'included', 'min_weight':tier, 'price':decimal_text(parcel[number][column], positive=True), 'eta':parcel[header+2][start_col], 'destinations':[], 'source_sheet':'海派', 'source_cell':f'{column}{number}'}) for name, first in (('FBA分区表',3), ('亚马逊仓库代码表',2)): for number, row in sheets[name].items(): if number < first or not row.get('A'): continue is_zone = name == 'FBA分区表' result['warehouses'].append({'code':row['A'].strip().upper(), 'zone':row.get('B','') if is_zone else '', 'state':row.get('C','') if is_zone else '', 'postal':row.get('D','') if is_zone else '', 'address':row.get('B','') if not is_zone else '', 'source_sheet':name, 'source_cell':f'A{number}'}) for number, row in sheets['UPS&FedEx偏远地区表'].items(): if number <= 23: continue for column, severity, fee in (('B','remote','1'), ('F','extended','2'), ('J','extreme',None)): postal = row.get(column, '') if postal: result['postcodes'].append({'postal':postal, 'severity':severity, 'fee':fee}) for name, rows in sheets.items(): if name in ('快速查询','UPS&FedEx偏远地区表','FBA分区表','亚马逊仓库代码表'): continue for number, row in rows.items(): if number <= 1 or (name in TRUCK_SHEETS and number <= 5): continue # Long detail text contains rules; short lookup rows contain service fees. for column, value in row.items(): if len(value) >= 20 and not re.search(r'公司地址|TEL:|船名航次|预计开船日期|关税查询网址', value): result['documents'].append({'sheet':name, 'cell':f'{column}{number}', 'text':value}) if name in ('一件代发收费','FDA申报清单'): text = ';'.join(row.values()) result['documents'].append({'sheet':name,'cell':f'A{number}','text':text}) return result