读取TXT数据转Excel列的Python脚本新增数据后失效求助

问题描述 #

编写的Python脚本用于将raw.txt文件中的行数据提取至Excel列，初始测试正常，但在添加含无效内容的新数据后脚本报错失效。

原脚本代码 #

import xlrd, xlwt, re
from svnscripts.timestampdirectory import createdir, path_dir
import os
import pandas as pd
import time
def clearcasevobs():
    pathdest = path_dir()
    dest = createdir()
    timestr = time.strftime("%Y-%m-%d")

    txtName = rf"{pathdest}\{timestr}-clearcaseRawData-vobsDetails.txt"
    workBook = xlwt.Workbook(encoding='ascii')
    workSheet = workBook.add_sheet('sheet1')

    fp = open(txtName, 'r+b')

    # header
    workSheet.write(0, 0, "Tag")
    workSheet.write(0, 1, "CreateDate")
    workSheet.write(0, 2, "Created By")
    workSheet.write(0, 3, "Storage Host Pathname")
    workSheet.write(0, 4, "Storage Global Pathname")
    workSheet.write(0, 5, "DB Schema Version")
    workSheet.write(0, 6, "Mod_by_rem_user")
    workSheet.write(0, 7, "Atomic Checkin")
    workSheet.write(0, 8, "Owner user")
    workSheet.write(0, 9, "Owner Group")
    workSheet.write(0, 10, "ACLs enabled")
    workSheet.write(0, 11, "FeatureLevel")

    row = 0
    entries = 0
    fullentry = []
    for linea in fp.readlines():

        str_linea = linea.decode('gb2312', 'ignore')
        str_linea = str_linea[:-2]  # str  string

        txt = str_linea
        arr = str_linea

        if arr[:9] == "versioned":
            txt = arr
            entries += 1
            s = txt.index("/")
            e = txt.index('"', s)
            txt = txt[s:e]
            fullentry.append(txt)

        elif arr.find("created") >= 0:
            entries += 1
            txt = arr
            s = txt.index("created")
            e = txt.index("by")
            txt1 = txt[s + 7:20]
            fullentry.append(txt1)

            txt2 = txt[e + 3:]
            fullentry.append(txt2)

        elif arr.find("VOB storage host:pathname") >= 0:
            entries += 1
            txt = arr
            s = txt.index('"')
            e = txt.index('"', s + 1)
            txt = txt[s + 1:e]
            fullentry.append(txt)

        elif arr.find("VOB storage global pathname") >= 0:
            entries += 1
            txt = arr
            s = txt.index('"')
            e = txt.index('"', s + 1)
            txt = txt[s + 1:e]
            fullentry.append(txt)

        elif arr.find("database schema version:") >= 0:
            entries += 1
            txt = arr
            txt = txt[-2:]
            fullentry.append(txt)

        elif arr.find("modification by remote privileged user:") >= 0:
            entries += 1
            txt = arr
            s = txt.index(':')
            txt = txt[s + 2:]
            fullentry.append(txt)

        elif arr.find("tomic checkin:") >= 0:
            entries += 1
            txt = arr
            s = txt.index(':')
            txt = txt[s + 2:]
            fullentry.append(txt)

        elif arr.find("owner ") >= 0:
            entries += 1
            txt = arr
            s = txt.index('owner')
            txt = txt[s + 5:]
            fullentry.append(txt)

        elif arr.find("group tmn") >= 0:
            if arr.find("tmn/root") == -1:
                entries += 1
                txt = arr
                s = txt.index('group')
                entries += 1
                txt = txt[s + 5:]
                fullentry.append(txt)

        elif arr.find("ACLs enabled:") >= 0:
            entries += 1
            txt = arr
            txt = txt[-2:]
            fullentry.append(txt)

        elif arr.find("FeatureLevel =") >= 0:
            entries += 1
            txt = arr
            txt = txt[-1:]
            fullentry.append(txt)

        if (row == 65536):
            break;

    finalarr = []
    finalarr1 = []
    temp = 0
    row = 1

    for r in fullentry:
        finalarr.append(r)
        temp += 1

        if temp == 12:
            finalarr1.append(finalarr)

            temp = 0

            col = 0
            for arr in finalarr:
                workSheet.write(row, col, arr)
                col += 1
            row += 1
            finalarr.clear()
            if (row == 65536):
                break;

    workBook.save(os.path.join(dest, "ClearcaseReport.xls"))
    fp.close()

问题原因分析 #

1. 无异常捕获机制：脚本大量使用index()方法查找字符串，当新TXT中出现不符合预期格式的行时（如目标关键词缺失），会直接抛出ValueError导致崩溃。
2. 依赖固定位置截取：例如txt[s + 7:20]这类按固定索引截取的逻辑，一旦数据格式稍有变化就会提取错误内容或引发异常。
3. 条目计数逻辑脆弱：通过累加entries和temp到12来拆分每个VOB的完整信息，若某条VOB数据缺失字段或多了无效行，会导致后续所有数据错位，最终写入Excel时匹配错误。
4. xls格式限制：使用xlwt生成的xls文件最大支持65536行，新数据量接近或超过该值时会写入失败。

稳定替代实现方案 #

改用pandas处理数据，按VOB条目分组解析，添加异常处理，生成支持更大数据量的xlsx格式文件：

import os
import time
import pandas as pd
from svnscripts.timestampdirectory import createdir, path_dir

def clearcasevobs():
    pathdest = path_dir()
    dest = createdir()
    timestr = time.strftime("%Y-%m-%d")
    txt_path = rf"{pathdest}\{timestr}-clearcaseRawData-vobsDetails.txt"
    
    # 定义列名
    columns = [
        "Tag", "CreateDate", "Created By", "Storage Host Pathname",
        "Storage Global Pathname", "DB Schema Version", "Mod_by_rem_user",
        "Atomic Checkin", "Owner user", "Owner Group", "ACLs enabled", "FeatureLevel"
    ]
    
    vob_data = []
    current_vob = {col: "" for col in columns}
    
    with open(txt_path, 'r', encoding='gb2312', errors='ignore') as fp:
        for line in fp:
            line = line.strip()
            if not line:
                continue
            
            # 匹配Tag（versioned开头的行）
            if line.startswith("versioned"):
                # 保存上一个VOB数据（如果已收集部分字段）
                if any(current_vob.values()):
                    vob_data.append(current_vob.copy())
                    current_vob = {col: "" for col in columns}
                try:
                    start = line.index("/")
                    end = line.index('"', start)
                    current_vob["Tag"] = line[start:end]
                except (ValueError, IndexError):
                    pass
            
            # 匹配创建日期和创建者
            elif "created" in line and "by" in line:
                try:
                    created_idx = line.index("created")
                    by_idx = line.index("by")
                    current_vob["CreateDate"] = line[created_idx+7:by_idx-1].strip()
                    current_vob["Created By"] = line[by_idx+3:].strip()
                except ValueError:
                    pass
            
            # 匹配存储主机路径
            elif "VOB storage host:pathname" in line:
                try:
                    start = line.index('"')
                    end = line.index('"', start+1)
                    current_vob["Storage Host Pathname"] = line[start+1:end]
                except ValueError:
                    pass
            
            # 匹配全局存储路径
            elif "VOB storage global pathname" in line:
                try:
                    start = line.index('"')
                    end = line.index('"', start+1)
                    current_vob["Storage Global Pathname"] = line[start+1:end]
                except ValueError:
                    pass
            
            # 匹配数据库版本
            elif "database schema version:" in line:
                try:
                    current_vob["DB Schema Version"] = line.split(":")[-1].strip()
                except IndexError:
                    pass
            
            # 匹配远程用户修改权限
            elif "modification by remote privileged user:" in line:
                try:
                    current_vob["Mod_by_rem_user"] = line.split(":")[-1].strip()
                except IndexError:
                    pass
            
            # 匹配原子提交设置
            elif "Atomic checkin:" in line or "tomic checkin:" in line:
                try:
                    current_vob["Atomic Checkin"] = line.split(":")[-1].strip()
                except IndexError:
                    pass
            
            # 匹配所有者用户
            elif "owner " in line:
                try:
                    current_vob["Owner user"] = line.split("owner")[-1].strip()
                except IndexError:
                    pass
            
            # 匹配所有者组
            elif "group tmn" in line and "tmn/root" not in line:
                try:
                    current_vob["Owner Group"] = line.split("group")[-1].strip()
                except IndexError:
                    pass
            
            # 匹配ACL启用状态
            elif "ACLs enabled:" in line:
                try:
                    current_vob["ACLs enabled"] = line.split(":")[-1].strip()
                except IndexError:
                    pass
            
            # 匹配FeatureLevel
            elif "FeatureLevel =" in line:
                try:
                    current_vob["FeatureLevel"] = line.split("=")[-1].strip()
                except IndexError:
                    pass
        
        # 加入最后一个VOB数据
        if any(current_vob.values()):
            vob_data.append(current_vob)
    
    # 生成DataFrame并保存为xlsx
    df = pd.DataFrame(vob_data, columns=columns)
    df.to_excel(os.path.join(dest, "ClearcaseReport.xlsx"), index=False, encoding='utf-8')

if __name__ == "__main__":
    clearcasevobs()

方案优势 #

1. 异常容错：对所有字符串操作添加try-except捕获异常，避免单个无效行导致整个脚本崩溃。
2. 按条目分组解析：以每个VOB为单位收集数据，不会因字段缺失或无效行导致后续数据错位。
3. 格式兼容：生成xlsx格式文件，突破xls的65536行限制，支持更大数据量。
4. 代码可读性：使用字典映射字段，逻辑更清晰，便于维护和扩展。

内容的提问来源于stack exchange，提问作者Champs