mirror of
https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git
synced 2026-11-08 01:06:59 +08:00
106 lines
3.8 KiB
Python
106 lines
3.8 KiB
Python
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
|
|
|
|
import pandas as pd
|
|
|
|
FIELD_TYPE_MAP = {
|
|
'int64': 'int',
|
|
'int32': 'int',
|
|
'float64': 'float',
|
|
'float32': 'float',
|
|
'datetime64': 'datetime',
|
|
'datetime64[ns]': 'datetime',
|
|
'object': 'string',
|
|
'string': 'string',
|
|
'bool': 'string',
|
|
}
|
|
|
|
USER_TYPE_TO_PANDAS = {
|
|
'int': 'int64',
|
|
'float': 'float64',
|
|
'datetime': 'datetime64[ns]',
|
|
'string': 'string',
|
|
}
|
|
|
|
|
|
def infer_field_type(dtype) -> str:
|
|
dtype_str = str(dtype)
|
|
return FIELD_TYPE_MAP.get(dtype_str, 'string')
|
|
|
|
|
|
def _round_import_integer(value):
|
|
"""Round explicitly selected integer values without a float intermediate."""
|
|
if pd.isna(value) or (isinstance(value, str) and not value.strip()):
|
|
return pd.NA
|
|
if isinstance(value, bool):
|
|
return int(value)
|
|
try:
|
|
number = Decimal(str(value))
|
|
except InvalidOperation as exc:
|
|
raise ValueError('Invalid numeric value for integer field') from exc
|
|
if not number.is_finite():
|
|
raise ValueError('Integer fields require finite numeric values')
|
|
rounded = number.to_integral_value(rounding=ROUND_HALF_UP)
|
|
if rounded < -(2 ** 63) or rounded > 2 ** 63 - 1:
|
|
raise ValueError('Rounded value is outside the signed 64-bit integer range')
|
|
return int(rounded)
|
|
|
|
|
|
def read_import_dataframe(save_path: str, sheet_name: str, field_mapping: dict):
|
|
"""Read stored cell values; round only columns explicitly selected as integers."""
|
|
dtype_dict = {
|
|
column: 'object' if field_type == 'int' else USER_TYPE_TO_PANDAS.get(field_type, 'string')
|
|
for column, field_type in field_mapping.items()
|
|
}
|
|
if save_path.endswith('.csv'):
|
|
df = pd.read_csv(save_path, engine='c', dtype=dtype_dict)
|
|
else:
|
|
df = pd.read_excel(save_path, sheet_name=sheet_name, engine='calamine', dtype=dtype_dict)
|
|
for column, field_type in field_mapping.items():
|
|
if field_type == 'int' and column in df.columns:
|
|
try:
|
|
# Build a nullable integer array directly: Series.map can coerce
|
|
# large integers plus missing values to lossy float64 values.
|
|
df[column] = pd.array([_round_import_integer(value) for value in df[column]], dtype='Int64')
|
|
except ValueError as exc:
|
|
raise ValueError(f"Column '{column}': {exc}") from exc
|
|
return df
|
|
|
|
|
|
def parse_excel_preview(save_path: str, max_rows: int = 10):
|
|
sheets_data = []
|
|
if save_path.endswith(".csv"):
|
|
df = pd.read_csv(save_path, engine='c')
|
|
fields = []
|
|
for col in df.columns:
|
|
fields.append({
|
|
"fieldName": col,
|
|
"fieldType": infer_field_type(df[col].dtype)
|
|
})
|
|
preview_df = df.head(max_rows).replace({pd.NA: None, float('nan'): None})
|
|
preview_data = preview_df.to_dict(orient='records')
|
|
sheets_data.append({
|
|
"sheetName": "Sheet1",
|
|
"fields": fields,
|
|
"data": preview_data,
|
|
"rows": len(df)
|
|
})
|
|
else:
|
|
sheet_names = pd.ExcelFile(save_path).sheet_names
|
|
for sheet_name in sheet_names:
|
|
df = pd.read_excel(save_path, sheet_name=sheet_name, engine='calamine')
|
|
fields = []
|
|
for col in df.columns:
|
|
fields.append({
|
|
"fieldName": col,
|
|
"fieldType": infer_field_type(df[col].dtype)
|
|
})
|
|
preview_df = df.head(max_rows).replace({pd.NA: None, float('nan'): None})
|
|
preview_data = preview_df.to_dict(orient='records')
|
|
sheets_data.append({
|
|
"sheetName": sheet_name,
|
|
"fields": fields,
|
|
"data": preview_data,
|
|
"rows": len(df)
|
|
})
|
|
return sheets_data
|