ai/card
1
0
Fork 0
Code Activity

fix gpt

Browse Source
This commit is contained in:
syui
2025-06-03 05:00:51 +09:00
parent 6cd8014f80
commit ef907660cc
43 changed files with 5255 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

290
api/app/ai_provider.py Normal file
View File

@ -0,0 +1,290 @@
"""AI Provider integration for ai.card"""
import os
import json
from typing import Optional, Dict, List, Any
from abc import ABC, abstractmethod
import logging
import httpx
from openai import OpenAI
import ollama
class AIProvider(ABC):
"""Base class for AI providers"""
@abstractmethod
async def chat(self, prompt: str, system_prompt: Optional[str] = None) -> str:
"""Generate a response based on prompt"""
pass
class OllamaProvider(AIProvider):
"""Ollama AI provider for ai.card"""
def __init__(self, model: str = "qwen3", host: Optional[str] = None):
self.model = model
self.host = host or os.getenv('OLLAMA_HOST', 'http://127.0.0.1:11434')
if not self.host.startswith('http'):
self.host = f'http://{self.host}'
self.client = ollama.Client(host=self.host, timeout=60.0)
self.logger = logging.getLogger(__name__)
self.logger.info(f"OllamaProvider initialized with host: {self.host}, model: {self.model}")
async def chat(self, prompt: str, system_prompt: Optional[str] = None) -> str:
"""Simple chat interface"""
try:
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
response = self.client.chat(
model=self.model,
messages=messages,
options={
"num_predict": 2000,
"temperature": 0.7,
"top_p": 0.9,
},
stream=False
)
return response['message']['content']
except Exception as e:
self.logger.error(f"Ollama chat failed: {e}")
return "I'm having trouble connecting to the AI model."
class OpenAIProvider(AIProvider):
"""OpenAI API provider with MCP function calling support"""
def __init__(self, model: str = "gpt-4o-mini", api_key: Optional[str] = None, mcp_client=None):
self.model = model
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
if not self.api_key:
raise ValueError("OpenAI API key not provided")
self.client = OpenAI(api_key=self.api_key)
self.logger = logging.getLogger(__name__)
self.mcp_client = mcp_client
def _get_mcp_tools(self) -> List[Dict[str, Any]]:
"""Generate OpenAI tools from MCP endpoints"""
if not self.mcp_client:
return []
tools = [
{
"type": "function",
"function": {
"name": "get_user_cards",
"description": "ユーザーが所有するカードの一覧を取得します",
"parameters": {
"type": "object",
"properties": {
"did": {
"type": "string",
"description": "ユーザーのDID"
},
"limit": {
"type": "integer",
"description": "取得するカード数の上限",
"default": 10
}
},
"required": ["did"]
}
}
},
{
"type": "function",
"function": {
"name": "draw_card",
"description": "ガチャを引いてカードを取得します",
"parameters": {
"type": "object",
"properties": {
"did": {
"type": "string",
"description": "ユーザーのDID"
},
"is_paid": {
"type": "boolean",
"description": "有料ガチャかどうか",
"default": False
}
},
"required": ["did"]
}
}
},
{
"type": "function",
"function": {
"name": "get_card_details",
"description": "特定のカードの詳細情報を取得します",
"parameters": {
"type": "object",
"properties": {
"card_id": {
"type": "integer",
"description": "カードID"
}
},
"required": ["card_id"]
}
}
},
{
"type": "function",
"function": {
"name": "analyze_card_collection",
"description": "ユーザーのカードコレクションを分析します",
"parameters": {
"type": "object",
"properties": {
"did": {
"type": "string",
"description": "ユーザーのDID"
}
},
"required": ["did"]
}
}
},
{
"type": "function",
"function": {
"name": "get_gacha_stats",
"description": "ガチャの統計情報を取得します",
"parameters": {
"type": "object",
"properties": {}
}
}
}
]
return tools
async def chat(self, prompt: str, system_prompt: Optional[str] = None) -> str:
"""Simple chat interface without MCP tools"""
try:
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=2000,
temperature=0.7
)
return response.choices[0].message.content
except Exception as e:
self.logger.error(f"OpenAI chat failed: {e}")
return "I'm having trouble connecting to the AI model."
async def chat_with_mcp(self, prompt: str, did: str = "user") -> str:
"""Chat interface with MCP function calling support"""
if not self.mcp_client:
return await self.chat(prompt)
try:
tools = self._get_mcp_tools()
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "あなたはai.cardシステムのアシスタントです。カードゲームの情報、ガチャ、コレクション分析などについて質問されたら、必要に応じてツールを使用して正確な情報を提供してください。"},
{"role": "user", "content": prompt}
],
tools=tools,
tool_choice="auto",
max_tokens=2000,
temperature=0.7
)
message = response.choices[0].message
# Handle tool calls
if message.tool_calls:
messages = [
{"role": "system", "content": "カードゲームシステムのツールを使って正確な情報を提供してください。"},
{"role": "user", "content": prompt},
{
"role": "assistant",
"content": message.content,
"tool_calls": [tc.model_dump() for tc in message.tool_calls]
}
]
# Execute each tool call
for tool_call in message.tool_calls:
tool_result = await self._execute_mcp_tool(tool_call, did)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": tool_call.function.name,
"content": json.dumps(tool_result, ensure_ascii=False)
})
# Get final response
final_response = self.client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=2000,
temperature=0.7
)
return final_response.choices[0].message.content
else:
return message.content
except Exception as e:
self.logger.error(f"OpenAI MCP chat failed: {e}")
return f"申し訳ありません。エラーが発生しました: {e}"
async def _execute_mcp_tool(self, tool_call, default_did: str = "user") -> Dict[str, Any]:
"""Execute MCP tool call"""
try:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
if function_name == "get_user_cards":
did = arguments.get("did", default_did)
limit = arguments.get("limit", 10)
return await self.mcp_client.get_user_cards(did, limit)
elif function_name == "draw_card":
did = arguments.get("did", default_did)
is_paid = arguments.get("is_paid", False)
return await self.mcp_client.draw_card(did, is_paid)
elif function_name == "get_card_details":
card_id = arguments.get("card_id")
return await self.mcp_client.get_card_details(card_id)
elif function_name == "analyze_card_collection":
did = arguments.get("did", default_did)
return await self.mcp_client.analyze_card_collection(did)
elif function_name == "get_gacha_stats":
return await self.mcp_client.get_gacha_stats()
else:
return {"error": f"未知のツール: {function_name}"}
except Exception as e:
return {"error": f"ツール実行エラー: {str(e)}"}
def create_ai_provider(provider: str = "ollama", model: Optional[str] = None, mcp_client=None, **kwargs) -> AIProvider:
"""Factory function to create AI providers"""
if provider == "ollama":
model = model or "qwen3"
return OllamaProvider(model=model, **kwargs)
elif provider == "openai":
model = model or "gpt-4o-mini"
return OpenAIProvider(model=model, mcp_client=mcp_client, **kwargs)
else:
raise ValueError(f"Unknown provider: {provider}")

8
api/app/core/config.py
View File

@ -35,7 +35,13 @@ class Settings(BaseSettings):
max_unique_cards: int = 1000 # Maximum number of unique cards
# CORS
cors_origins: list[str] = ["http://localhost:3000", "https://card.syui.ai"]
cors_origins: list[str] = [
"http://localhost:3000",
"http://localhost:5173",
"http://localhost:4173",
"https://card.syui.ai",
"https://xxxcard.syui.ai"
]
# Security
secret_key: str = "your-secret-key-change-this-in-production"

4
api/app/mcp_server.py
View File

@ -37,6 +37,10 @@ class AICardMcpServer:
self.server = FastMCP("aicard")
self._register_mcp_tools()
def get_app(self) -> FastAPI:
"""Get the FastAPI app instance"""
return self.app
def _register_mcp_tools(self):
"""Register all MCP tools"""

45
api/app/repositories/card.py
View File

@ -92,6 +92,51 @@ class CardRepository(BaseRepository[UserCard]):
obtained_at=card.obtained_at
)
self.session.add(registry)
async def get_total_card_count(self) -> int:
"""Get total number of cards obtained"""
result = await self.session.execute(
select(func.count(UserCard.id))
)
return result.scalar() or 0
async def get_cards_by_rarity(self) -> dict:
"""Get card count by rarity"""
result = await self.session.execute(
select(UserCard.status, func.count(UserCard.id))
.group_by(UserCard.status)
)
cards_by_rarity = {}
for status, count in result.all():
cards_by_rarity[status.value if hasattr(status, 'value') else str(status)] = count
return cards_by_rarity
async def get_recent_cards(self, limit: int = 10) -> List[dict]:
"""Get recent card activities"""
result = await self.session.execute(
select(
UserCard.card_id,
UserCard.status,
UserCard.obtained_at,
User.did.label('owner_did')
)
.join(User, UserCard.user_id == User.id)
.order_by(UserCard.obtained_at.desc())
.limit(limit)
)
activities = []
for row in result.all():
activities.append({
'card_id': row.card_id,
'status': row.status.value if hasattr(row.status, 'value') else str(row.status),
'obtained_at': row.obtained_at,
'owner_did': row.owner_did
})
return activities
class UniqueCardRepository(BaseRepository[UniqueCardRegistry]):

59
api/app/routes/cards.py
View File

@ -1,10 +1,11 @@
"""Card-related API routes"""
from typing import List
from typing import List, Dict
from fastapi import APIRouter, HTTPException, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.card import Card, CardDraw, CardDrawResult
from app.services.gacha import GachaService
from app.services.card_master import card_master_service
from app.repositories.user import UserRepository
from app.repositories.card import CardRepository, UniqueCardRepository
from app.db.base import get_session
@ -115,4 +116,58 @@ async def get_unique_cards(db: AsyncSession = Depends(get_session)):
"unique_id": str(uc.unique_id)
}
for uc in unique_cards
]
]
@router.get("/stats")
async def get_gacha_stats(db: AsyncSession = Depends(get_session)):
"""
ガチャ統計情報を取得
"""
try:
card_repo = CardRepository(db)
# 総ガチャ実行数
total_draws = await card_repo.get_total_card_count()
# レアリティ別カード数
cards_by_rarity = await card_repo.get_cards_by_rarity()
# 成功率計算（簡易版）
success_rates = {}
if total_draws > 0:
for rarity, count in cards_by_rarity.items():
success_rates[rarity] = count / total_draws
# 最近の活動（最新10件）
recent_cards = await card_repo.get_recent_cards(limit=10)
recent_activity = []
for card_data in recent_cards:
recent_activity.append({
"timestamp": card_data.get("obtained_at", "").isoformat() if card_data.get("obtained_at") else "",
"user_did": card_data.get("owner_did", "unknown"),
"card_name": f"Card #{card_data.get('card_id', 0)}",
"rarity": card_data.get("status", "common")
})
return {
"total_draws": total_draws,
"cards_by_rarity": cards_by_rarity,
"success_rates": success_rates,
"recent_activity": recent_activity
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Statistics error: {str(e)}")
@router.get("/master", response_model=List[Dict])
async def get_card_master_data():
"""
全カードマスターデータを取得（ai.jsonから）
"""
try:
cards = card_master_service.get_all_cards()
return cards
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to get card master data: {str(e)}")

142
api/app/services/card_master.py Normal file
View File

@ -0,0 +1,142 @@
"""
Card master data fetcher from external ai.json
"""
import httpx
import json
from typing import Dict, List, Optional
from functools import lru_cache
import logging
logger = logging.getLogger(__name__)
CARD_MASTER_URL = "https://git.syui.ai/ai/ai/raw/branch/main/ai.json"
# Default CP ranges for cards (matching existing gacha.py values)
DEFAULT_CP_RANGES = {
0: (10, 100),
1: (20, 120),
2: (30, 130),
3: (40, 140),
4: (50, 150),
5: (25, 125),
6: (15, 115),
7: (60, 160),
8: (80, 180),
9: (70, 170),
10: (90, 190),
11: (35, 135),
12: (65, 165),
13: (75, 175),
14: (100, 200),
15: (85, 185),
135: (95, 195), # world card
}
class CardMasterService:
def __init__(self):
self._cache = None
self._cache_time = 0
self._cache_duration = 3600 # 1 hour cache
@lru_cache(maxsize=1)
def fetch_card_master_data(self) -> Optional[Dict]:
"""Fetch card master data from external source"""
try:
response = httpx.get(CARD_MASTER_URL, timeout=10.0)
response.raise_for_status()
data = response.json()
return data
except Exception as e:
logger.error(f"Failed to fetch card master data: {e}")
return None
def get_card_info(self) -> Dict[int, Dict]:
"""Get card information in the format expected by gacha service"""
master_data = self.fetch_card_master_data()
if not master_data:
# Fallback to hardcoded data
return self._get_fallback_card_info()
try:
cards = master_data.get("ai", {}).get("card", {}).get("cards", [])
card_info = {}
for card in cards:
card_id = card.get("id")
if card_id is not None:
# Use name from JSON, fallback to English name
name = card.get("name", f"card_{card_id}")
# Get CP range from defaults
cp_range = DEFAULT_CP_RANGES.get(card_id, (50, 150))
card_info[card_id] = {
"name": name,
"base_cp_range": cp_range,
"ja_name": card.get("lang", {}).get("ja", {}).get("name", name),
"description": card.get("lang", {}).get("ja", {}).get("text", "")
}
return card_info
except Exception as e:
logger.error(f"Failed to parse card master data: {e}")
return self._get_fallback_card_info()
def _get_fallback_card_info(self) -> Dict[int, Dict]:
"""Fallback card info if external source fails"""
return {
0: {"name": "ai", "base_cp_range": (10, 100)},
1: {"name": "dream", "base_cp_range": (20, 120)},
2: {"name": "radiance", "base_cp_range": (30, 130)},
3: {"name": "neutron", "base_cp_range": (40, 140)},
4: {"name": "sun", "base_cp_range": (50, 150)},
5: {"name": "night", "base_cp_range": (25, 125)},
6: {"name": "snow", "base_cp_range": (15, 115)},
7: {"name": "thunder", "base_cp_range": (60, 160)},
8: {"name": "ultimate", "base_cp_range": (80, 180)},
9: {"name": "sword", "base_cp_range": (70, 170)},
10: {"name": "destruction", "base_cp_range": (90, 190)},
11: {"name": "earth", "base_cp_range": (35, 135)},
12: {"name": "galaxy", "base_cp_range": (65, 165)},
13: {"name": "create", "base_cp_range": (75, 175)},
14: {"name": "supernova", "base_cp_range": (100, 200)},
15: {"name": "world", "base_cp_range": (85, 185)},
}
def get_all_cards(self) -> List[Dict]:
"""Get all cards with full information"""
master_data = self.fetch_card_master_data()
if not master_data:
return []
try:
cards = master_data.get("ai", {}).get("card", {}).get("cards", [])
result = []
for card in cards:
card_id = card.get("id")
if card_id is not None:
cp_range = DEFAULT_CP_RANGES.get(card_id, (50, 150))
result.append({
"id": card_id,
"name": card.get("name", f"card_{card_id}"),
"ja_name": card.get("lang", {}).get("ja", {}).get("name", ""),
"description": card.get("lang", {}).get("ja", {}).get("text", ""),
"base_cp_min": cp_range[0],
"base_cp_max": cp_range[1]
})
return result
except Exception as e:
logger.error(f"Failed to get all cards: {e}")
return []
# Singleton instance
card_master_service = CardMasterService()

23
api/app/services/gacha.py
View File

@ -10,36 +10,19 @@ from app.models.card import Card, CardRarity
from app.repositories.user import UserRepository
from app.repositories.card import CardRepository, UniqueCardRepository
from app.db.models import DrawHistory
from app.services.card_master import card_master_service
class GachaService:
"""ガチャシステムのサービスクラス"""
# カード基本情報（ai.jsonから）
CARD_INFO = {
0: {"name": "ai", "base_cp_range": (10, 100)},
1: {"name": "dream", "base_cp_range": (20, 120)},
2: {"name": "radiance", "base_cp_range": (30, 130)},
3: {"name": "neutron", "base_cp_range": (40, 140)},
4: {"name": "sun", "base_cp_range": (50, 150)},
5: {"name": "night", "base_cp_range": (25, 125)},
6: {"name": "snow", "base_cp_range": (15, 115)},
7: {"name": "thunder", "base_cp_range": (60, 160)},
8: {"name": "ultimate", "base_cp_range": (80, 180)},
9: {"name": "sword", "base_cp_range": (70, 170)},
10: {"name": "destruction", "base_cp_range": (90, 190)},
11: {"name": "earth", "base_cp_range": (35, 135)},
12: {"name": "galaxy", "base_cp_range": (65, 165)},
13: {"name": "create", "base_cp_range": (75, 175)},
14: {"name": "supernova", "base_cp_range": (100, 200)},
15: {"name": "world", "base_cp_range": (85, 185)},
}
def __init__(self, session: AsyncSession):
self.session = session
self.user_repo = UserRepository(session)
self.card_repo = CardRepository(session)
self.unique_repo = UniqueCardRepository(session)
# Load card info from external source
self.CARD_INFO = card_master_service.get_card_info()
async def draw_card(self, user_did: str, is_paid: bool = False) -> Tuple[Card, bool]:
"""