まだ書きかけです。フォルダーツリーとか載せてません。

データをsqlliteのテーブルに登録するとき、
列(カラム)が新規データと差分が出てしまうのを自動で
解決したくてつくりました。

データ分析に使うので、データ型はINTで決め打ちになってます。
これをstreamlitに応用したら、また記事書きます。

import sqlite3
import csv

open_csv = open("test.csv")
read_csv = csv.reader(open_csv)
master_row_columns = []
for row in read_csv:
    master_row_columns.append(row)
master_columns = master_row_columns[0]

open_csv = open("test2.csv")
read_csv = csv.reader(open_csv)
insert_row_columns = []
for row in read_csv:
    insert_row_columns.append(row)
insert_columns = insert_row_columns[0]

master_columns_list = master_columns

print(master_columns)
print(insert_columns)
diff_list = set(master_columns) ^ set(insert_columns)
diff_list = list(diff_list)

insert_name = ''
for name in insert_columns:
    insert_name = insert_name  + name + ","
insert_name = insert_name[:-1]
print(insert_name)
print(diff_list)
for diff in diff_list:
    master_columns_list.append(diff)
print(master_columns_list)

insert_count_text = ''
for i in range(len(insert_columns)):
    insert_count_text = "?," + insert_count_text
insert_count_text = insert_count_text[:-1]
print(insert_count_text)

#test.dbを作成し、接続（すでに存在する場合は接続のみ）
con = sqlite3.connect("test.db")
cur = con.cursor()

#testテーブルを作成（IF NOT EXISTSは「存在しなければ作成する」という意味）
create_test = "CREATE TABLE IF NOT EXISTS test (id INTEGER, name TEXT, height INTEGER, weight INTEGER)"
cur.execute(create_test)

#マスターのカラム更新
if diff_list is not None:
    for column in diff_list:
        alter_columns_test = "ALTER TABLE test ADD COLUMN "+ column +" INTEGER"
        cur.execute(alter_columns_test)

#testテーブルのデータを削除（何回もコード実行すると同じデータ追加されるので）
delete_test = "DELETE FROM TEST"
cur.execute(delete_test)
#csvファイルの指定
open_csv = open("test2.csv")

#csvファイルを読み込む
read_csv = csv.reader(open_csv)

#next()関数を用いて最初の行(列名)はスキップさせる
next_row = next(read_csv)

#csvデータをINSERTする
rows = []
for row in read_csv:
    rows.append(row)

#executemany()で複数のINSERTを実行する
cur.executemany(
    "INSERT INTO test (" + insert_name +") VALUES ("+ insert_count_text +")", rows)

#テーブルの変更内容保存
#csvも閉じておきましょう
con.commit()
open_csv.close()

#testテーブルの確認
select_test = "SELECT * FROM test"
select_test_result = cur.execute(select_test)

with open('test.csv', 'w', newline="") as f:
    writer = csv.writer(f)
    writer.writerow(master_columns_list)
    writer.writerows(select_test_result)


print("—————————-")
print("fetchall")
print("—————————-")
print(cur.execute(select_test))
print(cur.fetchall())
print("—————————-")
print("for文")
print("—————————-")
for i in cur.execute(select_test):
    print(i)

#データベースの接続終了
con.close