#!/usr/bin/env python3 """ 総勘定元帳データ移行スクリプト バイナリファイル → MariaDB Usage: python3 migrate_to_mariadb.py --dry-run # SQLファイル出力のみ（DB接続不要） python3 migrate_to_mariadb.py # MariaDBに直接投入 """ import os import sys import argparse from datetime import datetime BASE_DIR = os.path.dirname(os.path.abspath(__file__)) SETTING_DIR = os.path.join(BASE_DIR, '設定') DATA_DIR = os.path.join(BASE_DIR, '振替伝票データ') ACCOUNT_FILE = os.path.join(SETTING_DIR, '勘定科目') CATEGORY_NAMES = { 0: '資産', 1: '負債', 2: '資本', 3: '仕入', 4: '売上', 5: '一般管理費', 6: '営業外収益', 7: '営業外費用', 8: '製造原価' } ACCOUNT_NAME_FIXES = { '40': '預り金', } LEGACY_ACCOUNTS = [ {'name': '清水銀行本店', 'category': 0, 'category_name': '資産'}, {'name': '材料在庫', 'category': 0, 'category_name': '資産'}, {'name': '電話加入権', 'category': 0, 'category_name': '資産'}, {'name': '別途積立金', 'category': 2, 'category_name': '資本'}, ] def read_pstring(data, offset): if offset >= len(data): return '', offset length = data[offset] raw = data[offset + 1:offset + 1 + length] s = raw.decode('shift_jis', errors='replace') return s, offset + 1 + length def parse_accounts(path): with open(path, 'rb') as f: data = f.read() accounts = [] for i in range(134): base = i * 48 pos = base name, pos = read_pstring(data, pos) color, pos = read_pstring(data, pos) code1, pos = read_pstring(data, pos) code2, pos = read_pstring(data, pos) name = name.strip() color = color.strip() code1 = code1.strip() code2 = code2.strip() if name == '***': continue if name == '預託金' and i == 129: continue accounts.append({ 'master_index': i, 'name': name, 'category': int(color) if color.isdigit() else 0, 'code1': int(code1) if code1.isdigit() else None, 'code2': int(code2) if code2.isdigit() else None, }) return accounts def parse_amount(s): s = s.strip().replace(',', '') if not s: return 0 try: return int(float(s)) except ValueError: return 0 def parse_journal_file(path): with open(path, 'rb') as f: data = f.read() if len(data) == 0: return [] n_records = len(data) // 160 entries = [] for i in range(n_records): rec = data[i * 160:(i + 1) * 160] pos = 0 date, pos = read_pstring(rec, pos) debit_amount_s, pos = read_pstring(rec, pos) debit_code_raw = rec[pos + 1:pos + 5].decode('ascii', errors='replace').strip() pos += 5 debit_account, pos = read_pstring(rec, pos) description, pos = read_pstring(rec, pos) slip_no, pos = read_pstring(rec, pos) credit_code_raw = rec[pos + 1:pos + 5].decode('ascii', errors='replace').strip() pos += 5 credit_account, pos = read_pstring(rec, pos) credit_amount_s, pos = read_pstring(rec, pos) kubun, pos = read_pstring(rec, pos) debit_account_name = debit_account.strip() credit_account_name = credit_account.strip() debit_account_name = ACCOUNT_NAME_FIXES.get(debit_account_name, debit_account_name) credit_account_name = ACCOUNT_NAME_FIXES.get(credit_account_name, credit_account_name) entries.append({ 'date': date.strip(), 'debit_amount': parse_amount(debit_amount_s), 'debit_account': debit_account_name, 'credit_amount': parse_amount(credit_amount_s), 'credit_account': credit_account_name, 'description': description.strip(), 'slip_no': slip_no.strip(), 'kubun': kubun.strip(), }) return entries def group_into_slips(entries): slips = [] current_lines = [] debit_total = 0 credit_total = 0 for entry in entries: current_lines.append(entry) debit_total += entry['debit_amount'] credit_total += entry['credit_amount'] if debit_total > 0 and debit_total == credit_total: slip_date = current_lines[0]['date'] slip_no = '' for line in current_lines: if line['slip_no']: slip_no = line['slip_no'] break slips.append({ 'date': slip_date, 'slip_no': slip_no, 'total_amount': debit_total, 'lines': list(current_lines), }) current_lines = [] debit_total = 0 credit_total = 0 if current_lines: print(f" WARNING: {len(current_lines)} orphan lines (debit={debit_total}, credit={credit_total})", file=sys.stderr) slip_date = current_lines[0]['date'] slips.append({ 'date': slip_date, 'slip_no': '', 'total_amount': max(debit_total, credit_total), 'lines': list(current_lines), }) return slips def parse_carryforward_file(path): with open(path, 'rb') as f: data = f.read() if len(data) == 0: return [] n_records = len(data) // 80 records = [] for i in range(n_records): base = i * 80 name, _ = read_pstring(data, base) block = data[base + 13:base + 80] kari, p = read_pstring(block, 0) kasi, _ = read_pstring(block, p) records.append({ 'name': name.strip(), 'debit_balance': parse_amount(kari), 'credit_balance': parse_amount(kasi), }) return records def sql_escape(s): if s is None: return 'NULL' return "'" + str(s).replace("\\", "\\\\").replace("'", "\\'") + "'" def generate_sql(accounts, all_slips, all_carryforwards): lines = [] lines.append("-- 総勘定元帳 データ移行SQL") lines.append(f"-- Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") lines.append("") lines.append("SET NAMES utf8mb4;") lines.append("SET FOREIGN_KEY_CHECKS = 0;") lines.append("") # DDL lines.append("-- ========== テーブル作成 ==========") lines.append("") lines.append("""CREATE TABLE IF NOT EXISTS `gl_accounts` ( `id` INT UNSIGNED NOT NULL AUTO_INCREMENT, `master_index` INT NOT NULL COMMENT '元データの配列インデックス(0-133)', `name` VARCHAR(20) NOT NULL COMMENT '科目名', `category` TINYINT UNSIGNED NOT NULL COMMENT '0=資産,1=負債,2=資本,3=仕入,4=売上,5=一般管理費,6=営業外収益,7=営業外費用,8=製造原価', `category_name` VARCHAR(10) NOT NULL, `code1` INT DEFAULT NULL, `code2` INT DEFAULT NULL, `sort_order` INT NOT NULL COMMENT '表示順', PRIMARY KEY (`id`), UNIQUE KEY `uk_name` (`name`), KEY `idx_category` (`category`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='勘定科目マスター';""") lines.append("") lines.append("""CREATE TABLE IF NOT EXISTS `gl_journal_slips` ( `id` INT UNSIGNED NOT NULL AUTO_INCREMENT, `slip_date` DATE NOT NULL COMMENT '伝票日付', `fiscal_year` SMALLINT NOT NULL COMMENT '会計年度(4月始まり)', `fiscal_month` TINYINT NOT NULL COMMENT '会計月(1-12)', `slip_no` VARCHAR(10) DEFAULT NULL COMMENT '伝票番号', `total_amount` BIGINT NOT NULL COMMENT '借方合計=貸方合計', `line_count` TINYINT UNSIGNED NOT NULL COMMENT '明細行数', `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (`id`), KEY `idx_date` (`slip_date`), KEY `idx_fiscal` (`fiscal_year`, `fiscal_month`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='振替伝票ヘッダー';""") lines.append("") lines.append("""CREATE TABLE IF NOT EXISTS `gl_journal_lines` ( `id` INT UNSIGNED NOT NULL AUTO_INCREMENT, `slip_id` INT UNSIGNED NOT NULL, `line_no` TINYINT UNSIGNED NOT NULL COMMENT '行番号(1始まり)', `debit_account_id` INT UNSIGNED DEFAULT NULL COMMENT '借方科目', `debit_amount` BIGINT NOT NULL DEFAULT 0, `credit_account_id` INT UNSIGNED DEFAULT NULL COMMENT '貸方科目', `credit_amount` BIGINT NOT NULL DEFAULT 0, `description` VARCHAR(100) DEFAULT NULL COMMENT '摘要', `kubun` CHAR(2) DEFAULT NULL COMMENT '区分', PRIMARY KEY (`id`), KEY `idx_slip` (`slip_id`), KEY `idx_debit_account` (`debit_account_id`), KEY `idx_credit_account` (`credit_account_id`), CONSTRAINT `fk_line_slip` FOREIGN KEY (`slip_id`) REFERENCES `gl_journal_slips` (`id`), CONSTRAINT `fk_line_debit` FOREIGN KEY (`debit_account_id`) REFERENCES `gl_accounts` (`id`), CONSTRAINT `fk_line_credit` FOREIGN KEY (`credit_account_id`) REFERENCES `gl_accounts` (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='振替伝票明細';""") lines.append("") lines.append("""CREATE TABLE IF NOT EXISTS `gl_carry_forwards` ( `id` INT UNSIGNED NOT NULL AUTO_INCREMENT, `fiscal_year` SMALLINT NOT NULL, `fiscal_month` TINYINT NOT NULL COMMENT '月(1-12)', `account_id` INT UNSIGNED NOT NULL, `debit_balance` BIGINT NOT NULL DEFAULT 0, `credit_balance` BIGINT NOT NULL DEFAULT 0, PRIMARY KEY (`id`), UNIQUE KEY `uk_year_month_account` (`fiscal_year`, `fiscal_month`, `account_id`), KEY `idx_account` (`account_id`), CONSTRAINT `fk_cf_account` FOREIGN KEY (`account_id`) REFERENCES `gl_accounts` (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='繰越残高';""") lines.append("") lines.append("""CREATE TABLE IF NOT EXISTS `gl_settings` ( `key_name` VARCHAR(50) NOT NULL, `value` VARCHAR(200) DEFAULT NULL, PRIMARY KEY (`key_name`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='設定';""") lines.append("") # Accounts lines.append("-- ========== 勘定科目データ ==========") lines.append("TRUNCATE TABLE `gl_carry_forwards`;") lines.append("TRUNCATE TABLE `gl_journal_lines`;") lines.append("TRUNCATE TABLE `gl_journal_slips`;") lines.append("TRUNCATE TABLE `gl_accounts`;") lines.append("") name_to_id = {} aid = 0 for acct in accounts: aid += 1 name_to_id[acct['name']] = aid cat_name = CATEGORY_NAMES.get(acct['category'], '不明') lines.append( f"INSERT INTO `gl_accounts` (`id`,`master_index`,`name`,`category`,`category_name`,`code1`,`code2`,`sort_order`) " f"VALUES ({aid},{acct['master_index']},{sql_escape(acct['name'])},{acct['category']},{sql_escape(cat_name)}," f"{acct['code1'] if acct['code1'] is not None else 'NULL'}," f"{acct['code2'] if acct['code2'] is not None else 'NULL'}," f"{aid});" ) lines.append("-- 旧科目（繰越データにのみ存在）") for legacy in LEGACY_ACCOUNTS: aid += 1 name_to_id[legacy['name']] = aid lines.append( f"INSERT INTO `gl_accounts` (`id`,`master_index`,`name`,`category`,`category_name`,`code1`,`code2`,`sort_order`) " f"VALUES ({aid},-1,{sql_escape(legacy['name'])},{legacy['category']},{sql_escape(legacy['category_name'])}," f"NULL,NULL,{aid});" ) lines.append("") # Settings lines.append("-- ========== 設定 ==========") # gl_settings は TRUNCATE 対象外（PKは key_name）。再実行時の # 重複キー停止を避けるため REPLACE で冪等化する。 lines.append(f"REPLACE INTO `gl_settings` VALUES ('company_name', {sql_escape('法月工芸社')});") lines.append(f"REPLACE INTO `gl_settings` VALUES ('decimal_mode', '3');") lines.append(f"REPLACE INTO `gl_settings` VALUES ('fiscal_start_month', '4');") lines.append(f"REPLACE INTO `gl_settings` VALUES ('migrated_at', {sql_escape(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))});") lines.append("") # Journal slips lines.append("-- ========== 振替伝票データ ==========") slip_id = 0 unknown_accounts = set() for year, month, slips in all_slips: if not slips: continue lines.append(f"-- {year}年{month}月 ({len(slips)} slips)") for slip in slips: slip_id += 1 slip_date = slip['date'] try: dt = datetime.strptime(slip_date, '%Y/%m/%d') date_sql = dt.strftime('%Y-%m-%d') cal_month = dt.month if cal_month >= 4: fiscal_year = dt.year else: fiscal_year = dt.year - 1 fiscal_month = ((cal_month - 4) % 12) + 1 except ValueError: date_sql = '1900-01-01' fiscal_year = 1900 fiscal_month = 1 lines.append( f"INSERT INTO `gl_journal_slips` (`id`,`slip_date`,`fiscal_year`,`fiscal_month`,`slip_no`,`total_amount`,`line_count`) " f"VALUES ({slip_id},{sql_escape(date_sql)},{fiscal_year},{fiscal_month}," f"{sql_escape(slip['slip_no']) if slip['slip_no'] else 'NULL'}," f"{slip['total_amount']},{len(slip['lines'])});" ) for line_no, line in enumerate(slip['lines'], 1): debit_aid = name_to_id.get(line['debit_account']) credit_aid = name_to_id.get(line['credit_account']) if line['debit_account'] and not debit_aid: unknown_accounts.add(line['debit_account']) if line['credit_account'] and not credit_aid: unknown_accounts.add(line['credit_account']) lines.append( f"INSERT INTO `gl_journal_lines` (`slip_id`,`line_no`,`debit_account_id`,`debit_amount`,`credit_account_id`,`credit_amount`,`description`,`kubun`) " f"VALUES ({slip_id},{line_no}," f"{debit_aid if debit_aid else 'NULL'},{line['debit_amount']}," f"{credit_aid if credit_aid else 'NULL'},{line['credit_amount']}," f"{sql_escape(line['description']) if line['description'] else 'NULL'}," f"{sql_escape(line['kubun']) if line['kubun'] else 'NULL'});" ) lines.append("") # Carry-forwards lines.append("-- ========== 繰越残高データ ==========") cf_seen = set() for year, month, records in all_carryforwards: if not records: continue non_zero = [r for r in records if r['debit_balance'] != 0 or r['credit_balance'] != 0] if not non_zero: continue lines.append(f"-- 繰越 {year}年{month}月 ({len(non_zero)} non-zero accounts)") cal_month = int(month) if cal_month >= 4: fiscal_year = int(year) else: fiscal_year = int(year) - 1 for r in records: acct_id = name_to_id.get(r['name']) if not acct_id: unknown_accounts.add(r['name']) continue if r['debit_balance'] == 0 and r['credit_balance'] == 0: continue fiscal_month = ((cal_month - 4) % 12) + 1 cf_key = (fiscal_year, fiscal_month, acct_id) if cf_key in cf_seen: continue cf_seen.add(cf_key) lines.append( f"INSERT INTO `gl_carry_forwards` (`fiscal_year`,`fiscal_month`,`account_id`,`debit_balance`,`credit_balance`) " f"VALUES ({fiscal_year},{fiscal_month},{acct_id},{r['debit_balance']},{r['credit_balance']});" ) lines.append("") lines.append("SET FOREIGN_KEY_CHECKS = 1;") lines.append("") lines.append("-- ========== 伝票番号の自動採番 ==========") lines.append("UPDATE gl_journal_slips s") lines.append("JOIN (") lines.append(" SELECT id, ROW_NUMBER() OVER (PARTITION BY fiscal_year ORDER BY slip_date, id) AS new_no") lines.append(" FROM gl_journal_slips") lines.append(") t ON s.id = t.id") lines.append("SET s.slip_no = t.new_no;") lines.append("") if unknown_accounts: lines.append(f"-- WARNING: Unknown account names found: {unknown_accounts}") return '\n'.join(lines), unknown_accounts def main(): parser = argparse.ArgumentParser(description='総勘定元帳データ移行') parser.add_argument('--dry-run', action='store_true', help='SQLファイル出力のみ') parser.add_argument('--output', default=os.path.join(BASE_DIR, 'migration.sql'), help='出力SQLファイル') args = parser.parse_args() print("=== 総勘定元帳データ移行 ===") print() # 1. Parse accounts print("1. 勘定科目マスター読み込み...") accounts = parse_accounts(ACCOUNT_FILE) print(f" {len(accounts)} accounts loaded") # 2. Parse journal files print("2. 振替伝票データ読み込み...") all_slips = [] total_entries = 0 total_slips = 0 years = sorted([d for d in os.listdir(DATA_DIR) if d.isdigit()]) for year in years: year_dir = os.path.join(DATA_DIR, year) for month_num in range(1, 13): month_str = f'{month_num:02d}' fname = f'振伝{year}:{month_str}' fpath = os.path.join(year_dir, fname) if not os.path.exists(fpath): continue if os.path.getsize(fpath) == 0: all_slips.append((year, month_str, [])) continue entries = parse_journal_file(fpath) slips = group_into_slips(entries) all_slips.append((year, month_str, slips)) total_entries += len(entries) total_slips += len(slips) print(f" {total_entries} entries → {total_slips} slips ({len(years)} years)") # 3. Parse carry-forward files print("3. 繰越残高データ読み込み...") all_carryforwards = [] total_cf = 0 for year in years: year_dir = os.path.join(DATA_DIR, year) for month_num in range(1, 13): month_str = f'{month_num:02d}' fname = f'繰越{year}:{month_str}' fpath = os.path.join(year_dir, fname) if not os.path.exists(fpath): continue records = parse_carryforward_file(fpath) all_carryforwards.append((year, month_str, records)) total_cf += len(records) print(f" {total_cf} carry-forward records ({len(all_carryforwards)} months)") # 4. Generate SQL print("4. SQL生成中...") sql, unknown = generate_sql(accounts, all_slips, all_carryforwards) if unknown: print(f" WARNING: Unknown account names: {unknown}") output_path = args.output with open(output_path, 'w', encoding='utf-8') as f: f.write(sql) print(f" → {output_path} ({len(sql):,} bytes)") # Stats print() print("=== 統計 ===") print(f" 勘定科目: {len(accounts)}") print(f" 振替伝票: {total_slips} slips / {total_entries} lines") print(f" 繰越残高: {total_cf} records") print(f" 年度範囲: {years[0]}-{years[-1]}") if args.dry_run: print() print("(dry-run モード: SQLファイルのみ出力)") else: print() print("DB投入するにはまずSQLファイルを確認し、MariaDBに流してください:") print(f" mysql -u USER -p DATABASE < {output_path}") if __name__ == '__main__': main()