出力
RequestHeader(Step1): {'User-Agent': 'python-requests/2.31.0', 'Accept-Encoding': 'gzip, deflate, br', 'Accept': '*/*', 'Connection': 'keep-alive'}
ResponseHeader(Step1): {'Date': 'Tue, 26 Nov 2024 13:01:41 GMT', 'Server': 'Apache', 'cache-control': 'max-age=0, private, must-revalidate', 'vary': 'Accept,Accept-Encoding', 'referrer-policy': 'strict-origin-when-cross-origin', 'x-permitted-cross-domain-policies': 'none', 'x-xss-protection': '1; mode=block', 'x-request-id': 'e89225cb-8a6f-4f5a-96b7-5ecc29522035', 'x-download-options': 'noopen', 'etag': 'W/"d186e1137efebcecd03dbed76f187228"', 'x-frame-options': 'SAMEORIGIN', 'x-content-type-options': 'nosniff', 'Status': '200 OK', 'Content-Type': 'text/html; charset=utf-8', 'Set-Cookie': '_redmine_session=a24zYWkrcER1dlE0cWdvYmtXakp3Wmh5MVV3UHNmeXN0T1lzUnI0WUhwTHl2MEpSSjQwRURQTWVoQm9mNzVxOTNTdjBmZ2dRMjdXWmZpQTJad0NxZHAvMFZOREY5b0dLY3h5amdCRllqeTdoVy83cHNGMnNSRGNFTk11VGdRbTBIOXJ5OEhiNWh3N0JheWJ0ZUZlOG1vWkdKU2NJUkZjTDBIT2NVZ25rSjVnZGpnMTJWV1phcVRNM3FYaGZoOEdJLS1lTXIwN1crdVlCeHYwR25Udml1QUNnPT0%3D--2ac961a36e84a103e4e10b6f922b512fa42c9f8e; path=/demo; httponly', 'X-Robots-Tag': 'noindex,nofollow,noarchive', 'Content-Encoding': 'gzip', 'Content-Length': '4006', 'Keep-Alive': 'timeout=4, max=100', 'Connection': 'Keep-Alive'}
StatusCode(Step1): 200
Step1: -RtbqaQr5NtAUXTJ0fdxi9JENT5gcnFW9-DmBkXPGOWwgN8RuT0x43D9G_dXwAxBCQfl3H4oss_u8Udxi8ZeGA
RequestHeader(Step2): {'User-Agent': 'python-requests/2.31.0', 'Accept-Encoding': 'gzip, deflate, br', 'Accept': '*/*', 'Connection': 'keep-alive', 'Cookie': '_redmine_session=a24zYWkrcER1dlE0cWdvYmtXakp3Wmh5MVV3UHNmeXN0T1lzUnI0WUhwTHl2MEpSSjQwRURQTWVoQm9mNzVxOTNTdjBmZ2dRMjdXWmZpQTJad0NxZHAvMFZOREY5b0dLY3h5amdCRllqeTdoVy83cHNGMnNSRGNFTk11VGdRbTBIOXJ5OEhiNWh3N0JheWJ0ZUZlOG1vWkdKU2NJUkZjTDBIT2NVZ25rSjVnZGpnMTJWV1phcVRNM3FYaGZoOEdJLS1lTXIwN1crdVlCeHYwR25Udml1QUNnPT0%3D--2ac961a36e84a103e4e10b6f922b512fa42c9f8e', 'Content-Length': '232', 'Content-Type': 'application/json'}
ResponseHeader(Step2): {'Date': 'Tue, 26 Nov 2024 13:01:41 GMT', 'Server': 'Apache', 'cache-control': 'no-cache', 'referrer-policy': 'strict-origin-when-cross-origin', 'x-permitted-cross-domain-policies': 'none', 'x-xss-protection': '1; mode=block', 'x-request-id': 'a128d7e0-5a67-4ff3-a4d4-6d614b35a1a9', 'x-download-options': 'noopen', 'x-frame-options': 'SAMEORIGIN', 'x-content-type-options': 'nosniff', 'location': 'https://my.redmine.jp/demo/projects/demo', 'Status': '302 Found', 'Content-Type': 'text/html; charset=utf-8', 'Set-Cookie': '_redmine_session=eEtRRTh0WEVKdk13ejV2T2VubzlmSDBCR3RETGI1N1Y2Wm1IdEZ3SUMrTWVRT3JrVm43SGVCL1kwZjR2QVBQOHR0TGJIbmVVQUxiOFZ6R3F1MnhpWkFwU1FWY2d6d0VJVTBTWE1DcXFLbFB3dzJXWVByVm8zbzRaRkozRkQ0ZWxtN2xydE5FdmNzQzdLRlo3NzZmdWZTUExvODdBZml5UGpBVFFQSDVGL1pMZWpZRVByUTdQVVl3dlI4am83ZktZZFRXYmxaSjlLVUJxRmVyR0N1SHBpNEFGdnRxem5FcXVUMzM4aFJrSXZxWT0tLTZCY3ZqQzZlTmQ3ZExBZGxpcHFpdVE9PQ%3D%3D--c4aaa35749baf1f8a68909df91a82af8b2f83970; path=/demo; httponly', 'X-Robots-Tag': 'noindex,nofollow,noarchive', 'Vary': 'Accept-Encoding', 'Content-Encoding': 'gzip', 'Content-Length': '20', 'Keep-Alive': 'timeout=4, max=100', 'Connection': 'Keep-Alive'}
StatusCode(Step2): 302

つまり原因は以下のどれかまたは複合
・パーセントエンコーディングのミス
・POSTパラメータのミス
・Cookieのミス

Sub SendRequest()
    Dim xmlhttp As Object
    Dim url As String
    Dim response As String, responseText As String
    Dim csrfToken As String
    Dim cookies As String
    Dim postData As String
    Dim buffer As Object
    
    ' URL設定
    Dim GET_LOGIN_URL As String
    Dim POST_LOGIN_URL As String
    GET_LOGIN_URL = "https://my.redmine.jp/demo/login"
    POST_LOGIN_URL = "https://my.redmine.jp/demo/login"
    
    ' CSRFトークン取得（GETリクエスト）
    Set buffer = GetLogin(GET_LOGIN_URL)
    
    response = buffer.getAllResponseHeaders()
    responseText = buffer.responseText
	
	
	msgbox response
	
    ' CSRFトークンとCookieを解析
    csrfToken = ParseMetaTag(responseText, "csrf-token")
    cookies = ParseCookies(response)
    
    If csrfToken <> "" Then
        ' POSTリクエストパラメータ設定
        postData = "authenticity_token=" & csrfToken & _
                   "&username=reporter" & _
                   "&password=reporter" & _
                   "&back_url=%2Fdemo%2Fprojects%2Fdemo" & _
                   "&login=ログイン"
        
        ' ログイン試行（POSTリクエスト）
        response = PostLogin(POST_LOGIN_URL, postData, cookies)
        msgbox response
    End If
End Sub

Function GetLogin(url As String) As Object
    Dim xmlhttp As Object
    Set xmlhttp = CreateObject("MSXML2.ServerXMLHTTP")
    
    ' GETリクエスト送信
    xmlhttp.Open "GET", url, False
    xmlhttp.Send
    
    ' レスポンスを返す
    GetLogin = xmlhttp
End Function

Function PostLogin(url As String, postData As String, cookies As String) As String
    Dim xmlhttp As Object
    Set xmlhttp = CreateObject("MSXML2.ServerXMLHTTP")
    
    ' POSTリクエスト送信
    xmlhttp.Open "POST", url, False
    xmlhttp.setRequestHeader "Content-Type", "application/x-www-form-urlencoded"
    if cookies <> "" then
    	xmlhttp.setRequestHeader "Cookie", cookies
    end if
    xmlhttp.Send postData
    
    ' レスポンスを返す
    PostLogin = xmlhttp.responseText
    PostLogin = xmlhttp.getAllResponseHeaders()
End Function

Function ParseMetaTag(html As String, name As String) As String
    Dim startPos As Long
    Dim endPos As Long
    Dim metaTag As String
    
    ' metaタグを検索
    startPos = InStr(html, "<meta name=""" & name & """ content=""")
    If startPos > 0 Then
        startPos = startPos + Len("<meta name=""" & name & """ content=""")
        endPos = InStr(startPos, html, """")
        metaTag = Mid(html, startPos, endPos - startPos)
        ParseMetaTag = metaTag
    Else
        ParseMetaTag = ""
    End If
End Function

Function ParseCookies(response As String) As String
    Dim startPos As Long
    Dim endPos As Long
    Dim cookies As String
    Dim cookie As String
    
    cookies = ""
    
    ' Set-Cookieヘッダを検索してCookieを取得
    startPos = InStr(1, response, "Set-Cookie:")
    Do While startPos > 0
        startPos = startPos + Len("Set-Cookie:")
        endPos = InStr(startPos, response, vbCrLf)
        cookie = Mid(response, startPos, endPos - startPos)
        If Len(cookies) > 0 Then
            cookies = cookies & "; " & Trim(cookie)
        Else
            cookies = Trim(cookie)
        End If
        startPos = InStr(startPos, response, "Set-Cookie:")
    Loop
    
    ParseCookies = cookies
End Function

import socket
import ssl
import urllib.parse

HTML_TOKEN_NAME = "csrf-token"
POST_TOKEN_NAME = "authenticity_token"
GET_LOGIN_URL = "https://my.redmine.jp/demo/login"
POST_LOGIN_URL = "https://my.redmine.jp/demo/login"
USERNAME = "reporter"
PASSWORD = "reporter"

def send_request(host, port, request):
    """ソケットを使ってリクエストを送信し、レスポンスを受信"""
    sock = socket.create_connection((host, port))
    sock = ssl.wrap_socket(sock)  # HTTPS通信対応
    sock.sendall(request.encode("utf-8"))
    
    # レスポンスを受信
    response = b""
    while True:
        data = sock.recv(4096)
        if not data:
            break
        response += data
    sock.close()
    return response.decode("utf-8")

def parse_headers_and_body(response):
    """HTTPレスポンスをヘッダーとボディに分割"""
    header_end = response.find("\r\n\r\n")
    headers = response[:header_end]
    body = response[header_end + 4:]
    return headers, body

def parse_meta_tag(html, name):
    """HTML内から指定したmetaタグのcontent属性を抽出"""
    meta_start = html.find(f'<meta name="{name}" content="')
    if meta_start == -1:
        return None
    content_start = meta_start + len(f'<meta name="{name}" content="')
    content_end = html.find('"', content_start)
    return html[content_start:content_end]

def parse_cookies(headers):
    """レスポンスヘッダーからCookieを抽出"""
    cookies = {}
    for line in headers.split("\r\n"):
        if line.lower().startswith("set-cookie:"):
            parts = line.split(": ", 1)[1].split(";")[0].split("=")
            if len(parts) == 2:
                cookies[parts[0].strip()] = parts[1].strip()
    return cookies

def manual_urlencode(params):
    """辞書形式のパラメータをURLエンコードする関数"""
    encoded_str = ""
    for key, value in params.items():
        if encoded_str:
            encoded_str += "&"  # パラメータ同士を&で区切る
        encoded_str += manual_quote(str(key)) + "=" + manual_quote(str(value))
    return encoded_str

def manual_quote(value):
    """URLエンコードを手動で実装"""
    safe_chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_.~"
    result = ""
    for char in value:
        if char in safe_chars:
            result += char
        else:
            result += "%" + format(ord(char), "02X")
    return result

def get_login(url):
    """GETリクエストを送信してCSRFトークンとCookieを取得"""
    parsed_url = urllib.parse.urlparse(url)
    host, path = parsed_url.netloc, parsed_url.path
    request = f"GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n"
    
    response = send_request(host, 443, request)
    headers, body = parse_headers_and_body(response)
    cookies = parse_cookies(headers)
    csrf_token = parse_meta_tag(body, HTML_TOKEN_NAME)
    return csrf_token, body, cookies

def post_login(url, parameter={}, cookies={}):
    """POSTリクエストを送信してログイン"""
    parsed_url = urllib.parse.urlparse(url)
    host, path = parsed_url.netloc, parsed_url.path
    encoded_params = manual_urlencode(parameter)
    cookie_header = "; ".join(f"{key}={value}" for key, value in cookies.items())
    
    request = (
        f"POST {path} HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        f"Content-Type: application/x-www-form-urlencoded\r\n"
        f"Content-Length: {len(encoded_params)}\r\n"
        f"Cookie: {cookie_header}\r\n"
        f"Connection: close\r\n\r\n"
        f"{encoded_params}"
    )
    
    response = send_request(host, 443, request)
    headers, body = parse_headers_and_body(response)
    print("-"*60)
    print("RequestHeader(Step2): ■")
    print(request)
    print("-"*60)
    print("ResponseHeader(Step2): ■")
    print(headers)
    print("-"*60)
    print("StatusCode(Step2):", headers.split("\r\n")[0])  # ステータスライン

if __name__ == "__main__":
    # GETリクエストでトークン取得
    data = get_login(url=GET_LOGIN_URL)
    print("-"*60)
    print("Step1 Token:", data[0])

    if data[0] is not None:
        # POSTリクエストでログイン
        parameters = {
            POST_TOKEN_NAME: data[0],
            "username": USERNAME,
            "password": PASSWORD,
            "back_url": "/demo/projects/demo",
            "login": "ログイン",
        }
        post_login(url=POST_LOGIN_URL, parameter=parameters, cookies=data[2])

C:\Users\tttak>C:\Users\tttak\Desktop\l2.txt.py
C:\Users\tttak\Desktop\l2.txt.py:15: DeprecationWarning: ssl.wrap_socket() is deprecated, use SSLContext.wrap_socket()
  sock = ssl.wrap_socket(sock)  # HTTPS通信対応
------------------------------------------------------------
Step1 Token: ceJ9TvmKjfQUeyV8lWe0n3TVzitETbRwm_ah3vS8n3y_00KvYqAMoRo26y0lAupchXkjHJUXERw0UhRI-66adw
------------------------------------------------------------
RequestHeader(Step2): ■
POST /demo/login HTTP/1.1
Host: my.redmine.jp
Content-Type: application/x-www-form-urlencoded
Content-Length: 203
Cookie: _redmine_session=ZUFUMVJIYklYYTZ3dno0ZGVMVHAzdldXTHU3U0NyUWhNSy9udjNCa21rRHpXYkNoL2gyYkhFa1NSNFowVG9JOGFhZ1BrRHdTMHRBdS9Mck1nN2Vtd1JxL3R4TllOVmo2RG5XZ25BSHhmSmNGblYvTnFPRWZpU2NXcVJLbStFYmhmdDlnTnZTejFtUVFydzdUS3c3SVI5Ky8zWG9LbmpIU1lBT2dYZWpOMDdqcHdMd1JVYzA3cWlVOTdxeXU1eU9yLS1XWEQ5c3Zmb1ZKWGRXa2o2R2lPZ2hBPT0%3D--b22ecbf71ed9cb15b602413321738bdc25e6c056
Connection: close

authenticity_token=ceJ9TvmKjfQUeyV8lWe0n3TVzitETbRwm_ah3vS8n3y_00KvYqAMoRo26y0lAupchXkjHJUXERw0UhRI-66adw&username=reporter&password=reporter&back_url=%2Fdemo%2Fprojects%2Fdemo&login=%30ED%30B0%30A4%30F3
------------------------------------------------------------
ResponseHeader(Step2): ■
HTTP/1.1 302 Found
Date: Wed, 27 Nov 2024 13:03:40 GMT
Server: Apache
cache-control: no-cache
referrer-policy: strict-origin-when-cross-origin
x-permitted-cross-domain-policies: none
x-xss-protection: 1; mode=block
x-request-id: bdbb6384-c38d-43f9-9049-bcda5642fd3b
x-download-options: noopen
x-frame-options: SAMEORIGIN
x-content-type-options: nosniff
location: https://my.redmine.jp/demo/projects/demo
content-length: 0
Status: 302 Found
Content-Type: text/html; charset=utf-8
Set-Cookie: _redmine_session=SEIwVDgzeEZobmdFRGZ3Y1hXcGZUdWRRRDE4STRmWFdDcElEVGdUK1RXM3Vld1R0cVFvVDhHYjZIaE1aR3BIZUk3Vi9XdWZVd1o1dzF3T3hsRGFGWnpiUWQ5NUliUDRqMi9DMkE1KzBBZWFRZStGdUVxNWkrOHBTakdpS1YwdVRUZytjMlp5Q0J6MkJiSHp6ZkFzQXVnSTI5S0pySERWYnNseWZuTVNnamxRVHJaZlUyWTYzM1YzYzl3dGRrdnVadzdWd3FteHg4YXNNelJBMUZpR1RNK244MUIwZkFTaWptSzh5UytVTmYrQT0tLTloeVVhK1lPSWNxM25QSHYrc0tUeFE9PQ%3D%3D--5bae0bc4886df499a08e492a088bd8d13d7fd35d; path=/demo; httponly
X-Robots-Tag: noindex,nofollow,noarchive
Vary: Accept-Encoding
Connection: close
------------------------------------------------------------
StatusCode(Step2): HTTP/1.1 302 Found

C:\Users\tttak>

import os, csv, html, re, traceback


INPUT_ERROR_LOG_CSV_DIRECTORY_PATH = r"C:\Users\tttak\Desktop\BAF\in"
INPUT_ERROR_LOGS_MERGE_ROW_NUMBER = 0 #合致行数欄
INPUT_ERROR_LOGS_ERROR_CODE_NUMBER = None #制限を加えるための禁止行の根拠
BAN_TARGET_ERROR_CODE_SET = {} #制限を加えるための禁止事項の一覧
INPUT_ERROR_LOGS_ACTION_ROW_NAME_ROW = 3 #名称欄

INPUT_SOURCE_CSV_DIRECTORY_PATH = r"C:\Users\tttak\Desktop\BAF\opt"

OUTPUT_MERGED_ERROR_LOG_LIST = r"C:\Users\tttak\Desktop\BAF\out"

REPLACE_TABLE_NAME_DICT = {"main":"主",}

BASE_CSS_TEXT = """
table {
	border-collapse: collapse;
	margin-top: 0px;
}
.title {
	margin-top: 0.1em;
	margin-bottom: 0.1em;
}
th, td {
	border: 1px solid #000000;
	padding: 3px;
}
th {
	text-align: center;
}
td {
	text-align: left;
}
thead {
	position: sticky;
	top: 0;
	color: #000000;
	background: #dfdfdf;
	border-bottom: 3px double #000000;
}
tbody {
	overflow-x: hidden;
	overflow-y: scroll;
	height: 100px;
}
td.none_col {
	background-color: #999999;
	user-select: none;
	text-align: center;
}
.red_mark {
	color: 550000;
	background-color: ffcccc;
}
""".strip()

def read_csv_file(csv_path, encoding="utf-8"):
	header, data = None, None
	with open(csv_path, mode="r", encoding=encoding, newline="") as file:
		reader = csv.reader(file)
		header = next(reader, [])  # ヘッダーを取得
		data = [row for row in reader]  # データをリストとして取得
	return (header, data)

def write_csv_file(csv_path, data, header=None, encoding="utf-8"):
	os.makedirs(os.path.dirname(csv_path), exist_ok=True)  # ディレクトリが存在しない場合は作成
	with open(csv_path, mode="w", encoding=encoding, newline="") as file:
		writer = csv.writer(file)
		if header is not None:
			writer.writerow(header)  # ヘッダーを書き込む
		writer.writerows(data)  # データを書き込む

def write_html_file(html_path, data, header=None, red_marker_col_list=None, title=None, css_text=None, join_col_number=None):
	os.makedirs(os.path.dirname(html_path), exist_ok=True)  # ディレクトリが存在しない場合は作成
	with open(html_path, mode="w", encoding="utf-8-sig") as file:
		file.write("<html><head>{0}{1}{2}</head><body>".format(
			"" if title is None else "<title>{0}</title>".format(title),
			"" if css_text is None else "<style>{0}</style>".format(css_text),
			"<style>.col_num_{0} {{ border-right: 3px double #000000; }}</style>".format(join_col_number) if join_col_number is not None else ""
		))
		if title is not None and type(title) == str:
			file.write("<h3 class=\"title\">{0}</h3><hr>".format(html.escape(str(title))))
		file.write("<table>")
		max_width = max([len(m) for m in data]) if len(data) > 0 else 0
		max_width = max_width if header is None or max_width > len(header) else len(header)
		if header is not None:
			buffer_row = list()
			for i in range(max_width):
				if i < len(header) and header[i] is not None:
					buffer_row.append("<th class=\"normal_col col_num_{1}\">{0}</th>".format(html.escape(str(header[i])), i+1))
				else:
					buffer_row.append("<th class=\"none_col col_num_{0}\">(Null)</th>".format(i+1))
			file.write("<thead><tr>{0}</tr></thead>".format("".join(buffer_row)))
		red_marker_col_list = [None for _ in data] if red_marker_col_list is None or len(data) != len(red_marker_col_list) else red_marker_col_list
		for d, rmc in zip(data, red_marker_col_list):
			buffer_row = list()
			for i in range(max_width):
				if i < len(d) and d[i] is not None:
					buffer_row.append("<td class=\"normal_col col_num_{1}{2}\">{0}</td>".format(html.escape(str(d[i])), i+1, " red_mark " if rmc is not None and rmc == i else ""))
				else:
					buffer_row.append("<td class=\"none_col col_num_{0}{1}\"></td>".format(i+1, " red_mark " if rmc is not None and rmc == i else ""))
			file.write("<tr>{0}</tr>".format("".join(buffer_row)))
		file.write("</table></body></html>")

def replace_name(raw_name, replace_dict=REPLACE_TABLE_NAME_DICT):
	return raw_name if raw_name not in replace_dict else replace_dict[raw_name]

if "__main__" == __name__:
	csv_name_list = [file for file in os.listdir(INPUT_ERROR_LOG_CSV_DIRECTORY_PATH) if file.endswith(".csv")]
	for csv_name_one in csv_name_list:
		print("● {0}".format(csv_name_one))
		error_csv_file_path = os.path.join(INPUT_ERROR_LOG_CSV_DIRECTORY_PATH, csv_name_one)
		source_csv_file_path = os.path.join(INPUT_SOURCE_CSV_DIRECTORY_PATH, csv_name_one)
		output_csv_file_path = os.path.join(OUTPUT_MERGED_ERROR_LOG_LIST, "{0}.{1}".format(replace_name(csv_name_one.rstrip(".csv")), "csv"))
		output_html_file_path = os.path.join(OUTPUT_MERGED_ERROR_LOG_LIST, "{0}.{1}".format(replace_name(csv_name_one.rstrip(".csv")), "html"))
		if not(os.path.isfile(error_csv_file_path)):
			raise Exception("Not found main csv. ('{0}')".format(error_csv_file_path.replace("'", "\\'")))
		buffer_msg, children_buffer_msg_list = ("(　)", "No problem."), list()
		if not(os.path.isfile(source_csv_file_path)):
			buffer_msg = ("(！)", "Not found a pair source file. ('{0}')".format(source_csv_file_path.replace("'", "\\'")))
			print("    {0} {1}".format(buffer_msg[0], buffer_msg[1]))
			continue
		try:
			header_m, data_m = read_csv_file(error_csv_file_path, encoding="utf-8")
			header_s, data_s = read_csv_file(source_csv_file_path, encoding="utf-8")
			result, red_marker_col_list = list(), list()
			for dm in data_m:
				mm = dm + [None for _ in header_s]
				buffer_row_index = dm[INPUT_ERROR_LOGS_ACTION_ROW_NAME_ROW] if INPUT_ERROR_LOGS_ACTION_ROW_NAME_ROW is not None and INPUT_ERROR_LOGS_ACTION_ROW_NAME_ROW < len(dm) else None
				red_marker_col = None
				if buffer_row_index in header_s:
					red_marker_col = header_s.index(buffer_row_index)
					red_marker_col = len(header_m) + red_marker_col
				red_marker_col_list.append(red_marker_col)
				buffer_row_index = dm[INPUT_ERROR_LOGS_ERROR_CODE_NUMBER] if INPUT_ERROR_LOGS_ERROR_CODE_NUMBER is not None and INPUT_ERROR_LOGS_ERROR_CODE_NUMBER < len(dm) else None
				error_code_number = None if INPUT_ERROR_LOGS_ERROR_CODE_NUMBER is None or re.match(r"^[0-9]+$", buffer_row_index) is None else int(buffer_row_index)
				if error_code_number in BAN_TARGET_ERROR_CODE_SET:
					children_buffer_msg_list.append(("(Ｃ)", "continue row. ({0})". format(error_code_number)))
					continue
				buffer_row_index = dm[INPUT_ERROR_LOGS_MERGE_ROW_NUMBER] if INPUT_ERROR_LOGS_MERGE_ROW_NUMBER is not None and INPUT_ERROR_LOGS_MERGE_ROW_NUMBER < len(dm) else None
				merge_number = None if INPUT_ERROR_LOGS_MERGE_ROW_NUMBER is None or re.match(r"^[0-9]+$", buffer_row_index) is None else int(buffer_row_index)
				if merge_number is not None:
					if len(data_s) > merge_number:
						mm = dm + data_s[merge_number]
						children_buffer_msg_list.append(("(　)", "matching row. ({0})". format(merge_number)))
					else:
						children_buffer_msg_list.append(("(！)", "Nothing row. ({0})". format(merge_number)))
				else:
					children_buffer_msg_list.append(("(！)", "Not found merged row. ('{0}')". format(buffer_row_index.replace("'", "\\'"))))
				result.append(mm)
			if len(data_m) != len(result):
				buffer_msg = ("(！)", "Failed to append. ({0} -> {1})".format(len(data_m), len(result)))
			write_csv_file(output_csv_file_path, result, header=header_m+header_s, encoding="utf-8")
			write_html_file(
				output_html_file_path, 
				result, 
				header=header_m+header_s, 
				red_marker_col_list=red_marker_col_list,
				title="フィルタ除外一覧（{0}）".format(replace_name(csv_name_one.rstrip(".csv"))),
				css_text=BASE_CSS_TEXT,
				join_col_number=None if len(header_m) < 1 else len(header_m)
			)
		except Exception as e:
			buffer_msg = ("(Ｅ)", "[{0}]: {1}".format(type(e).__name__, str(e)))
			print(traceback.format_exc())
		print("    {0} {1}".format(buffer_msg[0], buffer_msg[1]))
		for cbm in children_buffer_msg_list:
			print("        {0} {1}".format(cbm[0], cbm[1]))
		print()

import os, re, unicodedata, datetime
import openpyxl as oxl

CATCH_FILTER_RECORD_DIRECTORY_PATH_LIST = [
	os.path.join(os.path.dirname(os.path.abspath(__file__)), "LOG", "FILTER_RECORD_01"),
	os.path.join(os.path.dirname(os.path.abspath(__file__)), "LOG", "FILTER_RECORD_02"),
]
CATCH_FILTER_LOG_DIRECTORY_PATH = [
	os.path.join(os.path.dirname(os.path.abspath(__file__)), "LOG", "FILTER_LOG_01"),
	os.path.join(os.path.dirname(os.path.abspath(__file__)), "LOG", "FILTER_LOG_02"),
]
OUTPUT_EXCEL_FILE_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "LOG", "", "dump.xlsx")

def merge_two_data(
	main_data_2d_list, sub_data_2d_list, main_title_list, sub_title_list, 
	join_column_main_to_sub_dict, main_data_ban_column_set=set(), sub_data_ban_column_set=set(),
	only_first_match_mode=False
):
	if len(main_data_2d_list) > 0 and len({len(d) for d in main_data_2d_list}) != 1:
		raise ValueError("The structure of the main table is broken.({0})".format(", ".join(list({len(d) for d in main_data_2d_list}))))
	if len(sub_data_2d_list) > 0 and len({len(d) for d in sub_data_2d_list}) != 1:
		raise ValueError("The structure of the sub table is broken.".format(", ".join(list({len(d) for d in sub_data_2d_list}))))
	if len(main_data_2d_list) > 0 and len(main_data_2d_list[0]) != len(main_title_list):
		raise ValueError("The number of columns in the main table and the main table title are different. ({0}, {1})".format(len(main_data_2d_list[0]), len(main_title_list)))
	if len(sub_data_2d_list) > 0 and len(sub_data_2d_list[0]) != len(sub_title_list):
		raise ValueError("The number of columns in the sub table and the sub table title are different. ({0}, {1})".format(len(sub_data_2d_list[0]), len(sub_title_list)))
	for jck in join_column_main_to_sub_dict.keys():
		if jck not in main_title_list:
			raise ValueError("There are no join columns in the main table title. ('{0}')".format(str(jck).replace("'", "\\'")))
	for jcv in join_column_main_to_sub_dict.values():
		if jcv not in sub_title_list:
			raise ValueError("There are no join columns in the sub table title. ('{0}')".format(str(jcv).replace("'", "\\'")))
	join_index_main_list, join_index_sub_list = list(), list()
	for k, v in join_column_main_to_sub_dict.items():
		join_index_main_list.append(main_title_list.index(k))
		join_index_sub_list.append(sub_title_list.index(v))
	ban_main_column_set = {main_title_list.index(s) for s in list(main_data_ban_column_set)}
	ban_sub_column_set = {sub_title_list.index(s) for s in list(sub_data_ban_column_set)}
	buffer_title_list = [main_title_list[i] for i in range(len(main_title_list)) if i not in ban_main_column_set] + [sub_title_list[i] for i in range(len(sub_title_list)) if i not in ban_sub_column_set]
	join_point = len([main_title_list[i] for i in range(len(main_title_list)) if i not in ban_main_column_set])
	error_list = list()
	buffer_2d_data = list()
	for main_d in main_data_2d_list:
		main_join_value_list = [main_d[jimo] for jimo in join_index_main_list]
		main_buffer = [main_d[i] for i in range(len(main_d)) if i not in ban_main_column_set]
		reject_flag = True
		for sub_d in sub_data_2d_list:
			sub_join_value_list = [sub_d[jiso] for jiso in join_index_sub_list]
			if all([m==s for m, s in zip(main_join_value_list, sub_join_value_list)]):
				sub_buffer = [sub_d[i] for i in range(len(sub_d)) if i not in ban_sub_column_set]
				buffer_2d_data.append(main_buffer + sub_buffer)
				reject_flag = False
				if only_first_match_mode:
					break
		if reject_flag:
			error_list.append("Found an unmatch record. ({0})".format(", ".join(["'{0}':'{1}'".format(str(t).replace("'", "\\'"), str(v).replace("'", "\\'")) for t, v in zip(main_title_list, main_d)])))
	return (buffer_title_list, buffer_2d_data, join_point, error_list)


MOTHER_TITLE = "みよし市 ＦＢＳＳ(住民記録・印鑑登録) ＤＢ フィルタリングレコード一覧"

class Report_generator:
	def __init__(self, output_path, mother_title=MOTHER_TITLE):
		self.output_path = output_path
		self.wb = wb = oxl.Workbook()
		self.ws_name_set = set()
		self.begin_datetime_text = datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
		self.mother_title = str(mother_title)
		self.active_row_dict = dict()
	def set(
		self, sheet_name, title_text, title_list, data_2d_list, join_point=None, search_column_name=None,
		right_flag_list=None, filter_dict=None
	):
		right_flag_list_true = [(True if right_flag_list[i] else False ) if len(right_flag_list) > i else False for i in range(len(title_list))] if right_flag_list is not None else [False for _ in range(len(title_list))] 
		
		search_column_index = None
		if join_point is None:
			if search_column_name in title_list:
				search_column_index = title_list.index(search_column_name)
		else:
			if search_column_name in title_list[:join_point]:
				search_column_index = title_list.index(search_column_name)
		ws = self.wb.create_sheet(sheet_name, len(self.wb.sheetnames))
		self.ws_name_set.add(sheet_name)
		ws.freeze_panes = ws.cell(5, 1).coordinate #行の固定
		max_point = [4, len(title_list)+1]
		green_list = list()
		for y, d in enumerate(data_2d_list): #値
			memory_value, memory_index = d[search_column_index], None
			if join_point is None:
				if memory_value in title_list[join_point:]:
					memory_index = title_list.index(memory_value)
			else:
				if memory_value in title_list:
					memory_index = title_list.index(memory_value)
			for x, dd in enumerate(d):
				c = ws.cell(y+5, x+2)
				if right_flag_list_true[x]:
					c.value = int(dd)
					c.number_format = "0"
				else:
					c.value = str(dd)
					c.number_format = "@"
				if memory_index is not None and x == memory_index:
					green_list.append((y+5, x+2))
			max_point[0] = y+5
		side = oxl.styles.borders.Side(style="thin", color="000000")
		for y in range(4, max_point[0]+1): #枠
			for x in range(2, max_point[1]+1):
				if join_point is not None and x == join_point + 1:
					ws.cell(y, x).border = oxl.styles.borders.Border(
						top=side, 
						bottom=side, 
						left=side,
						right=oxl.styles.borders.Side(style="double", color="000000")
					)
				else:
					ws.cell(y, x).border = oxl.styles.borders.Border(top=side, bottom=side, left=side, right=side)
		for y in range(1, max_point[0]+1): #白塗り
			for x in range(1, max_point[1]+1):
				ws.cell(y, x).fill = oxl.styles.PatternFill(patternType="solid", fgColor="FFFFFF")
		for g in green_list:
			ws.cell(g[0], g[1]).fill = oxl.styles.PatternFill(patternType="solid", fgColor="A0FFA0")
		for x, tl in enumerate(title_list): #列タイトル
			c = ws.cell(4, x+2)
			c.value = str(tl)
			c.number_format = "@"
			c.font = oxl.styles.Font(bold = True)
			c.fill = oxl.styles.PatternFill(
				patternType="solid", 
				fgColor="A0A0FF" if join_point is not None and join_point > x else "D0D0FF"
			)
		for col in ws.columns: #シート幅の自動調整
			max_length = 0
			column = col[0].column_letter
			for cell in col:
				try:
					if len(str(cell.value)) > max_length:
						max_length = len(cell.value)
				except:
					pass
			adjusted_width = (max_length + 1) * 2
			ws.column_dimensions[column].width = adjusted_width + 2.5
		c = ws.cell(1, 1)
		c.value = str(self.mother_title)
		c.font =  oxl.styles.Font(bold=True, size=20)
		c = ws.cell(2, 2)
		c.value = "－ {0}".format(title_text)
		c.font =  oxl.styles.Font(bold=True, size=15)
		ws.row_dimensions[3].height = 7.5
		ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=join_point+1 if join_point is None else len(title_list)+1)
		ws.merge_cells(start_row=2, start_column=2, end_row=2, end_column=join_point+1 if join_point is None else len(title_list)+1)
		for row in ws: #シート全体のフォントを変更
			for cell in row:
				cell.font = oxl.styles.Font(name="HG明朝B")
		ws.auto_filter.ref = "${0}${1}:${2}${3}".format(
			oxl.utils.get_column_letter(2),
			4,
			oxl.utils.get_column_letter(max_point[1]),
			max_point[0]
		) #オートフィルタ設定
		self.active_row_dict[sheet_name] = {row[0].row:True for row in ws.iter_rows(min_row=5, max_row=max_point[0])}
		for k, v in filter_dict.items():
			buffer_index = None
			if k in title_list:
				buffer_index = title_list.index(str(k))
			if buffer_index is not None and type(v) in (list, tuple, set):
				ws.auto_filter.add_filter_column(buffer_index, list(v)) #オートフィルタ再指定
				for row in ws.iter_rows(min_row=5, max_row=max_point[0]): #オートフィルタは自動設定されないため、手動定義
					if row[buffer_index+2-1].value not in v:
						ws.row_dimensions[row[0].row].hidden = True
						self.active_row_dict[sheet_name][row[0].row] = False
	def dump(self, index_sheet_name="テーブル一覧", filter_title="有効"):
		for ws in self.wb.worksheets: #配下のsheet以外は削除
			if ws.title not in self.ws_name_set:
				self.wb.remove(ws)
		ws = self.wb.create_sheet(str(index_sheet_name), 0)
		max_point, blue_font_list = [4, 8], list()
		for y, wsx in enumerate([wx for wx in self.wb.worksheets if wx != ws]): #値
			c = ws.cell(y+5, 2)
			c.number_format = "0"
			c.value = y + 1
			c = ws.cell(y+5, 3)
			c.number_format = "@"
			c.value = str(wsx.title)
			c.hyperlink = "#{0}!A1".format(wsx.title)
			blue_font_list.append((y+5, 3))
			max_point[0] = y+5
			c = ws.cell(y+5, 4)
			c.number_format = "0"
			c.value = int(len([f for f in self.active_row_dict[wsx.title].values() if f]))
			c = ws.cell(y+5, 5)
			c.number_format = "0"
			c.value = int(len(self.active_row_dict[wsx.title]))
		side = oxl.styles.borders.Side(style="thin", color="000000")
		for y in range(4, max_point[0]+1): #枠
			for x in range(2, 6):
				ws.cell(y, x).border = oxl.styles.borders.Border(top=side, bottom=side, left=side, right=side)
		for y in range(1, max_point[0]+1): #白塗り
			for x in range(1, max_point[1]+1):
				ws.cell(y, x).fill = oxl.styles.PatternFill(patternType="solid", fgColor="FFFFFF")
		for x, tl in enumerate(["No.", "テーブル名称", "{0}行数".format(filter_title), "総検出行数"]): #列タイトル
			c = ws.cell(4, x+2)
			c.value = str(tl)
			c.number_format = "@"
			c.font = oxl.styles.Font(bold=True)
			c.fill = oxl.styles.PatternFill(patternType="solid", fgColor="A0A0FF")
		for col in ws.columns: #自動幅調整
			max_length = 0
			column = col[0].column_letter
			for cell in col:
				try:
					if len(str(cell.value)) > max_length:
						max_length = len(cell.value)
				except:
					pass
			adjusted_width = (max_length + 1) * 2
			ws.column_dimensions[column].width = adjusted_width
		ws.column_dimensions[ws.cell(1, 3).column_letter].width = 100
		c = ws.cell(1, 1)
		c.value = str(self.mother_title)
		c.font =  oxl.styles.Font(bold=True, size=20)
		c = ws.cell(2, 2)
		c.value = "－ {0}".format(index_sheet_name)
		c.font =  oxl.styles.Font(bold=True, size=15)
		c = ws.cell(1, 4)
		c.value = "作成日時: {0}".format(self.begin_datetime_text)
		c.fill = oxl.styles.PatternFill(patternType="solid", fgColor="FFFFFF")
		ws.row_dimensions[3].height = 7.5
		ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=3)
		ws.merge_cells(start_row=2, start_column=2, end_row=2, end_column=3)
		ws.merge_cells(start_row=1, start_column=4, end_row=1, end_column=8)
		for row in ws: #シート全体のフォントを変更
			for cell in row:
				if (cell.row, cell.column) in blue_font_list:
					cell.font =  oxl.styles.Font(
						name="HG明朝B",
						color="0000DD", 
						underline="single"
					)
				else:
					cell.font = oxl.styles.Font(name="HG明朝B")
	def save(self):
		os.makedirs(os.path.dirname(self.output_path), exist_ok=True)
		self.wb.save(self.output_path)
	def close(self):
		try:
			self.wb.close()
		except Exception:
			pass

def check_column_type_is_free_int(data_2d_list, title_list, regex_column_name=r"^.*(番号)$"):
	if len(data_2d_list) > 0 and len(data_2d_list[0]) != len(title_list):
		raise ValueError("The number of columns in the table and the table title are different. ({0}, {1})".format(len(data_2d_list[0]), len(title_list)))
	return_flag_list = [True for _ in range(len(title_list))]
	for i in range(len(title_list)):
		if regex_column_name is None or re.search(regex_column_name, title_list[i]) is not None:
			for d in data_2d_list:
				if d[i] is not None:
					try:
						int(d[i])
					except TypeError:
						return_flag_list[i] = False
		else:
			return_flag_list[i] = False
	return return_flag_list


EX_MAIN_TITLE_LIST = ["ログ番号", "区分", "接続コード", "モード", "ロジック"]
EX_MAIN_DATA_LIST = [
	["001", "1", "01", "データ１", "MAIN-01"],
	["002", "1", "01", "データ１", "MAIN-01"],
	["003", "1", "02", "データ１", "MAIN-01"],
	["004", "1", "03", "データ２", "MAIN-01"],
	["005", "2", "01", "データ３", "MAIN-10"],
	["006", "2", "02", "データ２", "MAIN-10"],
	["007", "2", "02", "データ２", "MAIN-10"],
]
EX_SUB_TITLE_LIST = ["区分", "接続コードA", "データ１" ,"データ２"]
EX_SUB_DATA_LIST = [
	["1", "01", "あ", "わ"],
	["1", "02", "い", "わ"],
	["1", "03", "い", "わ"],
	["2", "02", "あ", "を"],
]
BAN_EX_SUB_TITLE_SET = {"区分", "接続コードA"}

if "__main__" == __name__:
	title_list, data_2d_list, join_point, error_list = merge_two_data(
		EX_MAIN_DATA_LIST, EX_SUB_DATA_LIST, EX_MAIN_TITLE_LIST, EX_SUB_TITLE_LIST, 
		{"区分":"区分", "接続コード":"接続コードA"}, main_data_ban_column_set=set(), sub_data_ban_column_set=BAN_EX_SUB_TITLE_SET,
		only_first_match_mode=False
	)
	print("{0} ( {1} )".format(title_list, join_point))
	for d in data_2d_list:
		print(d)
	print()
	for e in error_list:
		print(e)
	
	right_flag_list = check_column_type_is_free_int(data_2d_list, title_list, regex_column_name=r"^.*(番号)$")
	rg = Report_generator(output_path=OUTPUT_EXCEL_FILE_PATH, mother_title=MOTHER_TITLE)
	rg.set(
		sheet_name="テスト", title_text="テスト", title_list=title_list, data_2d_list=data_2d_list, join_point=join_point, 
		search_column_name="モード", right_flag_list=right_flag_list, filter_dict={"区分":["1"]}
	)
	rg.dump(index_sheet_name="テーブル一覧", filter_title="主体除外")
	rg.save()
	rg.close()

生成スクリプト
SelfDecryptPage_x-Fujitsu-JEFマッピングスクリプト.html
SelfDecryptPage_x-Fujitsu-EBCDICマッピングスクリプト.html

まとめ
SelfDecryptPage_x-Fujitsuマッピングデータ.html

#!/usr/bin/env python3
"""
jif_analyze.py
大量の .JIF ファイルからバイナリ構造を推察するための解析ツール

主な機能:
- ヘッダ（先頭 N バイト）の頻度集計
- 先頭4バイトを uint32 として分布確認（例: 0x00000190 = 400 の発生頻度）
- バイト周波数ヒストグラムとエントロピー（ファイル単位、スライディングウィンドウ）
- 既知マジックナンバー（PNG/JPEG/GIF/TIFF/BMP/ZIP/zlib/PKCS）の位置検索
- JPEG/PNG 等の埋め込み抽出（見つかれば別ファイルに保存）
- 代表サンプルの16進ダンプ出力
- 結果を CSV と JSON で保存

要件: Python 3.8+
推奨: pandas, numpy, matplotlib（プロットを作る場合）
"""
import os
import sys
import argparse
import struct
import binascii
import json
import math
from collections import Counter, defaultdict
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor, as_completed

# --- 設定 ---
DEFAULT_HEAD_BYTES = 512         # 先頭 N バイトを取得
SLIDING_WINDOW = 1024            # スライドでエントロピーを見るウィンドウ（省略可）
SAMPLE_COUNT = 10                # 代表ヘッダごとに出力するサンプル数
KNOWN_MAGIC = {
    'JPEG_SOI': bytes.fromhex('FFD8FF'),
    'JPEG_SOI_ALT': bytes.fromhex('FFD8FFE0'),  # JFIF
    'PNG': bytes.fromhex('89504E470D0A1A0A'),
    'GIF87a': b'GIF87a',
    'GIF89a': b'GIF89a',
    'BMP': b'BM',
    'TIFF_LE': bytes.fromhex('49492A00'),
    'TIFF_BE': bytes.fromhex('4D4D002A'),
    'ZIP': bytes.fromhex('504B0304'),
    'ZLIB_78_01': bytes.fromhex('7801'),
    'ZLIB_78_9C': bytes.fromhex('789C'),
    'ZLIB_78_DA': bytes.fromhex('78DA'),
    'PDF': b'%PDF-',
}

# 出力ディレクトリ
OUT_DIR = Path("results_jif")
OUT_DIR.mkdir(exist_ok=True)

# ----------------------------------------------------------
# ユーティリティ
# ----------------------------------------------------------
def read_head(path: Path, n=DEFAULT_HEAD_BYTES):
    with path.open('rb') as f:
        return f.read(n)

def file_size(path: Path):
    return path.stat().st_size

def uint32_le(b: bytes):
    if len(b) < 4:
        return None
    return struct.unpack('<I', b[:4])[0]

def byte_freq(data: bytes):
    c = Counter(data)
    freq = [c.get(i, 0) for i in range(256)]
    return freq

def entropy_from_counts(counts, total):
    ent = 0.0
    for c in counts:
        if c == 0:
            continue
        p = c / total
        ent -= p * math.log2(p)
    return ent

def file_entropy(data: bytes):
    counts = byte_freq(data)
    return entropy_from_counts(counts, len(data)) if len(data)>0 else 0.0

def find_subseq_positions(data: bytes, subseq: bytes):
    """subseq を data 内で全て探してオフセットリストを返す"""
    pos = []
    start = 0
    while True:
        i = data.find(subseq, start)
        if i == -1:
            break
        pos.append(i)
        start = i + 1
    return pos

def hexdump_snippet(data: bytes, length=64):
    return binascii.hexlify(data[:length]).decode('ascii')

# ----------------------------------------------------------
# ファイル単位の解析（Worker）
# ----------------------------------------------------------
def analyze_file(path):
    try:
        head = read_head(path, DEFAULT_HEAD_BYTES)
        size = file_size(path)
        u32 = uint32_le(head[:4])
        head_hex = binascii.hexlify(head).decode('ascii')
        first8 = head[:8]

        # byte freq + entropy（先頭 N バイト）
        freq = byte_freq(head)
        ent = entropy_from_counts(freq, len(head)) if len(head)>0 else 0.0

        # 全ファイルから検出したい known magic の探索（ファイル全体を読む）
        positions = {}
        with path.open('rb') as f:
            full = f.read()
            for name, sig in KNOWN_MAGIC.items():
                # 最初の数ヶ所だけ見つける
                pos = find_subseq_positions(full, sig)
                if pos:
                    positions[name] = pos[:10]  # 最大10件だけ保存

        # JPEG の SOI+EOI の可能性（簡易）
        jpeg_ranges = []
        if 'JPEG_SOI' in positions or 'JPEG_SOI_ALT' in positions:
            # find EOI 0xFFD9
            soi_pos = positions.get('JPEG_SOI', []) + positions.get('JPEG_SOI_ALT', [])
            eoi = bytes.fromhex('FFD9')
            for s in soi_pos:
                # search for EOI after s
                idx = full.find(eoi, s+2)
                if idx != -1:
                    jpeg_ranges.append((s, idx+2))

        # 先頭4バイトのパターン（hex）
        head4_hex = binascii.hexlify(head[:4]).decode('ascii') if len(head)>=4 else ''

        return {
            'path': str(path),
            'name': path.name,
            'size': size,
            'head4_uint32_le': u32,
            'head4_hex': head4_hex,
            'head_hex': head_hex[:256],  # 長すぎないよう切る
            'entropy_head': ent,
            'found_magic': {k: v for k,v in positions.items()},
            'jpeg_ranges': jpeg_ranges,
        }
    except Exception as e:
        return {'path': str(path), 'error': str(e)}

# ----------------------------------------------------------
# 集計・レポート作成
# ----------------------------------------------------------
def aggregate_results(results):
    # 頻出ヘッダ（先頭16バイト）
    head_counter = Counter()
    u32_counter = Counter()
    magic_counter = Counter()
    entropies = []
    sizes = []

    for r in results:
        if 'error' in r:
            continue
        head16 = r.get('head_hex', '')[:32]
        head_counter[head16] += 1
        u32_counter[r.get('head4_uint32_le')] += 1
        sizes.append(r.get('size',0))
        entropies.append(r.get('entropy_head',0.0))
        for k in r.get('found_magic',{}).keys():
            magic_counter[k] += 1

    summary = {
        'total_files': len(results),
        'unique_top16_headers': len(head_counter),
        'top_headers': head_counter.most_common(20),
        'u32_distribution_sample': u32_counter.most_common(20),
        'magic_counts': magic_counter.most_common(),
        'size_stats': {
            'min': min(sizes) if sizes else 0,
            'max': max(sizes) if sizes else 0,
            'mean': sum(sizes)/len(sizes) if sizes else 0,
        },
        'entropy_stats': {
            'min': min(entropies) if entropies else 0,
            'max': max(entropies) if entropies else 0,
            'mean': sum(entropies)/len(entropies) if entropies else 0,
        }
    }
    return summary

# ----------------------------------------------------------
# 抽出処理（JPEG/PNG 等を見つけたらファイルに書き出す）
# ----------------------------------------------------------
def extract_embedded(results):
    extract_dir = OUT_DIR / "extracted"
    extract_dir.mkdir(exist_ok=True)
    count = 0
    for r in results:
        if 'error' in r:
            continue
        path = Path(r['path'])
        try:
            with path.open('rb') as f:
                data = f.read()
            # JPEG 抽出（簡易）: SOI..EOI の範囲を保存
            for (s,e) in r.get('jpeg_ranges', []):
                outp = extract_dir / f"{path.name}.jpeg_{s}_{e}.jpg"
                with outp.open('wb') as fo:
                    fo.write(data[s:e])
                count += 1

            # PNG 抽出: search png signature and try to find IEND chunk
            png_sig = KNOWN_MAGIC['PNG']
            start = 0
            while True:
                s = data.find(png_sig, start)
                if s == -1:
                    break
                # PNG end chunk: IEND (00 00 00 00 49 45 4E 44 AE 42 60 82)
                iend = b'\x00\x00\x00\x00IEND\xaeB`\x82'
                e = data.find(iend, s)
                if e != -1:
                    e += len(iend)
                    outp = extract_dir / f"{path.name}.png_{s}_{e}.png"
                    with outp.open('wb') as fo:
                        fo.write(data[s:e])
                    count += 1
                    start = e
                else:
                    start = s + 8
        except Exception as ex:
            print("extract error:", path, ex)

    return count

# ----------------------------------------------------------
# メイン
# ----------------------------------------------------------
def main(args):
    p = Path(args.input)
    files = list(p.rglob("*.jif")) + list(p.rglob("*.JIF"))
    print(f"found {len(files)} .JIF files under {p}")
    results = []
    # 並列で解析
    with ProcessPoolExecutor(max_workers=args.workers) as ex:
        futures = {ex.submit(analyze_file, f): f for f in files}
        for fut in as_completed(futures):
            res = fut.result()
            results.append(res)

    # 集計
    summary = aggregate_results(results)

    # JSON/CSV 出力
    OUT_DIR.mkdir(exist_ok=True)
    (OUT_DIR / "file_results.json").write_text(json.dumps(results, indent=2))
    (OUT_DIR / "summary.json").write_text(json.dumps(summary, indent=2))

    # 書き出しの簡易CSV
    import csv
    csvfile = OUT_DIR / "file_results.csv"
    with csvfile.open('w', newline='', encoding='utf-8') as fo:
        writer = csv.writer(fo)
        writer.writerow(['path','name','size','head4_uint32_le','head4_hex','entropy_head','found_magic'])
        for r in results:
            writer.writerow([r.get('path'), r.get('name'), r.get('size'),
                             r.get('head4_uint32_le'), r.get('head4_hex'),
                             r.get('entropy_head'), ';'.join(r.get('found_magic',{}).keys())])

    # 抽出
    extracted = extract_embedded(results)

    print("summary:", json.dumps(summary, indent=2))
    print(f"extracted {extracted} embedded files (jpeg/png) into {OUT_DIR / 'extracted'}")
    print("results written into:", OUT_DIR)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Analyze .JIF files to infer binary structure.")
    parser.add_argument("input", help="directory containing .JIF files")
    parser.add_argument("--workers", type=int, default=4, help="parallel workers")
    args = parser.parse_args()
    main(args)

#!/usr/bin/env python3
"""
bitmap_range_scan.py
多数の .JIF ファイルから、非圧縮ビットマップ（低エントロピー）領域を
統計的に推定するためのスクリプト。

出力:
 - results_bitmap/entropy_positions.json
 - results_bitmap/low_entropy_map.csv
 - 統計的に出現率が高い「低エントロピー開始位置」
"""

import os
import math
import json
from pathlib import Path
from collections import defaultdict, Counter
from concurrent.futures import ProcessPoolExecutor, as_completed

WINDOW = 2048          # 一度に読むバイト数（非圧縮画像だと数KB〜数十KB）
STEP = 512             # どれだけずらして読むか
LOW_ENTROPY = 6.5      # これ以下なら「非圧縮かもしれない」とみなす
OUT_DIR = Path("results_bitmap")
OUT_DIR.mkdir(exist_ok=True)

def entropy(data: bytes):
    if not data:
        return 0.0
    freq = [0]*256
    for b in data:
        freq[b] += 1
    ent = 0.0
    total = len(data)
    for c in freq:
        if c == 0:
            continue
        p = c / total
        ent -= p * math.log2(p)
    return ent

def process_file(path: Path):
    try:
        with path.open('rb') as f:
            data = f.read()

        size = len(data)
        low_positions = []

        # スライディングエントロピー
        for off in range(0, size-WINDOW, STEP):
            chunk = data[off:off+WINDOW]
            ent = entropy(chunk)
            if ent < LOW_ENTROPY:
                low_positions.append(off)

        return {
            "file": str(path),
            "size": size,
            "low_positions": low_positions
        }
    except Exception as e:
        return {"file": str(path), "error": str(e)}

def main():
    p = Path(".")
    files = list(p.rglob("*.jif")) + list(p.rglob("*.JIF"))
    print(f"Found {len(files)} files")

    results = []
    pos_counter = Counter()

    with ProcessPoolExecutor(max_workers=4) as ex:
        futs = {ex.submit(process_file, f): f for f in files}
        for fut in as_completed(futs):
            r = fut.result()
            results.append(r)
            if "low_positions" in r:
                for pos in r["low_positions"]:
                    pos_counter[pos] += 1

    # 保存
    (OUT_DIR / "entropy_positions.json").write_text(json.dumps(results, indent=2))

    # CSV 保存
    with (OUT_DIR / "low_entropy_map.csv").open("w", encoding="utf-8") as fo:
        fo.write("offset,count\n")
        for off, c in pos_counter.most_common():
            fo.write(f"{off},{c}\n")

    print("Top low-entropy offsets (possible bitmap start):")
    for off, c in pos_counter.most_common(20):
        print(f" offset {off}: {c} files")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
compress_analyze.py

JIFファイルの中から圧縮方式を推定するための統計解析スクリプト。
主な機能:
 - スライディングウィンドウエントロピー
 - 既知圧縮シグネチャの検出
 - バイト頻度解析
 - 全ファイルでの圧縮ブロックのオフセット統計
"""

import os
import math
import json
from pathlib import Path
from collections import Counter, defaultdict
from concurrent.futures import ProcessPoolExecutor, as_completed

WINDOW = 2048
STEP = 256
HIGH_ENT = 7.5     # 圧縮ブロックとして扱う閾値
LOW_ENT = 5.5      # 非圧縮 or ビットマップと思われる閾値
OUT = Path("results_compress")
OUT.mkdir(exist_ok=True)

# known compression signatures
COMP_SIG = {
    "ZLIB_78_01": bytes.fromhex("7801"),
    "ZLIB_78_9C": bytes.fromhex("789C"),
    "ZLIB_78_DA": bytes.fromhex("78DA"),
    "LZMA": bytes.fromhex("5D00008000"),
    "LZFSE": bytes.fromhex("62767832"),
    "LZ4_FRAME": bytes.fromhex("04224D18"),
    "JPEG": bytes.fromhex("FFD8FF"),
    "PNG": bytes.fromhex("89504E470D0A1A0A"),
    "JBIG2": bytes.fromhex("974A4232"),
    # CCITT G4 は明確なシグネチャが無いため後述の統計で判定
}

def entropy(data: bytes):
    if not data:
        return 0.0
    freq = [0]*256
    for b in data:
        freq[b] += 1
    ent = 0.0
    total = len(data)
    for c in freq:
        if c == 0:
            continue
        p = c / total
        ent -= p * math.log2(p)
    return ent

def find_signature(data: bytes):
    found = []
    for name, sig in COMP_SIG.items():
        idx = data.find(sig)
        if idx != -1:
            found.append((name, idx))
    return found

def process_file(path: Path):
    try:
        with path.open("rb") as f:
            data = f.read()

        size = len(data)

        # signature scan
        sigs = find_signature(data)

        # sliding entropy
        high_positions = []
        low_positions = []

        for off in range(0, size-WINDOW, STEP):
            ent = entropy(data[off:off+WINDOW])
            if ent > HIGH_ENT:
                high_positions.append(off)
            elif ent < LOW_ENT:
                low_positions.append(off)

        return {
            "file": str(path),
            "size": size,
            "signatures": sigs,
            "high_entropy": high_positions,
            "low_entropy": low_positions
        }

    except Exception as e:
        return {"file": str(path), "error": str(e)}


def main():
    files = list(Path(".").rglob("*.JIF")) + list(Path(".").rglob("*.jif"))
    print("Found", len(files), "files")

    results = []
    sig_counter = Counter()
    high_pos_counter = Counter()
    low_pos_counter = Counter()

    with ProcessPoolExecutor(max_workers=4) as ex:
        futs = {ex.submit(process_file, f): f for f in files}
        for fut in as_completed(futs):
            r = fut.result()
            results.append(r)

            if "signatures" in r:
                for name, pos in r["signatures"]:
                    sig_counter[name] += 1

            if "high_entropy" in r:
                for pos in r["high_entropy"]:
                    high_pos_counter[pos] += 1

            if "low_entropy" in r:
                for pos in r["low_entropy"]:
                    low_pos_counter[pos] += 1

    # save
    (OUT/"results.json").write_text(json.dumps(results, indent=2))
    (OUT/"signature_counts.json").write_text(json.dumps(sig_counter, indent=2))

    with (OUT/"high_entropy_map.csv").open("w") as fo:
        fo.write("offset,count\n")
        for off,c in high_pos_counter.most_common():
            fo.write(f"{off},{c}\n")

    with (OUT/"low_entropy_map.csv").open("w") as fo:
        fo.write("offset,count\n")
        for off,c in low_pos_counter.most_common():
            fo.write(f"{off},{c}\n")

    print("=== SIGNATURE COUNTS ===")
    print(sig_counter)
    print("Result saved in", OUT)


if __name__ == "__main__":
    main()