apple_music

apple_music

Apple Music connector -- reads tracks from the local Music.app via AppleScript.

Uses osascript subprocess calls to query Music.app on macOS. No API keys or network access required; everything stays local.

Requires macOS (sys.platform == "darwin") and the Music app to be available.

Classes

AppleMusicConnector

AppleMusicConnector()

Bases: BaseConnector

Read tracks from the local Music.app library via AppleScript.

Only works on macOS where osascript is available and Music.app is installed (ships with the OS).

Source code in src/openjarvis/connectors/apple_music.py

def __init__(self) -> None:
    self._status = SyncStatus()

Functions

is_connected

is_connected() -> bool

Return True if running on macOS and Music.app responds.

Source code in src/openjarvis/connectors/apple_music.py

def is_connected(self) -> bool:
    """Return True if running on macOS and Music.app responds."""
    if sys.platform != "darwin":
        return False
    result = _run_osascript(_LIBRARY_CHECK_SCRIPT, timeout=10)
    return result is not None

disconnect

disconnect() -> None

No-op -- local connector with no credentials to revoke.

Source code in src/openjarvis/connectors/apple_music.py

def disconnect(self) -> None:
    """No-op -- local connector with no credentials to revoke."""

sync

sync(*, since: Optional[datetime] = None, cursor: Optional[str] = None) -> Iterator[Document]

Query Music.app for all tracks and yield one Document per track.

PARAMETER	DESCRIPTION
since	If provided, only yield tracks whose last-played date is after this datetime. TYPE: Optional[datetime] DEFAULT: None
cursor	Not used for this local connector (included for API compatibility). TYPE: Optional[str] DEFAULT: None

YIELDS	DESCRIPTION
Document	One document per track in the Music library.

Source code in src/openjarvis/connectors/apple_music.py

def sync(
    self,
    *,
    since: Optional[datetime] = None,
    cursor: Optional[str] = None,  # noqa: ARG002
) -> Iterator[Document]:
    """Query Music.app for all tracks and yield one Document per track.

    Parameters
    ----------
    since:
        If provided, only yield tracks whose last-played date is after
        this datetime.
    cursor:
        Not used for this local connector (included for API
        compatibility).

    Yields
    ------
    Document
        One document per track in the Music library.
    """
    raw = _run_osascript(_TRACKS_SCRIPT)
    if raw is None:
        logger.warning("Could not retrieve tracks from Music.app")
        self._status.state = "error"
        self._status.error = "AppleScript query failed"
        return

    lines = [line for line in raw.split("\n") if line.strip()]
    self._status.items_total = len(lines)
    synced = 0

    for line in lines:
        parts = line.split("|||")
        if len(parts) < 7:
            logger.debug("Skipping malformed line: %r", line)
            continue

        name = parts[0].strip()
        artist = parts[1].strip()
        album = parts[2].strip()
        duration_raw = parts[3].strip()
        genre = parts[4].strip()
        play_count_raw = parts[5].strip()
        played_date_raw = parts[6].strip()

        # Parse numeric fields
        try:
            duration_s = round(float(duration_raw), 2)
        except (ValueError, TypeError):
            duration_s = 0.0
        try:
            play_count = int(play_count_raw)
        except (ValueError, TypeError):
            play_count = 0

        played_date = _parse_played_date(played_date_raw)
        timestamp = played_date if played_date else datetime.now()

        # Apply since filter
        if since is not None and played_date is not None:
            if played_date <= since:
                continue

        track_data = {
            "name": name,
            "artist": artist,
            "album": album,
            "duration_s": duration_s,
            "genre": genre,
            "play_count": play_count,
            "played_date": played_date_raw,
        }

        doc = Document(
            doc_id=_track_doc_id(name, artist),
            source="apple_music",
            doc_type="track",
            content=json.dumps(track_data),
            title=f"{name} \u2014 {artist}",
            author=artist,
            timestamp=timestamp,
            metadata={
                "album": album,
                "duration_s": duration_s,
                "genre": genre,
                "play_count": play_count,
            },
        )
        synced += 1
        yield doc

    self._status.items_synced = synced
    self._status.state = "idle"
    self._status.last_sync = datetime.now()
    self._status.error = None

sync_status

sync_status() -> SyncStatus

Return sync progress from the most recent :meth:sync call.

Source code in src/openjarvis/connectors/apple_music.py

def sync_status(self) -> SyncStatus:
    """Return sync progress from the most recent :meth:`sync` call."""
    return self._status