SQL Serverから出力したrptファイルを使いやすいようにcsvファイルに変換する【Python】

f:id:piro_suke:20160622005308j:plain

SQL Serverからエクスポートしたrpt形式ファイルを渡され、

CSVに変換してくれと依頼される。

「あれ、SQL Server Management Studioに

直接CSV出力する機能ありませんでしたっけ？」

と確認しても「しらん。やれ。」と言われる。

そんなよくある状況で助けてくれるrpt2csv変換スクリプトを作成した。

SQL Serverが吐くもの以外でrpt形式のファイルは見たことがないが、

中身は下記のような形式になっている：

カラム名1 カラム名2 カラム名3 ...
--------- ---------- ---------- ...
値1-1        値1-2         値1-3        ...
値2-1        値2-2        値2-3        ...

1行目がカラム名、

2行目がスペース区切りでハイフンがカラムサイズ分ずつ並び、

3行目以降がカラムサイズ分ずつスペース埋めでデータが並ぶ

フォーマットになっているようだ。

2行目でハイフンの数を元にカラムごとのサイズを取得し、

それを元に3行目以降のデータをカラムサイズごとに切り出し、

スペースをtrimすればデータを取り出すことができそうだ。

スクリプトを作成する

言語はPython3。2でも動くかな？

ファイルから1行ずつデータを取得するとして、

それぞれの行（カラム名行、ハイフン行、データ行）のデータを

変換する関数を作っていく。

1行目のカラム名行は、

連続するスペースを1つにしてからスペースで分割してリストにする。

import re

def create_header_list(line):
    line = re.sub(r'\s+', ' ', line)
    return line.split(" ")

2行目のハイフン行は、

スペースで分割してハイフンの文字数を取得して

カラム別文字数リストにする。

def create_length_list(line):
    col_list = line.split(" ")
    return [len(x) for x in col_list]

3行目以降のデータ行は、

2行目から生成したカラム別文字数リストを使って、

先頭から順番に文字数分ずつ部分文字列を切り出していく。

スペース埋めされているので、trimして、リストにまとめる。

def create_data_list(line, length_list):
    data_list = []
    char_index = 0 
    for length in length_list:
        new_length = char_index + length
        chunk = line[char_index:new_length].strip()
        data_list.append(chunk)
        char_index += length + 1 
    return data_list

行データをリストにしてしまえば、Pythonのcsvモジュールで

簡単にCSV化して出力できる。

rptファイルを読み込んでcsvファイルを出力するメイン処理を加えて、

全体として下記のようなスクリプトになった。

rptファイルの文字コードがBOM付きutf-8だったので、

読み込み時のエンコードを「utf-8-sig」と指定している。

rpt2csv.py

# -*- coding: utf-8 -*-

import re
import csv 

rpt_file_path = "＜rptファイルのパス＞"
csv_file_path = "＜csvファイルの出力先パス＞"

def create_header_list(line):
    line = re.sub(r'\s+', ' ', line)
    return line.split(" ")

def create_length_list(line):
    col_list = line.split(" ")
    return [len(x) for x in col_list]

def create_data_list(line, length_list):
    data_list = []
    char_index = 0 
    for length in length_list:
        new_length = char_index + length
        chunk = line[char_index:new_length].strip()
        data_list.append(chunk)
        char_index += length + 1 
    return data_list

line_number = 0 
header_list = []
length_list = []
with open(rpt_file_path, encoding="utf-8-sig") as rpt_file:
    with open(csv_file_path, "w", newline='', encoding="cp932") as out_csv_file:
        writer = csv.writer(out_csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_ALL)
        for line in rpt_file:
            line_number += 1

            if line_number == 1:
                header_list = create_header_list(line)
                writer.writerow(header_list)
                continue
            if line_number == 2:
                length_list = create_length_list(line)
                continue

            data_list = create_data_list(line, length_list)
            writer.writerow(data_list)