`\n\n"
"Supported codes:\n"
"- `en` = English\n"
"- `de` = Deutsch\n"
"- `fr` = Français\n"
"- `es` = Español\n"
"- `it` = Italiano\n"
"- `pt` = Português\n"
"- `nl` = Nederlands\n"
"- `ru` = Русский\n"
f"\nCurrent: **{current.upper()}**"
)
try:
lang_code = args.strip().lower()
supported = ['en', 'de', 'fr', 'es', 'it', 'pt', 'nl', 'ru', 'ja', 'zh']
if lang_code not in supported:
return f"❌ Language `{lang_code}` not supported. Use: en, de, fr, es, it, pt, nl, ru, ja, zh"
# Save user language preference
self.user_languages[str(user_id)] = lang_code
self._save_user_settings()
lang_names = {
'en': 'English', 'de': 'Deutsch', 'fr': 'Français',
'es': 'Español', 'it': 'Italiano', 'pt': 'Português',
'nl': 'Nederlands', 'ru': 'Русский', 'ja': '日本語', 'zh': '中文'
}
return f"🌍 **Language set to:** {lang_names.get(lang_code, lang_code.upper())}\n✅ Responses will be in {lang_names.get(lang_code, lang_code)} from now on!"
except Exception as e:
return f"❌ Language error: {str(e)[:100]}"
def _handle_setchannel(self, args: str, chat_system, server_id: str, message) -> str:
"""Handle !setchannel command - set server's bot response channel"""
if not args.strip():
current = self.server_channels.get(str(server_id), 'any')
return (
"📍 **Channel Settings:**\n"
"Usage: `!setchannel <#channel>` or `!setchannel any`\n\n"
"- Mention a channel: `!setchannel #bot-chat`\n"
"- Use a channel ID: `!setchannel 1234567890`\n"
"- Respond in any channel: `!setchannel any`\n\n"
f"Current setting: **{current}**"
)
try:
channel_arg = args.strip().lower()
# Handle "any" option
if channel_arg == 'any':
self.server_channels[str(server_id)] = 'any'
self._save_server_settings()
return "📍 **Channel setting:** Bot will respond in ANY channel\n✅ Setting updated!"
# Try to extract channel ID from mention or number
import re
channel_id_match = re.search(r'<#(\d+)>', channel_arg)
if channel_id_match:
channel_id = channel_id_match.group(1)
elif channel_arg.isdigit():
channel_id = channel_arg
else:
return "❌ Invalid format. Use `!setchannel <#channel>` or channel ID"
# Verify channel exists in this server (if message has guild)
if hasattr(message, 'guild') and message.guild:
try:
channel = message.guild.get_channel(int(channel_id))
if not channel:
return f"❌ Channel not found in this server. Use `!setchannel #channel`"
except:
pass
# Save server channel setting
self.server_channels[str(server_id)] = channel_id
self._save_server_settings()
return f"📍 **Channel set:** Bot will respond in <#{channel_id}>\n✅ Setting updated!"
except Exception as e:
return f"❌ Channel error: {str(e)[:100]}"
def _handle_calculate(self, args: str, chat_system) -> str:
"""Handle !calculate command - perform math calculations"""
if not args.strip():
return (
"🧮 **Math Calculator:**\n"
"Usage: `!calculate `\n\n"
"Examples:\n"
"- `!calculate 2 + 3 * 4` → 14\n"
"- `!calculate sqrt(16)` → 4.0\n"
"- `!calculate sin(pi/2)` → 1.0\n"
"- `!calculate 5!` → 120\n"
"- `!calculate log(100)` → 2.0\n\n"
"Also works in Deutsch:\n"
"- `!calculate berechne 2 + 3` → 5"
)
try:
from improvements_v5_3 import MathCalculator
calc = MathCalculator()
result_data = calc.parse_and_calculate(args)
# Check if calculation was successful
if not result_data.get('success', False):
return f"❌ Invalid expression: `{args}`\nError: {result_data.get('error', 'Unknown error')}"
# Get the formatted result
result = result_data.get('formatted_result', result_data.get('result'))
return f"🧮 **Calculation:**\n```\n{args} = {result}\n```"
except Exception as e:
return f"❌ Calculation error: {str(e)[:100]}"
# ═══════════════════════════════════════════════════════════════════════════════
# DISCORD BOT CLIENT - ACTUAL DISCORD.PY INTEGRATION
# ═══════════════════════════════════════════════════════════════════════════════
class DiscordBotClient:
"""
Actual Discord.py client that connects to Discord servers
- Runs in background thread
- Listens for messages in Discord
- Routes through DiscordBotManager
- Handles events and errors
"""
def __init__(self, bot_manager: DiscordBotManager):
self.bot_manager = bot_manager
self.client = None
self.running = False
self.thread = None
self.last_error = None
logger.info("🤖 Discord Bot Client initialized (not connected)")
async def start_bot(self, token: str):
"""Start Discord bot with token"""
try:
import discord
from discord.ext import commands
# Set up intents for the bot
intents = discord.Intents.default()
intents.message_content = True
intents.messages = True
# Create bot instance
self.client = commands.Bot(command_prefix=self.bot_manager.prefix, intents=intents)
# Register event handlers
@self.client.event
async def on_ready():
"""Bot connected to Discord"""
logger.info(f"✅ Discord bot connected as {self.client.user}")
logger.info(f"🤖 Bot is in {len(self.client.guilds)} servers")
# Set bot status/activity
try:
activity = discord.Activity(
type=discord.ActivityType.watching,
name=f"with AI | Use !help"
)
await self.client.change_presence(status=discord.Status.online, activity=activity)
except Exception as e:
logger.warning(f"Could not set bot status: {e}")
self.running = True
@self.client.event
async def on_message(message):
"""Handle incoming messages"""
try:
# Don't respond to ourselves
if message.author == self.client.user:
return
# Don't respond to other bots
if message.author.bot:
return
# Only process if message is a command or mentions us
if not message.content.startswith(self.bot_manager.prefix) and self.client.user not in message.mentions:
return
logger.info(f"Discord: {message.author} ({message.guild.name}): {message.content}")
# Process message through bot manager
response = self.bot_manager.process_message(
user_id=str(message.author.id),
message=message.content,
chat_system=globals().get('chat_system'), # Get global chat system
server_id=str(message.guild.id) if message.guild else 'dm',
message_obj=message
)
if response:
# Send response in Discord
try:
await message.reply(response, mention_author=False)
except Exception as e:
logger.error(f"Could not send Discord message: {e}")
await message.channel.send(f"❌ Error: {str(e)[:100]}")
except Exception as e:
logger.error(f"Error processing Discord message: {e}", exc_info=True)
try:
await message.channel.send(f"❌ Bot error: {str(e)[:100]}")
except:
pass
@self.client.event
async def on_error(event, *args, **kwargs):
"""Handle Discord errors"""
logger.error(f"Discord bot error in {event}: {__import__('sys').exc_info()}")
# Connect to Discord
logger.info(f"🤖 Connecting Discord bot...")
await self.client.start(token)
except ImportError:
logger.error("❌ discord.py not installed! Install with: pip install discord.py")
self.last_error = "discord.py not installed"
except Exception as e:
logger.error(f"❌ Failed to start Discord bot: {e}", exc_info=True)
self.last_error = str(e)
self.running = False
def run_in_thread(self, token: str):
"""Run bot in background thread"""
import asyncio
def run_bot():
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(self.start_bot(token))
except Exception as e:
logger.error(f"Discord bot thread error: {e}", exc_info=True)
self.last_error = str(e)
# Stop existing thread if running
if self.thread and self.thread.is_alive():
try:
if self.client:
asyncio.run_coroutine_threadsafe(self.client.close(), self.client.loop)
self.running = False
except:
pass
# Start new thread
self.thread = threading.Thread(target=run_bot, daemon=True, name="DiscordBot")
self.thread.start()
logger.info("🤖 Discord bot thread started")
def stop_bot(self):
"""Stop the Discord bot"""
try:
if self.client and self.client.loop:
asyncio.run_coroutine_threadsafe(self.client.close(), self.client.loop)
self.running = False
logger.info("🤖 Discord bot stopped")
except Exception as e:
logger.error(f"Error stopping Discord bot: {e}")
def is_connected(self) -> bool:
"""Check if bot is connected"""
try:
return self.client and self.client.is_ready() and self.running
except:
return False
def get_status(self) -> Dict:
"""Get bot connection status"""
return {
'enabled': self.bot_manager.enabled,
'connected': self.is_connected(),
'running': self.running,
'error': self.last_error,
'servers': len(self.client.guilds) if self.client else 0,
'commands_processed': self.bot_manager.stats['total_commands']
}
# ═══════════════════════════════════════════════════════════════════════════════
# SMART CHAT SYSTEM
# ═══════════════════════════════════════════════════════════════════════════════
class SmartChatSystem:
"""
Intelligent chat system that ties everything together - ULTRA IMPROVED
Features:
- Advanced context awareness
- Multi-turn conversation handling
- Smart response ranking & scoring
- Intelligent caching & memory
- Real-time learning from user feedback
"""
def __init__(self):
# Conversations storage with enhanced tracking
self.conversations: Dict[str, Dict] = {}
# Pre-trained responses
self.responses = self._load_responses()
# Advanced cache for frequently asked questions
self.response_cache = {}
self.cache_hits = 0
self.cache_misses = 0
# Learning system - track what works
self.learned_patterns = {}
self.response_performance = defaultdict(lambda: {'good': 0, 'bad': 0, 'avg_rating': 0})
# Statistics with more details
self.stats = {
'total_messages': 0,
'by_intent': defaultdict(int),
'by_language': defaultdict(int),
'feedback': {'positive': 0, 'negative': 0},
'response_times': [],
'cache_efficiency': 0.0,
'average_confidence': 0.0,
}
# Context manager for multi-turn conversations
self.conversation_context = defaultdict(lambda: {'history': deque(maxlen=10), 'metadata': {}})
logger.info(f"💬 Chat System v2 initialized with {len(self.responses)} pre-trained responses")
def _load_responses(self) -> Dict[str, Dict]:
"""Load pre-trained responses from training files - ULTRA IMPROVED WITH RANKING"""
responses = {}
tag_index = {}
category_stats = defaultdict(int)
# Try to load combined training file first (if available)
combined_file = AppConfig.BASE_DIR / "training_combined_all.json"
if combined_file.exists():
logger.info(f"🎯 Found combined training file! Using training_combined_all.json")
training_files = [combined_file]
else:
logger.info(f"📚 Combined file not found, loading individual training files...")
training_files = list(AppConfig.BASE_DIR.glob("training_*.json"))
logger.info(f"📚 Loading training data from {len(training_files)} file(s)...")
total_loaded = 0
for file_path in training_files:
try:
with open(file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
category = data.get('category', file_path.stem)
priority = data.get('priority', 0.5) # Priority for ranking
if 'training_data' in data:
for item in data['training_data']:
if 'input' in item and 'output' in item:
key = item['input'].lower()
tags = item.get('tags', [])
# Enhanced response object
responses[key] = {
'response': item['output'],
'source': 'pretrained',
'category': category,
'confidence': min(0.95, 0.7 + priority * 0.25), # Dynamic confidence
'uses': 0,
'rating': 0.0,
'tags': tags,
'metadata': item.get('metadata', {}),
'created_at': datetime.now().isoformat(),
'last_used': None,
'performance_score': 0.5
}
# Build tag index
for tag in tags:
tag_lower = tag.lower()
if tag_lower not in tag_index:
tag_index[tag_lower] = []
tag_index[tag_lower].append(key)
total_loaded += 1
category_stats[category] += 1
logger.info(f" ✅ {file_path.name}: {category} ({category_stats[category]} responses)")
except Exception as e:
logger.warning(f" ❌ Failed to load {file_path.name}: {e}")
# Store indices
self.tag_index = tag_index
logger.info(f"📊 Loaded {total_loaded} training examples from {len(category_stats)} categories")
logger.info(f"🏷️ Built index with {len(tag_index)} tags")
return responses
def _build_context_history(self, messages: deque) -> str:
"""Build context from conversation history with pronoun resolution (last 3 exchanges)"""
if not messages:
return ""
context_parts = []
last_messages = list(messages)[-6:] if len(messages) > 0 else []
tracked_topics = [] # Track main topics for pronoun resolution
for msg in last_messages:
if hasattr(msg, 'role') and hasattr(msg, 'content'):
role_prefix = "User" if "USER" in str(msg.role).upper() else "Assistant"
content = msg.content[:100]
# Extract nouns for pronoun resolution
if "USER" in str(msg.role).upper():
# Extract key nouns from user message for topic tracking
nouns = re.findall(r'\b[A-Z][a-z]+\b|\b(?:code|image|python|javascript|java|question|problem|issue)\b', content, re.IGNORECASE)
tracked_topics.extend(nouns[:2]) # Keep track of topics
context_parts.append(f"{role_prefix}: {content}")
# Add pronoun context if available
context_str = " | ".join(context_parts) if context_parts else ""
if tracked_topics:
context_str += f" [Topics: {', '.join(set(tracked_topics))}]"
return context_str
def _find_similar_response(self, message: str, context_history: str = "") -> Optional[Dict]:
"""Find similar response using advanced ranking - ULTRA IMPROVED"""
msg_lower = message.lower()
msg_words = set(msg_lower.split())
# 1. Cache check first
cache_key = hashlib.md5(msg_lower.encode()).hexdigest()[:16]
if cache_key in self.response_cache:
self.cache_hits += 1
logger.debug(f" ♻️ Cache hit for: {message[:40]}")
return self.response_cache[cache_key]
self.cache_misses += 1
# 2. Exact match (highest priority)
if msg_lower in self.responses:
match = self.responses[msg_lower]
match['uses'] += 1
match['last_used'] = datetime.now().isoformat()
logger.info(f" 🎯 Exact match found")
self.response_cache[cache_key] = match
return match
# 3. Similarity-based matching with advanced scoring
candidates = []
for pattern, data in self.responses.items():
# Multi-factor scoring
similarity = calculate_text_similarity(msg_lower, pattern)
# Tag matching bonus
tag_bonus = 0.0
for word in msg_words:
if word in self.tag_index:
if any(key == pattern for key in self.tag_index[word]):
tag_bonus += 0.15
# Performance history bonus
perf_score = data.get('performance_score', 0.5)
# Recency bonus (recently used = good)
recency_bonus = 0.0
if data.get('last_used'):
try:
last_used = datetime.fromisoformat(data['last_used'])
days_ago = (datetime.now() - last_used).days
if days_ago < 7:
recency_bonus = 0.1
except:
pass
# Combine scores
total_score = (
similarity * 0.5 + # Similarity weight
min(tag_bonus, 0.2) * 0.2 + # Tag matching weight (capped)
perf_score * 0.2 + # Performance weight
recency_bonus * 0.1 # Recency weight
)
if total_score > 0.35: # Lower threshold for more coverage
candidates.append((total_score, data, similarity))
if candidates:
# Sort by score
candidates.sort(key=lambda x: x[0], reverse=True)
score, best_match, sim = candidates[0]
if score > 0.35: # Confidence threshold
best_match['uses'] += 1
best_match['last_used'] = datetime.now().isoformat()
logger.info(f" ✨ Similar match found (score: {score:.2f})")
self.response_cache[cache_key] = best_match
return best_match
logger.debug(f" ❌ No good match for: {message[:40]}")
return None
def process_message(self,
session_id: str,
message: str,
language: Optional[str] = None) -> Dict[str, Any]:
"""Process incoming message"""
start_time = time.time()
# Update activity for autonomous trainer
autonomous_trainer.update_activity()
# Statistics
self.stats['total_messages'] += 1
# Detect language if not provided
if not language:
language = nlp_processor.detect_language(message)
self.stats['by_language'][language] += 1
# Get or create conversation
if session_id not in self.conversations:
self.conversations[session_id] = {
'messages': deque(maxlen=AppConfig.MAX_CONTEXT_LENGTH),
'language': language,
'created_at': datetime.now().isoformat()
}
conv = self.conversations[session_id]
# Add user message
conv['messages'].append(Message(
role=MessageRole.USER,
content=message
))
# Recognize intent
intent = nlp_processor.recognize_intent(message)
self.stats['by_intent'][intent['type'].value] += 1
logger.info(f"💬 [{session_id[:8]}] Intent: {intent['type'].value} | Lang: {language}")
# Route to appropriate handler
if intent['type'] == IntentType.CALCULATION:
response = self._handle_calculation(message, language)
elif intent['type'] == IntentType.IMAGE_GENERATION:
response = self._handle_image_generation(message, language)
elif intent['type'] == IntentType.CODE_GENERATION:
response = self._handle_code_generation(message, language)
elif intent['type'] == IntentType.KNOWLEDGE_QUERY:
response = self._handle_knowledge_query(message, language)
elif intent['type'] == IntentType.TRANSLATION:
response = self._handle_translation(message, language)
else:
response = self._handle_conversation(message, language, conv)
# Add assistant message
conv['messages'].append(Message(
role=MessageRole.ASSISTANT,
content=response.get('content', response.get('message', '')),
metadata=response
))
# Learning Integration: Process with Enhanced ML Learner
if ENHANCED_LEARNING_ENABLED and enhanced_learner:
try:
# Add to context manager for future reference
enhanced_learner.process_message_with_learning(
session_id=session_id,
message=message,
response=response.get('content', response.get('message', '')),
handler=intent['type'].value,
intent=intent['type'].value,
confidence=intent['confidence']
)
except Exception as e:
logger.debug(f"Learning integration error: {e}")
# Add timing
response['processing_time'] = time.time() - start_time
return response
def _handle_calculation(self, message: str, language: str) -> Dict:
"""Handle mathematical calculations - IMPROVED with safe eval"""
logger.info(f" 🧮 Processing calculation request: {message[:50]}...")
try:
# Try improved calculation first
try:
from ai_improvements import apply_all_improvements
result = apply_all_improvements('calculation', message)
if result and result.get('success'):
logger.info(f" ✅ Calculation: {result.get('expression')} = {result.get('result')}")
return {
'type': 'math',
'success': True,
'content': f"{result.get('expression')} = {result.get('result')}",
'result': result.get('result'),
'source': 'calculator_improved'
}
except Exception as improve_error:
logger.debug(f"Improved calculation failed: {improve_error}")
# Fallback to original method
if IMPROVEMENTS_AVAILABLE:
# Convert German math expressions if needed
expression = message
if language == 'de':
expression = german_handler.convert_german_math(message)
# Parse and calculate
result = math_calculator.parse_and_calculate(expression)
if result['success']:
explanation = math_calculator.explain_calculation(
result['expression'],
result['result'],
language
)
logger.info(f" ✅ Calculation completed: {result['expression']} = {result['formatted_result']}")
return {
'type': 'math',
'success': True,
'content': explanation,
'expression': result['expression'],
'result': result['result'],
'formatted_result': result['formatted_result'],
'source': 'calculator'
}
# If all else fails
return {
'type': 'text',
'success': False,
'content': f"Could not calculate: {message}",
'source': 'calculator_error'
}
except Exception as e:
logger.error(f" ❌ Exception in calculation: {e}", exc_info=True)
return {
'type': 'text',
'success': False,
'content': f"Calculation error: {str(e)}",
'source': 'system_error'
}
def _handle_image_generation(self, message: str, language: str) -> Dict:
"""Handle image generation request with improved style detection"""
# Extract image description
desc = re.sub(
r'(generate|create|make|draw|paint|design|bild|generiere|erstelle)\s+(image|bild|picture|foto|grafik)?\s*(of|von)?',
'',
message,
flags=re.IGNORECASE
).strip()
# Clean up remaining keywords and articles
# Remove: articles (a, an, the), common prepositions/words (with, in, style, me, my)
desc = re.sub(r'\b(a|an|the|with|in|style|me|my|for|of|von)\b', '', desc, flags=re.IGNORECASE).strip()
# Remove multiple spaces
desc = re.sub(r'\s+', ' ', desc).strip()
if not desc or len(desc) < 3:
logger.warning(f" ⚠️ Image description too short: '{desc}'")
desc = 'abstract art'
logger.info(f" 🎨 Image generation: '{desc}'")
# Detect style from message - improved with more keywords
style = ImageStyle.REALISTIC
msg_lower = message.lower()
style_keywords = {
ImageStyle.ANIME: ['anime', 'manga', 'cartoon', 'japanese'],
ImageStyle.ARTISTIC: ['art', 'painting', 'oil painting', 'watercolor', 'sketch', 'drawing', 'abstract'],
ImageStyle.THREE_D: ['3d', 'three-d', 'cg', 'render', 'model', 'digital'],
ImageStyle.REALISTIC: ['realistic', 'photorealistic', 'real', 'photograph', 'photo'],
ImageStyle.SURREAL: ['surreal', 'surrealism', 'dream', 'fantasy', 'magical', 'fantastical']
}
for style_type, keywords in style_keywords.items():
if any(kw in msg_lower for kw in keywords):
style = style_type
logger.info(f" ✨ Style detected: {style.value}")
break
# Generate image with WEB-TRAINED SYSTEM (learns from Google, Wikipedia, Unsplash)
try:
logger.info(f" 🎨 Attempting image generation...")
# Try web-trained generator first (if available)
web_result = None
try:
try:
from web_trained_generator_real import generate_web_trained_image
logger.info(f" 🧠 Using web-trained generator...")
web_result = generate_web_trained_image(prompt=desc, width=1024, height=1024, style=style.value)
if web_result.success and web_result.base64_data:
logger.info(f" ✅ Image generated! Quality: {web_result.quality_score:.0%}")
return {
'type': 'image',
'success': True,
'content': f'🧠 Image generated in {style.value} style',
'message': f'Image generated in {style.value} style',
'filename': web_result.filename,
'base64_data': web_result.base64_data,
'base64': web_result.base64_data,
'generation_time': web_result.generation_time,
'source': 'web_trained_pil',
'quality_score': web_result.quality_score,
'style': style.value
}
except (ModuleNotFoundError, ImportError):
logger.info(f" ℹ️ Web-trained generator not available, using custom generator...")
except Exception as e:
logger.warning(f" ⚠️ Web result error: {str(e)}")
# Fallback to custom generator
try:
from custom_image_generator import generate_image as custom_generate
logger.info(f" 🎨 Using custom image generator...")
custom_result = custom_generate(prompt=desc, width=1024, height=1024)
if custom_result and custom_result.get('success') and custom_result.get('base64_data'):
logger.info(f" ✅ Image generated! Method: {custom_result.get('method')}")
return {
'type': 'image',
'success': True,
'content': f'✨ Image generated in {style.value} style',
'message': f'Image generated in {style.value} style',
'filename': custom_result.get('filename', 'generated_image.png'),
'base64_data': custom_result.get('base64_data', ''),
'base64': custom_result.get('base64_data', ''),
'generation_time': custom_result.get('generation_time', 0),
'source': 'custom_generator',
'style': style.value
}
except Exception as e:
logger.warning(f" ❌ Custom generator error: {str(e)}")
# If all generation failed, inform user
logger.error(f" ❌ Image generation failed")
return {
'type': 'text',
'success': False,
'content': f"Image generation failed. Please try a different description.",
'source': 'image_generator_error'
}
except Exception as e:
logger.error(f" ❌ Exception in image generation: {e}", exc_info=True)
return {
'type': 'text',
'success': False,
'content': f"Error generating image: {str(e)}",
'source': 'system_error'
}
def _handle_code_generation(self, message: str, language: str) -> Dict:
"""Handle code generation request - Professional code for ALL languages"""
# 1. Try enhanced professional code generator (ALL LANGUAGES!)
try:
from enhanced_code_generator import generate_professional_code
result = generate_professional_code(message)
if result and result.get('code'):
logger.info(f" ✅ Generated professional {result.get('language').upper()} code")
return {
'type': 'code',
'success': True,
'code': result.get('code', ''),
'explanation': result.get('explanation', 'Generated code'),
'source': 'professional_generator',
'language': result.get('language', 'python'),
'confidence': 0.95,
'code_type': result.get('type', 'function')
}
except Exception as enhance_error:
logger.warning(f"Enhanced code generation warning: {enhance_error}")
# 2. Fallback to improved code generation
try:
from ai_improvements import apply_all_improvements
result = apply_all_improvements('code', message)
if result and result.get('success'):
logger.info(f" ✅ Generated code using improved templates")
return {
'type': 'code',
'success': True,
'code': result.get('code', ''),
'explanation': result.get('explanation', 'Generated code'),
'source': 'improved_generator',
'language': result.get('language', 'python'),
'confidence': 0.88
}
except Exception as improve_error:
logger.debug(f"Improved code generation failed: {improve_error}")
# 2. Detect what code is being requested
code_lang, code_request = self._detect_code_request(message)
logger.info(f" 💻 Code Request: {code_lang.upper()} - {code_request[:40]}...")
# 3. Check training data first
training_match = self._find_similar_response(message)
if training_match:
training_match['uses'] += 1
logger.info(f" ✅ Found matching code in training data")
return {
'type': 'code',
'success': True,
'code': training_match['response'],
'explanation': f"From training data: {training_match.get('category', 'code examples')}",
'source': f"training:{training_match.get('category', 'unknown')}",
'confidence': training_match['confidence'],
'language': code_lang
}
# 4. Generate quality code based on language and request
generated_code = self._generate_quality_code(code_lang, code_request)
if generated_code:
return {
'type': 'code',
'success': True,
'code': generated_code['code'],
'explanation': generated_code.get('explanation', ''),
'source': 'generated',
'language': code_lang,
'confidence': 0.85
}
# 5. Search web for code examples
logger.info(f" 🌐 Searching web for {code_lang} code examples...")
try:
results = web_learner.search_all_sources(f"{code_lang} {code_request}", max_results=3)
# Filter for code-related results
code_results = [
r for r in results
if r.source in [SourceType.STACKOVERFLOW, SourceType.GITHUB]
]
if code_results:
best = code_results[0]
logger.info(f" ✅ Found code example from {best.source.value}")
return {
'type': 'code',
'success': True,
'code': best.content,
'explanation': f"{best.title}\n\nFrom: {best.source.value}",
'source': best.source.value,
'url': best.url,
'language': code_lang,
'confidence': 0.75
}
except Exception as e:
logger.warning(f" ⚠️ Error searching for code: {e}")
# 6. Provide smart template with useful example
logger.info(f" 📝 Providing code template for {code_lang}")
template = self._generate_code_template(code_lang, code_request)
return {
'type': 'code',
'success': True,
'code': template['code'],
'explanation': template['explanation'],
'source': 'template',
'language': code_lang,
'tips': template.get('tips', [])
}
def _detect_code_request(self, message: str) -> Tuple[str, str]:
"""Detect programming language and what code is being requested"""
msg_lower = message.lower()
# Language detection patterns
language_patterns = {
'python': [r'\bpython\b', r'\.py\b', r'\bpip\b', r'\bdjango\b', r'\bflask\b'],
'javascript': [r'\bjavascript\b', r'\bjs\b', r'\bnode\b', r'\breact\b', r'\bvue\b'],
'typescript': [r'\btypescript\b', r'\bts\b'],
'java': [r'\bjava\b', r'\bspring\b', r'\bmaven\b'],
'csharp': [r'c#|csharp', r'\.net', r'\basync\b.*\bawait\b'],
'go': [r'\bgo\b', r'\bgoroutine\b'],
'rust': [r'\brust\b', r'\bcargo\b'],
'cpp': [r'\bc\+\+\b|cpp', r'\bstd::', r'\b#include\b'],
'sql': [r'\bsql\b', r'\bselect\b', r'\bdatabase\b'],
'html': [r'\bhtml\b', r'\btag\b'],
'css': [r'\bcss\b', r'\bstyle\b'],
'bash': [r'\bbash\b', r'\bshell\b', r'\bscript\b'],
}
detected_lang = 'python' # Default
for lang, patterns in language_patterns.items():
for pattern in patterns:
if re.search(pattern, msg_lower):
detected_lang = lang
break
# Extract what they want coded
code_request = message
remove_keywords = ['write', 'create', 'generate', 'make', 'write me', 'code for', 'function for']
for kw in remove_keywords:
code_request = re.sub(rf'\b{kw}\b', '', code_request, flags=re.IGNORECASE)
code_request = code_request.strip()
return detected_lang, code_request
def _generate_quality_code(self, language: str, request: str) -> Optional[Dict]:
"""Generate high-quality code with proper syntax and documentation"""
# Common code patterns mapped to templates
code_templates = {
'python': {
'hello': '#!/usr/bin/env python3\n"""Simple hello world program"""\n\ndef main():\n print("Hello, World!")\n\nif __name__ == "__main__":\n main()',
'function': 'def my_function(param1, param2):\n """Function documentation here"""\n result = param1 + param2\n return result',
'class': 'class MyClass:\n """Class documentation"""\n def __init__(self, name):\n self.name = name\n \n def method(self):\n """Method documentation"""\n return f"Hello, {self.name}"',
'loop': 'for i in range(10):\n print(f"Iteration {i}")',
'file': 'with open("file.txt", "r") as f:\n content = f.read()\n print(content)',
'list': 'numbers = [1, 2, 3, 4, 5]\nsquared = [x**2 for x in numbers]\nprint(squared)',
'dict': 'person = {"name": "John", "age": 30}\nprint(person["name"])',
},
'javascript': {
'hello': 'console.log("Hello, World!");',
'function': 'function myFunction(param1, param2) {\n return param1 + param2;\n}',
'arrow': 'const myFunction = (param1, param2) => {\n return param1 + param2;\n};',
'class': 'class MyClass {\n constructor(name) {\n this.name = name;\n }\n method() {\n return `Hello, ${this.name}`;\n }\n}',
'async': 'async function fetchData(url) {\n const response = await fetch(url);\n const data = await response.json();\n return data;\n}',
'loop': 'for (let i = 0; i < 10; i++) {\n console.log(`Iteration ${i}`);\n}',
'array': 'const numbers = [1, 2, 3, 4, 5];\nconst squared = numbers.map(x => x ** 2);\nconsole.log(squared);',
},
'java': {
'hello': 'public class HelloWorld {\n public static void main(String[] args) {\n System.out.println("Hello, World!");\n }\n}',
'function': 'public static int add(int a, int b) {\n return a + b;\n}',
'class': 'public class MyClass {\n private String name;\n \n public MyClass(String name) {\n this.name = name;\n }\n}',
'arraylist': 'List numbers = new ArrayList<>();\nnumbers.add(1);\nnumbers.add(2);\nSystem.out.println(numbers);',
},
'sql': {
'select': 'SELECT * FROM users WHERE age > 18;',
'insert': 'INSERT INTO users (name, email) VALUES ("John", "john@example.com");',
'update': 'UPDATE users SET age = 30 WHERE name = "John";',
'delete': 'DELETE FROM users WHERE id = 1;',
'join': 'SELECT u.name, o.total FROM users u JOIN orders o ON u.id = o.user_id;',
},
}
# Try to find matching template
lang_templates = code_templates.get(language, {})
request_lower = request.lower()
for keyword, code in lang_templates.items():
if keyword in request_lower:
return {'code': code, 'explanation': f"{language} {keyword} example"}
return None
def _generate_code_template(self, language: str, request: str) -> Dict:
"""Generate a useful code template when no exact match found"""
templates_by_lang = {
'python': {
'code': '''#!/usr/bin/env python3
"""
Title: {request}
Description: Solution for {request}
Author: NoahsKI
"""
def main():
"""Main function"""
print("Implementing: {request}")
# Your code here
pass
if __name__ == "__main__":
main()
''',
'explanation': f'Python template for: {request}',
'tips': [
'Use proper function names (snake_case)',
'Add docstrings to functions',
'Handle exceptions with try/except',
'Use list comprehensions for transformations'
]
},
'javascript': {
'code': '''/**
* {request}
* Description: Solution for {request}
*/
async function main() {{
console.log("Implementing: {request}");
// Your code here
}}
main().catch(error => console.error(error));
''',
'explanation': f'JavaScript template for: {request}',
'tips': [
'Use const/let instead of var',
'Use arrow functions for callbacks',
'Use async/await for asynchronous code',
'Add JSDoc comments for documentation'
]
},
'java': {
'code': '''/**
* {request}
* Solution for: {request}
*/
public class Solution {{
public static void main(String[] args) {{
System.out.println("Implementing: {request}");
// Your code here
}}
}}
''',
'explanation': f'Java template for: {request}',
'tips': [
'Use PascalCase for class names',
'Use camelCase for method/variable names',
'Add Javadoc comments /** */',
'Handle checked exceptions properly'
]
},
}
default_template = {
'code': f'# TODO: Implement {request}\n# Add your {language} code here\npass',
'explanation': f'{language.capitalize()} template for: {request}',
'tips': ['Use proper syntax for your language', 'Add comments explaining your code']
}
template = templates_by_lang.get(language, default_template)
# Format template with request
try:
for key in ['code', 'explanation']:
if isinstance(template.get(key), str):
template[key] = template[key].format(request=request)
except:
pass
return template
def _detect_code_language(self, code: str) -> str:
"""Detect programming language from code snippet"""
code_lower = code.lower()
language_patterns = {
'python': [r'\bdef\s+', r'\bimport\s+', r'\bclass\s+', r'print\(', r'#.*'],
'javascript': [r'\bfunction\s+', r'const\s+', r'let\s+', r'console\.log', r'\.js[\'\"]'],
'java': [r'\bpublic\s+class\s+', r'\bpublic\s+static\s+void\s+', r'import\s+java'],
'go': [r'\bfunc\s+', r'\bpackage\s+', r':='],
'rust': [r'\bfn\s+', r'\blet\s+', r'\bmut\s+'],
'cpp': [r'#include', r'\bint\s+main', r'std::'],
'sql': [r'\bSELECT\b', r'\bFROM\b', r'\bWHERE\b'],
}
for lang, patterns in language_patterns.items():
for pattern in patterns:
if re.search(pattern, code_lower):
return lang
return 'text' # Default fallback
def _handle_knowledge_query(self, message: str, language: str) -> Dict:
"""Handle knowledge query - IMPROVED: Training data, knowledge graph, web search with better synthesis"""
logger.info(f" ❓ Processing knowledge query: {message[:60]}...")
# 1. Try improved knowledge synthesis first
try:
from ai_improvements import apply_all_improvements
result = apply_all_improvements('knowledge', message)
if result and result.get('success'):
logger.info(f" ✅ Generated knowledge response using improved synthesis")
return {
'type': 'text',
'success': True,
'content': result.get('content', 'Knowledge synthesized'),
'source': 'improved_knowledge',
'confidence': 0.85
}
except Exception as improve_error:
logger.debug(f"Improved knowledge synthesis failed: {improve_error}")
# 2. Check training data with improved matching
training_match = self._find_similar_response(message)
if training_match:
training_match['uses'] += 1
logger.info(f" ✅ Found in training data (category: {training_match.get('category')})")
return {
'type': 'code' if any(kw in training_match['response'] for kw in ['class ', 'function ', 'def ', '```']) else 'text',
'success': True,
'content': training_match['response'],
'source': f"training:{training_match.get('category', 'unknown')}",
'confidence': training_match['confidence']
}
# 3. Check knowledge graph
try:
kg_result = knowledge_graph.get_knowledge(message)
if kg_result and kg_result.confidence > 0.6:
logger.info(f" 📚 Found in knowledge graph (confidence: {kg_result.confidence:.2f})")
return {
'type': 'text',
'success': True,
'content': kg_result.answer,
'source': 'knowledge_graph',
'confidence': kg_result.confidence,
'sources': kg_result.sources
}
except Exception as e:
logger.debug(f" ⚠️ Knowledge graph lookup error: {e}")
# 4. Search web as fallback
logger.info(f" 🌐 Searching web for answers...")
try:
results = web_learner.search_all_sources(message, max_results=5)
if results:
logger.info(f" ✅ Found {len(results)} relevant web results")
# Synthesize answer
answer = self._synthesize_answer(results, message)
# Store in knowledge graph for future use
try:
knowledge_graph.add_knowledge(
question=message,
answer=answer,
sources=[r.url for r in results[:3]],
language=language
)
logger.info(f" 💾 Cached answer in knowledge graph")
except Exception as e:
logger.debug(f" ⚠️ Could not cache in knowledge graph: {e}")
# Cache for next time
msg_key = message.lower()
self.responses[msg_key] = {
'response': answer,
'source': 'internet',
'category': 'learned',
'confidence': 0.7,
'uses': 1,
'tags': []
}
return {
'type': 'text',
'success': True,
'content': answer,
'sources': [r.to_dict() for r in results[:3]],
'confidence': 0.7,
'source': 'web_search'
}
else:
logger.warning(f" ❌ No web results found")
except Exception as e:
logger.error(f" ❌ Error during web search: {e}")
# Fallback response
return {
'type': 'text',
'success': False,
'content': "I couldn't find information about that. Could you rephrase your question or provide more context?",
'source': 'fallback'
}
def _synthesize_answer(self, results: List[SearchResult], query: str) -> str:
"""Synthesize answer from multiple sources with improved quality"""
if not results:
return "No results found. Please try a different search query."
# Take top 3 results
top_results = results[:3]
# Extract and rank sentences by relevance
ranked_sentences = []
for result in top_results:
content = result.content
if not content or len(content) < 20:
continue
# Split into sentences
result_sentences = re.split(r'[.!?]+\s+', content)
# Rate and keep relevant sentences
for sentence in result_sentences:
sentence = sentence.strip()
if len(sentence) > 30:
relevance = calculate_text_similarity(query, sentence)
if relevance > 0.15: # Lower threshold for more results
ranked_sentences.append((sentence, relevance, result.title))
# Sort by relevance
ranked_sentences.sort(key=lambda x: x[1], reverse=True)
# Build answer with proper structure
if ranked_sentences:
# Take top 4-6 most relevant sentences
answer_sentences = [s[0] for s in ranked_sentences[:6]]
answer = '. '.join(answer_sentences)
if answer and not answer.endswith('.'):
answer += '.'
return answer
# Fallback to first result
content = top_results[0].content
sentences = re.split(r'[.!?]+\s+', content)
sentences = [s.strip() for s in sentences if len(s.strip()) > 30][:3]
answer = '. '.join(sentences)
if answer and not answer.endswith('.'):
answer += '.'
return answer or top_results[0].content
def _handle_translation(self, message: str, language: str) -> Dict:
"""Handle translation request - IMPROVED: Multi-language support with patterns"""
try:
from ai_improvements import apply_all_improvements
result = apply_all_improvements('translation', message, language)
if result and result.get('success'):
logger.info(f"✅ Translation: {result.get('source_language')} → {result.get('target_language')}")
return result
# Fallback
return {
'type': 'text',
'success': False,
'content': 'Translation could not be completed. Please try another text.'
}
except Exception as e:
logger.error(f"Translation error: {e}")
return {
'type': 'text',
'success': False,
'content': f'Translation error: {str(e)}'
}
def _handle_conversation(self, message: str, language: str, conv: Dict) -> Dict:
"""Handle general conversation - ENHANCED with detailed, natural responses"""
msg_lower = message.lower()
# FIRST: Detect if this is a factual question that needs real answers
question_keywords = ['what', 'how', 'why', 'where', 'when', 'who', 'is', 'can you', 'could you', 'what\'s', 'how\'s', 'why\'s', 'was', 'wie', 'warum', 'wo', 'wann', 'wer', 'kannst', 'könntest', 'ist']
is_question = any(q in msg_lower for q in question_keywords) or '?' in message
# For factual questions, prioritize web search FIRST
if is_question:
try:
from web_searcher import search_and_get_answer
web_answer = search_and_get_answer(message)
if web_answer:
logger.info(f" ✅ Web search found answer for factual question: {message[:50]}")
return {
'type': 'text',
'success': True,
'content': web_answer,
'source': 'web_search',
'confidence': 0.8
}
except Exception as web_error:
logger.debug(f"Web search failed for question: {web_error}")
# 1. Try enhanced conversation engine first (detailed, natural responses) - but NOT for factual questions
if not is_question:
try:
from enhanced_conversation_engine import get_enhanced_response
context = self._build_context_history(conv.get('messages', []))
result = get_enhanced_response(message, language, context)
if result and result.get('content'):
logger.info(f" ✅ Generated enhanced conversation response")
return result
except Exception as enhance_error:
logger.debug(f"Enhanced conversation failed: {enhance_error}")
# 2. Build context from conversation history (last 3 exchanges)
context_history = self._build_context_history(conv.get('messages', []))
# 3. Try improved conversation response
try:
from ai_improvements import apply_all_improvements
result = apply_all_improvements('conversation', message)
if result and result.get('success'):
logger.info(f" ✅ Generated conversation response using improved patterns")
return {
'type': 'text',
'success': True,
'content': result.get('content', 'Response generated'),
'source': 'improved_conversation',
'confidence': 0.82
}
except Exception as improve_error:
logger.debug(f"Improved conversation response failed: {improve_error}")
# 4. Check training data for conversations
training_match = self._find_similar_response(message, context_history)
if training_match:
training_match['uses'] += 1
return {
'type': 'text',
'success': True,
'content': training_match['response'],
'source': f"training:{training_match.get('category', 'conversation')}"
}
# 3. Multi-language greetings with varied responses
greeting_patterns = {
'en': {'keywords': ['hello', 'hi', 'hey', 'howdy', 'greetings'],
'responses': [
'Hello! How can I help you today?',
'Hi there! What can I do for you?',
'Hey! What do you need assistance with?',
'Welcome! How can I assist you?'
]},
'de': {'keywords': ['hallo', 'hi', 'guten tag', 'moin', 'guten morgen'],
'responses': [
'Hallo! Wie kann ich dir heute helfen?',
'Hi! Was kann ich für dich tun?',
'Willkommen! Woran kann ich dir helfen?',
'Guten Tag! Was möchtest du wissen?'
]}
}
# Check greetings
for lang_code, greeting_data in greeting_patterns.items():
if any(g in msg_lower for g in greeting_data['keywords']):
return {
'type': 'text',
'success': True,
'content': random.choice(greeting_data['responses']),
'source': 'conversation'
}
# 4. Thanks with varied responses
thanks_patterns = {
'en': {'keywords': ['thank', 'thx', 'thanks', 'appreciate', 'grateful'],
'responses': [
"You're welcome! Anything else I can help with?",
"Happy to help! Is there something else?",
"My pleasure! What else can I do for you?",
"Glad I could help! Need anything else?"
]},
'de': {'keywords': ['danke', 'merci', 'thx', 'dank', 'dankbar'],
'responses': [
'Gerne geschehen! Kann ich dir noch helfen?',
'Sehr gerne! Was sonst noch?',
'Das freut mich! Brauchst du noch was?',
'Jederzeit! Wie kann ich sonst noch helfen?'
]}
}
# Check thanks
for lang_code, thanks_data in thanks_patterns.items():
if any(t in msg_lower for t in thanks_data['keywords']):
return {
'type': 'text',
'success': True,
'content': random.choice(thanks_data['responses']),
'source': 'conversation'
}
# 5. Goodbye with varied responses
goodbye_patterns = {
'en': {'keywords': ['bye', 'goodbye', 'see you', 'farewell', 'catch you'],
'responses': [
'Goodbye! Feel free to come back anytime!',
'See you later! Have a great day!',
'Bye! Come back if you need anything!',
'Take care! Looking forward to helping you again!'
]},
'de': {'keywords': ['auf wiedersehen', 'tschüss', 'ciao', 'bis dann', 'ade'],
'responses': [
'Auf Wiedersehen! Komm gerne wieder!',
'Tschüss! Hab einen schönen Tag!',
'Bis bald! Es war schön, dir zu helfen!',
'Mach\'s gut! Wir sehen uns!'
]}
}
# Check goodbye
for lang_code, goodbye_data in goodbye_patterns.items():
if any(b in msg_lower for b in goodbye_data['keywords']):
return {
'type': 'text',
'success': True,
'content': random.choice(goodbye_data['responses']),
'source': 'conversation'
}
# 6. Default: search web for general conversation (if not a question handled above)
if not is_question:
logger.info(f" 🌐 General conversation, searching web for context...")
results = web_learner.search_all_sources(message, max_results=3)
if results:
answer = self._synthesize_answer(results, message)
return {
'type': 'text',
'success': True,
'content': answer,
'source': 'web_search',
'confidence': 0.65
}
# 8. Ultimate fallback with helpful suggestion
fallback_responses = [
"I'm not sure about that. Can you tell me more or ask something specific?",
"I don't have direct information on that. Could you rephrase the question?",
"That's an interesting topic! Can you be more specific about what you want to know?",
"I'd like to help more! Can you provide more details or context?",
"Ich bin mir da nicht sicher. Kannst du mir mehr Details geben?"
]
return {
'type': 'text',
'success': True,
'content': random.choice(fallback_responses),
'source': 'fallback'
}
def get_response(self, message: str, language: str = 'auto', session_id: str = None) -> Dict[str, Any]:
"""
Main entry point for getting chat responses - IMPROVED with better logging and response quality
This is the method called by the API endpoint
"""
start_time = time.time()
try:
# Generate session ID if not provided
if not session_id:
session_id = str(uuid.uuid4())
logger.info(f"💭 Processing message - Session: {session_id[:8]} | Msg length: {len(message)}")
# Process message using existing process_message method
result = self.process_message(session_id, message, language)
# Ensure 'content' field exists for API compatibility
if 'content' not in result:
result['content'] = result.get('message', result.get('response', 'No response generated'))
# Add response quality metrics
result['quality'] = {
'length': len(result.get('content', '')),
'has_sources': 'sources' in result and len(result.get('sources', [])) > 0,
'confidence': result.get('confidence', 0.5)
}
processing_time = time.time() - start_time
result['processing_time'] = processing_time
logger.info(f"✅ Response generated in {processing_time:.2f}s - Type: {result.get('type', 'unknown')}")
return result
except Exception as e:
logger.error(f"❌ Error in get_response: {e}", exc_info=True)
processing_time = time.time() - start_time
return {
'success': False,
'content': f"Sorry, I encountered an error: {str(e)}",
'type': 'error',
'source': 'system_error',
'processing_time': processing_time
}
def record_feedback(self, session_id: str, message_id: str, feedback_type: str):
"""Record feedback on response"""
if feedback_type == 'positive':
self.stats['feedback']['positive'] += 1
elif feedback_type == 'negative':
self.stats['feedback']['negative'] += 1
def get_stats(self) -> Dict:
"""Get chat statistics"""
total_feedback = sum(self.stats['feedback'].values())
satisfaction = (
self.stats['feedback']['positive'] / total_feedback
if total_feedback > 0
else 0
)
return {
'total_messages': self.stats['total_messages'],
'by_intent': dict(self.stats['by_intent']),
'by_language': dict(self.stats['by_language']),
'feedback': dict(self.stats['feedback']),
'satisfaction_rate': f"{satisfaction * 100:.1f}%",
'active_conversations': len(self.conversations),
'cached_responses': len(self.responses)
}
# ═══════════════════════════════════════════════════════════════════════════════
# PERSISTENT CHAT SESSION MANAGER
# ═══════════════════════════════════════════════════════════════════════════════
class ChatSessionManager:
"""
Manages persistent chat sessions for users
- Save/load conversation history
- Session management
- User preferences
- Chat persistence
"""
def __init__(self):
self.sessions_dir = Path('noahski_data/chats')
self.sessions_dir.mkdir(parents=True, exist_ok=True)
self.active_sessions = {}
self.session_timeout = 86400 * 7 # 7 days in seconds
logger.info("💬 Chat Session Manager initialized")
def create_session(self, user_id: str, chat_name: str) -> Dict:
"""Create a new persistent chat session"""
try:
session_id = f"{user_id}_{chat_name}_{int(time.time())}"
session = {
'id': session_id,
'user_id': user_id,
'name': chat_name,
'created': datetime.now().isoformat(),
'last_active': datetime.now().isoformat(),
'messages': [],
'metadata': {'character': None, 'context': None}
}
self.active_sessions[session_id] = session
self._save_session(session)
logger.info(f"✅ Created chat session: {session_id}")
return {'success': True, 'session_id': session_id, 'session': session}
except Exception as e:
logger.error(f"❌ Error creating session: {e}")
return {'success': False, 'error': str(e)}
def list_sessions(self, user_id: str) -> List[Dict]:
"""List all sessions for a user"""
try:
sessions = []
for session_file in self.sessions_dir.glob(f"{user_id}_*.json"):
with open(session_file, 'r', encoding='utf-8') as f:
session = json.load(f)
sessions.append({
'id': session['id'],
'name': session['name'],
'created': session['created'],
'last_active': session['last_active'],
'message_count': len(session.get('messages', []))
})
return sorted(sessions, key=lambda x: x['last_active'], reverse=True)
except Exception as e:
logger.error(f"Error listing sessions: {e}")
return []
def get_session(self, session_id: str) -> Optional[Dict]:
"""Get session data"""
try:
# Check cache first
if session_id in self.active_sessions:
return self.active_sessions[session_id]
# Load from disk
user_id = session_id.split('_')[0]
for session_file in self.sessions_dir.glob(f"{user_id}_*.json"):
with open(session_file, 'r', encoding='utf-8') as f:
session = json.load(f)
if session['id'] == session_id:
self.active_sessions[session_id] = session
return session
return None
except Exception as e:
logger.error(f"Error getting session: {e}")
return None
def add_message(self, session_id: str, role: str, content: str, metadata: Dict = None) -> bool:
"""Add message to session"""
try:
session = self.get_session(session_id)
if not session:
return False
message = {
'timestamp': datetime.now().isoformat(),
'role': role, # 'user' or 'bot'
'content': content,
'metadata': metadata or {}
}
session['messages'].append(message)
session['last_active'] = datetime.now().isoformat()
self.active_sessions[session_id] = session
self._save_session(session)
return True
except Exception as e:
logger.error(f"Error adding message: {e}")
return False
def delete_session(self, session_id: str) -> bool:
"""Delete a session"""
try:
user_id = session_id.split('_')[0]
# Find and delete the file
for session_file in self.sessions_dir.glob(f"{user_id}_*.json"):
with open(session_file, 'r', encoding='utf-8') as f:
session = json.load(f)
if session['id'] == session_id:
session_file.unlink()
if session_id in self.active_sessions:
del self.active_sessions[session_id]
logger.info(f"✅ Deleted session: {session_id}")
return True
return False
except Exception as e:
logger.error(f"Error deleting session: {e}")
return False
def set_character(self, session_id: str, character: str) -> bool:
"""Set bot character for this session"""
try:
session = self.get_session(session_id)
if not session:
return False
session['metadata']['character'] = character
self._save_session(session)
return True
except Exception as e:
logger.error(f"Error setting character: {e}")
return False
def _save_session(self, session: Dict):
"""Save session to disk"""
try:
user_id = session['user_id']
session_file = self.sessions_dir / f"{user_id}_{session['id']}.json"
with open(session_file, 'w', encoding='utf-8') as f:
json.dump(session, f, indent=2, ensure_ascii=False)
except Exception as e:
logger.error(f"Error saving session: {e}")
def get_stats(self) -> Dict:
"""Get session statistics"""
total_messages = sum(
len(s.get('messages', []))
for s in self.active_sessions.values()
)
return {
'active_sessions': len(self.active_sessions),
'total_messages': total_messages,
'sessions_dir': str(self.sessions_dir)
}
class SecurityManager:
"""Manages security operations including tokens, rate limiting, CSRF protection"""
def __init__(self):
self.rate_limit_store = {} # {ip: [timestamps]}
self.rate_limit_requests = 100 # Max requests
self.rate_limit_window = 3600 # Per hour
self.csrf_tokens = {} # {token_id: token_data}
def validate_token(self, token: str) -> bool:
"""Validate token format and integrity"""
if not token or len(token) < 10:
return False
try:
# Basic JWT-like validation
if token.startswith('Bearer '):
token = token[7:]
return len(token) > 10 and isinstance(token, str)
except:
return False
def encrypt_token(self, token: str, key: str = 'noahski_default_key') -> str:
"""Encrypt token using simple encoding (replace with cryptography lib for production)"""
try:
# For production, use: from cryptography.fernet import Fernet
# For now, use base64 encoding as placeholder
import base64
token_bytes = token.encode()
encoded = base64.b64encode(token_bytes).decode()
return f"encrypted_{encoded}"
except Exception as e:
logger.error(f"Token encryption error: {e}")
return token
def decrypt_token(self, encrypted_token: str) -> str:
"""Decrypt token"""
try:
if not encrypted_token.startswith('encrypted_'):
return encrypted_token
import base64
encoded = encrypted_token.replace('encrypted_', '')
token = base64.b64decode(encoded).decode()
return token
except Exception as e:
logger.error(f"Token decryption error: {e}")
return encrypted_token
def check_rate_limit(self, ip: str, limit: int = 100, window: int = 3600) -> bool:
"""Check if client has exceeded rate limit (returns True if allowed)"""
now = time.time()
if ip not in self.rate_limit_store:
self.rate_limit_store[ip] = []
# Remove old timestamps outside the window
self.rate_limit_store[ip] = [
ts for ts in self.rate_limit_store[ip]
if now - ts < window
]
# Check limit
if len(self.rate_limit_store[ip]) >= limit:
return False # Rate limit exceeded
# Add current timestamp
self.rate_limit_store[ip].append(now)
return True # Within rate limit
def generate_csrf_token(self) -> str:
"""Generate CSRF protection token"""
import uuid
token = f"csrf_{uuid.uuid4().hex}_{int(time.time())}"
self.csrf_tokens[token] = {
'created': time.time(),
'expires': time.time() + 3600 # 1 hour
}
return token
def validate_csrf_token(self, token: str) -> bool:
"""Validate CSRF token"""
if token not in self.csrf_tokens:
return False
token_data = self.csrf_tokens[token]
if time.time() > token_data['expires']:
del self.csrf_tokens[token] # Remove expired token
return False
return True
def sanitize_html(self, text: str) -> str:
"""Remove HTML/script tags from text"""
dangerous_patterns = [
'', ' bool:
"""Validate email format"""
import re
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
def get_client_ip(self, request) -> str:
"""Get client IP from request"""
if request.environ.get('HTTP_X_FORWARDED_FOR'):
return request.environ['HTTP_X_FORWARDED_FOR'].split(',')[0].strip()
return request.remote_addr or 'unknown'
# Global Discord Bot Manager
discord_bot = DiscordBotManager()
# Global Discord Bot Client (actual Discord.py connection)
discord_client = DiscordBotClient(discord_bot)
# Global chat system
chat_system = SmartChatSystem()
# === Enhanced Response System (Better answers + Google search + Fix undefined) ===
try:
from enhanced_response_integration import setup_all_enhancements
setup_all_enhancements(app, chat_system)
ENHANCED_RESPONSES_ENABLED = True
logger.info("✅ Enhanced Response System enabled (Google search, better answers, undefined fix)")
except Exception as e:
ENHANCED_RESPONSES_ENABLED = False
logger.warning(f"⚠️ Enhanced Response System not available: {e}")
# Integrate Universal ML Learning with all handlers
try:
from ml_integration import integrate_ml_with_handlers
integrate_ml_with_handlers(chat_system)
except Exception as e:
logger.error(f"❌ Failed to integrate ML: {e}")
# Integrate Enhanced ML Learning System (Context, Feedback, Python, WebLearning)
try:
from enhanced_ml_learner import get_enhanced_ml_learner
from context_manager import get_context_manager
from feedback_learner import get_feedback_learner
enhanced_learner = get_enhanced_ml_learner()
context_manager = get_context_manager()
feedback_learner = get_feedback_learner()
ENHANCED_LEARNING_ENABLED = True
logger.info("✅ Enhanced ML Learning System integrated (Context, Feedback, Python, Web)")
except Exception as e:
ENHANCED_LEARNING_ENABLED = False
enhanced_learner = None
context_manager = None
feedback_learner = None
logger.warning(f"⚠️ Enhanced ML Learning not fully available: {e}")
# ═══════════════════════════════════════════════════════════════════════════════
# NEURAL NETWORK INITIALIZATION FOR RESPONSE QUALITY PREDICTION
# Optimized for HF Spaces with error handling
# ═══════════════════════════════════════════════════════════════════════════════
NEURAL_NETWORK_ENABLED = False
neural_network = None
if ENHANCED_LEARNING_ENABLED and not AppConfig.IS_HF_SPACE: # Skip NN training on HF for speed
try:
import numpy as np
from enhanced_ml_learner import SimpleNeuralNetwork
logger.info("🧠 Initializing Neural Network for Response Quality Prediction...")
# Create neural network: 20 input features → 16 hidden → 8 hidden → 1 quality score (0-1)
neural_network = SimpleNeuralNetwork([20, 16, 8, 1], learning_rate=0.05)
# Generate training data (response quality features)
# Features: response length, keyword matches, confidence score, context relevance, etc.
np.random.seed(42)
n_samples = 100 if AppConfig.IS_HF_SPACE else 200 # Less samples on HF
# Create training data with response quality patterns
X_train = np.random.randn(n_samples, 20) * 0.5 + 0.5 # Normalized features
# Quality labels: higher quality if certain feature patterns exist
quality_scores = []
for sample in X_train:
# Simple heuristic: average of features as quality
base_quality = np.mean(np.abs(sample[:10])) * 0.8 # 80% from first 10 features
base_quality += np.mean(np.abs(sample[10:15])) * 0.15 # 15% from middle
base_quality += np.mean(np.abs(sample[15:])) * 0.05 # 5% from last
quality_scores.append(np.clip(base_quality, 0, 1))
y_train = np.array(quality_scores).reshape(-1, 1)
# Train the network with fewer epochs on HF Spaces
epochs = 10 if AppConfig.IS_HF_SPACE else 25
logger.info(f" Training on {n_samples} samples with 20 features ({epochs} epochs)...")
history = neural_network.train(X_train, y_train, epochs=epochs, batch_size=32)
# Store in enhanced learner
if enhanced_learner:
enhanced_learner.neural_networks['response_quality'] = neural_network
NEURAL_NETWORK_ENABLED = True
logger.info(f"✅ Neural Network trained successfully!")
logger.info(f" Final Loss: {history['losses'][-1]:.6f}")
logger.info(f" Final Accuracy: {history['accuracies'][-1]:.2%}")
except Exception as e:
NEURAL_NETWORK_ENABLED = False
neural_network = None
logger.warning(f"⚠️ Neural Network initialization failed: {e}")
# Don't crash on HF Spaces if NN fails
if not AppConfig.IS_HF_SPACE:
import traceback
traceback.print_exc()
elif ENHANCED_LEARNING_ENABLED:
logger.info("⚡ Neural Network training skipped on HF Spaces for performance")
NEURAL_NETWORK_ENABLED = False # Use NN from cache if available later
# Integrate AGI Core System (150+ capabilities)
try:
from agi_core_system import get_agi_core
agi_core = get_agi_core()
AGI_SYSTEM_ENABLED = True
logger.info("✅ AGI Core System v6.0 integrated (150+ capabilities)")
except Exception as e:
AGI_SYSTEM_ENABLED = False
agi_core = None
logger.warning(f"⚠️ AGI System not available: {e}")
# Integrate Settings Manager (User Settings & Configuration)
try:
from settings_manager import get_settings_manager
settings_manager = get_settings_manager()
SETTINGS_MANAGER_ENABLED = True
logger.info("✅ Settings Manager initialized")
except Exception as e:
SETTINGS_MANAGER_ENABLED = False
settings_manager = None
logger.warning(f"⚠️ Settings Manager not available: {e}")
# Global chat session manager (persistent chats)
session_manager = ChatSessionManager()
# Global security manager
security_manager = SecurityManager()
# ═══════════════════════════════════════════════════════════════════════════════
# AUTONOMOUS LEARNING SYSTEM INTEGRATION
# ═══════════════════════════════════════════════════════════════════════════════
# Import autonomous learning module
try:
from autonomous_learning import AutonomousLearningSystem
autonomous_learning_system = AutonomousLearningSystem()
autonomous_learning_system.start()
AUTONOMOUS_LEARNING_ENABLED = True
logger.info("✅ Autonomous Learning System integrated and started")
except Exception as e:
AUTONOMOUS_LEARNING_ENABLED = False
autonomous_learning_system = None
logger.warning(f"⚠️ Autonomous Learning System not available: {e}")
# ═══════════════════════════════════════════════════════════════════════════════
# FLASK APPLICATION
# ═══════════════════════════════════════════════════════════════════════════════
# Initialize Flask app
app = Flask(__name__)
if AUTH_PLUGIN_AVAILABLE:
register_auth_routes(app)
logger.info("✓ Authentication routes registered")
@app.route('/login') # ← NICHT eingerückt! Muss auf gleicher Ebene wie @app.route sein
def login_page():
"""Serve login page"""
try:
return send_from_directory('.', 'auth_login.html')
except FileNotFoundError:
logger.error("auth_login.html not found")
return jsonify({'error': 'Login page not found'}), 404
except Exception as e:
logger.error(f"Error serving login page: {e}")
return jsonify({'error': 'Error loading login page'}), 500
app.secret_key = AppConfig.SECRET_KEY
CORS(app, supports_credentials=True, origins=AppConfig.ALLOWED_ORIGINS)
# ─────────────────────────────────────────────────────────────────────────────
# SECURITY MIDDLEWARE
# ─────────────────────────────────────────────────────────────────────────────
@app.before_request
def before_request_security():
"""Apply security checks before processing request"""
try:
# Check rate limit
client_ip = security_manager.get_client_ip(request)
# Skip rate limiting for health checks and static files
if request.path not in ['/health', '/static', '/favicon.ico']:
# DISABLED FOR DEBUGGING: if not security_manager.check_rate_limit(client_ip, limit=100, window=3600):
if False: # Rate limiting disabled for debugging
logger.warning(f"Rate limit exceeded for IP: {client_ip}")
return jsonify({
'success': False,
'error': 'Rate limit exceeded. Please try again later.'
}), 429
# Store client IP in request context for logging
request.client_ip = client_ip
except Exception as e:
logger.error(f"Security check error: {e}")
@app.after_request
def after_request_security(response):
"""Add security headers to all responses"""
try:
# Prevent clickjacking
response.headers['X-Frame-Options'] = 'SAMEORIGIN'
# Prevent MIME type sniffing
response.headers['X-Content-Type-Options'] = 'nosniff'
# Enable XSS protection
response.headers['X-XSS-Protection'] = '1; mode=block'
# Content Security Policy (basic)
response.headers['Content-Security-Policy'] = (
"default-src 'self'; "
"script-src 'self' 'unsafe-inline' 'unsafe-eval'; "
"style-src 'self' 'unsafe-inline'; "
"img-src 'self' data: https:; "
"font-src 'self' data:;"
)
# Referrer Policy
response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
# Feature Policy / Permissions Policy
response.headers['Permissions-Policy'] = (
'camera=(), microphone=(), geolocation=()'
)
# HSTS (optional - only on HTTPS)
if request.scheme == 'https':
response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
return response
except Exception as e:
logger.error(f"Security header error: {e}")
return response
#route
@app.route('/', methods=['GET'])
def index():
"""Root route - redirects to /main"""
return redirect('/main')
@app.route('/main', methods=['GET'])
def main_page():
"""Main route - zeigt Login oder Hauptseite je nach Auth-Status"""
try:
logger.info("=== MAIN ROUTE CALLED ===")
# Check if auth plugin is available
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
logger.error("Auth plugin not available!")
return redirect('/login')
# Token aus Cookie holen
token = request.cookies.get('auth_token')
logger.info(f"Cookie token: {token[:20] if token else 'None'}...")
# Fallback: Token aus Session
if not token:
token = session.get('auth_token')
logger.info(f"Session token: {token[:20] if token else 'None'}...")
# Wenn Token vorhanden, validieren
if token:
try:
logger.info(f"Validating token...")
valid, email = auth_plugin.validate_token(token)
logger.info(f"Token valid: {valid}, Email: {email}")
if valid:
# ✓ Eingeloggt -> Hauptseite zeigen
logger.info(f"✓ User {email} Zugriff auf Hauptseite")
try:
return send_from_directory('.', 'index_ultimate.html')
except FileNotFoundError:
logger.error("index_ultimate.html not found")
return jsonify({'error': 'Main page not found'}), 404
else:
logger.warning(f"Token invalid")
except Exception as e:
logger.error(f"Token-Validierung fehlgeschlagen: {e}", exc_info=True)
# ✗ Nicht eingeloggt -> REDIRECT zu /login (nicht direkt HTML zeigen!)
logger.info("Nicht eingeloggt -> Redirect zu /login")
return redirect('/login')
except Exception as e:
logger.error(f"Fehler in index route: {e}", exc_info=True)
return redirect('/login')
@app.route('/register', methods=['GET'])
def register_page():
"""Register-Seite anzeigen"""
try:
# Prüfe ob User bereits eingeloggt ist
token = request.cookies.get('auth_token') or session.get('auth_token')
if token:
try:
valid, _ = auth_plugin.validate_token(token)
if valid:
# User ist bereits eingeloggt -> Redirect zur Hauptseite
logger.info("User bereits eingeloggt, redirect zu /")
return redirect('/')
except:
pass
# Zeige Register-Seite
logger.info("Zeige Register-Seite")
try:
return send_from_directory('.', 'auth_register.html')
except FileNotFoundError:
logger.error("auth_register.html nicht gefunden!")
return jsonify({'error': 'Register page not found'}), 404
except FileNotFoundError:
logger.error("auth_register.html nicht gefunden!")
return jsonify({'error': 'Register page not found'}), 404
except Exception as e:
logger.error(f"Fehler in register route: {e}", exc_info=True)
return redirect('/login')
@app.route('/settings', methods=['GET'])
def settings_page():
"""Settings-Seite anzeigen"""
try:
# Prüfe ob User eingeloggt ist
token = request.cookies.get('auth_token') or session.get('auth_token')
if token and auth_plugin:
try:
valid, _ = auth_plugin.validate_token(token)
if valid:
logger.info("Zeige Settings-Seite")
try:
return send_from_directory('.', 'settings.html')
except FileNotFoundError:
logger.error("settings.html not found")
return jsonify({'error': 'Settings page not found'}), 404
except:
pass
# Nicht eingeloggt -> Redirect zu Login
logger.info("Nicht eingeloggt -> Redirect zu /login")
return redirect('/login')
except FileNotFoundError:
logger.error("settings.html nicht gefunden!")
return jsonify({'error': 'Settings page not found'}), 404
except Exception as e:
logger.error(f"Fehler in settings route: {e}", exc_info=True)
return redirect('/login')
@app.route('/accsettings', methods=['GET'])
def account_settings_page():
"""Account Settings-Seite anzeigen"""
try:
# Prüfe ob User eingeloggt ist
token = request.cookies.get('auth_token') or session.get('auth_token')
if token and auth_plugin:
try:
valid, _ = auth_plugin.validate_token(token)
if valid:
logger.info("Zeige Account Settings-Seite")
try:
return send_from_directory('.', 'settings.html')
except FileNotFoundError:
logger.error("settings.html not found for account settings")
return jsonify({'error': 'Settings page not found'}), 404
except:
pass
# Nicht eingeloggt -> Redirect zu Login
logger.info("Nicht eingeloggt -> Redirect zu /login")
return redirect('/login')
except FileNotFoundError:
logger.error("settings.html nicht gefunden!")
return jsonify({'error': 'Settings page not found'}), 404
except Exception as e:
logger.error(f"Fehler in account settings route: {e}", exc_info=True)
return redirect('/login')
@app.route('/images/', methods=['GET'])
def serve_image(filename):
"""Serve generated images from the images directory"""
try:
# Clean the filename - replace backslashes with forward slashes
filename = filename.replace('\\', '/')
# Build the full path
image_path = os.path.join('noahski_data', 'generated_media', os.path.basename(filename))
if os.path.exists(image_path) and os.path.isfile(image_path):
try:
return send_from_directory(os.path.dirname(image_path), os.path.basename(image_path), mimetype='image/png')
except Exception as e:
logger.error(f"Error serving image: {e}")
return jsonify({'error': 'Error loading image'}), 500
else:
logger.warning(f"Image not found: {image_path}")
return jsonify({'error': 'Image not found'}), 404
except Exception as e:
logger.error(f"Error serving image: {e}")
return jsonify({'error': 'Error serving image'}), 500
# ═══════════════════════════════════════════════════════════════════════════════
# SIMPLE REGISTRATION API (without email verification)
# ═══════════════════════════════════════════════════════════════════════════════
@app.route('/api/auth/register', methods=['POST'])
def api_register_simple():
"""Simple registration without email verification"""
try:
# Check if auth plugin is available
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'message': 'Authentication system not available'
}), 503
data = request.get_json()
if not data:
return jsonify({
'success': False,
'message': 'No data provided'
}), 400
name = data.get('name', '').strip()
email = data.get('email', '').strip().lower()
password = data.get('password', '')
# Validierung
if not email or not password or not name:
return jsonify({
'success': False,
'message': 'Name, Email and Password are required'
}), 400
if len(name) < 2:
return jsonify({
'success': False,
'message': 'Name must be at least 2 characters'
}), 400
if len(password) < 6:
return jsonify({
'success': False,
'message': 'Password must be at least 6 characters'
}), 400
# Email-Validierung
import re
email_regex = r'^[^\s@]+@[^\s@]+\.[^\s@]+$'
if not re.match(email_regex, email):
return jsonify({
'success': False,
'message': 'Invalid email address'
}), 400
# Prüfe ob User bereits existiert
if auth_plugin.users and email in auth_plugin.users:
return jsonify({
'success': False,
'message': 'Email already registered. Please login instead.'
}), 400
# Erstelle User direkt (ohne Email-Verifizierung)
import time
import hashlib
import secrets
from plugins.auth_plugin import User
# Hash password with salt (same method as auth_plugin)
salt = secrets.token_hex(16)
pwd_hash = hashlib.pbkdf2_hmac('sha256', password.encode(), salt.encode(), 100000)
password_hash = f"{salt}${pwd_hash.hex()}"
# Create user object
user = User(
email=email,
password_hash=password_hash,
username=name,
created_at=time.time(),
verified=True # Direkt verifiziert (keine Email-Bestätigung)
)
# Save user
auth_plugin.users[email] = user
auth_plugin._save_users()
logger.info(f"✅ New user registered: {email} ({name})")
return jsonify({
'success': True,
'message': 'Account created successfully! You can now login.'
}), 200
except Exception as e:
logger.error(f"❌ Registration error: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Registration failed: {str(e)}'
}), 500
# ═══════════════════════════════════════════════════════════════════════════════
# SIMPLE LOGIN API (wrapper for /auth/login)
# ═══════════════════════════════════════════════════════════════════════════════
@app.route('/api/auth/login', methods=['POST'])
def api_login_simple():
"""Simple login API endpoint"""
try:
# Check if auth plugin is available
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'message': 'Authentication system not available'
}), 503
data = request.get_json()
if not data:
return jsonify({
'success': False,
'message': 'No data provided'
}), 400
email = data.get('email', '').strip().lower()
password = data.get('password', '')
if not email or not password:
return jsonify({
'success': False,
'message': 'Email and password are required'
}), 400
# Get client info
ip_address = request.remote_addr
user_agent = request.headers.get('User-Agent')
# Call auth plugin login method
success, message, token = auth_plugin.login(email, password, ip_address, user_agent)
if success and token:
# Create response with cookie
response = jsonify({
'success': True,
'message': message,
'token': token
})
# Set cookie
response.set_cookie(
'auth_token',
token,
max_age=86400, # 24 hours
httponly=True,
secure=False, # Set to True in production with HTTPS
samesite='Lax'
)
# Also save in session as fallback
session['auth_token'] = token
logger.info(f"✅ User logged in: {email}")
return response, 200
else:
return jsonify({
'success': False,
'message': message
}), 401
except Exception as e:
logger.error(f"❌ Login error: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Login failed: {str(e)}'
}), 500
@app.route('/api/logout', methods=['POST', 'GET'])
def api_logout_simple():
"""Simple logout API endpoint"""
try:
# Clear session
session.clear()
# Create response with redirect
response = jsonify({
'success': True,
'message': 'Logged out successfully'
})
# Clear cookie
response.set_cookie('auth_token', '', expires=0)
logger.info("✅ User logged out")
return response, 200
except Exception as e:
logger.error(f"❌ Logout error: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Logout failed: {str(e)}'
}), 500
@app.route('/api/auth/validate', methods=['POST', 'GET'])
def api_validate_simple():
"""Simple token validation API endpoint"""
try:
# Check if auth plugin is available
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'valid': False,
'message': 'Authentication system not available'
}), 503
# Get token from Authorization header
token = request.headers.get('Authorization')
if token and token.startswith('Bearer '):
token = token[7:]
# Fallback: get from cookie
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({
'success': False,
'valid': False,
'message': 'No token provided'
}), 400
# Validate token
valid, email = auth_plugin.validate_token(token)
if valid:
return jsonify({
'success': True,
'valid': True,
'email': email
}), 200
else:
return jsonify({
'success': False,
'valid': False,
'message': 'Invalid or expired token'
}), 401
except Exception as e:
logger.error(f"❌ Validate error: {e}", exc_info=True)
return jsonify({
'success': False,
'valid': False,
'message': f'Validation failed: {str(e)}'
}), 500
@app.route('/api/auth/user', methods=['GET'])
def api_user_info():
"""Get current user information"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'message': 'Authentication system not available'
}), 503
# Get token
token = request.headers.get('Authorization')
if token and token.startswith('Bearer '):
token = token[7:]
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({
'success': False,
'message': 'No token provided'
}), 401
# Validate token
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({
'success': False,
'message': 'Invalid token'
}), 401
# Get user info
user_info = auth_plugin.get_user_info(email)
if user_info:
return jsonify({
'success': True,
'user': user_info
}), 200
else:
return jsonify({
'success': False,
'message': 'User not found'
}), 404
except Exception as e:
logger.error(f"❌ User info error: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Failed to get user info: {str(e)}'
}), 500
@app.route('/api/auth/update-settings', methods=['POST'])
def api_update_settings():
"""Update user settings (profile, discord bot, preferences) - WITH SETTINGS MANAGER"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'message': 'Authentication system not available'
}), 503
# Get token
token = request.headers.get('Authorization')
if token and token.startswith('Bearer '):
token = token[7:]
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({
'success': False,
'message': 'No token provided'
}), 401
# Validate token
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({
'success': False,
'message': 'Invalid token'
}), 401
# Get data
data = request.get_json()
if not data:
return jsonify({
'success': False,
'message': 'No settings provided'
}), 400
# Prepare settings dictionary with all possible fields
updates = {}
# Theme, Language, Notifications
if 'theme' in data:
updates['theme'] = data['theme']
if 'language' in data:
updates['language'] = data['language']
if 'notifications' in data:
updates['notifications_enabled'] = data['notifications']
# Chat settings
chat_settings = {}
if 'chat_theme' in data:
chat_settings['theme'] = data['chat_theme']
if 'font_size' in data:
chat_settings['font_size'] = data['font_size']
if 'show_timestamps' in data:
chat_settings['show_timestamps'] = data['show_timestamps']
if 'auto_scroll' in data:
chat_settings['auto_scroll'] = data['auto_scroll']
if chat_settings:
updates['chat'] = chat_settings
# Discord Bot settings
discord_settings = {}
if 'discord_bot_enabled' in data:
discord_settings['enabled'] = data['discord_bot_enabled']
if 'discord_bot_prefix' in data:
discord_settings['prefix'] = data['discord_bot_prefix']
if 'discord_bot_status' in data:
discord_settings['status'] = data['discord_bot_status']
if 'discord_bot_status_text' in data:
discord_settings['status_text'] = data['discord_bot_status_text']
if discord_settings:
updates['discord_bot'] = discord_settings
# Preferences
prefs = {}
if 'agi_enabled' in data:
prefs['agi_enabled'] = data['agi_enabled']
if 'show_reasoning' in data:
prefs['show_reasoning'] = data['show_reasoning']
if 'auto_complete' in data:
prefs['auto_complete'] = data['auto_complete']
if prefs:
updates['preferences'] = prefs
# Privacy settings
privacy = {}
if 'save_chat_history' in data:
privacy['save_chat_history'] = data['save_chat_history']
if 'allow_feedback' in data:
privacy['allow_feedback'] = data['allow_feedback']
if 'data_retention_days' in data:
privacy['data_retention_days'] = data['data_retention_days']
if privacy:
updates['privacy'] = privacy
# Use Settings Manager to save
if SETTINGS_MANAGER_ENABLED and settings_manager:
result = settings_manager.update_user_settings(email, updates)
if result.get('success'):
logger.info(f"✅ Settings updated via Settings Manager for: {email}")
# Apply discord bot settings if provided
if discord_settings and discord_settings.get('enabled'):
try:
discord_bot.enable_bot(
discord_settings.get('prefix', '!'),
discord_settings.get('status', 'online')
)
logger.info(f"🤖 Discord Bot configured for: {email}")
except Exception as e:
logger.warning(f"⚠️ Could not apply Discord settings: {e}")
return jsonify({
'success': True,
'message': 'Settings saved successfully',
'settings': result.get('settings', updates)
}), 200
else:
return jsonify({
'success': False,
'message': result.get('error', 'Failed to save settings')
}), 500
else:
# Fallback: save to file
try:
settings_dir = Path('noahski_data/auth')
settings_dir.mkdir(parents=True, exist_ok=True)
settings_file = settings_dir / f'{email.replace("@", "_at_")}_settings.json'
with open(settings_file, 'w', encoding='utf-8') as f:
json.dump({
'email': email,
'settings': updates,
'updated_at': datetime.now().isoformat()
}, f, indent=2)
logger.info(f"✅ Settings saved (fallback) for: {email}")
return jsonify({
'success': True,
'message': 'Settings saved successfully',
'settings': updates
}), 200
except Exception as e:
logger.error(f"❌ Error saving settings: {e}")
return jsonify({
'success': False,
'message': f'Error saving settings: {str(e)}'
}), 500
except Exception as e:
logger.error(f"Error in update settings: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Error updating settings: {str(e)}'
}), 500
@app.route('/api/auth/get-settings', methods=['GET'])
def api_get_settings():
"""Get user settings - WITH SETTINGS MANAGER"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'message': 'Authentication system not available'
}), 503
# Get token
token = request.headers.get('Authorization')
if token and token.startswith('Bearer '):
token = token[7:]
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({
'success': False,
'message': 'No token provided'
}), 401
# Validate token
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({
'success': False,
'message': 'Invalid token'
}), 401
# Use Settings Manager to load
if SETTINGS_MANAGER_ENABLED and settings_manager:
user_settings = settings_manager.get_user_settings(email)
return jsonify({
'success': True,
'settings': user_settings,
'timestamp': datetime.now().isoformat()
}), 200
else:
# Fallback: load from file
try:
settings_dir = Path('noahski_data/auth')
settings_file = settings_dir / f'{email.replace("@", "_at_")}_settings.json'
if settings_file.exists():
with open(settings_file, 'r', encoding='utf-8') as f:
data = json.load(f)
return jsonify({
'success': True,
'settings': data.get('settings', {}),
'updated_at': data.get('updated_at')
}), 200
else:
# Return default settings
return jsonify({
'success': True,
'settings': {
'theme': 'dark',
'language': 'de',
'notifications_enabled': True,
'discord_bot': {
'enabled': False,
'prefix': '!',
'status': 'online'
}
}
}), 200
except Exception as e:
logger.error(f"Error loading settings (fallback): {e}")
return jsonify({
'success': True,
'settings': {
'theme': 'dark',
'language': 'de',
'notifications_enabled': True,
'discord_bot': {
'enabled': False,
'prefix': '!',
'status': 'online'
}
}
}), 200
except Exception as e:
logger.error(f"Error in get settings: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Error getting settings: {str(e)}'
}), 500
@app.route('/api/settings/apply-theme', methods=['POST'])
def api_apply_theme():
"""Apply theme setting and return configuration"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({'success': False, 'message': 'Auth not available'}), 503
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({'success': False, 'message': 'No token'}), 401
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({'success': False, 'message': 'Invalid token'}), 401
data = request.get_json()
theme = data.get('theme', 'dark')
if SETTINGS_MANAGER_ENABLED and settings_manager:
settings_manager.apply_theme(email, theme)
logger.info(f"🎨 Theme '{theme}' applied for user: {email}")
return jsonify({
'success': True,
'theme': theme,
'applied': True,
'css_variables': {
'primary': '#667eea' if theme == 'dark' else '#8b5cf6',
'bg': '#1f2937' if theme == 'dark' else '#ffffff',
'text': '#ffffff' if theme == 'dark' else '#1f2937'
}
}), 200
except Exception as e:
logger.error(f"Theme error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/settings/apply-language', methods=['POST'])
def api_apply_language():
"""Apply language setting"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({'success': False, 'message': 'Auth not available'}), 503
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({'success': False, 'message': 'No token'}), 401
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({'success': False, 'message': 'Invalid token'}), 401
data = request.get_json()
language = data.get('language', 'de')
if SETTINGS_MANAGER_ENABLED and settings_manager:
settings_manager.apply_language(email, language)
logger.info(f"🌍 Language '{language}' applied for user: {email}")
return jsonify({
'success': True,
'language': language,
'applied': True,
'region': language
}), 200
except Exception as e:
logger.error(f"Language error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/settings/reset', methods=['POST'])
def api_reset_settings():
"""Reset settings to defaults"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({'success': False, 'message': 'Auth not available'}), 503
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({'success': False, 'message': 'No token'}), 401
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({'success': False, 'message': 'Invalid token'}), 401
if SETTINGS_MANAGER_ENABLED and settings_manager:
success = settings_manager.reset_user_settings(email)
if success:
logger.info(f"🔄 Settings reset for user: {email}")
return jsonify({
'success': True,
'message': 'Settings reset to defaults',
'settings': settings_manager.get_user_settings(email)
}), 200
else:
return jsonify({'success': False, 'message': 'Reset failed'}), 500
else:
return jsonify({'success': False, 'message': 'Settings manager not available'}), 503
except Exception as e:
logger.error(f"Reset error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/settings/export', methods=['GET'])
def api_export_settings():
"""Export user settings as JSON"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({'success': False, 'message': 'Auth not available'}), 503
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({'success': False, 'message': 'No token'}), 401
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({'success': False, 'message': 'Invalid token'}), 401
if SETTINGS_MANAGER_ENABLED and settings_manager:
result = settings_manager.export_user_data(email)
logger.info(f"📤 Settings exported for user: {email}")
return jsonify(result), 200
else:
return jsonify({'success': False, 'message': 'Settings manager not available'}), 503
except Exception as e:
logger.error(f"Export error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/auth/stats', methods=['GET'])
def api_auth_stats():
"""Get user-specific statistics"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'message': 'Authentication system not available'
}), 503
# Get token
token = request.headers.get('Authorization')
if token and token.startswith('Bearer '):
token = token[7:]
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({
'success': False,
'message': 'No token provided'
}), 401
# Validate token
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({
'success': False,
'message': 'Invalid token'
}), 401
# Get user info
user_info = auth_plugin.get_user_info(email)
if not user_info:
return jsonify({
'success': False,
'message': 'User not found'
}), 404
# Calculate account age in days
account_age_days = 0
if 'created_at' in user_info:
created_timestamp = user_info['created_at']
if isinstance(created_timestamp, (int, float)):
created_date = datetime.fromtimestamp(created_timestamp)
account_age_days = (datetime.now() - created_date).days
# Return user stats
return jsonify({
'success': True,
'account_age_days': account_age_days,
'active_sessions': 1,
'messages_sent': getattr(chat_system, 'stats', {}).get('total_messages', 0),
'image_generation_count': getattr(image_generator, 'stats', {}).get('total_generated', 0),
'total_interactions': account_age_days * 10 # Placeholder
}), 200
except Exception as e:
logger.error(f"Error in auth stats: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Error getting stats: {str(e)}'
}), 500
@app.route('/api/auth/change-password', methods=['POST'])
def api_change_password():
"""Change user password"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'message': 'Authentication system not available'
}), 503
# Get token
token = request.headers.get('Authorization')
if token and token.startswith('Bearer '):
token = token[7:]
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({
'success': False,
'message': 'No token provided'
}), 401
# Validate token
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({
'success': False,
'message': 'Invalid token'
}), 401
# Get data
data = request.get_json()
current_password = data.get('current_password')
new_password = data.get('new_password')
if not current_password or not new_password:
return jsonify({
'success': False,
'message': 'Current and new password required'
}), 400
if len(new_password) < 6:
return jsonify({
'success': False,
'message': 'New password must be at least 6 characters'
}), 400
# Verify current password
if email not in auth_plugin.users:
return jsonify({
'success': False,
'message': 'User not found'
}), 404
user = auth_plugin.users[email]
if not auth_plugin._verify_password(current_password, user.password_hash):
return jsonify({
'success': False,
'message': 'Current password is incorrect'
}), 401
# Hash new password
import secrets
import hashlib
salt = secrets.token_hex(16)
pwd_hash = hashlib.pbkdf2_hmac('sha256', new_password.encode(), salt.encode(), 100000)
new_password_hash = f"{salt}${pwd_hash.hex()}"
# Update password
user.password_hash = new_password_hash
auth_plugin._save_users()
logger.info(f"✅ Password changed for user: {email}")
return jsonify({
'success': True,
'message': 'Password updated successfully'
}), 200
except Exception as e:
logger.error(f"❌ Change password error: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Failed to change password: {str(e)}'
}), 500
@app.route('/api/discord/configure', methods=['POST'])
def api_configure_discord():
"""Configure Discord Bot"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'message': 'Authentication system not available'
}), 503
# Get token
token = request.headers.get('Authorization')
if token and token.startswith('Bearer '):
token = token[7:]
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({
'success': False,
'message': 'No token provided'
}), 401
# Validate token
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({
'success': False,
'message': 'Invalid token'
}), 401
# Get data
data = request.get_json()
if not data:
return jsonify({
'success': False,
'message': 'No configuration provided'
}), 400
# Configure Discord Bot
if data.get('enabled'):
token = data.get('token')
prefix = data.get('prefix', '!')
status = data.get('status', 'online')
if not token:
return jsonify({
'success': False,
'message': 'Discord bot token required'
}), 400
discord_bot.enable_bot(token, prefix, status)
# Start actual Discord bot in background thread
try:
discord_client.run_in_thread(token)
logger.info(f"✅ Discord bot client started for user: {email}")
except Exception as e:
logger.error(f"❌ Failed to start Discord bot client: {e}")
logger.info(f"✅ Discord bot enabled for user: {email}")
else:
discord_bot.disable_bot()
# Stop actual Discord bot
try:
discord_client.stop_bot()
logger.info(f"✅ Discord bot client stopped for user: {email}")
except Exception as e:
logger.error(f"❌ Failed to stop Discord bot client: {e}")
logger.info(f"✅ Discord bot disabled for user: {email}")
return jsonify({
'success': True,
'message': 'Discord bot configured successfully',
'enabled': discord_bot.enabled
}), 200
except Exception as e:
logger.error(f"Discord config error: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Error configuring Discord: {str(e)}'
}), 500
@app.route('/api/discord/status', methods=['GET'])
def api_discord_status():
"""Get Discord Bot Status"""
try:
return jsonify({
'success': True,
'enabled': discord_bot.enabled,
'prefix': discord_bot.prefix,
'status': discord_bot.status,
'stats': {
'total_commands': discord_bot.stats['total_commands'],
'success_rate': f"{(discord_bot.stats['command_success'] / max(discord_bot.stats['total_commands'], 1)) * 100:.1f}%",
'active_users': len(discord_bot.stats['active_users'])
}
}), 200
except Exception as e:
logger.error(f"Discord status error: {e}")
return jsonify({
'success': False,
'message': f'Error getting status: {str(e)}'
}), 500
@app.route('/api/auth/delete-account', methods=['POST'])
def api_delete_account():
"""Delete user account"""
try:
if not AUTH_PLUGIN_AVAILABLE or not auth_plugin:
return jsonify({
'success': False,
'message': 'Authentication system not available'
}), 503
# Get token
token = request.headers.get('Authorization')
if token and token.startswith('Bearer '):
token = token[7:]
if not token:
token = request.cookies.get('auth_token')
if not token:
return jsonify({
'success': False,
'message': 'No token provided'
}), 401
# Validate token
valid, email = auth_plugin.validate_token(token)
if not valid:
return jsonify({
'success': False,
'message': 'Invalid token'
}), 401
# Delete user
if email in auth_plugin.users:
del auth_plugin.users[email]
auth_plugin._save_users()
# Logout (invalidate token)
if token in auth_plugin.sessions:
del auth_plugin.sessions[token]
auth_plugin._save_sessions()
logger.info(f"✅ Account deleted: {email}")
return jsonify({
'success': True,
'message': 'Account deleted successfully'
}), 200
else:
return jsonify({
'success': False,
'message': 'User not found'
}), 404
except Exception as e:
logger.error(f"❌ Delete account error: {e}", exc_info=True)
return jsonify({
'success': False,
'message': f'Failed to delete account: {str(e)}'
}), 500
# ═══════════════════════════════════════════════════════════════════════════════
# AUTONOMOUS LEARNING API ENDPOINTS
# ═══════════════════════════════════════════════════════════════════════════════
@app.route('/api/autonomous/trigger', methods=['POST'])
def autonomous_trigger():
"""Manually trigger autonomous learning session"""
if not AUTONOMOUS_LEARNING_ENABLED:
return jsonify({'error': 'Autonomous learning not available'}), 503
try:
if autonomous_learning_system.running and \
autonomous_learning_system.learning_thread and \
autonomous_learning_system.learning_thread.is_alive():
return jsonify({
'success': False,
'message': 'Learning already in progress'
})
autonomous_learning_system._start_learning_session()
return jsonify({
'success': True,
'message': 'Learning session started'
})
except Exception as e:
logger.error(f"Trigger learning error: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/api/autonomous/search', methods=['POST'])
def autonomous_search():
"""Search autonomous learning knowledge base"""
if not AUTONOMOUS_LEARNING_ENABLED:
return jsonify({'error': 'Autonomous learning not available'}), 503
try:
data = request.get_json()
query = data.get('query', '')
top_k = data.get('top_k', 5)
results = autonomous_learning_system.vector_db.search(query, top_k=top_k)
return jsonify({
'success': True,
'results': [
{
'title': entry.title,
'source': entry.source,
'domain': entry.domain,
'confidence': entry.confidence,
'quality_score': entry.quality_score,
'similarity': float(similarity),
'snippet': entry.content[:300]
}
for entry, similarity in results
]
})
except Exception as e:
logger.error(f"Knowledge search error: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/media/', methods=['GET'])
def serve_media(filename):
"""Serve generated media files"""
try:
# Sanitize filename
filename = sanitize_filename(filename)
filepath = AppConfig.IMAGES_DIR / filename
if not filepath.exists():
return jsonify({'error': 'File not found'}), 404
try:
return send_from_directory(str(filepath.parent), filepath.name)
except Exception as e:
logger.error(f"Error serving media file: {e}")
return jsonify({'error': 'Error loading file'}), 500
except Exception as e:
logger.error(f"Media serve error: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/api/analyze/image', methods=['POST'])
def api_analyze_image():
"""Analyze uploaded image: OCR, objects, colors, metadata"""
try:
if 'file' not in request.files:
return jsonify({'success': False, 'error': 'No file provided'}), 400
file = request.files['file']
if not file or file.filename == '':
return jsonify({'success': False, 'error': 'No file selected'}), 400
# Save file temporarily
import secrets
temp_filename = secrets.token_hex(8) + '_' + file.filename
temp_path = AppConfig.UPLOADS_DIR / temp_filename
AppConfig.UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
file.save(str(temp_path))
# Analyze image
result = file_analyzer.analyze_image(str(temp_path))
# Clean up
try:
temp_path.unlink()
except:
pass
return jsonify(result), 200 if result.get('success') else 400
except Exception as e:
logger.error(f"Image analysis error: {e}", exc_info=True)
return jsonify({'success': False, 'error': str(e)[:100]}), 500
@app.route('/api/analyze/document', methods=['POST'])
def api_analyze_document():
"""Analyze uploaded document: PDF, DOCX, XLSX, TXT"""
try:
if 'file' not in request.files:
return jsonify({'success': False, 'error': 'No file provided'}), 400
file = request.files['file']
if not file or file.filename == '':
return jsonify({'success': False, 'error': 'No file selected'}), 400
# Check file type
allowed_extensions = ['.pdf', '.docx', '.doc', '.xlsx', '.xls', '.txt', '.pptx']
file_ext = Path(file.filename).suffix.lower()
if file_ext not in allowed_extensions:
return jsonify({
'success': False,
'error': f'File type {file_ext} not supported',
'supported': allowed_extensions
}), 400
# Save file temporarily
import secrets
temp_filename = secrets.token_hex(8) + file.filename
temp_path = AppConfig.UPLOADS_DIR / temp_filename
AppConfig.UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
file.save(str(temp_path))
# Analyze document
result = file_analyzer.analyze_document(str(temp_path))
# Clean up
try:
temp_path.unlink()
except:
pass
return jsonify(result), 200 if result.get('success') else 400
except Exception as e:
logger.error(f"Document analysis error: {e}", exc_info=True)
return jsonify({'success': False, 'error': str(e)[:100]}), 500
@app.route('/api/analyze/file', methods=['POST'])
def api_analyze_file():
"""Analyze any uploaded file"""
try:
if 'file' not in request.files:
return jsonify({'success': False, 'error': 'No file provided'}), 400
file = request.files['file']
if not file or file.filename == '':
return jsonify({'success': False, 'error': 'No file selected'}), 400
# Save file temporarily
import secrets
temp_filename = secrets.token_hex(8) + file.filename
temp_path = AppConfig.UPLOADS_DIR / temp_filename
AppConfig.UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
file.save(str(temp_path))
# Analyze file
result = file_analyzer.get_file_summary(str(temp_path))
# Clean up
try:
temp_path.unlink()
except:
pass
return jsonify(result), 200 if result.get('success') else 400
except Exception as e:
logger.error(f"File analysis error: {e}", exc_info=True)
return jsonify({'success': False, 'error': str(e)[:100]}), 500
@app.route('/api/analyzer/stats', methods=['GET'])
def api_analyzer_stats():
"""Get file analyzer statistics"""
try:
stats = file_analyzer.get_stats()
return jsonify({
'success': True,
'stats': stats,
'image_generator_stats': image_generator.get_stats()
}), 200
except Exception as e:
logger.error(f"Stats error: {e}")
return jsonify({'success': False, 'error': str(e)[:100]}), 500
# ==================== CHAT SESSION MANAGEMENT ENDPOINTS ====================
def sanitize_input(text: str, max_length: int = 10000) -> str:
"""Sanitize user input - removes null bytes and limits length"""
if not text:
return ""
text = text.replace('\x00', '') # Remove null bytes
text = text[:max_length] # Limit length
return text.strip()
def extract_response_features(message: str, response: str, confidence: float = 0.5) -> np.ndarray:
"""
Extract features from message and response for neural network prediction
Returns: 20-dimensional feature vector
"""
try:
import numpy as np
features = np.zeros(20)
# Feature 1-5: Response characteristics
response_length = len(response)
features[0] = min(response_length / 1000, 1.0) # Length (normalized)
features[1] = len(response.split()) / 100 if response else 0 # Word count
features[2] = confidence # Confidence score
# Feature 6-10: Message analysis
message_length = len(message)
features[5] = min(message_length / 500, 1.0) # Message length
features[6] = message.count('?') / max(1, len(message.split())) # Question marks
features[7] = message.count('!') / max(1, len(message.split())) # Exclamation marks
# Feature 11-15: Response quality indicators
if response:
# Check for code in response
features[10] = 1.0 if '```' in response or 'def ' in response or 'class ' in response else 0.3
# Check for structured info
features[11] = 1.0 if any(x in response for x in ['•', '-', '1.', '2.']) else 0.5
# Check for completeness
features[12] = 1.0 if response.endswith(('.', '!', '?', ')')) else 0.7
# Check for technical depth
features[13] = min(response.count(' ') / 100, 1.0)
# Feature 16-20: Additional factors
features[15] = 1.0 if 'error' not in response.lower() else 0.2
features[16] = 1.0 if 'success' in response.lower() else 0.6
features[17] = min(len(set(response.split())) / max(1, len(response.split())), 1.0) # Diversity
features[18] = np.clip(np.random.random() * 0.1 + 0.9, 0, 1) # Add small randomness
features[19] = min(response.count('\n') / 10, 1.0) # Formatting
# Normalize all features to 0-1 range
features = np.clip(features, 0, 1)
return features
except Exception as e:
logger.debug(f"Feature extraction error: {e}")
return np.random.rand(20) # Return random features on error
def predict_response_quality(response: str, message: str = "", confidence: float = 0.5) -> float:
"""
Use neural network to predict response quality (0-1 score)
"""
try:
if not NEURAL_NETWORK_ENABLED or neural_network is None:
return 0.5 # Default middle quality if NN not available
# Extract features
features = extract_response_features(message, response, confidence)
# Predict quality with neural network
features_batch = features.reshape(1, -1)
quality_prediction = neural_network.predict(features_batch)
quality_score = float(quality_prediction[0][0])
# Ensure valid range
quality_score = np.clip(quality_score, 0.0, 1.0)
return quality_score
except Exception as e:
logger.debug(f"Quality prediction error: {e}")
return 0.5
def enhance_response_with_quality_score(response_dict: Dict) -> Dict:
"""
Add neural network quality prediction to response
"""
try:
if NEURAL_NETWORK_ENABLED and neural_network:
message = response_dict.get('content', '')
confidence = response_dict.get('confidence', 0.5)
# Predict quality
quality_score = predict_response_quality(message, "", confidence)
# Add to response
response_dict['quality_score'] = quality_score
response_dict['quality_level'] = (
'excellent' if quality_score > 0.8 else
'good' if quality_score > 0.6 else
'fair' if quality_score > 0.4 else
'poor'
)
else:
response_dict['quality_score'] = None
response_dict['quality_level'] = 'unknown'
return response_dict
except Exception as e:
logger.debug(f"Quality enhancement error: {e}")
return response_dict
@app.route('/api/chats/create', methods=['POST'])
def api_create_chat():
"""Create a new chat session"""
try:
user_id = request.headers.get('X-User-ID', 'anonymous')
data = request.get_json() or {}
chat_name = sanitize_input(data.get('name', 'Chat'))
if not chat_name:
chat_name = 'Chat'
result = session_manager.create_session(user_id, chat_name)
return jsonify(result), 200
except Exception as e:
logger.error(f"Create chat error: {e}")
return jsonify({'success': False, 'error': str(e)[:100]}), 500
@app.route('/api/chats/list', methods=['GET'])
def api_list_chats():
"""List all chats for current user"""
try:
user_id = request.headers.get('X-User-ID', 'anonymous')
sessions = session_manager.list_sessions(user_id)
return jsonify({'success': True, 'chats': sessions}), 200
except Exception as e:
logger.error(f"List chats error: {e}")
return jsonify({'success': False, 'error': str(e)[:100]}), 500
@app.route('/api/chats/', methods=['GET'])
def api_get_chat(session_id: str):
"""Get specific chat session"""
try:
session = session_manager.get_session(session_id)
if not session:
return jsonify({'success': False, 'error': 'Chat not found'}), 404
return jsonify({'success': True, 'chat': session}), 200
except Exception as e:
logger.error(f"Get chat error: {e}")
return jsonify({'success': False, 'error': str(e)[:100]}), 500
@app.route('/api/chats/', methods=['DELETE'])
def api_delete_chat(session_id: str):
"""Delete a chat session"""
try:
result = session_manager.delete_session(session_id)
return jsonify(result), 200 if result.get('success') else 404
except Exception as e:
logger.error(f"Delete chat error: {e}")
return jsonify({'success': False, 'error': str(e)[:100]}), 500
@app.route('/api/chats//character', methods=['POST'])
def api_set_character(session_id: str):
"""Set character/personality for chat"""
try:
data = request.get_json() or {}
character = sanitize_input(data.get('character', ''))
result = session_manager.set_character(session_id, character)
return jsonify(result), 200 if result.get('success') else 404
except Exception as e:
logger.error(f"Set character error: {e}")
return jsonify({'success': False, 'error': str(e)[:100]}), 500
@app.route('/api/chats//message', methods=['POST'])
def api_add_message(session_id: str):
"""Add message to chat session"""
try:
data = request.get_json() or {}
role = data.get('role', 'user')
content = sanitize_input(data.get('content', ''))
if role not in ['user', 'bot', 'system']:
return jsonify({'success': False, 'error': 'Invalid role'}), 400
if not content:
return jsonify({'success': False, 'error': 'Empty message'}), 400
session = session_manager.get_session(session_id)
if not session:
return jsonify({'success': False, 'error': 'Chat not found'}), 404
result = session_manager.add_message(session_id, role, content)
return jsonify(result), 200
except Exception as e:
logger.error(f"Add message error: {e}")
return jsonify({'success': False, 'error': str(e)[:100]}), 500
@app.route('/health', methods=['GET'])
def health():
"""Health check endpoint"""
return jsonify({
'status': 'healthy',
'version': '3.0.0',
'uptime': format_duration(time.time() - app_start_time),
'features': AppConfig.FEATURES,
'components': {
'cache': len(cache.cache) > 0,
'web_learner': True,
'image_generator': True,
'chat_system': True,
'autonomous_trainer': autonomous_trainer.running,
'knowledge_graph': len(knowledge_graph.nodes) > 0
}
})
@app.route('/api/config', methods=['GET'])
def get_config():
"""Get current configuration"""
return jsonify(AppConfig.get_config_summary())
# ───────────────────────────────────────────────────────────────────────────────
# CHAT API ROUTES
# ───────────────────────────────────────────────────────────────────────────────
@app.route('/api/chat', methods=['POST'])
def api_chat():
"""Main chat endpoint with persistent session support and unified AI routing"""
request_start_time = time.time()
try:
data = request.get_json()
if not data:
return jsonify({
'error': 'No data provided',
'success': False,
'content': 'Error: No data provided'
}), 400
# Accept both 'message' and 'prompt' for compatibility
user_message = data.get('message') or data.get('prompt')
if not user_message:
return jsonify({
'error': 'No message or prompt provided',
'success': False,
'content': 'Error: Please provide a message or prompt'
}), 400
# Validate message length - No limit for code generation!
# Allow unlimited length for better code generation
if len(user_message) > 100000: # Very generous limit
return jsonify({
'error': 'Message too long (max 100000 characters)',
'success': False,
'content': 'Error: Message is too long'
}), 413
# Sanitize input
user_message = sanitize_input(user_message)
language = data.get('language', 'auto')
session_id = data.get('session_id', None)
use_agi = data.get('use_agi', True) # Enable AGI by default
show_reasoning = data.get('show_reasoning', False)
# ═══════════════════════════════════════════════════════════════════════════
# PRIORITY 1: SEARCH TRAINING DATA FIRST
# Use pre-trained answers if available (best quality)
# ═══════════════════════════════════════════════════════════════════════════
training_match = None
response_dict = None # Initialize
try:
from training_data_searcher import training_searcher
training_match = training_searcher.search_training(user_message)
if training_match:
logger.info(f"🎓 Using training answer for: {user_message[:50]}")
response_text = training_match.get('answer', '')
response_dict = {
'content': response_text,
'type': 'text',
'source': 'training_data',
'category': training_match.get('category', 'general'),
'confidence': training_match.get('confidence', 0.9)
}
# Skip unified handler and go straight to response building
unified_result = None
except Exception as e:
logger.warning(f"⚠️ Training search error: {e}")
training_match = None
# ═══════════════════════════════════════════════════════════════════════════
# If no training match found, use unified handler or regular AI
# ═══════════════════════════════════════════════════════════════════════════
unified_handler = get_unified_handler()
unified_result = None
if not training_match and unified_handler and not data.get('disable_unified'):
try:
# Use unified handler to detect what user wants
unified_result = unified_handler.process_message(user_message, user_id=session_id or 'default')
logger.info(f"📊 Unified Result: success={unified_result.get('success')}, task_type={unified_result.get('task_type')}, confidence={unified_result.get('confidence'):.0%}")
logger.debug(f" Full result: {unified_result}")
if unified_result.get('success') and unified_result.get('task_type') != 'unknown':
logger.info(f"🎯 Unified Handler: {unified_result.get('task_type')} ({unified_result.get('confidence'):.0%})")
# Format response based on task type
task_type = unified_result.get('task_type')
result_data = unified_result.get('result', {})
logger.info(f"🔀 ROUTING based on task_type={task_type}")
logger.debug(f" result_data={result_data}")
# Check if handler returned an error FIRST
if result_data.get('error'):
response_text = f"❌ Error: {result_data.get('error', 'Unknown error')}"
response_dict = {
'content': response_text,
'type': 'error',
'source': task_type,
'error': result_data.get('error')
}
# Extract main response based on task
elif task_type == 'code_generation' and result_data.get('code'):
logger.warning(f"[CODE_BLOCK_REACHED] task_type={task_type}, has code={result_data.get('code') is not None}")
code = result_data.get('code', '')
# Detect code language from the code itself
code_language = 'python'
if 'function' in code or 'const' in code or 'let' in code:
code_language = 'javascript'
elif 'SELECT' in code or 'INSERT' in code or 'UPDATE' in code:
code_language = 'sql'
elif code.strip().startswith(' 0.8 else 'medium'
except Exception as e:
logger.warning(f"⚠️ Wikipedia enhancement error: {e}")
# Continue with original response if enhancement fails
response_data['response_confidence'] = 0.5
# ═══════════════════════════════════════════════════════════════════════════
# NEURAL NETWORK QUALITY PREDICTION
# Use trained neural network to add quality score to response
# ═══════════════════════════════════════════════════════════════════════════
if NEURAL_NETWORK_ENABLED and response_type == 'text':
try:
confidence_score = response_data.get('response_confidence') or response_data.get('confidence', 0.5)
response_data = enhance_response_with_quality_score(response_data)
# Log quality prediction
quality_score = response_data.get('quality_score')
quality_level = response_data.get('quality_level', 'unknown')
if quality_score is not None:
logger.info(f"🧠 Neural Network Quality: {quality_level.upper()} ({quality_score:.1%})")
except Exception as e:
logger.debug(f"Neural network quality prediction failed: {e}")
response_data['quality_score'] = None
return jsonify(response_data)
except ValueError as e:
logger.error(f"Chat validation error: {e}")
return jsonify({
'success': False,
'error': str(e),
'content': 'Error: Invalid request data',
'message': 'Error: Invalid request data',
'response': 'Error: Invalid request data'
}), 400
except Exception as e:
logger.error(f"Chat error: {e}", exc_info=True)
return jsonify({
'success': False,
'error': 'Internal server error',
'content': 'Sorry, I encountered an unexpected error. Please try again.',
'message': 'Sorry, I encountered an unexpected error. Please try again.',
'response': 'Sorry, I encountered an unexpected error. Please try again.'
}), 500
@app.route('/api/correct', methods=['POST'])
def api_correct():
"""Error correction endpoint - User corrects AI response and AI learns from it"""
"""VALIDIERT user-Korrektionen - nur gute Antworten werden gelernt!"""
try:
data = request.get_json()
if not data:
return jsonify({'success': False, 'error': 'No data provided'}), 400
original_query = data.get('query', '')
original_response = data.get('response', '')
correction = data.get('correction', '')
if not original_query or not original_response:
return jsonify({'success': False, 'error': 'Missing query or response'}), 400
if not correction or len(correction.strip()) < 2:
return jsonify({
'success': False,
'error': 'Correction too short',
'validation': {
'is_valid': False,
'reason': '❌ Deine Antwort ist zu kurz'
}
}), 400
# Log the error and learn from it
if WIKIPEDIA_LEARNING_ENABLED and wiki_fallback_learner:
try:
# ✅ VALIDATE correction BEFORE learning
validation = wiki_fallback_learner.validate_correction(original_query, correction)
logger.info(f"📊 Validation Result: {validation}")
# If validation fails, reject the correction
if not validation['is_valid']:
logger.warning(f"❌ Correction rejected: {validation['reason']}")
return jsonify({
'success': False,
'error': validation['reason'],
'validation': validation,
'learned': False
}), 400
# ✅ Validation passed - now learn from it
wiki_fallback_learner.log_error(
original_query=original_query,
original_response=original_response,
correction=correction
)
# Try to search Wikipedia and get better answer
enhanced, sources, metadata = enhance_ai_response(
response=original_response,
query=original_query,
force_search=True # Force Wikipedia search for correction
)
logger.info(f"📚 Error Correction: Learned from user feedback for '{original_query[:50]}'")
return jsonify({
'success': True,
'message': f'✅ Gelernt! Score: {validation["score"]:.0%}',
'corrected_response': enhanced,
'sources': sources,
'metadata': metadata,
'learned': True,
'validation': validation
})
except Exception as e:
logger.error(f"Error in correction learning: {e}")
return jsonify({
'success': False,
'error': f'Error processing correction: {str(e)}'
}), 500
else:
return jsonify({
'success': False,
'error': 'Wikipedia learning system not available'
}), 503
except Exception as e:
logger.error(f"Error in API correct endpoint: {e}")
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/learning-stats', methods=['GET'])
def api_learning_stats():
"""Get Wikipedia learning system statistics"""
try:
if WIKIPEDIA_LEARNING_ENABLED and wiki_fallback_learner:
# Get statistics from the learner
stats = {
'learned_facts': len(wiki_fallback_learner.learned_facts) if hasattr(wiki_fallback_learner, 'learned_facts') else 0,
'error_log_size': len(wiki_fallback_learner.error_log) if hasattr(wiki_fallback_learner, 'error_log') else 0,
'last_update': wiki_fallback_learner.last_update if hasattr(wiki_fallback_learner, 'last_update') else 'unknown',
'system_enabled': True,
'enhancement_method': 'Wikipedia API',
'confidence_threshold': 0.75,
'reliability': 'high'
}
# Try to load detailed statistics from files
try:
import json
from pathlib import Path
learned_path = Path('learned_facts.json')
error_log_path = Path('error_learning_log.json')
if learned_path.exists():
with open(learned_path, 'r', encoding='utf-8') as f:
learned_data = json.load(f)
stats['learned_facts'] = len(learned_data)
if error_log_path.exists():
with open(error_log_path, 'r', encoding='utf-8') as f:
error_data = json.load(f)
stats['error_log_size'] = len(error_data)
stats['error_categories'] = {
'total_errors': len(error_data),
'recent_errors': len([e for e in error_data if True]) # All for now
}
except:
pass
logger.info(f"📊 Learning Stats requested: {stats}")
return jsonify({
'success': True,
'statistics': stats
})
else:
return jsonify({
'success': False,
'error': 'Wikipedia learning system not available',
'system_enabled': False
}), 503
except Exception as e:
logger.error(f"Error getting learning stats: {e}")
return jsonify({
'success': False,
'error': str(e)
}), 500
@app.route('/api/feedback', methods=['POST'])
def api_feedback():
"""Record user feedback for learning - Enhanced ML Learning"""
try:
data = request.get_json()
if not data:
return jsonify({'success': False, 'error': 'No data provided'}), 400
session_id = data.get('session_id', '')
message_id = data.get('message_id', '')
handler = data.get('handler', 'general')
response_text = data.get('response', '')
feedback_type = data.get('type', 'neutral') # positive, negative, neutral
rating = data.get('rating') # 1-5
comment = data.get('comment', '')
corrections = data.get('corrections') # User's correction if negative
if not session_id or not message_id:
return jsonify({'success': False, 'error': 'Missing session_id or message_id'}), 400
# Record in feedback learner
if ENHANCED_LEARNING_ENABLED and feedback_learner:
try:
feedback_record = feedback_learner.record_feedback(
session_id=session_id,
message_id=message_id,
handler=handler,
response=response_text,
feedback_type=feedback_type,
rating=rating,
user_comment=comment,
corrections=corrections
)
# Get quality score
quality = feedback_learner.get_handler_quality_score(handler)
# Save to disk
feedback_learner.save_feedback_history()
logger.info(f"Feedback recorded: {handler} - {feedback_type} - Rating: {rating}")
return jsonify({
'success': True,
'feedback_recorded': True,
'handler': handler,
'quality_score': quality['quality_score'],
'rating': quality['rating']
})
except Exception as e:
logger.error(f"Error recording feedback: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
else:
return jsonify({'success': False, 'error': 'Feedback learning not available'}), 503
except Exception as e:
logger.error(f"Feedback endpoint error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/tokens-info', methods=['GET'])
def api_tokens_info():
"""Get current token usage information"""
try:
token_info = {}
if context_manager:
token_info = context_manager.get_token_status()
else:
token_info = {
'total_tokens': 0,
'remaining_tokens': 10000000,
'max_tokens': 10000000,
'percentage': 0
}
return jsonify({
'success': True,
'tokens': {
'used': token_info.get('total_tokens', 0),
'remaining': token_info.get('remaining_tokens', 10000000),
'max': token_info.get('max_tokens', 10000000),
'percentage_used': round((token_info.get('total_tokens', 0) / 10000000) * 100, 2),
'message': f"Tokens used: {token_info.get('total_tokens', 0)} / 10,000,000 ({round((token_info.get('total_tokens', 0) / 10000000) * 100, 2)}%)"
}
})
except Exception as e:
logger.error(f"Token info error: {e}")
return jsonify({
'success': False,
'error': str(e),
'tokens': {
'used': 0,
'remaining': 10000000,
'max': 10000000,
'percentage_used': 0,
'message': 'Tokens used: 0 / 10,000,000 (0%)'
}
}), 500
@app.route('/api/learning-status', methods=['GET'])
def api_learning_status():
"""Get overall learning status - Enhanced ML Learning"""
try:
if ENHANCED_LEARNING_ENABLED and enhanced_learner:
status = enhanced_learner.get_overall_learning_status()
return jsonify({
'success': True,
'learning_enabled': True,
'context_awareness': status['context']['total_interactions'],
'feedback_quality': status['feedback_quality']['overall_quality'],
'web_learning_topics': status['web_learning']['topics_learned'],
'python_patterns': status['python_patterns_learned'],
'learning_metrics': status['learning_metrics'],
'top_topics': status['context'].get('top_topics', [])
})
else:
return jsonify({'success': False, 'learning_enabled': False}), 503
except Exception as e:
logger.error(f"Learning status error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/agi-reasoning', methods=['POST'])
def api_agi_reasoning():
"""Multi-step reasoning with transparency"""
try:
data = request.get_json()
problem = data.get('problem', '')
steps = data.get('steps', 5)
if not AGI_SYSTEM_ENABLED or not agi_core:
return jsonify({'error': 'AGI system not available'}), 503
reasoning_chain = agi_core.reasoning_chain(problem, num_steps=min(steps, 10))
return jsonify({
'success': True,
'problem': problem,
'reasoning_steps': reasoning_chain,
'final_reasoning': reasoning_chain[-1] if reasoning_chain else None,
'total_steps': len(reasoning_chain)
})
except Exception as e:
logger.error(f"Reasoning error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/agi-memory', methods=['GET', 'POST'])
def api_agi_memory():
"""Memory operations (retrieve, store, consolidate)"""
try:
if not AGI_SYSTEM_ENABLED or not agi_core:
return jsonify({'error': 'AGI system not available'}), 503
if request.method == 'GET':
query = request.args.get('query', '')
memory_type = request.args.get('type', 'all')
results = agi_core.memory.retrieve(query, memory_type)
return jsonify({
'success': True,
'query': query,
'memory_type': memory_type,
'results_found': len(results),
'memories': results[:10]
})
elif request.method == 'POST':
data = request.get_json()
action = data.get('action', 'store')
if action == 'store':
agi_core.memory.store_long_term(data.get('key', ''), data.get('value', {}))
return jsonify({'success': True, 'action': 'Memory stored'})
elif action == 'consolidate':
result = agi_core.memory.consolidate_memories()
return jsonify({'success': True, 'consolidation': result})
else:
return jsonify({'error': 'Unknown action'}), 400
except Exception as e:
logger.error(f"Memory error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/agi-capabilities', methods=['GET'])
def api_agi_capabilities():
"""Check AGI system capabilities and status"""
try:
if not AGI_SYSTEM_ENABLED or not agi_core:
return jsonify({
'success': True,
'agi_enabled': False,
'capabilities': 0
})
status = agi_core.status_report()
return jsonify({
'success': True,
'agi_enabled': True,
'system': status['system'],
'version': status['version'],
'status': status['status'],
'modules_active': status['modules_active'],
'total_capabilities': status['capabilities'],
'uptime_seconds': status['uptime_seconds'],
'memory_items': status['memory_items'],
'knowledge_entities': status['knowledge_entities'],
'tools_available': status['tools_available'],
'safety_checks': status['safety_checks']
})
except Exception as e:
logger.error(f"Capabilities error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/agi-self-improve', methods=['POST'])
def api_agi_self_improve():
"""Trigger self-improvement cycle"""
try:
if not AGI_SYSTEM_ENABLED or not agi_core:
return jsonify({'error': 'AGI system not available'}), 503
improvement_result = agi_core.self_improve()
return jsonify({
'success': True,
'current_performance': improvement_result['current_performance'],
'improvements_identified': len(improvement_result['suggestions']),
'suggestions': improvement_result['suggestions'][:5],
'expected_performance_gain': improvement_result['new_performance'] - improvement_result['current_performance']
})
except Exception as e:
logger.error(f"Self-improvement error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/stats', methods=['GET'])
def api_stats():
"""Get system statistics"""
try:
# Safe access to all attributes with defaults
stats = {
'total_interactions': len(getattr(cache, 'cache', {})),
'internet_searches': getattr(web_learner, 'search_count', 0),
'auto_learned_from_internet': len(getattr(web_learner, 'learned_facts', [])),
'contextual_responses': len(getattr(chat_system, 'responses', {})),
'images_generated': getattr(image_generator, 'generation_count', 0),
'translations': 0,
'positive_feedback': getattr(chat_system, 'positive_feedback', 0),
'negative_feedback': getattr(chat_system, 'negative_feedback', 0),
'total_responses_known': len(getattr(chat_system, 'responses', {})),
'cache_size': len(getattr(cache, 'cache', {})),
'learning_rate': '95%',
'network_status': 'online'
}
# Add autonomous learning stats if available
if AUTONOMOUS_LEARNING_ENABLED and autonomous_learning_system:
try:
autonomous_stats = autonomous_learning_system.get_stats()
stats['knowledge_entries'] = autonomous_stats.get('knowledge_entries', 0)
except:
stats['knowledge_entries'] = 0
return jsonify(stats)
except Exception as e:
logger.error(f"Stats error: {e}", exc_info=True)
# Return minimal stats on error
return jsonify({
'total_interactions': 0,
'internet_searches': 0,
'auto_learned_from_internet': 0,
'contextual_responses': 0,
'images_generated': 0,
'translations': 0,
'positive_feedback': 0,
'negative_feedback': 0,
'total_responses_known': 0,
'cache_size': 0,
'learning_rate': '0%',
'network_status': 'online',
'error': str(e)
})
# ═══════════════════════════════════════════════════════════════════════════════
# UNIFIED CHAT HANDLER - ALL AI FUNCTIONS IN ONE CHAT
# ═══════════════════════════════════════════════════════════════════════════════
# Initialize unified chat handler (lazy loading on first use)
unified_chat_handler = None
def get_unified_handler():
"""Lazy load unified chat handler"""
global unified_chat_handler
if unified_chat_handler is None:
try:
from unified_chat_handler import UnifiedChatHandler
# Pass training data responses to handler
unified_chat_handler = UnifiedChatHandler(responses=chat_system.responses)
logger.info("✓ Unified Chat Handler initialized with training data")
except Exception as e:
logger.error(f"Failed to initialize Unified Chat Handler: {e}")
return None
return unified_chat_handler
@app.route('/unified-chat', methods=['GET'])
def unified_chat_page():
"""Serve unified chat interface"""
try:
return render_template('unified_chat.html')
except Exception as e:
logger.error(f"Unified chat page error: {e}")
return f"Error loading chat
{str(e)}
", 500
@app.route('/api/unified-chat', methods=['POST'])
def api_unified_chat():
"""
Unified chat endpoint - routes all messages to appropriate handlers
Request:
{
"message": "user message",
"user_id": "optional_user_id"
}
Response:
{
"success": true,
"task_type": "code_generation|image_generation|...",
"confidence": 0.95,
"matched_keywords": ["keywords"],
"result": {...},
"handler": "detected_handler_name"
}
"""
try:
data = request.get_json()
if not data:
return jsonify({
'success': False,
'error': 'No data provided'
}), 400
user_message = data.get('message', '').strip()
if not user_message:
return jsonify({
'success': False,
'error': 'No message provided'
}), 400
user_id = data.get('user_id', 'default')
# Get unified handler
handler = get_unified_handler()
if handler is None:
# Fallback to basic chat
logger.warning("Unified handler not available, using fallback")
return jsonify({
'success': True,
'task_type': 'conversation',
'confidence': 0.5,
'matched_keywords': [],
'result': {
'type': 'conversation',
'response': chat_system.get_response(user_message)['response']
},
'handler': 'fallback_chat'
})
# Process message with unified handler
result = handler.process_message(user_message, user_id=user_id)
# Log unified chat interaction
try:
interaction_log = {
'timestamp': datetime.now().isoformat(),
'user_id': user_id,
'message': user_message,
'task_type': result.get('task_type'),
'confidence': result.get('confidence'),
'handler': result.get('handler')
}
# Could save to database or file here
except:
pass
return jsonify(result)
except Exception as e:
logger.error(f"Unified chat error: {e}", exc_info=True)
return jsonify({
'success': False,
'error': str(e),
'task_type': 'unknown'
}), 500
@app.route('/api/unified-chat/handlers', methods=['GET'])
def api_get_handlers():
"""Get available handlers in unified chat system"""
try:
handler = get_unified_handler()
if handler is None:
return jsonify({'success': False, 'error': 'Handler not available'}), 503
info = handler.get_handler_info()
return jsonify({
'success': True,
'info': info
})
except Exception as e:
logger.error(f"Error getting handlers: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/image-training-stats', methods=['GET'])
def api_image_training_stats():
"""Get image generation training statistics"""
try:
from web_image_training import get_web_training_stats
web_stats = get_web_training_stats()
return jsonify({
'success': True,
'web_training': web_stats,
'system_status': 'trained',
'message': f"System trained on {web_stats['total_sources']} sources (Wikipedia: {web_stats['wikipedia']}, Unsplash: {web_stats['unsplash']}, Google: {web_stats['google']})"
})
except Exception as e:
logger.warning(f"Image training stats error: {e}")
return jsonify({
'success': True,
'web_training': {
'total_sources': 0,
'wikipedia': 0,
'unsplash': 0,
'google': 0,
'subjects_trained': 0
},
'system_status': 'initializing',
'message': 'Image training system initializing...'
})
@app.route('/api/reset', methods=['POST'])
def api_reset():
"""Reset conversation"""
try:
# Clear session or conversation history if needed
session.clear()
return jsonify({
'success': True,
'message': 'Conversation reset successfully'
})
except Exception as e:
logger.error(f"Reset error: {e}", exc_info=True)
return jsonify({'error': str(e)}), 500
# ═════════════════════════════════════════════════════════════════════════════
# NEW ENDPOINTS: IMAGE GENERATION, CODE GENERATION, MATH CALCULATION
# ═════════════════════════════════════════════════════════════════════════════
@app.route('/api/generate-image', methods=['POST'])
def api_generate_image():
"""Generate image from prompt with style selection
Uses hybrid approach:
1. Try custom_image_generator first (most reliable)
2. If fails, try UltraImageGenerator with all fallbacks
3. Guaranteed to return valid PNG
"""
try:
if not AppConfig.FEATURES.get('image_generation'):
return jsonify({
'success': False,
'error': 'Image generation is disabled'
}), 503
data = request.get_json() or {}
prompt = data.get('prompt', '').strip()
if not prompt:
return jsonify({
'success': False,
'error': 'Prompt is required'
}), 400
# Get parameters - always use 4K resolution
width = 3840 # Force 4K UHD
height = 2160 # Force 4K UHD
style = data.get('style', 'realistic')
quality = data.get('quality', 'high')
# Generate image
logger.info(f"🎨 Generating 4K image: {prompt[:50]}... (3840x2160)")
# PRIORITY 1: Try REAL PHOTOS first (Unsplash, Pexels, Pixabay)
try:
logger.info(" 🖼️ [1/4] Trying premium photo sources...")
from premium_image_fetcher import fetch_premium_image
premium_result = fetch_premium_image(prompt, width, height)
if premium_result.get('success'):
logger.info(f" ✅ Premium photo found: {premium_result.get('method')}")
normalized_filename = os.path.basename(premium_result.get('filename', '')).replace('\\', '/')
# Get file size
try:
file_size = os.path.getsize(premium_result['filename'])
size_mb = file_size / (1024 * 1024)
except:
size_mb = 0
return jsonify({
'success': True,
'image_url': f"/images/{normalized_filename}",
'image_path': premium_result.get('filename'),
'prompt': prompt,
'style': style,
'size': f"{width}x{height}",
'quality': quality,
'generation_time': 0,
'source': f"{premium_result.get('method')} (Real Photo!)"
}), 200
except Exception as premium_e:
logger.info(f" ℹ️ Premium sources unavailable: {str(premium_e)[:50]}")
# PRIORITY 2: Try custom_image_generator (BMP 4K)
logger.info(f" 🎨 [2/4] Trying custom_image_generator: {prompt[:30]}...")
try:
from custom_image_generator import generate_image as custom_generate
custom_result = custom_generate(prompt=prompt, width=width, height=height, style=style)
# custom_result is a dict, not an object
if isinstance(custom_result, dict):
if custom_result.get('success') and custom_result.get('base64_data'):
logger.info(f" ✅ Custom generator success! Method: {custom_result.get('method')}")
normalized_filename = os.path.basename(custom_result.get('filename', '')).replace('\\', '/')
return jsonify({
'success': True,
'image_url': f"/images/{normalized_filename}",
'image_path': custom_result.get('filename'),
'image_data': custom_result.get('base64_data'),
'prompt': prompt,
'style': style,
'size': f"{width}x{height}",
'quality': quality,
'generation_time': custom_result.get('generation_time', 0),
'source': custom_result.get('method', 'custom_generator')
}), 200
else:
logger.warning(f" ⚠️ Custom generator returned success=False or no data: {custom_result.get('error', 'unknown')}")
else:
logger.warning(f" ⚠️ Custom generator returned non-dict: {type(custom_result)}")
except ImportError as ie:
logger.warning(f" ⚠️ Import error: {str(ie)}")
except Exception as custom_e:
logger.warning(f" ⚠️ Custom generator exception: {str(custom_e)}", exc_info=True)
# PRIORITY 3: Try UltraImageGenerator with all APIs
logger.info(" 🔄 [3/4] Trying UltraImageGenerator with all APIs...")
result = image_generator.generate(
prompt=prompt,
width=width,
height=height,
style=style,
quality=quality
)
if result.success:
logger.info(f" ✅ UltraImageGenerator success! API: {result.api_used if hasattr(result, 'api_used') else 'unknown'}")
normalized_filename = os.path.basename(result.filename).replace('\\', '/')
return jsonify({
'success': True,
'image_url': f"/images/{normalized_filename}",
'image_path': result.filename,
'image_data': result.base64_data,
'prompt': result.prompt if hasattr(result, 'prompt') else prompt,
'style': style,
'size': f"{width}x{height}",
'quality': quality,
'generation_time': result.generation_time if hasattr(result, 'generation_time') else 0,
'source': result.api_used if hasattr(result, 'api_used') else 'unknown'
}), 200
else:
logger.error(f" ❌ All generation methods failed: {result.error}")
return jsonify({
'success': False,
'error': 'Image generation failed',
'details': result.error if hasattr(result, 'error') else str(result)
}), 500
except Exception as e:
logger.error(f"Image generation error: {e}", exc_info=True)
# ULTIMATE FALLBACK: Generate a simple image locally using PIL
try:
logger.warning(f" 🚨 Using ultimate fallback PIL generator...")
from PIL import Image
from io import BytesIO
import base64
img = Image.new('RGB', (width, height), color=(100, 150, 200))
buffer = BytesIO()
img.save(buffer, format='PNG')
image_b64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
return jsonify({
'success': True,
'image_data': image_b64,
'prompt': prompt,
'style': style,
'size': f"{width}x{height}",
'source': 'fallback_pil',
'warning': 'Using fallback solid color image'
}), 200
except Exception as fallback_e:
logger.error(f" ❌ Ultimate fallback failed: {fallback_e}")
return jsonify({
'success': False,
'error': f'Image generation error: {str(e)}'
}), 500
@app.route('/api/generate-image-custom', methods=['POST'])
def api_generate_image_custom():
"""Generate image using custom, reliable method - GUARANTEED TO WORK"""
try:
from custom_image_generator import generate_image
data = request.get_json() or {}
prompt = data.get('prompt', '').strip()
if not prompt:
return jsonify({
'success': False,
'error': 'Prompt is required'
}), 400
# Get parameters
width = data.get('width', 512)
height = data.get('height', 512)
style = data.get('style', 'default')
# Validate dimensions
width = max(256, min(1024, width))
height = max(256, min(1024, height))
logger.info(f"🎨 [CUSTOM] Generating image: {prompt[:50]}... ({width}x{height})")
# Generate using custom method
result = generate_image(
prompt=prompt,
width=width,
height=height,
style=style
)
if result.success:
logger.info(f" ✅ Custom image generated successfully ({result.method} method)")
return jsonify({
'success': True,
'filename': str(result.filename),
'base64_data': result.base64_data,
'prompt': prompt,
'style': style,
'size': f"{width}x{height}",
'generation_time': result.generation_time,
'method': result.method,
'message': f'Image generated using {result.method.upper()} method (CUSTOM GENERATOR)'
}), 200
else:
logger.error(f" ❌ Custom image generation failed: {result.error}")
return jsonify({
'success': False,
'error': result.error,
'message': 'Custom image generation failed'
}), 500
except ImportError:
logger.error("Custom image generator module not found")
return jsonify({
'success': False,
'error': 'Custom image generator not available'
}), 503
except Exception as e:
logger.error(f"Custom image generation error: {e}", exc_info=True)
return jsonify({
'success': False,
'error': f'Custom image generation error: {str(e)}'
}), 500
@app.route('/api/generate-code', methods=['POST'])
def api_generate_code():
"""Generate or analyze code"""
try:
if not IMPROVEMENTS_AVAILABLE:
return jsonify({
'success': False,
'error': 'Code generation module not available'
}), 503
data = request.get_json() or {}
action = data.get('action', 'generate') # generate, analyze,explain
code_or_description = data.get('code') or data.get('description') or ''
language = data.get('language', 'python').lower()
if not code_or_description:
return jsonify({
'success': False,
'error': 'Code or description is required'
}), 400
if action == 'generate':
result = code_generator.generate_function(code_or_description, language)
return jsonify({
'success': result['success'],
'code': result.get('code', ''),
'language': language,
'templates': result.get('templates', []),
'error': result.get('error')
}), 200 if result['success'] else 400
elif action == 'analyze':
result = code_generator.analyze_code(code_or_description, language)
return jsonify({
'success': True,
'language': language,
'lines': result.get('lines', 0),
'issues': result.get('issues', []),
'suggestions': result.get('suggestions', []),
'complexity': result.get('complexity', 'unknown')
}), 200
else:
return jsonify({
'success': False,
'error': f'Unknown action: {action}',
'supported_actions': ['generate', 'analyze', 'explain']
}), 400
except Exception as e:
logger.error(f"Code generation error: {e}", exc_info=True)
return jsonify({
'success': False,
'error': f'Code generation error: {str(e)}'
}), 500
@app.route('/api/calculate', methods=['POST'])
def api_calculate():
"""Calculate mathematical expressions"""
try:
if not IMPROVEMENTS_AVAILABLE:
return jsonify({
'success': False,
'error': 'Calculator module not available'
}), 503
data = request.get_json() or {}
expression = data.get('expression', '').strip()
language = data.get('language', 'en')
if not expression:
return jsonify({
'success': False,
'error': 'Expression is required'
}), 400
# Detect if German math expression
if language == 'auto':
lang_detect = language_detector.detect(expression)
language = lang_detect['detected_language']
# Convert German math to standard if needed
if language == 'de':
expression = german_handler.convert_german_math(expression)
# Calculate
logger.info(f"🧮 Calculating: {expression}")
result = math_calculator.parse_and_calculate(expression)
if result['success']:
explanation = math_calculator.explain_calculation(
result['expression'],
result['result'],
language
)
return jsonify({
'success': True,
'expression': result['expression'],
'result': result['result'],
'formatted_result': result['formatted_result'],
'explanation': explanation,
'language': language
}), 200
else:
return jsonify({
'success': False,
'error': result.get('error', 'Calculation failed'),
'message': result.get('message', 'Could not calculate expression')
}), 400
except Exception as e:
logger.error(f"Calculation error: {e}", exc_info=True)
return jsonify({
'success': False,
'error': f'Calculation error: {str(e)}'
}), 500
@app.route('/api/detect-language', methods=['POST'])
def api_detect_language():
"""Detect user language"""
try:
if not IMPROVEMENTS_AVAILABLE:
return jsonify({
'success': False,
'error': 'Language detection not available'
}), 503
data = request.get_json() or {}
text = data.get('text', '').strip()
if not text:
return jsonify({
'success': False,
'error': 'Text is required'
}), 400
result = language_detector.detect(text)
return jsonify({
'success': True,
'detected_language': result['detected_language'],
'language_name': result['language_name'],
'confidence': result['confidence']
}), 200
except Exception as e:
logger.error(f"Language detection error: {e}", exc_info=True)
return jsonify({
'success': False,
'error': f'Language detection error: {str(e)}'
}), 500
# ═══════════════════════════════════════════════════════════════════════════════
# NEURAL NETWORK API ENDPOINTS
# ═══════════════════════════════════════════════════════════════════════════════
@app.route('/api/neural/status', methods=['GET'])
def neural_network_status():
"""Get neural network training and status information"""
try:
if not NEURAL_NETWORK_ENABLED or neural_network is None:
return jsonify({
'success': False,
'enabled': False,
'message': 'Neural Network not enabled'
}), 503
status = {
'success': True,
'enabled': True,
'architecture': neural_network.layer_sizes,
'learning_rate': neural_network.learning_rate,
'trained_epochs': neural_network.training_history['epochs'],
'final_loss': neural_network.training_history['losses'][-1] if neural_network.training_history['losses'] else 0,
'final_accuracy': neural_network.training_history['accuracies'][-1] if neural_network.training_history['accuracies'] else 0,
'total_parameters': sum(
layer.weights.size + layer.biases.size
for layer in neural_network.layers
),
'training_samples': len(neural_network.training_history.get('losses', []))
}
return jsonify(status), 200
except Exception as e:
logger.error(f"Neural network status error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/neural/predict', methods=['POST'])
def neural_network_predict():
"""
Test neural network with custom input features
Input: 20-dimensional feature vector
Output: Quality prediction (0-1)
"""
try:
if not NEURAL_NETWORK_ENABLED or neural_network is None:
return jsonify({
'success': False,
'error': 'Neural Network not enabled'
}), 503
data = request.get_json() or {}
# Option 1: Provide custom features
if 'features' in data:
features = np.array(data['features']).reshape(1, -1)
if features.shape[1] != 20:
return jsonify({
'success': False,
'error': f'Expected 20 features, got {features.shape[1]}'
}), 400
# Option 2: Extract from response text
elif 'response' in data and 'message' in data:
response = data['response']
message = data['message']
confidence = data.get('confidence', 0.5)
features = extract_response_features(message, response, confidence).reshape(1, -1)
else:
return jsonify({
'success': False,
'error': 'Provide either "features" array or "message"/"response" text'
}), 400
# Make prediction
prediction = neural_network.predict(features)
quality_score = float(prediction[0][0])
quality_score = np.clip(quality_score, 0.0, 1.0)
return jsonify({
'success': True,
'quality_score': quality_score,
'quality_level': (
'excellent' if quality_score > 0.8 else
'good' if quality_score > 0.6 else
'fair' if quality_score > 0.4 else
'poor'
),
'features_used': 20
}), 200
except Exception as e:
logger.error(f"Neural prediction error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/neural/retrain', methods=['POST'])
def neural_network_retrain():
"""
Retrain neural network with new data (admin only)
"""
try:
if not NEURAL_NETWORK_ENABLED or neural_network is None:
return jsonify({
'success': False,
'error': 'Neural Network not enabled'
}), 503
data = request.get_json() or {}
epochs = data.get('epochs', 20)
batch_size = data.get('batch_size', 32)
# Validate parameters
if not (5 <= epochs <= 200):
return jsonify({
'success': False,
'error': 'Epochs must be between 5 and 200'
}), 400
if not (8 <= batch_size <= 128):
return jsonify({
'success': False,
'error': 'Batch size must be between 8 and 128'
}), 400
logger.info(f"🧠 Retraining Neural Network: {epochs} epochs, batch_size={batch_size}")
# Generate fresh training data
import numpy as np
n_samples = 300
X_train = np.random.randn(n_samples, 20) * 0.5 + 0.5
quality_scores = []
for sample in X_train:
base_quality = np.mean(np.abs(sample[:10])) * 0.8
base_quality += np.mean(np.abs(sample[10:15])) * 0.15
base_quality += np.mean(np.abs(sample[15:])) * 0.05
quality_scores.append(np.clip(base_quality, 0, 1))
y_train = np.array(quality_scores).reshape(-1, 1)
# Train
history = neural_network.train(X_train, y_train, epochs=epochs, batch_size=batch_size)
return jsonify({
'success': True,
'message': f'Network retrained on {n_samples} samples',
'epochs': epochs,
'final_loss': history['losses'][-1],
'final_accuracy': history['accuracies'][-1],
'training_history': {
'losses': history['losses'][-10:], # Last 10 losses
'accuracies': history['accuracies'][-10:] # Last 10 accuracies
}
}), 200
except Exception as e:
logger.error(f"Neural retrain error: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
# ───────────────────────────────────────────────────────────────────────────────
# ERROR HANDLERS
# ───────────────────────────────────────────────────────────────────────────────
@app.errorhandler(404)
def not_found(error):
return jsonify({'error': 'Not found'}), 404
@app.errorhandler(500)
def internal_error(error):
logger.error(f"Internal error: {error}")
return jsonify({'error': 'Internal server error'}), 500
# ───────────────────────────────────────────────────────────────────────────────
# SHUTDOWN HANDLER
# ───────────────────────────────────────────────────────────────────────────────
def shutdown_handler():
"""Clean shutdown"""
logger.info("🛑 Shutting down NoahsKI...")
# Stop autonomous trainer
autonomous_trainer.stop()
# Stop autonomous learning
if AUTONOMOUS_LEARNING_ENABLED:
autonomous_learning_system.stop()
# Save cache
cache.save()
# Save knowledge graph
knowledge_graph._save_graph()
# Save web learning cache
web_learner._save_cache()
logger.info("✅ Shutdown complete")
import atexit
atexit.register(shutdown_handler)
# ═══════════════════════════════════════════════════════════════════════════════
# MAIN ENTRY POINT
# ═══════════════════════════════════════════════════════════════════════════════
app_start_time = time.time()
if __name__ == '__main__':
logger.info("=" * 80)
logger.info(" NoahsKI ULTRA - Starting Server")
logger.info("=" * 80)
logger.info(f" Version: 4.0.0 ULTIMATE")
logger.info(f" Host: {AppConfig.SERVER_HOST}:{AppConfig.SERVER_PORT}")
logger.info(f" Debug: {AppConfig.DEBUG_MODE}")
logger.info("=" * 80)
logger.info(" Loaded Components:")
logger.info(f" ✓ Cache System ({len(cache.cache)} entries)")
logger.info(f" ✓ Image Generator ({len(image_generator.apis)} APIs)")
logger.info(f" ✓ Web Learner (15+ sources)")
logger.info(f" ✓ Autonomous Trainer ({len(autonomous_trainer.training_topics)} topics)")
logger.info(f" ✓ Knowledge Graph ({len(knowledge_graph.nodes)} nodes)")
logger.info(f" ✓ Chat System ({len(chat_system.responses)} responses)")
if AUTONOMOUS_LEARNING_ENABLED:
stats = autonomous_learning_system.get_stats()
logger.info(f" ✓ Autonomous Learning (RAG + FAISS, {stats['knowledge_entries']} entries)")
logger.info("=" * 80)
logger.info(" 🚀 Server starting...")
logger.info("=" * 80)
try:
app.run(
host=AppConfig.SERVER_HOST,
port=AppConfig.SERVER_PORT,
debug=AppConfig.DEBUG_MODE,
threaded=AppConfig.THREADED
)
except KeyboardInterrupt:
logger.info("\n🛑 Received shutdown signal")
shutdown_handler()
except Exception as e:
logger.error(f"❌ Fatal error: {e}", exc_info=True)
shutdown_handler()