ponshu-room-lite/tools/synology/ai-proxy/server.py

#!/usr/bin/env python3
"""
Ponshu Room AI Proxy Server
============================
役割: アプリからのリクエストを受け取り、Gemini APIに中継
機能:
  - APIキーの保護（アプリにAPIキーを埋め込まない）
  - デバイスID認証（許可されたデバイスのみアクセス可能）
  - レート制限（1日10回まで/デバイス）
  - 使用状況のログ記録
"""

import os
import json
import hashlib
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional

from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import httpx

# ==========================================
# 設定
# ==========================================
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "")
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-1.5-flash")
RATE_LIMIT_PER_DAY = int(os.getenv("RATE_LIMIT_PER_DAY", "10"))
DATA_DIR = Path("/app/data")
USAGE_FILE = DATA_DIR / "usage.json"

# アプリ起動時にデータディレクトリを作成
DATA_DIR.mkdir(exist_ok=True)

# ==========================================
# FastAPI アプリ初期化
# ==========================================
app = FastAPI(
    title="Ponshu Room AI Proxy",
    description="Gemini API Proxy with Rate Limiting",
    version="1.0.0"
)

# CORS設定（モバイルアプリからのリクエストを許可）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 本番環境では特定のオリジンに制限すべき
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ==========================================
# データモデル
# ==========================================
class AnalyzeRequest(BaseModel):
    """画像解析リクエスト"""
    device_id: str  # デバイスID（SHA256ハッシュ化されたもの）
    images: List[str]  # Base64エンコードされた画像データ
    prompt: Optional[str] = None  # カスタムプロンプト（オプション）

class AnalyzeResponse(BaseModel):
    """画像解析レスポンス"""
    success: bool
    data: Optional[Dict] = None
    error: Optional[str] = None
    usage: Dict[str, int] = {}  # {"today": 3, "limit": 10}

# ==========================================
# 使用状況管理
# ==========================================
def load_usage_data() -> Dict:
    """使用状況データを読み込む"""
    if not USAGE_FILE.exists():
        return {}

    try:
        with open(USAGE_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}

def save_usage_data(data: Dict) -> None:
    """使用状況データを保存"""
    with open(USAGE_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

def get_today_key() -> str:
    """今日の日付キー（YYYY-MM-DD）を取得"""
    return datetime.now().strftime("%Y-%m-%d")

def check_rate_limit(device_id: str) -> tuple[bool, int]:
    """
    レート制限をチェック

    Returns:
        (制限内か, 今日の使用回数)
    """
    usage_data = load_usage_data()
    today = get_today_key()

    # デバイスIDのエントリがない場合は作成
    if device_id not in usage_data:
        usage_data[device_id] = {}

    # 今日の使用回数を取得
    today_count = usage_data[device_id].get(today, 0)

    # 制限チェック
    is_within_limit = today_count < RATE_LIMIT_PER_DAY

    return is_within_limit, today_count

def increment_usage(device_id: str) -> int:
    """
    使用回数をインクリメント

    Returns:
        更新後の今日の使用回数
    """
    usage_data = load_usage_data()
    today = get_today_key()

    if device_id not in usage_data:
        usage_data[device_id] = {}

    usage_data[device_id][today] = usage_data[device_id].get(today, 0) + 1

    save_usage_data(usage_data)

    return usage_data[device_id][today]

def cleanup_old_usage_data() -> None:
    """30日以上前のデータを削除（データサイズ管理）"""
    usage_data = load_usage_data()
    cutoff_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")

    for device_id in usage_data:
        dates_to_remove = [
            date for date in usage_data[device_id]
            if date < cutoff_date
        ]
        for date in dates_to_remove:
            del usage_data[device_id][date]

    save_usage_data(usage_data)

# ==========================================
# APIエンドポイント
# ==========================================
@app.get("/")
async def root():
    """ヘルスチェック"""
    return {
        "service": "Ponshu Room AI Proxy",
        "status": "running",
        "version": "1.0.0"
    }

@app.get("/health")
async def health_check():
    """詳細なヘルスチェック"""
    return {
        "status": "healthy",
        "gemini_api_key_configured": bool(GEMINI_API_KEY),
        "rate_limit": RATE_LIMIT_PER_DAY,
        "model": GEMINI_MODEL
    }

@app.post("/analyze", response_model=AnalyzeResponse)
async def analyze_sake_label(request: AnalyzeRequest):
    """
    日本酒ラベル画像を解析

    レート制限: 1デバイスあたり1日10回まで
    """

    # 1. APIキーのチェック
    if not GEMINI_API_KEY:
        raise HTTPException(
            status_code=500,
            detail="Gemini APIキーが設定されていません"
        )

    # 2. デバイスIDの検証（SHA256形式かチェック）
    if len(request.device_id) != 64:  # SHA256は64文字
        raise HTTPException(
            status_code=400,
            detail="無効なデバイスIDです"
        )

    # 3. レート制限チェック
    is_within_limit, today_count = check_rate_limit(request.device_id)

    if not is_within_limit:
        return AnalyzeResponse(
            success=False,
            error=f"本日の解析上限（{RATE_LIMIT_PER_DAY}回）に達しました。明日またお試しください。",
            usage={
                "today": today_count,
                "limit": RATE_LIMIT_PER_DAY
            }
        )

    # 4. Gemini APIにリクエストを送信
    try:
        # プロンプトのデフォルト値
        prompt = request.prompt or """
この日本酒のラベル画像（複数枚ある場合は表・裏など）を分析してください。
全ての画像から情報を統合し、以下の情報をJSON形式で返してください：

{
  "name": "銘柄名（例：獺祭 純米大吟醸）",
  "brand": "蔵元名（例：旭酒造）",
  "prefecture": "都道府県名（例：山口県）",
  "type": "特定名称（例：純米大吟醸, 本醸造, 普通酒 など）",
  "description": "この日本酒の特徴を100文字程度で説明（裏ラベルの情報があれば活用してください）",
  "catchCopy": "この日本酒を一言で表現する詩的なキャッチコピー（20文字以内）",
  "confidenceScore": 0から100の整数,
  "flavorTags": ["フルーティ", "辛口"],
  "tasteStats": {
    "aroma": 3,
    "sweetness": 3,
    "acidity": 3,
    "bitterness": 3,
    "body": 3
  },
  "alcoholContent": アルコール度数（数値のみ。例: 15.5）,
  "polishingRatio": 精米歩合（整数のみ。例: 50）,
  "sakeMeterValue": 日本酒度（数値。例: +3.0, -1.5）,
  "riceVariety": "酒米（例：山田錦、五百万石）",
  "yeast": "酵母（例：きょうかい9号、M310）",
  "manufacturingYearMonth": "製造年月（例：2023.10, 2023年10月）"
}

**tasteStatsの説明 (1-5の整数)**:
- aroma: 香りの強さ
- sweetness: 甘み
- acidity: 酸味
- bitterness: ビター感/キレ
- body: コク・ボディ

読み取れない情報は null を返してください。
JSONのみを返し、他の文章は含めないでください。
"""

        # Gemini APIエンドポイント
        gemini_url = f"https://generativelanguage.googleapis.com/v1beta/models/{GEMINI_MODEL}:generateContent"

        # リクエストボディの構築
        parts = [{"text": prompt}]
        for image_data in request.images:
            parts.append({
                "inline_data": {
                    "mime_type": "image/jpeg",
                    "data": image_data
                }
            })

        gemini_request = {
            "contents": [{
                "parts": parts
            }],
            "safetySettings": [
                {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"},
                {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"}
            ]
        }

        # HTTPクライアントでリクエスト送信
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                gemini_url,
                json=gemini_request,
                params={"key": GEMINI_API_KEY}
            )
            response.raise_for_status()

            gemini_response = response.json()

        # レスポンスから結果を抽出
        if "candidates" not in gemini_response or not gemini_response["candidates"]:
            raise HTTPException(
                status_code=500,
                detail="Gemini APIからの応答が不正です"
            )

        text_result = gemini_response["candidates"][0]["content"]["parts"][0]["text"]

        # JSONをパース（マークダウンのコードブロックを削除）
        text_result = text_result.strip().replace("```json", "").replace("```", "").strip()
        result_json = json.loads(text_result)

        # 5. 使用回数をインクリメント
        new_count = increment_usage(request.device_id)

        # 6. 成功レスポンス
        return AnalyzeResponse(
            success=True,
            data=result_json,
            usage={
                "today": new_count,
                "limit": RATE_LIMIT_PER_DAY
            }
        )

    except httpx.HTTPStatusError as e:
        raise HTTPException(
            status_code=e.response.status_code,
            detail=f"Gemini API error: {e.response.text}"
        )
    except json.JSONDecodeError:
        raise HTTPException(
            status_code=500,
            detail="Gemini APIからのJSON解析に失敗しました"
        )
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"予期しないエラー: {str(e)}"
        )

@app.get("/usage/{device_id}")
async def get_usage(device_id: str):
    """デバイスの使用状況を取得"""
    if len(device_id) != 64:
        raise HTTPException(status_code=400, detail="無効なデバイスIDです")

    _, today_count = check_rate_limit(device_id)

    return {
        "device_id": device_id[:8] + "...",  # プライバシー保護のため一部のみ表示
        "today": today_count,
        "limit": RATE_LIMIT_PER_DAY,
        "remaining": max(0, RATE_LIMIT_PER_DAY - today_count),
        "date": get_today_key()
    }

# ==========================================
# 起動時処理
# ==========================================
@app.on_event("startup")
async def startup_event():
    """サーバー起動時に古いデータをクリーンアップ"""
    cleanup_old_usage_data()
    print(f"🚀 AI Proxy Server started")
    print(f"   Model: {GEMINI_MODEL}")
    print(f"   Rate Limit: {RATE_LIMIT_PER_DAY} requests/day")

# ==========================================
# メイン実行
# ==========================================
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)