ラベル Python の投稿を表示しています。 すべての投稿を表示
ラベル Python の投稿を表示しています。 すべての投稿を表示

2026年8月3日月曜日

Stable diffusion forge Neo を試す

Stable diffusion forge Neo を試す

概要

WAI-illustrious-SDXL v17.0 が出たのでそれを試すために forge Neo が必要だったのでインストールから試してみました

環境

  • macOS 26.5.2 (m2 pro mac mini)
  • Python 3.13.14
  • StableDiffusion webui forge Neo 2.28

Python 準備

  • pyenv install 3.13.14

プロジェクトのクローン

  • git clone https://github.com/Haoming02/sd-webui-forge-classic sd-webui-forge-neo --branch neo --depth 1
  • cd sd-webui-forge-neo
  • pyenv local 3.13.14
  • pip install uv

プロジェクトのセットアップ

  • uv venv venv --python 3.13 --seed
  • vim webui-user.sh
# export TORCH_COMMAND="pip install torch==2.12.0 torchvision==0.27.0"

↓

export TORCH_COMMAND="pip install torch==2.12.0 torchvision==0.27.0"

webui 起動

  • chmod +x webui-user.sh
  • ./webui-user.sh

webui 動作確認

localhost:7860 にアクセスできれば OK です

モデルのダウンロードとインストール

https://civitai.red/models/827184/wai-illustrious-sdxl ここからダウンロードします
6.46GB ほどあります

ダウンロードしたら以下に配置します

ls -l sd-webui-forge-neo/models/Stable-diffusion/

モデルを再配置したら念のため webui を再起動しましょう

動作確認

再度 webui にアクセスしモデルが読み込まれていることを確認します
そしてプロンプトとパラメータを入力します

  • Steps: 15-30
  • CFG scale: 5-7
  • Sampler: Euler a
  • Size: 1024x1024
  • Positive Prompt: ,masterpiece,best quality,amazing quality,
  • Negative Prompt: bad quality,worst quality,worst detail,sketch,censor,

あとはプロンプトを適当に調整し「Generate」してください

最後に

macOS でも普通に動作しました

参考サイト

時刻: 0 件のコメント:
ラベル: ,

2026年7月17日金曜日

Gemini CLI + litellm で独自のエンドポイントを使う方法

Gemini CLI + litellm で独自のエンドポイントを使う方法

概要

前回 Codex 版を紹介しましたが今回は Gemini CLI での方法を紹介します

環境

  • Ubuntu 24.04
  • nodejs 22.21.0
  • gemini-cli 0.51.0

インストール

  • npm install -g @google/gemini-cli

構成概要

  • gemini-cli -> python proxy (docker) -> litellm (docker) -> custom endpoint

という流れになっています
python proxy は nginx などでも代用できるケースはありますが今回はリクエストを削る必要があるため python (fastapi) を使っています

理由は custom endpoint が gemini に完全に互換しているわけではなく使える機能により受け付けないパラメータがあるためそれを削るために用意しています
なので custom endpoint が完全互換であれば litellm だけで OK です

compose.yaml

gemini-proxy が python 製のプロキシになります
fastapi で gemini-cli から来たリクエストを受け使えないパラメータを削った後 litellm にリクエストします

services:
  litellm:
    image: docker.litellm.ai/berriai/litellm:latest
    container_name: litellm
    environment:
      - AI_SERVICE_API_KEY=${AI_SERVICE_API_KEY}
    ports:
      - "4000:4000"
    volumes:
      - ./litellm_config.yaml:/app/config.yaml
    restart: unless-stopped
    command: ["--config", "/app/config.yaml"]

  gemini-proxy:
    # Python proxy for gemini CLI (native Gemini API format).
    # Replaces nginx because request body JSON transformation is required:
    # strips unsupported 'id' field from function_call/function_response,
    # rewrites :streamGenerateContent?alt=sse → :streamGenerateContentSse,
    # and injects AI_SERVICE_API_KEY as x-goog-api-key header.
    # Configure gemini CLI to use port 4001 instead of 4000.
    build:
      context: .
      dockerfile: Dockerfile.proxy
    container_name: gemini-proxy
    environment:
      - AI_SERVICE_API_KEY=${AI_SERVICE_API_KEY}
    ports:
      - "4001:4001"
    restart: unless-stopped

proxy.py

API を互換させるためのプロキシです
先ほども紹介しましたが custom endpoint が完全互換の場合は不要です

#!/usr/bin/env python3
"""
Proxy for gemini CLI → Your custom Gemini endpoint.

Handles:
- URL mapping: :streamGenerateContent?alt=sse → :streamGenerateContentSse
- Request body: strip unsupported 'id' field from function_call / function_response
- Auth: replace incoming key with AI_SERVICE_API_KEY as x-goog-api-key header
"""

import json
import os

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import Response, StreamingResponse

CUSTOM_BASE = os.environ.get(
    "CUSTOM_BASE",
    "https://your-gemini-custom-endpoint",
)
API_KEY = os.environ["AI_SERVICE_API_KEY"]

app = FastAPI()


def strip_unsupported_fields(body: dict) -> dict:
    """Strip 'id' from function_call/function_response — field not supported by this endpoint."""
    for content in body.get("contents", []):
        for part in content.get("parts", []):
            for key in ("functionCall", "function_call"):
                if isinstance(part.get(key), dict):
                    part[key].pop("id", None)
            for key in ("functionResponse", "function_response"):
                if isinstance(part.get(key), dict):
                    part[key].pop("id", None)
    return body


@app.post("/v1beta/models/{model}:{method}")
async def proxy(request: Request, model: str, method: str):
    is_streaming = method == "streamGenerateContent"
    # Custom endpoint uses a dedicated SSE endpoint instead of ?alt=sse query param
    target_method = "streamGenerateContentSse" if is_streaming else method
    target_url = f"{CUSTOM_BASE}/{model}:{target_method}"

    body_bytes = await request.body()
    try:
        body = strip_unsupported_fields(json.loads(body_bytes))
        body_bytes = json.dumps(body, ensure_ascii=False).encode()
    except (json.JSONDecodeError, AttributeError):
        pass

    upstream_headers = {"Content-Type": "application/json", "x-goog-api-key": API_KEY}

    if is_streaming:

        async def generate():
            async with httpx.AsyncClient(timeout=120.0) as client:
                async with client.stream(
                    "POST", target_url, content=body_bytes, headers=upstream_headers
                ) as resp:
                    async for chunk in resp.aiter_bytes():
                        yield chunk

        return StreamingResponse(
            generate(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )
    else:
        async with httpx.AsyncClient(timeout=120.0) as client:
            resp = await client.post(
                target_url, content=body_bytes, headers=upstream_headers
            )
        return Response(
            content=resp.content,
            status_code=resp.status_code,
            media_type=resp.headers.get("content-type", "application/json"),
        )

Dockerfile.proxy

先程の proxy.py をビルドしてイメージを作成する dockerfile です

FROM python:3.12-slim
WORKDIR /app
RUN pip install --no-cache-dir fastapi httpx uvicorn
COPY proxy.py .
CMD ["uvicorn", "proxy:app", "--host", "0.0.0.0", "--port", "4001"]

litellm_config.yaml

gemini-3.1-flash-lite 側は基本不要ですが gemini-cli がデフォルトで gemini-3.1-flash-lite のモデルの定義を探しに行くのでエラーにならないように追加しています
pass_through_endpoints でパスのオーバライドをしていますがこれも状況によるので必須ではないです

model_list:
  - model_name: gemini-3.5-flash
    litellm_params:
      # gemini/ prefix is required to use API key auth instead of Vertex AI ADC.
      # Model name must match model_name above to avoid LiteLLM response model override error
      # that corrupts streaming responses. drop_params: true handles unsupported fields.
      model: gemini/gemini-3.5-flash
      api_base: https://your-gemini-custom-endpoint:generateContent
      api_key: os.environ/AI_SERVICE_API_KEY
      max_tokens: 4096
      temperature: 0.2
      rpm: 30
      tpm: 60000
  - model_name: gemini-3.1-flash-lite
    litellm_params:
      # gemini/ prefix is required to use API key auth instead of Vertex AI ADC.
      model: gemini/gemini-3.5-flash
      api_base: https://your-gemini-custom-endpoint:generateContent
      api_key: os.environ/AI_SERVICE_API_KEY

litellm_settings:
  drop_params: true

general_settings:
  global_max_parallel_requests: 1
  max_parallel_requests: 1
  # Pass-through: forward native Gemini API requests (/v1beta/models/...) directly
  # to the upstream endpoint, bypassing LiteLLM's model routing and SDK layer.
  # This avoids the "cannot override response model" bug caused by GenerateContentResponse
  # not having a 'model' attribute in LiteLLM's google_genai provider path.
  pass_through_endpoints:
    - path: "/v1beta/models/{path:path}"
      target: "https://your-gemini-custom-endpoint/gemini/{path}"
      headers:
        x-goog-api-key: os.environ/AI_SERVICE_API_KEY

router_settings:
  num_retries: 0
  retry_after: 20
  allowed_fails: 1
  cooldown_time: 180
  timeout: 120
  stream_timeout: 120

server_settings:
  port: 4000

動作確認

  • docker compose up -d
  • export GOOGLE_GEMINI_BASE_URL="http://localhost:4001"
  • gemini --model=gemini-3.5-flash

で「hi」とか打ってレスポンスが返ってくればとりあえず接続はできています

最後に

gemini-cli + litellm + custom endpoint の設定方法を紹介しました

独自のエンドポイントが提供している gemini などがあり API が完全互換していない場合は自分でリクエストやレスポンスを細工する必要があります
今回は fastapi + litellm で対応しましたが方法は様々なので自分に合う方法で互換させれば OK です

基本はエラーに合わせてプロキシや litellm 側で調整するしかないです

{
  "error": {
    "code": 400,
    "message": "Invalid JSON payload received. Unknown name \"id\" at 'contents[2].parts[0].function_call': Cannot find field.\nInvalid JSON payload received. Unknown name \"id\" at 'contents[3].parts[0].function_response': Cannot find field.",
    "status": "INVALID_ARGUMENT",
    "details": [
      {
        "@type": "type.googleapis.com/google.rpc.BadRequest",
        "fieldViolations": [
          {
            "field": "contents[2].parts[0].function_call",
            "description": "Invalid JSON payload received. Unknown name \"id\" at 'contents[2].parts[0].function_call': Cannot find field."
          },
          {
            "field": "contents[3].parts[0].function_response",
            "description": "Invalid JSON payload received. Unknown name \"id\" at 'contents[3].parts[0].function_response': Cannot find field."
          }
        ]
      }
    ]
  }
}

参考サイト

時刻: 0 件のコメント:
ラベル: , , ,

2026年5月12日火曜日

LTX2 を M2 Pro Mac mini で動かす

LTX2 を M2 Pro Mac mini で動かす

概要

MLX 版があるのでそれを使って M2 Pro 上で動作させてみます

環境

  • M2 Pro mac mini (16GB)
  • macOS 26.4.1
  • LTX2-MLX c4f8c44
  • Python 3.12.13

インストール

  • git clone --depth=1 https://github.com/Acelogic/LTX-2-MLX.git
  • cd LTX-2-MLX
  • pyenv local 3.12.13
  • uv sync --all-groups

モデルダウンロード

  • uv run scripts/download_weights.py

注意事項

gemma3 はライセンスに同意する必要がある

  • https://huggingface.co/google/gemma-3-12b-it -> Acknowledge license
  • huggingface のアカウントが必要
  • エラーになった場合は一度 rm -rf weights/gemma-3-12b してから再度 uv run scripts/download_weights.py する

distilled は 43GB あるので動かない場合は distilled-fp8 27GB を変わりにダウンロードし使うこと

  • uv run scripts/download_weights.py 実行時に 3 - Custom - Choose individual weights を選択しダウンロードするモデルを個別に選択する

動画生成(通常)

uv run python scripts/generate.py "A realistic cat riding a skateboard on a street, smoothly moving forward, natural motion, full body visible, cinematic lighting, high detail, 4k, camera tracking shot, shallow depth of field" --height 512 --width 512

おそらくこれは M2 Pro mac mini ではエラーになります

動画生成(fp8版)

uv run python scripts/generate.py "A realistic cat riding a skateboard on a street, smoothly moving forward, natural motion, full body visible, cinematic lighting, high detail, 4k, camera tracking shot, shallow depth of field" --height 512 --width 512 --fp8

画像サイズを下げれば何とかこれでもいけるかもしれませんがおそらくこれも不可能です

動画生成(fp8版+低メモリ)

uv run python scripts/generate.py "A realistic cat riding a skateboard on a street" --height 64 --width 96 --fp8 --low-memory --frames 9

おそらくこれでないと M2 Pro mac mini のスペックでは無理かなと思います
画像サイズとプロンプトの長さ、フレーム数も下げています

動画生成(fp8版+低メモリ+fp16フラグ)

uv run python scripts/generate.py "A realistic cat riding a skateboard on a street" --height 64 --width 96 --fp8 --low-memory --frames 9 --fp16 --pipeline one-stage --cfg 5.0

トラブルシューティング

libc++abi: terminating due to uncaught exception of type std::runtime_error: [METAL] Command buffer execution failed: Impacting Interactivity (0000000e:kIOGPUCommandBufferCallbackErrorImpactingInteractivity)

基本的にはメモリ不足が原因です

There appear to be 1 leaked semaphore objects to clean up at shutdown

これもメモリ不足の場合はほとんどです

最後に

LTX2-MLX を M2 Pro mac mini で動かしてみました
何とか動きますが量子化済みのモデルでないと動かないのと生成する動画サイズもかなり小さく短いものでないと無理でした

本当にギリギリなのでもしかしたら環境によっては動かない可能性もあるのでご了承ください

参考サイト

時刻: 0 件のコメント:
ラベル: , ,

2026年5月4日月曜日

kabuステーションAPIを使って株を売買するPythonスクリプト

kabuステーションAPIを使って株を売買するPythonスクリプト

概要

前回 kabuステーションAPIを使用して現在のポジションを取得しました
今回は株を売買する方法を紹介します

環境

  • M2 Pro Mac mini
  • macOS 26.4.1
  • UTM 4.7.5
    • Windows11 Home
  • kabuステーション 5.40.0.0
  • nginx 1.30.0
  • Python 3.12.13

株を購入する Python スクリプト

  • vim buy_stock.py
import time

import requests

# =========================
# 設定
# =========================
API_PASSWORD = "xxx"  # kabuステーションで設定したAPIパスワード
BASE_URL = "http://192.168.65.9:28081/kabusapi"  # 検証環境

SYMBOL = "1306"  # TOPIX ETF
EXCHANGE = 1  # 東証
QTY = 10  # 最小単位（検証なので10でOK）


# =========================
# トークン取得
# =========================
def get_token():
    url = f"{BASE_URL}/token"
    res = requests.post(url, json={"APIPassword": API_PASSWORD})

    if res.status_code != 200:
        print("トークン取得エラー:", res.text)
        return None

    return res.json()["Token"]


# =========================
# 共通ヘッダ
# =========================
def headers(token):
    return {"X-API-KEY": token}


# =========================
# 成行買い
# =========================
def buy_market(token):
    url = f"{BASE_URL}/sendorder"

    body = {
        "Symbol": SYMBOL,  # 銘柄コード
        "Exchange": EXCHANGE,  # 市場コード
        "SecurityType": 1,  # 商品種別 (1: 株式)
        "Side": 2,  # 売買区分 (2: 買)
        "CashMargin": 1,  # 信用区分 (1: 現物)
        "DelivType": 2,  # 受渡区分 (2: 預り金)
        "FundType": "02",  # 資産区分 (02: 保護)
        "AccountType": 4,  # 口座種別 (4: 特定)
        "Qty": QTY,  # 注文数量
        "FrontOrderType": 10,  # 執行条件 (10: 成行)
        "Price": 0,  # 注文価格
        "ExpireDay": 0,  # 注文有効期限 (0: 本日)
    }

    res = requests.post(url, json=body, headers=headers(token))

    if res.status_code != 200:
        print("注文エラー:", res.status_code, res.text)
        return False

    print("注文成功:", res.json())
    return True


# =========================
# ポジション取得
# =========================
def get_positions(token):
    url = f"{BASE_URL}/positions"
    res = requests.get(url, headers=headers(token))

    if res.status_code != 200:
        print("ポジション取得エラー:", res.text)
        return None

    return res.json()


# =========================
# メイン
# =========================
def main():
    token = get_token()
    if not token:
        return

    # ① 買う
    if not buy_market(token):
        return

    # 約定待ち（重要）
    print("約定待ち...")
    time.sleep(3)

    # ② ポジション取得
    positions = get_positions(token)

    print("=== 保有ポジション ===")
    if not positions:
        print("ポジションなし")
        return

    for p in positions:
        print(f"""
銘柄: {p.get('Symbol')}
数量: {p.get('LeavesQty')}
取得単価: {p.get('Price')}
現在値: {p.get('CurrentPrice')}
評価損益: {p.get('ProfitLoss')}
        """)


# =========================
if __name__ == "__main__":
    main()

売りの場合

上記のスクリプトで

  • Side -> 1
  • FundType -> " " (半角スペース2つ)

にすれば売りもできます

ログ格納場所

%appdata%\KabuS\Log

API リファレンス

https://kabucom.github.io/kabusapi/reference/index.html#operation/sendorderPost

body の各種パラメータはこちらを参照してください

あとは Github の Issue にいろいろ情報があります

https://github.com/kabucom/kabusapi/issues/

トラブルシューティング

注文エラー: 400 {"Code":1002,"Message":"数量は単位株数の倍数で入力してください"}

QTY = 10 以上にしましょう

注文エラー: 400 {"Code":4001005,"Message":"パラメータ変換エラー - 詳細はkabuSログファイルを確認してください"}

余計なパラメータがあったり不足しているパラメータがあるとこれが頻発します
自分は FundType が足りていませんでした

最後に

どうやら検証環境の場合は実際に購入することはできないようです
とりあえず以下のようになれば注文のリクエストは成功しています

注文成功: {'Result': 0, 'OrderId': None}

あとは各種パラーメタを調整すれば好きな銘柄の売買ができます

参考サイト

時刻: 0 件のコメント:
ラベル: ,

2026年5月3日日曜日

kabuステーションAPIを使って現在のポジション一覧を取得するPythonスクリプト

kabuステーションAPIを使って現在のポジション一覧を取得するPythonスクリプト

概要

前回 kabuステーションAPIを使用して認証トークンを取得しました
今回はポジションの一覧を取得します

環境

  • M2 Pro Mac mini
  • macOS 26.4.1
  • UTM 4.7.5
    • Windows11 Home
  • kabuステーション 5.40.0.0
  • nginx 1.30.0
  • Python 3.12.13

ポジションの一覧を取得する Python スクリプト

  • vim list_positions.py
import requests

# =========================
# 設定
# =========================
API_PASSWORD = "xxx"  # kabuステーションで設定したAPIパスワード
BASE_URL = "http://192.168.65.9:28081/kabusapi"  # 検証環境


# =========================
# トークン取得
# =========================
def get_token():
    url = f"{BASE_URL}/token"

    body = {"APIPassword": API_PASSWORD}

    res = requests.post(url, json=body)

    if res.status_code != 200:
        print("トークン取得エラー:", res.status_code, res.text)
        return None

    return res.json().get("Token")


# =========================
# ポジション取得
# =========================
def get_positions(token):
    url = f"{BASE_URL}/positions"

    headers = {"X-API-KEY": token}

    res = requests.get(url, headers=headers)

    # トークン期限切れ対応
    if res.status_code == 401:
        print("トークン期限切れ → 再取得")
        token = get_token()
        if not token:
            return None

        headers["X-API-KEY"] = token
        res = requests.get(url, headers=headers)

    if res.status_code != 200:
        print("ポジション取得エラー:", res.status_code, res.text)
        return None

    return res.json()


# =========================
# メイン処理
# =========================
def main():
    token = get_token()

    if not token:
        print("トークン取得失敗")
        return

    positions = get_positions(token)

    if not positions:
        print("ポジションなし or 取得失敗")
        return

    print("=== 保有ポジション ===")
    for p in positions:
        print(f"""
銘柄: {p.get('Symbol')}
数量: {p.get('LeavesQty')}
取得単価: {p.get('Price')}
現在値: {p.get('CurrentPrice')}
評価損益: {p.get('ProfitLoss')}
        """)


# =========================
if __name__ == "__main__":
    main()
  • uv run python list_positions.py

最後に

次回は実際に売買する方法を紹介します

時刻: 0 件のコメント:
ラベル: ,

2026年5月2日土曜日

kabuステーションAPIをコールして認証トークンを取得する

kabuステーションAPIをコールして認証トークンを取得する

概要

前回 kabuステーションをWindowsのVM上に構築しました
今回は実際に API をコールしトークンを取得します

環境

  • M2 Pro Mac mini
  • macOS 26.4.1
  • UTM 4.7.5
    • Windows11 Home
  • kabuステーション 5.40.0.0
  • nginx 1.30.0
  • Python 3.12.13

Windows 上に nginx を構築

Python スクリプトを Mac 上からコールするので kabuステーションが外部からアクセスできるように nginx を構築します

Windows 上で直接 Python スクリプトを実行してもいいですがいろいろとハマるので Mac 上から実行します

worker_processes  1;
events {
    worker_connections  1024;
}
http {
    include       mime.types;
    default_type  application/octet-stream;
    sendfile        on;
    keepalive_timeout  65;

    map $http_upgrade $connection_upgrade { 
    default upgrade;
    ''      close;
    } 

    server {
        listen       28081;
        server_name  localhost;

        proxy_http_version 1.1;
        proxy_set_header Host localhost;
        proxy_set_header Upgrade $http_upgrade; 
        proxy_set_header Connection $connection_upgrade;

        location / {
            proxy_pass   http://127.0.0.1:18081/;
        }
    }
}

28081 ポートでアクセスできるようにします
本番の場合は 18080 に proxy します

とりあえずテストなのでバイナリを直接ダブルクリックで実行すれば OK です
起動できているかは localhost:28081/kabusapi にアクセスしてみましょう
停止する場合はタスクマネージャーを使って nginx.exe プロセスを停止します

トークン取得用スクリプト

Windows 上で直接作成しても OK ですし Mac 側で編集/作成して VM 側に共有しても OK です

requests が必要なのでインストールしておきましょう

  • vim fetch_token.py
import requests

# =========================
# 設定
# =========================
API_PASSWORD = "xxx"  # kabuステーションで設定したAPIパスワード
BASE_URL = "http://192.168.65.9:28081/kabusapi"  # 検証環境


# =========================
# トークン取得
# =========================
def get_token():
    url = f"{BASE_URL}/token"

    body = {"APIPassword": API_PASSWORD}

    res = requests.post(url, json=body)

    if res.status_code != 200:
        print("ERROR:", res.status_code, res.text)
        return None

    data = res.json()
    return data.get("Token")


if __name__ == "__main__":
    token = get_token()

    if token:
        print("取得成功:")
        print(token)
    else:
        print("トークン取得失敗")
  • uv run python fetch_token.py

これでトークンが表示されれば OK です
API_PASSWORD は前回 API を有効にした際に設定したパスワードを入力しましょう

最後に

kabu ステーション API を使って API をコールするための認証トークンを取得してみました
kabu ステーション API は Windows 上に構築しスクリプトは Mac や Linux 上からコールするようにしましょう
今回はとりあえずバイナリを直接実行していますが本格的に取引を行う場合は自動起動やプロセス監視/ログ監視が必要になると思います

次回は API を使ってポジションの一覧を取得する方法を紹介します

参考サイト

時刻: 0 件のコメント:
ラベル: ,

2026年4月30日木曜日

Mac に kabuステーションAPI環境を構築する

Mac に kabuステーションAPI環境を構築する

概要

まずは構築しAPIを有効にします

環境

  • M2 Pro Mac mini
  • macOS 26.4.1
  • UTM 4.7.5
    • Windows11 Home
  • kabuステーション 5.40.0.0

UTM/CrystalFetch インストール

公式サイトからダウンロードしインストールします

CrystalFetch で Windows11 の ISO イメージ取得

AppleSillicon 上で動作させるのでチェックをオンにします

UTM で Windows11 の VM 作成

  • ライセンスキーがあれば入れる
  • 各種 Windows のセットアップ
  • display output is not active に注意
    • インストール時は「virtio-ramfb-gl (GPU Supported)」 Retinaオン
    • Windowsインストール後は「virtio-grpu-gl-pci (GPU Supported)」Retinaオフ
  • 全画面時の解除はCtrl+Altでマウスのフォーカスを外す

kabuステーションのダウンロードとインストール

https://download.r30.kabu.co.jp/ap/kabustation/ から exe をダウンロードしインストールします

三菱UFJ eスマート証券のアカウントでログイン

口座番号とパスワードでログインします

kabuステーションの設定からAPIを有効化

  • 右上の「歯車」
  • 「システム設定」
  • API タブ
  • 「APIを利用する」にチェック
  • APIパスワードを両方入力
  • kabuステーションを再起動し右上のAPIマークが有効になっていることを確認

動作確認

ブラウザで http://localhost:18081/kabusapi にアクセスしリクエスト不正になることを確認します

最後に

次回は API をコールするためのトークンを取得します

時刻: 0 件のコメント:
ラベル: ,

2026年4月21日火曜日

pipenv から uv に移行する

pipenv から uv に移行する

概要

自分のプロジェクトを pipenv から uv に置き換えたので手順を紹介します

環境

  • macOS 26.3.1
  • Python 3.12.13
  • uv 0.11.7

流れ

  1. pyproject.toml の作成
  2. Dockerfile の修正

python 準備

  • cat .python-version
3.12.13

uv インストール

  • pip install uv

pyproject.toml の作成

[project]
name = "ai-gallery"
version = "0.1.0"
description = "AI Gallery application"
requires-python = ">=3.12"
dependencies = [
    "flask",
    "google-cloud-storage",
    "redis",
    "gunicorn",
    "google-genai",
    "pillow",
    "py-vapid",
    "pywebpush",
]

[dependency-groups]
dev = [
    "black",
    "isort",
    "pyright",
    "pytest",
    "bandit",
]

[tool.black]
line-length = 88
target-version = ["py312"]

[tool.isort]
profile = "black"
line_length = 88

[tool.pyright]
pythonVersion = "3.12"

uv sync

  • uv sync

これで uv.lock が作成されます
仮想環境のパスはプロジェクト直下の .venv に作成されます

Dockerfile 修正

pipenv -> uv に変更します

diff --git a/Dockerfile b/Dockerfile
index 499f280..4ec6df7 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,12 +1,16 @@
 FROM python:3.12.13-alpine3.23
 
-COPY . /home
 WORKDIR /home
 
-RUN pip install pipenv
-RUN pipenv install
+COPY pyproject.toml uv.lock ./
+
+RUN --mount=type=cache,target=/root/.cache/pip \
+    pip install uv && \
+    uv export --no-dev | pip install --no-cache-dir -r /dev/stdin
+
+COPY . .
 
 ENV WAIT=on
 ENV MODE=1
 
-CMD ["pipenv", "run", "python", "app.py"]
+CMD ["python", "app.py"]

今回は system 環境にインストールするようにしています

uv sync でも以下のような流れで動作するのでこの当たりは好きなように修正してください

RUN --mount=type=cache,target=/root/.cache/pip \
    pip install uv && \
    uv sync --frozen --no-install-project --no-dev
CMD ["uv", "run", "--no-sync", "gunicorn", "-w", "1", "-b", "0.0.0.0:8080", "app:app"]

--no-sync オプションは実行時に uv sync を実行しないための設定です

一応 docker build -> run して動作確認しておくといいでしょう

README 修正

pipenv で実行している部分を uv に書き換えます
例えば以下のような部分を書き換えていきます

-- pipenv install -d
+- uv sync --all-groups

-- pipenv run isort . && pipenv run black . && pipenv run pyright . && npm run format
+- uv run isort . && uv run black . && uv run pyright . && npm run format

-- pipenv run pytest test/ && pipenv run bandit -r . && REDIS_PASSWORD="" pipenv run gunicorn -w 1 -b 0.0.0.0:8080 app:app
+- uv run pytest test/ && uv run bandit -r . && REDIS_PASSWORD="" uv run gunicorn -w 1 -b 0.0.0.0:8080 app:app

-- pipenv requirements > requirements.txt
+- uv export > requirements.txt

.dockerignore 修正

node_modules/
.venv/

compose.yaml 修正

pipenv -> uv のコマンド修正をします

command: pipenv run python check_new_message.py

↓

command: uv run python check_new_message.py

各種シェルスクリプトの修正

あれば修正します
同様に pipenv -> uv で OK です

Github Actions 修正

各種 yaml ファイルを修正しましょう

diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index 50d8f31..48e8a98 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -16,11 +16,11 @@ jobs:
         uses: actions/setup-python@v5
       - name: Install dependencies
         run: |
-          python -m pip install --upgrade pipenv
-          pipenv install -d
+          python -m pip install --upgrade uv
+          uv sync --all-groups
       - name: Lint with isort
         run: |
-          pipenv run isort --check --diff .
+          uv run isort --check --diff .
       - name: Lint with black
         run: |
-          pipenv run black --check .
+          uv run black --check .

実際にアクションを実行して動作確認しましょう

vscode の設定の修正

  • .vscode/settings.json
diff --git a/.vscode/settings.json b/.vscode/settings.json
index 8a2fea7..1bd2c9b 100644
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@@ -1,4 +1,3 @@
 {
-  "python-envs.defaultEnvManager": "ms-python.python:pipenv",
   "python-envs.pythonProjects": []
 }

これでプロジェクト配下にある .venv を自動で読み込んでくれます

.env ファイルを使っている場合の注意事項

  • uv run --env-file .env

のように命じ的に指定しないと読み込んでくれないので注意しましょう

移行後

  • Pipfile と Pipfile.lock の削除
  • .isort.cfg の削除
  • pip uninstall pipenv
  • venv の削除
    • rm -rf ~/.local/share/virtualenvs/ai-gallery-nT0ArqSL
  • pipenv コマンドがプロジェクト内で使われていないか確認
    • git grep 'pipenv'
  • GAE などにもデプロイしている場合は動作確認
uv export > requirements.txt
gcloud app deploy -q

最後に

アプリ自体を修正するのは簡単ですが CI や各種スクリプトで使われているコマンドを書き換えるのが面倒かなという印象です

時刻: 0 件のコメント:
ラベル: ,

2026年2月13日金曜日

サーバから Chrome ブラウザプッシュする実装の流れ

サーバから Chrome ブラウザプッシュする実装の流れ

概要

実装の流れを紹介します

環境

  • macOS 26.2
  • Python 3.12.11

鍵の作成

こちらを参考に作成します
鍵ファイルと JavaScript 用の鍵文字列を作成すれば OK です

サブスクリプション情報の保存 (Python)

JavaScrpt 側で通知の許可をした場合に通知先の情報をサーバ側に保存しておきます
今回は Redis で紹介しますが永続化できるものであれば何でも OK です

@app.route("/subscribe", methods=["POST"])
def subscribe():
    """フロントから送られてくる Subscription 情報を Redis に保存"""
    data = request.get_json()
    client_id = data.get("client_id")
    subscription_info = data.get("subscription")

    redis_conn = get_kv()
    # client_id ごとに購読情報を保存
    redis_conn.set(f"push_sub:{client_id}", json.dumps(subscription_info))
    return jsonify({"ok": True})

プッシュ送信処理 (Python)

あとで使いますが実際にプッシュを送信する処理です
メールアドレス部分はダミーでも問題ないですが一応ちゃんと設定した方がいいです (公開アプリの場合)

VAPID_PRIVATE_KEY_PATH = os.getenv("VAPID_PRIVATE_KEY_PATH", "./private_key.pem")
VAPID_CLAIMS = {"sub": "mailto:user@mail"}

def send_web_push(subscription_info, message_body):
    try:
        webpush(
            subscription_info=json.loads(subscription_info),
            data=json.dumps(message_body),
            vapid_private_key=VAPID_PRIVATE_KEY_PATH,
            vapid_claims=VAPID_CLAIMS,  # type: ignore
        )
    except WebPushException as ex:
        print("Push failed:", repr(ex))

ServiceWorker (JavaScript)

リモートプッシュはこの ServiceWorker がないと動きません
Python からのプッシュをこの ServiceWorker が受け取り通知を表示する感じです

通知のスタイルなどはシンプルにしているので必要であればアイコンなど設定してください

self.addEventListener("push", function (event) {
  console.log("Push received");
  let data = { title: "新着", body: "メッセージがあります" };

  try {
    if (event.data) {
      data = event.data.json(); // ここでエラーが起きると通知が出ない
    }
  } catch (e) {
    console.error("JSON parse error:", e);
  }

  const options = {
    body: data.body,
  };

  event.waitUntil(self.registration.showNotification(data.title, options));
});

ブラウザ側のサブスクリプション登録 (JavaScript)

ブラウザで特定のサイトなどにアクセスすると通知を許可しますかなど出るサイトがありますがそれを実装します
許可するをクリックしたら必要な情報を Python 側に送ります

urlBase64ToUint8Array に鍵文字列は冒頭で作成したものを使ってください

function urlBase64ToUint8Array(base64String) {
  const padding = "=".repeat((4 - (base64String.length % 4)) % 4);
  const base64 = (base64String + padding).replace(/-/g, "+").replace(/_/g, "/");
  const rawData = window.atob(base64);
  const outputArray = new Uint8Array(rawData.length);
  for (let i = 0; i < rawData.length; ++i) {
    outputArray[i] = rawData.charCodeAt(i);
  }
  return outputArray;
}

async function setupPush() {
  const register = await navigator.serviceWorker.register("/static/js/sw.js");

  // すでに購読済みかチェック
  let subscription = await register.pushManager.getSubscription();

  // 未購読なら新規作成
  if (!subscription) {
    subscription = await register.pushManager.subscribe({
      userVisibleOnly: true,
      applicationServerKey: urlBase64ToUint8Array(
        "xxx",
      ),
    });
  }

  // サーバーへ送信（常に送ることで client_id との紐付けを確実に更新する）
  await fetch("/subscribe", {
    method: "POST",
    body: JSON.stringify({
      client_id: CLIENT_ID,
      subscription: subscription,
    }),
    headers: { "Content-Type": "application/json" },
  });
}

通知許可ダイアログ表示 (JavaScript)

ページが読み込まれた場合に自動で出してもいいですが嫌がられるケースもあるので特定のボタンを押したら許可する流れにします

document.addEventListener("DOMContentLoaded", () => {
  // Click title to set client_id and request notification permission
  const title = document.getElementById("notification");
  if (title) {
    title.addEventListener("click", async () => {
      // 通知の許可を求める
      if ("Notification" in window) {
        const permission = await Notification.requestPermission();
        if (permission === "granted") {
          // ここで実行！
          await setupPush();
        }
      }
    });
  }
  // 追加: すでに許可済みなら、ページ読み込み時に購読情報を同期する
  if ("Notification" in window && Notification.permission === "granted") {
    setupPush().catch((err) => console.error("Auto-setup failed:", err));
  }
});

プッシュ通知送信タイミング (Python)

あとは任意のタイミングでサーバ側で送信すれば OK です
当然ですが送信しまくると大変なことになるので条件など設けて適切なタイミングで送るようにしましょう

@app.route("/send", methods=["POST"])
def redis_chat_send():
    # 1... 既存の処理
    # 2. プッシュ通知の送信 (自分以外の購読者へ)
    try:
        # push_sub: で始まるキーをすべて取得
        # ※ ユーザー数が多い場合はメンバーリストを別途管理するのが理想ですが、
        # 現状のチャットルーム運用であれば keys で十分です
        sub_keys = redis_conn.keys("push_sub:*")

        for sub_key in sub_keys:  # type: ignore
            target_id = sub_key.decode().split(":")[-1]

            # 送信者（自分）以外にのみ送る
            if target_id != client_id:
                sub_info = redis_conn.get(sub_key)
                if sub_info:
                    logger.info(f"Sending push to {target_id}")
                    # 非同期処理にするのが理想ですが、まずは直列で実装
                    send_web_push(
                        sub_info,
                        {"title": "新着メッセージ", "body": f"{client_id}: {msg}"},
                    )
    except Exception as e:
        logger.error(f"Push notification error: {e}")

    return jsonify({"ok": True})

サブスクリプション確認

  • Redis 側にキーが登録されていること
  • 開発者コンソールで ServiceWorker が登録されていること

プッシュが届かない場合

  • Mac であれば通知の設定を確認しましょう
    • Mac の Chrome はデフォルトでオフになっている
    • Chrome で許可していても Mac 側で拒否していることが多い
  • サイトが https 化されていない
    • localhost ならテストできるようです
    • IP ベースのサイトや http のサイトでは動作しません

スマホ or タブレット版でプッシュ通知が届かない場合

  • ラップトップ版と違いかなり作りが違う
  • 単純に通知を許可することができない
  • iOS 側で当然許可する必要があるがそれでもサブスクリプションできないことがある
  • サイトを PWA 化して上げる必要などがあるかも

最後に

参考サイト

時刻: 0 件のコメント:
ラベル: , ,
登録: 投稿 (Atom)