Cleanup repo + harden offline install and watchdog
Repo hygiene: - Remove dead scripts (.wait-for-display, .display-keepalive, .keep-screen-alive) - Remove stale backup/scratch (install.sh.bak, New.txt, stale .pyc) - Stop tracking machine-specific/generated state (config/app_config.json, playlists/server_playlist.json, src/player_auth.json, generated launchers) - Prune duplicate wheel versions (one per package) - Reorganize working_files -> dev/, dev/tests/, documentation/dev-notes/, scripts/ - Extend .gitignore; fix no-op rules in repo/.gitignore Offline install (install.sh): - Non-fatal system-dep step; per-package apt retry; .deb bundle install - Skip pip self-upgrade offline; guard online fallbacks behind connectivity probe - Dependency-free fetch_url (curl/wget/urllib); --no-reboot flag - Fix corrupted literal backslash-bang escapes (shebangs, test -f, banner) - Idempotent cron install (no more stacked @reboot lines) Autostart / watchdog: - start.sh: single-instance flock lock (.player_watchdog.lock) acquired only after a display is confirmed, preventing parallel watchdogs - Set XDG_RUNTIME_DIR when missing; exit early if headless - run_player.sh: activate .venv; stop_player.sh: clear lock file
This commit is contained in:
SKE087
Executable
+185
@@ -0,0 +1,185 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script for Kiwy-Signage authentication with DigiServer v2
|
||||
Run this to verify authentication is working before updating main.py
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add src directory to path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
|
||||
|
||||
from player_auth import PlayerAuth
|
||||
import json
|
||||
|
||||
def load_app_config():
|
||||
"""Load existing app_config.json"""
|
||||
# Try multiple possible locations
|
||||
possible_paths = [
|
||||
'config/app_config.json',
|
||||
'resources/app_config.txt',
|
||||
'src/config/app_config.json',
|
||||
'../config/app_config.json'
|
||||
]
|
||||
|
||||
for config_file in possible_paths:
|
||||
if os.path.exists(config_file):
|
||||
print(f" Found config: {config_file}")
|
||||
with open(config_file, 'r') as f:
|
||||
return json.load(f)
|
||||
|
||||
print(f"❌ Config file not found! Tried:")
|
||||
for path in possible_paths:
|
||||
print(f" - {path}")
|
||||
return None
|
||||
|
||||
def test_authentication():
|
||||
"""Test authentication with DigiServer v2"""
|
||||
print("=" * 60)
|
||||
print("Kiwy-Signage Authentication Test")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
# Load config
|
||||
print("📁 Loading configuration...")
|
||||
config = load_app_config()
|
||||
if not config:
|
||||
return False
|
||||
|
||||
server_ip = config.get('server_ip', '')
|
||||
hostname = config.get('screen_name', '')
|
||||
quickconnect = config.get('quickconnect_key', '')
|
||||
port = config.get('port', '')
|
||||
|
||||
print(f" Server: {server_ip}:{port}")
|
||||
print(f" Hostname: {hostname}")
|
||||
print(f" Quick Connect: {'*' * len(quickconnect)}")
|
||||
print()
|
||||
|
||||
# Build server URL
|
||||
import re
|
||||
ip_pattern = r'^\d+\.\d+\.\d+\.\d+$'
|
||||
if re.match(ip_pattern, server_ip):
|
||||
server_url = f'http://{server_ip}:{port}'
|
||||
else:
|
||||
server_url = f'http://{server_ip}'
|
||||
|
||||
print(f"🌐 Server URL: {server_url}")
|
||||
print()
|
||||
|
||||
# Test server connection
|
||||
print("🔌 Testing server connection...")
|
||||
try:
|
||||
import requests
|
||||
response = requests.get(f"{server_url}/api/health", timeout=5)
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
print(f" ✅ Server is healthy (version: {data.get('version')})")
|
||||
else:
|
||||
print(f" ⚠️ Server responded with status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f" ❌ Cannot connect to server: {e}")
|
||||
print()
|
||||
print("💡 Make sure DigiServer v2 is running and accessible!")
|
||||
return False
|
||||
print()
|
||||
|
||||
# Initialize auth
|
||||
print("🔐 Initializing authentication...")
|
||||
auth = PlayerAuth(config_file='src/player_auth.json')
|
||||
|
||||
# Check if already authenticated
|
||||
if auth.is_authenticated():
|
||||
print(f" ℹ️ Found existing authentication")
|
||||
print(f" Player: {auth.get_player_name()}")
|
||||
print()
|
||||
|
||||
print("✓ Verifying saved authentication...")
|
||||
valid, info = auth.verify_auth()
|
||||
|
||||
if valid:
|
||||
print(f" ✅ Authentication is valid!")
|
||||
print(f" Player ID: {info['player_id']}")
|
||||
print(f" Player Name: {info['player_name']}")
|
||||
print(f" Group ID: {info.get('group_id', 'None')}")
|
||||
print(f" Orientation: {info.get('orientation', 'Landscape')}")
|
||||
print()
|
||||
|
||||
# Test playlist fetch
|
||||
print("📋 Testing playlist fetch...")
|
||||
playlist_data = auth.get_playlist()
|
||||
if playlist_data:
|
||||
version = playlist_data.get('playlist_version', 0)
|
||||
content_count = len(playlist_data.get('playlist', []))
|
||||
print(f" ✅ Playlist received!")
|
||||
print(f" Version: {version}")
|
||||
print(f" Content items: {content_count}")
|
||||
else:
|
||||
print(f" ⚠️ Could not fetch playlist")
|
||||
print()
|
||||
|
||||
# Test heartbeat
|
||||
print("💓 Testing heartbeat...")
|
||||
if auth.send_heartbeat(status='online'):
|
||||
print(f" ✅ Heartbeat sent successfully")
|
||||
else:
|
||||
print(f" ⚠️ Heartbeat failed")
|
||||
print()
|
||||
|
||||
print("=" * 60)
|
||||
print("✅ All tests passed! Player is ready to use.")
|
||||
print("=" * 60)
|
||||
return True
|
||||
else:
|
||||
print(f" ❌ Saved authentication is expired or invalid")
|
||||
print(f" Re-authenticating...")
|
||||
print()
|
||||
|
||||
# Need to authenticate
|
||||
print("🔑 Authenticating with server...")
|
||||
success, error = auth.authenticate(
|
||||
server_url=server_url,
|
||||
hostname=hostname,
|
||||
quickconnect_code=quickconnect
|
||||
)
|
||||
|
||||
if success:
|
||||
print(f" ✅ Authentication successful!")
|
||||
print(f" Player: {auth.get_player_name()}")
|
||||
print(f" Player ID: {auth.get_player_id()}")
|
||||
print()
|
||||
|
||||
# Save confirmation
|
||||
print(f"💾 Authentication saved to: src/player_auth.json")
|
||||
print()
|
||||
|
||||
print("=" * 60)
|
||||
print("✅ Authentication successful! Player is ready to use.")
|
||||
print("=" * 60)
|
||||
return True
|
||||
else:
|
||||
print(f" ❌ Authentication failed: {error}")
|
||||
print()
|
||||
print("💡 Troubleshooting:")
|
||||
print(" 1. Check player exists in DigiServer v2 (hostname must match)")
|
||||
print(" 2. Verify quickconnect_key matches server configuration")
|
||||
print(" 3. Check server logs for authentication attempts")
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("❌ Authentication test failed")
|
||||
print("=" * 60)
|
||||
return False
|
||||
|
||||
if __name__ == '__main__':
|
||||
try:
|
||||
success = test_authentication()
|
||||
sys.exit(0 if success else 1)
|
||||
except KeyboardInterrupt:
|
||||
print("\n⚠️ Test cancelled by user")
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"\n❌ Unexpected error: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,162 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script for USB card reader functionality
|
||||
"""
|
||||
|
||||
import evdev
|
||||
from evdev import InputDevice, categorize, ecodes
|
||||
import time
|
||||
|
||||
def list_input_devices():
|
||||
"""List all available input devices"""
|
||||
print("\n=== Available Input Devices ===")
|
||||
devices = [evdev.InputDevice(path) for path in evdev.list_devices()]
|
||||
|
||||
# Exclusion keywords that help identify non-card-reader devices
|
||||
exclusion_keywords = [
|
||||
'touch', 'touchscreen', 'mouse', 'mice', 'trackpad',
|
||||
'touchpad', 'pen', 'stylus', 'video', 'button', 'lid'
|
||||
]
|
||||
|
||||
for i, device in enumerate(devices):
|
||||
device_name_lower = device.name.lower()
|
||||
is_excluded = any(keyword in device_name_lower for keyword in exclusion_keywords)
|
||||
is_likely_card = 'card' in device_name_lower or 'reader' in device_name_lower or 'rfid' in device_name_lower
|
||||
|
||||
print(f"\n[{i}] {device.path}")
|
||||
print(f" Name: {device.name}")
|
||||
print(f" Phys: {device.phys}")
|
||||
|
||||
capabilities = device.capabilities()
|
||||
if ecodes.EV_KEY in capabilities:
|
||||
print(f" Type: Keyboard/HID Input Device")
|
||||
|
||||
# Add helpful hints
|
||||
if is_likely_card:
|
||||
print(f" ** LIKELY CARD READER **")
|
||||
elif is_excluded:
|
||||
print(f" (Excluded: appears to be touch/mouse/other non-card device)")
|
||||
elif 'usb' in device_name_lower and 'keyboard' in device_name_lower:
|
||||
print(f" (USB Keyboard - could be card reader)")
|
||||
|
||||
return devices
|
||||
|
||||
def test_card_reader(device_index=None):
|
||||
"""Test reading from a card reader device"""
|
||||
devices = [evdev.InputDevice(path) for path in evdev.list_devices()]
|
||||
|
||||
# Exclusion keywords (same as in main app)
|
||||
exclusion_keywords = [
|
||||
'touch', 'touchscreen', 'mouse', 'mice', 'trackpad',
|
||||
'touchpad', 'pen', 'stylus', 'video', 'button', 'lid'
|
||||
]
|
||||
|
||||
if device_index is not None:
|
||||
if device_index >= len(devices):
|
||||
print(f"Error: Device index {device_index} out of range")
|
||||
return
|
||||
device = devices[device_index]
|
||||
else:
|
||||
# Try to find a card reader automatically using same logic as main app
|
||||
device = None
|
||||
|
||||
# Priority 1: Explicit card readers
|
||||
for dev in devices:
|
||||
device_name_lower = dev.name.lower()
|
||||
if any(keyword in device_name_lower for keyword in exclusion_keywords):
|
||||
continue
|
||||
if 'card' in device_name_lower or 'reader' in device_name_lower or 'rfid' in device_name_lower or 'hid' in device_name_lower:
|
||||
capabilities = dev.capabilities()
|
||||
if ecodes.EV_KEY in capabilities:
|
||||
device = dev
|
||||
print(f"Found card reader: {dev.name}")
|
||||
break
|
||||
|
||||
# Priority 2: USB keyboards
|
||||
if not device:
|
||||
for dev in devices:
|
||||
device_name_lower = dev.name.lower()
|
||||
if any(keyword in device_name_lower for keyword in exclusion_keywords):
|
||||
continue
|
||||
if 'usb' in device_name_lower and 'keyboard' in device_name_lower:
|
||||
capabilities = dev.capabilities()
|
||||
if ecodes.EV_KEY in capabilities:
|
||||
device = dev
|
||||
print(f"Using USB keyboard as card reader: {dev.name}")
|
||||
break
|
||||
|
||||
# Priority 3: Any non-excluded keyboard
|
||||
if not device:
|
||||
for dev in devices:
|
||||
device_name_lower = dev.name.lower()
|
||||
if any(keyword in device_name_lower for keyword in exclusion_keywords):
|
||||
continue
|
||||
capabilities = dev.capabilities()
|
||||
if ecodes.EV_KEY in capabilities:
|
||||
device = dev
|
||||
print(f"Using keyboard device as card reader: {dev.name}")
|
||||
break
|
||||
|
||||
if not device:
|
||||
print("No suitable input device found!")
|
||||
return
|
||||
|
||||
print(f"\n=== Testing Card Reader ===")
|
||||
print(f"Device: {device.name}")
|
||||
print(f"Path: {device.path}")
|
||||
print("\nSwipe your card now (press Ctrl+C to exit)...\n")
|
||||
|
||||
card_data = ""
|
||||
|
||||
try:
|
||||
for event in device.read_loop():
|
||||
if event.type == ecodes.EV_KEY:
|
||||
key_event = categorize(event)
|
||||
|
||||
if key_event.keystate == 1: # Key down
|
||||
key_code = key_event.keycode
|
||||
|
||||
# Handle Enter key (card read complete)
|
||||
if key_code == 'KEY_ENTER':
|
||||
print(f"\n✓ Card data received: '{card_data}'")
|
||||
print(f" Length: {len(card_data)} characters")
|
||||
print(f" Processed ID: card_{card_data.strip().upper()}")
|
||||
print("\nReady for next card swipe...")
|
||||
card_data = ""
|
||||
|
||||
# Build card data string
|
||||
elif key_code.startswith('KEY_'):
|
||||
char = key_code.replace('KEY_', '')
|
||||
if len(char) == 1: # Single character
|
||||
card_data += char
|
||||
print(f"Reading: {card_data}", end='\r', flush=True)
|
||||
elif char.isdigit(): # Handle numeric keys
|
||||
card_data += char
|
||||
print(f"Reading: {card_data}", end='\r', flush=True)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n\nTest stopped by user")
|
||||
except Exception as e:
|
||||
print(f"\nError: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("USB Card Reader Test Tool")
|
||||
print("=" * 50)
|
||||
|
||||
devices = list_input_devices()
|
||||
|
||||
if not devices:
|
||||
print("\nNo input devices found!")
|
||||
exit(1)
|
||||
|
||||
print("\n" + "=" * 50)
|
||||
choice = input("\nEnter device number to test (or press Enter for auto-detect): ").strip()
|
||||
|
||||
if choice:
|
||||
try:
|
||||
device_index = int(choice)
|
||||
test_card_reader(device_index)
|
||||
except ValueError:
|
||||
print("Invalid device number!")
|
||||
else:
|
||||
test_card_reader()
|
||||
@@ -0,0 +1,138 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test server connection and playlist fetch."""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add src directory to path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
|
||||
|
||||
from player_auth import PlayerAuth
|
||||
|
||||
# Load config
|
||||
config_file = 'config/app_config.json'
|
||||
with open(config_file, 'r') as f:
|
||||
config = json.load(f)
|
||||
|
||||
print("=" * 80)
|
||||
print("SERVER CONNECTION TEST")
|
||||
print("=" * 80)
|
||||
|
||||
server_ip = config.get("server_ip", "")
|
||||
screen_name = config.get("screen_name", "")
|
||||
quickconnect_key = config.get("quickconnect_key", "")
|
||||
port = config.get("port", "")
|
||||
|
||||
print(f"\nConfiguration:")
|
||||
print(f" Server: {server_ip}")
|
||||
print(f" Port: {port}")
|
||||
print(f" Screen Name: {screen_name}")
|
||||
print(f" QuickConnect: {quickconnect_key}")
|
||||
|
||||
# Build server URL
|
||||
if server_ip.startswith('http://') or server_ip.startswith('https://'):
|
||||
server_url = server_ip
|
||||
# If it has https but port 443 is specified, ensure port is included if non-standard
|
||||
if not ':' in server_ip.replace('https://', '').replace('http://', ''):
|
||||
if port and port != '443' and port != '80':
|
||||
server_url = f"{server_ip}:{port}"
|
||||
else:
|
||||
# Use https for port 443, http for others
|
||||
protocol = "https" if port == "443" else "http"
|
||||
server_url = f"{protocol}://{server_ip}:{port}"
|
||||
|
||||
print(f"\nServer URL: {server_url}")
|
||||
|
||||
# Test authentication
|
||||
print("\n" + "=" * 80)
|
||||
print("1. TESTING AUTHENTICATION")
|
||||
print("=" * 80)
|
||||
|
||||
auth = PlayerAuth('src/player_auth.json')
|
||||
|
||||
# Check if already authenticated
|
||||
if auth.is_authenticated():
|
||||
print("✓ Found existing authentication")
|
||||
valid, message = auth.verify_auth()
|
||||
if valid:
|
||||
print(f"✓ Auth is valid: {message}")
|
||||
else:
|
||||
print(f"✗ Auth expired: {message}")
|
||||
print("\nRe-authenticating...")
|
||||
success, error = auth.authenticate(
|
||||
server_url=server_url,
|
||||
hostname=screen_name,
|
||||
quickconnect_code=quickconnect_key
|
||||
)
|
||||
if success:
|
||||
print(f"✓ Re-authentication successful!")
|
||||
else:
|
||||
print(f"✗ Re-authentication failed: {error}")
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("No existing authentication found. Authenticating...")
|
||||
success, error = auth.authenticate(
|
||||
server_url=server_url,
|
||||
hostname=screen_name,
|
||||
quickconnect_code=quickconnect_key
|
||||
)
|
||||
if success:
|
||||
print(f"✓ Authentication successful!")
|
||||
else:
|
||||
print(f"✗ Authentication failed: {error}")
|
||||
sys.exit(1)
|
||||
|
||||
# Test playlist fetch
|
||||
print("\n" + "=" * 80)
|
||||
print("2. TESTING PLAYLIST FETCH")
|
||||
print("=" * 80)
|
||||
|
||||
playlist_data = auth.get_playlist()
|
||||
|
||||
if playlist_data:
|
||||
print(f"✓ Playlist fetched successfully!")
|
||||
print(f"\nPlaylist Version: {playlist_data.get('playlist_version', 'N/A')}")
|
||||
print(f"Number of items: {len(playlist_data.get('playlist', []))}")
|
||||
|
||||
print("\n" + "-" * 80)
|
||||
print("PLAYLIST ITEMS:")
|
||||
print("-" * 80)
|
||||
|
||||
for idx, item in enumerate(playlist_data.get('playlist', []), 1):
|
||||
print(f"\n{idx}. File: {item.get('file_name', 'N/A')}")
|
||||
print(f" URL: {item.get('url', 'N/A')}")
|
||||
print(f" Duration: {item.get('duration', 'N/A')}s")
|
||||
|
||||
# Check if URL is relative or absolute
|
||||
url = item.get('url', '')
|
||||
if url.startswith('http://') or url.startswith('https://'):
|
||||
print(f" Type: Absolute URL")
|
||||
else:
|
||||
print(f" Type: Relative path (will fail to download!)")
|
||||
|
||||
# Save full response
|
||||
with open('server_response_debug.json', 'w') as f:
|
||||
json.dump(playlist_data, f, indent=2)
|
||||
print(f"\n✓ Full response saved to: server_response_debug.json")
|
||||
|
||||
print("\n" + "=" * 80)
|
||||
print("SUMMARY")
|
||||
print("=" * 80)
|
||||
print(f"Server has: {len(playlist_data.get('playlist', []))} files")
|
||||
print(f"Local has: 3 files (from playlists/server_playlist_v8.json)")
|
||||
|
||||
if len(playlist_data.get('playlist', [])) > 3:
|
||||
print(f"\n⚠️ PROBLEM: Server has {len(playlist_data.get('playlist', []))} files but only 3 were saved!")
|
||||
print("\nMissing files are likely:")
|
||||
local_files = ['music.jpg', '130414-746934884.mp4', 'IMG_0386.jpeg']
|
||||
server_files = [item.get('file_name', '') for item in playlist_data.get('playlist', [])]
|
||||
missing = [f for f in server_files if f not in local_files]
|
||||
for f in missing:
|
||||
print(f" - {f}")
|
||||
|
||||
else:
|
||||
print("✗ Failed to fetch playlist")
|
||||
sys.exit(1)
|
||||
|
||||
print("\n" + "=" * 80)
|
||||
@@ -0,0 +1,55 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Direct API test to check server playlist."""
|
||||
|
||||
import requests
|
||||
import json
|
||||
|
||||
# Try with the saved auth
|
||||
auth_file = 'src/player_auth.json'
|
||||
with open(auth_file, 'r') as f:
|
||||
auth_data = json.load(f)
|
||||
|
||||
server_url = auth_data['server_url']
|
||||
auth_code = auth_data['auth_code']
|
||||
|
||||
print("=" * 80)
|
||||
print("DIRECT API TEST")
|
||||
print("=" * 80)
|
||||
print(f"Server: {server_url}")
|
||||
print(f"Auth code: {auth_code[:20]}...")
|
||||
print()
|
||||
|
||||
# Try to get playlist
|
||||
try:
|
||||
url = f"{server_url}/api/player/playlist"
|
||||
headers = {
|
||||
'Authorization': f'Bearer {auth_code}'
|
||||
}
|
||||
|
||||
print(f"Fetching: {url}")
|
||||
response = requests.get(url, headers=headers, timeout=10)
|
||||
|
||||
print(f"Status: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
print(f"\nPlaylist version: {data.get('playlist_version', 'N/A')}")
|
||||
print(f"Number of items: {len(data.get('playlist', []))}")
|
||||
|
||||
print("\nPlaylist items:")
|
||||
for idx, item in enumerate(data.get('playlist', []), 1):
|
||||
print(f"\n {idx}. {item.get('file_name', 'N/A')}")
|
||||
print(f" URL: {item.get('url', 'N/A')}")
|
||||
print(f" Duration: {item.get('duration', 'N/A')}s")
|
||||
|
||||
# Save full response
|
||||
with open('server_playlist_full.json', 'w') as f:
|
||||
json.dump(data, f, indent=2)
|
||||
print(f"\nFull response saved to: server_playlist_full.json")
|
||||
else:
|
||||
print(f"Error: {response.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
|
||||
print("\n" + "=" * 80)
|
||||
@@ -0,0 +1,165 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script to diagnose edited media upload issues
|
||||
Run this to test if the server endpoint exists and works correctly
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import requests
|
||||
from pathlib import Path
|
||||
|
||||
# Add src to path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
|
||||
|
||||
def test_upload():
|
||||
"""Test the edited media upload functionality"""
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("EDITED MEDIA UPLOAD DIAGNOSTICS")
|
||||
print("="*60)
|
||||
|
||||
# Get authentication
|
||||
try:
|
||||
from get_playlists_v2 import get_auth_instance
|
||||
auth = get_auth_instance()
|
||||
|
||||
if not auth or not auth.is_authenticated():
|
||||
print("❌ ERROR: Not authenticated!")
|
||||
print(" Please ensure player_auth.json exists and is valid")
|
||||
return False
|
||||
|
||||
server_url = auth.auth_data.get('server_url')
|
||||
auth_code = auth.auth_data.get('auth_code')
|
||||
|
||||
print(f"\n✓ Authentication successful")
|
||||
print(f" Server URL: {server_url}")
|
||||
print(f" Auth Code: {auth_code[:20]}..." if auth_code else " Auth Code: None")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Authentication error: {e}")
|
||||
return False
|
||||
|
||||
# Check for edited media files
|
||||
edited_media_dir = os.path.join(os.path.dirname(__file__), 'media', 'edited_media')
|
||||
edited_files = list(Path(edited_media_dir).glob('*_e_v*.jpg'))
|
||||
metadata_files = list(Path(edited_media_dir).glob('*_metadata.json'))
|
||||
|
||||
print(f"\n✓ Edited Media Directory: {edited_media_dir}")
|
||||
print(f" Edited images: {len(edited_files)}")
|
||||
print(f" Metadata files: {len(metadata_files)}")
|
||||
|
||||
if not edited_files:
|
||||
print("\n⚠️ No edited images found!")
|
||||
print(" Create an edit first, then run this test")
|
||||
return False
|
||||
|
||||
# Test with the first edited image
|
||||
image_path = str(edited_files[0])
|
||||
metadata_file = str(edited_files[0]).replace('.jpg', '_metadata.json')
|
||||
|
||||
if not os.path.exists(metadata_file):
|
||||
print(f"\n❌ Metadata file not found: {metadata_file}")
|
||||
return False
|
||||
|
||||
print(f"\nTesting upload with:")
|
||||
print(f" Image: {os.path.basename(image_path)}")
|
||||
print(f" Size: {os.path.getsize(image_path):,} bytes")
|
||||
print(f" Metadata: {os.path.basename(metadata_file)}")
|
||||
|
||||
# Load and display metadata
|
||||
with open(metadata_file, 'r') as f:
|
||||
metadata = json.load(f)
|
||||
|
||||
print(f"\nMetadata content:")
|
||||
for key, value in metadata.items():
|
||||
print(f" {key}: {value}")
|
||||
|
||||
# Add original_filename if not present
|
||||
metadata['original_filename'] = os.path.basename(metadata['original_path'])
|
||||
|
||||
# Prepare upload request
|
||||
upload_url = f"{server_url}/api/player-edit-media"
|
||||
headers = {'Authorization': f'Bearer {auth_code}'}
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print("TESTING UPLOAD...")
|
||||
print(f"{'='*60}")
|
||||
print(f"Endpoint: {upload_url}")
|
||||
print(f"Headers: Authorization: Bearer {auth_code[:20]}...")
|
||||
|
||||
try:
|
||||
with open(image_path, 'rb') as img_file:
|
||||
files = {
|
||||
'image_file': (metadata['original_filename'], img_file, 'image/jpeg')
|
||||
}
|
||||
data = {
|
||||
'metadata': json.dumps(metadata),
|
||||
'original_file': metadata['original_filename']
|
||||
}
|
||||
|
||||
print(f"\nSending request (30s timeout, SSL verify=False)...")
|
||||
response = requests.post(
|
||||
upload_url,
|
||||
headers=headers,
|
||||
files=files,
|
||||
data=data,
|
||||
timeout=30,
|
||||
verify=False
|
||||
)
|
||||
|
||||
print(f"\n✓ Response received!")
|
||||
print(f" Status Code: {response.status_code}")
|
||||
print(f" Headers: {dict(response.headers)}")
|
||||
|
||||
if response.status_code == 200:
|
||||
print(f"\n✅ SUCCESS! Server accepted the upload")
|
||||
print(f" Response: {response.json()}")
|
||||
return True
|
||||
elif response.status_code == 404:
|
||||
print(f"\n❌ ENDPOINT NOT FOUND (404)")
|
||||
print(f" The server does NOT have /api/player-edit-media endpoint")
|
||||
print(f" Server may need to implement this feature")
|
||||
elif response.status_code == 401:
|
||||
print(f"\n❌ AUTHENTICATION FAILED (401)")
|
||||
print(f" Check your auth_code in player_auth.json")
|
||||
else:
|
||||
print(f"\n❌ REQUEST FAILED (Status: {response.status_code})")
|
||||
print(f" Response: {response.text}")
|
||||
|
||||
return False
|
||||
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
print(f"\n❌ CONNECTION ERROR")
|
||||
print(f" Cannot reach server at {server_url}")
|
||||
print(f" Error: {e}")
|
||||
return False
|
||||
except requests.exceptions.Timeout as e:
|
||||
print(f"\n❌ TIMEOUT")
|
||||
print(f" Server did not respond within 30 seconds")
|
||||
print(f" Error: {e}")
|
||||
return False
|
||||
except requests.exceptions.SSLError as e:
|
||||
print(f"\n❌ SSL ERROR")
|
||||
print(f" Error: {e}")
|
||||
print(f" Tip: Try adding verify=False to requests")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"\n❌ UNEXPECTED ERROR")
|
||||
print(f" Error: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if __name__ == '__main__':
|
||||
success = test_upload()
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
if success:
|
||||
print("✅ UPLOAD TEST PASSED - Server accepts edited media!")
|
||||
else:
|
||||
print("❌ UPLOAD TEST FAILED - See details above")
|
||||
print(f"{'='*60}\n")
|
||||
|
||||
sys.exit(0 if success else 1)
|
||||
@@ -0,0 +1,82 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script to verify the enhanced logging without running the full GUI
|
||||
"""
|
||||
import os
|
||||
import json
|
||||
|
||||
def simulate_playback_check():
|
||||
"""Simulate the playback logic to see what would happen"""
|
||||
|
||||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
media_dir = os.path.join(base_dir, 'media')
|
||||
playlists_dir = os.path.join(base_dir, 'playlists')
|
||||
|
||||
# Supported extensions
|
||||
VIDEO_EXTENSIONS = ['.mp4', '.avi', '.mkv', '.mov', '.webm']
|
||||
IMAGE_EXTENSIONS = ['.jpg', '.jpeg', '.png', '.bmp', '.gif']
|
||||
|
||||
# Load playlist
|
||||
playlist_file = os.path.join(playlists_dir, 'server_playlist_v8.json')
|
||||
with open(playlist_file, 'r') as f:
|
||||
data = json.load(f)
|
||||
|
||||
playlist = data.get('playlist', [])
|
||||
|
||||
print("=" * 80)
|
||||
print("SIMULATING PLAYBACK SEQUENCE")
|
||||
print("=" * 80)
|
||||
|
||||
for idx, media_item in enumerate(playlist):
|
||||
file_name = media_item.get('file_name', '')
|
||||
duration = media_item.get('duration', 10)
|
||||
|
||||
print(f"\n[STEP {idx + 1}] ===== Playing item {idx + 1}/{len(playlist)} =====")
|
||||
print(f" File: {file_name}")
|
||||
print(f" Duration: {duration}s")
|
||||
|
||||
# Construct path
|
||||
media_path = os.path.join(media_dir, file_name)
|
||||
print(f" Full path: {media_path}")
|
||||
|
||||
# Check existence
|
||||
if not os.path.exists(media_path):
|
||||
print(f" \u274c Media file not found: {media_path}")
|
||||
print(f" ACTION: Skipping to next media...")
|
||||
continue
|
||||
|
||||
file_size = os.path.getsize(media_path)
|
||||
print(f" \u2713 File exists (size: {file_size:,} bytes)")
|
||||
|
||||
# Check extension
|
||||
file_extension = os.path.splitext(file_name)[1].lower()
|
||||
print(f" Extension: {file_extension}")
|
||||
|
||||
if file_extension in VIDEO_EXTENSIONS:
|
||||
print(f" Media type: VIDEO")
|
||||
print(f" ACTION: play_video('{media_path}', {duration})")
|
||||
print(f" - Creating Video widget...")
|
||||
print(f" - Adding to content area...")
|
||||
print(f" - Scheduling next media in {duration}s")
|
||||
print(f" \u2713 Media started successfully")
|
||||
elif file_extension in IMAGE_EXTENSIONS:
|
||||
print(f" Media type: IMAGE")
|
||||
print(f" ACTION: play_image('{media_path}', {duration})")
|
||||
print(f" - Creating AsyncImage widget...")
|
||||
print(f" - Adding to content area...")
|
||||
print(f" - Scheduling next media in {duration}s")
|
||||
print(f" \u2713 Image displayed successfully")
|
||||
else:
|
||||
print(f" \u274c Unsupported media type: {file_extension}")
|
||||
print(f" Supported: .mp4/.avi/.mkv/.mov/.webm/.jpg/.jpeg/.png/.bmp/.gif")
|
||||
print(f" ACTION: Skipping to next media...")
|
||||
continue
|
||||
|
||||
print(f"\n [After {duration}s] Transitioning to next media (was index {idx})")
|
||||
|
||||
print("\n" + "=" * 80)
|
||||
print("END OF PLAYLIST - Would restart from beginning")
|
||||
print("=" * 80)
|
||||
|
||||
if __name__ == '__main__':
|
||||
simulate_playback_check()
|
||||
@@ -0,0 +1,45 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test script to check what playlist the server is actually returning."""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add src directory to path
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
|
||||
|
||||
from get_playlists_v2 import fetch_server_playlist
|
||||
|
||||
# Load config
|
||||
config_file = 'config/app_config.json'
|
||||
with open(config_file, 'r') as f:
|
||||
config = json.load(f)
|
||||
|
||||
print("=" * 80)
|
||||
print("TESTING SERVER PLAYLIST FETCH")
|
||||
print("=" * 80)
|
||||
|
||||
# Fetch playlist from server
|
||||
print("\n1. Fetching playlist from server...")
|
||||
server_data = fetch_server_playlist(config)
|
||||
|
||||
print(f"\n2. Server Response:")
|
||||
print(f" Version: {server_data.get('version', 'N/A')}")
|
||||
print(f" Playlist items: {len(server_data.get('playlist', []))}")
|
||||
|
||||
print(f"\n3. Detailed Playlist Items:")
|
||||
for idx, item in enumerate(server_data.get('playlist', []), 1):
|
||||
print(f"\n Item {idx}:")
|
||||
print(f" file_name: {item.get('file_name', 'N/A')}")
|
||||
print(f" url: {item.get('url', 'N/A')}")
|
||||
print(f" duration: {item.get('duration', 'N/A')}")
|
||||
|
||||
print("\n" + "=" * 80)
|
||||
print(f"TOTAL: Server has {len(server_data.get('playlist', []))} files")
|
||||
print("=" * 80)
|
||||
|
||||
# Save to file for inspection
|
||||
output_file = 'server_response_debug.json'
|
||||
with open(output_file, 'w') as f:
|
||||
json.dump(server_data, f, indent=2)
|
||||
print(f"\nFull server response saved to: {output_file}")
|
||||
Reference in New Issue
Block a user