Python Pandas on_bad_lines – расширение информации из предупреждения по умолчанию «warn» ⇐ Python

[Anonymous]

Расширение этого вопроса о stackoverflow:
Используйте on_bad_lines для записи недопустимых строк из pandas.read_csv в файл.
Поведение on_bad_lines по умолчанию ='warn' — вывести следующее:
2024-11-14 10:06:33,515 - WARNING - C:\xxx\S_EQUI.CSV: Skipping line 46272: expected 82 fields, saw 92
Skipping line 47655: expected 82 fields, saw 172

Это результат использования метода Pandas под названием _alert_malformed, который:
def _alert_malformed(self, msg: str, row_num: int) -> None:
"""
Alert a user about a malformed row, depending on value of
`self.on_bad_lines` enum.

If `self.on_bad_lines` is ERROR, the alert will be `ParserError`.
If `self.on_bad_lines` is WARN, the alert will be printed out.

Parameters
----------
msg: str
The error message to display.
row_num: int
The row number where the parsing error occurred.
Because this row number is displayed, we 1-index,
even though we 0-index internally.
"""
if self.on_bad_lines == self.BadLineHandleMethod.ERROR:
raise ParserError(msg)
if self.on_bad_lines == self.BadLineHandleMethod.WARN:
warnings.warn(
f"Skipping line {row_num}: {msg}\n",
ParserWarning,
stacklevel=find_stack_level(),
)

Как расширить сообщение этого метода и вывод row_no в созданную мной пользовательскую вызываемую функцию, включающую bad_line?
Текущий сценарий:< /p>
def write_bad_line(file_path, line, fp, sep=';' ) -> None:
'''Writes a bad line to a file, then raises a warning'''
global current_file_path
if not file_path:
file_path = current_file_path
time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
complete_list = [time, str(file_path)] + line
complete_list = sep.join(complete_list) + '\n'

fp.write(complete_list)
warnings.warn(f'Bad Line - number of columns and number of delimiters do not match: {sep.join(line)}', pd.errors.ParserWarning)
return None

def bad_lines_processor(line: list) -> None:
'''Wrapper function to process bad lines'''
global current_file_path, bad_lines_fp
write_bad_line(file_path=current_file_path, line = line, fp=bad_lines_fp)

bad_lines_fp = open('bad_lines.csv', 'a')

def process_csv_file(file_path, json_data, output_writer, delimiter=';'):
'''Processes a single CSV file and writes it to the Excel writer.'''

# Set the current file path for logging purposes
# set_current_file_path(file_path) # Commented out for debugging purposes.

tab_name = os.path.splitext(os.path.basename(file_path))[0].replace('.csv', '')

with warnings.catch_warnings(record=True) as w:
df = pd.read_csv(file_path, sep=delimiter, header=0, dtype=str, engine='python', on_bad_lines=bad_lines_processor)

if w:
for warning in w:
logger.warning('%s: %s', file_path, warning.message)


Подробнее здесь: https://stackoverflow.com/questions/791 ... lt-warning