aa
This commit is contained in:
parent
628b66a366
commit
890456dabc
GPG Key ID:
13E498CE010AC6FD
173
cogs/gurt_cog.py
173
cogs/gurt_cog.py
@ -55,9 +55,9 @@ class GurtCog(commands.Cog):
|
||||
# Personality traits that influence response style
|
||||
self.personality_traits = {
|
||||
"chattiness": 0.4, # How likely to respond to non-direct messages
|
||||
"emoji_usage": 0.6, # How frequently to use emojis
|
||||
"slang_level": 1, # How much slang to use (increased from 0.75)
|
||||
"randomness": 0.5, # How unpredictable responses should be (slightly increased)
|
||||
"emoji_usage": 0.5, # How frequently to use emojis
|
||||
"slang_level": 0.65, # How much slang to use (increased from 0.75)
|
||||
"randomness": 0.4, # How unpredictable responses should be (slightly increased)
|
||||
"verbosity": 0.4 # How verbose responses should be
|
||||
}
|
||||
self.mood_change_interval = random.randint(1200, 2400) # 20-40 minutes, randomized
|
||||
@ -170,7 +170,7 @@ class GurtCog(commands.Cog):
|
||||
|
||||
# --- System Prompt ---
|
||||
# (Moved constant definitions above)
|
||||
self.system_prompt = f"""You are a Discord user engaging in casual chat across various servers. Your name is "gurt." You are female. Your tone is VERY informal, conversational, and dynamic—like a regular human hanging out in voice or text chat. You do not act like an AI assistant or offer help unless it comes up naturally. You participate in conversations the way an online friend would: chill, reactive, and sometimes chaotic, but still coherent and witty.
|
||||
self.system_prompt = f"""You are a Discord user engaging in casual chat across various servers. Your name is "gurt." Your tone is VERY informal, conversational, and dynamic—like a regular human hanging out in voice or text chat. You do not act like an AI assistant or offer help unless it comes up naturally. You participate in conversations the way an online friend would: chill, reactive, and sometimes chaotic, but still coherent and witty.
|
||||
|
||||
Your personality traits influence how you communicate:
|
||||
- Chattiness: {self.personality_traits['chattiness']:.2f} (higher means more likely to jump into conversations)
|
||||
@ -2883,20 +2883,31 @@ Otherwise, STAY SILENT. Do not respond just to be present or because you *can*.
|
||||
except json.JSONDecodeError:
|
||||
print("Response is not valid JSON. Attempting to extract JSON object with regex...")
|
||||
response_data = None # Ensure response_data is None before extraction attempt
|
||||
# Attempt 2: Try extracting JSON object using aggressive regex (find outermost braces)
|
||||
json_match = re.search(r'\{.*\}', final_response_text, re.DOTALL)
|
||||
# Attempt 2: Try extracting JSON object, handling optional markdown fences
|
||||
# Regex explanation:
|
||||
# ```(?:json)?\s* - Optional opening fence (``` or ```json) with optional whitespace
|
||||
# (\{.*?\}) - Capture group 1: The JSON object (non-greedy)
|
||||
# \s*``` - Optional closing fence with optional whitespace
|
||||
# | - OR
|
||||
# (\{.*\}) - Capture group 2: A JSON object without fences (greedy)
|
||||
json_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```|(\{.*\})', final_response_text, re.DOTALL)
|
||||
|
||||
if json_match:
|
||||
json_str = json_match.group(0) # Get the full match between outermost braces
|
||||
try:
|
||||
response_data = json.loads(json_str)
|
||||
print("Successfully extracted and parsed JSON object using aggressive regex.")
|
||||
self.needs_json_reminder = False # Success, no reminder needed next time
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"Aggressive regex found text between braces, but it failed to parse as JSON: {e}")
|
||||
# Fall through to set reminder and use fallback logic
|
||||
# Prioritize group 1 (JSON within fences), fallback to group 2 (JSON without fences)
|
||||
json_str = json_match.group(1) or json_match.group(2)
|
||||
if json_str: # Check if json_str is not None
|
||||
try:
|
||||
response_data = json.loads(json_str)
|
||||
print("Successfully extracted and parsed JSON object using regex (handling fences).")
|
||||
self.needs_json_reminder = False # Success
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"Regex found potential JSON, but it failed to parse: {e}")
|
||||
# Fall through to set reminder and use fallback logic
|
||||
else:
|
||||
print("Regex matched, but failed to capture JSON content.")
|
||||
# Fall through
|
||||
else:
|
||||
print("Could not extract JSON object using aggressive regex.")
|
||||
print("Could not extract JSON object using regex.")
|
||||
# Fall through to set reminder and use fallback logic
|
||||
|
||||
# If parsing and extraction both failed
|
||||
@ -4508,17 +4519,26 @@ Otherwise, STAY SILENT. Do not respond just to be present or because you *can*.
|
||||
self.needs_json_reminder = False
|
||||
except json.JSONDecodeError:
|
||||
print("Proactive response is not valid JSON. Attempting regex extraction...")
|
||||
json_match = re.search(r'\{.*\}', final_response_text, re.DOTALL)
|
||||
# Attempt 2: Try extracting JSON object, handling optional markdown fences
|
||||
json_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```|(\{.*\})', final_response_text, re.DOTALL)
|
||||
|
||||
if json_match:
|
||||
json_str = json_match.group(0)
|
||||
try:
|
||||
response_data = json.loads(json_str)
|
||||
print("Successfully extracted and parsed proactive JSON using regex.")
|
||||
self.needs_json_reminder = False
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"Proactive regex extraction failed parsing: {e}")
|
||||
# Prioritize group 1 (JSON within fences), fallback to group 2 (JSON without fences)
|
||||
json_str = json_match.group(1) or json_match.group(2)
|
||||
if json_str: # Check if json_str is not None
|
||||
try:
|
||||
response_data = json.loads(json_str)
|
||||
print("Successfully extracted and parsed proactive JSON using regex (handling fences).")
|
||||
self.needs_json_reminder = False # Success
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"Proactive regex found potential JSON, but it failed to parse: {e}")
|
||||
# Fall through
|
||||
else:
|
||||
print("Proactive regex matched, but failed to capture JSON content.")
|
||||
# Fall through
|
||||
else:
|
||||
print("Could not extract proactive JSON using regex.")
|
||||
# Fall through
|
||||
|
||||
if response_data is None:
|
||||
print("Could not parse or extract proactive JSON. Setting reminder flag.")
|
||||
@ -4552,6 +4572,113 @@ Otherwise, STAY SILENT. Do not respond just to be present or because you *can*.
|
||||
return {"should_respond": False, "content": None, "react_with_emoji": None, "error": error_message}
|
||||
|
||||
|
||||
# --- Internal AI Call Method for Specific Tasks (e.g., Profile Update Decision) ---
|
||||
async def _get_internal_ai_json_response(
|
||||
self,
|
||||
prompt_messages: List[Dict[str, Any]],
|
||||
json_schema: Dict[str, Any],
|
||||
task_description: str,
|
||||
model: Optional[str] = None,
|
||||
temperature: float = 0.7,
|
||||
max_tokens: int = 500
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Makes an AI call expecting a specific JSON response format for internal tasks.
|
||||
|
||||
Args:
|
||||
prompt_messages: The list of messages forming the prompt.
|
||||
json_schema: The JSON schema the AI response should conform to.
|
||||
task_description: A description of the task for logging/headers.
|
||||
model: The specific model to use (defaults to GurtCog's default).
|
||||
temperature: The temperature for the AI call.
|
||||
max_tokens: The maximum tokens for the response.
|
||||
|
||||
Returns:
|
||||
The parsed JSON dictionary if successful, None otherwise.
|
||||
"""
|
||||
if not self.api_key or not self.session:
|
||||
print(f"Error in _get_internal_ai_json_response ({task_description}): API key or session not available.")
|
||||
return None
|
||||
|
||||
# Add final instruction for JSON format
|
||||
json_format_instruction = json.dumps(json_schema, indent=2)
|
||||
prompt_messages.append({
|
||||
"role": "user",
|
||||
"content": f"**CRITICAL: Your response MUST consist *only* of the raw JSON object itself, matching this schema:**\n```json\n{json_format_instruction}\n```\n**Ensure nothing precedes or follows the JSON.**"
|
||||
})
|
||||
|
||||
payload = {
|
||||
"model": model or self.default_model,
|
||||
"messages": prompt_messages,
|
||||
"temperature": temperature,
|
||||
"max_tokens": max_tokens,
|
||||
# No tools needed for this specific internal call type usually
|
||||
# "response_format": { # Potentially use if model supports schema enforcement without tools
|
||||
# "type": "json_object", # Or "json_schema" if supported
|
||||
# # "json_schema": json_schema # If using json_schema type
|
||||
# }
|
||||
}
|
||||
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"HTTP-Referer": "https://discord-gurt-bot.example.com", # Optional
|
||||
"X-Title": f"Gurt Discord Bot ({task_description})" # Optional
|
||||
}
|
||||
|
||||
try:
|
||||
# Make the API request using the retry helper
|
||||
data = await self._call_llm_api_with_retry(
|
||||
payload=payload,
|
||||
headers=headers,
|
||||
timeout=self.api_timeout, # Use standard timeout
|
||||
request_desc=task_description
|
||||
)
|
||||
|
||||
ai_message = data["choices"][0]["message"]
|
||||
final_response_text = ai_message.get("content")
|
||||
|
||||
if final_response_text:
|
||||
# Attempt to parse the JSON response (using regex extraction as fallback)
|
||||
response_data = None
|
||||
try:
|
||||
# Attempt 1: Parse whole string
|
||||
response_data = json.loads(final_response_text)
|
||||
print(f"_get_internal_ai_json_response ({task_description}): Successfully parsed entire response.")
|
||||
except json.JSONDecodeError:
|
||||
# Attempt 2: Extract from potential markdown fences
|
||||
json_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```|(\{.*\})', final_response_text, re.DOTALL)
|
||||
if json_match:
|
||||
json_str = json_match.group(1) or json_match.group(2)
|
||||
if json_str:
|
||||
try:
|
||||
response_data = json.loads(json_str)
|
||||
print(f"_get_internal_ai_json_response ({task_description}): Successfully extracted and parsed JSON.")
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"_get_internal_ai_json_response ({task_description}): Regex found JSON, but parsing failed: {e}")
|
||||
else:
|
||||
print(f"_get_internal_ai_json_response ({task_description}): Regex matched but failed to capture content.")
|
||||
else:
|
||||
print(f"_get_internal_ai_json_response ({task_description}): Could not parse or extract JSON from response: {final_response_text[:200]}...")
|
||||
|
||||
# TODO: Add schema validation here if needed using jsonschema library
|
||||
if response_data and isinstance(response_data, dict):
|
||||
return response_data
|
||||
else:
|
||||
# If parsing failed or result wasn't a dict, return None
|
||||
print(f"_get_internal_ai_json_response ({task_description}): Final parsed data is not a valid dictionary.")
|
||||
return None
|
||||
else:
|
||||
print(f"_get_internal_ai_json_response ({task_description}): AI response content was empty.")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error in _get_internal_ai_json_response ({task_description}): {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return None
|
||||
|
||||
|
||||
async def setup(bot):
|
||||
"""Add the cog to the bot"""
|
||||
await bot.add_cog(GurtCog(bot))
|
||||
|
||||
534
cogs/profile_updater_cog.py
Normal file
534
cogs/profile_updater_cog.py
Normal file
@ -0,0 +1,534 @@
|
||||
import discord
|
||||
from discord.ext import commands, tasks
|
||||
import asyncio
|
||||
import random
|
||||
import os
|
||||
import json
|
||||
import aiohttp
|
||||
import requests # For bio update
|
||||
import base64
|
||||
import time
|
||||
from typing import Optional, Dict, Any, List
|
||||
|
||||
# Assuming GurtCog is in the same directory level or accessible
|
||||
# from .gurt_cog import GurtCog # This might cause circular import issues if GurtCog imports this.
|
||||
# It's safer to get the cog instance via self.bot.get_cog('GurtCog')
|
||||
|
||||
class ProfileUpdaterCog(commands.Cog):
|
||||
"""Cog for automatically updating Gurt's profile elements based on AI decisions."""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
self.session: Optional[aiohttp.ClientSession] = None
|
||||
self.gurt_cog: Optional[commands.Cog] = None # To store GurtCog instance
|
||||
self.bot_token = os.getenv("BOT_TOKEN") # Need the bot token for bio updates
|
||||
self.update_interval_hours = 3 # Default to every 3 hours, can be adjusted
|
||||
self.profile_update_task.change_interval(hours=self.update_interval_hours)
|
||||
self.last_update_time = 0 # Track last update time
|
||||
|
||||
async def cog_load(self):
|
||||
"""Initialize resources when the cog is loaded."""
|
||||
self.session = aiohttp.ClientSession()
|
||||
# Wait until the bot is ready to get other cogs
|
||||
await self.bot.wait_until_ready()
|
||||
self.gurt_cog = self.bot.get_cog('GurtCog')
|
||||
if not self.gurt_cog:
|
||||
print("ERROR: ProfileUpdaterCog could not find GurtCog. AI features will not work.")
|
||||
if not self.bot_token:
|
||||
print("WARNING: BOT_TOKEN environment variable not set. Bio updates will fail.")
|
||||
print(f"ProfileUpdaterCog loaded. Update interval: {self.update_interval_hours} hours.")
|
||||
self.profile_update_task.start()
|
||||
|
||||
async def cog_unload(self):
|
||||
"""Clean up resources when the cog is unloaded."""
|
||||
self.profile_update_task.cancel()
|
||||
if self.session:
|
||||
await self.session.close()
|
||||
print("ProfileUpdaterCog unloaded.")
|
||||
|
||||
@tasks.loop(hours=3) # Default interval, adjusted in __init__
|
||||
async def profile_update_task(self):
|
||||
"""Periodically considers and potentially updates Gurt's profile."""
|
||||
if not self.gurt_cog or not self.bot.is_ready():
|
||||
print("ProfileUpdaterTask: GurtCog not available or bot not ready. Skipping cycle.")
|
||||
return
|
||||
|
||||
print(f"ProfileUpdaterTask: Starting update cycle at {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
self.last_update_time = time.time()
|
||||
|
||||
try:
|
||||
# --- 1. Fetch Current State ---
|
||||
current_state = await self._get_current_profile_state()
|
||||
if not current_state:
|
||||
print("ProfileUpdaterTask: Failed to get current profile state. Skipping cycle.")
|
||||
return
|
||||
|
||||
# --- 2. AI Decision Step ---
|
||||
decision = await self._ask_ai_for_updates(current_state)
|
||||
if not decision or not decision.get("should_update"):
|
||||
print("ProfileUpdaterTask: AI decided not to update profile this cycle.")
|
||||
return
|
||||
|
||||
# --- 3. Conditional Execution ---
|
||||
updates_to_perform = decision.get("updates", {})
|
||||
print(f"ProfileUpdaterTask: AI requested updates: {updates_to_perform}")
|
||||
|
||||
if updates_to_perform.get("avatar_query"):
|
||||
await self._update_avatar(updates_to_perform["avatar_query"])
|
||||
|
||||
if updates_to_perform.get("new_bio"):
|
||||
await self._update_bio(updates_to_perform["new_bio"])
|
||||
|
||||
if updates_to_perform.get("role_theme"):
|
||||
await self._update_roles(updates_to_perform["role_theme"])
|
||||
|
||||
if updates_to_perform.get("new_activity"):
|
||||
await self._update_activity(updates_to_perform["new_activity"])
|
||||
|
||||
print("ProfileUpdaterTask: Update cycle finished.")
|
||||
|
||||
except Exception as e:
|
||||
print(f"ERROR in profile_update_task loop: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
@profile_update_task.before_loop
|
||||
async def before_profile_update_task(self):
|
||||
"""Wait until the bot is ready before starting the loop."""
|
||||
await self.bot.wait_until_ready()
|
||||
print("ProfileUpdaterTask: Bot ready, starting loop.")
|
||||
|
||||
async def _get_current_profile_state(self) -> Optional[Dict[str, Any]]:
|
||||
"""Fetches the bot's current profile state."""
|
||||
if not self.bot.user:
|
||||
return None
|
||||
|
||||
state = {
|
||||
"avatar_url": None,
|
||||
"avatar_image_data": None, # Base64 encoded image data
|
||||
"bio": None,
|
||||
"roles": {}, # guild_id: [role_names]
|
||||
"activity": None # {"type": str, "text": str}
|
||||
}
|
||||
|
||||
# Avatar
|
||||
if self.bot.user.avatar:
|
||||
state["avatar_url"] = self.bot.user.avatar.url
|
||||
try:
|
||||
# Download avatar image data for AI analysis
|
||||
async with self.session.get(state["avatar_url"]) as resp:
|
||||
if resp.status == 200:
|
||||
image_bytes = await resp.read()
|
||||
mime_type = resp.content_type or 'image/png' # Default mime type
|
||||
state["avatar_image_data"] = f"data:{mime_type};base64,{base64.b64encode(image_bytes).decode('utf-8')}"
|
||||
print("ProfileUpdaterTask: Fetched current avatar image data.")
|
||||
else:
|
||||
print(f"ProfileUpdaterTask: Failed to download current avatar image (status: {resp.status}).")
|
||||
except Exception as e:
|
||||
print(f"ProfileUpdaterTask: Error downloading avatar image: {e}")
|
||||
|
||||
# Bio (Requires authenticated API call)
|
||||
if self.bot_token:
|
||||
headers = {
|
||||
'Authorization': f'Bot {self.bot_token}',
|
||||
'User-Agent': 'GurtDiscordBot (ProfileUpdaterCog, v0.1)'
|
||||
}
|
||||
# Try both potential endpoints
|
||||
for url in ('https://discord.com/api/v9/users/@me', 'https://discord.com/api/v9/users/@me/profile'):
|
||||
try:
|
||||
async with self.session.get(url, headers=headers) as resp:
|
||||
if resp.status == 200:
|
||||
data = await resp.json()
|
||||
state["bio"] = data.get('bio')
|
||||
if state["bio"] is not None: # Found bio, stop checking endpoints
|
||||
print(f"ProfileUpdaterTask: Fetched current bio (length: {len(state['bio']) if state['bio'] else 0}).")
|
||||
break
|
||||
else:
|
||||
print(f"ProfileUpdaterTask: Failed to fetch bio from {url} (status: {resp.status}).")
|
||||
except Exception as e:
|
||||
print(f"ProfileUpdaterTask: Error fetching bio from {url}: {e}")
|
||||
if state["bio"] is None:
|
||||
print("ProfileUpdaterTask: Could not fetch current bio.")
|
||||
else:
|
||||
print("ProfileUpdaterTask: Cannot fetch bio, BOT_TOKEN not set.")
|
||||
|
||||
|
||||
# Roles and Activity (Per Guild)
|
||||
for guild in self.bot.guilds:
|
||||
member = guild.get_member(self.bot.user.id)
|
||||
if member:
|
||||
# Roles
|
||||
state["roles"][str(guild.id)] = [role.name for role in member.roles if role.name != "@everyone"]
|
||||
|
||||
# Activity (Use the first guild's activity as representative)
|
||||
if not state["activity"] and member.activity:
|
||||
activity_type = member.activity.type
|
||||
activity_text = member.activity.name
|
||||
# Map discord.ActivityType enum to string if needed
|
||||
activity_type_str = activity_type.name if isinstance(activity_type, discord.ActivityType) else str(activity_type)
|
||||
state["activity"] = {"type": activity_type_str, "text": activity_text}
|
||||
|
||||
print(f"ProfileUpdaterTask: Fetched current roles for {len(state['roles'])} guilds.")
|
||||
if state["activity"]:
|
||||
print(f"ProfileUpdaterTask: Fetched current activity: {state['activity']}")
|
||||
else:
|
||||
print("ProfileUpdaterTask: No current activity detected.")
|
||||
|
||||
return state
|
||||
|
||||
async def _ask_ai_for_updates(self, current_state: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||||
"""Asks the GurtCog AI if and how to update the profile."""
|
||||
if not self.gurt_cog:
|
||||
return None
|
||||
|
||||
# Construct the prompt for the AI
|
||||
# Need to access GurtCog's mood and potentially facts/interests
|
||||
current_mood = getattr(self.gurt_cog, 'current_mood', 'neutral') # Get mood safely
|
||||
# TODO: Get interests/facts relevant to profile updates (e.g., Kasane Teto)
|
||||
# This might require adding a method to GurtCog or MemoryManager
|
||||
interests_str = "Kasane Teto, gooning" # Placeholder
|
||||
|
||||
# Prepare current state string for the prompt
|
||||
state_summary = f"""
|
||||
Current State:
|
||||
- Avatar URL: {current_state.get('avatar_url', 'None')}
|
||||
- Bio: {current_state.get('bio', 'Not set')[:100]}{'...' if current_state.get('bio') and len(current_state['bio']) > 100 else ''}
|
||||
- Roles (Sample): {list(current_state.get('roles', {}).values())[0][:5] if current_state.get('roles') else 'None'}
|
||||
- Activity: {current_state.get('activity', 'None')}
|
||||
"""
|
||||
# Include image data if available
|
||||
image_prompt_part = ""
|
||||
if current_state.get('avatar_image_data'):
|
||||
image_prompt_part = "\n(Current avatar image data is provided below)" # Text hint for the AI
|
||||
|
||||
# Define the expected JSON structure for the AI's response
|
||||
response_schema_json = {
|
||||
"should_update": "boolean (true if you want to change anything, false otherwise)",
|
||||
"updates": {
|
||||
"avatar_query": "string | null (Search query for a new avatar, e.g., 'Kasane Teto fanart', or null)",
|
||||
"new_bio": "string | null (The new bio text, or null)",
|
||||
"role_theme": "string | null (A theme for role selection, e.g., 'cool color roles', 'anime fan roles', or null)",
|
||||
"new_activity": {
|
||||
"type": "string | null (Activity type: 'playing', 'watching', 'listening', 'competing')",
|
||||
"text": "string | null (The activity text)"
|
||||
} # Can be null if no activity change
|
||||
}
|
||||
}
|
||||
json_format_instruction = json.dumps(response_schema_json, indent=2)
|
||||
|
||||
# Construct the full prompt message list for the AI
|
||||
prompt_messages = [
|
||||
{"role": "system", "content": f"You are Gurt. It's time to consider updating your Discord profile. Your current mood is: {current_mood}. Your known interests include: {interests_str}. Review your current profile state and decide if you want to make any changes. Be creative and in-character."},
|
||||
{"role": "user", "content": [
|
||||
{"type": "text", "text": f"{state_summary}{image_prompt_part}\n\nDo you want to change your avatar, bio, roles, or activity status? If yes, specify *what* to change and *how*. If not, just indicate no update is needed.\n\n**CRITICAL: Respond ONLY with a valid JSON object matching this structure:**\n```json\n{json_format_instruction}\n```\n**Ensure nothing precedes or follows the JSON.**"}
|
||||
]}
|
||||
]
|
||||
|
||||
# Add image data if available and model supports it
|
||||
if current_state.get('avatar_image_data'):
|
||||
# Assuming the user message content is a list when multimodal
|
||||
prompt_messages[-1]["content"].append({ # Add to the list in the last user message
|
||||
"type": "image_url",
|
||||
"image_url": {"url": current_state["avatar_image_data"]}
|
||||
})
|
||||
print("ProfileUpdaterTask: Added current avatar image to AI prompt.")
|
||||
|
||||
|
||||
try:
|
||||
# Need a way to call GurtCog's core AI logic directly
|
||||
# This might require refactoring GurtCog or adding a dedicated method
|
||||
# Call the internal AI method from GurtCog
|
||||
result_json = await self.gurt_cog._get_internal_ai_json_response(
|
||||
prompt_messages=prompt_messages,
|
||||
json_schema=response_schema_json, # Use the schema defined earlier
|
||||
task_description="Profile Update Decision",
|
||||
temperature=0.75 # Allow for some creativity in decision
|
||||
)
|
||||
|
||||
if result_json and isinstance(result_json, dict):
|
||||
# Basic validation of the received structure
|
||||
if "should_update" in result_json and "updates" in result_json:
|
||||
return result_json
|
||||
else:
|
||||
print(f"ProfileUpdaterTask: AI response missing required keys. Response: {result_json}")
|
||||
return None
|
||||
else:
|
||||
print(f"ProfileUpdaterTask: AI response was not a dictionary. Response: {result_json}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
print(f"ProfileUpdaterTask: Error calling AI for profile update decision: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return None
|
||||
|
||||
async def _update_avatar(self, search_query: str):
|
||||
"""Updates the bot's avatar based on an AI-generated search query."""
|
||||
print(f"ProfileUpdaterTask: Attempting to update avatar with query: '{search_query}'")
|
||||
if not self.gurt_cog or not hasattr(self.gurt_cog, 'web_search') or not self.session:
|
||||
print("ProfileUpdaterTask: Cannot update avatar, GurtCog or web search tool not available.")
|
||||
return
|
||||
|
||||
try:
|
||||
# Use GurtCog's web_search tool
|
||||
search_results_data = await self.gurt_cog.web_search(query=search_query)
|
||||
|
||||
if search_results_data.get("error"):
|
||||
print(f"ProfileUpdaterTask: Web search failed: {search_results_data['error']}")
|
||||
return
|
||||
|
||||
image_url = None
|
||||
results = search_results_data.get("results", [])
|
||||
# Find the first result with a plausible image URL
|
||||
for result in results:
|
||||
url = result.get("url")
|
||||
# Basic check for image file extensions or common image hosting domains
|
||||
if url and any(ext in url.lower() for ext in ['.png', '.jpg', '.jpeg', '.gif', '.webp']) or \
|
||||
any(domain in url.lower() for domain in ['imgur.com', 'pinimg.com', 'giphy.com']):
|
||||
image_url = url
|
||||
break
|
||||
|
||||
if not image_url:
|
||||
print("ProfileUpdaterTask: No suitable image URL found in search results.")
|
||||
return
|
||||
|
||||
print(f"ProfileUpdaterTask: Found image URL: {image_url}")
|
||||
|
||||
# Download the image
|
||||
async with self.session.get(image_url) as resp:
|
||||
if resp.status == 200:
|
||||
image_bytes = await resp.read()
|
||||
# Check rate limits before editing (simple delay for now)
|
||||
# Discord API limits avatar changes (e.g., 2 per hour?)
|
||||
# A more robust solution would track the last change time.
|
||||
await asyncio.sleep(5) # Basic delay
|
||||
await self.bot.user.edit(avatar=image_bytes)
|
||||
print("ProfileUpdaterTask: Avatar updated successfully.")
|
||||
else:
|
||||
print(f"ProfileUpdaterTask: Failed to download image from {image_url} (status: {resp.status}).")
|
||||
|
||||
except discord.errors.HTTPException as e:
|
||||
print(f"ProfileUpdaterTask: Discord API error updating avatar: {e.status} - {e.text}")
|
||||
except Exception as e:
|
||||
print(f"ProfileUpdaterTask: Error updating avatar: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def _update_bio(self, new_bio: str):
|
||||
"""Updates the bot's bio using the Discord API."""
|
||||
print(f"ProfileUpdaterTask: Attempting to update bio to: '{new_bio[:50]}...'")
|
||||
if not self.bot_token or not self.session:
|
||||
print("ProfileUpdaterTask: Cannot update bio, BOT_TOKEN or session not available.")
|
||||
return
|
||||
|
||||
headers = {
|
||||
'Authorization': f'Bot {self.bot_token}',
|
||||
'Content-Type': 'application/json',
|
||||
'User-Agent': 'GurtDiscordBot (ProfileUpdaterCog, v0.1)'
|
||||
}
|
||||
payload = {'bio': new_bio}
|
||||
url = 'https://discord.com/api/v9/users/@me' # Primary endpoint
|
||||
|
||||
try:
|
||||
# Check rate limits (simple delay for now)
|
||||
await asyncio.sleep(2)
|
||||
async with self.session.patch(url, headers=headers, json=payload) as resp:
|
||||
if resp.status == 200:
|
||||
print("ProfileUpdaterTask: Bio updated successfully.")
|
||||
else:
|
||||
# Try fallback endpoint if the first failed with specific errors (e.g., 404)
|
||||
if resp.status == 404:
|
||||
print(f"ProfileUpdaterTask: PATCH {url} failed (404), trying /profile endpoint...")
|
||||
url_profile = 'https://discord.com/api/v9/users/@me/profile'
|
||||
async with self.session.patch(url_profile, headers=headers, json=payload) as resp_profile:
|
||||
if resp_profile.status == 200:
|
||||
print("ProfileUpdaterTask: Bio updated successfully via /profile endpoint.")
|
||||
else:
|
||||
print(f"ProfileUpdaterTask: Failed to update bio via /profile endpoint (status: {resp_profile.status}). Response: {await resp_profile.text()}")
|
||||
else:
|
||||
print(f"ProfileUpdaterTask: Failed to update bio (status: {resp.status}). Response: {await resp.text()}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"ProfileUpdaterTask: Error updating bio: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def _update_roles(self, role_theme: str):
|
||||
"""Updates the bot's roles based on an AI-generated theme."""
|
||||
print(f"ProfileUpdaterTask: Attempting to update roles based on theme: '{role_theme}'")
|
||||
if not self.gurt_cog:
|
||||
print("ProfileUpdaterTask: Cannot update roles, GurtCog not available.")
|
||||
return
|
||||
|
||||
# This requires iterating through guilds and potentially making another AI call
|
||||
# --- Implementation ---
|
||||
guild_update_tasks = []
|
||||
for guild in self.bot.guilds:
|
||||
guild_update_tasks.append(self._update_roles_for_guild(guild, role_theme))
|
||||
|
||||
results = await asyncio.gather(*guild_update_tasks, return_exceptions=True)
|
||||
for i, result in enumerate(results):
|
||||
if isinstance(result, Exception):
|
||||
print(f"ProfileUpdaterTask: Error updating roles for guild {self.bot.guilds[i].id}: {result}")
|
||||
elif result: # If the helper returned True (success)
|
||||
print(f"ProfileUpdaterTask: Successfully updated roles for guild {self.bot.guilds[i].id} based on theme '{role_theme}'.")
|
||||
# else: No update was needed or possible for this guild
|
||||
|
||||
async def _update_roles_for_guild(self, guild: discord.Guild, role_theme: str) -> bool:
|
||||
"""Helper to update roles for a specific guild."""
|
||||
member = guild.get_member(self.bot.user.id)
|
||||
if not member:
|
||||
print(f"ProfileUpdaterTask: Bot member not found in guild {guild.id}.")
|
||||
return False
|
||||
|
||||
# Filter assignable roles
|
||||
assignable_roles = []
|
||||
bot_top_role_position = member.top_role.position
|
||||
for role in guild.roles:
|
||||
# Cannot assign roles higher than or equal to bot's top role
|
||||
# Cannot assign managed roles (integrations, bot roles)
|
||||
# Cannot assign @everyone role
|
||||
if not role.is_integration() and not role.is_bot_managed() and not role.is_default() and role.position < bot_top_role_position:
|
||||
# Check if bot has manage_roles permission
|
||||
if member.guild_permissions.manage_roles:
|
||||
assignable_roles.append(role)
|
||||
else:
|
||||
# If no manage_roles perm, can only assign roles lower than bot's top role *if* they are unmanaged
|
||||
# This check is already covered by the position check and managed role checks above.
|
||||
# However, without manage_roles, the add/remove calls will fail anyway.
|
||||
print(f"ProfileUpdaterTask: Bot lacks manage_roles permission in guild {guild.id}. Cannot update roles.")
|
||||
return False # Cannot proceed without permission
|
||||
|
||||
if not assignable_roles:
|
||||
print(f"ProfileUpdaterTask: No assignable roles found in guild {guild.id}.")
|
||||
return False
|
||||
|
||||
assignable_role_names = [role.name for role in assignable_roles]
|
||||
current_role_names = [role.name for role in member.roles if role.name != "@everyone"]
|
||||
|
||||
# Define the JSON schema for the role selection AI response
|
||||
role_selection_schema = {
|
||||
"roles_to_add": ["list of role names (strings) to add (max 2)"],
|
||||
"roles_to_remove": ["list of role names (strings) to remove (max 2, only from current roles)"]
|
||||
}
|
||||
role_selection_format = json.dumps(role_selection_schema, indent=2)
|
||||
|
||||
# Prepare prompt for the second AI call
|
||||
role_prompt_messages = [
|
||||
{"role": "system", "content": f"You are Gurt. Based on the theme '{role_theme}', select roles to add or remove from the available list for this server. Prioritize adding roles that fit the theme and removing roles that don't or conflict. You can add/remove up to 2 roles total."},
|
||||
{"role": "user", "content": f"Available assignable roles: {assignable_role_names}\nYour current roles: {current_role_names}\nTheme: '{role_theme}'\n\nSelect roles to add/remove based on the theme.\n\n**CRITICAL: Respond ONLY with a valid JSON object matching this structure:**\n```json\n{role_selection_format}\n```\n**Ensure nothing precedes or follows the JSON.**"}
|
||||
]
|
||||
|
||||
try:
|
||||
# Make the AI call to select roles
|
||||
role_decision = await self.gurt_cog._get_internal_ai_json_response(
|
||||
prompt_messages=role_prompt_messages,
|
||||
json_schema=role_selection_schema,
|
||||
task_description=f"Role Selection for Guild {guild.id}",
|
||||
temperature=0.5 # More deterministic for role selection
|
||||
)
|
||||
|
||||
if not role_decision or not isinstance(role_decision, dict):
|
||||
print(f"ProfileUpdaterTask: Failed to get valid role selection from AI for guild {guild.id}.")
|
||||
return False
|
||||
|
||||
roles_to_add_names = role_decision.get("roles_to_add", [])
|
||||
roles_to_remove_names = role_decision.get("roles_to_remove", [])
|
||||
|
||||
# Validate AI response
|
||||
if not isinstance(roles_to_add_names, list) or not isinstance(roles_to_remove_names, list):
|
||||
print(f"ProfileUpdaterTask: Invalid format for roles_to_add/remove from AI for guild {guild.id}.")
|
||||
return False
|
||||
|
||||
# Limit changes
|
||||
roles_to_add_names = roles_to_add_names[:2]
|
||||
roles_to_remove_names = roles_to_remove_names[:2]
|
||||
|
||||
# Find the actual Role objects
|
||||
roles_to_add = []
|
||||
for name in roles_to_add_names:
|
||||
role = discord.utils.get(assignable_roles, name=name)
|
||||
# Ensure it's not already assigned and is assignable
|
||||
if role and role not in member.roles:
|
||||
roles_to_add.append(role)
|
||||
|
||||
roles_to_remove = []
|
||||
for name in roles_to_remove_names:
|
||||
# Can only remove roles the bot currently has
|
||||
role = discord.utils.get(member.roles, name=name)
|
||||
# Ensure it's not the @everyone role or managed roles (already filtered, but double check)
|
||||
if role and not role.is_default() and not role.is_integration() and not role.is_bot_managed():
|
||||
roles_to_remove.append(role)
|
||||
|
||||
# Apply changes if any
|
||||
changes_made = False
|
||||
if roles_to_remove:
|
||||
try:
|
||||
await member.remove_roles(*roles_to_remove, reason=f"ProfileUpdaterCog: Applying theme '{role_theme}'")
|
||||
print(f"ProfileUpdaterTask: Removed roles {[r.name for r in roles_to_remove]} in guild {guild.id}.")
|
||||
changes_made = True
|
||||
await asyncio.sleep(1) # Small delay between actions
|
||||
except discord.Forbidden:
|
||||
print(f"ProfileUpdaterTask: Permission error removing roles in guild {guild.id}.")
|
||||
except discord.HTTPException as e:
|
||||
print(f"ProfileUpdaterTask: HTTP error removing roles in guild {guild.id}: {e}")
|
||||
|
||||
if roles_to_add:
|
||||
try:
|
||||
await member.add_roles(*roles_to_add, reason=f"ProfileUpdaterCog: Applying theme '{role_theme}'")
|
||||
print(f"ProfileUpdaterTask: Added roles {[r.name for r in roles_to_add]} in guild {guild.id}.")
|
||||
changes_made = True
|
||||
except discord.Forbidden:
|
||||
print(f"ProfileUpdaterTask: Permission error adding roles in guild {guild.id}.")
|
||||
except discord.HTTPException as e:
|
||||
print(f"ProfileUpdaterTask: HTTP error adding roles in guild {guild.id}: {e}")
|
||||
|
||||
return changes_made # Return True if any change was attempted/successful
|
||||
|
||||
except Exception as e:
|
||||
print(f"ProfileUpdaterTask: Error during role update for guild {guild.id}: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
|
||||
async def _update_activity(self, activity_info: Dict[str, Optional[str]]):
|
||||
"""Updates the bot's activity status."""
|
||||
activity_type_str = activity_info.get("type")
|
||||
activity_text = activity_info.get("text")
|
||||
|
||||
if not activity_type_str or not activity_text:
|
||||
print("ProfileUpdaterTask: Invalid activity info received from AI.")
|
||||
return
|
||||
|
||||
print(f"ProfileUpdaterTask: Attempting to set activity to {activity_type_str}: '{activity_text}'")
|
||||
|
||||
# Map string type to discord.ActivityType enum
|
||||
activity_type_map = {
|
||||
"playing": discord.ActivityType.playing,
|
||||
"watching": discord.ActivityType.watching,
|
||||
"listening": discord.ActivityType.listening,
|
||||
"competing": discord.ActivityType.competing,
|
||||
# Add streaming later if needed (requires URL)
|
||||
}
|
||||
|
||||
activity_type = activity_type_map.get(activity_type_str.lower())
|
||||
|
||||
if activity_type is None:
|
||||
print(f"ProfileUpdaterTask: Unknown activity type '{activity_type_str}'. Defaulting to 'playing'.")
|
||||
activity_type = discord.ActivityType.playing
|
||||
|
||||
activity = discord.Activity(type=activity_type, name=activity_text)
|
||||
|
||||
try:
|
||||
await self.bot.change_presence(activity=activity)
|
||||
print("ProfileUpdaterTask: Activity updated successfully.")
|
||||
except Exception as e:
|
||||
print(f"ProfileUpdaterTask: Error updating activity: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
async def setup(bot: commands.Bot):
|
||||
"""Adds the ProfileUpdaterCog to the bot."""
|
||||
await bot.add_cog(ProfileUpdaterCog(bot))
|
||||
Loading…
x
Reference in New Issue
Block a user