create_safe_string,
loose_compare_strings,
)
+from music_assistant.helpers.database import UNSET
from music_assistant.helpers.json import serialize_to_json
from music_assistant.models.music_provider import MusicProvider
"external_ids": serialize_to_json(item.external_ids),
"search_name": create_safe_string(item.name, True, True),
"search_sort_name": create_safe_string(item.sort_name or "", True, True),
+ "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
},
)
# update/set provider_mappings table
),
"search_name": create_safe_string(name, True, True),
"search_sort_name": create_safe_string(sort_name or "", True, True),
+ "timestamp_added": int(update.date_added.timestamp())
+ if update.date_added
+ else UNSET,
},
)
# update/set provider_mappings table
)
from music_assistant.controllers.media.base import MediaControllerBase
from music_assistant.helpers.compare import compare_artist, compare_strings, create_safe_string
+from music_assistant.helpers.database import UNSET
from music_assistant.helpers.json import serialize_to_json
if TYPE_CHECKING:
"metadata": serialize_to_json(item.metadata),
"search_name": create_safe_string(item.name, True, True),
"search_sort_name": create_safe_string(item.sort_name or "", True, True),
+ "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
},
)
# update/set provider_mappings table
"metadata": serialize_to_json(metadata),
"search_name": create_safe_string(name, True, True),
"search_sort_name": create_safe_string(sort_name or "", True, True),
+ "timestamp_added": int(update.date_added.timestamp())
+ if update.date_added
+ else UNSET,
},
)
self.logger.debug("updated %s in database: %s", update.name, db_id)
create_safe_string,
loose_compare_strings,
)
+from music_assistant.helpers.database import UNSET
from music_assistant.helpers.datetime import utc_timestamp
from music_assistant.helpers.json import serialize_to_json
from music_assistant.models.music_provider import MusicProvider
"duration": item.duration,
"search_name": create_safe_string(item.name, True, True),
"search_sort_name": create_safe_string(item.sort_name or "", True, True),
+ "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
},
)
# update/set provider_mappings table
"duration": update.duration if overwrite else cur_item.duration or update.duration,
"search_name": create_safe_string(name, True, True),
"search_sort_name": create_safe_string(sort_name or "", True, True),
+ "timestamp_added": int(update.date_added.timestamp())
+ if update.date_added
+ else UNSET,
},
)
# update/set provider_mappings table
from abc import ABCMeta, abstractmethod
from collections.abc import Iterable
from contextlib import suppress
+from datetime import datetime
from typing import TYPE_CHECKING, Any, TypeVar, cast, final
from music_assistant_models.enums import EventType, ExternalID, MediaType, ProviderFeature
db_row_dict["provider"] = "library"
db_row_dict["favorite"] = bool(db_row_dict["favorite"])
db_row_dict["item_id"] = str(db_row_dict["item_id"])
+ db_row_dict["date_added"] = datetime.fromtimestamp(
+ db_row_dict["timestamp_added"]
+ ).isoformat()
for key in JSON_KEYS:
if key not in db_row_dict:
from music_assistant.constants import DB_TABLE_PLAYLISTS
from music_assistant.helpers.compare import create_safe_string
+from music_assistant.helpers.database import UNSET
from music_assistant.helpers.json import serialize_to_json
from music_assistant.helpers.uri import create_uri, parse_uri
from music_assistant.helpers.util import guard_single_request
"external_ids": serialize_to_json(item.external_ids),
"search_name": create_safe_string(item.name, True, True),
"search_sort_name": create_safe_string(item.sort_name or "", True, True),
+ "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
},
)
# update/set provider_mappings table
),
"search_name": create_safe_string(name, True, True),
"search_sort_name": create_safe_string(sort_name or "", True, True),
+ "timestamp_added": int(update.date_added.timestamp())
+ if update.date_added
+ else UNSET,
},
)
# update/set provider_mappings table
create_safe_string,
loose_compare_strings,
)
+from music_assistant.helpers.database import UNSET
from music_assistant.helpers.json import serialize_to_json
from music_assistant.models.music_provider import MusicProvider
"total_episodes": item.total_episodes or 0,
"search_name": create_safe_string(item.name, True, True),
"search_sort_name": create_safe_string(item.sort_name or "", True, True),
+ "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
},
)
# update/set provider_mappings table
"total_episodes": cur_item.total_episodes or update.total_episodes or 0,
"search_name": create_safe_string(name, True, True),
"search_sort_name": create_safe_string(sort_name or "", True, True),
+ "timestamp_added": int(update.date_added.timestamp())
+ if update.date_added
+ else UNSET,
},
)
# update/set provider_mappings table
create_safe_string,
loose_compare_strings,
)
+from music_assistant.helpers.database import UNSET
from music_assistant.helpers.json import serialize_to_json
from music_assistant.models.music_provider import MusicProvider
"search_sort_name": create_safe_string(
item.sort_name if item.sort_name is not None else "", True, True
),
+ "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
},
)
# update/set provider_mappings table
),
"search_name": create_safe_string(name, True, True),
"search_sort_name": create_safe_string(sort_name or "", True, True),
+ "timestamp_added": int(update.date_added.timestamp())
+ if update.date_added
+ else UNSET,
},
)
# update/set provider_mappings table
create_safe_string,
loose_compare_strings,
)
+from music_assistant.helpers.database import UNSET
from music_assistant.helpers.json import serialize_to_json
from music_assistant.models.music_provider import MusicProvider
"metadata": serialize_to_json(item.metadata),
"search_name": create_safe_string(item.name, True, True),
"search_sort_name": create_safe_string(item.sort_name or "", True, True),
+ "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
},
)
# update/set provider_mappings table
),
"search_name": create_safe_string(name, True, True),
"search_sort_name": create_safe_string(sort_name or "", True, True),
+ "timestamp_added": int(update.date_added.timestamp())
+ if update.date_added
+ else UNSET,
},
)
# update/set provider_mappings table
LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.database")
+
+class _UnsetType:
+ """Sentinel value to indicate a field should use the database default."""
+
+ _instance: _UnsetType | None = None
+
+ def __new__(cls) -> _UnsetType:
+ """Create singleton instance."""
+ if cls._instance is None:
+ cls._instance = super().__new__(cls)
+ return cls._instance
+
+ def __repr__(self) -> str:
+ """Return string representation."""
+ return "UNSET"
+
+ def __bool__(self) -> bool:
+ """Return False for boolean context."""
+ return False
+
+
+UNSET: _UnsetType = _UnsetType()
+
ENABLE_DEBUG = os.environ.get("PYTHONDEVMODE") == "1"
allow_replace: bool = False,
) -> int:
"""Insert data in given table."""
+ # Filter out UNSET values so database defaults are used
+ values = {k: v for k, v in values.items() if v is not UNSET}
keys = tuple(values.keys())
if allow_replace:
sql_query = f"INSERT OR REPLACE INTO {table}({','.join(keys)})"
async def upsert(self, table: str, values: dict[str, Any]) -> None:
"""Upsert data in given table."""
+ # Filter out UNSET values so database defaults are used
+ values = {k: v for k, v in values.items() if v is not UNSET}
keys = tuple(values.keys())
sql_query = (
f"INSERT INTO {table}({','.join(keys)}) VALUES ({','.join(f':{x}' for x in keys)})"
values: dict[str, Any],
) -> Mapping[str, Any]:
"""Update record."""
+ # Filter out UNSET values so those fields are not updated
+ values = {k: v for k, v in values.items() if v is not UNSET}
keys = tuple(values.keys())
sql_query = f"UPDATE {table} SET {','.join(f'{x}=:{x}' for x in keys)} WHERE "
sql_query += " AND ".join(f"{x} = :{x}" for x in match)
library_item = await self.mass.music.artists.update_item_in_library(
library_item.item_id, prov_item
)
+ elif prov_item.date_added and library_item.date_added != prov_item.date_added:
+ # update date_added if it changed
+ library_item = await self.mass.music.artists.update_item_in_library(
+ library_item.item_id, prov_item
+ )
if not library_item.favorite and prov_item.favorite:
# existing library item not favorite but should be
await self.mass.music.artists.set_favorite(library_item.item_id, True)
library_item = await self.mass.music.albums.update_item_in_library(
library_item.item_id, prov_item
)
+ elif prov_item.date_added and library_item.date_added != prov_item.date_added:
+ # update date_added if it changed
+ library_item = await self.mass.music.albums.update_item_in_library(
+ library_item.item_id, prov_item
+ )
if not library_item.favorite and prov_item.favorite:
# existing library item not favorite but should be
await self.mass.music.albums.set_favorite(library_item.item_id, True)
library_item = await self.mass.music.audiobooks.update_item_in_library(
library_item.item_id, prov_item
)
+ elif prov_item.date_added and library_item.date_added != prov_item.date_added:
+ # update date_added if it changed
+ library_item = await self.mass.music.audiobooks.update_item_in_library(
+ library_item.item_id, prov_item
+ )
if not library_item.favorite and prov_item.favorite:
# existing library item not favorite but should be
await self.mass.music.audiobooks.set_favorite(library_item.item_id, True)
-
# check if resume_position_ms or fully_played changed
if (
prov_item.resume_position_ms is not None
library_item = await self.mass.music.playlists.update_item_in_library(
library_item.item_id, prov_item
)
+ elif prov_item.date_added and library_item.date_added != prov_item.date_added:
+ # update date_added if it changed
+ library_item = await self.mass.music.playlists.update_item_in_library(
+ library_item.item_id, prov_item
+ )
if not library_item.favorite and prov_item.favorite:
# existing library item not favorite but should be
await self.mass.music.playlists.set_favorite(library_item.item_id, True)
-
cur_db_ids.add(int(library_item.item_id))
await asyncio.sleep(0) # yield to eventloop
# optionally sync playlist tracks
library_item = await self.mass.music.tracks.update_item_in_library(
library_item.item_id, prov_item
)
+ elif prov_item.date_added and library_item.date_added != prov_item.date_added:
+ # update date_added if it changed
+ library_item = await self.mass.music.tracks.update_item_in_library(
+ library_item.item_id, prov_item
+ )
if not library_item.favorite and prov_item.favorite:
# existing library item not favorite but should be
await self.mass.music.tracks.set_favorite(library_item.item_id, True)
-
cur_db_ids.add(int(library_item.item_id))
await asyncio.sleep(0) # yield to eventloop
except MusicAssistantError as err:
library_item = await self.mass.music.podcasts.update_item_in_library(
library_item.item_id, prov_item
)
+ elif prov_item.date_added and library_item.date_added != prov_item.date_added:
+ # update date_added if it changed
+ library_item = await self.mass.music.podcasts.update_item_in_library(
+ library_item.item_id, prov_item
+ )
if not library_item.favorite and prov_item.favorite:
# existing library item not favorite but should be
await self.mass.music.podcasts.set_favorite(library_item.item_id, True)
-
cur_db_ids.add(int(library_item.item_id))
await asyncio.sleep(0) # yield to eventloop
library_item = await self.mass.music.radio.update_item_in_library(
library_item.item_id, prov_item
)
+ elif prov_item.date_added and library_item.date_added != prov_item.date_added:
+ # update date_added if it changed
+ library_item = await self.mass.music.radio.update_item_in_library(
+ library_item.item_id, prov_item
+ )
if not library_item.favorite and prov_item.favorite:
# existing library item not favorite but should be
await self.mass.music.radio.set_favorite(library_item.item_id, True)
-
cur_db_ids.add(int(library_item.item_id))
await asyncio.sleep(0) # yield to eventloop
import time
import urllib.parse
from collections.abc import AsyncGenerator, Iterator, Sequence
+from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
},
disc_number=tags.disc or 0,
track_number=tags.track or 0,
+ date_added=(
+ datetime.fromtimestamp(file_item.created_at, tz=UTC)
+ if file_item.created_at
+ else None
+ ),
)
if isrc_tags := tags.isrc:
# album
album = track.album = (
- await self._parse_album(track_path=file_item.relative_path, track_tags=tags)
+ await self._parse_album(
+ track_path=file_item.relative_path,
+ track_tags=tags,
+ track_created_at=file_item.created_at,
+ )
if tags.album
else None
)
)
return episode
- async def _parse_album(self, track_path: str, track_tags: AudioTags) -> Album:
- """Parse Album metadata from Track tags."""
+ async def _parse_album(
+ self, track_path: str, track_tags: AudioTags, track_created_at: int | None = None
+ ) -> Album:
+ """Parse Album metadata from Track tags.
+
+ :param track_path: Path to the track file.
+ :param track_tags: Audio tags from the track.
+ :param track_created_at: Creation timestamp of the track file (Unix epoch).
+ """
assert track_tags.album
# work out if we have an album and/or disc folder
# track_dir is the folder level where the tracks are located
in_library=True,
)
},
+ date_added=(
+ datetime.fromtimestamp(track_created_at, tz=UTC) if track_created_at else None
+ ),
)
if track_tags.barcode:
album.external_ids.add((ExternalID.BARCODE, track_tags.barcode))
- is_dir: Boolean if item is directory (not file).
- checksum: Checksum for this path (usually last modified time) None for dir.
- file_size : File size in number of bytes or None if unknown (or not a file).
+ - created_at: File creation timestamp (Unix epoch) or None for directories.
"""
filename: str
is_dir: bool
checksum: str | None = None
file_size: int | None = None
+ created_at: int | None = None # file creation timestamp (Unix epoch)
@property
def ext(self) -> str | None:
# This can raise OSError for files with invalid encoding (e.g., emojis on SMB mounts)
# Let the caller handle the exception
stat = entry.stat(follow_symlinks=False)
+ # st_birthtime is available on macOS/Windows, st_ctime on Linux
+ # (on Linux st_ctime is metadata change time, not creation time)
+ created_at = int(getattr(stat, "st_birthtime", stat.st_ctime))
return cls(
filename=entry.name,
relative_path=get_relative_path(base_path, entry.path),
is_dir=False,
checksum=str(int(stat.st_mtime)),
file_size=stat.st_size,
+ created_at=created_at,
)
projects / music-assistant-server.git / commitdiff