import os import logging import pandas as pd from pathlib import Path from typing import List, Tuple, Union from mylib.pdfzh_translator import OpenAITranslator from mylib.read_encoding_cvs import read_csv from mylib.logging_config import setup_logging # Setup custom logging setup_logging() logger = logging.getLogger('mylib.translate_utils') def column_letter_to_index(col_letter: str) -> int: """将列字母转换为列索引(从0开始) Args: col_letter: 列字母(如 'A', 'B', 'AA' 等) Returns: 列索引(从0开始) """ try: col_index = 0 for i, char in enumerate(reversed(col_letter.upper())): col_index += (ord(char) - ord('A') + 1) * (26 ** i) return col_index - 1 except Exception as e: logger.error(f"列字母转换时出错: {e}") raise def read_csv_with_header(file_path: str, header_row: int = 1, encoding: str = None) -> pd.DataFrame: """读取CSV文件并正确处理标题行 Args: file_path: CSV文件路径 header_row: 标题行号(从0开始),默认为1(第2行) encoding: 文件编码 Returns: pandas DataFrame """ try: if not os.path.exists(file_path): logger.error(f"文件不存在: {file_path}") raise FileNotFoundError(f"文件不存在: {file_path}") # 读取所有数据 data = read_csv(file_path, encoding) if not data: logger.error("读取的文件为空") raise ValueError("读取的文件为空") # 确保header_row在有效范围内 if header_row >= len(data): logger.error(f"标题行 {header_row} 超出文件范围") raise ValueError(f"标题行 {header_row} 超出文件范围") # 使用指定行作为列名,前面的行丢弃 df = pd.DataFrame(data[header_row+1:], columns=data[header_row]) logger.info(f"成功读取CSV文件,使用第{header_row+1}行作为标题行") logger.info(f"列标题: {df.columns.tolist()}") return df except Exception as e: logger.error(f"读取CSV文件时出错: {e}") raise def extract_column_data(df: pd.DataFrame, column_identifier: Union[str, int], start_row: int = 2, header_row: int = 1) -> pd.Series: """提取指定列的数据,默认从第3行开始 Args: df: pandas DataFrame column_identifier: 要提取的列名或列号(从0开始),也可以是列字母(如 'A', 'B') start_row: 开始提取的行号,默认为2(第3行) header_row: 标题行号,默认为1(第2行) Returns: 包含指定列数据的Series """ try: if df.empty: logger.error("DataFrame为空") return pd.Series() # 处理列号或列名或列字母 if isinstance(column_identifier, str) and column_identifier.isalpha(): column_identifier = column_letter_to_index(column_identifier) if isinstance(column_identifier, int): if column_identifier < 0 or column_identifier >= len(df.columns): logger.error(f"列号 {column_identifier} 超出范围") raise ValueError(f"列号 {column_identifier} 超出范围") column_identifier = df.columns[column_identifier] # 确保列名存在 if column_identifier not in df.columns: logger.error(f"列名 {column_identifier} 不存在") raise ValueError(f"列名 {column_identifier} 不存在") # 确保开始行在有效范围内 if start_row >= len(df) or start_row < 0: logger.error(f"开始行 {start_row} 超出范围") raise ValueError(f"开始行 {start_row} 超出范围") # 提取指定列的数据 column_data = df.iloc[start_row:][column_identifier] logger.info(f"成功提取列 {column_identifier} 数据,从第{start_row}行开始,共{len(column_data)}条数据") return column_data except Exception as e: logger.error(f"提取列数据时出错: {e}") raise def test_column_extraction(input_file: str): """测试列提取功能 Args: input_file: 输入CSV文件路径 """ try: if not os.path.exists(input_file): logger.error(f"文件不存在: {input_file}") raise FileNotFoundError(f"文件不存在: {input_file}") # 读取CSV文件 df = read_csv_with_header(input_file, header_row=1) # 提取第二列的数据,从第三行开始 column_data = extract_column_data(df, column_identifier=1, start_row=2, header_row=1) # 打印提取的数据 print("提取的列数据:") print(column_data) except Exception as e: logger.error(f"测试列提取时出错: {e}") if __name__ == '__main__': output_dir = Path('temp') input_file = output_dir / "测试.csv" test_column_extraction('/home/mrh/code/excel_tool/temp/测试.csv')