Coverage for netbox_data_import/flat_workbook.py: 99%
135 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-09 20:50 +0000
1# SPDX-License-Identifier: Apache-2.0
2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""Interpret flat workbooks without accessing the ORM."""
5from __future__ import annotations
7from typing import Any
9from dataclasses import dataclass, field
10from io import BytesIO
12import openpyxl
14from .adapters import SourceUnreadable
15from .catalog import CANDIDATE_TARGET_PREFIX
16from .transform_regex import TransformPattern, TransformPatternError
17from .values import comparison_key
19EXTRA_JSON_PREFIX = "extra_json:"
20_MAX_UNUSED_SAMPLES = 5
23@dataclass(frozen=True)
24class TransformRule:
25 """One regex that splits a source column into up to two Target Fields."""
27 source_column: str
28 pattern: str
29 group_1_target: str = ""
30 group_2_target: str = ""
33@dataclass(frozen=True)
34class FlatWorkbookConfig:
35 """Everything the flat adapter needs to interpret one workbook."""
37 sheet_name: str
38 column_map: dict[str, tuple[str, ...]] = field(default_factory=dict)
39 transform_rules: tuple[TransformRule, ...] = ()
40 capture_extra_data: bool = False
43@dataclass(frozen=True)
44class _CompiledTransformRule:
45 """One transform rule whose pattern is ready for every source row."""
47 source_column: str
48 pattern: TransformPattern
49 group_1_target: str
50 group_2_target: str
53def _compile_transform_rules(rules: tuple[TransformRule, ...]) -> tuple[_CompiledTransformRule, ...]:
54 """Compile all configured patterns once and name invalid source columns."""
55 compiled_rules = []
56 for rule in rules:
57 try:
58 pattern = TransformPattern.compile(rule.pattern)
59 except TransformPatternError as exc:
60 raise SourceUnreadable(
61 f"Invalid regex pattern '{rule.pattern}' in transform rule for column '{rule.source_column}': {exc}"
62 ) from exc
63 compiled_rules.append(
64 _CompiledTransformRule(
65 source_column=rule.source_column,
66 pattern=pattern,
67 group_1_target=rule.group_1_target,
68 group_2_target=rule.group_2_target,
69 )
70 )
71 return tuple(compiled_rules)
74def _text(value) -> str:
75 """Return the trimmed text of a cell value, empty for None."""
76 return "" if value is None else str(value).strip()
79def _header_index_map(sheet) -> dict[str, int]:
80 """Map each header name in row 1 to its column index; the first of a duplicate wins."""
81 headers: dict[str, int] = {}
82 for index, cell in enumerate(sheet[1]):
83 if cell.value is not None:
84 name = str(cell.value).strip()
85 if name not in headers:
86 headers[name] = index
87 return headers
90def _cell(raw_row, index) -> object:
91 """Return the raw cell at *index*, or None when the row is short."""
92 return raw_row[index] if index is not None and index < len(raw_row) else None
95def _merge_row_values(row_number: int, raw_row, headers: dict[str, int], column_map) -> dict[str, Any]:
96 """Build one row dict, recording a conflict when two source columns disagree."""
97 row: dict[str, Any] = {"_row_number": row_number}
98 for target_field, source_columns in column_map.items():
99 values: dict[str, Any] = {}
100 for source_column in source_columns:
101 value = _cell(raw_row, headers.get(source_column))
102 if isinstance(value, str):
103 value = value.strip()
104 if value is not None and str(value).strip():
105 values[source_column] = value
107 if not values:
108 continue
109 if target_field.startswith(CANDIDATE_TARGET_PREFIX):
110 candidate_target = target_field.removeprefix(CANDIDATE_TARGET_PREFIX)
111 row.setdefault("_candidate_values", {})[candidate_target] = {
112 source_column: str(value) for source_column, value in values.items()
113 }
114 continue
115 if len({comparison_key(target_field, value) for value in values.values()}) == 1:
116 row[target_field] = next(iter(values.values()))
117 else:
118 row[target_field] = None
119 row.setdefault("_conflicts", {})[target_field] = {
120 source_column: str(value) for source_column, value in values.items()
121 }
122 return row
125def promote_extra_json_fields(row: dict) -> None:
126 """Move every `extra_json:<name>` entry into the row's captured extra columns."""
127 for key in [key for key in list(row) if isinstance(key, str) and key.startswith(EXTRA_JSON_PREFIX)]:
128 value = row.pop(key)
129 if value not in (None, ""):
130 row.setdefault("_extra_columns", {})[key[len(EXTRA_JSON_PREFIX) :]] = value
133def _apply_transform_rules(
134 row: dict,
135 raw_row,
136 headers: dict[str, int],
137 rules: tuple[_CompiledTransformRule, ...],
138) -> None:
139 """Apply each safe transform rule in place."""
140 for rule in rules:
141 raw_value = _cell(raw_row, headers.get(rule.source_column))
142 if raw_value is None:
143 continue
144 text = str(raw_value).strip()
145 captures = rule.pattern.capture_groups(text)
146 if captures is None:
147 continue
148 if rule.group_1_target and len(captures) >= 1:
149 row[rule.group_1_target] = captures[0]
150 if rule.group_2_target and len(captures) >= 2:
151 row[rule.group_2_target] = captures[1]
154def _collect_unmapped_values(raw_row, headers, unmapped_columns, unused_stats, keep_stats, capture) -> dict[str, str]:
155 """Return this row's unmapped values and add them to *unused_stats* in place."""
156 extra: dict[str, str] = {}
157 for column in unmapped_columns:
158 text = _text(_cell(raw_row, headers[column]))
159 if not text:
160 continue
161 if capture:
162 extra[column] = text
163 if keep_stats:
164 entry = unused_stats.setdefault(column, {"count": 0, "samples": []})
165 entry["count"] += 1
166 if len(entry["samples"]) < _MAX_UNUSED_SAMPLES:
167 entry["samples"].append(text)
168 return extra
171def _open_sheet(content: bytes, sheet_name: str):
172 """Return the named worksheet, or say which sheets the file actually has."""
173 try:
174 book = openpyxl.load_workbook(BytesIO(content), data_only=True)
175 except Exception as exc:
176 raise SourceUnreadable(f"Cannot open Excel file: {exc}") from exc
177 if sheet_name not in book.sheetnames:
178 available = ", ".join(book.sheetnames)
179 raise SourceUnreadable(f"Sheet '{sheet_name}' not found. Available sheets: {available}")
180 return book[sheet_name]
183def interpret(content: bytes, config: FlatWorkbookConfig, *, collect_unused: bool = False):
184 """Return the rows one workbook carries, and the unmapped-column tally when asked for it."""
185 sheet = _open_sheet(content, config.sheet_name)
186 headers = _header_index_map(sheet)
187 mapped = {column for columns in config.column_map.values() for column in columns}
188 transformed = {rule.source_column for rule in config.transform_rules}
189 consumed_columns = mapped | transformed
190 unmapped_columns = [column for column in headers if column not in consumed_columns]
191 unused_stats: dict[str, dict] = {}
192 transform_rules = _compile_transform_rules(config.transform_rules)
194 rows = []
195 for row_number, raw_row in enumerate(sheet.iter_rows(min_row=2, values_only=True), start=2):
196 if all(value is None for value in raw_row):
197 continue
198 row = _merge_row_values(row_number, raw_row, headers, config.column_map)
199 _apply_transform_rules(row, raw_row, headers, transform_rules)
200 promote_extra_json_fields(row)
201 if collect_unused or config.capture_extra_data:
202 extra = _collect_unmapped_values(
203 raw_row, headers, unmapped_columns, unused_stats, collect_unused, config.capture_extra_data
204 )
205 if extra:
206 row.setdefault("_extra_columns", {}).update(extra)
207 rows.append(row)
209 return rows, unused_stats
212__all__ = (
213 "EXTRA_JSON_PREFIX",
214 "FlatWorkbookConfig",
215 "TransformRule",
216 "interpret",
217 "promote_extra_json_fields",
218)