97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
"""Test script to verify data flow and formatting."""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add src to path
|
|
sys.path.insert(0, str(Path(__file__).parent / "src"))
|
|
|
|
from jellycleanarr.config import settings
|
|
from jellycleanarr.api.client import JellyfinClient
|
|
from jellycleanarr.services.stats_service import StatsService
|
|
from jellycleanarr.utils.formatters import format_duration, format_number
|
|
|
|
|
|
async def test_data_flow():
|
|
"""Test the complete data flow from API to formatted output."""
|
|
print("=" * 80)
|
|
print("Testing Jellycleanarr Data Flow")
|
|
print("=" * 80)
|
|
|
|
# Initialize client
|
|
print(f"\n1. Connecting to: {settings.jellyfin_address}")
|
|
client = JellyfinClient(settings.jellyfin_address, settings.jellyfin_api_key)
|
|
|
|
try:
|
|
# Test connection
|
|
await client.test_connection()
|
|
print("[OK] Connection successful")
|
|
|
|
# Initialize stats service
|
|
stats_service = StatsService(client)
|
|
|
|
# Get user stats
|
|
print("\n2. Fetching user stats...")
|
|
most_played, least_played = await stats_service.get_user_stats()
|
|
print(f"[OK] Got {len(most_played)} most active users")
|
|
print(f"[OK] Got {len(least_played)} least active users")
|
|
|
|
# Display raw data for first user
|
|
if most_played:
|
|
print("\n3. Raw data for most active user:")
|
|
user = most_played[0]
|
|
print(f" Username: {user.username}")
|
|
print(f" Total plays: {user.total_plays}")
|
|
print(f" Total duration (raw): {user.total_duration_seconds}")
|
|
print(f" Unique items: {user.unique_items_played}")
|
|
|
|
# Test formatting
|
|
print("\n4. Testing formatters:")
|
|
try:
|
|
formatted_duration = format_duration(user.total_duration_seconds)
|
|
print(f" Formatted duration: {formatted_duration}")
|
|
except Exception as e:
|
|
print(f" ERROR formatting duration: {e}")
|
|
|
|
formatted_plays = format_number(user.total_plays)
|
|
print(f" Formatted plays: {formatted_plays}")
|
|
formatted_items = format_number(user.unique_items_played)
|
|
print(f" Formatted items: {formatted_items}")
|
|
|
|
# Create formatted row (as TUI would)
|
|
print("\n5. Creating table row (as TUI would):")
|
|
row = [
|
|
user.username,
|
|
format_number(user.total_plays),
|
|
format_duration(user.total_duration_seconds),
|
|
format_number(user.unique_items_played),
|
|
]
|
|
print(f" Row: {row}")
|
|
|
|
# Convert to strings (as StatsTable does)
|
|
string_row = [str(cell) for cell in row]
|
|
print(f" String row: {string_row}")
|
|
|
|
# Display all users
|
|
print("\n6. All formatted users:")
|
|
print("-" * 80)
|
|
print(f"{'User':<20} {'Plays':>12} {'Duration':>15} {'Items':>13}")
|
|
print("-" * 80)
|
|
for user in most_played:
|
|
print(f"{user.username:<20} {format_number(user.total_plays):>12} "
|
|
f"{format_duration(user.total_duration_seconds):>15} "
|
|
f"{format_number(user.unique_items_played):>13}")
|
|
|
|
except Exception as e:
|
|
print(f"\nERROR: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
await client.close()
|
|
print("\n[OK] Connection closed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_data_flow())
|