386 lines
13 KiB
Python
386 lines
13 KiB
Python
# Description: Main entry point for Lemontropia Suite
|
|
# Provides CLI interface for user testing the Data Capture Engine
|
|
# Run with: python main.py
|
|
|
|
import sys
|
|
import asyncio
|
|
import signal
|
|
from pathlib import Path
|
|
from decimal import Decimal
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
# Add core to path
|
|
core_dir = Path(__file__).parent / "core"
|
|
sys.path.insert(0, str(core_dir))
|
|
|
|
from core.database import DatabaseManager
|
|
from core.project_manager import ProjectManager, ProjectData, SessionData, LootEvent
|
|
from core.log_watcher import LogWatcher, MockLogGenerator
|
|
|
|
# Configure logging for user visibility
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s | %(levelname)-8s | %(message)s',
|
|
datefmt='%H:%M:%S'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LemontropiaApp:
|
|
"""
|
|
Main application class for user testing.
|
|
|
|
Provides interactive CLI for:
|
|
- Project management (create, list, archive)
|
|
- Session control (start, stop)
|
|
- Live log watching (mock mode)
|
|
- Data viewing (loot, stats)
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.db = DatabaseManager()
|
|
self.pm = ProjectManager(self.db)
|
|
self.watcher = None
|
|
self._running = False
|
|
|
|
# Initialize database
|
|
logger.info("🍋 Initializing Lemontropia Suite...")
|
|
if not self.db.initialize():
|
|
logger.error("Failed to initialize database!")
|
|
sys.exit(1)
|
|
|
|
logger.info("✅ Database ready")
|
|
|
|
# Ensure test data exists
|
|
self._ensure_mock_data()
|
|
|
|
def _ensure_mock_data(self):
|
|
"""Create mock chat.log if it doesn't exist."""
|
|
test_data_dir = Path(__file__).parent / "test-data"
|
|
test_data_dir.mkdir(exist_ok=True)
|
|
|
|
mock_log = test_data_dir / "mock-chat.log"
|
|
if not mock_log.exists():
|
|
logger.info("📝 Generating mock chat.log for testing...")
|
|
MockLogGenerator.create_mock_file(mock_log, lines=50)
|
|
logger.info(f"✅ Mock log created: {mock_log}")
|
|
|
|
def print_header(self):
|
|
"""Print application header."""
|
|
print("\n" + "="*60)
|
|
print(" 🍋 LEMONTROPIA SUITE — User Test Build v0.1.0")
|
|
print(" Core Data Capture Engine — Mock Mode")
|
|
print("="*60)
|
|
|
|
def print_menu(self):
|
|
"""Print main menu."""
|
|
print("\n📋 MAIN MENU")
|
|
print("-" * 40)
|
|
print(" 1. 🎯 Create New Project")
|
|
print(" 2. 📂 List All Projects")
|
|
print(" 3. ▶️ Start Live Session (Mock Mode)")
|
|
print(" 4. 📊 View Project Stats")
|
|
print(" 5. 🗄️ Archive Project")
|
|
print(" 6. 🧹 Reset Database (WARNING)")
|
|
print(" 0. 🚪 Exit")
|
|
print("-" * 40)
|
|
|
|
def create_project(self):
|
|
"""Create a new project."""
|
|
print("\n🎯 CREATE NEW PROJECT")
|
|
print("-" * 40)
|
|
|
|
name = input("Project name: ").strip()
|
|
if not name:
|
|
print("❌ Name required")
|
|
return
|
|
|
|
print("\nProject types:")
|
|
print(" 1. hunt (Hunting)")
|
|
print(" 2. mine (Mining)")
|
|
print(" 3. craft (Crafting)")
|
|
print(" 4. inventory (Inventory)")
|
|
|
|
type_choice = input("Select type (1-4): ").strip()
|
|
type_map = {'1': 'hunt', '2': 'mine', '3': 'craft', '4': 'inventory'}
|
|
project_type = type_map.get(type_choice, 'hunt')
|
|
|
|
project = self.pm.create_project(name, project_type)
|
|
print(f"\n✅ Created project: {project.name} (ID: {project.id}, Type: {project_type})")
|
|
|
|
# Show current projects
|
|
self.list_projects()
|
|
|
|
def list_projects(self):
|
|
"""List all projects."""
|
|
print("\n📂 PROJECTS")
|
|
print("-" * 60)
|
|
|
|
projects = self.pm.list_projects()
|
|
|
|
if not projects:
|
|
print(" No projects found. Create one first!")
|
|
return
|
|
|
|
print(f" {'ID':<5} {'Name':<20} {'Type':<10} {'Status':<10} {'Created'}")
|
|
print(" " + "-" * 58)
|
|
|
|
for p in projects:
|
|
created = p.created_at.strftime("%Y-%m-%d") if p.created_at else "N/A"
|
|
print(f" {p.id:<5} {p.name:<20} {p.type:<10} {p.status:<10} {created}")
|
|
|
|
print(f"\n Total: {len(projects)} project(s)")
|
|
|
|
async def start_live_session(self):
|
|
"""Start a live session with mock log watching."""
|
|
print("\n▶️ START LIVE SESSION (Mock Mode)")
|
|
print("-" * 60)
|
|
|
|
# Select project
|
|
projects = self.pm.list_projects(status='active')
|
|
if not projects:
|
|
print("❌ No active projects. Create one first (Option 1).")
|
|
return
|
|
|
|
print("Select project:")
|
|
for i, p in enumerate(projects, 1):
|
|
print(f" {i}. {p.name} ({p.type})")
|
|
|
|
choice = input(f"\nSelect (1-{len(projects)}): ").strip()
|
|
try:
|
|
project = projects[int(choice) - 1]
|
|
except (ValueError, IndexError):
|
|
print("❌ Invalid selection")
|
|
return
|
|
|
|
# Start session
|
|
session = self.pm.start_session(project.id, notes="User test session")
|
|
print(f"✅ Session started: ID {session.id}")
|
|
|
|
# Setup log watcher
|
|
test_data_dir = Path(__file__).parent / "test-data"
|
|
mock_log = test_data_dir / "mock-chat.log"
|
|
|
|
self.watcher = LogWatcher(str(mock_log), poll_interval=2.0, mock_mode=True)
|
|
|
|
# Stats tracking
|
|
stats = {'loot': 0, 'globals': 0, 'hofs': 0, 'skills': 0, 'total_ped': Decimal('0.0')}
|
|
|
|
def on_event(event):
|
|
"""Handle log events."""
|
|
if event.event_type == 'loot':
|
|
loot = LootEvent(
|
|
item_name=event.data.get('item_name', 'Unknown'),
|
|
quantity=event.data.get('quantity', 1),
|
|
value_ped=event.data.get('value_ped', Decimal('0.0')),
|
|
event_type='regular',
|
|
raw_log_line=event.raw_line
|
|
)
|
|
self.pm.record_loot(loot)
|
|
stats['loot'] += 1
|
|
stats['total_ped'] += loot.value_ped
|
|
print(f" 💰 Loot: {loot.item_name} x{loot.quantity} ({loot.value_ped} PED)")
|
|
|
|
elif event.event_type == 'global':
|
|
stats['globals'] += 1
|
|
print(f" 🌍 GLOBAL: {event.data.get('player_name')} found {event.data.get('value_ped')} PED!")
|
|
|
|
elif event.event_type == 'hof':
|
|
stats['hofs'] += 1
|
|
print(f" 🏆 HALL OF FAME: {event.data.get('value_ped')} PED!")
|
|
|
|
elif event.event_type == 'skill':
|
|
stats['skills'] += 1
|
|
print(f" 📈 Skill: {event.data.get('skill_name')} +{event.data.get('gained')}")
|
|
|
|
# Subscribe to events
|
|
self.watcher.subscribe('loot', on_event)
|
|
self.watcher.subscribe('global', on_event)
|
|
self.watcher.subscribe('hof', on_event)
|
|
self.watcher.subscribe('skill', on_event)
|
|
|
|
print("\n🔴 LIVE SESSION RUNNING")
|
|
print(" Watching mock chat.log for events...")
|
|
print(" Press Ctrl+C to stop\n")
|
|
|
|
await self.watcher.start()
|
|
|
|
try:
|
|
while self._running:
|
|
await asyncio.sleep(1)
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
await self.watcher.stop()
|
|
|
|
# End session
|
|
self.pm.end_session(session.id)
|
|
|
|
print("\n✅ Session ended")
|
|
print(f"\n📊 SESSION SUMMARY:")
|
|
print(f" Loot events: {stats['loot']}")
|
|
print(f" Globals: {stats['globals']}")
|
|
print(f" HoFs: {stats['hofs']}")
|
|
print(f" Skills: {stats['skills']}")
|
|
print(f" Total PED: {stats['total_ped']}")
|
|
|
|
def view_project_stats(self):
|
|
"""View statistics for a project."""
|
|
print("\n📊 PROJECT STATISTICS")
|
|
print("-" * 60)
|
|
|
|
projects = self.pm.list_projects()
|
|
if not projects:
|
|
print(" No projects found.")
|
|
return
|
|
|
|
print("Select project:")
|
|
for i, p in enumerate(projects, 1):
|
|
print(f" {i}. {p.name}")
|
|
|
|
choice = input(f"\nSelect (1-{len(projects)}): ").strip()
|
|
try:
|
|
project = projects[int(choice) - 1]
|
|
except (ValueError, IndexError):
|
|
print("❌ Invalid selection")
|
|
return
|
|
|
|
summary = self.pm.get_project_summary(project.id)
|
|
if not summary:
|
|
print("❌ Could not load summary")
|
|
return
|
|
|
|
print(f"\n📈 STATS FOR: {summary['name']}")
|
|
print("-" * 40)
|
|
print(f" Type: {summary['type']}")
|
|
print(f" Status: {summary['status']}")
|
|
print(f" Sessions: {summary['session_count']}")
|
|
print(f" Total Spent: {summary['total_spent']} PED")
|
|
print(f" Total Return: {summary['total_return']} PED")
|
|
print(f" Net Profit: {summary['net_profit']} PED")
|
|
print(f" Globals: {summary['global_count']}")
|
|
print(f" HoFs: {summary['hof_count']}")
|
|
|
|
if summary['total_spent'] > 0:
|
|
roi = (summary['net_profit'] / summary['total_spent']) * 100
|
|
print(f" ROI: {roi:.2f}%")
|
|
|
|
def archive_project(self):
|
|
"""Archive a project."""
|
|
print("\n🗄️ ARCHIVE PROJECT")
|
|
print("-" * 40)
|
|
|
|
projects = self.pm.list_projects()
|
|
if not projects:
|
|
print(" No projects found.")
|
|
return
|
|
|
|
print("Select project to archive:")
|
|
for i, p in enumerate(projects, 1):
|
|
print(f" {i}. {p.name} ({p.status})")
|
|
|
|
choice = input(f"\nSelect (1-{len(projects)}): ").strip()
|
|
try:
|
|
project = projects[int(choice) - 1]
|
|
except (ValueError, IndexError):
|
|
print("❌ Invalid selection")
|
|
return
|
|
|
|
confirm = input(f"Archive '{project.name}'? (yes/no): ").strip().lower()
|
|
if confirm == 'yes':
|
|
self.pm.archive_project(project.id)
|
|
print(f"✅ Archived: {project.name}")
|
|
else:
|
|
print("Cancelled")
|
|
|
|
def reset_database(self):
|
|
"""Reset database (for testing)."""
|
|
print("\n🧹 RESET DATABASE")
|
|
print("-" * 40)
|
|
print("⚠️ WARNING: This will delete all data!")
|
|
|
|
confirm = input("Type 'RESET' to confirm: ").strip()
|
|
if confirm == 'RESET':
|
|
db_path = self.db.db_path
|
|
self.db.close()
|
|
|
|
import os
|
|
if db_path.exists():
|
|
os.remove(db_path)
|
|
print(f"✅ Database deleted: {db_path}")
|
|
|
|
# Reinitialize
|
|
self.db = DatabaseManager()
|
|
self.db.initialize()
|
|
self.pm = ProjectManager(self.db)
|
|
print("✅ Database reinitialized")
|
|
else:
|
|
print("Cancelled")
|
|
|
|
async def run(self):
|
|
"""Main application loop."""
|
|
self._running = True
|
|
|
|
# Setup signal handlers
|
|
def signal_handler():
|
|
self._running = False
|
|
print("\n\n🛑 Shutting down...")
|
|
|
|
signal.signal(signal.SIGINT, lambda s, f: signal_handler())
|
|
|
|
self.print_header()
|
|
|
|
while self._running:
|
|
self.print_menu()
|
|
choice = input("Select option: ").strip()
|
|
|
|
try:
|
|
if choice == '1':
|
|
self.create_project()
|
|
elif choice == '2':
|
|
self.list_projects()
|
|
elif choice == '3':
|
|
await self.start_live_session()
|
|
elif choice == '4':
|
|
self.view_project_stats()
|
|
elif choice == '5':
|
|
self.archive_project()
|
|
elif choice == '6':
|
|
self.reset_database()
|
|
elif choice == '0':
|
|
self._running = False
|
|
print("\n🍋 Thank you for testing Lemontropia Suite!")
|
|
else:
|
|
print("❌ Invalid option")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
print(f"❌ Error: {e}")
|
|
|
|
# Cleanup
|
|
self.db.close()
|
|
|
|
|
|
def main():
|
|
"""Application entry point."""
|
|
print("""
|
|
🍋 ==========================================
|
|
LEMONTROPIA SUITE — User Test Build
|
|
Data Capture Engine v0.1.0
|
|
==========================================
|
|
""")
|
|
|
|
app = LemontropiaApp()
|
|
|
|
try:
|
|
asyncio.run(app.run())
|
|
except KeyboardInterrupt:
|
|
print("\n\n🛑 Interrupted by user")
|
|
finally:
|
|
print("\n👋 Goodbye!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|