2025-06-26

Python GUIの進捗ウインドウ

呼び出し例

            show_progress_wx("キャプチャ中...")
            try:
                update_progress_wx("キャプチャ中...")
                update_progress_wx("リサイズ中...")
            except Exception as e:
                close_progress_wx()

進捗ウインドウ

import multiprocessing
import wx
import sys
import time
import os
import threading

LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'log')
PROGRESS_FILE = os.path.abspath(os.path.join(LOG_DIR, 'progress.txt'))
CLOSE_FLAG_FILE = os.path.abspath(os.path.join(LOG_DIR, 'progress_close.flag'))

# サブプロセスで動く進捗ウィンドウ
class ProgressFrame(wx.Frame):
    def __init__(self, file_path, close_flag_path):
        super().__init__(None, title="進捗", style=wx.DEFAULT_FRAME_STYLE & ~(wx.RESIZE_BORDER | wx.MAXIMIZE_BOX))
        self.panel = wx.Panel(self)
        self.label = wx.StaticText(self.panel, label="処理中...", pos=(20, 20), size=(260, 30))
        self.SetSize((300, 100))
        self.Centre()
        self.file_path = file_path
        self.close_flag_path = close_flag_path
        self.last_msg = None
        self.timer = wx.Timer(self)
        self.Bind(wx.EVT_TIMER, self.on_timer, self.timer)
        self.timer.Start(200)
        self.Bind(wx.EVT_CLOSE, self.on_close)
        self.Show()

    def on_timer(self, event):
        try:
            # 終了フラグファイルがあればウィンドウを閉じる
            if os.path.exists(self.close_flag_path):
                self.timer.Stop()
                self.Destroy()
                return
            if not os.path.exists(self.file_path):
                return
            with open(self.file_path, "r", encoding="utf-8") as f:
                msg = f.read().strip()
            if msg != self.last_msg:
                self.label.SetLabel(msg)
                self.last_msg = msg
        except Exception:
            pass

    def on_close(self, event):
        self.timer.Stop()
        self.Destroy()

# サブプロセス本体
def progress_window_proc(file_path, close_flag_path):
    app = wx.App(False)
    frame = ProgressFrame(file_path, close_flag_path)
    app.MainLoop()

# 親プロセス側API
_progress_proc = None
_progress_file = None
_close_flag_file = None

def show_progress_wx(message="処理中...", file_path=None, close_flag_path=None):
    global _progress_proc, _progress_file, _close_flag_file
    if file_path is None:
        # log/progress.txtをデフォルトに
        if not os.path.exists(LOG_DIR):
            os.makedirs(LOG_DIR, exist_ok=True)
        file_path = PROGRESS_FILE
    if close_flag_path is None:
        close_flag_path = CLOSE_FLAG_FILE
    # 終了フラグを消しておく
    if os.path.exists(close_flag_path):
        try:
            os.remove(close_flag_path)
        except Exception:
            pass
    if _progress_proc is not None and _progress_proc.is_alive():
        update_progress_wx(message, file_path)
        return file_path
    with open(file_path, "w", encoding="utf-8") as f:
        f.write(message)
    _progress_proc = multiprocessing.get_context("spawn").Process(target=progress_window_proc, args=(file_path, close_flag_path))
    _progress_proc.start()
    _progress_file = file_path
    _close_flag_file = close_flag_path
    time.sleep(0.2)  # ウィンドウ起動待ち
    update_progress_wx(message, file_path)
    return file_path

def update_progress_wx(message, file_path=None):
    global _progress_file
    if file_path is None:
        file_path = _progress_file
    if file_path:
        try:
            with open(file_path, "w", encoding="utf-8") as f:
                f.write(message)
        except Exception:
            pass

def close_progress_wx(file_path=None, close_flag_path=None):
    global _progress_proc, _progress_file, _close_flag_file
    if close_flag_path is None:
        close_flag_path = _close_flag_file
    # 終了フラグファイルを作成
    if close_flag_path:
        try:
            with open(close_flag_path, "w", encoding="utf-8") as f:
                f.write("close")
        except Exception:
            pass
    if _progress_proc is not None:
        _progress_proc.join(timeout=1)
    # ファイルは削除しない
    _progress_proc = None
    _progress_file = None
    _close_flag_file = None

if __name__ == "__main__":
    import time
    fpath = show_progress_wx("最初のメッセージ")
    time.sleep(2)
    update_progress_wx("進捗を更新中...", fpath)
    time.sleep(2)
    close_progress_wx(fpath)

2025-01-13

python fastapi ログ出力（汎用メモ）

fastapiで下記を行う

リクエストボディ、レスポンスボディのログを自動出力
リクエストセッションごとに固有のIDをログに出力

main.py

from contextvars import ContextVar
from asgi_correlation_id import CorrelationIdMiddleware
import logging.config
from fastapi import FastAPI, Request
from pydantic import BaseModel

from schema import Item
from my_logger import setup_logger
# from my_logger_middleware import add_request_id, log_requests
from my_logger_middleware import log_requests


# # Context variable for request ID
# request_id_context: ContextVar[str] = ContextVar("request_id", default="")

# # Helper function to get the current request ID
# def get_request_id() -> str:
#     return request_id_context.get()

# Logger setup
setup_logger()
logger = logging.getLogger("appLogger")

# FastAPI application
app = FastAPI()
# app.middleware("http")(add_request_id)
app.add_middleware(CorrelationIdMiddleware, header_name="X-Request-ID")
app.middleware("http")(log_requests)

logger = logging.getLogger("appLogger")


@app.get("/example")
def example_endpoint(name: str):
    logger.info("Processing request - message: Starting processing")

    # print(f'item: {item}')
    # response_data = item
    response_data = {"message": "This is response body!", "code": "dummy"}

    logger.info("Processing request - message: Finished processing")

    return response_data

class RequestBodyModel(BaseModel):
    key: str
    value: str

@app.post("/log-request-body")
async def log_request_body(request: Request, body: RequestBodyModel):
    body_dict = body.dict()
    # logger.info(f"Request received - correlation_id: {current_correlation_id}, body: {body_dict}")
    logger.info('ログ確認')
    response_data = {"message": "Request body logged successfully!", "received": body_dict}

    # logger.info(f"Response sent - correlation_id: {current_correlation_id}, response: {response_data}")
    return response_data

@app.get("/")
def read_root():
    logger.info("GET ルートメソッド")
    return {"Hello": "World"}

my_logger_middleware.py

# from contextvars import ContextVar
# import uuid
from fastapi import Request
from fastapi.responses import JSONResponse
import logging

logger = logging.getLogger("appLogger")


# Context variable for request ID
# print('ContextVar')
# print(ContextVar)
# request_id_context: ContextVar[str] = ContextVar("request_id", default="")

# async def add_request_id(request: Request, call_next):
#     request_id = str(uuid.uuid4())
#     request_id_context.set(request_id)
#     response = await call_next(request)
#     return response


# リクエスト・レスポンス ロギング
async def log_requests(request: Request, call_next):
    # print(f'request: {request}')

    # リクエストボディロギング
    request_body = await request.body()
    request_message = request_body.decode("utf-8").replace("\\n", " ").replace("\\r", " ")
    logger.info(f"Request received - body: {request_message}")

    # OpenAPI関連の場合、処理終了
    if request.url.path.startswith(("/docs", "/redoc", "/openapi.json")):
        response = await call_next(request)
        return response

    # 各固有処理
    response = await call_next(request)

    # レスポンスロギング
    response_body = b""
    async for chunk in response.body_iterator:
        response_body += chunk
    response_message = response_body.decode("utf-8").replace("\\n", " ").replace("\\r", " ")
    logger.info(f"Response sent - body: {response_message}")

    return JSONResponse(content=response_body.decode("utf-8"))

my_logger.py

import os
import logging.config
import yaml
from datetime import datetime
# from logging import Filter
# from my_logger_middleware import request_id_context

# # add_request_id
# class AddRequestIdFilter(Filter):
#     def filter(self, record):
#         try:
#             # リクエストIDを取得しレコードに設定
#             record.request_id = request_id_context.get()
#         except LookupError:
#             # フォールバック値を設定
#             record.request_id = "N/A"
#         return True

def setup_logger():
    LOG_DIR = "..\logs"
    CONFIG_DIR = "..\config"
    os.makedirs(LOG_DIR, exist_ok=True)
    log_config_path = os.path.join(CONFIG_DIR, "logconf.yaml")
    with open(log_config_path, "r") as f:
        log_config = yaml.safe_load(f)

    # ログファイル名に日付設定
    log_filename = os.path.join(LOG_DIR, f"applog_{datetime.now().strftime('%Y%m%d')}.log")
    log_config['handlers']['file']['filename'] = log_filename

    logging.config.dictConfig(log_config)

logconf.yaml

version: 1
formatters:
  detailed:
    # format: "%(asctime)s - %(levelname)s - [request_id=%(request_id)s] [correlation_id=%(correlation_id)s] %(message)s"
    format: "%(asctime)s - %(levelname)s - [request_id=%(correlation_id)s] %(message)s"
filters:
  correlation_id:
    (): asgi_correlation_id.CorrelationIdFilter
  # add_request_id:
  #   (): my_logger.AddRequestIdFilter
handlers:
  file:
    class: logging.FileHandler
    level: DEBUG
    formatter: detailed
    filters: [correlation_id]
    # filters: [add_request_id]
    filename: applog_default.log
    encoding: utf-8
loggers:
  appLogger:
    level: DEBUG
    handlers: [file]
    propagate: no
root:
  level: DEBUG
  handlers: [file]

環境作成

uv init fastapi-app
cd .\fastapi-app\
uv add fastapi --extra standard
uv add asgi-correlation-id

2021-05-23

【GAS】Googleドキュメントのファイル一覧をスプレッドシートに書き出す

Google Apps Script Google Apps Script-スプレッドシート

参考
qiita.com

/** @OnlyCurrentDoc */
function outputFileListLink() {
  //フォルダIDを指定する
  var folder_id = "1K-Zig9uIKt9iH2XWQYe3nrngyzkyB3eA"; //URLの「～～folders/{この部分}」

  //フォルダ情報を取得する
  var folder = DriveApp.getFolderById(folder_id);
  //フォルダ情報からファイル一覧を取得する
  var files = folder.getFiles();

  var list = [];  //この変数のファイル名・URLが入っていきます
  //フォルダ内のファイル一覧を1件ずつ読み込む
  while(files.hasNext()) {
    //ファイル情報を1件読む
    var buff = files.next();
    //ファイル名取得
    var xlsName = buff.getName();
    //ファイル名の1文字目から"_"が最初に出てきた位置までの文字列を取得
    var nameIndex = xlsName.indexOf("_"); //何文字目に"_"が入っているか取得
    xlsName = xlsName.substring(0, nameIndex);  //1文字目から指定文字目までの文字列を取得

    //ファイルのURLを取得
    var xlsLink = buff.getUrl();
    //URLからダウンロード用のリンク文字列を作成
    xlsLink = xlsLink.replace("file/d/", "uc?export=download&id="); //文字を置換
    xlsLink = xlsLink.replace("/view?usp=drivesdk","")  //文字を置換
    xlsLink = "=HYPERLINK(\"" + xlsLink + "\")" //ハイパーリンクの関数を設定

    //ファイル名、URLを格納
    list.push([xlsName, xlsLink]);
  };

  // ファイルが0件なら処理終了
  if(list.length == 0) {
    return;
  }

  // 出力先シートを取得する
  var ss = SpreadsheetApp.getActive();  //アクティブになっているスプレッドシートを取得
  var sheet = ss.getSheetByName("シート2"); //シート2を取得
  // シートの内容をクリア
  sheet.clearContents();
  // 出力範囲を取得
  var rowIndex = 1; // The starting row of a range.
  var colIndex = 1; // The starting row of a column.
  range = sheet.getRange(
    rowIndex,      // 行始点位置
    colIndex,      // 列始点位置
    list.length,   // 行終点位置(ファイルの数を指定)
    list[0].length // 列終点位置(2(ファイルとURL)を指定)
  );

  // 対象の範囲にまとめて書き出します
  range.setValues(list);
}