tklib.tkcgi.tkFormData のソースコード
"""
CGI/WSGI環境からのフォームデータとファイルアップロードを解析するためのユーティリティモジュール。
このモジュールは、HTTPリクエストのGETクエリ文字列やPOSTリクエストのmultipart/form-data形式のデータを
簡単に解析し、フォームデータやアップロードされたファイルにアクセスするためのクラスと関数を提供します。
:doc:`tkFormData_usage`
"""
import os
import sys
import urllib.parse
from urllib.parse import parse_qs
#from email.parser import BytesParser
#from email.policy import default
[ドキュメント]
class tkFormData:
"""
CGI/WSGI環境からのHTTPリクエストデータを解析し、フォームデータとアップロードされたファイルを抽出するクラス。
GETリクエストのクエリ文字列と、POSTリクエストの `application/x-www-form-urlencoded`
および `multipart/form-data` 形式のデータを処理します。
"""
def __init__(self, environ, input_stream):
"""
tkFormDataクラスのインスタンスを初期化し、フォームデータとファイルを解析します。
環境変数と入力ストリームから、GETのクエリ文字列とPOSTのmultipartデータを解析します。
:param environ: dict, WSGI環境変数またはそれに準ずる辞書。
:param input_stream: file-like object, リクエストボディを読み込むためのファイルライクオブジェクト。通常は `sys.stdin.buffer`。
"""
self.form_data = {}
self.files = {}
# print("<H3>parse query</H3>")
self._parse_query_string(environ)
# print("query parameters:", self.form_data)
# print("<H3>parse multipart</H3>")
self._parse_multipart_data(environ, input_stream)
def _parse_query_string(self, environ):
"""
CGI/WSGI環境変数からクエリ文字列を解析し、フォームデータとして保存します。
`QUERY_STRING` 環境変数を `urllib.parse.parse_qs` を使用して解析します。
:param environ: dict, WSGI環境変数またはそれに準ずる辞書。
"""
query_string = environ.get("QUERY_STRING", "")
parsed_query = urllib.parse.parse_qs(query_string)
# クエリパラメーターを辞書形式に保存
for key, value in parsed_query.items():
self.form_data[key] = value[0] # 単一の値を保存
def _parse_multipart_data(self, environ, input_stream):
"""
HTTPリクエストボディから `multipart/form-data` 形式のデータを解析し、フォームデータとファイルを抽出します。
`CONTENT_TYPE` ヘッダーからboundaryを特定し、リクエストボディを分割して
各パートのヘッダーとコンテンツを解析します。ファイルは `self.files` に、
通常のフォームデータは `self.form_data` に保存されます。
:param environ: dict, WSGI環境変数またはそれに準ずる辞書。`CONTENT_TYPE` を含みます。
:param input_stream: file-like object, リクエストボディを読み込むためのファイルライクオブジェクト。
"""
content_type = environ.get("CONTENT_TYPE", "")
# print("content_type=", content_type, "<br>")
if not content_type.startswith("multipart/form-data"):
# print("not multipart/form-data: return<br>")
return # multipartでない場合はスキップ
# Boundaryを抽出
boundary = content_type.split("boundary=")[-1].strip()
boundary_bytes = f"--{boundary}".encode()
end_boundary_bytes = f"--{boundary}--".encode()
# print("boundary=", boundary, "<br>\n")
# データを読み込む
data = input_stream.read()
self.query_string = data
# print("data=", data, "<br>\n")
# 各パートを分割
parts = data.split(boundary_bytes)
for part in parts:
# print("<hr>part:", part, "<br>\n")
if not part or part == end_boundary_bytes:
continue
# ヘッダー部分とコンテンツ部分を分割
if b"\r\n\r\n" not in part:
# print("Invalid part structure: skipping<br>\n")
continue
headers, content = part.split(b"\r\n\r\n", 1)
headers = headers.decode().strip().split("\r\n")
# コンテンツディスポジションを解析
disposition = [h for h in headers if h.startswith("Content-Disposition")][0]
_, params = disposition.split(": ", 1)
params = {kv.split("=")[0]: kv.split("=")[1].strip('"') for kv in params.split("; ") if "=" in kv}
name = params.get("name")
filename = params.get("filename")
if filename: # ファイルの場合
# print("filename:", filename, "<br>\n")
self.files[name] = {
"filename": filename,
"content": content.strip()
}
else: # フォームデータの場合
self.form_data[name] = content.strip().decode()
# print(f"self.form_data[{name}]:", self.form_data[name], "<br>\n")
# print("_parse_multipart_data: end<br>\n")
[ドキュメント]
def get_form_data(self):
"""
解析されたフォームデータ（キーと値のペア）を返します。
:returns: dict, 解析されたフォームデータを格納する辞書。
"""
return self.form_data
[ドキュメント]
def get_file(self, name):
"""
指定された名前のアップロードファイルを返します。
:param name: str, 取得したいファイルのフォームフィールド名。
:returns: dict or None, ファイルの情報 (`filename`, `content`) を含む辞書、またはファイルが見つからない場合は `None`。
"""
return self.files.get(name)
[ドキュメント]
def get_params():
"""
HTTPリクエストメソッドに基づいて、GETまたはPOSTリクエストのパラメータとファイル情報を取得します。
`REQUEST_METHOD` 環境変数をチェックし、GETリクエストの場合は `QUERY_STRING` を、
POSTリクエストの場合は `CONTENT_TYPE` に応じて `tkFormData` クラスを使用して
リクエストボディを解析します。
:returns: tuple, `(request_method, query_string, query_params, form_storage)` を含むタプル。
- `request_method` (str): リクエストメソッド (例: 'GET', 'POST')。
- `query_string` (str or bytes): オリジナルのクエリ文字列またはPOSTボディの生データ。
- `query_params` (dict): 解析されたクエリパラメータまたはフォームデータ。
- `form_storage` (tkFormData or None): `multipart/form-data` の場合に `tkFormData` オブジェクト、それ以外は `None`。
"""
query_string = ""
request_method = os.environ.get('REQUEST_METHOD', '')
# print(f"{request_method=}<br>\n")
if request_method == 'GET':
# クエリパラメータを取得（GETリクエスト用）
query_string = os.environ.get('QUERY_STRING', '')
query_params = parse_qs(query_string)
# print("<hr>\n")
# print("GET:<br>")
# print(" query string: ", query_string, "<br>\n")
# print(" query params: ", query_params, "<br>\n")
elif request_method == 'POST':
# print("<hr>\n")
# print("POST:<br>")
content_type = os.environ.get('CONTENT_TYPE', '')
form_storage = None
if content_type.startswith('multipart/form-data'):
form_storage = tkFormData(os.environ, sys.stdin.buffer)
query_params = form_storage.form_data
query_string = form_storage.query_string
# print(" query_params:", query_params, "</br>\n")
# print(" form_storage=", form_storage, "<br>\n")
# if form_storage:
# file_inf = form_storage.files
# for key in file_inf.keys():
# print(f" key: {key}<br>\n")
# print(f" filename:", file_inf[key]['filename'], "</br>\n")
## content = file_inf[key]['content']
## outfile = "d:/a.xlsx"
## print(f"Save content to [{outfile}]<br>\n")
## with open(outfile, "wb") as f:
## f.write(content)
else:
print("Unknown request method<br>\n")
return request_method, query_string, query_params, form_storage