-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathexcel_file.py
More file actions
313 lines (252 loc) · 10.3 KB
/
excel_file.py
File metadata and controls
313 lines (252 loc) · 10.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
"""
Excel file entity for reading and hashing Excel files.
Encapsulates file I/O and content-based hashing for incremental imports.
"""
from pathlib import Path
from typing import Optional, List, Dict, Literal
import pandas as pd
import hashlib
from excel_to_sql.exceptions import ExcelFileError
class ExcelFile:
"""
Represents an Excel file with reading and hashing capabilities.
Usage:
file = ExcelFile("path/to/file.xlsx")
df = file.read() # Read as DataFrame
hash_value = file.content_hash # Get SHA-256 of content
sheets = file.list_sheets() # List all sheet names
"""
def __init__(self, path: Path | str) -> None:
"""
Initialize Excel file entity.
Args:
path: Path to Excel file
"""
self._path = Path(path)
self._hash_cache: Optional[Dict[str, str]] = None
# ──────────────────────────────────────────────────────────────
# PROPERTIES
# ──────────────────────────────────────────────────────────────
@property
def path(self) -> Path:
"""File path."""
return self._path
@property
def name(self) -> str:
"""Filename with extension."""
return self._path.name
@property
def exists(self) -> bool:
"""Check if file exists."""
return self._path.exists()
@property
def sheet_names(self) -> List[str]:
"""Get list of sheet names."""
try:
excel_file = pd.ExcelFile(self._path, engine="openpyxl")
return excel_file.sheet_names
except Exception:
return []
# ──────────────────────────────────────────────────────────────
# PUBLIC METHODS
# ──────────────────────────────────────────────────────────────
def read(
self,
sheet_name: str | None = None,
header: int | None | Literal["detect"] = 0
) -> pd.DataFrame:
"""
Read Excel file as DataFrame.
Args:
sheet_name: Sheet to read (default: first sheet)
header: Row to use as header (0-based). Use "detect" for automatic
header detection. None means no header. (default: 0)
Returns:
Pandas DataFrame with raw data
Raises:
FileNotFoundError: If file doesn't exist
ValueError: If file is invalid/corrupted
"""
if not self.exists:
raise FileNotFoundError(f"Excel file not found: {self._path}")
if self._path.suffix.lower() not in {".xlsx", ".xls"}:
raise ValueError(f"Not an Excel file: {self._path}")
try:
# Use sheet_name=0 to read first sheet when None specified
# (pd.read_excel returns dict when sheet_name=None)
actual_sheet = 0 if sheet_name is None else sheet_name
# Handle automatic header detection
if header == "detect":
from excel_to_sql.auto_pilot.header_detector import HeaderDetector
detector = HeaderDetector()
header_row = detector.detect_header_row(self._path, actual_sheet)
return pd.read_excel(self._path, sheet_name=actual_sheet, header=header_row, engine="openpyxl")
return pd.read_excel(self._path, sheet_name=actual_sheet, header=header, engine="openpyxl")
except (FileNotFoundError, PermissionError):
# Re-raise filesystem errors as-is
raise
except pd.errors.EmptyDataError as e:
raise ExcelFileError(
f"Excel file is empty: {self._path.name}",
file_path=str(self._path),
operation="read"
) from e
except pd.errors.ParserError as e:
raise ExcelFileError(
f"Invalid Excel file format: {self._path.name}",
file_path=str(self._path),
operation="read"
) from e
except Exception as e:
raise ExcelFileError(
f"Failed to read Excel file: {self._path.name}",
file_path=str(self._path),
operation="read"
) from e
def read_all_sheets(self) -> Dict[str, pd.DataFrame]:
"""
Read all sheets from Excel file.
Returns:
Dictionary mapping sheet names to DataFrames
Example:
file = ExcelFile("data.xlsx")
sheets = file.read_all_sheets()
for sheet_name, df in sheets.items():
print(f"{sheet_name}: {len(df)} rows")
"""
if not self.exists:
raise FileNotFoundError(f"Excel file not found: {self._path}")
try:
return pd.read_excel(self._path, sheet_name=None, engine="openpyxl")
except (FileNotFoundError, PermissionError):
raise
except pd.errors.EmptyDataError as e:
raise ExcelFileError(
f"Excel file is empty: {self._path.name}",
file_path=str(self._path),
operation="read_all_sheets"
) from e
except pd.errors.ParserError as e:
raise ExcelFileError(
f"Invalid Excel file format: {self._path.name}",
file_path=str(self._path),
operation="read_all_sheets"
) from e
except Exception as e:
raise ExcelFileError(
f"Failed to read Excel file: {self._path.name}",
file_path=str(self._path),
operation="read_all_sheets"
) from e
def read_sheets(self, sheet_names: List[str]) -> Dict[str, pd.DataFrame]:
"""
Read specific sheets from Excel file.
Args:
sheet_names: List of sheet names to read
Returns:
Dictionary mapping sheet names to DataFrames
Example:
file = ExcelFile("data.xlsx")
sheets = file.read_sheets(["Products", "Categories"])
"""
result = {}
for sheet_name in sheet_names:
try:
result[sheet_name] = self.read(sheet_name)
except ExcelFileError:
# Re-raise ExcelFileError as-is
raise
except Exception as e:
raise ExcelFileError(
f"Failed to read sheet: {sheet_name}",
file_path=str(self._path),
operation="read_sheets"
) from e
return result
@property
def content_hash(self) -> str:
"""
SHA-256 hash of DataFrame content (not file bytes).
Lazy computation - cached after first call.
Hash is based on first sheet content.
Deprecated: Use get_content_hash(sheet_name) for multi-sheet support.
"""
if self._hash_cache is None:
df = self.read()
self._hash_cache = {"": self._compute_hash(df)}
return self._hash_cache.get("", "")
def get_content_hash(self, sheet_name: Optional[str] = None) -> str:
"""
Get SHA-256 hash of a specific sheet's content.
Args:
sheet_name: Sheet name (default: first sheet)
Returns:
Hexadecimal SHA-256 hash
Hash is based on:
- Column names (sorted)
- Row values (converted to string)
- NOT file metadata (mtime, size)
"""
cache_key = sheet_name or ""
if self._hash_cache is None:
self._hash_cache = {}
if cache_key not in self._hash_cache:
df = self.read(sheet_name)
self._hash_cache[cache_key] = self._compute_hash(df)
return self._hash_cache[cache_key]
def get_combined_hash(self) -> str:
"""
Get combined hash of all sheets.
Returns:
Hexadecimal SHA-256 hash of all sheets combined
Useful for detecting changes in multi-sheet files.
"""
sheets = self.read_all_sheets()
# Combine all sheets into one hash
combined = ""
for sheet_name in sorted(sheets.keys()):
sheet_hash = self._compute_hash(sheets[sheet_name])
combined += f"{sheet_name}:{sheet_hash}"
return hashlib.sha256(combined.encode()).hexdigest()
def validate(self) -> bool:
"""
Quick validation that file is readable.
Returns:
True if file can be read
Doesn't load full data - just checks:
- File exists
- Has .xlsx extension
- Can read sheet names
"""
if not self.exists:
return False
if self._path.suffix.lower() not in {".xlsx", ".xls"}:
return False
try:
pd.ExcelFile(self._path, engine="openpyxl")
return True
except Exception:
return False
# ──────────────────────────────────────────────────────────────
# PRIVATE METHODS
# ──────────────────────────────────────────────────────────────
def _compute_hash(self, df: pd.DataFrame) -> str:
"""
Compute hash of DataFrame content.
Args:
df: DataFrame to hash
Returns:
Hexadecimal SHA-256 hash
Hash computation strategy:
- Sort columns for consistency
- Convert to string representation
- Hash the resulting string
This ensures that same data produces same hash,
regardless of original column order.
"""
# Sort columns for consistency
df_sorted = df[sorted(df.columns)]
# Convert to string representation
content_str = df_sorted.to_string(index=True)
# Compute SHA-256
return hashlib.sha256(content_str.encode()).hexdigest()