"""Layout exporter service — orchestrator.
Wires queries, processing, and Excel writing into a single API.
"""
from __future__ import annotations
import zipfile
from io import BytesIO
from pathlib import Path
from typing import Optional
from sqlalchemy.orm import Session
from dpmcore.services.layout_exporter import processing, queries
from dpmcore.services.layout_exporter.excel_writer import ExcelLayoutWriter
from dpmcore.services.layout_exporter.models import (
DimensionMember,
ExportConfig,
ReleaseWindow,
TableLayout,
)
def _table_window(tv: object) -> ReleaseWindow:
"""Release window of a table version, used outside a module export."""
return ReleaseWindow(
start_release_id=getattr(tv, "start_release_id", None),
end_release_id=getattr(tv, "end_release_id", None),
)
[docs]
class LayoutExporterService:
"""Export annotated table layouts to Excel workbooks.
Args:
session: An open SQLAlchemy session.
"""
[docs]
def __init__(self, session: Session) -> None:
"""Bind the service to an open SQLAlchemy session."""
self.session = session
[docs]
def export_module(
self,
module_code: str,
release_code: Optional[str] = None,
output_path: Optional[str] = None,
config: Optional[ExportConfig] = None,
) -> Path:
"""Export all tables in a module to a single workbook.
Args:
module_code: Module version code (e.g., 'FINREP9').
release_code: Optional release code filter.
output_path: Output file path. Defaults to '{module_code}.xlsx'.
config: Export configuration flags.
Returns:
Path to the written file.
"""
table_versions = queries.load_module_table_versions(
self.session,
module_code,
release_code,
)
if not table_versions:
msg = f"No tables found for module '{module_code}'"
if release_code:
msg += f" at release '{release_code}'"
raise ValueError(msg)
# Codes are read in the module version's release window: a
# workbook must show the codes in force while that module
# version was reportable, not today's.
module_version = queries.load_module_version(
self.session,
module_code,
release_code,
)
window = ReleaseWindow(
start_release_id=getattr(module_version, "start_release_id", None),
end_release_id=getattr(module_version, "end_release_id", None),
)
layouts = []
for tv in table_versions:
layout = self._build_layout(tv, window, config)
if layout.rows or layout.columns:
layouts.append(layout)
out = Path(output_path or f"{module_code}.xlsx")
self._write_workbook(layouts, out, config)
return out
[docs]
def export_tables(
self,
table_codes: list[str],
release_code: Optional[str] = None,
output_path: Optional[str] = None,
config: Optional[ExportConfig] = None,
) -> Path:
"""Export specific tables to a workbook.
Args:
table_codes: List of table codes (e.g., ['F_01.01', 'F_01.02']).
release_code: Optional release code filter.
output_path: Output file path.
config: Export configuration flags.
Returns:
Path to the written file.
"""
layouts = []
for code in table_codes:
tv = queries.load_table_version(
self.session,
code,
release_code,
)
if tv is None:
continue
layout = self._build_layout(tv, _table_window(tv), config)
if layout.rows or layout.columns:
layouts.append(layout)
if not layouts:
raise ValueError(f"No valid tables found for codes: {table_codes}")
out = Path(output_path or "tables.xlsx")
self._write_workbook(layouts, out, config)
return out
[docs]
def build_layout(
self,
table_code: str,
release_code: Optional[str] = None,
config: Optional[ExportConfig] = None,
) -> TableLayout:
"""Build the intermediate layout for a single table.
Useful for programmatic access without Excel generation.
"""
tv = queries.load_table_version(
self.session,
table_code,
release_code,
)
if tv is None:
raise ValueError(f"Table '{table_code}' not found")
return self._build_layout(tv, _table_window(tv), config)
def _build_layout( # noqa: C901
self,
tv: object,
window: Optional[ReleaseWindow] = None,
config: Optional[ExportConfig] = None,
) -> TableLayout:
"""Core pipeline: query -> process -> TableLayout.
``window`` is the release window codes are read in — the module
version's when exporting a module, the table version's own
otherwise.
"""
if window is None:
window = _table_window(tv)
cfg = config or ExportConfig()
# Load all headers
raw_headers = queries.load_headers(self.session, tv.table_vid) # type: ignore[attr-defined]
# Collect context_ids, property_ids, and subcategory_vids.
# The table's own context is loaded with the headers': cells
# inherit it, and derived cells need it to be complete.
context_ids: set[int] = set()
table_context_id = getattr(tv, "context_id", None)
if table_context_id:
context_ids.add(table_context_id)
property_ids: set[int] = set()
subcategory_vids: set[int] = set()
for _tvh, _header, hv in raw_headers:
if hv.context_id:
context_ids.add(hv.context_id)
if hv.property_id:
property_ids.add(hv.property_id)
if hv.subcategory_vid:
subcategory_vids.add(hv.subcategory_vid)
# Batch load categorisations
context_cats = queries.load_categorisations(
self.session,
context_ids,
window,
)
property_cats = queries.load_property_as_categorisation(
self.session,
property_ids,
window,
)
subcategory_info = queries.load_subcategory_info(
self.session,
subcategory_vids,
)
# Build sorted headers
columns, rows, sheets = processing.build_layout_headers(
raw_headers,
context_cats,
property_cats,
subcategory_info,
)
all_headers = columns + rows + sheets
# Load cells
raw_cells = queries.load_cells(self.session, tv.table_vid) # type: ignore[attr-defined]
# Collect variable_vids for data point categorisations
variable_vids: set[int] = set()
for tvc, _cell in raw_cells:
if tvc.variable_vid:
variable_vids.add(tvc.variable_vid)
# Also collect key variable VIDs from key headers. Open rows
# put their key on a column, open sheets on the Z header.
for header in all_headers:
if header.key_variable_vid:
variable_vids.add(header.key_variable_vid)
dp_cats = queries.load_dp_categorisations(
self.session,
variable_vids,
window,
)
variable_info = queries.load_variable_info(
self.session,
variable_vids,
)
# Properties reached through the headers, for whatever has no
# variable to name them.
property_info = (
queries.load_property_info(self.session, property_ids)
if cfg.derive_missing_variables
else {}
)
# Populate key variable fields on key headers
for header in all_headers:
key_vid = header.key_variable_vid
if key_vid and key_vid in variable_info:
v_id, dtype, prop_name = variable_info[key_vid]
header.key_variable_id = v_id
header.key_data_type_code = dtype
header.key_property_name = prop_name
elif header.is_key and header.property_id in property_info:
# No key variable generated yet: the header names the
# property itself.
dtype, prop_name = property_info[header.property_id]
header.key_data_type_code = dtype
header.key_property_name = prop_name
# Build synthetic key dimension annotations for key headers.
# The key variable names the property; failing that, so does
# the header.
key_variable_vids: set[int] = {
h.key_variable_vid for h in all_headers if h.key_variable_vid
}
key_vid_prop_ids = (
queries.load_key_variable_property_ids(
self.session,
key_variable_vids,
)
if key_variable_vids
else {}
)
key_prop_by_header: dict[int, int] = {}
for header in all_headers:
if not header.is_key:
continue
prop_id = key_vid_prop_ids.get(header.key_variable_vid or 0)
if prop_id is None and cfg.derive_missing_variables:
prop_id = header.property_id
if prop_id:
key_prop_by_header[header.header_id] = prop_id
if key_prop_by_header:
key_prop_cats = queries.load_property_as_categorisation(
self.session,
set(key_prop_by_header.values()),
window,
)
for header in all_headers:
prop_id = key_prop_by_header.get(header.header_id)
atm_dm = key_prop_cats.get(prop_id) if prop_id else None
if atm_dm:
header.key_categorisations = [
DimensionMember(
property_id=atm_dm.property_id,
dimension_label=atm_dm.member_label,
dimension_code=atm_dm.member_code,
domain_code=atm_dm.domain_code,
member_label="",
member_code="",
),
]
# Build cell data
row_ids = {h.header_id for h in rows}
col_ids = {h.header_id for h in columns}
sheet_ids = {h.header_id for h in sheets}
# Add default sheet ID if no explicit sheets
if not sheet_ids:
for _tvc, cell in raw_cells:
if cell.sheet_id:
sheet_ids.add(cell.sheet_id)
cells = processing.build_cells(
raw_cells,
row_ids,
col_ids,
sheet_ids,
dp_cats,
variable_info,
)
# Cells whose variable has not been generated yet
if cfg.derive_missing_variables:
processing.derive_missing_cell_data(
cells,
all_headers,
property_info,
context_cats.get(table_context_id, [])
if table_context_id
else [],
)
# Possible values of enumerated cells and key headers
enum_subcat_vids = processing.resolve_enumeration_sources(
cells,
all_headers,
)
if enum_subcat_vids:
enumerations = queries.load_enumerations(
self.session,
enum_subcat_vids,
window,
)
processing.attach_enumerations(cells, all_headers, enumerations)
return processing.build_table_layout(tv, columns, rows, sheets, cells)
def _write_workbook(
self,
layouts: list[TableLayout],
output_path: Path,
config: Optional[ExportConfig] = None,
) -> None:
"""Write layouts to an Excel file."""
writer = ExcelLayoutWriter(layouts, config)
wb = writer.write()
output_path.parent.mkdir(parents=True, exist_ok=True)
wb.save(str(output_path))
_fix_xlsx_timestamps(output_path)
def _fix_xlsx_timestamps(path: Path) -> None:
"""Fix openpyxl bug that writes invalid ISO timestamps.
openpyxl serialises timestamps as e.g. ``2026-03-31T16:00:00+00:00Z``
which is invalid (either ``+00:00`` or ``Z``, not both).
Excel shows a repair dialog on open. This rewrites the offending XML
in-place.
"""
buf = BytesIO(path.read_bytes())
with zipfile.ZipFile(buf, "r") as zin:
names = zin.namelist()
if "docProps/core.xml" not in names:
return
core_xml = zin.read("docProps/core.xml").decode("utf-8")
if "+00:00Z" not in core_xml:
return
fixed_xml = core_xml.replace("+00:00Z", "Z")
out = BytesIO()
with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as zout:
for name in names:
data = (
fixed_xml.encode("utf-8")
if name == "docProps/core.xml"
else zin.read(name)
)
zout.writestr(name, data)
path.write_bytes(out.getvalue())