"""
tkCGIApplicationモジュール
このモジュールは、PythonでCGIアプリケーションを構築するためのフレームワークを提供します。
ルーティング、テンプレートレンダリング、エラーハンドリング、セキュリティ設定、メール送信、
外部APIリクエストなどの機能を含んでいます。
詳細説明:
CGI環境でのウェブアプリケーション開発を簡素化し、リクエストの解析からレスポンスの生成までを
一貫してサポートします。GET/POSTパラメータの処理、ファイルアップロード、セキュリティ対策、
ロギング、アカウント認証機能などを備え、堅牢なアプリケーション開発を支援します。
:doc:`tkCGIApplication_usage`
"""
import os
import platform
import sys
import traceback
from dotenv import load_dotenv
#import toml
from datetime import datetime
import re
import builtins
import functools
import inspect
import logging
import requests
import json
import yaml
import html
import urllib.parse
from jinja2 import Template, Environment, FileSystemLoader
import smtplib
from tkBase import *
import tkFormData
from tkHTMLDocument import tkHTMLElement
"""
frame = inspect.currentframe()
print(f"File: {inspect.getfile(frame)}")
print(f"Function: {inspect.getframeinfo(frame).function}")
print(f"Line: {inspect.getframeinfo(frame).lineno}")
"""
[ドキュメント]
def handle_exception(exc_type, exc_value, exc_tb):
"""
捕捉された例外をHTML形式でクライアントに出力します。
詳細説明:
標準エラー出力にトレースバック情報をHTMLの`<pre>`タグで囲んで出力し、
Webブラウザで読みやすい形式でエラーを表示します。
:param exc_type: 例外の型。
:param exc_value: 例外インスタンス。
:param exc_tb: トレースバックオブジェクト。
:returns: なし
"""
print("Content-Type: text/html\n")
print("<html><body>")
print("<h2>Error:</h2>")
print("<pre>")
traceback.print_exception(exc_type, exc_value, exc_tb, file=sys.stdout)
print("</pre>")
print("</body></html>")
[ドキュメント]
class tkResponse:
"""
CGIアプリケーションからのレスポンスを表現するシンプルなクラスです。
詳細説明:
操作の成功/失敗を示すブール値と、付随するメッセージを保持します。
インスタンスはブール値として評価可能です。
"""
def __init__(self, res = False, message = ""):
"""
tkResponseの新しいインスタンスを初期化します。
:param res: bool, レスポンスが成功した場合はTrue、失敗した場合はFalse。
:param message: str, レスポンスに関連するメッセージ。
:returns: なし
"""
self.res = res
self.message = message
def __bool__(self):
"""
インスタンスがブール値として評価されたときの動作を定義します。
:returns: bool, `self.res`の値を返します。
"""
return self.res
[ドキュメント]
class tkCGIApplication:
"""
CGIアプリケーションを構築するためのフレームワーククラスです。
詳細説明:
このクラスは、CGI環境でのWebアプリケーション開発を簡素化するために設計されています。
リクエストの解析（GET/POSTパラメータ、PATH_INFO）、HTML/JSON出力の管理、
ルーティング、テンプレートレンダリング、エラーハンドリング、ロギング、
アカウント認証、外部HTTPリクエストの送信、メール通知などの機能を提供します。
"""
# クラス変数としてオリジナルの print 関数を退避
print_original = builtins.print
def __init__(self, import_name, static_folder = 'static', static_url_path = '/static', template_folder = 'templates', error_handler = True):
"""
tkCGIApplicationの新しいインスタンスを初期化します。
詳細説明:
アプリケーションのルートパス、静的ファイルやテンプレートのパス、
CGI環境変数から取得するリクエスト情報などを設定します。
また、必要に応じてカスタムエラーハンドラーをセットアップします。
:param import_name: str, アプリケーションのモジュール名。このモジュールに基づいてルートパスが決定されます。
:param static_folder: str, 静的ファイルが配置されているディレクトリの名前（ルートパスからの相対パス）。デフォルトは'static'。
:param static_url_path: str, 静的ファイルにアクセスするためのURLパス。デフォルトは'/static'。
:param template_folder: str, テンプレートファイルが配置されているディレクトリの名前（ルートパスからの相対パス）。デフォルトは'templates'。
:param error_handler: bool, Trueの場合、カスタム例外ハンドラーを設定します。デフォルトはTrue。
:returns: なし
"""
if error_handler:
self.set_error_handler()
self.import_name = import_name
if import_name in sys.modules:
module = sys.modules[import_name]
self.root_path = os.path.dirname(os.path.abspath(module.__file__))
else:
self.root_path = os.getcwd()
self.os_name = platform.system()
self.script_path = sys.argv[0]
self.script_fullpath = os.path.abspath(self.script_path)
self.script_dir = os.path.dirname(self.script_fullpath)
self.script_filename = os.path.basename(self.script_path)
self.script_filebody, self.script_ext = os.path.splitext(self.script_filename)
self.static_url_path = static_url_path
self.static_folder = os.path.abspath(os.path.join(self.root_path, static_folder))
self.templates_dir = os.path.abspath(os.path.join(self.root_path, template_folder))
self.data_folder = None
self.print = builtins.print
# CGIで起動した場合
self.server_name = os.environ.get("SERVER_NAME", "localhost")
self.script_name = os.environ.get("SCRIPT_NAME", "")
# CLIで起動した場合
if self.script_name == "":
self.script_name = sys.argv[0]
self.query_string = os.environ.get("QUERY_STRING", "")
self.full_url = f"http://{self.server_name}{self.script_name}"
if self.query_string:
self.full_url += f"?{self.query_string}"
self.query_string = None
self.query = None
self.method = self.get_method()
#pythonの出力をutf8に固定
if self.method != "":
sys.stdout.reconfigure(encoding='utf-8')
self.output_target = None # html|json
self.html_initialized = False
self.deny_iframe = True
self.security_policy = "default-src 'self'; script-src 'self'; style-src 'self'"
self.routes = {}
self.error_handlers = {}
[ドキュメント]
def setup_config(self, security_level = 5):
"""
指定されたセキュリティレベルに基づいて、アプリケーションのセキュリティ設定を構成します。
詳細説明:
`security_level`の値に応じて、iFrameの埋め込み許可(X-Frame-Options)や
コンテンツセキュリティポリシー(Content-Security-Policy)ヘッダーを設定します。
レベルが高いほど厳格なポリシーが適用されます。
:param security_level: int, 設定するセキュリティレベル。1から5の範囲。
5: iFrame拒否, CSPを厳格に自己サイトに限定。
3: iFrame拒否, CSPを自己サイトに限定 (スクリプト/スタイル)。
2: iFrame拒否, CSPでインラインスクリプト/スタイルを許可。
1: iFrame許可, CSPでインラインスクリプト/スタイルを許可。
0以下: iFrame許可, CSPなし。
:returns: なし
"""
self.security_level = security_level
self.security_policy = ""
if security_level >= 5:
self.deny_iframe = True
self.security_policy = "default-src 'self'; script-src 'self'; style-src 'self'"
elif security_level >= 3:
self.deny_iframe = True
self.security_policy = "script-src 'self'; style-src 'self'"
elif security_level >= 2:
self.deny_iframe = True
self.security_policy = "script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline';";
elif security_level >= 1:
self.deny_iframe = False
self.security_policy = "script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline';";
else:
self.deny_iframe = False
self.security_policy = ""
[ドキュメント]
def init_html(self, target = "html", charset="utf-8", title="title", html_ver="HTML5", force_init=False):
"""
HTML/JSONレスポンスの初期ヘッダーとHTML構造の開始を出力します。
詳細説明:
`target`が"json"の場合はJSON形式のヘッダーを出力し、
それ以外の場合はHTMLのDOCTYPE宣言、`<html>`, `<head>`, `<body>`タグの開始を出力します。
セキュリティポリシーや文字セット、タイトルも設定されます。
既に初期化されている場合は、`force_init`がTrueでない限り再初期化は行われません。
:param target: str, 出力形式。"html"または"json"を指定します。デフォルトは"html"。
:param charset: str, HTMLページの文字エンコーディング。デフォルトは"utf-8"。
:param title: str, HTMLページの`<title>`タグの内容。デフォルトは"title"。
:param html_ver: str, HTMLのバージョン。現在"HTML5"のみが対応しています。デフォルトは"HTML5"。
:param force_init: bool, Trueの場合、既に初期化されていても強制的に再初期化を行います。デフォルトはFalse。
:returns: なし
"""
if not force_init and self.html_initialized:
return
self.html_initialized = True
self.output_target = target
if self.output_target == "json":
if charset is None or charset == "":
self.print_original(f"Content-Type: application/json")
else:
self.print_original(f"Content-Type: application/json; charset={charset}")
self.print_original()
else:
# HTMLヘッダーを出力
self.print_original(f"Content-Type: text/html; charset={charset}")
# self.print_original("Access-Control-Allow-Origin: *")
if self.deny_iframe:
self.print_original('X-Frame-Options: DENY')
if self.security_policy != "":
self.print_original(f"Content-Security-Policy: {self.security_policy}")
self.print_original()
self.print_original("<!DOCTYPE html>")
self.print_original("<html>")
self.print_original("<head>")
if html_ver == "HTML5":
self.print_original(f" <meta charset=\"{charset}\">")
# self.print_original(f" <meta http-equiv=\"Content-Type\" content=\"text/html; charset={charset}\">")
else:
self.print_original(f" <meta http-equiv=\"Content-Type\" content=\"text/html; charset={charset}\">")
self.print_original(f" <title>{title}</title>")
self.print_original("</head>")
self.print_original("<body>")
[ドキュメント]
def end_html(self):
"""
HTMLの終了タグ(`</body>`と`</html>`)を出力します。
詳細説明:
`init_html`でHTML出力が初期化されている場合にのみ、これらのタグを出力します。
出力形式が"json"の場合は何もしません。
:returns: なし
"""
if self.html_initialized and self.output_target != "json":
self.print_original("</body>")
self.print_original("</html>")
[ドキュメント]
def set_error_handler(self):
"""
システムレベルの例外ハンドラーを`handle_exception`関数に設定します。
詳細説明:
これにより、未処理の例外が発生した場合に、指定された`handle_exception`関数が呼び出され、
HTML形式でエラー情報がWebブラウザに表示されるようになります。
:returns: なし
"""
sys.excepthook = handle_exception
[ドキュメント]
def set_mail_params(self, info = {}):
"""
メール送信に使用するパラメータを設定します。
詳細説明:
スクリプト名、テンプレートパス、送信元/送信先メールアドレス、件名、名前などの
基本的なメールパラメータを初期化し、`info`辞書で提供される追加情報で更新します。
:param info: dict, メール送信に関する追加情報を含む辞書。デフォルトは空の辞書。
:returns: なし
"""
params = {
"script_filebody": self.script_filebody,
"template_path" : os.path.join(self.templates_dir, info.get("template_filename", "")),
"from" : info.get('from', "default_from@example.com"),
"to" : info.get('to', "default_to@example.com"),
"subject" : info.get('subject', "Notification mail"),
"recipient_name" : info.get('recipient_name', "No recipient name"),
"sender_name" : info.get('sender_name', "No sender name"),
}
params.update(info)
self.mail_params = params
[ドキュメント]
def start_logging(self, log_path):
"""
ロギングシステムを開始し、ログファイルをセットアップします。
詳細説明:
指定された`log_path`が存在しない場合、ディレクトリを作成します。
その後、`setup_logging`を呼び出してロギングの設定を行います。
ログファイルのオープンに失敗した場合は、エラーメッセージをHTMLで出力し、アプリケーションを終了します。
:param log_path: str, ログファイルのフルパス。
:returns: なし
"""
self.log_path = log_path
if log_path:
try:
self.setup_logging()
except Exception as e:
self.init_html(target = 'html', charset = 'utf-8')
self.print(f"Error in {self.script_filebody}:tkCGIApplication.start_logging(): Could not open logging due to {e}")
exit()
[ドキュメント]
def setup_logging(self):
"""
Pythonのloggingモジュールを設定します。
詳細説明:
ログファイルが保存されるディレクトリが存在しない場合は作成します。
指定されたログパスに書き込み権限があるかを確認し、`logging.basicConfig`を呼び出して
ログレベル（DEBUG）とログフォーマットを設定します。
書き込み権限がない場合は、エラーメッセージを出力し、システムを終了します。
:returns: なし
"""
if not os.path.exists(os.path.dirname(self.log_path)):
os.makedirs(os.path.dirname(self.log_path))
if os.access(os.path.dirname(self.log_path), os.W_OK):
# log_format = ("[%(asctime)s] In %(pathname)s:%(name)s %(funcName)s(): line:%(lineno)d: %(message)s")
# log_format = ("[%(asctime)s] In %(filename)s:%(name)s %(funcName)s(): line:%(lineno)d: %(message)s")
log_format = (f"[%(asctime)s] In {self.script_filebody}:%(filename)s:%(name)s %(funcName)s(): line:%(lineno)d: %(message)s")
# logging.basicConfig(level = logging.CRITICAL, filename=self.log_path, format = log_format)
# logging.basicConfig(level = logging.ERROR, filename=self.log_path, format = log_format)
# logging.basicConfig(level = logging.WARNING, filename=self.log_path, format = log_format)
# logging.basicConfig(level = logging.INFO, filename=self.log_path, format = log_format)
logging.basicConfig(level = logging.DEBUG, filename=self.log_path, format = log_format)
else:
self.print({'result': False, 'message': 'Logging directory is not writable.'})
raise SystemExit("Critical error: Unable to initialize logging.")
[ドキュメント]
def load_config(self, config_path):
"""
指定されたパスから設定ファイルを読み込みます。
詳細説明:
設定ファイルはINI形式であることを想定し、`read_ini`関数を使用して解析されます。
ファイルが見つからない、またはフォーマットが無効な場合はエラーメッセージを出力します。
:param config_path: str, 設定ファイルへのパス。
:returns: dict or None, 設定情報を格納した辞書、またはエラーが発生した場合はNone。
"""
if not os.path.exists(config_path):
print(f"Error in {self.script_filebody}:tkCGIApplication.load_config: Can not find [{config_path}]")
return None
inf = read_ini(config_path)
if not inf:
print(f"Error in {self.script_filebody}:tkCGIApplication.load_config: Invalid format in [{config_path}]")
return None
return inf
[ドキュメント]
def load_key_info(self, key_info_path):
"""
指定されたパスからキー情報ファイルを読み込みます。
詳細説明:
キー情報ファイルはINI形式であることを想定し、`read_ini`関数を使用して解析されます。
ファイルが見つからない場合はエラーメッセージを出力します。
:param key_info_path: str, キー情報ファイルへのパス。
:returns: dict or None, キー情報を格納した辞書、またはエラーが発生した場合はNone。
"""
if not os.path.exists(key_info_path):
print(f"Error in {self.script_filebody}:tkCGIApplication.load_key_info: Can not find [{key_info_path}]")
return None
return read_ini(key_info_path)
[ドキュメント]
def read_config(self, path):
"""
設定ファイルを読み込みます。
詳細説明:
`path`がNoneの場合、スクリプトのフルパスと`{dirname}/{filebody}.cfg`のテンプレートを使用して
設定ファイルのパスを生成します。その後、INI形式のファイルを読み込みます。
:param path: str or None, 設定ファイルへのパス。Noneの場合、デフォルトのパスが生成されます。
:returns: dict, 読み込まれた設定情報を含む辞書。
:raises RuntimeError: 設定ファイルの読み込みに失敗した場合。
"""
if path is None:
path = self.replace_path(None, template=["{dirname}", "{filebody}.cfg"])
self.config_path = path
try:
conf = read_ini(path)
except Exception as e:
raise RuntimeError(f"Error in {self.script_filebody}:tkCGIAppliaction.read_config(): Failed to read config file: {e}")
return conf
[ドキュメント]
def replace_path(self, path = None, template = None, ext_dict = {}):
"""
指定されたパスとテンプレートを使用して新しいパス文字列を生成します。
詳細説明:
テンプレート文字列内の`{dirname}`、`{filebody}`、`{ext}`などのプレースホルダを、
元のパスの情報や`ext_dict`の内容で置き換えます。テンプレートは文字列または文字列のリストで指定可能です。
:param path: str or None, 元となるパス。Noneの場合、`self.script_fullpath`が使用されます。
:param template: str or list of str or None, 新しいパスを生成するためのテンプレート文字列または文字列のリスト。
Noneの場合、デフォルトで`"{dirname}/{filebody}-out.txt"`が使用されます。
:param ext_dict: dict, テンプレート内のカスタムプレースホルダを置き換えるための辞書。
:returns: str, 生成された新しいパス文字列。
"""
if template is None:
template = os.path.join("{dirname}", "{filebody}-out.txt")
if path is None:
path = self.script_fullpath
if type(template) is list:
p = []
for s in template:
s2 = replace_path(path, s, ext_dict)
p.append(s2)
return os.path.join(p[0], *p[1:])
else:
return replace_path(path, template, ext_dict)
[ドキュメント]
def get_method(self):
"""
HTTPリクエストメソッドを取得します。
詳細説明:
CGI環境変数`REQUEST_METHOD`からリクエストメソッド（例: 'GET', 'POST'）を取得します。
環境変数が設定されていない場合は空文字列を返します。
:returns: str, HTTPリクエストメソッド。
"""
return os.environ.get('REQUEST_METHOD', '') # 'GET','POST', ''
[ドキュメント]
def get_path_info(self):
"""
CGI環境変数`PATH_INFO`の値を取得します。
詳細説明:
`PATH_INFO`は、CGIスクリプト名に続く追加のパス情報を提供します。
:returns: str, `PATH_INFO`の値。
"""
inf = os.environ.get('PATH_INFO', '')
return inf
[ドキュメント]
def get_path_args(self):
"""
`PATH_INFO`からパス引数のリストを取得します。
詳細説明:
`PATH_INFO`の文字列をスラッシュで分割し、空の要素を除外してリストを生成します。
最初の要素には先頭にスラッシュが追加されます。
:returns: list of str, パス引数のリスト。
"""
inf = os.environ.get('PATH_INFO', '')
if inf == '/':
return ['/']
args = [s for s in inf.strip('/').split('/') if s]
if len(args) > 0:
args[0] = '/' + args[0]
return args
[ドキュメント]
def get_client_inf(self):
"""
クライアントとリクエストに関する情報をCGI環境変数から取得します。
詳細説明:
クライアントのIPアドレス、ポート、ユーザーエージェント、ホスト名、
`PATH_INFO`、`REQUEST_METHOD`、`QUERY_STRING`、`CONTENT_LENGTH`、
および`HTTP_`で始まるすべてのヘッダー情報を辞書として返します。
:returns: dict, クライアントとリクエストの詳細情報。
"""
client_ip = os.environ.get('REMOTE_ADDR', 'unknown')
client_port = os.environ.get('REMOTE_PORT', 'unknown')
user_agent = os.environ.get('HTTP_USER_AGENT', 'unknown')
host = os.environ.get('HTTP_HOST', 'unknown')
inf = {
"host": host,
"ip_address": client_ip,
"port": client_port,
"user_agent": user_agent,
"PATH_INFO": os.environ.get('PATH_INFO', ''),
"REQUEST_METHOD": os.environ.get('REQUEST_METHOD', ''),
"QUERY_STRING": os.environ.get('QUERY_STRING', ''),
"CONTENT_LENGTH": os.environ.get('CONTENT_LENGTH', ''),
}
for key, value in os.environ.items():
if key.startswith("HTTP_"):
inf[key] = value
return inf
[ドキュメント]
def reload(self):
"""
クライアントのWebページを再ロードするJavaScriptコードを出力します。
:returns: なし
"""
print("<script>")
print("window.location.reload();")
print("</script>")
[ドキュメント]
def redirect(self, new_url):
"""
クライアントを新しいURLにリダイレクトするJavaScriptコードを出力します。
:param new_url: str, リダイレクト先のURL。
:returns: なし
"""
print("<script>")
print("window.location.href = {new_url};")
print("</script>")
[ドキュメント]
def download(self, file_path):
"""
指定されたファイルをクライアントにダウンロードさせます。
詳細説明:
ファイルが存在しない場合は404 Not Foundエラーを返します。
存在する場合は、`Content-Type: application/octet-stream` ヘッダーと
`Content-Disposition: attachment` ヘッダーを設定してファイルをバイナリで出力します。
:param file_path: str, ダウンロードさせるファイルのフルパス。
:returns: なし (HTTPレスポンスを直接出力し、プロセスを終了)
"""
if not os.path.exists(file_path):
print("Status: 404 Not Found")
print("Content-Type: text/plain")
print()
print(f"File [{file_path}] not found")
exit()
print("Content-Type: application/octet-stream")
print(f"Content-Disposition: attachment; filename={os.path.basename(file_path)}")
print()
with open(file_path, "rb") as file:
sys.stdout.buffer.write(file.read())
[ドキュメント]
def textarea_copy(self, text, textarea_id = "textarea_id", rows = 5, cols = 100, copy_button_text = 'copy'):
"""
指定されたテキストを含むtextareaと、その内容をクリップボードにコピーするボタンを出力します。
詳細説明:
JavaScriptを使用して、ユーザーがボタンをクリックした際にtextareaの内容をコピーする機能を提供します。
:param text: str, textareaに表示するテキスト。
:param textarea_id: str, textarea要素のID。JavaScriptで参照されます。デフォルトは"textarea_id"。
:param rows: int, textareaの行数。デフォルトは5。
:param cols: int, textareaの列数。デフォルトは100。
:param copy_button_text: str, コピーボタンに表示するテキスト。デフォルトは'copy'。
:returns: なし
"""
self.print("""
<script>
function copyText() {
var textarea = document.getElementById("{textarea_id}");
textarea.select();
document.execCommand("copy");
alert("Copied: " + textarea.value);
}
</script>
<button onclick="copyText()">{copy_button_text}</button>
<textarea id="textArea" rows="{rows}" cols="{cols}">{text}</textarea><br>
""")
[ドキュメント]
def error(self, *args):
"""
指定されたメッセージを赤い太字のHTMLで出力します。
:param args: str, 出力するエラーメッセージ。複数の引数を指定できます。
:returns: なし
"""
print("<H3><font color='red'>")
print(*args)
print("</font></H3>")
[ドキュメント]
def write(self, *args):
"""
引数が文字列ならそのまま出力し、tkHTMLElementオブジェクトなら`.outerHTML`を出力します。
詳細説明:
HTML出力が初期化されていない場合、自動的に`init_html`を呼び出して初期化します。
引数が文字列の場合はそのまま`print_original`で出力し、
`outerHTML`属性を持つオブジェクト（例: `tkHTMLElement`のインスタンス）の場合は、その属性値を出力します。
それ以外の型の場合、エラーメッセージを出力します。
:param args: str or tkHTMLElement, 出力する内容。複数の引数を指定できます。
:returns: なし
"""
if not self.html_initialized:
self.init_html()
for arg in args:
if isinstance(arg, str):
self.print_original(arg)
# elif isinstance(arg, tkHTMLElement):
else:
if hasattr(arg, 'outerHTML'):
self.print_original(arg.outerHTML)
else:
print("[Error] Argument is a dictionary but does not have an 'outerHTML' attribute.")
# else:
# print(f"Error in {self.script_filebody}:tkCGIApplication.write(): Unsupported argument type for arg=[{arg}]:", type(arg))
[ドキュメント]
def print_custom(self, *args, **kwargs):
"""
CGIアプリケーションのカスタムprint関数として機能し、標準出力をラップします。
詳細説明:
HTML出力が初期化されていない場合、自動的に`init_html`を呼び出して初期化します。
その後、与えられた引数をオリジナルの`builtins.print`関数で出力します。
これは、CGI出力の制御を可能にするために`builtins.print`をリダイレクトする際に使用されます。
:param args: 任意の型の可変長引数。`print`関数に渡されます。
:param kwargs: 任意のキーワード引数。`print`関数に渡されます。
:returns: なし
"""
if not self.html_initialized:
self.init_html()
# `<pre>` タグで囲んで表示
# self.print_original("<pre>")
self.print_original(*args, **kwargs)
# self.print_original("</pre>")
[ドキュメント]
def redirect_print(self, print_func=None):
"""
`builtins.print`関数をカスタムの出力関数にリダイレクトします。
詳細説明:
`print_func`が指定された場合、その関数にリダイレクトします。
`print_func`がNoneの場合、このインスタンスの`print_custom`メソッドにリダイレクトします。
これにより、アプリケーションの出力制御をCGI環境に合わせてカスタマイズできます。
:param print_func: function or None, `builtins.print`の代わりに呼び出す関数。
Noneの場合、`self.print_custom`が使用されます。
:returns: なし
"""
if print_func:
builtins.print = print_func
else:
builtins.print = self.print_custom
[ドキュメント]
def route(self, action):
"""
URLアクションをハンドラー関数にマップするデコレータです。
詳細説明:
このデコレータは、指定された`action`文字列と、デコレートされた関数を
アプリケーションのルーティングテーブル(`self.routes`)に登録します。
これにより、特定のアクション名に対応する関数を呼び出すことができます。
:param action: str, この関数が処理するURLアクションのキー。
:returns: function, デコレートされた関数をラップするラッパー関数。
"""
def decorator(func):
self.routes[action] = func
@functools.wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return decorator
[ドキュメント]
def register_error_handler(self, code, handler):
"""
特定のHTTPステータスコードに対するエラーハンドラー関数を登録します。
:param code: int, 処理するHTTPステータスコード（例: 404）。
:param handler: function, 指定されたエラーコードが発生したときに呼び出される関数。
:returns: なし
"""
self.error_handlers[code] = handler
[ドキュメント]
def handle_error(self, code):
"""
登録されているエラーハンドラーに従ってエラーを処理します。
詳細説明:
指定された`code`に対応するハンドラーが登録されていればそれを呼び出します。
ハンドラーが見つからない場合は、デフォルトのJSON形式エラーメッセージを出力します。
:param code: int, 発生したエラーのHTTPステータスコード。
:returns: any or None, エラーハンドラーの戻り値、またはデフォルトのエラー処理の場合はNone。
"""
handler = self.error_handlers.get(code)
if handler:
return handler()
else:
# デフォルトエラー応答
print(json.dumps({'result': False, 'message': f"Error {code} occurred."}, indent=4))
logging.error(f"Unhandled error {code}")
[ドキュメント]
def get_account_list(self):
"""
アカウント情報が記述されたYAMLファイルを読み込み、アカウントリストを取得します。
詳細説明:
`self.account_path`に設定されたファイルからYAML形式のアカウント情報を読み込みます。
ファイルが存在しない、または解析エラーが発生した場合は、ロギングを行いNoneを返します。
:returns: list of dict or None, 各辞書が単一のアカウント情報を含むリスト、またはエラー発生時はNone。
"""
if not self.account_path: return None
logging.info(f"reading {self.account_path}")
if os.path.exists(self.account_path):
with open(self.account_path, 'r') as file:
try:
accounts = yaml.safe_load(file)
logging.info(f"Accounts loaded successfully")
"""
for idx, inf in enumerate(accounts):
for key, val in inf.items():
logging.info(f" {idx}: {key}: {val}")
"""
return accounts
except yaml.YAMLError as e:
logging.error(f"Error parsing YAML file: {e}")
return None
else:
logging.warning(f"{self.account_path} file not found.")
return None
[ドキュメント]
def get_account(self, user, account_list):
"""
アカウントリストから指定されたユーザー名に一致するアカウント情報を検索します。
詳細説明:
`account_list`内の辞書オブジェクトを繰り返し処理し、'user'キーの値が`user`に一致する
最初のアカウント情報を返します。見つからない場合はNoneを返します。
:param user: str, 検索するユーザー名。
:param account_list: list of dict, アカウント情報を含む辞書のリスト。
:returns: dict or None, マッチしたアカウント情報を含む辞書、または見つからない場合はNone。
"""
# print()
# print(f"{account_list=}")
# print()
for inf in account_list:
# print(f"{self.script_filebody}:tkCGIApplication.get_account(): {user=} {inf=}")
if inf.get('user') == user: # 'user'キーが存在し、値が一致
return inf # マッチしたオブジェクトを返す
return None # 見つからない場合は None を返す
[ドキュメント]
def is_valid_root_dir(self, rootDir):
"""
指定されたルートディレクトリが、親ディレクトリへの不正なアクセスを含まないか検証します。
詳細説明:
セキュリティ対策として、パスに`../`や`./`が含まれていないかを確認します。
:param rootDir: str, 検証するルートディレクトリパス。
:returns: bool, 不正なパス要素が含まれていない場合はTrue、それ以外はFalse。
"""
if '../' in rootDir: return False
if './' in rootDir: return False
return True
[ドキュメント]
def is_valid_filemask(self, filemask):
"""
ファイルマスクが有効であるかを確認します。
詳細説明:
現時点では常にTrueを返しますが、将来的にファイルマスクの正規表現の妥当性などを検証するロジックを
追加するためのプレースホルダーです。
:param filemask: str, 検証するファイルマスク。
:returns: bool, 常にTrue。
"""
return True
[ドキュメント]
def is_allowed_path(self, path, allowed_dirs, file_masks = [], rejected_file_masks = []):
"""
指定されたパスが許可されたディレクトリとファイルマスクの条件を満たしているか検証します。
詳細説明:
1. パスが`allowed_dirs`リストに含まれるディレクトリ内に存在するかを確認します。
2. パス内のファイル名が`file_masks`リストのいずれかの正規表現にマッチするかを確認します。
3. パス内のファイル名が`rejected_file_masks`リストのいずれの正規表現にもマッチしないことを確認します。
これらの条件すべてを満たした場合のみ`tkResponse(True)`を返します。
:param path: str, 検証するファイルパス。
:param allowed_dirs: list of str, 許可されたディレクトリパスのリスト。
:param file_masks: list of str, ファイル名を許可するための正規表現文字列のリスト。デフォルトは空のリスト。
:param rejected_file_masks: list of str, ファイル名を拒否するための正規表現文字列のリスト。デフォルトは空のリスト。
:returns: tkResponse, 許可された場合は`tkResponse(True, ...)`、拒否された場合は`tkResponse(False, message)`。
"""
# path を dir と filename に分割
directory, filename = os.path.split(path)
directory = directory.replace('\\', '/')
if not directory.endswith('/'):
directory += '/'
# dir が allowed_dirs リスト変数に含まれていなければ False を返す
if allowed_dirs and directory not in allowed_dirs:
# print(" not allowed dir")
return tkResponse(False, f"Directory [{directory}] in [{path}] is not allowed for dirs in [{allowed_dirs}]")
# allowed_dirs が None でも "" でもない場合
# file_masks のすべての正規表現に filename がマッチしなければ False を返す
is_allowed = False
if file_masks:
for mask in file_masks:
if re.search(mask, filename, re.IGNORECASE):
# print(" passed")
is_allowed = True
break
# else:
# print(" not passed")
if not is_allowed:
# print(" not allowed file mask")
return tkResponse(False, f"File name [{filename}] in [{path}] is not allowed for file_masks in [{file_masks}]")
# rejected_file_masks が None でも "" でもない場合
# その中のいずれかの正規表現に filename がマッチすれば False を返す
if rejected_file_masks:
for mask in rejected_file_masks:
if re.search(mask, filename, re.IGNORECASE):
print(" rejected mask")
return tkResponse(False, f"[{path}] was rejected due to rejected_file_masks in [{rejected_file_masks}]")
# その他の場合は True を返す
return tkResponse(True, f"[{path}] passed permission screening")
[ドキュメント]
def is_valid_ip(self, account, ip):
"""
クライアントのIPアドレスが指定されたアカウントで許可されているか検証します。
詳細説明:
アカウント情報に含まれる'IPaddress'リスト内の各正規表現パターンに対して、
提供された`ip`がマッチするかを確認します。
:param account: dict, アカウント情報を含む辞書。'IPaddress'キーに許可されたIPパターンリストを持つことが期待されます。
:param ip: str, 検証するクライアントのIPアドレス。
:returns: bool, IPアドレスが許可されている場合はTrue、それ以外はFalse。
"""
return any(re.match(pattern, ip) for pattern in account['IPaddress'])
[ドキュメント]
def is_valid_password(self, account, password):
"""
提供されたパスワードが指定されたアカウントのパスワードと一致するか検証します。
:param account: dict, アカウント情報を含む辞書。'password'キーに正しいパスワードを持つことが期待されます。
:param password: str, 検証するパスワード。
:returns: bool, パスワードが一致する場合はTrue、それ以外はFalse。
"""
return account['password'] == password
[ドキュメント]
def validate_account(self, func):
"""
アカウント認証を行うデコレータです。デコレートされた関数を実行する前に、
ユーザー名、パスワード、IPアドレスの検証を行います。
詳細説明:
CGIリクエストのクエリ文字列から`user`と`password`を取得し、
`get_account_list`と`get_account`を使用してアカウント情報をロードします。
`is_valid_password`と`is_valid_ip`を使用してそれぞれパスワードとIPアドレスを検証します。
認証に失敗した場合は、JSON形式のエラーメッセージを出力し、メール通知を送信して関数の実行を中止します。
認証に成功した場合のみ、デコレートされた関数を実行します。
:param func: function, 認証が必要な関数。
:returns: function, 認証ロジックをラップした関数。
"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
# print()
# print("start {self.script_filebody}:tkCGIApplication.validate_account():")
logging.info(f"Validating account for function: {func.__name__}")
# QUERY_STRINGから必要な情報を取得
user = self.query.get('user', None)
password = self.query.get('password', None)
clientIP = os.getenv('REMOTE_ADDR', '')
# print("Requested account inf:")
# print(f" {user=}")
# print(f" {password=}")
# print(f" {clientIP=}")
# アカウントリストの取得
account_list = self.get_account_list()
if not account_list:
message = f"Error in {self.script_filebody}:tkCGIApplication.validate_account(): Failed to load account list from [{self.account_path}]: [{account_list}]"
logging.error(message)
print(json.dumps({'result': False, 'message': message}, indent=4))
self.send_mail(self.key_info, params = self.mail_params, message = message)
return
# アカウント取得
account = self.get_account(user, account_list)
# print(f"{self.script_filebody}:tkCGIApplication.validate_account(): {account=}")
if not account:
message = f"Error in {self.script_filebody}:tkCGIApplication.validate_account(): Failed to get account information for user={user} from [{self.account_path}]: [{account_list}]"
logging.warning(f"Unauthorized user access attempt: user={user}")
print(json.dumps({'result': False, 'message': "Invalid user or unauthorized access."}, indent=4))
self.send_mail(self.key_info, params = self.mail_params, message = message)
return
# print(f" account passed")
# パスワードの確認
if not self.is_valid_password(account, password):
logging.warning(f"Invalid password attempt for user: {user}")
print(json.dumps({'result': False, 'message': "Invalid password."}, indent=4))
self.send_mail(self.key_info, params = self.mail_params, message = message)
return
# print(f" password passed")
# IPアドレスの確認
if not self.is_valid_ip(account, clientIP):
logging.warning(f"Unauthorized IP access attempt for user: {user}, IP: {clientIP}")
print(json.dumps({'result': False, 'message': "Unauthorized IP address."}, indent=4))
self.send_mail(self.key_info, params = self.mail_params, message = message)
return
# print(f" IP address passed")
# アカウントが有効であれば関数を実行
# print(f"{self.script_filebody}:tkCGIApplication.validate_account(): call {func=}")
self.account_list = account_list
self.account = account
try:
return func(*args, **kwargs)
except Exception as e:
logging.error(f"Route function [{func.__name__}] calling error due to [{e}]")
return None
return wrapper
[ドキュメント]
def logging(self, func):
"""
関数のエントリーとエグジットをログに記録するデコレータです。
詳細説明:
デコレートされた関数が呼び出される直前と、その実行が完了した直後に、
関数の名前とともに情報メッセージをログに出力します。
:param func: function, ロギングを適用する関数。
:returns: function, ロギング機能を追加したラッパー関数。
"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
logging.info(f"Entered to [{func.__name__}]")
ret = func(*args, **kwargs)
logging.info(f"Exited from [{func.__name__}]")
return ret
return wrapper
[ドキュメント]
def send_mail(self, server_info, params = None, message = 'no message'):
"""
メールを送信します。
詳細説明:
`server_info`に指定されたSMTPサーバー情報と、`params`に含まれる送信元/送信先、件名、テンプレート情報を使用してメールを送信します。
テンプレートファイルが存在しない場合や、SMTPサーバーへの接続/ログイン、メール送信中にエラーが発生した場合は、
エラーメッセージをログに出力し、Falseを返します。
:param server_info: dict, SMTPサーバー接続情報（`smtp_server`, `port`, `username`, `password`）。
:param params: dict or None, メール送信パラメータ（`template_path`, `from`, `to`, `subject`など）。Noneの場合、処理を中断します。
:param message: str, ログメッセージとして使用される追加メッセージ。デフォルトは'no message'。
:returns: bool, メール送信が成功した場合はTrue、失敗した場合はFalse。
"""
'''
server_info: smtp_server, port, user_name, password
params: from, to
'''
if params is None:
return False
template_path = params['template_path']
params['date'] = datetime.now().strftime('%Y-%m-%d')
params['time'] = datetime.now().strftime('%H:%M:%S')
if not os.path.exists(template_path):
message = f"Error in {self.script_filebody}:tkCGIApplication.send_mail(): Can not find template file [{template_path}]"
print(message)
logging.error(message)
# func(*args, **kwargs) は存在しないためコメントアウト
# return func(*args, **kwargs)
return False # エラーが発生したためFalseを返す
with open(template_path, 'r', encoding='utf-8') as file:
template_str = file.read()
template = Template(template_str)
rendered_message = template.render(params)
try:
server = smtplib.SMTP(server_info.get('smtp_server', 'no server given'), server_info.get('port', 25))
except Exception as e:
message = f"Error in {self.script_filebody}:tkCGIApplication.send_mail(): Can not open SMTP server [{params['smtp_server']}]"
print(message)
logging.error(message)
return False
username = server_info.get('username', '')
password = server_info.get('password', '')
if username and password:
try:
server.login(username, password)
except Exception as e:
message = f"Error in {self.script_filebody}:tkCGIApplication.send_mail(): Failed to loigin SMTP server with user [{username}]"
print(message)
logging.error(message)
return False
try:
server.sendmail(params['from'], params['to'], rendered_message)
message = f"email sent to [{params['to']}] from [{params['from']}]"
logging.info(message)
except Exception as e:
message = f"Error in {self.script_filebody}:tkCGIApplication.send_mail(): Failed to send e-mail due to [{e}]"
print(message)
logging.error(message)
return False # エラーが発生したためFalseを返す
return True
[ドキュメント]
def email(self, inf, server_info):
"""
デコレートされた関数が呼び出された後にメールを送信するデコレータです。
詳細説明:
デコレートされた関数が実行された後、`self.mail_params`と`server_info`を使用して
`send_mail`メソッドを呼び出します。これにより、関数の実行結果や特定のイベントに応じて
自動的に通知メールを送信できます。
:param inf: dict, メール送信パラメータに追加する情報。
:param server_info: dict, SMTPサーバー接続情報。
:returns: function, メール送信ロジックをラップした関数。
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
params = self.mail_params.copy()
params['func_name'] = func.__name__
# `inf`の内容を`params`にマージするが、既存コードにはそのロジックがないため追加しない
# params.update(inf) # もし必要であれば追加する
ret = self.send_mail(server_info, params = params, message = '')
return func(*args, **kwargs)
return wrapper
return decorator
[ドキュメント]
def parse_query_string(self, query_string):
"""
クエリ文字列を解析し、パラメータの辞書を返します。
詳細説明:
`query_string`を`&`で分割し、各パラメータを`=`でキーと値に分けます。
キーと値は`sanitize`メソッドでサニタイズされ、値はURLデコードされます。
キーが空のパラメータは無視されます。
:param query_string: str, 解析するクエリ文字列。
:returns: dict, パースされたパラメータを格納した辞書。
"""
query = {}
for param in query_string.split('&'):
if '=' in param:
key, value = param.split('=', 1)
if key == '': continue
key = self.sanitize(value = key)
query[key] = self.sanitize(value = urllib.parse.unquote(value))
else:
if param == '': continue
key = self.sanitize(value = param)
query[key] = ''
# logging.warning(f"Malformed parameter in query string: {param}")
return query
[ドキュメント]
def get_params(self):
"""
GET、POST、またはコマンドライン引数からパラメータを取得します。
詳細説明:
既にパラメータが取得されている場合はキャッシュされた`self.query`を返します。
そうでない場合、`REQUEST_METHOD`に応じて以下の順でパラメータを解析します。
- POSTリクエスト: `CONTENT_TYPE`が`multipart/form-data`の場合は`tkFormData`を使用。
それ以外の場合は`sys.stdin`からボディを読み込みます。
- GETリクエスト: `QUERY_STRING`環境変数を使用。
- CLI実行: `sys.argv`の引数を使用。
取得したパラメータは`parse_query_string`で解析され、`PATH_INFO`からも`action`パラメータが抽出されます。
すべてのパラメータは`self.query`に格納され、返されます。
:returns: dict, 取得したパラメータの辞書。
"""
if self.query:
return self.query
self.form_data = None
request_method = os.environ.get('REQUEST_METHOD', '')
self.request_method = request_method
if request_method == 'POST':
# form-dataがmultipart/form-dataで送られた場合
content_type = os.environ.get('CONTENT_TYPE', '')
if content_type.startswith('multipart/form-data'):
request_method, query_string, query_params, form_storage = tkFormData.get_params()
self.query_string = query_string
self.query = query_params
self.form_data = form_storage
return self.query
# その他
content_length = os.environ.get('CONTENT_LENGTH', 0)
if content_length is not None and content_length.isdigit():
content_length = int(content_length)
self.query_string = sys.stdin.read(content_length)
else:
self.query_string = ''
elif os.environ.get('QUERY_STRING'):
self.query_string = os.environ.get('QUERY_STRING')
else:
self.query_string = '&'.join(sys.argv[1:])
params = self.parse_query_string(self.query_string)
self.path_info = self.get_path_info()
_params = self.get_path_args()
if len(_params) > 0:
params["action"] = _params[0]
self.query = params
return self.query
[ドキュメント]
def sanitize(self, value, allowed_characters = None):
"""
ユーザー入力をサニタイズする関数。HTMLタグを無効化し、特定の文字をエスケープします。
詳細説明:
入力が文字列でない場合は空文字列を返します。
`html.escape`を使用してHTML特殊文字をエスケープします。
`allowed_characters`が指定されている場合、それらの文字以外の文字をアンダースコアに置換します。
:param value: str, ユーザー入力の文字列。
:param allowed_characters: str or None, 許可する文字の集合を含む文字列。Noneの場合、文字置換は行われません。
:returns: str, サニタイズされた文字列。
"""
if not isinstance(value, str):
return ''
sanitized_value = html.escape(value)
if allowed_characters:
sanitized_value = re.sub(f"[^{re.escape(allowed_characters)}]", "_", sanitized_value)
return sanitized_value
[ドキュメント]
def desanitize(self, sanitized_input):
"""
サニタイズされた入力を元の形式に戻す関数。HTMLエスケープを解除します。
:param sanitized_input: str, サニタイズされた文字列。
:returns: str, 元の形式の文字列。
"""
if not isinstance(sanitized_input, str):
return ''
return html.unescape(sanitized_input)
[ドキュメント]
def render_template(self, template_file, params, extract_body = True):
"""
Jinja2テンプレートをレンダリングします。
詳細説明:
`template_file`がファイルパスの場合、その内容を読み込みます。
`extract_body`がTrueの場合、HTML文字列から`<body>`タグ内のコンテンツのみを抽出します。
指定された`params`辞書を使用してテンプレートをレンダリングし、結果の文字列を返します。
:param template_file: str, テンプレートファイル名（`self.templates_dir`からの相対パス）またはテンプレート文字列そのもの。
:param params: dict, テンプレートに渡す変数を格納した辞書。
:param extract_body: bool, Trueの場合、HTMLテンプレートから`<body>`タグの内容のみを抽出してレンダリングします。デフォルトはTrue。
:returns: str, レンダリングされたテンプレートの結果文字列。
"""
# template_stringがパスの場合、読み込む
template_string = None
template_path = os.path.join(self.templates_dir, template_file)
# print(f"{template_path=}<br>\n")
if os.path.isfile(template_path):
# print("this is file<br>")
try:
template_string = open(template_path).read()
except Exception as e:
print(f"Could not read [{template_path}] due to error [{e}]")
pass
# print(f"{template_string=}<br>")
if template_string is None:
template_string = template_file
if extract_body:
pattern = re.compile(r"<body>(.*?)</body>", re.IGNORECASE | re.DOTALL)
match = re.search(pattern, template_string)
if match:
template_string = match.group(1)
template_string = match.group(1)
template = Template(template_string)
text = template.render(params);
return text
[ドキュメント]
def render_template_from_file(self, template_name, context):
"""
指定されたファイル名のJinja2テンプレートをレンダリングします。
詳細説明:
`self.templates_dir`をテンプレートのルートとし、`template_name`で指定された
テンプレートファイルを読み込み、`context`辞書を適用してレンダリングします。
:param template_name: str, テンプレートディレクトリからの相対パスで指定されたテンプレートファイル名。
:param context: dict, テンプレートに渡す変数を格納した辞書。
:returns: str, レンダリングされたテンプレートの結果文字列。
"""
# テンプレートディレクトリ(スクリプトの実行ディレクトリからの相対パス)
env = Environment(loader=FileSystemLoader(self.templates_dir))
template = env.get_template(template_name)
return template.render(context)
[ドキュメント]
def get(self, url, params = None, headers=None):
"""
指定されたURLに対してHTTP GETリクエストを送信します。
:param url: str, リクエスト先のURL。
:param params: dict or None, クエリパラメータとして追加する辞書。デフォルトはNone。
:param headers: dict or None, リクエストヘッダーとして追加する辞書。デフォルトはNone。
:returns: requests.Response, HTTPレスポンスオブジェクト。
"""
return self.request("GET", url, params=params, headers=headers)
[ドキュメント]
def post(self, url, params=None, headers=None, body=None):
"""
指定されたURLに対してHTTP POSTリクエストを送信します。
詳細説明:
`body`が辞書型の場合、JSON文字列に変換され、`Content-Type: application/json`ヘッダーが設定されます。
:param url: str, リクエスト先のURL。
:param params: dict or None, クエリパラメータとして追加する辞書。デフォルトはNone。
:param headers: dict or None, リクエストヘッダーとして追加する辞書。デフォルトはNone。
:param body: dict or str or None, リクエストボディとして送信するデータ。辞書の場合はJSONに変換されます。デフォルトはNone。
:returns: requests.Response, HTTPレスポンスオブジェクト。
"""
if isinstance(body, dict):
body = json.dumps(body) # 辞書型データをJSONに変換
if headers is None:
headers = {}
headers["Content-Type"] = "application/json"
# headers["Content-Type"] = "application/json; charset=utf-8"
return self.request("POST", url, params=params, headers=headers, body=body)
[ドキュメント]
def put(self, url, params=None, headers=None, body=None):
"""
指定されたURLに対してHTTP PUTリクエストを送信します。
詳細説明:
`body`が辞書型の場合、JSON文字列に変換され、`Content-Type: application/json`ヘッダーが設定されます。
:param url: str, リクエスト先のURL。
:param params: dict or None, クエリパラメータとして追加する辞書。デフォルトはNone。
:param headers: dict or None, リクエストヘッダーとして追加する辞書。デフォルトはNone。
:param body: dict or str or None, リクエストボディとして送信するデータ。辞書の場合はJSONに変換されます。デフォルトはNone。
:returns: requests.Response, HTTPレスポンスオブジェクト。
"""
if isinstance(body, dict):
body = json.dumps(body)
if headers is None:
headers = {}
headers["Content-Type"] = "application/json"
# headers["Content-Type"] = "application/json; charset=utf-8"
return self.request("PUT", url, params=params, headers=headers, body=body)
[ドキュメント]
def delete(self, url, params=None, headers=None):
"""
指定されたURLに対してHTTP DELETEリクエストを送信します。
:param url: str, リクエスト先のURL。
:param params: dict or None, クエリパラメータとして追加する辞書。デフォルトはNone。
:param headers: dict or None, リクエストヘッダーとして追加する辞書。デフォルトはNone。
:returns: requests.Response, HTTPレスポンスオブジェクト。
"""
return self.request("DELETE", url, params=params, headers=headers)
[ドキュメント]
def request(self, method, url, params=None, headers=None, body=None):
"""
HTTPリクエストを実行してレスポンスを返します。
詳細説明:
指定されたHTTPメソッド（GET, POST, PUT, DELETE）に応じて、`requests`ライブラリを使用して
HTTPリクエストを送信します。GETおよびDELETEリクエストでは`body`は使用できません。
リクエスト中にエラーが発生した場合は、エラーメッセージを含む辞書を返します。
:param method: str, HTTPメソッド (GET, POST, PUT, DELETE)。
:param url: str, リクエスト先URL。
:param params: dict or None, クエリパラメータとして追加する辞書。
:param headers: dict or None, ヘッダー情報を含む辞書。
:param body: str or bytes or dict or None, ボディデータ（POSTやPUTの際に必要に応じて直接指定）。
:returns: requests.Response or dict, HTTPレスポンスオブジェクト、またはエラー発生時はエラー情報を含む辞書。
"""
try:
# GETリクエストではparamsをクエリパラメータとして使用
if method.upper() == "GET":
if body is not None:
print(f"Error in {self.script_filebody}:tkCGIApplication.request(): data should not given for GET")
exit()
response = requests.get(url, params=params, headers=headers)
# POSTリクエストではparamsをボディデータとして使用
elif method.upper() == "POST":
response = requests.post(url, params=params, data=body, headers=headers)
# PUTリクエストではparamsをボディデータとして使用
elif method.upper() == "PUT":
response = requests.put(url, params=params, data=body, headers=headers)
# DELETEリクエストではparamsをクエリパラメータとして使用
elif method.upper() == "DELETE":
if body is not None:
print(f"Error in {self.script_filebody}:tkCGIApplication.request(): data should not given for DELETE")
exit()
response = requests.delete(url, params=params, headers=headers)
# その他のHTTPメソッドはエラーとして処理
else:
return {"error": f"Unsupported HTTP method: {method}"}
return response
"""
# レスポンスデータを整形して返す
return {
"status": response.status_code,
"reason": response.reason,
"headers": dict(response.headers),
"body": response.text,
"json": response.json,
}
"""
except requests.exceptions.RequestException as e:
# エラーハンドリング
return {"error": str(e)}
[ドキュメント]
def call_route_function(self, action, query, *args, **kwargs):
"""
指定されたアクション名に対応するルーティング関数を呼び出します。
詳細説明:
`self.routes`辞書から`action`キーに対応する関数を探し、存在すればそれを実行します。
アクションが登録されていない場合は、警告をログに記録し、未対応アクションを示す辞書を返します。
:param action: str, 呼び出す関数のキーとなるアクション名。
:param query: dict, リクエストからパースされたクエリパラメータの辞書。
:param args: tuple, 関数に渡す位置引数。
:param kwargs: dict, 関数に渡すキーワード引数。
:returns: any, ルーティング関数の戻り値、または未対応アクションを示す辞書。
"""
# print()
# print(f"start {self.script_filebody}:tkCGIApplication.call_route_function(): {action=} {query=}")
if action in self.routes:
# 対応する関数を呼び出す
route_function = self.routes[action]
return route_function(*args, **kwargs)
else:
# 未定義のアクションへの応答
logging.warning(f"Undefined action: {action}")
return {'result': False, 'message': f"Action '{action}' is not supported."}
[ドキュメント]
def run(self, redirect = True, error_handler = None): #, target = "html"):
"""
CGIアプリケーションのエントリーポイントです。リクエストを処理し、レスポンスを生成します。
詳細説明:
`redirect`がTrueの場合、`builtins.print`をカスタムの出力関数にリダイレクトします。
`get_params`を呼び出してリクエストパラメータを解析し、`action`パラメータに基づいて
登録されたルーティング関数を呼び出します。
ルーティング関数が見つからない場合は、エラーメッセージをJSON形式で出力します。
ルーティング関数の戻り値に応じて、HTMLまたはJSON形式で最終的な出力を処理します。
:param redirect: bool, Trueの場合、`builtins.print`を`self.print_custom`にリダイレクトします。デフォルトはTrue。
:param error_handler: function or None, カスタムエラーハンドラー。現在実装は使用されていません。デフォルトはNone。
:returns: any or None, ルーティング関数の戻り値、またはエラー発生時はNone。
"""
# print をリダイレクト
if redirect:
self.redirect_print()
# エラーハンドラーを登録（デフォルトハンドラーを使用可能）
'''
if error_handler is None:
self.register_error_handler(
404, lambda: print(json.dumps({'result': False, 'message': "404 Not Found."}, indent=4))
)
else:
self.register_error_handler(404, error_handler)
'''
# QUERY_STRING を取得
self.query = self.get_params()
# アクションのルーティング
action = self.query.get('action', '/')
if action not in self.routes and '/' + action in self.routes:
action = f"/{action}"
if action in self.routes:
ret = self.call_route_function(action, self.query)
# if ret:
# ret = ret.strip()
# if type(ret) is str and len(ret) >= 1 and ret[0] == '{':
if self.output_target == 'json' and ret is not None:
# `ret`が文字列であればそのまま出力、そうでなければJSONとしてダンプ
if isinstance(ret, str):
self.write(ret)
else:
self.write(json.dumps(ret, indent=4))
elif isinstance(ret, str) or isinstance(ret, tkHTMLElement):
self.write(ret)
self.end_html()
elif ret is None: # ルーティング関数がNoneを返した場合 (例: 認証失敗時など)
pass # すでにエラーメッセージが出力されているか、何も出力しない
else: # その他の型の場合、JSONとしてダンプを試みる
self.write(json.dumps(ret, indent=4)) # デフォルトでJSONとして出力
# self.end_html() # JSON出力なのでHTML終了タグは不要
return ret
else:
logging.error(f"action [{action}] is not supported.")
response = {
'result': False,
'message': f"action [{action}] is not supported."
}
# ルートが見つからない場合、JSONヘッダーが未設定の可能性もあるため、
# HTML初期化せずにContent-Typeを明示してJSONを出力する
self.init_html(target='json', force_init=True)
self.print_original(json.dumps(response, indent=4))
return None
# self.end_html() は、HTML出力が初期化され、かつJSONターゲットでない場合のみ呼び出す
# 上記の条件分岐で適切な出力が行われているため、ここに到達するケースは減るが、
# 念のため、初期化状態とターゲットを確認して呼び出す
if self.html_initialized and self.output_target != 'json':
self.end_html()
if __name__ == "__main__":
pass