stan44/Fbrowser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement database integration

Browse Source 
This commit introduces database integration using FireflyDB for storing and retrieving audio file metadata.

- Integrated FireflyDB for persistent storage of metadata.
- Added methods to check for existing metadata in the database and retrieve it.
- Modified the Organizer class to use FireflyDB for processing metadata.
- Added auto-scanning and metadata extraction upon directory opening in Fbrowser.
- Created archiver.py and metaextract.py to house the ArchiveExtractor and MetadataExtractor classes respectively.
- Added .gitignore entries for Firefly related files.
- Added Mock MIT License, Contributor License Agreement, and Pro Edition License Agreement files.
This commit is contained in:
Stan44 2025-04-11 22:30:39 -05:00
parent 3b909bb756
commit 98ee9ce50a
10 changed files with 1039 additions and 249 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
.gitignore vendored
View File

@ -111,3 +111,13 @@ maintest.py
# Generated MIDI files - optional
*.mid
*.midi
# Firefly Related Files
Firefly.exe
Firefly.XML
libfirefly.XML
FireflyClient.exe
Firefly.exe
Firefly.dll
redis_viewer.py
/templates/

37
Contributor License Agreement.md Normal file
View File

@ -0,0 +1,37 @@
fbroswer Contributor License Agreement (CLA)
This Contributor License Agreement ("Agreement") is between you ("Contributor") and [Stan ] ("Project") and governs your contributions to the fbroswer Community Edition (the "Project").
1. Definitions
1.1. "Contribution" means any original work of authorship, including any modifications or additions to an existing work, that is intentionally submitted by Contributor to the Project for inclusion in, or documentation of, any of the products owned or managed by the Project (the "Work").
1.2. "Contributor" means the individual or legal entity who submits a Contribution to the Project.
1.3. "Project" means the open-source fbroswer Community Edition and its associated repositories, documentation, and websites.
2. Grant of Rights
2.1. Subject to the terms of this Agreement, Contributor hereby grants to Project and to recipients of software distributed by Project a perpetual, worldwide, non-exclusive, royalty-free, irrevocable, sublicensable, and transferable license to:
a) reproduce, prepare derivative works of, publicly display, publicly perform, sublicense, and distribute the Contribution and such derivative works;
b) incorporate the Contribution into any form, medium, or technology now known or later developed.
3. Moral Rights
3.1. To the extent Contributor has moral rights in the Contribution, Contributor hereby irrevocably transfers and waives such moral rights to the fullest extent permitted by applicable law.
4. Warranties and Representations
4.1. Contributor represents that:
a) Contributor is entitled to grant the rights to the Contribution under this Agreement;
b) the Contribution is Contributor's original creation and does not violate any third-party rights;
c) if the Contribution includes material from third parties, Contributor has obtained any necessary permissions and clearly identified such material.
4.2. Contributor provides the Contribution "AS IS," without warranty of any kind, and Project disclaims all warranties, express or implied.
5. No Compensation
5.1. Contributor agrees that the rights granted hereunder are granted without expectation of monetary compensation.
6. Contribution Process
6.1. To make a Contribution, Contributor may submit a pull request, patch, or other submission through the Project's contribution process. By doing so, Contributor agrees that the Contribution is subject to this Agreement.
7. Termination
7.1. This Agreement and the licenses granted herein will terminate automatically if Contributor fails to comply with any term of this Agreement. Upon termination, Project may cease to distribute the Contribution but retains the rights granted prior to termination.
8. General 8.1. This Agreement is governed by the laws of [United States], without regard to conflict-of-law principles. 8.2. If any provision of this Agreement is held invalid or unenforceable, the remaining provisions will remain in full force and effect. 8.3. This Agreement constitutes the entire agreement between the parties regarding the Contribution and supersedes all prior or contemporaneous understandings.
By submitting a Contribution, you accept and agree to the terms of this Agreement.

140
Fbrowser.py
View File

@ -31,15 +31,13 @@ from PyQt6.QtGui import QFileSystemModel, QCursor
from PyQt6.QtMultimedia import QMediaPlayer, QAudioOutput
from MidPlay import MidPlay
from timer_m import Timer_Ui
from ScanOrg101 import (
Organizer,
FileFilterProxyModel,
DirectoryFilterProxyModel,
ArchiveExtractor,
mutagen,
)
import mutagen
warnings.filterwarnings("ignore", category=DeprecationWarning)
@ -71,7 +69,12 @@ class MetadataTooltip(QFrame):
}
"""
)
self.organizer = Organizer(
use_db=True,
db_host="localhost", # Make sure this matches your FireflyDB server
db_port=6379,
db_password="", # Add password if needed
)
layout = QVBoxLayout(self)
self.title_label = QLabel("")
self.title_label.setObjectName("title")
@ -327,6 +330,9 @@ class Fbrowser(QMainWindow):
self.file_filter_model.mapFromSource(self.list_model.index(path))
)
# Automatically scan the directory and extract metadata
self.auto_scan_directory(path)
def on_item_double_clicked(self, index):
"""on double click do stuff"""
source_index = self.file_filter_model.mapToSource(index)
@ -666,9 +672,31 @@ class Fbrowser(QMainWindow):
def extract_hover_metadata(self, file_path):
"""Extract metadata for the hovered file"""
# First check if we have this metadata in the database
if hasattr(self.organizer, "db") and self.organizer.db:
db_metadata = self.organizer.get_metadata_from_db(file_path)
if db_metadata:
# Add the size field which might not be in the database
try:
db_metadata["size"] = (
f"{os.path.getsize(file_path) / (1024*1024):.2f} MB"
)
except:
db_metadata["size"] = "Unknown"
# Cache the metadata
self.metadata_cache[file_path] = db_metadata
# Show the tooltip if we're still hovering over the same file
if self.hover_path == file_path:
self.show_metadata_tooltip(db_metadata)
return
# If not in database, extract it directly
try:
# Use mutagen directly for speed
audio = mutagen.File(file_path) # type: ignore
audio = mutagen.File(file_path)
if not audio:
return
@ -689,6 +717,10 @@ class Fbrowser(QMainWindow):
# Cache the metadata
self.metadata_cache[file_path] = metadata
# Store in database for future use
if hasattr(self.organizer, "db") and self.organizer.db:
self.organizer.store_metadata(metadata)
# Show the tooltip if we're still hovering over the same file
if self.hover_path == file_path:
self.show_metadata_tooltip(metadata)
@ -728,6 +760,104 @@ class Fbrowser(QMainWindow):
# Auto-hide after 8 seconds
QTimer.singleShot(8000, self.metadata_tooltip.hide)
def closeEvent(self, event):
"""Handle application close event"""
# Close database connection
if hasattr(self, "organizer") and self.organizer:
self.organizer.close()
event.accept()
def auto_scan_directory(self, path):
"""Automatically scan a directory and extract metadata when opened"""
# Only scan directories that are likely to contain audio files
# to avoid unnecessary scanning of system folders
if not self.should_auto_scan(path):
return
print(f"Auto-scanning directory: {path}")
# Initialize organizer with database connection if not already done
if not hasattr(self.organizer, "db") or self.organizer.db is None:
self.organizer = Organizer(
use_db=True, db_host="localhost", db_port=6379, db_password=""
)
# Connect organizer signals
self.organizer.on_progress_update = self.update_progress
# Custom scan handler that checks for existing metadata
def scan_and_extract():
# Filter files that don't have metadata in DB
files_to_process = []
for file_path in self.organizer.file_list:
if file_path.lower().endswith(
(".mp3", ".wav", ".flac", ".m4a", ".wma", ".mid", ".midi")
):
if not self.organizer.has_metadata_in_db(file_path):
files_to_process.append(file_path)
if files_to_process:
print(
f"Found {len(files_to_process)} files without metadata, extracting..."
)
self.organizer.file_list = files_to_process
self.organizer.extract_metadata()
else:
print(
"All files already have metadata in database, skipping extraction"
)
self.progress_bar.setValue(100)
# When scan completes, check and extract metadata as needed
self.organizer.on_scan_complete = scan_and_extract
# Start scan
self.organizer.start_scan(path)
def auto_extract_metadata(self):
"""Automatically extract metadata after scanning"""
if not self.organizer.file_list:
print("No audio files found to extract metadata from")
return
print(
f"Auto-extracting metadata for {len(self.organizer.file_list)} files"
)
# Connect signals
self.organizer.on_progress_update = self.update_progress # type: ignore
self.organizer.on_metadata_complete = lambda: print("Metadata extraction complete") # type: ignore
# Start extraction
self.organizer.extract_metadata()
def should_auto_scan(self, path):
"""Determine if a directory should be automatically scanned"""
# Skip system directories and very large directories
if any(
system_dir in path
for system_dir in [
"/Windows",
"C:\\Windows",
"/System",
"C:\\Program Files",
"C:\\Program Files (x86)",
"/usr/bin",
"/bin",
]
):
return False
# Check if the directory has a reasonable number of files
try:
file_count = len(os.listdir(path))
if file_count > 1000: # Skip very large directories
return False
except (PermissionError, OSError):
return False
return True
if __name__ == "__main__":
app = QApplication(sys.argv)

21
Mock MIT License.md Normal file
View File

@ -0,0 +1,21 @@
MIT License
Copyright (c) [Year] [Your Name or Organization]
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the “Software”), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

62
Mock Pro Edition License Agreement.md Normal file
View File

@ -0,0 +1,62 @@
fbroswer Pro Edition License Agreement
======================================
This License Agreement (“Agreement”) is a binding contract between you (“Licensee”)
and [Your Name or Organization] (“Licensor”) for the licensed software product
known as “fbroswer Pro Edition” including all updates, add‑ons, and accompanying
documentation (collectively, the “Software”).
1. GRANT OF LICENSE
Licensor hereby grants Licensee a non‑exclusive, non‑transferable, revocable
license to install and use one copy of the Software on up to [N] machines
owned or controlled by Licensee, solely for Licensee’s internal business or
personal use, subject to payment of the license fee described below.
2. LICENSE FEE & PAYMENT
a. Licensee agrees to pay Licensor a one‑time license fee of USD $[Amount].
b. All payments are non‑refundable. Licensor may suspend license rights for
non‑payment or late payment beyond [30] days after invoice date.
3. RESTRICTIONS
Licensee shall not, and shall not permit others to:
- Reverse‑engineer, decompile, or disassemble the Software except as expressly
permitted by law.
- Rent, lease, sublicense, or distribute the Software to third parties.
- Remove or alter any proprietary notices or labels on the Software.
4. SUPPORT & MAINTENANCE
Licensor will provide free updates and bug‑fix releases for a period of [12]
months from the date of purchase. Extended maintenance contracts are available
separately.
5. OWNERSHIP
The Software is licensed, not sold. Licensor and its suppliers retain all
rights, title, and interest in and to the Software, including all intellectual
property rights.
6. WARRANTY DISCLAIMER
THE SOFTWARE IS PROVIDED “AS IS” WITHOUT WARRANTY OF ANY KIND. LICENSOR
DISCLAIMS ALL WARRANTIES, WHETHER EXPRESS, IMPLIED, OR STATUTORY, INCLUDING
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND NON‑INFRINGEMENT.
7. LIMITATION OF LIABILITY
IN NO EVENT SHALL LICENSOR BE LIABLE FOR ANY INDIRECT, INCIDENTAL, SPECIAL,
OR CONSEQUENTIAL DAMAGES ARISING OUT OF OR IN CONNECTION WITH THE SOFTWARE,
EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGES.
8. TERMINATION
This Agreement will terminate automatically if Licensee fails to comply with
any term herein. Upon termination, Licensee must cease all use of the Software
and destroy all copies in its possession.
9. GOVERNING LAW
This Agreement shall be governed by and construed in accordance with the laws
of [Your State/Country], without regard to its conflict of law principles.
10. ENTIRE AGREEMENT
This Agreement constitutes the entire agreement between the parties
concerning the Software and supersedes all prior or contemporaneous
understandings or agreements.
By installing, copying, or otherwise using the Software, Licensee agrees to be
bound by the terms of this Agreement.

449
ScanOrg101.py
View File

@ -3,13 +3,15 @@ ScanOrg101.py - Enhanced file scanning and organization module
"""
# flake8: noqa: E501
import os
import concurrent.futures
import zipfile
import py7zr
import rarfile # typed: ignore
import mutagen
from collections import deque
import time
from PyQt6.QtCore import Qt, QThread, QSortFilterProxyModel, pyqtSignal
from dbman import FireflyDB
from metaextract import MetadataExtractor, mutagen
from archiver import ArchiveExtractor
# Directory Filter Proxy Model
@ -313,228 +315,38 @@ class FileScanner(QThread):
self.stop_requested = True
# Metadata Extractor
class MetadataExtractor(QThread):
metadata_extracted = pyqtSignal(dict)
extraction_complete = pyqtSignal()
progress_update = pyqtSignal(int)
def __init__(self, file_list):
super().__init__()
self.file_list = file_list
self.stop_requested = False
self.metadata_cache = {}
def run(self):
total_files = len(self.file_list)
processed_files = 0
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = []
for file_path in self.file_list:
if self.stop_requested:
break
if file_path in self.metadata_cache:
self.metadata_extracted.emit(
self.metadata_cache[file_path]
)
processed_files += 1
self.progress_update.emit(
int(processed_files / total_files * 100)
)
else:
futures.append(
executor.submit(self.extract_metadata, file_path)
)
for future in concurrent.futures.as_completed(futures):
if self.stop_requested:
break
try:
metadata = future.result()
if metadata:
self.metadata_extracted.emit(metadata)
except Exception as e:
print(f"Error extracting metadata: {e}")
processed_files += 1
self.progress_update.emit(
int(processed_files / total_files * 100)
)
self.extraction_complete.emit()
def extract_metadata(self, file_path):
try:
if not os.path.isfile(file_path):
return None
# Skip non-audio files
if not file_path.lower().endswith(
(".mp3", ".wav", ".flac", ".m4a", ".wma", ".mid", ".midi")
):
return None
audio = mutagen.File(file_path) # type: ignore
if not audio:
return None
metadata = {
"file_path": file_path,
"artist": self._get_tag(audio, "artist", "Unknown Artist"),
"album": self._get_tag(audio, "album", "Unknown Album"),
"title": self._get_tag(
audio, "title", os.path.basename(file_path)
),
"genre": self._get_tag(audio, "genre", "Unknown Genre"),
"year": self._get_tag(audio, "date", "Unknown Year"),
}
# Cache the result
self.metadata_cache[file_path] = metadata
return metadata
except Exception as e:
print(f"Error processing {file_path}: {e}")
return None
def _get_tag(self, audio, tag_name, default_value):
"""Helper method to safely extract tags from audio files"""
try:
if tag_name in audio:
value = audio[tag_name]
if isinstance(value, list) and len(value) > 0:
return str(value[0])
return str(value)
except Exception:
pass
return default_value
def stop(self):
self.stop_requested = True
# Archive Extractor not fully tested or implemented
class ArchiveExtractor(QThread):
extraction_progress = pyqtSignal(int)
extraction_complete = pyqtSignal(list) # Emits list of extracted files
extraction_error = pyqtSignal(str)
def __init__(self, archive_path, extraction_dir):
super().__init__()
self.archive_path = archive_path
self.extraction_dir = extraction_dir
self.stop_requested = False
def run(self):
try:
extracted_files = []
if self.archive_path.lower().endswith(".zip"):
extracted_files = self._extract_zip()
elif self.archive_path.lower().endswith(".7z"):
extracted_files = self._extract_7z()
elif self.archive_path.lower().endswith(".rar"):
extracted_files = self._extract_rar()
else:
self.extraction_error.emit(
f"Unsupported archive format: {self.archive_path}"
)
return
self.extraction_complete.emit(extracted_files)
except Exception as e:
self.extraction_error.emit(f"Extraction error: {str(e)}")
def _extract_zip(self):
extracted_files = []
try:
with zipfile.ZipFile(self.archive_path, "r") as zip_ref:
file_list = zip_ref.namelist()
total_files = len(file_list)
for i, file in enumerate(file_list):
if self.stop_requested:
break
zip_ref.extract(file, self.extraction_dir)
extracted_files.append(
os.path.join(self.extraction_dir, file)
)
self.extraction_progress.emit(
int((i + 1) / total_files * 100)
)
except Exception as e:
self.extraction_error.emit(f"ZIP extraction error: {str(e)}")
return extracted_files
def _extract_7z(self):
extracted_files = []
try:
with py7zr.SevenZipFile(self.archive_path, mode="r") as z:
file_list = z.getnames()
total_files = len(file_list)
for i, file in enumerate(file_list):
if self.stop_requested:
break
z.extract(self.extraction_dir, [file])
extracted_files.append(
os.path.join(self.extraction_dir, file)
)
self.extraction_progress.emit(
int((i + 1) / total_files * 100)
)
except Exception as e:
self.extraction_error.emit(f"7Z extraction error: {str(e)}")
return extracted_files
def _extract_rar(self):
extracted_files = []
try:
with rarfile.RarFile(self.archive_path) as rf:
file_list = rf.namelist()
total_files = len(file_list)
for i, file in enumerate(file_list):
if self.stop_requested:
break
rf.extract(file, self.extraction_dir)
extracted_files.append(
os.path.join(self.extraction_dir, file)
)
self.extraction_progress.emit(
int((i + 1) / total_files * 100)
)
except Exception as e:
self.extraction_error.emit(f"RAR extraction error: {str(e)}")
return extracted_files
def stop(self):
self.stop_requested = True
# Main Organizer class
class Organizer:
def __init__(self):
def __init__(
self, use_db=False, db_host="localhost", db_port=6379, db_password=None
):
self.file_list = []
self.dir_list = []
self.scanner = None
self.metadata_extractor = None
self.archive_extractor = None
# Metadata organization
self.artists = set()
self.albums = set()
self.genres = set()
self.years = set()
# Signals for UI updates
self.on_scan_complete = None
self.on_progress_update = None
self.on_metadata_complete = None
# Database integration - use FireflyDB from dbman.py
self.use_db = use_db
self.db_manager = FireflyDB()
if use_db:
self.db_manager.connect_to(use_db, db_host, db_port, db_password)
self.db = self.db_manager.db
else:
self.db = None
def close(self):
"""Close database connection when done"""
if hasattr(self, "db_manager"):
self.db_manager.close()
self.db = None
def start_scan(self, path):
"""Start scanning a directory for files"""
self.file_list.clear()
@ -574,11 +386,21 @@ class Organizer:
self.scanner.wait()
def extract_metadata(self):
"""Extract metadata from audio files"""
"""Extract metadata from audio files using MetadataExtractor from metaextract.py"""
if not self.file_list:
print("No files to extract metadata from")
return
# Verify database connection if enabled
if self.use_db and self.db:
if not self.db_manager.verify_database_connection():
print(
"Warning: Database verification failed, continuing without database"
)
self.use_db = False
self.db = None
# Use MetadataExtractor from metaextract.py
self.metadata_extractor = MetadataExtractor(self.file_list)
self.metadata_extractor.metadata_extracted.connect(
self.process_metadata
@ -593,59 +415,198 @@ class Organizer:
self.on_progress_update
)
# Set the callback for metadata completion
self.metadata_extractor.on_metadata_complete = (
self.on_metadata_complete
)
self.metadata_extractor.start()
def process_metadata(self, metadata):
"""Process extracted metadata"""
"""Process extracted metadata using FireflyDB from dbman.py"""
# Use the database manager to process metadata
if hasattr(self, "db_manager"):
self.db_manager.process_metadata(metadata)
# Also update local sets for UI display
if "artist" in metadata and metadata["artist"]:
if not hasattr(self, "artists"):
self.artists = set()
self.artists.add(metadata["artist"])
if "album" in metadata and metadata["album"]:
if not hasattr(self, "albums"):
self.albums = set()
self.albums.add(metadata["album"])
if "genre" in metadata and metadata["genre"]:
if not hasattr(self, "genres"):
self.genres = set()
self.genres.add(metadata["genre"])
if "year" in metadata and metadata["year"]:
if not hasattr(self, "years"):
self.years = set()
self.years.add(metadata["year"])
def metadata_extraction_complete(self):
"""Handle metadata extraction completion"""
print(
f"Metadata extraction complete. Artists: {len(self.artists)},\
Albums: {len(self.albums)}, Genres: {len(self.genres)},\
Years: {len(self.years)}"
)
if self.on_metadata_complete:
self.on_metadata_complete()
def extract_archives(self):
"""Extract archives"""
if not self.file_list:
print("No files to extract archives from")
return
self.archive_extractor = ArchiveExtractor(
self.file_list, extraction_dir=None
)
self.archive_extractor.extraction_complete.connect(
self.archive_extraction_complete
f"Metadata extraction complete. Artists: {len(getattr(self, 'artists', []))}, "
f"Albums: {len(getattr(self, 'albums', []))}, Genres: {len(getattr(self, 'genres', []))}, "
f"Years: {len(getattr(self, 'years', []))}"
)
# Connect progress signal if handler exists
def extract_archive(self, archive_path, extraction_dir):
"""Extract an archive to specified directory using ArchiveExtractor from archiver.py"""
if not os.path.isfile(archive_path):
print(f"Error: Archive file {archive_path} does not exist")
return False
if not os.path.isdir(extraction_dir):
print(
f"Error: Extraction directory {extraction_dir} does not exist"
)
return False
print(f"Extracting archive {archive_path} to {extraction_dir}")
# Create an ArchiveExtractor instance from archiver.py
self.archive_extractor = ArchiveExtractor(archive_path, extraction_dir)
# Connect signals
if self.on_progress_update:
self.archive_extractor.extraction_progress.connect(
self.on_progress_update
)
# Define completion handler
def on_extraction_complete(extracted_files):
print(
f"Archive extraction complete. Extracted {len(extracted_files)} files."
)
# Add extracted files to our file list if they match our criteria
for file_path in extracted_files:
if os.path.isfile(file_path) and any(
file_path.lower().endswith(ext)
for ext in [
".mp3",
".wav",
".flac",
".m4a",
".wma",
".mid",
".midi",
]
):
self.file_list.append(file_path)
# Automatically extract metadata from audio files if enabled
audio_files = [
f
for f in extracted_files
if any(
f.lower().endswith(ext)
for ext in [
".mp3",
".wav",
".flac",
".m4a",
".wma",
".mid",
".midi",
]
)
]
if audio_files and self.use_db:
print(
f"Found {len(audio_files)} audio files in archive, extracting metadata..."
)
temp_extractor = MetadataExtractor(audio_files)
temp_extractor.metadata_extracted.connect(
self.process_metadata
)
temp_extractor.start()
# Connect completion signal
self.archive_extractor.extraction_complete.connect(
on_extraction_complete
)
# Define error handler
def on_extraction_error(error_message):
print(f"Archive extraction error: {error_message}")
# Connect error signal
self.archive_extractor.extraction_error.connect(on_extraction_error)
# Start extraction
self.archive_extractor.start()
return True
def archive_extraction_complete(self):
"""Handle archive extraction completion"""
print("Archive extraction complete.")
def extract_archive_to_directory(
self, archive_path, target_directory=None
):
"""Extract an archive to a specified directory or to a subdirectory in the same location"""
if not os.path.isfile(archive_path):
print(f"Error: Archive file {archive_path} does not exist")
return False
if self.on_progress_update:
self.on_progress_update(100)
# If no target directory is specified, create one based on the archive name
if not target_directory:
archive_name = os.path.splitext(os.path.basename(archive_path))[0]
target_directory = os.path.join(
os.path.dirname(archive_path), archive_name
)
def stop_extraction(self):
"""Stop the current extraction operation"""
if self.archive_extractor:
self.archive_extractor.stop()
self.archive_extractor.wait()
# Create the directory if it doesn't exist
if not os.path.exists(target_directory):
try:
os.makedirs(target_directory)
print(f"Created directory {target_directory}")
except OSError as e:
print(f"Error creating directory {target_directory}: {e}")
return False
return self.extract_archive(archive_path, target_directory)
def has_metadata_in_db(self, file_path):
"""Check if metadata for a file already exists in the database"""
if not self.use_db or not self.db:
return False
try:
# Delegate to the db_manager
return self.db_manager.has_metadata_in_db(file_path)
except Exception as e:
print(f"Error checking metadata in database: {e}")
return False
def get_metadata_from_db(self, file_path):
"""Retrieve metadata for a file from the database"""
if not self.use_db or not self.db:
return None
try:
# Delegate to the db_manager
return self.db_manager.get_metadata_from_db(file_path)
except Exception as e:
print(f"Error retrieving metadata from database: {e}")
return None
def store_metadata(self, metadata):
"""Store audio file metadata in the database"""
if not self.use_db or not self.db:
print("Database usage is disabled, not storing metadata")
return False
try:
# Delegate to the db_manager
return self.db_manager.store_metadata(metadata)
except Exception as e:
print(f"Error storing metadata in database: {e}")
import traceback
traceback.print_exc()
return False

148
archiver.py Normal file
View File

@ -0,0 +1,148 @@
import os
import zipfile
import py7zr
import rarfile # typed: ignore
from PyQt6.QtCore import QThread, pyqtSignal
# Archive Extractor not fully tested or implemented
class ArchiveExtractor(QThread):
extraction_progress = pyqtSignal(int)
extraction_complete = pyqtSignal(list) # Emits list of extracted files
extraction_error = pyqtSignal(str)
def __init__(self, archive_path, extraction_dir):
super().__init__()
self.archive_path = archive_path
self.extraction_dir = extraction_dir
self.stop_requested = False
self.file_list = []
# Signals for UI updates
self.on_scan_complete = None
self.on_progress_update = None
self.on_metadata_complete = None
def run(self):
try:
extracted_files = []
if self.archive_path.lower().endswith(".zip"):
extracted_files = self._extract_zip()
elif self.archive_path.lower().endswith(".7z"):
extracted_files = self._extract_7z()
elif self.archive_path.lower().endswith(".rar"):
extracted_files = self._extract_rar()
else:
self.extraction_error.emit(
f"Unsupported archive format: {self.archive_path}"
)
return
self.extraction_complete.emit(extracted_files)
except Exception as e:
self.extraction_error.emit(f"Extraction error: {str(e)}")
def _extract_zip(self):
extracted_files = []
try:
with zipfile.ZipFile(self.archive_path, "r") as zip_ref:
file_list = zip_ref.namelist()
total_files = len(file_list)
for i, file in enumerate(file_list):
if self.stop_requested:
break
zip_ref.extract(file, self.extraction_dir)
extracted_files.append(
os.path.join(self.extraction_dir, file)
)
self.extraction_progress.emit(
int((i + 1) / total_files * 100)
)
except Exception as e:
self.extraction_error.emit(f"ZIP extraction error: {str(e)}")
return extracted_files
def _extract_7z(self):
extracted_files = []
try:
with py7zr.SevenZipFile(self.archive_path, mode="r") as z:
file_list = z.getnames()
total_files = len(file_list)
for i, file in enumerate(file_list):
if self.stop_requested:
break
z.extract(self.extraction_dir, [file])
extracted_files.append(
os.path.join(self.extraction_dir, file)
)
self.extraction_progress.emit(
int((i + 1) / total_files * 100)
)
except Exception as e:
self.extraction_error.emit(f"7Z extraction error: {str(e)}")
return extracted_files
def _extract_rar(self):
extracted_files = []
try:
with rarfile.RarFile(self.archive_path) as rf:
file_list = rf.namelist()
total_files = len(file_list)
for i, file in enumerate(file_list):
if self.stop_requested:
break
rf.extract(file, self.extraction_dir)
extracted_files.append(
os.path.join(self.extraction_dir, file)
)
self.extraction_progress.emit(
int((i + 1) / total_files * 100)
)
except Exception as e:
self.extraction_error.emit(f"RAR extraction error: {str(e)}")
return extracted_files
def stop(self):
self.stop_requested = True
def extract_archives(self):
"""Extract archives"""
if not self.file_list:
print("No files to extract archives from")
return
self.archive_extractor = ArchiveExtractor(
self.file_list, extraction_dir=None
)
self.archive_extractor.extraction_complete.connect(
self.archive_extraction_complete
)
# Connect progress signal if handler exists
if self.on_progress_update:
self.archive_extractor.extraction_progress.connect(
self.on_progress_update
)
self.archive_extractor.start()
def archive_extraction_complete(self):
"""Handle archive extraction completion"""
print("Archive extraction complete.")
if self.on_progress_update:
self.on_progress_update(100)
def stop_extraction(self):
"""Stop the current extraction operation"""
if self.archive_extractor:
self.archive_extractor.stop()
self.archive_extractor.wait()

241
dbman.py Normal file
View File

@ -0,0 +1,241 @@
from datetime import datetime
from ifireflylib import IFireflyClient as FireflyDatabase
class FireflyDB:
def __init__(self):
# Initialize with default values
self.use_db = False
self.db = None
self.artists = set()
self.albums = set()
self.genres = set()
self.years = set()
def connect_to(
self, use_db=False, db_host="localhost", db_port=6379, db_password=None
):
# Set the use_db flag
self.use_db = use_db
# Metadata organization
self.artists = set()
self.albums = set()
self.genres = set()
self.years = set()
if use_db:
try:
print(f"Connecting to FireflyDB at {db_host}:{db_port}")
self.db = FireflyDatabase(
host=db_host, port=db_port, password=db_password
)
# Test connection
if not self.db.ping():
print(
"Warning: Could not connect to FireflyDB. Continuing without database."
)
self.use_db = False
self.db = None
else:
print("Successfully connected to FireflyDB")
# Store connection timestamp
timestamp = str(datetime.now())
print(f"Setting last_connection timestamp: {timestamp}")
self.db.string_ops.string_set("last_connection", timestamp)
print("Connection timestamp stored successfully")
except Exception as e:
print(f"Error connecting to FireflyDB: {e}")
import traceback
traceback.print_exc()
self.use_db = False
self.db = None
def close(self):
"""Close database connection when done"""
if self.db:
self.db.close()
self.db = None
def has_metadata_in_db(self, file_path):
"""Check if metadata for a file already exists in the database"""
if not self.use_db or not self.db:
return False
try:
key = f"audio:{file_path}"
# Use hash_exists to check if the key exists in the database
exists = self.db.hash_ops.hash_exists(key, "title")
return exists
except Exception as e:
print(f"Error checking metadata in database: {e}")
return False
def store_metadata(self, metadata):
"""Store audio file metadata in the database"""
if not self.use_db or not self.db:
print("Database usage is disabled, not storing metadata")
return False
try:
# Use the file path as a unique key
file_path = metadata["file_path"]
key = f"audio:{file_path}"
# Log the storage attempt
print(f"Storing metadata for {file_path} in FireflyDB")
print(f"Metadata fields: {list(metadata.keys())}")
# Store as a hash with all metadata fields
success = True
for field, value in metadata.items():
if field != "file_path": # Skip using file_path as a field
print(f" Setting field {field}={value}")
# Use the direct hash_set method instead of hash_ops
result = self.db.hash_ops.hash_set(key, field, value)
if not result:
print(
f" Failed to store field {field} for {file_path}"
)
success = False
else:
print(f" Successfully stored field {field}")
# Add to index lists for quick lookup
if "artist" in metadata and metadata["artist"]:
artist_key = f"index:artist:{metadata['artist']}"
print(f" Adding to artist index: {artist_key}")
self.db.list_ops.list_right_push(artist_key, file_path)
if "album" in metadata and metadata["album"]:
album_key = f"index:album:{metadata['album']}"
print(f" Adding to album index: {album_key}")
self.db.list_ops.list_right_push(album_key, file_path)
if "genre" in metadata and metadata["genre"]:
genre_key = f"index:genre:{metadata['genre']}"
print(f" Adding to genre index: {genre_key}")
self.db.list_ops.list_right_push(genre_key, file_path)
# Add to a master list of all audio files for easy retrieval
print(" Adding to master audio files list")
self.db.list_ops.list_right_push("all_audio_files", file_path)
# Store timestamp of when metadata was added
self.db.hash_ops.hash_set(key, "timestamp", str(datetime.now()))
# Verify storage by retrieving one field
if "title" in metadata:
retrieved_title = self.db.hash_ops.hash_get(key, "title")
print(f" Verification - Retrieved title: {retrieved_title}")
if retrieved_title != metadata["title"]:
print(
f" Verification failed: expected '{metadata['title']}', got '{retrieved_title}'"
)
success = False
print(
f"Metadata storage {'successful' if success else 'partially failed'} for {file_path}"
)
return success
except Exception as e:
print(f"Error storing metadata in database: {e}")
import traceback
traceback.print_exc()
return False
def get_metadata_from_db(self, file_path):
"""Retrieve metadata for a file from the database"""
if not self.use_db or not self.db:
return None
try:
key = f"audio:{file_path}"
metadata = self.db.hash_ops.hash_get_all(key)
if metadata:
# Add the file path to the metadata
metadata["file_path"] = file_path
return metadata
return None
except Exception as e:
print(f"Error retrieving metadata from database: {e}")
return None
def search_by_artist(self, artist):
"""Search for files by artist"""
if not self.use_db or not self.db:
return []
try:
key = f"index:artist:{artist}"
return self.db.list_range(key, 0, -1)
except Exception as e:
print(f"Error searching by artist: {e}")
return []
# Similar methods for album and genre searches
def process_metadata(self, metadata):
"""Process extracted metadata"""
if "artist" in metadata and metadata["artist"]:
self.artists.add(metadata["artist"])
if "album" in metadata and metadata["album"]:
self.albums.add(metadata["album"])
if "genre" in metadata and metadata["genre"]:
self.genres.add(metadata["genre"])
if "year" in metadata and metadata["year"]:
self.years.add(metadata["year"])
# Store in database if enabled
if self.use_db and self.db:
db_success = self.store_metadata(metadata)
if db_success:
print(
f"Successfully stored metadata for {metadata.get('file_path', 'unknown file')} in database"
)
else:
print(
f"Failed to store metadata for {metadata.get('file_path', 'unknown file')} in database"
)
def verify_database_connection(self):
"""Verify that the database connection is working properly"""
if not self.use_db or not self.db:
print("Database usage is disabled")
return False
try:
# Test ping
ping_result = self.db.ping()
if not ping_result:
print("Database ping failed")
return False
# Test basic operations
test_key = "test:connection"
test_value = "connection_test"
# Test string operations
string_set_result = self.db.string_set(test_key, test_value)
if not string_set_result:
print("Failed to set test string in database")
return False
string_get_result = self.db.string_get(test_key)
if string_get_result != test_value:
print(
f"String get test failed. Expected '{test_value}', got '{string_get_result}'"
)
return False
# Clean up
self.db.delete(test_key)
print("Database connection verified successfully")
return True
except Exception as e:
print(f"Database verification failed with error: {e}")
return False

40
ifireflylib.pyi Normal file
View File

@ -0,0 +1,40 @@
from typing import Dict, List, Optional
class IFireflyClient:
def __init__(
self,
host: str = "localhost",
port: int = 6379,
password: Optional[str] = None,
) -> None: ...
def ping(self) -> bool: ...
def close(self) -> None: ...
# String operations
def string_ops(self, key: str) -> Dict[str, str]: ...
def string_set(self, key: str, value: str) -> bool: ...
def string_get(self, key: str) -> Optional[str]: ...
def string_delete(self, key: str) -> bool: ...
# List operations
def list_right_push(self, key: str, value: str) -> int: ...
def list_left_push(self, key: str, value: str) -> int: ...
def list_right_pop(self, key: str) -> Optional[str]: ...
def list_left_pop(self, key: str) -> Optional[str]: ...
def list_range(self, key: str, start: int, end: int) -> List[str]: ...
def list_length(self, key: str) -> int: ...
def list_ops(self, key: str) -> List[str]: ...
# Hash operations
def hash_ops(self, key: str) -> Dict[str, str]: ...
def hash_set(self, key: str, field: str, value: str) -> bool: ...
def hash_get(self, key: str, field: str) -> Optional[str]: ...
def hash_delete(self, key: str, field: str) -> bool: ...
def hash_get_all(self, key: str) -> Dict[str, str]: ...
def hash_exists(self, key: str, field: str) -> bool: ...
# Connection management
def execute_command(self, command: str, *args) -> Optional[str]: ...
def on_progress_update(self, progress: int) -> None: ...
def on_metadata_complete(self) -> None: ...
def on_scan_complete(self) -> None: ...

140
metaextract.py Normal file
View File

@ -0,0 +1,140 @@
import os
import concurrent.futures
import mutagen
from datetime import datetime
from PyQt6.QtCore import QThread, pyqtSignal
# Metadata Extractor
class MetadataExtractor(QThread):
metadata_extracted = pyqtSignal(dict)
extraction_complete = pyqtSignal()
progress_update = pyqtSignal(int)
def __init__(self, file_list):
super().__init__()
self.file_list = file_list
self.stop_requested = False
self.metadata_cache = {}
self.on_metadata_complete = None
# Metadata organization
self.artists = set()
self.albums = set()
self.genres = set()
self.years = set()
def run(self):
total_files = len(self.file_list)
processed_files = 0
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = []
for file_path in self.file_list:
if self.stop_requested:
break
if file_path in self.metadata_cache:
self.metadata_extracted.emit(
self.metadata_cache[file_path]
)
processed_files += 1
self.progress_update.emit(
int(processed_files / total_files * 100)
)
else:
futures.append(
executor.submit(self.extract_metadata, file_path)
)
for future in concurrent.futures.as_completed(futures):
if self.stop_requested:
break
try:
metadata = future.result()
if metadata:
self.metadata_extracted.emit(metadata)
except Exception as e:
print(f"Error extracting metadata: {e}")
processed_files += 1
self.progress_update.emit(
int(processed_files / total_files * 100)
)
self.extraction_complete.emit()
def extract_metadata(self, file_path):
try:
if not os.path.isfile(file_path):
return None
# Skip non-audio files
if not file_path.lower().endswith(
(".mp3", ".wav", ".flac", ".m4a", ".wma", ".mid", ".midi")
):
return None
print(f"Extracting metadata for {file_path}")
audio = mutagen.File(file_path) # type: ignore
if not audio:
print(f"No metadata found for {file_path}")
return None
# Get file size in MB
file_size_mb = os.path.getsize(file_path) / (1024 * 1024)
metadata = {
"file_path": file_path,
"artist": self._get_tag(audio, "artist", "Unknown Artist"),
"album": self._get_tag(audio, "album", "Unknown Album"),
"title": self._get_tag(
audio, "title", os.path.basename(file_path)
),
"genre": self._get_tag(audio, "genre", "Unknown Genre"),
"year": self._get_tag(audio, "date", "Unknown Year"),
"size_mb": f"{file_size_mb:.2f}",
"filename": os.path.basename(file_path),
"extension": os.path.splitext(file_path)[1].lower(),
"last_modified": str(
datetime.fromtimestamp(os.path.getmtime(file_path))
),
"extracted_at": str(datetime.now()),
}
print(f"Extracted metadata: {metadata}")
# Cache the result
self.metadata_cache[file_path] = metadata
return metadata
except Exception as e:
print(f"Error processing {file_path}: {e}")
import traceback
traceback.print_exc()
return None
def _get_tag(self, audio, tag_name, default_value):
"""Helper method to safely extract tags from audio files"""
try:
if tag_name in audio:
value = audio[tag_name]
if isinstance(value, list) and len(value) > 0:
return str(value[0])
return str(value)
except Exception:
pass
return default_value
def stop(self):
self.stop_requested = True
def metadata_extraction_complete(self):
"""Handle metadata extraction completion"""
print(
f"Metadata extraction complete. Artists: {len(self.artists)},\
Albums: {len(self.albums)}, Genres: {len(self.genres)},\
Years: {len(self.years)}"
)
if self.on_metadata_complete:
self.on_metadata_complete()