You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python脚本输出的SQLite数据库、Excel及CSV文件全部损坏的问题排查求助

Python脚本输出的SQLite数据库、Excel及CSV文件全部损坏的问题排查求助

大家好,我最近写了一个Python脚本,用来处理Baselight和Xytech的元数据文件,还会结合视频文件做帧范围匹配,最后输出到SQLite数据库、Excel和CSV文件里。脚本的核心功能如下:

  • 读取并解析Baselight和Xytech的文本元数据
  • 将解析后的结构化数据存入SQLite数据库
  • 通过ffprobe获取视频时长,筛选出视频实际包含的有效帧范围
  • 把匹配成功的帧数据导出到Excel文件,未匹配的导出到CSV文件

但现在遇到了一个非常棘手的问题:脚本运行全程没有报错或崩溃,但所有输出的文件都无法正常使用:

  • SQLite数据库文件用专业查看器打开时,直接提示文件损坏
  • 生成的Excel文件(output.xls)打开时,Excel会弹出格式错误提示,无法正常加载内容
  • CSV文件里的内容要么是乱码,要么是无效的异常数据

我已经做了这些排查尝试,但还是没找到问题根源:

  • 调试过解析后的元数据,确认数据格式完全正确,没有异常值或空数据
  • 给数据库插入、文件写入环节都加了错误处理逻辑,运行时没有抛出任何异常
  • 反复确认了输出目录的读写权限,完全没问题
  • 每次运行脚本前都会删除之前的损坏文件,避免旧文件覆盖干扰
  • 用最小规模的有效输入文件测试,结果还是一样,输出文件依旧损坏

下面是我的完整脚本代码,麻烦各位帮忙看看哪里可能出了问题:

import os
import sqlite3
import subprocess
from openpyxl import Workbook
import csv

def read_file(file_path):
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"File not found: {file_path}")
    with open(file_path, 'r') as file:
        return file.readlines()

# Parse Baselight data
def parse_baselight(data):
    parsed_frames = []
    for line in data:
        if "<err>" in line:
            continue
        components = line.strip().split()
        if len(components) < 2:
            continue
        filename = components[0]
        frame_data = components[1:]
        numeric_frames = sorted(set(validate_numeric(frame) for frame in frame_data if validate_numeric(frame)))
        if numeric_frames:
            frames = format_frames(numeric_frames)
            parsed_frames.append((filename, frames))
    return parsed_frames

def validate_numeric(value):
    try:
        return int(value)
    except ValueError:
        return None

def format_frames(frames):
    ranges = []
    start = frames[0]
    end = frames[0]
    for i in range(1, len(frames)):
        if frames[i] == end + 1:
            end = frames[i]
        else:
            ranges.append(f"{start}-{end}" if start != end else f"{start}")
            start = frames[i]
            end = frames[i]
    ranges.append(f"{start}-{end}" if start != end else f"{start}")
    return ", ".join(ranges)

def parse_xytech(data):
    parsed_orders = []
    for line in data:
        if '/' in line:
            components = line.strip().split('/')
            location = components[1].strip()
            workorder = components[-1].strip()
            parsed_orders.append((location, workorder))
    return parsed_orders

def populate_database(baselight_data, xytech_data):
    conn = sqlite3.connect("thecrucible_databasedb")
    cursor = conn.cursor()

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS Baselight (
            Location TEXT,
            Frames TEXT
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS Xytech (
            Location TEXT,
            Workorder TEXT
        )
    """)

    cursor.executemany("INSERT INTO Baselight (Location, Frames) VALUES (?, ?)", baselight_data)
    cursor.executemany("INSERT INTO Xytech (Location, Workorder) VALUES (?, ?)", xytech_data)

    conn.commit()
    conn.close()

def get_video_length(video_path):
    result = subprocess.run(
        [
            "ffprobe",
            "-i", video_path,
            "-v", "error",
            "-show_entries", "format=duration",
            "-of", "default=noprint_wrappers=1:nokey=1"
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    try:
        return float(result.stdout.strip())
    except ValueError:
        print("Error extracting video length.")
        return 0

def find_matching_ranges(video_length, frame_ranges):
    matching_ranges = []
    fps = 24
    video_frames = int(video_length * fps)
    for frame_range in frame_ranges:
        if '-' in frame_range:
            start, end = map(int, frame_range.split('-'))
        else:
            start = end = int(frame_range)
        if start <= video_frames and end <= video_frames:
            matching_ranges.append(frame_range)
    return matching_ranges

def export_to_xls(data, output_path):
    wb = Workbook()
    ws = wb.active
    ws.title = "Matched Data"
    ws.append(["Filename", "Frame Range"])
    for row in data:
        ws.append(row)
    wb.save(output_path)

def export_unused_to_csv(unused_data, output_path):
    with open(output_path, 'w', newline='') as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(["Filename", "Frame Range"])
        writer.writerows(unused_data)

注:实际运行的脚本里已经导入了所有必要的模块(比如os、sqlite3等),贴代码时特意补全了之前遗漏的导入语句,这点不用纠结。现在实在找不到问题在哪,恳请各位大佬帮忙分析一下,谢谢!

备注:内容来源于stack exchange,提问作者Elijah Lockett

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.14 16:37:58