168 lines
5.3 KiB
Python
168 lines
5.3 KiB
Python
from pathlib import Path
|
|
import re
|
|
import zipfile
|
|
from xml.etree import ElementTree as ET
|
|
|
|
|
|
MAIN_NS = {"a": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
|
|
REL_NS = {"r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships"}
|
|
PACKAGE_REL_NS = {"pr": "http://schemas.openxmlformats.org/package/2006/relationships"}
|
|
|
|
|
|
def load_single_bid_market_matrix(excel_path, trading_market, sheet_name="Sheet1", excel_layout=None):
|
|
|
|
excel_file = Path(excel_path)
|
|
if not excel_file.exists():
|
|
raise FileNotFoundError(f"Excel input file not found: {excel_file}")
|
|
|
|
layout = excel_layout or {}
|
|
|
|
with zipfile.ZipFile(excel_file) as workbook_zip:
|
|
shared_strings = _load_shared_strings(workbook_zip)
|
|
sheet_xml_path = _resolve_sheet_xml_path(workbook_zip, sheet_name)
|
|
sheet_root = ET.fromstring(workbook_zip.read(sheet_xml_path))
|
|
cell_map = _load_cell_map(sheet_root, shared_strings)
|
|
|
|
start_column = layout.get("column_start", "B")
|
|
market_price_rows = layout.get("market_price_rows", {})
|
|
common_data_start_row = int(layout.get("common_data_start_row", 6))
|
|
price_row = int(market_price_rows.get(trading_market, 2))
|
|
_, start_col = _split_cell_reference(f"{start_column}1")
|
|
max_row, max_col = _max_used_extent(cell_map, price_row, start_col)
|
|
|
|
matrix = []
|
|
for row_index in [price_row, *range(common_data_start_row, max_row + 1)]:
|
|
row_values = []
|
|
for col_index in range(start_col, max_col + 1):
|
|
row_values.append(cell_map.get((row_index, col_index), ""))
|
|
matrix.append(row_values)
|
|
|
|
matrix = _trim_trailing_empty_rows_and_columns(matrix)
|
|
if not matrix or not matrix[0]:
|
|
raise ValueError(f"No bidding data found in {excel_file} for market {trading_market}.")
|
|
|
|
tsv = "\n".join("\t".join(value for value in row) for row in matrix)
|
|
return {
|
|
"matrix": matrix,
|
|
"tsv": tsv,
|
|
"row_count": len(matrix),
|
|
"column_count": len(matrix[0]),
|
|
"first_value": matrix[0][0],
|
|
}
|
|
|
|
|
|
def _load_shared_strings(workbook_zip):
|
|
|
|
try:
|
|
root = ET.fromstring(workbook_zip.read("xl/sharedStrings.xml"))
|
|
except KeyError:
|
|
return []
|
|
|
|
values = []
|
|
for item in root.findall("a:si", MAIN_NS):
|
|
text_parts = [node.text or "" for node in item.findall(".//a:t", MAIN_NS)]
|
|
values.append("".join(text_parts))
|
|
return values
|
|
|
|
|
|
def _resolve_sheet_xml_path(workbook_zip, sheet_name):
|
|
|
|
workbook_root = ET.fromstring(workbook_zip.read("xl/workbook.xml"))
|
|
relationships_root = ET.fromstring(workbook_zip.read("xl/_rels/workbook.xml.rels"))
|
|
|
|
relationship_map = {
|
|
rel.attrib["Id"]: rel.attrib["Target"]
|
|
for rel in relationships_root.findall("pr:Relationship", PACKAGE_REL_NS)
|
|
}
|
|
|
|
for sheet in workbook_root.findall("a:sheets/a:sheet", {**MAIN_NS, **REL_NS}):
|
|
if sheet.attrib.get("name") != sheet_name:
|
|
continue
|
|
|
|
relationship_id = sheet.attrib.get(f"{{{REL_NS['r']}}}id")
|
|
target = relationship_map[relationship_id]
|
|
target = target.lstrip("/")
|
|
if not target.startswith("xl/"):
|
|
target = f"xl/{target}"
|
|
return target
|
|
|
|
raise KeyError(f"Sheet '{sheet_name}' not found in workbook.")
|
|
|
|
|
|
def _load_cell_map(sheet_root, shared_strings):
|
|
|
|
cell_map = {}
|
|
|
|
for row in sheet_root.findall(".//a:sheetData/a:row", MAIN_NS):
|
|
for cell in row.findall("a:c", MAIN_NS):
|
|
reference = cell.attrib.get("r", "")
|
|
row_index, col_index = _split_cell_reference(reference)
|
|
cell_map[(row_index, col_index)] = _cell_value(cell, shared_strings)
|
|
|
|
return cell_map
|
|
|
|
|
|
def _cell_value(cell, shared_strings):
|
|
|
|
cell_type = cell.attrib.get("t")
|
|
|
|
if cell_type == "inlineStr":
|
|
text_nodes = cell.findall(".//a:t", MAIN_NS)
|
|
return "".join(node.text or "" for node in text_nodes)
|
|
|
|
value_node = cell.find("a:v", MAIN_NS)
|
|
if value_node is None or value_node.text is None:
|
|
return ""
|
|
|
|
raw_value = value_node.text
|
|
if cell_type == "s":
|
|
return shared_strings[int(raw_value)]
|
|
|
|
return raw_value
|
|
|
|
|
|
def _split_cell_reference(reference):
|
|
|
|
match = re.match(r"([A-Z]+)(\d+)", reference)
|
|
if not match:
|
|
raise ValueError(f"Unsupported cell reference: {reference}")
|
|
|
|
column_letters, row_number = match.groups()
|
|
return int(row_number), _column_letters_to_number(column_letters)
|
|
|
|
|
|
def _column_letters_to_number(column_letters):
|
|
|
|
result = 0
|
|
for character in column_letters:
|
|
result = result * 26 + (ord(character) - ord("A") + 1)
|
|
return result
|
|
|
|
|
|
def _max_used_extent(cell_map, start_row, start_col):
|
|
|
|
used_rows = [row for row, col in cell_map if row >= start_row and col >= start_col and cell_map[(row, col)] != ""]
|
|
used_cols = [col for row, col in cell_map if row >= start_row and col >= start_col and cell_map[(row, col)] != ""]
|
|
|
|
if not used_rows or not used_cols:
|
|
raise ValueError("No populated cells were found in the requested data area.")
|
|
|
|
return max(used_rows), max(used_cols)
|
|
|
|
|
|
def _trim_trailing_empty_rows_and_columns(matrix):
|
|
|
|
while matrix and all(value == "" for value in matrix[-1]):
|
|
matrix.pop()
|
|
|
|
if not matrix:
|
|
return matrix
|
|
|
|
while matrix and matrix[0]:
|
|
if any(row[-1] != "" for row in matrix):
|
|
break
|
|
for row in matrix:
|
|
row.pop()
|
|
|
|
return matrix
|