Auto-Generated Class Diagrams¶
These diagrams are automatically generated from the source code using pyreverse.
Auto-Updated
These diagrams update automatically when the documentation is built. They reflect the current state of the codebase.
Core Track Processing¶
classDiagram
class ArtistRenamer {
console_logger : Logger
error_logger : Logger
has_mapping : bool
rename_tracks(tracks: list[TrackDict]) list[TrackDict]
}
class BaseProcessor {
analytics : AnalyticsProtocol
config : AppConfig
console_logger : Logger
dry_run : bool
error_logger : Logger
get_dry_run_actions() list[dict[str, Any]]
}
class BatchTrackFetcher {
analytics : NoneType
ap_client : AppleScriptClientProtocol
cache_service : CacheServiceProtocol
config : AppConfig
console_logger : Logger
dry_run : bool
error_logger : Logger
fetch_all_tracks(batch_size: int) list[TrackDict]
}
class GenreManager {
track_processor
deduplicate_tracks_by_id(tracks: list[TrackDict]) list[TrackDict]
filter_tracks_for_incremental_update(tracks: list[TrackDict], last_run_time: datetime | None) list[TrackDict]
get_dry_run_actions() list[dict[str, Any]]
is_missing_or_unknown_genre(track: TrackDict) bool
parse_date_added(track: TrackDict) datetime | None
process_batch_results(batch_results: list[Any], updated_tracks: list[TrackDict], change_logs: list[ChangeLogEntry]) None
test_filter_tracks_for_update(artist_tracks: list[TrackDict], last_run: datetime | None, force_flag: bool, dominant_genre: str | None) list[TrackDict]
test_gather_with_error_handling(tasks: list[Any], operation_name: str) list[Any]
test_process_artist_genres(artist_name: str, all_artist_tracks: list[TrackDict], force_update: bool, applescript_semaphore: asyncio.Semaphore, tracks_to_update: list[TrackDict] | None) tuple[list[TrackDict], list[ChangeLogEntry]]
test_process_single_artist_wrapper(artist_name: str, artist_tracks: list[TrackDict], last_run: datetime | None, force: bool, artist_semaphore: asyncio.Semaphore, applescript_semaphore: asyncio.Semaphore | None) tuple[list[TrackDict], list[ChangeLogEntry]]
test_select_tracks_to_update_for_artist(artist_tracks: list[TrackDict], last_run: datetime | None, force_flag: bool, dominant_genre: str | None) list[TrackDict]
test_update_track_genre(track: TrackDict, new_genre: str, force_update: bool) tuple[TrackDict | None, ChangeLogEntry | None]
update_genres_by_artist_async(tracks: list[TrackDict], last_run_time: datetime | None, force: bool, fresh: bool) tuple[list[TrackDict], list[ChangeLogEntry]]
}
class IncrementalFilterService {
filter_tracks_for_incremental_update(tracks: list[TrackDict], last_run_time: datetime | None) list[TrackDict]
get_dry_run_actions() list[dict[str, Any]]
}
class PrereleaseHandler {
config
console_logger
pending_verification
prerelease_recheck_days
handle_prerelease_tracks(artist: str, album: str, album_tracks: list[TrackDict]) tuple[bool, list[TrackDict]]
}
class TrackCacheManager {
cache_service : CacheServiceProtocol
console_logger : Logger
snapshot_service : LibrarySnapshotServiceProtocol | None
can_use_snapshot() bool
get_cached_tracks(cache_key: str) Sequence[TrackDict] | None
load_snapshot() list[TrackDict] | None
merge_tracks(existing: list[TrackDict], updates: list[TrackDict]) list[TrackDict]
update_snapshot(tracks: list[TrackDict], processed_track_ids: Sequence[str] | None) None
}
class TrackDelta {
new_ids : list[str]
removed_ids : list[str]
updated_ids : list[str]
has_removals() bool
has_updates() bool
is_empty() bool
}
class TrackProcessor {
analytics
ap_client : AppleScriptClientProtocol
artist_renamer : ArtistRenamer | None
batch_fetcher
cache_manager
cache_service : CacheServiceProtocol
config
console_logger
dry_run : bool
dry_run_mode : str
dry_run_test_artists : set[str]
error_logger
security_validator : SecurityValidator
snapshot_service : NoneType
update_executor
fetch_tracks_async(artist: str | None, force_refresh: bool, dry_run_test_tracks: list[TrackDict] | None, ignore_test_filter: bool) list[TrackDict]
fetch_tracks_by_ids(track_ids: list[str]) list[TrackDict]
fetch_tracks_in_batches(batch_size: int) list[TrackDict]
get_dry_run_actions() list[dict[str, Any]]
set_artist_renamer(renamer: ArtistRenamer) None
set_dry_run_context(mode: str, test_artists: set[str]) None
update_artist_async(track: TrackDict, new_artist_name: str) bool
update_track_async(track_id: str, new_track_name: str | None, new_album_name: str | None, new_genre: str | None, new_year: str | None, track_status: str | None, original_artist: str | None, original_album: str | None, original_track: str | None) bool
}
class TrackUpdateExecutor {
analytics : AnalyticsProtocol
ap_client : AppleScriptClientProtocol
cache_service : CacheServiceProtocol
config : AppConfig
console_logger : Logger
dry_run : bool
error_logger : Logger
security_validator : SecurityValidator
get_dry_run_actions() list[dict[str, Any]]
set_dry_run(dry_run: bool) None
update_artist_async(track: TrackDict, new_artist_name: str) bool
update_track_async(track_id: str, new_track_name: str | None, new_album_name: str | None, new_genre: str | None, new_year: str | None, track_status: str | None, original_artist: str | None, original_album: str | None, original_track: str | None) bool
}
class TrackUpdater {
config
console_logger
error_logger
retry_handler
track_processor
record_successful_updates(tracks: list[TrackDict], year: str, artist: str, album: str, updated_tracks: list[TrackDict], changes_log: list[ChangeLogEntry]) None
update_album_tracks_bulk_async(tracks: list[TrackDict], year: str, artist: str, album: str) tuple[int, int]
update_tracks_for_album(artist: str, album: str, album_tracks: list[TrackDict], year: str, updated_tracks: list[TrackDict], changes_log: list[ChangeLogEntry]) None
}
class YearBatchProcessor {
analytics
config
console_logger
dry_run : bool
error_logger
year_determinator
get_dry_run_actions() list[dict[str, Any]]
group_tracks_by_album(tracks: list[TrackDict]) dict[tuple[str, str], list[TrackDict]]
process_albums_in_batches(grouped_albums: dict[tuple[str, str], list[TrackDict]], updated_tracks: list[TrackDict], changes_log: list[ChangeLogEntry], force: bool) None
update_album_tracks_bulk_async(tracks: list[TrackDict], year: str, artist: str, album: str) tuple[int, int]
}
class YearConsistencyChecker {
console_logger
dominance_min_share : float
parity_threshold : int
suspicion_threshold_years : int
top_years_count : int
get_consensus_release_year(tracks: list[TrackDict]) str | None
get_dominant_year(tracks: list[TrackDict]) str | None
get_earliest_track_added_year(tracks: list[TrackDict]) int | None
get_most_common_year(tracks: list[TrackDict]) str | None
}
class YearDeterminator {
cache_service
config
consistency_checker
console_logger
error_logger
external_api
fallback_handler
future_year_threshold
pending_verification
prerelease_recheck_days
skip_prerelease
check_prerelease_status(artist: str, album: str, album_tracks: list[TrackDict]) bool
check_suspicious_album(artist: str, album: str, album_tracks: list[TrackDict]) bool
determine_album_year(artist: str, album: str, album_tracks: list[TrackDict], force: bool) str | None
extract_future_years(album_tracks: list[TrackDict]) list[int]
extract_release_years(album_tracks: list[TrackDict]) list[str]
handle_future_years(artist: str, album: str, album_tracks: list[TrackDict], future_years: list[int]) bool
should_skip_album(album_tracks: list[TrackDict], artist: str, album: str, force: bool) tuple[bool, str]
}
class YearFallbackHandler {
absurd_year_threshold
api_orchestrator : NoneType
console_logger
fallback_enabled
min_confidence_for_new_year : int
pending_verification
trust_api_score_threshold : int
year_difference_threshold
apply_year_fallback(proposed_year: str, album_tracks: list[TrackDict], is_definitive: bool, confidence_score: int, artist: str, album: str, year_scores: dict[str, int] | None, release_year: str | None) str | None
get_existing_year_from_tracks(tracks: list[TrackDict]) str | None
is_year_change_dramatic(existing: str, proposed: str) bool
}
class YearRetriever {
DEFAULT_ABSURD_YEAR_THRESHOLD : int
DEFAULT_FALLBACK_ENABLED : bool
DEFAULT_SUSPICION_THRESHOLD_YEARS : int
DEFAULT_YEAR_DIFFERENCE_THRESHOLD : int
DOMINANCE_MIN_SHARE : float
MAX_RETRY_DELAY_SECONDS : float
MIN_VALID_YEAR : int
PARITY_THRESHOLD : int
SUSPICIOUS_ALBUM_MIN_LEN : int
SUSPICIOUS_MANY_YEARS : int
TOP_YEARS_COUNT : int
absurd_year_threshold
analytics : AnalyticsProtocol
cache_service : CacheServiceProtocol
config : AppConfig
console_logger : Logger
dry_run : bool
error_logger : Logger
external_api : ExternalApiServiceProtocol
fallback_enabled
min_confidence_for_new_year : int
normalize_collaboration_artist : staticmethod
pending_verification : PendingVerificationServiceProtocol
suspicion_threshold_years
track_processor
year_consistency_checker
year_difference_threshold
year_fallback_handler
get_album_years_with_logs(tracks: list[TrackDict], force: bool) tuple[list[TrackDict], list[ChangeLogEntry]]
get_dry_run_actions() list[dict[str, Any]]
get_last_updated_tracks() list[TrackDict]
process_album_years(tracks: list[TrackDict], force: bool, fresh: bool) bool
set_last_updated_tracks(tracks: list[TrackDict]) None
update_album_tracks_bulk_async(tracks: list[TrackDict], year: str, artist: str, album: str) tuple[int, int]
update_years_from_discogs(tracks: list[TrackDict], force: bool) tuple[list[TrackDict], list[ChangeLogEntry]]
}
GenreManager --|> BaseProcessor
IncrementalFilterService --|> BaseProcessor
BatchTrackFetcher --* TrackProcessor : batch_fetcher
TrackCacheManager --* TrackProcessor : cache_manager
TrackUpdateExecutor --* TrackProcessor : update_executor
YearConsistencyChecker --* YearRetriever : year_consistency_checker
YearFallbackHandler --* YearRetriever : year_fallback_handler
ArtistRenamer --o TrackProcessor : artist_renamer
TrackProcessor --o GenreManager : track_processor
TrackProcessor --o YearRetriever : track_processorData Models¶
classDiagram
class AlbumCacheEntry {
album : str
artist : str
confidence : int
timestamp : float
year : str
}
class AlbumType {
name
}
class AlbumTypeDetectionConfig {
compilation_patterns : list[str] | None
reissue_patterns : list[str] | None
soundtrack_patterns : list[str] | None
special_patterns : list[str] | None
various_artists_names : list[str] | None
}
class AlbumTypeInfo {
album_type
detected_pattern : str | None
strategy
}
class AlbumTypePatterns {
compilation : frozenset[str]
reissue : frozenset[str]
special : frozenset[str]
from_config(config: AppConfig) AlbumTypePatterns
from_defaults() AlbumTypePatterns
}
class AnalyticsConfig {
compact_time : bool
duration_thresholds
enable_gc_collect : bool
enabled : bool
max_events : Optional[int]
time_format : str
}
class AnalyticsProtocol {
batch_mode(message: str, console: Console | None) AbstractAsyncContextManager[Status]
execute_async_wrapped_call(func: Callable[..., Any], event_type: str) Any
execute_sync_wrapped_call(func: Callable[..., Any], event_type: str) Any
}
class ApiAuthConfig {
contact_email : str
discogs_token : str
musicbrainz_app_name : str
}
class AppConfig {
album_type_detection : Optional[AlbumTypeDetectionConfig]
album_years_cache_file : str
analytics
api_cache_file : str
apple_script_concurrency : Optional[int]
apple_script_rate_limit : Optional[AppleScriptRateLimitConfig]
apple_scripts_dir : str
applescript_retry : Optional[AppleScriptRetryConfig]
applescript_timeout_seconds : Optional[int]
applescript_timeouts : Optional[AppleScriptTimeoutsConfig]
artist_renamer : Optional[ArtistRenamerConfig]
batch_processing : Optional[BatchProcessingConfig]
cache_ttl_seconds : Optional[int]
caching : Optional[CachingConfig]
cleaning
database_verification
development
dry_run : bool
exceptions
experimental : Optional[ExperimentalConfig]
genre_update
incremental_interval_minutes : Optional[int]
logging
logs_base_dir : str
max_generic_entries : Optional[int]
max_retries : Optional[int]
music_library_path : str
pending_verification : Optional[PendingVerificationConfig]
python_settings
reporting : Optional[ReportingConfig]
retry_delay_seconds : Optional[float]
test_artists : Optional[list[str]]
year_retrieval
migrate_legacy_test_artists() AppConfig
}
class AppleScriptClientProtocol {
apple_scripts_dir : str | None
fetch_all_track_ids(timeout: float | None) list[str]
fetch_tracks_by_ids(track_ids: list[str], batch_size: int, timeout: float | None) list[dict[str, str]]
initialize() None
run_script(script_name: str, arguments: list[str] | None, timeout: float | None, context_artist: str | None, context_album: str | None, context_track: str | None, label: str | None) str | None
}
class AppleScriptFieldIndex {
name
}
class AppleScriptRateLimitConfig {
enabled : bool
requests_per_window : Optional[int]
window_size_seconds : Optional[float]
}
class AppleScriptRetryConfig {
base_delay_seconds : Optional[float]
jitter_range : Optional[float]
max_delay_seconds : Optional[float]
max_retries : Optional[int]
operation_timeout_seconds : Optional[float]
}
class AppleScriptTimeoutsConfig {
batch_update : Optional[int]
default : Optional[int]
full_library_fetch : Optional[int]
ids_batch_fetch : Optional[int]
single_artist_fetch : Optional[int]
}
class ArtistRenamerConfig {
config_path : str
}
class BatchProcessingConfig {
batch_size : Optional[int]
ids_batch_size : Optional[int]
}
class CacheServiceProtocol {
clear() None
generate_album_key(artist: str, album: str) str
get_album_year_entry_from_cache(artist: str, album: str) AlbumCacheEntry | None
get_album_year_from_cache(artist: str, album: str) str | None
get_async(key_data: Literal['ALL'], compute_func: None) list[TrackDict]
get_cached_api_result(artist: str, album: str, source: str) CachedApiResult | None
get_last_run_timestamp() datetime
initialize() None
invalidate(key_data: CacheableKey) None
invalidate_album_cache(artist: str, album: str) None
invalidate_all_albums() None
invalidate_for_track(track: TrackDict) None
load_cache() None
save_cache() None
set(key_data: CacheableKey, value: CacheableValue, ttl: int | None) None
set_async(key_data: CacheableKey, value: CacheableValue, ttl: int | None) None
set_cached_api_result(artist: str, album: str, source: str, year: str | None) None
store_album_year_in_cache(artist: str, album: str, year: str, confidence: int) None
sync_cache() None
}
class CachedApiResult {
album : str
api_response : dict[str, Any] | None
artist : str
metadata : Optional[dict[str, Any]]
source : str
timestamp : float
ttl : int | None
year : str | None
model_dump() dict[str, Any]
}
class CachingConfig {
album_cache_sync_interval : Optional[int]
api_result_cache_path : str
cleanup_error_retry_delay : Optional[int]
cleanup_interval_seconds : Optional[int]
default_ttl_seconds : Optional[int]
library_snapshot : Optional[LibrarySnapshotConfig]
negative_result_ttl : Optional[float]
}
class ChangeDisplayMode {
name
}
class ChangeLogEntry {
album_name : str
artist : str
change_type : str
new_album_name : str | None
new_genre : str | None
new_track_name : str | None
old_album_name : str | None
old_genre : str | None
old_track_name : str | None
timestamp : str
track_id : str
track_name : str
year_before_mgu : str | None
year_set_by_mgu : str | None
}
class CleaningConfig {
album_suffixes_to_remove : list[str]
remaster_keywords : list[str]
}
class CodeAction {
code : str
}
class CodeActionExtended {
args : Optional[list[str]]
code : str
type : Literal['code']
}
class DatabaseVerificationConfig {
auto_verify_days : Optional[int]
batch_size : Optional[int]
}
class DevelopmentConfig {
debug_mode : bool
test_artists : list[str]
parse_test_artists(v: str | list[str] | tuple[str, ...]) list[str]
}
class DiscogsArtist {
id : int
name : str
resource_url : str | None
}
class DurationThresholds {
long_max : Optional[float]
medium_max : Optional[float]
short_max : Optional[float]
}
class ExceptionsConfig {
track_cleaning : list[TrackCleaningException]
}
class ExperimentalConfig {
batch_updates_enabled : bool
max_batch_size : Optional[int]
}
class ExternalApiServiceProtocol {
close() None
get_album_year(artist: str, album: str, current_library_year: str | None, earliest_track_added_year: int | None) tuple[str | None, bool, int, dict[str, int]]
get_artist_activity_period(artist_norm: str) tuple[int | None, int | None]
get_artist_start_year(artist_norm: str) int | None
get_year_from_discogs(artist: str, album: str) str | None
initialize(force: bool) None
}
class FallbackConfig {
enabled : bool
trust_api_score_threshold : Optional[float]
year_difference_threshold : Optional[int]
}
class GenreUpdateConfig {
batch_size : Optional[int]
concurrent_limit : Optional[int]
}
class LibraryCacheMetadata {
last_force_scan_time : str | None
last_full_scan : datetime
library_mtime : datetime
snapshot_hash : str
track_count : int
version : str
from_dict(data: dict[str, Any]) LibraryCacheMetadata
to_dict() dict[str, Any]
}
class LibraryDeltaCache {
field_hashes : dict[str, str]
last_run : datetime
processed_track_ids : set[str]
tracked_since : datetime | None
add_processed_ids(track_ids: Iterable[str]) None
from_dict(data: dict[str, Any]) LibraryDeltaCache
should_reset() bool
to_dict() dict[str, Any]
}
class LibrarySnapshotConfig {
cache_file : str
compress : bool
compress_level : Optional[int]
delta_enabled : bool
enabled : bool
max_age_hours : Optional[int]
}
class LibrarySnapshotServiceProtocol {
get_library_mtime() datetime
get_snapshot_metadata() LibraryCacheMetadata | None
is_delta_enabled() bool
is_enabled() bool
is_snapshot_valid() bool
load_delta() LibraryDeltaCache | None
load_snapshot() list[TrackDict] | None
save_delta(delta: LibraryDeltaCache) None
save_snapshot(tracks: Sequence[TrackDict]) str
update_snapshot_metadata(metadata: LibraryCacheMetadata) None
}
class LogLevel {
name
}
class LogLevelsConfig {
analytics_file
console
main_file
normalize_log_level(value: str) str
}
class LoggingConfig {
analytics_log_file : str
changes_report_file : str
csv_output_file : str
dry_run_report_file : str
last_db_verify_log : str
last_incremental_run_file : str
levels
main_log_file : str
max_runs : Optional[int]
pending_verification_file : str
}
class LogicConfig {
absurd_year_threshold : Optional[int]
definitive_score_diff : Optional[int]
definitive_score_threshold : Optional[int]
major_market_codes : list[str]
min_confidence_for_new_year : Optional[float]
min_valid_year : Optional[int]
preferred_countries : list[str]
suspicion_threshold_years : Optional[int]
}
class MBArtist {
disambiguation : str | None
id : str
name : str
sort_name : str | None
type : str | None
}
class MBArtistCredit {
artist
joinphrase : str | None
name : str | None
}
class MusicBrainzArtist {
disambiguation : str | None
id : str
name : str
sort_name : str | None
}
class MusicBrainzRelease {
artist_credit : Optional[list[MBArtistCredit]]
country : str | None
date : str | None
id : str
release_group : MusicBrainzReleaseGroup | None
status : str | None
title : str
}
class MusicBrainzReleaseGroup {
first_release_date : str | None
id : str
primary_type : str | None
secondary_types : Optional[list[str]]
title : str
}
class MusicBrainzSearchResult {
count : int
offset : int
releases : Optional[list[MusicBrainzRelease]]
}
class PendingAlbumEntry {
album : str
artist : str
attempt_count : int
metadata : str
reason
timestamp : datetime
}
class PendingVerificationConfig {
auto_verify_days : Optional[int]
}
class PendingVerificationServiceProtocol {
generate_problematic_albums_report(min_attempts: int) int
get_all_pending_albums() list[PendingAlbumEntry]
get_attempt_count(artist: str, album: str) int
get_entry(artist: str, album: str) PendingAlbumEntry | None
initialize() None
is_verification_needed(artist: str, album: str) bool
mark_for_verification(artist: str, album: str, reason: str, metadata: dict[str, Any] | None, recheck_days: int | None) None
remove_from_pending(artist: str, album: str) None
should_auto_verify() bool
update_verification_timestamp() None
}
class PreferredApi {
name
}
class ProcessingConfig {
adaptive_delay : bool
batch_size : Optional[int]
cache_ttl_days : Optional[int]
delay_between_batches : Optional[float]
future_year_threshold : Optional[int]
pending_verification_interval_days : Optional[int]
prerelease_handling : str
prerelease_recheck_days : Optional[int]
skip_prerelease : bool
}
class PythonSettings {
prevent_bytecode : bool
}
class RateLimiterProtocol {
acquire() float
get_stats() dict[str, Any]
release() None
}
class RateLimitsConfig {
concurrent_api_calls : Optional[int]
discogs_requests_per_minute : Optional[int]
musicbrainz_requests_per_second : Optional[float]
}
class ReissueDetectionConfig {
reissue_keywords : list[str]
}
class ReportingConfig {
change_display_mode
min_attempts_for_report : Optional[float]
problematic_albums_path : str
}
class RevertTarget {
album : str | None
track_id : str | None
track_name : str
year_before_mgu : str
}
class ScoringConfig {
album_exact_match_bonus : int
album_substring_penalty : Optional[int]
album_unrelated_penalty : Optional[int]
album_variation_bonus : int
artist_cross_script_penalty : Optional[int]
artist_exact_match_bonus : int
artist_mismatch_penalty : Optional[int]
artist_substring_penalty : Optional[int]
base_score : int
country_artist_match_bonus : int
country_major_market_bonus : int
current_year_penalty : int
future_year_penalty : Optional[int]
mb_release_group_match_bonus : int
perfect_match_bonus : int
reissue_penalty : Optional[int]
soundtrack_compensation_bonus : int
source_discogs_bonus : int
source_itunes_bonus : int
source_mb_bonus : int
status_bootleg_penalty : Optional[int]
status_official_bonus : int
status_promo_penalty : Optional[int]
type_album_bonus : int
type_compilation_live_penalty : Optional[int]
type_ep_single_penalty : Optional[int]
year_after_end_penalty : Optional[int]
year_before_start_penalty : Optional[int]
year_diff_max_penalty : Optional[int]
year_diff_penalty_scale : Optional[int]
year_near_start_bonus : int
}
class ScriptAction {
args : list[Any] | None
script : str
}
class ScriptActionExtended {
args : Optional[list[str]]
content : str
script_path : str
type : Literal['script']
}
class ScriptApiPriority {
fallback : Optional[list[str]]
primary : list[str]
}
class ScriptType {
name
}
class SearchStrategy {
name
}
class SearchStrategyInfo {
detected_pattern : str | None
modified_album : str | None
modified_artist : str | None
strategy
}
class SecurityValidationError {
dangerous_pattern : str | None
field : str | None
}
class SecurityValidator {
logger
sanitize_string(value: str, field_name: str | None) str
validate_api_input(data: dict[str, Any], max_depth: int, _current_depth: int) dict[str, Any]
validate_file_path(file_path: str, allowed_base_paths: list[str] | None) str
validate_track_data(track_data: dict[str, Any]) dict[str, Any]
}
class SupportsDictConversion {
model_dump() dict[str, Any]
}
class TrackCleaningException {
album : str
artist : str
}
class TrackDict {
album : str
album_artist : str | None
artist : str
date_added : str | None
genre : str | None
id : str
last_modified : str | None
model_config : ConfigDict
name : str
original_album : str | None
original_artist : str | None
original_pos : int | None
release_year : str | None
track_status : str | None
year : str | None
year_before_mgu : str | None
year_set_by_mgu : str | None
copy() TrackDict
get(key: str, default: TrackFieldValue) TrackFieldValue
}
class TrackProcessorProtocol {
update_artist_async(track: TrackDict, new_artist_name: str) bool
}
class VerificationReason {
name
from_string(value: str) VerificationReason
}
class YearHandlingStrategy {
name
}
class YearRetrievalConfig {
api_auth
enabled : bool
fallback : Optional[FallbackConfig]
logic
preferred_api
processing
rate_limits
reissue_detection
scoring
script_api_priorities : Optional[dict[str, ScriptApiPriority]]
}
class YearRevertProcessorProtocol {
fetch_tracks_async(artist: str | None, force_refresh: bool, dry_run_test_tracks: list[TrackDict] | None, ignore_test_filter: bool) list[TrackDict]
update_track_async(track_id: str, new_track_name: str | None, new_album_name: str | None, new_genre: str | None, new_year: str | None, track_status: str | None, original_artist: str | None, original_album: str | None, original_track: str | None) bool
}
AlbumTypeInfo --> AlbumType : album_type
AlbumTypeInfo --> YearHandlingStrategy : strategy
PendingAlbumEntry --> VerificationReason : reason
SearchStrategyInfo --> SearchStrategy : strategy
AppConfig --> AnalyticsConfig : analytics
YearRetrievalConfig --> ApiAuthConfig : api_auth
ReportingConfig --> ChangeDisplayMode : change_display_mode
AppConfig --> CleaningConfig : cleaning
AppConfig --> DatabaseVerificationConfig : database_verification
AppConfig --> DevelopmentConfig : development
AnalyticsConfig --> DurationThresholds : duration_thresholds
AppConfig --> ExceptionsConfig : exceptions
AppConfig --> GenreUpdateConfig : genre_update
LogLevelsConfig --> LogLevel : console
LogLevelsConfig --> LogLevel : main_file
LogLevelsConfig --> LogLevel : analytics_file
LoggingConfig --> LogLevelsConfig : levels
AppConfig --> LoggingConfig : logging
YearRetrievalConfig --> LogicConfig : logic
MBArtistCredit --> MBArtist : artist
YearRetrievalConfig --> PreferredApi : preferred_api
YearRetrievalConfig --> ProcessingConfig : processing
AppConfig --> PythonSettings : python_settings
YearRetrievalConfig --> RateLimitsConfig : rate_limits
YearRetrievalConfig --> ReissueDetectionConfig : reissue_detection
YearRetrievalConfig --> ScoringConfig : scoring
AppConfig --> YearRetrievalConfig : year_retrievalAPI Clients¶
classDiagram
class Alias {
locale : str | None
name : str | None
primary : bool | None
sort_name : str | None
type : str | None
}
class ApiRateLimiter {
call_times : list[float]
lock : Lock
requests_per_window : int
total_requests : int
total_wait_time : float
window_seconds : float
acquire() float
get_stats() dict[str, Any]
release()* None
}
class ApiRequestExecutor {
api_call_durations : dict[str, list[float]]
cache_service
cache_ttl_days
console_logger
default_max_retries
default_retry_delay
discogs_token
error_logger
rate_limiters
request_counts : dict[str, int]
session : NoneType, aiohttp.ClientSession | None
user_agent
execute_request(api_name: str, url: str, params: dict[str, str] | None, headers_override: dict[str, str] | None, max_retries: int | None, base_delay: float | None, timeout_override: float | None) dict[str, Any] | None
set_session(session: aiohttp.ClientSession | None) None
}
class AppleMusicClient {
base_url : str
console_logger : Logger
country_code : str
entity : str
error_logger : Logger
limit
make_api_request_func : Callable[..., Coroutine[Any, Any, dict[str, Any] | None]]
score_release_func : Callable[..., float]
get_artist_start_year(artist_norm: str) int | None
get_scored_releases(artist_norm: str, album_norm: str) list[ScoredRelease]
}
class Area {
name : str | None
sort_name : str | None
type : str | None
}
class ArtistCredit {
artist : MBArtist
joinphrase : str | None
name : str | None
}
class ArtistPeriodContext {
end_year : int | None
start_year : int | None
}
class BaseApiClient {
compilation_pattern
console_logger : Logger
error_logger : Logger
}
class CoverArtArchive {
artwork : bool
back : bool
count : int
front : bool
}
class DiscogsClient {
analytics : Analytics
cache_service
cache_ttl_days : int
config
scoring_config
token : str
get_scored_releases(artist_norm: str, album_norm: str, artist_region: str | None) list[ScoredRelease]
get_year_from_discogs(artist: str, album: str) str | None
}
class DiscogsFormat {
descriptions : list[str]
name : str
qty : str | None
text : str | None
}
class DiscogsMasterRelease {
data_quality : str
genres : list[str]
id : int
main_release : int
main_release_url : str
resource_url : str
styles : list[str]
title : str
uri : str
year : int | None
}
class DiscogsRelease {
country : str | None
cover_image : str | None
formats : list[DiscogsFormat]
genre : list[str]
id : int | str
label : list[str]
master_id : int | None
master_url : str | None
released : str | None
resource_url : str
style : list[str]
thumb : str | None
title : str
type : str
uri : str
year : int | str | None
}
class DiscogsSearchResponse {
pagination : dict[str, Any]
results : list[DiscogsRelease]
}
class ExternalApiOrchestrator {
analytics : Analytics
api_call_durations : dict
applemusic_client
artist_period_context : ArtistPeriodContext | None
cache_service : CacheOrchestrator
cache_ttl_days
config : AppConfig
console_logger : Logger
contact_email : NoneType, str
current_year
default_api_max_retries
default_api_retry_delay
definitive_score_diff
definitive_score_threshold
discogs_client
discogs_token
error_logger : Logger
future_year_threshold
major_market_codes
min_valid_year
musicbrainz_app_name
musicbrainz_client
pending_verification_service : PendingVerificationService
preferred_api : str
prerelease_recheck_days
rate_limiters : dict
rate_limits_config
release_scorer
remaster_keywords : list[str]
request_counts : dict
request_executor
scoring_config
secure_config : NoneType, SecureConfig | None
session : NoneType, aiohttp.ClientSession | None
skip_prerelease
user_agent
year_score_resolver
year_search_coordinator
close() None
get_album_year(artist: str, album: str, current_library_year: str | None, earliest_track_added_year: int | None) tuple[str | None, bool, int, dict[str, int]]
get_artist_activity_period(artist_norm: str) tuple[int | None, int | None]
get_artist_start_year(artist_norm: str) int | None
get_year_from_discogs(artist: str, album: str) str | None
initialize(force: bool) None
}
class Label {
disambiguation : str | None
id : str
label_code : str | None
name : str
}
class LabelInfo {
catalog_number : str | None
label : Label | None
}
class LifeSpan {
begin : str | None
end : str | None
ended : bool | None
}
class Medium {
disc_count : int | None
format : str | None
position : int | None
title : str | None
track_count : int | None
tracks : list[Track] | None
}
class MusicBrainzClient {
analytics : Analytics
get_artist_activity_period(artist_norm: str) tuple[str | None, str | None]
get_artist_info(artist_norm: str, include_aliases: bool) dict[str, Any] | None
get_artist_region(artist_norm: str) str | None
get_canonical_artist_name(artist_norm: str) str | None
get_scored_releases(artist_norm: str, album_norm: str, artist_region: str | None) list[ScoredRelease]
}
class MusicBrainzReleasesResponse {
release_count : int
release_offset : int
releases : list[Release]
}
class Recording {
artist_credit : list[ArtistCredit] | None
disambiguation : str | None
id : str
length : int | None
title : str
}
class Release {
artist_credit : list[ArtistCredit] | None
barcode : str | None
country : str | None
cover_art_archive : CoverArtArchive | None
date : str | None
disambiguation : str | None
id : str
label_info : list[LabelInfo] | None
media : list[Medium] | None
packaging : str | None
release_events : list[ReleaseEvent] | None
release_group : dict[str, Any] | None
status : str | None
status_id : str | None
text_representation : TextRepresentation | None
title : str
year : int | None
}
class ReleaseEvent {
area : Area | None
date : str | None
}
class ReleaseGroup {
artist_credit : list[ArtistCredit] | None
disambiguation : str | None
first_release_date : str | None
id : str
primary_type : str | None
primary_type_id : str | None
releases : list[Release] | None
secondary_type_ids : list[str] | None
secondary_types : list[str] | None
title : str
}
class ReleaseScorer {
YEAR_LENGTH : int
artist_period_context : ArtistPeriodContext | None, NoneType
console_logger
current_year
definitive_score_threshold : int
major_market_codes : list
min_valid_year : int
remaster_keywords : list
scoring_config : ScoringConfig
clear_artist_period_context() None
score_original_release(release: dict[str, Any], artist_norm: str, album_norm: str) int
set_artist_period_context(context: ArtistPeriodContext | None) None
}
class ScoredRelease {
album_type : str | None
artist : str | None
barcode : str | None
catalog_number : str | None
country : str | None
disambiguation : str | None
format : str | None
is_reissue : NotRequired[bool]
label : str | None
score : float
source : str
status : str | None
title : str
year : str | None
}
class TextRepresentation {
language : str | None
script : str | None
}
class Track {
id : str
length : int | None
number : str | None
position : int | None
recording : Recording | None
title : str
}
class YearScoreResolver {
console_logger
current_year
definitive_score_diff
definitive_score_threshold
min_valid_year
remaster_keywords : list
aggregate_year_scores(all_releases: list[ScoredRelease]) defaultdict[str, list[int]]
select_best_year(year_scores: defaultdict[str, list[int]], all_releases: list[ScoredRelease] | None, existing_year: str | None) tuple[str, bool, int]
}
class YearSearchCoordinator {
applemusic_client
config
console_logger
discogs_client
error_logger
musicbrainz_client
preferred_api
release_scorer
fetch_all_api_results(artist_norm: str, album_norm: str, artist_region: str | None, log_artist: str, log_album: str) list[ScoredRelease]
}
class _RegionAwareApi {
get_scored_releases(artist_norm: str, album_norm: str, artist_region: str | None) list[ScoredRelease]
}
class _SimpleApi {
get_scored_releases(artist_norm: str, album_norm: str) list[ScoredRelease]
}
DiscogsClient --|> BaseApiClient
MusicBrainzClient --|> BaseApiClient
AppleMusicClient --* ExternalApiOrchestrator : applemusic_client
DiscogsClient --* ExternalApiOrchestrator : discogs_client
MusicBrainzClient --* ExternalApiOrchestrator : musicbrainz_client
ApiRequestExecutor --* ExternalApiOrchestrator : request_executor
YearScoreResolver --* ExternalApiOrchestrator : year_score_resolver
ArtistPeriodContext --* ExternalApiOrchestrator : artist_period_context
ReleaseScorer --* ExternalApiOrchestrator : release_scorer
YearSearchCoordinator --* ExternalApiOrchestrator : year_search_coordinatorCache Services¶
classDiagram
class AlbumCacheService {
album_years_cache : dict[str, AlbumCacheEntry]
album_years_cache_file : Path
cache_config
config : AppConfig
logger
policy
get_album_year(artist: str, album: str) str | None
get_album_year_entry(artist: str, album: str) AlbumCacheEntry | None
get_stats() dict[str, Any]
initialize() None
invalidate_album(artist: str, album: str) None
invalidate_all() None
save_to_disk() None
store_album_year(artist: str, album: str, year: str, confidence: int) None
}
class ApiCacheService {
api_cache : dict[str, CachedApiResult]
api_cache_file : Path
cache_config
config : AppConfig
event_manager
logger
cleanup_expired() int
emit_track_modified(track_id: str, old_artist: str, old_album: str) None
emit_track_removed(track_id: str, artist: str, album: str) None
get_cached_result(artist: str, album: str, source: str) CachedApiResult | None
get_stats() dict[str, Any]
initialize() None
invalidate_all() None
invalidate_for_album(artist: str, album: str) None
save_to_disk() None
set_cached_result(artist: str, album: str, source: str, success: bool, data: dict[str, Any] | None, metadata: dict[str, Any] | None) None
shutdown() None
}
class CacheContentType {
name
}
class CacheEvent {
event_type
fingerprint : str | None
metadata : dict[str, Any] | None
track_id : str | None
}
class CacheEventType {
name
}
class CacheOrchestrator {
album_service
album_years_cache : dict[str, tuple[str, str, str]]
api_cache : dict[str, Any]
api_service
cache : dict[str, Any]
config : AppConfig
config_manager : NoneType
generic_service
logger
clear() None
generate_album_key(artist: str, album: str) str
get(key_data: CacheableKey) CacheableValue | None
get_album_year(artist: str, album: str) str | None
get_album_year_entry_from_cache(artist: str, album: str) AlbumCacheEntry | None
get_album_year_from_cache(artist: str, album: str) str | None
get_async(key_data: CacheableKey, compute_func: Callable[[], asyncio.Future[CacheableValue]] | None) list[TrackDict] | CacheableValue
get_cached_api_result(artist: str, album: str, source: str) CachedApiResult | None
get_last_run_timestamp() datetime
initialize() None
invalidate(key_data: CacheableKey) None
invalidate_album_cache(artist: str, album: str) None
invalidate_all() None
invalidate_all_albums() None
invalidate_for_track(track: TrackDict) None
load_cache()* None
save_all_to_disk() None
save_cache() None
set(key_data: CacheableKey, value: CacheableValue, ttl: int | None) None
set_async(key_data: CacheableKey, value: CacheableValue, ttl: int | None) None
set_cached_api_result(artist: str, album: str, source: str, year: str | None) None
shutdown() None
store_album_year(artist: str, album: str, year: str, confidence: int) None
store_album_year_in_cache(artist: str, album: str, year: str, confidence: int) None
sync_cache() None
}
class CachePolicy {
cleanup_threshold : float
content_type
description : str
invalidation_strategy
max_size_mb : int | None
ttl_seconds : int
}
class EventDrivenCacheManager {
config
logger : NoneType, RootLogger
emit_event(event: CacheEvent) None
register_event_handler(event_type: CacheEventType, handler: Callable[[CacheEvent], None]) None
should_invalidate_for_event(content_type: CacheContentType, event: CacheEvent) bool
}
class FingerprintGenerationError {
track_data : dict[str, Any] | None
}
class FingerprintGenerator {
ENCODING : str
HASH_ALGORITHM : str
OPTIONAL_PROPERTIES : ClassVar[dict[str, Any]]
REQUIRED_PROPERTIES : ClassVar[set[str]]
logger
generate_track_fingerprint(track_data: dict[str, Any]) str
validate_fingerprint(fingerprint: Any) bool
}
class GenericCacheService {
cache : OrderedDict[str, tuple[CacheableValue, float]]
cache_config
cache_file : Path
config : AppConfig
default_ttl
logger
max_size : int
cleanup_expired() int
enforce_size_limits() int
get(key_data: CacheableKey) CacheableValue | None
get_stats() dict[str, Any]
initialize() None
invalidate(key_data: CacheableKey) bool
invalidate_all() None
save_to_disk() None
set(key_data: CacheableKey, value: CacheableValue, ttl: int | None) None
stop_cleanup_task() None
}
class InvalidationStrategy {
name
}
class LibrarySnapshotService {
compress
compress_level
config : AppConfig
delta_enabled
enabled
logger
max_age : timedelta
clear_snapshot() bool
compute_smart_delta(applescript_client: AppleScriptClientProtocol, force: bool) TrackDelta | None
compute_snapshot_hash(payload: Sequence[dict[str, Any]]) str
get_library_mtime() datetime
get_snapshot_metadata() LibraryCacheMetadata | None
initialize() None
is_delta_enabled() bool
is_enabled() bool
is_snapshot_valid() bool
load_delta() LibraryDeltaCache | None
load_snapshot() list[TrackDict] | None
save_delta(delta: LibraryDeltaCache) None
save_snapshot(tracks: Sequence[TrackDict]) str
should_force_scan(force_flag: bool) tuple[bool, str]
update_snapshot_metadata(metadata: LibraryCacheMetadata) None
}
class SmartCacheConfig {
DAY : int
DEFAULT_NEGATIVE_RESULT_TTL : int
HOUR : int
INFINITE_TTL : int
MINUTE : int
MONTH : int
WEEK : int
logger : NoneType, RootLogger
get_policy(content_type: CacheContentType) CachePolicy
get_ttl(content_type: CacheContentType) int
is_persistent_cache(content_type: CacheContentType) bool
should_use_event_invalidation(content_type: CacheContentType) bool
should_use_fingerprint_validation(content_type: CacheContentType) bool
}
class UnifiedHashService {
ALGORITHM : str
hash_album_key(artist: str, album: str) str
hash_api_key(artist: str, album: str, source: str) str
hash_generic_key(data: Any) str
hash_pending_key(track_id: str) str
}
CachePolicy --> CacheContentType : content_type
CacheEvent --> CacheEventType : event_type
CachePolicy --> InvalidationStrategy : invalidation_strategy
AlbumCacheService --* CacheOrchestrator : album_service
ApiCacheService --* CacheOrchestrator : api_service
EventDrivenCacheManager --* ApiCacheService : event_manager
SmartCacheConfig --* AlbumCacheService : cache_config
SmartCacheConfig --* ApiCacheService : cache_config
SmartCacheConfig --* GenericCacheService : cache_config
GenericCacheService --* CacheOrchestrator : generic_service
SmartCacheConfig --o EventDrivenCacheManager : configApple Music Integration¶
classDiagram
class AppleScriptClient {
analytics : Analytics
apple_scripts_dir
config : AppConfig
console_logger : NoneType, RootLogger
error_logger : NoneType, RootLogger
executor
file_validator
rate_limiter : AppleScriptRateLimiter | None
sanitizer
semaphore : asyncio.Semaphore | None
fetch_all_track_ids(timeout: float | None) list[str]
fetch_tracks_by_ids(track_ids: list[str], batch_size: int, timeout: float | None) list[dict[str, str]]
initialize() None
run_script(script_name: str, arguments: list[str] | None, timeout: float | None, context_artist: str | None, context_album: str | None, context_track: str | None, label: str | None) str | None
}
class AppleScriptExecutionError {
label : str
}
class AppleScriptExecutor {
apple_scripts_directory : str | None
console_logger : Logger
error_logger : Logger
rate_limiter : AppleScriptRateLimiter | None
retry_handler : DatabaseRetryHandler | None
semaphore : asyncio.Semaphore | None
cleanup_process(proc: asyncio.subprocess.Process, label: str) None
handle_subprocess_execution(cmd: list[str], label: str, timeout_seconds: float) str | None
log_script_success(label: str, script_result: str, elapsed: float) None
run_osascript(cmd: list[str], label: str, timeout_seconds: float) str | None
update_rate_limiter(rate_limiter: AppleScriptRateLimiter) None
update_semaphore(semaphore: asyncio.Semaphore) None
}
class AppleScriptFileValidator {
apple_scripts_directory : str | None
console_logger : Logger
error_logger : Logger
validate_script_file_access(script_path: str) bool
validate_script_path(script_path: str) bool
}
class AppleScriptRateLimiter {
logger
max_concurrent : int
request_timestamps : deque[float]
requests_per_window : int
semaphore : asyncio.Semaphore | None
total_requests : int
total_wait_time : float
window_seconds : float
acquire() float
get_stats() RateLimiterStats
initialize() None
release() None
}
class AppleScriptSanitizationError {
dangerous_pattern : str | None
}
class AppleScriptSanitizer {
logger
create_safe_command(script_code: str, arguments: list[str] | None) list[str]
sanitize_string(value: Any) str
validate_script_code(script_code: str | None, allow_music_app: bool) None
}
class RateLimiterStats {
avg_wait_time : float
current_calls_in_window : int
requests_per_window : int
total_requests : int
total_wait_time : float
window_seconds : float
}
AppleScriptExecutor --* AppleScriptClient : executor
AppleScriptFileValidator --* AppleScriptClient : file_validator
AppleScriptRateLimiter --* AppleScriptClient : rate_limiter
AppleScriptSanitizer --* AppleScriptClient : sanitizer
AppleScriptRateLimiter --o AppleScriptExecutor : rate_limiterApplication Layer¶
classDiagram
class BatchProcessor {
console_logger : Logger
error_logger : Logger
music_updater
print_summary(results: dict[str, list[str]], total: int) None
process_artists(artists: list[str], operation: str, force: bool) dict[str, list[str]]
process_from_file(file_path: str, operation: str, force: bool) dict[str, list[str]]
}
class CLI {
parser : ArgumentParser
parse_args(args: list[str] | None) argparse.Namespace
print_help() None
}
class Config {
config_path : str | None
resolved_path : str
load() AppConfig
}
class CryptographyError {
details : dict
message : str
}
class CryptographyManager {
key_file_path : Path
logger : Logger
decrypt_token(encrypted_token: str, key: str | None, passphrase: str | None) str
encrypt_token(token: str, key: str | None, passphrase: str | None) str
get_secure_config_status() dict[str, Any]
is_token_encrypted(token: str) bool
rotate_key(new_passphrase: str | None, backup_old_key: bool) None
}
class DatabaseVerifier {
analytics
ap_client : AppleScriptClientProtocol
config
console_logger : Logger
db_verify_logger : Logger
dry_run : bool
error_logger : Logger
can_run_incremental(force_run: bool) bool
get_dry_run_actions() list[DryRunAction]
should_auto_verify() bool
update_last_incremental_run() None
verify_and_clean_track_database(force: bool, apply_test_filter: bool) int
}
class DecryptionError {
}
class DryRunAction {
action : str
count : int
track_ids : list[str]
}
class EncryptionError {
}
class GenreUpdateService {
get_tracks_for_genre_update(artist: str | None) list[TrackDict] | None
run_update_genres(artist: str | None, force: bool) None
set_test_artists(test_artists: set[str] | None) None
}
class InvalidKeyError {
}
class InvalidTokenError {
}
class KeyGenerationError {
}
class MusicUpdater {
analytics
app_config
artist_renamer : ArtistRenamer
cleaning_service
console_logger
database_verifier
deps : DependencyContainer
dry_run_mode : str
dry_run_test_artists : set[str]
error_logger
genre_manager : GenreManager
genre_service
incremental_filter : IncrementalFilterService
snapshot_manager
track_processor : TrackProcessor
year_retriever : YearRetriever
year_service
run_clean_artist(artist: str) None
run_main_pipeline(force: bool, fresh: bool) None
run_restore_release_years(artist: str | None, album: str | None, threshold: int) None
run_revert_years(artist: str, album: str | None, backup_csv: str | None) None
run_update_genres(artist: str | None, force: bool) None
run_update_years(artist: str | None, force: bool, fresh: bool) None
run_verify_database(force: bool) None
run_verify_pending(_force: bool) None
set_dry_run_context(mode: str, test_artists: set[str]) None
}
class Orchestrator {
COMMANDS_BYPASSING_MUSIC_CHECK : ClassVar[set[str]]
config
console_logger
deps : DependencyContainer
error_logger
music_updater
run_command(args: argparse.Namespace) None
}
class PipelineSnapshotManager {
clear() None
get_snapshot() list[TrackDict] | None
merge_smart_delta(snapshot_tracks: list[TrackDict], delta: TrackDelta) list[TrackDict] | None
persist_to_disk() bool
reset() None
set_snapshot(tracks: list[TrackDict]) None
update_tracks(updated_tracks: Iterable[TrackDict]) None
}
class TrackCleaningService {
clean_all_metadata_with_logs(tracks: list[TrackDict]) list[ChangeLogEntry]
extract_and_clean_metadata(track: TrackDict) tuple[TrackFieldValue, str, TrackFieldValue, TrackFieldValue, str, str]
process_all_tracks(tracks: list[TrackDict], artist: str) tuple[list[TrackDict], list[ChangeLogEntry]]
process_single_track(track: TrackDict, artist_override: str | None) tuple[TrackDict | None, ChangeLogEntry | None]
}
class YearUpdateService {
get_tracks_for_year_update(artist: str | None) list[TrackDict] | None
run_restore_release_years(artist: str | None, album: str | None, threshold: int) None
run_revert_years(artist: str, album: str | None, backup_csv: str | None) None
run_update_years(artist: str | None, force: bool, fresh: bool) None
set_test_artists(test_artists: set[str] | None) None
update_all_years_with_logs(tracks: list[TrackDict], force: bool, fresh: bool) list[ChangeLogEntry]
}
DecryptionError --|> CryptographyError
EncryptionError --|> CryptographyError
InvalidKeyError --|> CryptographyError
InvalidTokenError --|> CryptographyError
KeyGenerationError --|> CryptographyError
DatabaseVerifier --* MusicUpdater : database_verifier
GenreUpdateService --* MusicUpdater : genre_service
MusicUpdater --* Orchestrator : music_updater
PipelineSnapshotManager --* MusicUpdater : snapshot_manager
TrackCleaningService --* MusicUpdater : cleaning_service
YearUpdateService --* MusicUpdater : year_service
MusicUpdater --o BatchProcessor : music_updaterMetrics & Analytics¶
classDiagram
class Analytics {
GC_COLLECTION_THRESHOLD : int
analytics_logger
call_counts : dict[str, int]
compact_time
config : AppConfig
console_logger
decorator_overhead : dict[str, float]
enable_gc_collect
enabled
error_logger
events : list[dict[str, Any]]
instance_id : int
long_max
max_events
medium_max
short_max
success_counts : dict[str, int]
time_format
batch_mode(message: str, console: Console | None) AsyncIterator[Status]
clear_old_events(days: int) int
execute_async_wrapped_call(func: Callable[..., Any], event_type: str) Any
execute_sync_wrapped_call(func: Callable[..., Any], event_type: str) Any
execute_wrapped_call(func: Callable[..., Any], event_type: str, is_async: bool) Any
generate_reports(force_mode: bool) None
get_stats(function_filter: str | list[str] | None) dict[str, Any]
log_summary() None
merge_with(other: Analytics) None
track(event_type: str) Callable[[Callable[..., Any]], Callable[..., Any]]
}
class CallInfo {
event_type : str
func_name : str
success : bool
}
class ChangeType {
GENRE : str
NAME : str
OTHER : str
YEAR : str
}
class Format {
ARROW : str
COL_WIDTH_10 : int
COL_WIDTH_30 : int
COL_WIDTH_38 : int
COL_WIDTH_40 : int
HEADER_ITEM_NAME : str
HEADER_ITEM_TYPE : str
HEADER_OLD_NEW : str
ITEM_TYPE_ALBUM : str
ITEM_TYPE_OTHER : str
ITEM_TYPE_TRACK : str
SEPARATOR_100 : int
SEPARATOR_80 : int
TRUNCATE_SUFFIX : str
}
class Key {
ALBUM : str
ARTIST : str
CHANGE_TYPE : str
NEW_ALBUM_NAME : str
NEW_GENRE : str
NEW_TRACK_NAME : str
OLD_ALBUM_NAME : str
OLD_GENRE : str
OLD_TRACK_NAME : str
TIMESTAMP : str
TRACK_NAME : str
YEAR_BEFORE_MGU : str
YEAR_SET_BY_MGU : str
}
class LoggerContainer {
analytics : Logger
console : Logger
error : Logger
}
class Misc {
CHANGES_REPORT_TYPE : str
DURATION_FIELD : str
EMOJI_CHANGE : str
EMOJI_REPORT : str
UNKNOWN : str
UNKNOWN_ALBUM : str
UNKNOWN_ARTIST : str
UNKNOWN_TRACK : str
}
class ParsedTrackFields {
date_added : str
last_modified : str
track_status : str
year : str
}
class TimingInfo {
duration : float
end : float
overhead : float
start : float
}