単位_Python

Slackデータの保管に関する考察 (API)
#SlackCollector[v48-6_20241015].html

ファイルのカジュアルな復号化に関する考察（Base64→DataURL）

ディレクトリ構造の精査とマッピングに関する小考察
get_directory_map.html

PythonでPostgreSQLとやりとりする
https://zenn.dev/collabostyle/articles/36e822520182d3

psycopg2
https://pypi.org/project/psycopg2/

ディレクトリ内の構造とファイルのハッシュ値を取得して整合性を検証するスクリプト
ファイル構造正常判定.html

def wareki_to_seireki(wareki: int) -> str:
    # 年号の対応表
    era_map = {
        1: (1868, "明治"),
        2: (1912, "大正"),
        3: (1926, "昭和"),
        4: (1989, "平成"),
        5: (2019, "令和")
    }
    
    # 7桁の数値を文字列に変換し、各部分を取得
    str_wareki = str(wareki).zfill(7)  # 桁不足の際に補完
    era_digit = int(str_wareki[0])  # 年号部分
    year = int(str_wareki[1:3])  # 年度部分
    month_day = str_wareki[3:7]  # 月日部分
    
    if era_digit not in era_map:
        raise ValueError("無効な年号の値です")
    
    base_year, era_name = era_map[era_digit]
    seireki_year = base_year + (year - 1)
    
    return f"{seireki_year}年{month_day[:2]}月{month_day[2:]}日 ({era_name}{year}年)"

def seireki_to_wareki(seireki: int, month: int, day: int) -> str:
    era_map = [
        (2019, "令和", 5),
        (1989, "平成", 4),
        (1926, "昭和", 3),
        (1912, "大正", 2),
        (1868, "明治", 1)
    ]
    
    for base_year, era_name, era_digit in era_map:
        if seireki >= base_year:
            wareki_year = seireki - base_year + 1
            return f"{era_digit}{str(wareki_year).zfill(2)}{str(month).zfill(2)}{str(day).zfill(2)}"
    
    raise ValueError("無効な西暦の値です")

# 使用例
print(wareki_to_seireki(5190701))  # 令和19年7月1日 -> 2037年7月1日
print(seireki_to_wareki(2037, 7, 1))  # 2037年7月1日 -> 5190701

定数管理スクリプト
#!/usr/bin/python
# -*- coding: utf-8 -*-

import site
import sys, warnings, copy
from types import MappingProxyType

class ConstantError(TypeError): # #定数用独自例外処理クラス (名称だけのクラス) ###INSIDE###
	"""定数への再代入例外処理用Exceptionクラス"""
	def __init__(self, message): #生成時に実行される ###INSIDE###
		super().__init__(message)

class Constant: #定数管理クラス
	def __init__(self, recursion_limit=20): #生成時に実行
		if recursion_limit is not None and type(recursion_limit) in [int, float]:
			self.set_constant_class_recursion_limit(recursion_limit)
		else:
			self.set_constant_class_recursion_limit(0)
	
	def __getattr__(self, name): #要素が見つからない場合に呼び出される関数 ###INSIDE###
		raise AttributeError("'{0}' does not exist. You can't refer to something that doesn't exist.".format(name))
	
	def __setattr__(self, name, value): #クラスインスタンスに値を代入する際に呼び出される関数 ###INSIDE###
		if name in self.__dict__: #インスタンス変数にすでに存在していないかを確認
			raise ConstantError("The constant named '{0}' cannot be changed.".format(name))
		elif name in set(dir(self)): #インスタンス内にすでに存在していないかを確認 (変数ではないが存在しているオブジェクト)
			raise ConstantError("'{0}' is already reserved.".format(name))
		else:
			try:
				self.__dict__[name] = copy.deepcopy(self.read_only_object_converter(value)) #参照渡しによる改変禁止
			except TypeError as e: #UnPickleエラー(ディープコピー断念)
				try:
					self.__dict__[name] = copy.copy(self.read_only_object_converter(value))
				except TypeError as e: #UnPickleエラー(シャーロコピー断念)
					self.__dict__[name] = self.read_only_object_converter(value)
	
	def __delattr__(self, name): #インスタンスから定数を消させないための関数 ###INSIDE###
		if name in self.__dict__: #インスタンス変数にすでに存在していないかを確認
			raise ConstantError("The constant named '{0}' cannot be deleted.".format(name))
		elif name in set(dir(self)): #インスタンス内にすでに存在していないかを確認 (変数ではないが存在しているオブジェクト)
			raise ConstantError("'{0}' is already reserved.".format(name))
		else:
			raise AttributeError("'{0}' does not exist. You can't delete something that doesn't exist.".format(name))
	
	def read_only_object_converter(self, value, recursion_limit=None): #list, set, dict を変更不可にする関数
		recursion_limit_buffer = None
		if recursion_limit is None:
			try:
				recursion_limit_buffer = int(self.CONSTANT_CLASS_RECURSION_LIMIT)
			except Exception: #例外が発生した場合は処理を行わない
				recursion_limit_buffer = 0
		else:
			try:
				recursion_limit_buffer = int(recursion_limit)
			except Exception: #例外が発生した場合は処理を行わない
				recursion_limit_buffer = 0
		try:
			if recursion_limit_buffer < 1:
				raise RecursionError("'recursion_limit's value is zero.")
			if type(value) in (dict, MappingProxyType):
				value_buffer = dict()
				for k, v in dict(value).items():
					value_buffer[k] = self.read_only_object_converter(v, recursion_limit=recursion_limit_buffer-1)
				return MappingProxyType(value_buffer)
			elif type(value) in (set, frozenset):
				value_buffer = set()
				for v in list(value):
					value_buffer.add(self.read_only_object_converter(v, recursion_limit=recursion_limit_buffer-1))
				return frozenset(value_buffer)
			elif type(value) in (list, tuple):
				value_buffer = list()
				for v in list(value):
					value_buffer.append(self.read_only_object_converter(v, recursion_limit=recursion_limit_buffer-1))
				return tuple(value_buffer)
			else:
				return value
		except RecursionError:
			return value

	def read_only_object_reverse_converter(self, value, recursion_limit=None): #tuple, frozenset, MappingProxyType を変更可能にする関数
		recursion_limit_buffer = None
		if recursion_limit is None:
			try:
				recursion_limit_buffer = int(self.CONSTANT_CLASS_RECURSION_LIMIT)
			except Exception: #例外が発生した場合は処理を行わない
				recursion_limit_buffer = 0
		else:
			try:
				recursion_limit_buffer = int(recursion_limit)
			except Exception: #例外が発生した場合は処理を行わない
				recursion_limit_buffer = 0
		try:
			if recursion_limit_buffer < 1:
				raise RecursionError("'recursion_limit's value is zero.")
			if type(value) in (dict, MappingProxyType):
				value_buffer = dict()
				for k, v in dict(value).items():
					value_buffer[k] = self.read_only_object_reverse_converter(v, recursion_limit=recursion_limit_buffer-1)
				return dict(value_buffer)
			elif type(value) in (set, frozenset):
				value_buffer = set()
				for v in list(value):
					value_buffer.add(self.read_only_object_reverse_converter(v, recursion_limit=recursion_limit_buffer-1))
				return set(value_buffer)
			elif type(value) in (list, tuple):
				value_buffer = list()
				for v in list(value):
					value_buffer.append(self.read_only_object_reverse_converter(v, recursion_limit=recursion_limit_buffer-1))
				return list(value_buffer)
			else:
				return value
		except RecursionError:
			return value
	
	def get_constant(self, constant_name_list=None, ignore_mode=True): #定数の辞書を返す関数
		try:
			return_dict = {}
			for k, v in self.__dict__.items():
				if constant_name_list is None or k in constant_name_list:
					return_dict[k] = v
			return return_dict
		except BaseException as e:
			if not(ignore_mode) or type(e) == KeyboardInterrupt:
				raise
			return None
	
	def show_constant(self, constant_name_list=None, ignore_mode=True): #定数をまとめた表を表示する関数
		try:
			show_dict = self.get_constant(constant_name_list=constant_name_list, ignore_mode=False)
			show_dict_key_max_len = max([len(str(k)) for k in show_dict.keys()]) if len(show_dict) > 0 else 0
			show_dict_value_max_len = max([len(str(v)) for v in show_dict.values()]) if len(show_dict) > 0 else 0
			print("[ Constant list ({0}) ]".format(len(show_dict)))
			counter, counter_max_len = 0, len(str(len(show_dict) - 1))
			for k, v in sorted(show_dict.items(), key=lambda x:x[0]):
				v_buffer = str(v).replace("\n", "\\n")
				v_buffer = "'{0}'".format(v_buffer) if v is not None and type(v) == str else v_buffer
				print("{3:{4}d} : {0:<{2}} = {1:<{5}}".format(str(k).replace("\n", "\\n"), v_buffer, show_dict_key_max_len, counter, counter_max_len, show_dict_value_max_len))
				counter += 1
			print()
			return True
		except BaseException as e:
			if not(ignore_mode) or type(e) == KeyboardInterrupt:
				raise
			return False
	
	def get_reserved_object(self, ignore_mode=True): #予約済みオブジェクトの名前リストを返す関数
		return_list = []
		try:
			for n in list(set(dir(self)) - set(self.__dict__.keys())):
				return_list.append(n)
			return return_list
		except BaseException as e:
			if not(ignore_mode) or type(e) == KeyboardInterrupt:
				raise
			return None
	
	def show_reserved_object(self, ignore_mode=True): #予約済みオブジェクトをまとめた表を表示する関数
		try:
			show_list = self.get_reserved_object(ignore_mode=False)
			show_list_max_len = max([len(str(n)) for n in show_list]) if len(show_list) > 0 else 0
			print("[ Reserved object list ({0}) ]".format(len(show_list)))
			counter, counter_max_len = 0, len(str(len(show_list) - 1))
			for n in sorted(show_list):
				print("{1:{2}d} : {0:<{3}}".format(str(n).replace("\n", "\\n"), counter, counter_max_len, show_list_max_len))
				counter += 1
			print()
			return True
		except BaseException as e:
			if not(ignore_mode) or type(e) == KeyboardInterrupt:
				raise
			return False
	
	def overwrite_constant(self, constant_name, new_constant_value, check_for_reserved_object_flag=False, ignore_mode=True): #定数を上書きできる関数 ※非推奨
		try:
			if constant_name in set(dir(self)) and constant_name not in self.__dict__ and check_for_reserved_object_flag:
				raise ConstantError("'{0}' is already reserved.".format(constant_name))
			self.__dict__[constant_name] = new_constant_value
			return True
		except (ConstantError, KeyError) as e:
			if not(ignore_mode):
				raise
			return False
	
	def delete_constant(self, constant_name, ignore_mode=True): #定数を削除できる関数 ※非推奨
		try:
			if constant_name in set(dir(self)) and constant_name not in self.__dict__:
				raise ConstantError("'{0}' is already reserved.".format(constant_name))
			del self.__dict__[constant_name]
			return True
		except (ConstantError, KeyError) as e:
			if not(ignore_mode):
				raise
			return False
	
	def set_constant_class_recursion_limit(self, recursion_limit=None, ignore_mode=True): #変更不可能化処理の再帰上限を設定する関数
		try:
			if recursion_limit is not None:
				recursion_limit_buffer = 0 if int(recursion_limit) < 0 else int(recursion_limit)
				self.__dict__["CONSTANT_CLASS_RECURSION_LIMIT"] = recursion_limit_buffer
			return self.__dict__["CONSTANT_CLASS_RECURSION_LIMIT"]
		except (KeyError, ValueError):
			if not(ignore_mode):
				raise
			return None

CONST = Constant() #定数クラスをインスタンス化

埋めたい
def padding_text_list(text_list, to_length=None, padding_chr=" ", right_mode=False):
	if len(text_list) < 1:
		return list()
    padding_chr_true = " " if padding_chr is None or len(str(padding_chr)) < 1 else str(padding_chr)[0]
    max_len = max([len(str(text)) for text in text_list]) if to_length is None else int(to_length)
    if right_mode:
        return [str(text).rstrip(padding_chr_true).rjust(max_len, padding_chr_true) for text in text_list]
    else:
        return [str(text).lstrip(padding_chr_true).ljust(max_len, padding_chr_true) for text in text_list]

ソートからの変換器生成
def create_convert_dict(data_2d_list, key_function, value_function, sort_key_function=None):
    # ソートが必要な場合は指定された関数でソート
    if sort_key_function:
        data_2d_list = sorted(data_2d_list, key=sort_key_function)

    # 辞書の生成（上書き対応）
    result_dict = {key_function(item): value_function(item) for item in data_2d_list}

    return result_dict

# 使用例
dict_of_key_value = create_convert_dict(
    data_2d_list=[["1", "A", "う"], ["2", "B", "い"], ["2", "C", "あ"]],
    key_function=lambda x: x[0],
    value_function=lambda x: x[2],
    sort_key_function=lambda x: (x[0], x[1]),
)

固定長ファイル変換その１（骨格）
import unicodedata

def get_display_width(text):
    return sum([2 if unicodedata.east_asian_width(ch) in ("A", "F", "W") else 1 for ch in text]) if len(text) > 0 else 0

class Fixed_length_file_converter:
    def __init__(self, config_dict_list):
        self.config_dict_list = []
        self.record_size = 0
        counter = 1
        for cd in config_dict_list:
            if "SIZE" in cd and "NAME" in cd:
                func = cd.get("FUNCTION", None)
                cd_n = {
                    "NUMBER": counter,
                    "NAME": cd["NAME"],
                    "SIZE": cd["SIZE"],
                    "FUNCTION": func
                }
                self.config_dict_list.append(cd_n)
                self.record_size += cd["SIZE"]
            else:
                raise Exception("The config dictionary list must have 'NAME' and 'SIZE' parameters.")
            counter += 1

    def print_config(self):
        headers, col_widths = ["NUMBER", "NAME", "SIZE", "FUNCTION"]. [8, 20, 6, 20]
        # Head
        line = ""
        for header, width in zip(headers, col_widths):
            line += header.ljust(width)
        print(line)
        print("-" * sum(col_widths))
        
        # Body
        for cd in self.config_dict_list:
            line = ""
            line += str(cd["NUMBER"]).ljust(col_widths[0])
            line += str(cd["NAME"]).ljust(col_widths[1])
            line += str(cd["SIZE"]).ljust(col_widths[2])
            func_name = cd["FUNCTION"].__name__ if cd["FUNCTION"] else "None"
            line += func_name.ljust(col_widths[3])
            print(line)

    def read(self, path):
	    records = []
	    try:
	        with open(path, mode="rb") as bfr:
	            data = bfr.read()
	            total_size = len(data)
	            
	            # 余りチェック
	            remainder = total_size % self.record_size
	            if remainder != 0:
	                raise ValueException(
	                	"{0} bytes leftover at the end of file. (total: {1} bytes, record_size: {2})".format(
	                		remainder,
	                		total_size,
	                		self.record_size
	                	}
	                )
	            
	            for i in range(0, total_size - remainder, self.record_size):
	                record_data = data[i:i + self.record_size]
	                record = []
	                offset = 0
	                for field in self.config_dict_list:
	                    size = field["SIZE"]
	                    field_data = record_data[offset:offset + size]
	                    decoded = field_data
	                    if field["FUNCTION"]:
	                        decoded = field["FUNCTION"](field_data)
	                    record.append(decoded)
	                    offset += size
	                records.append(record)
	    except Exception as e:
	        raise Exception("Failed to read file: [{0}]: {1}".format(type(e).__name__, str(e)))
	    return records

固定長ファイル変換その２
import unicodedata

def get_display_width(text):
	return sum([2 if unicodedata.east_asian_width(ch) in ("A", "F", "W") else 1 for ch in text]) if len(text) > 0 else 0

def convert_1byte_ebcdic(bytes): #EBCDIC変換
	return bytes.decode("cp930")

def convert_1byte_pac10(bytes):
	#PAC10進数（Packed Decimal）を整数に変換する関数
	data_list = list()
	for byte in bytes:
		data_list.append((byte & 0b11110000) >> 4)
		data_list.append(byte & 0b00001111)
	for s in data_list[:-1]:
		if not(0b0000 <= s and s <= 0b1001):
			raise ValueError("'{0}' is not a valid PAC decimal number.".format(s))
	sign = "+"
	if data_list[-1] not in (0b1100, 0b1101):
		raise ValueError("'{0}' is not a valid PAC sign.".format(data_list[-1]))
	elif data_list[-1] == 0b1101:
		sign = "-"
	result_string = "".join([str(s) for s in data_list[:-1]])
	return int("{0}{1}".format(sign, result_string))

def convert_2byte_jef(bytes): #JEF変換
	#【2バイト文字のみ前提】JEFのバイト列をJISコード経由でPythonのUnicode文字列に変換する
	result = []
	i = 0
	while i < len(bytes):
		if i + 1 >= len(bytes):
			break  # 奇数個だったら無視
		b1 = bytes[i]
		b2 = bytes[i + 1]
		jef_code = (b1 << 8) | b2

		# JISコードに戻す
		jis_code = jef_code - 0x8080
		high = (jis_code >> 8) & 0xFF
		low = jis_code & 0xFF

		# JISバイト列作成
		jis_bytes = bytes([high, low])
		try:
			# JISコードを一度Unicodeに変換する
			char = jis_bytes.decode("iso2022_jp")
			result.append(char)
		except Exception:
			result.append("？")  #デコードエラー
		i += 2

	return "".join(result)


class Fixed_length_file_converter:
	def __init__(self, config_dict_list):
		self.config_dict_list = []
		self.record_size = 0
		counter = 1
		for cd in config_dict_list:
			if "SIZE" in cd and "NAME" in cd:
				func = cd.get("FUNCTION", None)
				cd_n = {
					"NUMBER": counter,
					"NAME": cd["NAME"],
					"SIZE": cd["SIZE"],
					"FUNCTION": func
				}
				self.config_dict_list.append(cd_n)
				self.record_size += cd["SIZE"]
			else:
				raise Exception("The config dictionary list must have 'NAME' and 'SIZE' parameters.")
			counter += 1

	def print_config(self):
		headers, col_widths = ["NUMBER", "NAME", "SIZE", "FUNCTION"], [8, 20, 6, 20]
		# Head
		line = ""
		for header, width in zip(headers, col_widths):
			line += header.ljust(width)
		print(line)
		print("-" * sum(col_widths))
		
		# Body
		for cd in self.config_dict_list:
			line = ""
			line += str(cd["NUMBER"]).ljust(col_widths[0])
			line += str(cd["NAME"]).ljust(col_widths[1])
			line += str(cd["SIZE"]).ljust(col_widths[2])
			func_name = cd["FUNCTION"].__name__ if cd["FUNCTION"] else "None"
			line += func_name.ljust(col_widths[3])
			print(line)

	def read(self, path):
		records = []
		try:
			with open(path, mode="rb") as bfr:
				data = bfr.read()
				total_size = len(data)
				
				# 余りチェック
				remainder = total_size % self.record_size
				if remainder != 0:
					raise ValueException(
						"{0} bytes leftover at the end of file. (total: {1} bytes, record_size: {2})".format(
							remainder,
							total_size,
							self.record_size
						)
					)
				
				for i in range(0, total_size - remainder, self.record_size):
					record_data = data[i:i + self.record_size]
					record = []
					offset = 0
					for field in self.config_dict_list:
						size = field["SIZE"]
						field_data = record_data[offset:offset + size]
						decoded = field_data
						if field["FUNCTION"]:
							decoded = field["FUNCTION"](field_data)
						record.append(decoded)
						offset += size
					records.append(record)
		except Exception as e:
			raise Exception("Failed to read file: [{0}]: {1}".format(type(e).__name__, str(e)))
		return records


if "__main__" == __name__:
	print(convert_1byte_pac10([0x08, 0x12, 0x34, 0x5D]))

#住民票コード - 「平成14年総務省告示第436号」
#チェックデジット生成方式: モジュラス11 ウェイト2～7
def get_checkdigit(source_code_string): #チェックデジット生成 (モジュラス11 ウェイト2～7)
    value_list = [int(s) for s in reversed(str(source_code_string))]
    weight_list = [i % 6 + 2 for i in range(len(source_code_string))]
    sum_value = 0
    for v, w in zip(value_list, weight_list):
        sum_value +=  v * w
    return 11 - sum_value % 11 if sum_value % 11 >= 2 else 0

Slackアーカイブシステム（g_ver=2.3）
#SlackCollector[v48-7_20250601].html

Oracle接続系をベースにしたプログラムの塊（設定削除済み）
参考となるPythonプログラムコード(OracleDB接続)の塊.html

・PostgreSQL
https://www.postgresql.org/download/windows/

・PythonでのPostgreSQLデータベースラッパー（ビルドがめんどいからBinaryでヨシ）
https://pypi.org/project/psycopg2-binary/

#SlackCollector[v48-7_20250601].zip
の分割データ

Oracle接続

Oracle接続コア部分

TIFFファイルをbase64CSVにするコード
import os, base64, csv, datetime

LOAD_DIRECTORY = os.path.join(os.path.dirname(os.path.abspath(__file__)), "DATA")
OUTPUT_FILE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "OUTPUT.csv")
IF_FUNCTION = lambda x: str(x).lower().endswith(".tiff")
ROW_FUNCTION = [
	lambda x, y : "{0:0>10s}".format(os.path.splitext(os.path.basename(str(x)))[0]), 
	lambda x, y : str(os.path.basename(x)),
	lambda x, y : str(y)
]
TITLE_LIST = ["登録番号", "ファイル名", "データ"]

def package_binary_to_2d_list(
	if_function=IF_FUNCTION, row_function=ROW_FUNCTION, load_directory_path=LOAD_DIRECTORY,
	output_file_path=OUTPUT_FILE_PATH, title_list=TITLE_LIST
):
	with open(output_file_path, "w", newline="", encoding="utf-8") as fw:
		cfw = csv.writer(fw)
		cfw.writerow(title_list)
		for file_name in os.listdir(load_directory_path):
			file_path = os.path.join(load_directory_path, file_name)
			if os.path.isfile(file_path) and (file_name):
				with open(file_path, mode="rb") as fr:
					binary_data = fr.read()
					base64_data = base64.b64encode(binary_data).decode("ascii")
					row = list()
					for f in row_function:
						row.append(f(file_path, base64_data))
					cfw.writerow(row)

if __name__ == "__main__":
	print("開始: {0}".format(datetime.datetime.now()))
	package_binary_to_2d_list(
		if_function=IF_FUNCTION, 
		row_function=ROW_FUNCTION, 
		load_directory_path=LOAD_DIRECTORY,
		output_file_path=OUTPUT_FILE_PATH, 
		title_list=TITLE_LIST
	)
	print("終了: {0}".format(datetime.datetime.now()))
	

順番に辞書入りリストを並べたい
def reorder_dict_list_with_validation(data):
	if not data:
		return []
	mynum_to_dict = {}
	nextnum_set = set()
	
	# 準備と検査：辞書作成 & 重複 mynum チェック
	for d in data:
		mynum = d.get("mynum")
		nextnum = d.get("nextnum")
		if mynum in mynum_to_dict:
			raise ValueError("重複した「自ノード」が存在します: {0}".format(mynum))
		mynum_to_dict[mynum] = d
		if nextnum is not None:
			nextnum_set.add(nextnum)
	# 始点を決定：他のノードの nextnum に登場しない mynum が始点
	start_candidates = set(mynum_to_dict.keys()) - nextnum_set
	if len(start_candidates) != 1:
		raise ValueError("始点が1つに特定できません（候補: {0}）".format(start_candidates))
	start = start_candidates.pop()
	# 順に辿る：ループや存在しない nextnum を検出
	visited, result = set(), list()
	while start is not None:
		if start in visited:
			raise ValueError("循環を検出しました: {0} が再訪されました".format(start))
		node = mynum_to_dict.get(start)
		if node is None:
			raise ValueError("存在しない「次ノード」を参照しています: {0}".format(start))
		result.append(node)
		visited.add(start)
		start = node["nextnum"]

	# 全ノードが使用されたか確認
	if len(visited) != len(data):
		unused = set(mynum_to_dict.keys()) - visited
		raise ValueError("未使用のノードが存在します: {0}".format(unused))
	return result


def test(data):
	print("-"*60)
	try:
		ordered = reorder_dict_list_with_validation(data)
		for item in ordered:
			print(item)
	except Exception as e:
		print("[{0}]: {1}".format(type(e).__name__, e))
	print()

data = [{"mynum":1, "nextnum":3}, {"mynum":2, "nextnum":1}, {"mynum":3, "nextnum":None}]
test(data)

# 循環あり
data = [{"mynum":1, "nextnum":2}, {"mynum":2, "nextnum":1}]
test(data)

# 存在しない nextnum を参照
data = [{"mynum":1, "nextnum":2}]
test(data)

# 始点が複数
data = [{"mynum":1, "nextnum":3}, {"mynum":2, "nextnum":3}, {"mynum":3, "nextnum":None}]
test(data)

# mynum 重複
data = [{"mynum":1, "nextnum":2}, {"mynum":1, "nextnum":None}]
test(data)

生成_CSV→表

Pythonを持ち運びたい
https://atuagekyouju.hatenablog.com/entry/portable-execution-environment-for-python

ログ出力
import sys, subprocess, time, os, unicodedata, traceback

def len_2b(text):
	return sum([2 if unicodedata.east_asian_width(s) in "AFW" else 1 for s in str(text)]) if len(str(text)) else 0

def ensure_directory_exists(file_path):
	directory_path = os.path.dirname(os.path.abspath(file_path))
	if not(os.path.exists(directory_path)):
		os.makedirs(directory_path, exist_ok=True)

def write_with_retry(path, content, retry_interval=10, max_retries=None, encoding="utf-8", overwrite_mode=True):
	retries, flag = 0, False
	ensure_directory_exists(path)
	while True:
		try:
			with open(path, "a" if overwrite_mode else "w", encoding=encoding) as f:
				f.write(content)
			flag = True
			break
		except (PermissionError, OSError) as e:
			retries += 1
			if max_retries is not None and retries >= max_retries:
				raise RuntimeError("Max retries reached while writing to \"{0}\": {1}".format(path, e))
			time.sleep(retry_interval)
	return flag

def main(argv):
	return_flag = 255
	if len(argv) < 3:
		print("Usage: python {0} <output_file_path> <command> [args...]".format(argv[0]))
		return_flag = 3
	else:
		output_path, command = argv[1], argv[2:]
		try:
			try:
				result = subprocess.run(command, capture_output=True, text=True)
				buffer = "#Command:".format(" ".join(command))
				content = "\n".join([buffer, "-"*len_2b(buffer), result.stdout, result.stderr])
				flag = write_with_retry(output_path, content)
				return_flag = 0 if flag else 2
			except FileNotFoundError as e:
				write_with_retry(output_path, content)
				content = "\n".join([
					"#Command  : {0}".format(" ".join(command)),
					"#Exception: [0]: {1}".format(type(e).__name__, e),
					traceback.format_exc()
				])
				print(content)
				return_flag = 4
		except Exception as e:
			content = "\n".join([
				"#Command  : {0}".format(" ".join(command)),
				"#Exception: [0]: {1}".format(type(e).__name__, e),
				traceback.format_exc()
			])
			print(content)
			return_flag = 5
	return return_flag

if __name__ == "__main__":
	return_flag = main(sys.argv)
	sys.exit(return_flag)

hex_text = "C5BEBDD0BCE8BEC3"
encode_list = [
    "ascii", "big5", "big5hkscs", "cp037", "cp273", "cp424", "cp437", "cp500", "cp720", "cp737", "cp775", 
    "cp850", "cp852", "cp855", "cp856", "cp857", "cp858", "cp860", "cp861", "cp862", "cp863", "cp864", 
    "cp865", "cp866", "cp869", "cp874", "cp875", "cp932", "cp949", "cp950", "cp1006", "cp1026", "cp1125", 
    "cp1140", "cp1250", "cp1251", "cp1252", "cp1253", "cp1254", "cp1255", "cp1256", "cp1257", "cp1258", 
    "euc_jp", "euc_jis_2004", "euc_jisx0213", "euc_kr", "gb2312", "gbk", "gb18030", "hz", "iso2022_jp", 
    "iso2022_jp_1", "iso2022_jp_2", "iso2022_jp_2004", "iso2022_jp_3", "iso2022_jp_ext", "iso2022_kr", 
    "latin_1", "iso8859_2", "iso8859_3", "iso8859_4", "iso8859_5", "iso8859_6", "iso8859_7", "iso8859_8", 
    "iso8859_9", "iso8859_10", "iso8859_11", "iso8859_13", "iso8859_14", "iso8859_15", "iso8859_16", 
    "johab", "koi8_r", "koi8_t", "koi8_u", "kz1048", "mac_cyrillic", "mac_greek", "mac_iceland", 
    "mac_latin2", "mac_roman", "mac_turkish", "ptcp154", "shift_jis", "shift_jis_2004", "shift_jisx0213", 
    "utf_32", "utf_32_be", "utf_32_le", "utf_16", "utf_16_be", "utf_16_le", "utf_7", "utf_8"
]
print("変換前: {0}".format(hex_text))
print("-"*60)
for en in encode_list:
	print("変換後: {0:<15s}: {1}".format(en, bytes.fromhex(hex_text).decode(en, errors="replace")))

ジョブネット可視化？
import json
from pathlib import Path

def export_custom_graph_html(nodes, edges, html_path="graph.html"):
    html_template = """<!DOCTYPE html>
<html lang="ja">
<head>
  <meta charset="UTF-8">
  <title>Systemwalkerジョブネット可視化</title>
  <script src="https://unpkg.com/cytoscape@3.26.0/dist/cytoscape.min.js"></script>
  <style>
    html, body {{
      margin: 0;
      padding: 0;
      height: 100%;
      overflow: hidden;
      font-family: sans-serif;
    }}
    #container {{
      display: flex;
      height: 100%;
    }}
    #cy {{
      flex: 1;
      background: #FFFFFF;
    }}
    #info {{
      width: 300px;
      background: #F7F7F7;
      border-left: 1px solid #ccc;
      padding: 10px;
      overflow-y: auto;
      font-size: 13px;
    }}
    #info h2 {{
      font-size: 16px;
      border-bottom: 1px solid #ccc;
      margin-bottom: 5px;
    }}
    table {{
      width: 100%;
      border-collapse: collapse;
    }}
    td {{
      padding: 4px 6px;
      border-bottom: 1px solid #ddd;
    }}
    td.key {{
      font-weight: bold;
      width: 40%;
      color: #333;
    }}
  </style>
</head>
<body>
<div id="container">
  <div id="cy"></div>
  <div id="info">
    <h2>ノード情報</h2>
    <p>ノードをクリックすると属性が表示されます。</p>
  </div>
</div>
<script>
const elements = {data};

const cy = cytoscape({{
  container: document.getElementById('cy'),
  elements: elements,
  layout: {{
    name: 'preset',
    padding: 50
  }},
  style: [
    {{
      selector: 'node',
      style: {{
        'label': 'data(label)',
        'text-valign': 'center',
        'text-halign': 'center',
        'font-size': 12,
        'color': '#FFFFFF',
        'width': 'label',
        'padding': '8px',
        'shape': 'roundrectangle',
        'background-color': ele => {{
          if (ele.data('type') === 'message') return '#FF9900';
          else if (ele.data('type') === 'jobnet') return '#3C8DBC';
          else if (ele.data('type') === 'group') return 'transparent';
          else return '#4CAF50';
        }},
      }}
    }},
    {{
      selector: 'node[type="group"]',
      style: {{
        'shape': 'roundrectangle',
        'background-opacity': 0.05,
        'border-color': '#888',
        'border-width': 2,
        'label': 'data(label)',
        'font-size': 16,
        'text-valign': 'top',
        'text-halign': 'center'
      }}
    }},
    {{
      selector: '$node > node',
      style: {{
        'padding': '10px'
      }}
    }},
    {{
      selector: 'edge',
      style: {{
        'width': 2,
        'line-color': '#888',
        'target-arrow-shape': 'triangle',
        'target-arrow-color': '#888',
        'line-style': ele => ele.data('type') === 'message' ? 'dashed' : 'solid',
        'label': 'data(label)',
        'font-size': 10,
        'color': '#333',
        'curve-style': 'bezier'
      }}
    }}
  ]
}});

// クリックされたノードの情報を表示
cy.on('tap', 'node', function(evt) {{
  const data = evt.target.data();
  const info = document.getElementById('info');
  let html = "<h2>ノード情報</h2><table>";
  for (const [key, value] of Object.entries(data)) {{
    if (key != 'id') {{
      let true_key = String(key).replace('label', '名称')
      html += '<tr><td class="key">' + true_key + '</td><td>' + value + '</td></tr>';
    }}
  }}
  html += "</table>";
  info.innerHTML = html;
}});

// 位置保存ログ（ドラッグ時）
cy.on('dragfree', function(evt) {{
  let positions = cy.nodes().map(n => ({{
    id: n.id(),
    position: n.position()
  }}));
  console.log("ノード位置情報:", JSON.stringify(positions, null, 2));
}});
</script>
</body>
</html>
"""
    Path(html_path).write_text(html_template.format(
    	data=json.dumps(nodes+edges, ensure_ascii=False, indent=2)
    ), encoding="utf-8")
    print(f":white_check_mark: HTML出力完了: {html_path}")



nodes = [
    {"data": {"id": "J001", "label": "ジョブ1", "属性":"G"}, "position": {"x": 100, "y": 200}},
    {"data": {"id": "J002", "label": "ジョブ2"}, "position": {"x": 300, "y": 200}},
    {"data": {"id": "MSG01", "label": ":envelope_with_arrow: メッセージA", "type": "message"}, "position": {"x": 200, "y": 50}},
]
edges = [
    {"data": {"source": "J001", "target": "J002", "label": "依存"}},
    {"data": {"source": "MSG01", "target": "J002", "label": "起動", "type": "message"}},
]

export_custom_graph_html(nodes, edges, html_path="graph.html")

ノード相対座標を絶対座標に再定義することが出来るのでは？
import random

# 子ノードの定義（相対座標）
# 各親IDごとに子ノードのリストを定義
parents = {
    "P1": [{"id": "C1", "x": 0, "y": 0}, {"id": "C2", "x": 100, "y": 50}],
    "P2": [{"id": "C3", "x": 0, "y": 0}, {"id": "C4", "x": 150, "y": 100}],
    "P3": [{"id": "C5", "x": 0, "y": 0}, {"id": "C6", "x": 50, "y": 50}]
}

# 各親のバウンディングボックス取得
def compute_parent_size(children, margin=20):
    min_x = min(child["x"] for child in children)
    max_x = max(child["x"] for child in children)
    min_y = min(child["y"] for child in children)
    max_y = max(child["y"] for child in children)
    width = max_x - min_x + margin * 2
    height = max_y - min_y + margin * 2
    return width, height

# 親ノードの配置（グリッド風）
def layout_parents(parents, margin=50):
    positions = {}
    x_offset = 0
    y_offset = 0
    row_height = 0
    max_row_width = 800

    for pid, children in parents.items():
        width, height = compute_parent_size(children)
        if x_offset + width > max_row_width:
            x_offset = 0
            y_offset += row_height + margin
            row_height = 0
        positions[pid] = (x_offset, y_offset)
        x_offset += width + margin
        row_height = max(row_height, height)
    return positions

# 子ノードの絶対座標を計算
def compute_absolute_positions(parents, parent_positions, margin=20):
    absolute = []
    for pid, children in parents.items():
        parent_x, parent_y = parent_positions[pid]
        for child in children:
            abs_x = parent_x + child["x"] + margin
            abs_y = parent_y + child["y"] + margin
            absolute.append({
                "id": child["id"],
                "parent": pid,
                "x": abs_x,
                "y": abs_y
            })
    return absolute

# 実行
parent_positions = layout_parents(parents)
absolute_children = compute_absolute_positions(parents, parent_positions)

# 出力確認
for node in absolute_children:
    print(f"Node {node['id']} (parent {node['parent']}) => absolute x={node['x']}, y={node['y']}")

座標制約を決定瀬瀬々

https://pypi.org/project/geopandas/

https://pypi.org/project/pillow/

https://pypi.org/project/matplotlib/

出典：「国土数値情報（行政区域データ）」（国土交通省）（https://nlftp.mlit.go.jp/ksj/gml/datalist/KsjTmplt-N03-v3_0.html）（2025年08月04日取得）を加工して作成

市区町村描画

警告：メルカトル図表のため、都道府県を中央に置くことでゆがみを軽減する

モード対応完全版

独立ファイル構成チェック機構

独立ファイル構成チェック機構（interpreter除外）
Inspecter_noninterpreter.html

独立ファイル構成チェック機構(バッチ修正)
Inspecter.html
Inspecter_NoI.html

Ver. 1.3.1
Inspecter_1_3_1(1).html
Inspecter_1_3_1.html

独立ファイル構成チェック機構: Ver:1.5.0
Inspecter_1_5_0.html
Inspecter_1_5_0_NonInterpreter.html
Inspecter_1_5_0.zip
Inspecter_1_5_0_NonInterpreter.zip

独立ファイル構成チェック機構: Ver:1.6.0（プロセスに比較辞書を渡す部分を高速化）
Inspecter_1_6_0_NonInterpreter.zip
Inspecter_1_6_0_NonInterpreter.html

独立ファイル構成チェック機構: Ver:1.6.1（設定変更無視フラグの宣伝性強化）

＜バグによりこのバージョンを破棄＞

独立ファイル構成チェック機構: Ver:1.6.4（設定変更無視フラグの宣伝性強化）
Inspecter_1_6_4_NonInterpreter.zip
Inspecter_1_6_4.zip
Inspecter_1_6_4.html
Inspecter_1_6_4_NonInterpreter.html

独立ファイル構成チェック機構: Ver:1.6.5（PermissionError時にエラーが波及する問題があったため、修正）
Inspecter_1_6_5_NonInterpreter.html
Inspecter_1_6_5.html
Inspecter_1_6_5.zip
Inspecter_1_6_5_NonInterpreter.zip

sample json

Ver: 1.6.6
Inspecter_1_6_6.zip
Inspecter_1_6_6_NonInterpreter.zip
Inspecter_1_6_6.html
Inspecter_1_6_6_NonInterpreter.html

V10　CSV系列
import csv

def read_csv(path, encoding="utf-8"):
    with open(path, mode="r", encoding=encoding, newline="") as f:
        reader = csv.reader(f)
        return list(reader)

def write_csv(path, rows, encoding="utf-8"):
    with open(path, mode="w", encoding=encoding, newline="") as f:
        writer = csv.writer(f, quoting=csv.QUOTE_ALL, lineterminator="\n")
        writer.writerows(rows)

画像の自動結合ツール開発。
import os

os.chdir(os.path.dirname(os.path.abspath(__file__)))

import cv2
import numpy as np
from pathlib import Path
from collections import defaultdict

# ==============================
# ディレクトリ設定
# ==============================
INPUT_ROOT = Path("./INPUT")
OUTPUT_ROOT = Path("./OUTPUT")

# ==============================
# パラメータ設定（要調整）
# ==============================
TOP_IGNORE_PX = 80        # 下側画像の上部固定領域
BOTTOM_IGNORE_PX = 80     # 上側画像の下部固定領域

MIN_OVERLAP_PX = 30
MAX_OVERLAP_PX = 500

MAX_IDENTICAL_RATIO = 0.01  # 完全一致行がこれ以上なら無効


# ==============================
# 完全一致行の割合を計算
# ==============================
def identical_row_ratio(a, b):
    identical = 0
    h = a.shape[0]
    for i in range(h):
        if np.array_equal(a[i], b[i]):
            identical += 1
    return identical / h if h > 0 else 1.0


# ==============================
# 重なり検知
# ==============================
def detect_overlap(img_upper, img_lower):
    h1 = img_upper.shape[0]
    h2 = img_lower.shape[0]

    gray1 = cv2.cvtColor(img_upper, cv2.COLOR_BGR2GRAY)
    gray2 = cv2.cvtColor(img_lower, cv2.COLOR_BGR2GRAY)

    upper_end = h1 - BOTTOM_IGNORE_PX
    lower_start = TOP_IGNORE_PX

    best_score = -1.0
    best_overlap = 0

    max_search = min(
        MAX_OVERLAP_PX,
        upper_end,
        h2 - lower_start
    )

    for overlap in range(MIN_OVERLAP_PX, max_search):
        part1 = gray1[upper_end - overlap : upper_end]
        part2 = gray2[lower_start : lower_start + overlap]

        # 完全一致行が多すぎる overlap は除外
        if identical_row_ratio(part1, part2) > MAX_IDENTICAL_RATIO:
            continue

        score = cv2.matchTemplate(
            part1, part2, cv2.TM_CCOEFF_NORMED
        )[0][0]

        if score > best_score:
            best_score = score
            best_overlap = overlap

    return best_overlap


# ==============================
# 画像結合
# ==============================
def stitch_images(images):
    result = images[0]

    for next_img in images[1:]:
        overlap = detect_overlap(result, next_img)

        if overlap > 0:
            upper = result[:-overlap, :]
            lower = next_img[overlap:, :]
            result = np.vstack([upper, lower])
        else:
            # overlap が検出できなかった場合は単純結合
            result = np.vstack([result, next_img])

    return result


# ==============================
# ディレクトリ単位処理
# ==============================
def process_directory(input_dir: Path):
    groups = defaultdict(list)

    for bmp in sorted(input_dir.glob("*.bmp")):
        key = bmp.name[0]
        groups[key].append(bmp)

    output_dir = OUTPUT_ROOT / input_dir.name
    output_dir.mkdir(parents=True, exist_ok=True)

    for key, files in groups.items():
        images = [cv2.imread(str(f)) for f in files]

        if any(img is None for img in images):
            raise RuntimeError(f"画像読み込み失敗: {files}")

        stitched = stitch_images(images)
        output_path = output_dir / f"{key}.bmp"
        cv2.imwrite(str(output_path), stitched)

        print(f"[OK] 出力: {output_path}")


# ==============================
# メイン処理
# ==============================
def main():
    OUTPUT_ROOT.mkdir(exist_ok=True)

    for subdir in INPUT_ROOT.iterdir():
        if subdir.is_dir():
            print(f"処理中: {subdir}")
            process_directory(subdir)


if __name__ == "__main__":
    main()

プロトタイプCSV解析機
import logging
import openpyxl
from openpyxl.styles import PatternFill, Font, Border, Side
from openpyxl.cell.text import InlineFont
from openpyxl.cell.rich_text import TextBlock, CellRichText
from openpyxl.utils import get_column_letter
from datetime import datetime
from typing import List, Any, Callable, Tuple, Set, Dict, Optional

class ExcelDataProcessor:
    def __init__(
        self,
        data_2d_list: List[List[Any]],
        output_excel_file_path: str,
        output_log_file_path: str,
        output_summary_file_path: str,
        header_list: Optional[List[str]] = None,
        output_excel_design_row_mode: str = 'NONE', 
        output_excel_design_column_mode: str = 'NONE',
        filter_function: Callable[[Any], str] = lambda x: str(x).strip() if x is not None else "",
        check_function: Callable[[str], Tuple[bool, Set[int]]] = lambda x: (False, set()),
        target_cell_color: str = "FFFF00", # 対象セル/行の色 (Yellow)
        target_char_color: str = "FF0000", # 対象文字の色 (Red)
        header_default_color: str = "D9D9D9", # 【追加】ヘッダーのデフォルト背景色 (Gray)
        no_column_default_color: str = "E0E0E0" # 【追加】No.列のデフォルト背景色 (Light Gray)
    ):
        self.raw_data = data_2d_list
        self.excel_path = output_excel_file_path
        self.log_path = output_log_file_path
        self.summary_path = output_summary_file_path
        self.header_list = header_list
        self.row_mode = output_excel_design_row_mode.upper()
        self.col_mode = output_excel_design_column_mode.upper()
        self.filter_func = filter_function
        self.check_func = check_function
        
        # 色設定 (#除去)
        self.target_bg_color = target_cell_color.replace("#", "")
        self.target_font_color = target_char_color.replace("#", "")
        self.header_default_bg = header_default_color.replace("#", "")
        self.no_col_default_bg = no_column_default_color.replace("#", "")
        
        self._setup_logger()
        
        self.analysis_result = [] 
        self.target_cols_indices = set() 
        self.summary_stats = {"total_rows": 0, "target_rows": 0, "total_targets_found": 0}

    def _setup_logger(self):
        logging.basicConfig(
            filename=self.log_path,
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            force=True
        )
        self.logger = logging.getLogger(__name__)

    def _prepare_data(self):
        if self.header_list:
            headers = self.header_list
            data_rows = self.raw_data
        else:
            if not self.raw_data:
                raise ValueError("データが存在しません。")
            headers = self.raw_data[0]
            data_rows = self.raw_data[1:]
        return headers, data_rows

    def analyze(self):
        self.logger.info("解析を開始します。")
        headers, data_rows = self._prepare_data()
        
        processed_rows = []
        for r_idx, row in enumerate(data_rows):
            row_info = {
                "original_row_idx": r_idx,
                "has_target": False,
                "cells": []
            }
            
            for c_idx, cell_raw in enumerate(row):
                filtered_val = self.filter_func(cell_raw)
                is_target, target_indices = self.check_func(filtered_val)
                
                cell_info = {
                    "value": filtered_val,
                    "is_target": is_target,
                    "target_indices": target_indices
                }
                row_info["cells"].append(cell_info)
                
                if is_target:
                    row_info["has_target"] = True
                    self.target_cols_indices.add(c_idx)
                    self.summary_stats["total_targets_found"] += 1
            
            processed_rows.append(row_info)
            if row_info["has_target"]:
                self.summary_stats["target_rows"] += 1

        self.summary_stats["total_rows"] = len(processed_rows)
        self.headers = headers
        self.processed_data = processed_rows
        self.logger.info(f"解析完了: {self.summary_stats}")

    def _create_rich_text(self, text: str, target_indices: Set[int]) -> CellRichText:
        """【修正済み】InlineFontを使用したリッチテキスト生成"""
        if not target_indices:
            return CellRichText(text)
            
        rich_parts = []
        default_font = InlineFont(color="000000") 
        target_font = InlineFont(color=self.target_font_color, b=True)
        
        current_str = ""
        current_is_target = -1 
        
        for i, char in enumerate(text):
            is_target_char = 1 if i in target_indices else 0
            if current_is_target == -1:
                current_is_target = is_target_char
                current_str += char
            elif current_is_target == is_target_char:
                current_str += char
            else:
                font_to_use = target_font if current_is_target == 1 else default_font
                rich_parts.append(TextBlock(font_to_use, current_str))
                current_str = char
                current_is_target = is_target_char
        
        if current_str:
            font_to_use = target_font if current_is_target == 1 else default_font
            rich_parts.append(TextBlock(font_to_use, current_str))

        return CellRichText(rich_parts)

    def export_excel(self):
        self.logger.info("Excel出力を開始します。")
        wb = openpyxl.Workbook()
        ws = wb.active
        ws.title = "CheckResult"
        
        # --- スタイル定義 ---
        # 1. 塗りつぶし
        fill_target = PatternFill(start_color=self.target_bg_color, end_color=self.target_bg_color, fill_type="solid")
        fill_header_target = PatternFill(start_color="FFD700", end_color="FFD700", fill_type="solid") # 列ターゲット(Gold)
        
        # 【追加】デフォルト背景色
        fill_header_default = PatternFill(start_color=self.header_default_bg, end_color=self.header_default_bg, fill_type="solid")
        fill_no_col_default = PatternFill(start_color=self.no_col_default_bg, end_color=self.no_col_default_bg, fill_type="solid")
        
        # 【追加】枠線（格子線）
        thin_side = Side(style='thin', color="000000")
        border_all = Border(left=thin_side, right=thin_side, top=thin_side, bottom=thin_side)

        # --- 列・行のフィルタリング ---
        output_col_indices = []
        for i in range(len(self.headers)):
            if self.col_mode == 'DELETE':
                if i in self.target_cols_indices:
                    output_col_indices.append(i)
            else:
                output_col_indices.append(i)
        
        rows_to_write = []
        for row_info in self.processed_data:
            if self.row_mode == 'DELETE' and not row_info["has_target"]:
                continue
            rows_to_write.append(row_info)

        # --- ヘッダー書き込み ---
        # A1: No.列ヘッダー
        cell_no = ws.cell(row=1, column=1, value="No.")
        cell_no.fill = fill_header_default
        cell_no.border = border_all
        
        excel_col_idx = 2
        for original_col_idx in output_col_indices:
            cell = ws.cell(row=1, column=excel_col_idx, value=str(self.headers[original_col_idx]))
            
            # 色設定：ターゲット列ならGold、それ以外はデフォルトヘッダー色
            if original_col_idx in self.target_cols_indices:
                cell.fill = fill_header_target
            else:
                cell.fill = fill_header_default
            
            # 枠線適用
            cell.border = border_all
            
            # 列非表示設定 (HIDEモード)
            if self.col_mode == 'HIDE' and original_col_idx not in self.target_cols_indices:
                col_letter = get_column_letter(excel_col_idx)
                ws.column_dimensions[col_letter].hidden = True
                
            excel_col_idx += 1

        # --- データ書き込み ---
        current_excel_row = 2
        for row_info in rows_to_write:
            # 1. No.列
            no_cell = ws.cell(row=current_excel_row, column=1, value=row_info["original_row_idx"] + 1)
            
            # 色設定：ターゲット行ならYellow、それ以外はデフォルトNo列色
            if row_info["has_target"]:
                no_cell.fill = fill_target
            else:
                no_cell.fill = fill_no_col_default
            
            no_cell.border = border_all
            
            # 行非表示設定 (HIDEモード)
            if self.row_mode == 'HIDE' and not row_info["has_target"]:
                ws.row_dimensions[current_excel_row].hidden = True

            # 2. 実データ列
            excel_col_idx = 2
            for original_col_idx in output_col_indices:
                if original_col_idx < len(row_info["cells"]):
                    cell_data = row_info["cells"][original_col_idx]
                    cell = ws.cell(row=current_excel_row, column=excel_col_idx)
                    
                    # リッチテキスト
                    if cell_data["is_target"] and cell_data["target_indices"]:
                        cell.value = self._create_rich_text(cell_data["value"], cell_data["target_indices"])
                    else:
                        cell.value = cell_data["value"]
                    
                    # 背景色
                    if cell_data["is_target"]:
                        cell.fill = fill_target
                    
                    # 枠線
                    cell.border = border_all
                
                excel_col_idx += 1
            
            current_excel_row += 1

        try:
            wb.save(self.excel_path)
            self.logger.info(f"Excel保存成功: {self.excel_path}")
        except Exception as e:
            self.logger.error(f"Excel保存失敗: {e}")
            raise

    def export_summary(self):
        try:
            with open(self.summary_path, 'w', encoding='utf-8') as f:
                f.write("=== Summary Report ===\n")
                f.write(f"Executed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write(f"Total Rows Processed: {self.summary_stats['total_rows']}\n")
                f.write(f"Target Rows Found: {self.summary_stats['target_rows']}\n")
                f.write(f"Total Target Cells/Hits: {self.summary_stats['total_targets_found']}\n")
            self.logger.info(f"集計ファイル保存成功: {self.summary_path}")
        except Exception as e:
            self.logger.error(f"集計ファイル保存失敗: {e}")

    def execute(self):
        self.analyze()
        self.export_excel()
        self.export_summary()
        print(f"処理完了。ログを確認してください: {self.log_path}")

# --- 実行例 ---
if __name__ == "__main__":
    def my_filter(val):
        return str(val).strip() if val is not None else ""

    def my_check(text):
        target = "Error"
        indices = set()
        if target in text:
            # 簡易的な検索（重複考慮なし版）
            start = text.find(target)
            while start != -1:
                for i in range(start, start + len(target)):
                    indices.add(i)
                start = text.find(target, start + 1)
            return True, indices
        return False, set()

    raw_data = [
        ["ID", "Log Message", "Status"],
        [1, "System started", "OK"],
        [2, "Connection Error occurred", "Fail"],
        [3, "Process completed", "OK"],
        [4, "Unknown Error code 99", "Fail"]
    ]

    processor = ExcelDataProcessor(
        data_2d_list=raw_data,
        output_excel_file_path="result_v2.xlsx",
        output_log_file_path="process_v2.log",
        output_summary_file_path="summary_v2.txt",
        header_default_color="#B9B9B9", # ヘッダーのグレー
        no_column_default_color="#E0E0E0", # No列の薄いグレー
        target_cell_color="#FFFF00", # ターゲットセルの黄色
        filter_function=my_filter,
        check_function=my_check
    )

    processor.execute()

hex(ord("あ")).lstrip("0x")

基底ロジック

import os
import shutil
import hashlib
import re
import logging

# ==========================================
# 設定・定数
# ==========================================
class ExitCode:
	SUCCESS = 0
	ERR_NOT_FOUND = 11
	ERR_HASH_CALC = 12
	ERR_VERIFY_FAIL = 20
	ERR_CLEANUP_FAIL = 31
	ERR_CRITICAL = 32

class TransferMode(Enum):
	MOVE = 'move'  # 成功時に元ファイルをBackupへ移動
	COPY = 'copy'  # 成功時に元ファイルを残す（またはBackupへコピー）

# ==========================================
# ユーティリティ関数
# ==========================================
def calculate_file_hash(filepath, algorithm="sha256"):
	"""ファイルのハッシュ値を計算する"""
	hash_func = hashlib.new(algorithm)
	chunk_size = 1024 * 8
	try:
		with open(filepath, "rb") as f:
			while chunk := f.read(chunk_size):
				hash_func.update(chunk)
		return hash_func.hexdigest()
	except Exception as e:
		logging.error(f"ハッシュ計算エラー: {filepath} - {e}")
		return None

def setup_logger(log_dir):
	"""ログ設定"""
	if not os.path.exists(log_dir):
		os.makedirs(log_dir)
	
	timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
	log_file = os.path.join(log_dir, f"transfer_{timestamp}.log")
	
	logging.basicConfig(
		level=logging.INFO,
		format='%(asctime)s [%(levelname)s] %(message)s',
		handlers=[
			logging.FileHandler(log_file, encoding='utf-8'),
			logging.StreamHandler()
		]
	)
	return log_file

# ==========================================
# メイン処理クラス
# ==========================================
class FileTransferProcessor:
	def __init__(self, input_dir, file_regex, output_dir, mode, log_dir):
		self.input_dir = input_dir
		self.file_pattern = re.compile(file_regex)
		self.output_dir = output_dir
		self.mode = mode
		
		# Backup/Errorフォルダの設定（図のStep7, 8に基づく）
		self.backup_dir = os.path.join(input_dir, 'backup')
		self.error_dir = os.path.join(output_dir, 'error')
		
		# ログ初期化
		setup_logger(log_dir)

	def _ensure_dirs(self):
		"""必要なディレクトリの作成"""
		os.makedirs(self.output_dir, exist_ok=True)
		os.makedirs(self.backup_dir, exist_ok=True)
		os.makedirs(self.error_dir, exist_ok=True)

	def process_single_file(self, filename):
		source_path = os.path.join(self.input_dir, filename)
		dest_path = os.path.join(self.output_dir, filename)
		
		logging.info(f"--- 処理開始: {filename} ---")

		# -------------------------------------------------
		# ① 対象ファイルの存在を確認する
		# -------------------------------------------------
		if not os.path.exists(source_path):
			logging.error(f"Step1: ファイルが存在しません: {source_path}")
			return ExitCode.ERR_NOT_FOUND

		# -------------------------------------------------
		# ② 対象ファイルのハッシュ値記録 (Pre-Transfer)
		# -------------------------------------------------
		src_hash_pre = calculate_file_hash(source_path)
		if src_hash_pre is None:
			logging.error("Step2: 事前ハッシュ取得失敗")
			return ExitCode.ERR_HASH_CALC
		logging.info(f"Step2: 元ハッシュ(Pre): {src_hash_pre}")

		# -------------------------------------------------
		# ③ 対象ファイルのコピー
		# -------------------------------------------------
		copy_success = False
		try:
			shutil.copy2(source_path, dest_path)
			copy_success = True
			logging.info("Step3: コピー実行完了")
		except Exception as e:
			logging.error(f"Step3: コピー処理中にエラー発生: {e}")
			copy_success = False

		# -------------------------------------------------
		# ④ ②で記録したハッシュ値との照合 (Source再検証: コピー中の変更検出)
		# -------------------------------------------------
		src_hash_post = calculate_file_hash(source_path)
		source_unchanged = (src_hash_pre == src_hash_post)
		
		if not source_unchanged:
			logging.warning(f"Step4: 警告！コピー中に元ファイルが変更されました。\nPre: {src_hash_pre}\nPost:{src_hash_post}")

		# -------------------------------------------------
		# ⑤ ②で記録したハッシュ値との照合 (Dest検証: コピー成功確認)
		# -------------------------------------------------
		dest_hash = None
		transfer_valid = False
		if os.path.exists(dest_path):
			dest_hash = calculate_file_hash(dest_path)
			transfer_valid = (src_hash_pre == dest_hash)
		
		if not transfer_valid:
			logging.warning(f"Step5: 警告！転送先ファイルのハッシュが一致しません。\nSrc: {src_hash_pre}\nDst: {dest_hash}")

		# -------------------------------------------------
		# ⑥ 判定 (正常終了/差異なしを判定)
		# -------------------------------------------------
		is_success = copy_success and source_unchanged and transfer_valid
		
		if is_success:
			logging.info("Step6: 判定OK - 全整合性確認完了")
			
			# -------------------------------------------------
			# ⑦ コピー元ファイルを削除/Backupに移動 (成功時)
			# -------------------------------------------------
			try:
				if self.mode == TransferMode.MOVE:
					# Moveモード: Backupへ移動（または削除）
					shutil.move(source_path, os.path.join(self.backup_dir, filename))
					logging.info(f"Step7: 元ファイルをBackupへ移動しました: {filename}")
				else:
					# Copyモード: 何もしない（あるいは済スタンプを押すなど）
					logging.info("Step7: Copyモードのため元ファイルは保持されます")
				
				return ExitCode.SUCCESS

			except Exception as e:
				logging.error(f"Step7: 事後処理(Backup移動)でエラー: {e}")
				return ExitCode.ERR_CLEANUP_FAIL

		else:
			logging.error("Step6: 判定NG - 転送または検証に失敗しました")
			
			# -------------------------------------------------
			# ⑧ 例外処理: コピーしたファイルを削除/Errorに移動 (失敗時)
			# -------------------------------------------------
			try:
				if os.path.exists(dest_path):
					target_error_path = os.path.join(self.error_dir, filename)
					# 既に同名エラーファイルがある場合のハンドリング（上書き）
					if os.path.exists(target_error_path):
						os.remove(target_error_path)
					
					shutil.move(dest_path, target_error_path)
					logging.info(f"Step8: 不完全な転送ファイルをErrorへ移動しました: {filename}")
				else:
					logging.info("Step8:転送先ファイルが存在しないため移動処理スキップ")
				
				# エラーの原因に応じたコードを返す（ここでは簡略化して汎用エラー）
				return ExitCode.ERR_VERIFY_FAIL
			
			except Exception as e:
				logging.error(f"Step8: エラー処理中の致命的エラー: {e}")
				return ExitCode.ERR_CRITICAL

	def run(self):
		logging.info("=== ファイル転送バッチ開始 ===")
		self._ensure_dirs()
		
		# 入力ディレクトリ走査
		try:
			files = os.listdir(self.input_dir)
		except FileNotFoundError:
			logging.critical(f"入力ディレクトリが見つかりません: {self.input_dir}")
			return

		target_files = [f for f in files if self.file_pattern.match(f)]
		logging.info(f"対象ファイル数: {len(target_files)}")

		for filename in target_files:
			result = self.process_single_file(filename)
			if result == ExitCode.SUCCESS:
				logging.info(f"[{filename}] 完了: SUCCESS")
			else:
				logging.error(f"[{filename}] 終了: ERROR Code {result}")
		
		logging.info("=== ファイル転送バッチ終了 ===")

# ==========================================
# エントリーポイント
# ==========================================
if __name__ == "__main__":
	# 引数パーサーの設定
	parser = argparse.ArgumentParser(description='堅牢なファイル転送プログラム')
	
	# 入力必須項目
	parser.add_argument('--input_dir', required=True, help='入力ディレクトリパス')
	parser.add_argument('--regex', required=True, help='入力ファイル正規表現 (例: ".*\.txt")')
	parser.add_argument('--output_dir', required=True, help='出力ディレクトリパス')
	parser.add_argument('--log_dir', required=True, help='ログ出力ディレクトリ')
	
	# オプション項目
	parser.add_argument('--mode', choices=['move', 'copy'], default='copy', 
						help='動作モード (move: 元ファイルをBackupへ移動, copy: 元ファイルを維持)')

	args = parser.parse_args()

	mode_enum = TransferMode.MOVE if args.mode == 'move' else TransferMode.COPY

	processor = FileTransferProcessor(
		input_dir=args.input_dir,
		file_regex=args.regex,
		output_dir=args.output_dir,
		mode=mode_enum,
		log_dir=args.log_dir
	)
	
	processor.run()