Raincloud/synology-thumbgen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

color coded cli output!

Browse Source
This commit is contained in:
Raincloud
2026-03-22 12:58:19 -06:00
parent a8205447e3
commit 53430f768f
4 changed files with 565 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
psthumbgen.py
+46 -34
View File
@@ -72,6 +72,18 @@ def _ffmpeg_enabled():
return False
_ANSI_BLUE = "\033[94m"
_ANSI_GREEN = "\033[92m"
_ANSI_RST = "\033[0m"
def _print_icaros(msg):
print(f"{_ANSI_BLUE}{msg}{_ANSI_RST}")
def _print_ffmpeg(msg):
print(f"{_ANSI_GREEN}{msg}{_ANSI_RST}")
def find_files(dir):
# Only process formats that Synology doesn't handle well
@@ -136,13 +148,13 @@ def create_video_thumbnails(source_path, dest_dir):
# Try Windows thumbnail extraction first (leverages Icaros provider when present)
windows_thumb = extract_windows_thumbnail(source_path)
if windows_thumb:
print(f" -> SUCCESS: Using Windows/Icaros provider")
_print_icaros(f" -> SUCCESS: Using Windows/Icaros provider")
generate_synology_thumbnails(windows_thumb, dest_dir, include_video_screenshot=True)
return
# Optionally fall back to FFmpeg (disabled by default)
if _ffmpeg_enabled():
print(f"Windows/Icaros extraction failed for {source_path}, using FFmpeg...")
_print_ffmpeg(f"Windows/Icaros extraction failed for {source_path}, using FFmpeg...")
create_video_thumbnails_ffmpeg(source_path, dest_dir)
else:
print(f"Windows/Icaros extraction failed for {source_path}, FFmpeg disabled (THUMBGEN_ENABLE_FFMPEG=0). Skipping.")
@@ -175,15 +187,15 @@ def create_video_thumbnails_ffmpeg(source_path, dest_dir):
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
print(f"Warning: FFmpeg failed for {source_path}: {result.stderr}")
_print_ffmpeg(f"Warning: FFmpeg failed for {source_path}: {result.stderr}")
continue
except FileNotFoundError:
print("Warning: FFmpeg not found. Video thumbnails will be skipped.")
print("Install FFmpeg: https://ffmpeg.org/download.html")
_print_ffmpeg("Warning: FFmpeg not found. Video thumbnails will be skipped.")
_print_ffmpeg("Install FFmpeg: https://ffmpeg.org/download.html")
break
except Exception as e:
print(f"Error processing video {source_path}: {e}")
_print_ffmpeg(f"Error processing video {source_path}: {e}")
def create_psd_thumbnails(source_path, dest_dir):
@@ -315,15 +327,15 @@ def extract_windows_thumbnail(file_path):
# Tier 3: Try Icaros cache extraction
try:
print(f" -> Trying Icaros cache extraction...")
_print_icaros(f" -> Trying Icaros cache extraction...")
icaros_thumb = extract_icaros_thumbnail(file_path)
if icaros_thumb:
print(f" -> Found thumbnail in Icaros cache for {filename}")
_print_icaros(f" -> Found thumbnail in Icaros cache for {filename}")
return icaros_thumb
else:
print(f" -> No thumbnail in Icaros cache for {filename}")
_print_icaros(f" -> No thumbnail in Icaros cache for {filename}")
except Exception as e:
print(f" -> Icaros cache extraction failed: {e}")
_print_icaros(f" -> Icaros cache extraction failed: {e}")
print(f" -> Windows thumbnail extraction failed for {filename}")
return None
@@ -679,28 +691,28 @@ def extract_icaros_thumbnail(file_path):
# Locate Icaros cache directory
icaros_cache_dir = _get_icaros_cache_dir()
if not icaros_cache_dir:
print(f" -> No Icaros cache directory found")
_print_icaros(f" -> No Icaros cache directory found")
return None
# Debug: Show cache directory contents (all files)
try:
cache_files = os.listdir(icaros_cache_dir)
print(f" -> Icaros cache has {len(cache_files)} files")
_print_icaros(f" -> Icaros cache has {len(cache_files)} files")
if len(cache_files) > 0:
print(f" -> All cache files: {cache_files}")
_print_icaros(f" -> All cache files: {cache_files}")
except Exception:
pass
# Discover .icdb databases by size preference
icdb_paths = _list_icdb_databases(icaros_cache_dir)
if not icdb_paths:
print(f" -> No .icdb database files found")
_print_icaros(f" -> No .icdb database files found")
return None
# Lookup a precise index for this file from Icaros_idx.icdb (SQLite or binary)
mapped_index, preferred_db = _lookup_icaros_index(icaros_cache_dir, file_path)
if mapped_index is None:
print(f" -> No exact Icaros index entry for this file; skipping Icaros")
_print_icaros(f" -> No exact Icaros index entry for this file; skipping Icaros")
return None
# Try preferred DB first if provided
@@ -711,16 +723,16 @@ def extract_icaros_thumbnail(file_path):
for icdb_path in ordered_dbs:
img = extract_from_icdb_database(icdb_path, file_path, forced_index=mapped_index)
if img:
print(f" -> Found thumbnail in {os.path.basename(icdb_path)} via mapped index")
_print_icaros(f" -> Found thumbnail in {os.path.basename(icdb_path)} via mapped index")
return img
print(f" -> Mapped index did not resolve a thumbnail in any DB")
_print_icaros(f" -> Mapped index did not resolve a thumbnail in any DB")
return None
except ImportError:
print(f" -> Required modules not available for .icdb extraction")
_print_icaros(f" -> Required modules not available for .icdb extraction")
return None
except Exception as e:
print(f" -> Icaros cache extraction error: {e}")
_print_icaros(f" -> Icaros cache extraction error: {e}")
return None
@@ -814,12 +826,12 @@ def _build_icaros_index_map(cache_dir):
conn.close()
_ICAROS_INDEX_CACHE = mapping
if mapping:
print(f" -> Loaded Icaros index map for {len(mapping)} files")
_print_icaros(f" -> Loaded Icaros index map for {len(mapping)} files")
else:
print(f" -> Icaros index database present but no usable mapping found")
_print_icaros(f" -> Icaros index database present but no usable mapping found")
return _ICAROS_INDEX_CACHE
except Exception as e:
print(f" -> Failed to read Icaros index: {e}")
_print_icaros(f" -> Failed to read Icaros index: {e}")
_ICAROS_INDEX_CACHE = {}
return _ICAROS_INDEX_CACHE
@@ -988,7 +1000,7 @@ def extract_from_icdb_database(icdb_path, file_path, forced_index=None):
# Verify ICDB signature
if not data.startswith(b'ICDB'):
print(f" -> Invalid ICDB signature in {icdb_path}")
_print_icaros(f" -> Invalid ICDB signature in {icdb_path}")
return None
# Search for JPEG images in the file
@@ -1002,10 +1014,10 @@ def extract_from_icdb_database(icdb_path, file_path, forced_index=None):
pos += 1
if not jpeg_positions:
print(f" -> No JPEG images found in {icdb_path}")
_print_icaros(f" -> No JPEG images found in {icdb_path}")
return None
print(f" -> Found {len(jpeg_positions)} JPEG images in {icdb_path}")
_print_icaros(f" -> Found {len(jpeg_positions)} JPEG images in {icdb_path}")
# If we have a mapped index, try it directly first
if isinstance(forced_index, int) and len(jpeg_positions) > 0:
@@ -1021,11 +1033,11 @@ def extract_from_icdb_database(icdb_path, file_path, forced_index=None):
jpeg_data = data[jpeg_start:jpeg_end]
try:
img = Image.open(io.BytesIO(jpeg_data))
print(f" -> Used mapped index {pos_index} from {os.path.basename(icdb_path)}")
_print_icaros(f" -> Used mapped index {pos_index} from {os.path.basename(icdb_path)}")
return img.copy()
except Exception:
# If mapped position invalid, continue to heuristic
print(f" -> Mapped index {pos_index} invalid; falling back to heuristic")
_print_icaros(f" -> Mapped index {pos_index} invalid; falling back to heuristic")
# Heuristic alphabetical mapping (fallback)
def get_alphabetical_position(target_file_path):
@@ -1079,10 +1091,10 @@ def extract_from_icdb_database(icdb_path, file_path, forced_index=None):
try:
position = all_files.index(target_rel_path)
print(f" -> Alphabetical position estimate: {position} for {target_filename}")
_print_icaros(f" -> Alphabetical position estimate: {position} for {target_filename}")
return position
except ValueError:
print(f" -> File {target_filename} not found in current monitored directories")
_print_icaros(f" -> File {target_filename} not found in current monitored directories")
# Return a reasonable fallback position
return len(all_files) // 2 # Middle position as fallback
@@ -1093,16 +1105,16 @@ def extract_from_icdb_database(icdb_path, file_path, forced_index=None):
if cache_position >= len(jpeg_positions):
# Use modulo to wrap around (cache might be from larger file set)
cache_position = cache_position % len(jpeg_positions)
print(f" -> Wrapped position to {cache_position} (cache has {len(jpeg_positions)} entries)")
_print_icaros(f" -> Wrapped position to {cache_position} (cache has {len(jpeg_positions)} entries)")
jpeg_start = jpeg_positions[cache_position]
print(f" -> Alphabetical mapping -> JPEG {cache_position} for {os.path.basename(file_path)}")
_print_icaros(f" -> Alphabetical mapping -> JPEG {cache_position} for {os.path.basename(file_path)}")
# Find the end of the JPEG (FF D9 marker)
jpeg_end = data.find(b'\xff\xd9', jpeg_start)
if jpeg_end == -1:
print(f" -> Could not find JPEG end marker")
_print_icaros(f" -> Could not find JPEG end marker")
return None
jpeg_end += 2 # Include the end marker
@@ -1110,11 +1122,11 @@ def extract_from_icdb_database(icdb_path, file_path, forced_index=None):
# Try to create PIL Image from the JPEG data
img = Image.open(io.BytesIO(jpeg_data))
print(f" -> Successfully extracted {img.size[0]}x{img.size[1]} JPEG from {icdb_path}")
_print_icaros(f" -> Successfully extracted {img.size[0]}x{img.size[1]} JPEG from {icdb_path}")
return img.copy()
except Exception as e:
print(f" -> Error reading ICDB file {icdb_path}: {e}")
_print_icaros(f" -> Error reading ICDB file {icdb_path}: {e}")
return None