diff --git a/.github/workflows/schedule.yaml b/.github/workflows/schedule.yaml new file mode 100644 index 0000000..60b5068 --- /dev/null +++ b/.github/workflows/schedule.yaml @@ -0,0 +1,28 @@ +name: Scheduled tasks + +on: + schedule: + - cron: '13 12 * * *' + workflow_dispatch: + +jobs: + updatecli: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v3 + + - run: pipx install poetry + + - name: Setup Python + uses: actions/setup-python@v4 + with: + python-version: '3.x' + cache: 'poetry' + + - run: poetry install + + - name: Run Updatecli in Apply mode + run: poetry run pytest + env: + PYTEST_ADDOPTS: ${{ secrets.PYTEST_ADDOPTS }} diff --git a/README.md b/README.md index d79da83..10e313a 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,26 @@ poetry install poetry run pytest ``` +## Testing + +We run regular e2e acceptance tests on real data through pytest using our own ACRCloud account +as the data source. +These tests are scheduled to run at a regular interval in GitHub Actions with +[.github/workflows/schedule.yaml](./.github/workflows/schedule.yaml). + +If you would like to run the acceptance test suite locally using your own key, you can supply +a bearer token on the command line: + +```bash +poetry run pytest --acr-bearer-token="" +``` + +To configure this in CI we use and recommend setting this via `PYTEST_ADDOPTS` env: + +```bash +PYTEST_ADDOPTS='--acr-bearer-token=""' +``` + ## Release Management The CI/CD setup uses semantic commit messages following the [conventional commits standard](https://www.conventionalcommits.org/en/v1.0.0/). diff --git a/acrclient/client.py b/acrclient/client.py index aa766b7..39784e4 100644 --- a/acrclient/client.py +++ b/acrclient/client.py @@ -1,37 +1,120 @@ -import requests +from typing import Any, Optional +from requests import Session +from requests.adapters import HTTPAdapter, Retry +from requests.auth import AuthBase +from requests.models import PreparedRequest, Response -class Client: - """ACRCloud client to fetch metadata. +from .models import ( + GetBmCsProjectsResultsParams, + GetBmCsProjectsResultsResponse, + GetBmCsProjectsResultsResponseRecord, + ListBmCsProjectsParams, + ListBmCsProjectsResponse, + ListBmCsProjectsResponseRecord, + ListBmCsProjectsStreamsParams, + ListBmCsProjectsStreamsResponse, + ListBmCsProjectsStreamsResponseRecord, +) - Args: - bearer_token: The bearer token for ACRCloud. - """ - def __init__(self, bearer_token, base_url="https://eu-api-v2.acrcloud.com"): - self.base_url = base_url +class _Auth(AuthBase): # pylint: disable=too-few-public-methods + """Bearer token style auth for ACRCloud.""" + + def __init__(self, bearer_token: str) -> None: self.bearer_token = bearer_token - def request(self, path, headers=None, **kwargs): - """Fetch JSON data from ACRCloud API with set Access Key param.""" + def __call__(self, request: PreparedRequest) -> PreparedRequest: + request.headers["Authorization"] = f"Bearer {self.bearer_token}" + return request - url_params = { - **kwargs, - } - if not headers: - headers = {} - if self.bearer_token: - headers["Authorization"] = f"Bearer {self.bearer_token}" - - response = requests.get( - url=f"{self.base_url}{path}", params=url_params, headers=headers, timeout=10 + +class Client: + """ACRCloud client to fetch metadata.""" + def __init__( + self, + bearer_token: str, + base_url: str = "https://api-v2.acrcloud.com", + retries: int = 5, + backoff_factor: float = 0.1, + ): + """ + Parameters: + bearer_token: The bearer token for ACRCloud. + """ + self.base_url = base_url + self.auth = _Auth(bearer_token=bearer_token) + self._session = Session() + self._session.mount( + "https://", + HTTPAdapter( + max_retries=Retry( + total=retries, + backoff_factor=backoff_factor, + ) + ), ) + + def get( + self, + path: str, + params: Any = None, + **kwargs, + ) -> Response: + """Fetch JSON data from ACRCloud API with set Access Key param.""" + url = f"{self.base_url}{path}" + if not kwargs.get("timeout"): + kwargs["timeout"] = 60 + + # pylint: disable-next=missing-timeout + response = self._session.get(url=url, auth=self.auth, params=params, **kwargs) response.raise_for_status() + return response + + def json( + self, + path: str, + params: Any = None, + **kwargs, + ) -> Any: + response = self.get(path, params=params, **kwargs) return response.json() - def get_bm_cs_projects_results(self, project_id, stream_id, headers=None, **kwargs): - return self.request( + def list_bm_cs_projects( + self, + params: Optional[ListBmCsProjectsParams] = None, + **kwargs, + ) -> ListBmCsProjectsResponse: + data = self.json( + path="/api/bm-cs-projects", + params=params, + **kwargs, + ).get("data") + return [ListBmCsProjectsResponseRecord(**r) for r in data] + + def list_bm_cs_projects_streams( + self, + project_id: int, + params: Optional[ListBmCsProjectsStreamsParams] = None, + **kwargs, + ) -> ListBmCsProjectsStreamsResponse: + data = self.json( + path=f"/api/bm-cs-projects/{project_id}/streams", + params=params, + **kwargs, + ).get("data") + return [ListBmCsProjectsStreamsResponseRecord(**r) for r in data] + + def get_bm_cs_projects_results( + self, + project_id: int, + stream_id: str, + params: Optional[GetBmCsProjectsResultsParams] = None, + **kwargs, + ) -> GetBmCsProjectsResultsResponse: + data = self.json( path=f"/api/bm-cs-projects/{project_id}/streams/{stream_id}/results", - headers=headers, + params=params, **kwargs, ).get("data") + return [GetBmCsProjectsResultsResponseRecord(**r) for r in data] diff --git a/acrclient/models.py b/acrclient/models.py new file mode 100644 index 0000000..9c09465 --- /dev/null +++ b/acrclient/models.py @@ -0,0 +1,906 @@ +# pylint: disable=too-few-public-methods,too-many-locals,too-many-arguments,too-many-instance-attributes +from typing import Optional, TypedDict + + +class Bucket: + """Bucket with ACR data.""" + + # pylint: disable=redefined-builtin,invalid-name + def __init__( + self, + id: int, + uid: int, + name: str, + type: str, + state: int, + region: str, + metadata_template: str, + labels: list[str], + net_type: int, + tracker: int, + created_at: str, + updated_at: str, + num: int, + size: str | int, + access_permission: str, + *args, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.id: int = id + self.uid: int = uid + self.name: str = name + self.type: str = type + self.state: int = state + self.region: str = region + self.metadata_template: str = metadata_template + self.labels: list[str] = labels + self.net_type: int = net_type + self.tracker: int = tracker + self.created_at: str = created_at + self.updated_at: str = updated_at + self.num: int = num + self.size: str | int = size + self.access_permission: str = access_permission + + +class Artist: + """Artist for Music record.""" + + # pylint: disable=redefined-builtin,invalid-name + def __init__( + self, + name: str, + *args, + id: Optional[str | int | dict] = None, + seokey: Optional[str] = None, + isni: Optional[str] = None, + role: Optional[str] = None, + roles: Optional[list] = None, + langs: Optional[list] = None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.name: str = name + self.id: Optional[str | int | dict] = id + self.seokey: Optional[str] = seokey + self.isni: Optional[str] = isni + self.role: Optional[str] = role + self.roles: Optional[list] = roles + self.langs: Optional[list] = langs + + +class Album: + """Album for Music record.""" + + # pylint: disable=redefined-builtin,invalid-name + def __init__( + self, + *args, + id: Optional[str | int] = None, + name: Optional[str] = None, + image: Optional[str] = None, + langs: Optional[dict] = None, + cd_id: Optional[str] = None, + type: Optional[str] = None, + upc: Optional[str] = None, + release_date: Optional[str] = None, + **kwargs, + ) -> None: + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.id: Optional[str | int] = id + self.name: Optional[str] = name + self.image: Optional[str] = image + self.cd_id: Optional[str] = cd_id + self.type: Optional[str] = type + self.upc: Optional[str] = upc + self.release_date: Optional[str] = release_date + + self.langs: Optional[list[Lang]] = None + if langs: + self.langs = [Lang(*l) for l in langs] + + +class Contributors: + """Contributors for Music record.""" + + def __init__( + self, + *args, + composers: Optional[list[str]] = None, + lyricists: Optional[list[str]] = None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.composers: Optional[list[str]] = composers + self.lyricists: Optional[list[str]] = lyricists + + +class ExternalIds: + """External IDs for Music record.""" + + def __init__( + self, + *args, + isrc: Optional[str | list[str]] = None, + upc: Optional[str | list[str]] = None, + iswc: Optional[str] = None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.isrc: Optional[str | list[str]] = isrc + self.upc: Optional[str | list[str]] = upc + self.iswc: Optional[str] = iswc + + +class ExternalMetadataDeezer: + """External metadata from Deezer.""" + + def __init__( + self, + album: dict, + track: dict, + artists: list[dict], + *args, + **kwargs, + ) -> None: + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.album = album + self.track = track + self.artists = artists + + +class ExternalMetadataSpotify: + """External metadata from Spotify.""" + + def __init__( + self, + album: dict, + track: dict, + artists: list[dict], + *args, + **kwargs, + ) -> None: + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.album = album + self.track = track + self.artists = artists + + +class ExternalMetadataYoutube: + """External metadata from Youtube.""" + + def __init__( + self, + vid: str, + *args, + **kwargs, + ) -> None: + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.vid: str = vid + + +class ExternalMetadataMusicBrainz: + """MusicBrainz external metadata for Music record.""" + + def __init__( + self, + track, + *args, + **kwargs, + ) -> None: + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.track = track + + +class ExternalMetadataLyricFind: + """Lyricfind external metadata for Music record.""" + + def __init__( + self, + lfid: str, + *args, + **kwargs, + ) -> None: + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.lfid: str = lfid + + +class MusicStoryData: + """Musicstory data part for external metadata in Music record.""" + + # pylint: disable=redefined-builtin,invalid-name + def __init__( + self, + *args, + id, + **kwargs, + ) -> None: + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.id = id + + +class ExternalMetadataMusicStory: + """Musicstory external metadata for Music record.""" + + def __init__( + self, + track, + *args, + album=None, + release=None, + **kwargs, + ) -> None: + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.track: MusicStoryData = MusicStoryData(**track) + self.album: Optional[MusicStoryData] + if album: + self.album = MusicStoryData(**album) + self.release: Optional[MusicStoryData] = None + if release: + self.release = MusicStoryData(**release) + + +class ExternalMetadata: + """External metadata for Music record.""" + + def __init__( + self, + *args, + deezer: list = None, + spotify: list = None, + youtube: list = None, + musicstory: Optional[list | dict] = None, + lyricfind: Optional[list | dict] = None, + musicbrainz: Optional[list] = None, + **kwargs, + ) -> None: + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + + if deezer: + self.deezer = [ + deezer + if not isinstance(deezer, list) + else [ExternalMetadataDeezer(**d) for d in deezer] + ] + if spotify: + self.spotify = [ + spotify + if not isinstance(spotify, list) + else [ExternalMetadataSpotify(**s) for s in spotify] + ] + if youtube: + self.youtube = [ + youtube + if not isinstance(youtube, list) + else [ExternalMetadataYoutube(**y) for y in youtube] + ] + + self.musicstory: Optional[ + list[ExternalMetadataMusicStory] | ExternalMetadataMusicStory + ] = None + if musicstory: + self.musicstory = ( + ExternalMetadataMusicStory(**musicstory) + if not isinstance(musicstory, list) + else [ExternalMetadataMusicStory(**m) for m in musicstory] + ) + self.lyricfind: Optional[ + list[ExternalMetadataLyricFind] | ExternalMetadataLyricFind + ] = None + if lyricfind: + self.lyricfind = ( + ExternalMetadataLyricFind(**lyricfind) + if not isinstance(lyricfind, list) + else [ExternalMetadataLyricFind(**l) for l in lyricfind] + ) + self.musicbrianz: Optional[list[ExternalMetadataMusicBrainz | str]] = None + if musicbrainz: + self.musicbrainz = [ + mb if isinstance(mb, str) else ExternalMetadataMusicBrainz(**mb) + for mb in musicbrainz + ] + + +class Genre: + """Genre of Music record.""" + + # pylint: disable=redefined-builtin,invalid-name + def __init__( + self, + name: str, + *args, + id: Optional[int | str] = None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.name: str = name + self.id: Optional[int | str] = id + + +class WorkCreator: + """Creator of work in Music record.""" + + def __init__( + self, + ipi: int, + name: str, + *args, + affiliation: Optional[dict] = None, + last_name: Optional[str] = None, + role: Optional[str] = None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.ipi: int = ipi + self.name: str = name + self.affiliation: Optional[dict] = affiliation + self.last_name: Optional[str] = last_name + self.role: Optional[str] = role + + +class WorkWorkAgency: + """Agency in work of work in Music record.""" + + def __init__( + self, + code: str, + *args, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.code: str = code + + +class WorkWork: + """Work part of work in Music record.""" + + def __init__( + self, + code: str, + agency, + *args, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.code: str = code + self.agency = WorkWorkAgency(**agency) + + +class Work: + """Work part of Music record.""" + + # pylint: disable=redefined-builtin,invalid-name + def __init__( + self, + id, + name: str, + iswc, + creators, + *args, + other_names=None, + works=None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.id = id + self.name = name + self.iswc = iswc + self.creators = [WorkCreator(**c) for c in creators] + + self.other_names = other_names + + self.works = None + if works: + self.works = [WorkWork(**w) for w in works] + + +class Lang: + """Lang part of Music record.""" + + def __init__( + self, + code: str, + name: str, + *args, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.code: str = code + self.name: str = name + + +class ReleaseByTerritories: + """ReleaseByTerritories part of Music record.""" + + def __init__( + self, + territories: list[str], + release_date: str, + *args, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.territories: list[str] = territories + self.release_date: str = release_date + + +class Music: + """ACR abstraction of a "Music" resource.""" + + def __init__( + self, + title: str, + acrid: str, + score: int, + result_from: int, + duration_ms: int, + sample_begin_time_offset_ms: int, + sample_end_time_offset_ms: int, + db_begin_time_offset_ms: int, + db_end_time_offset_ms: int, + play_offset_ms: int, + external_ids, + external_metadata, + album, + *args, + artists: Optional[list[dict]] = None, + release_date: Optional[str] = None, + label: Optional[str] = None, + language: Optional[str] = None, + exids: Optional[str] = None, + source: Optional[str] = None, + ppm: Optional[str] = None, + bpm: Optional[str | bool] = None, + rights_claim_policy: Optional[str] = None, + updated_at: Optional[str] = None, + langs=None, + contributors=None, + rights_claim=None, + lyrics=None, + genres=None, + works=None, + release_by_territories=None, + **kwargs, + ): + # remove fields that should not be in the top-level of a music record + # can we report these to ACRCloud as part of auto-remediation? + for field in [ + "163", + "amco", + "andyou", + "anghami", + "artstis", + "awa", + "base_score", + "bugs", + "deezer", + "disco_3453684", + "joox", + "kkbox", + "lyricfind", + "merlin", + "musicbrainz", + "musicstory", + "mwg", + "nct", + "omusic", + "partner", + "partners", + "qqmusic", + "rec_times", + "rights_owners", + "sme", + "Soundcharts", + "spotify", + "trackitdown", + "umg", + "wmg", + "works_ids", + "youtube", + ]: + kwargs.pop(field, None) + + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.title: str = title + self.acrid: str = acrid + + self.score: int = score + self.duration_ms: int = duration_ms + self.result_from: int = result_from + + self.sample_begin_time_offset_ms: int = sample_begin_time_offset_ms + self.sample_end_time_offset_ms: int = sample_end_time_offset_ms + self.db_begin_time_offset_ms: int = db_begin_time_offset_ms + self.db_end_time_offset_ms: int = db_end_time_offset_ms + self.play_offset_ms: int = play_offset_ms + + self.external_ids: ExternalIds = ExternalIds(**external_ids) + self.external_metadata: ExternalMetadata | list[ExternalMetadata] = ( + [ExternalMetadata(*e) for e in external_metadata] + if isinstance(external_metadata, list) + else ExternalMetadata(**external_metadata) + ) + self.album: Album = Album(**album) + + self.artists: list[Artist] = [Artist(**a) for a in artists] if artists else [] + self.release_date: Optional[str] = release_date + self.label: Optional[str] = label + self.language: Optional[str] = language + self.exids: Optional[str] = exids + self.source: Optional[str] = source + self.ppm: Optional[str] = ppm + self.bpm: Optional[str | bool] = bpm + self.rights_claim_policy: Optional[str] = rights_claim_policy + self.updated_at: Optional[str] = updated_at + + self.contributors: Optional[Contributors] = None + if contributors: + self.contributors = Contributors(**contributors) + + self.rights_claim: Optional[dict] = None + if rights_claim: + self.rights_claim = rights_claim + + self.lyrics: Optional[dict] = None + if lyrics: + self.lyrics = lyrics + + self.langs: Optional[list[Lang | str]] = None + if langs: + self.langs = [l if isinstance(l, str) else Lang(**l) for l in langs] + + self.genres: Optional[list[Genre]] = None + if genres: + self.genres = [Genre(**g) for g in genres] + + self.works: Optional[list[Work]] = None + if works: + self.works = [Work(**w) for w in works] + + self.release_by_territories: Optional[list] = None + if release_by_territories: + self.release_by_territories = [ + ReleaseByTerritories(**rbt) for rbt in release_by_territories + ] + + +class CustomFile: + """CustomFile record.""" + + # pylint: disable=redefined-builtin,invalid-name,redefined-outer-name + def __init__( + self, + title: str, + acrid: str, + bucket_id: str, + count: int, + score: int, + duration_ms: int, + sample_begin_time_offset_ms: int, + sample_end_time_offset_ms: int, + db_begin_time_offset_ms: int, + db_end_time_offset_ms: int, + play_offset_ms: int, + *args, + album: Optional[str] = None, + artists: Optional[str] = None, + isrc: Optional[str] = None, + label: Optional[str] = None, + release_date: Optional[str] = None, + Artist: Optional[str] = None, + Artists: Optional[str] = None, + Title: Optional[str] = None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.title: str = title + self.acrid: str = acrid + self.bucket_id: str = bucket_id + + self.count: int = count + self.score: int = score + self.duration_ms: int = duration_ms + + self.sample_begin_time_offset_ms: int = sample_begin_time_offset_ms + self.sample_end_time_offset_ms: int = sample_end_time_offset_ms + self.db_begin_time_offset_ms: int = db_begin_time_offset_ms + self.db_end_time_offset_ms: int = db_end_time_offset_ms + self.play_offset_ms: int = play_offset_ms + + self.album: Optional[str] = album + self.artists: Optional[str] = artists + self.isrc: Optional[str] = isrc + self.label: Optional[str] = label + self.release_date: Optional[str] = release_date + + self.Artist: Optional[str] = Artist + self.Artists: Optional[str] = Artists + self.Title: Optional[str] = Title + + +class ListBmCsProjectsParams(TypedDict): + """Parameters for listing base projects.""" + + region: str + """eu-west-1,us-west-2,ap-southeast-1""" + type: str + """AVR,LCD,HR""" + page: str + """Page number""" + per_page: str + """The results number per page""" + + +class BmCsProjectsConfig: + """Configuration in a ProjectsStreamsRecord.""" + + # pylint: disable=redefined-builtin,invalid-name + def __init__( + self, + record: dict, + *args, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.record = record + + +class ListBmCsProjectsResponseRecord: + """Project record in base proects response.""" + + # pylint: disable=redefined-builtin,invalid-name + def __init__( + self, + id: int, + name: str, + uid: int, + type: str, + region: str, + state: int, + access_key: str, + bucket_group: str, + noise: int, + created_at: str, + updated_at: str, + iswc: int, + buckets: list, + status_check: int, + external_ids: list[str], + metadata_template: str, + result_callback_url: str, + result_callback_send_type: str, + result_callback_send_noresult: str, + state_notification_url: str, + state_notification_email: str, + state_notification_email_frequency: str, + state_notification_email_custom_interval: int, + result_callback_retry: int, + monitoring_num: int, + config: dict, + *args, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.id: int = id + self.name: str = name + self.uid: int = uid + self.type: str = type + self.region: str = region + self.state: int = state + self.access_key: str = access_key + self.bucket_group: str = bucket_group + self.noise: int = noise + self.created_at: str = created_at + self.updated_at: str = updated_at + self.iswc: int = iswc + self.buckets: list[Bucket] = [Bucket(**b) for b in buckets] + self.status_check: int = status_check + self.external_ids: list = external_ids + self.metadata_template: str = metadata_template + self.result_callback_url: str = result_callback_url + self.result_callback_send_type: str = result_callback_send_type + self.result_callback_send_noresult: str = result_callback_send_noresult + self.state_notification_url: str = state_notification_url + self.state_notification_email: str = state_notification_email + self.state_notification_email_frequency: str = ( + state_notification_email_frequency + ) + self.state_notification_email_custom_interval: int = ( + state_notification_email_custom_interval + ) + self.result_callback_retry: int = result_callback_retry + self.monitoring_num: int = monitoring_num + self.config: BmCsProjectsConfig = BmCsProjectsConfig(**config) + + +ListBmCsProjectsResponse = list[ListBmCsProjectsResponseRecord] + + +class ListBmCsProjectsStreamsParams(TypedDict): + """Parameters for listing base projects.""" + + timemap: int + """0 or 1""" + state: str + """All,Running,Timeout,Paused,Invalid URL,Mute,Other. Default is All.""" + search_value: str + """Search by Name, StreamID, URL, User-defind, Remark""" + sort: str + """sort by 'created_at', 'stream_id', 'name', Default is 'created_at'""" + order: str + """order by desc or asc, default is desc.""" + + +class BmCsProjectsStreamsConfig: + """Configuration in a ProjectsStreamsRecord.""" + + # pylint: disable=redefined-builtin,invalid-name + def __init__( + self, + id: int, + name: str, + uid: int, + rec_length: int, + interval: int, + rec_timeout: int, + monitor_timeout: int, + noise: int, + delay: int, + record: dict, + created_at: str, + updated_at: str, + *args, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.id: int = id + self.name: str = name + self.uid: int = uid + self.rec_length: int = rec_length + self.interval: int = interval + self.rec_timeout: int = rec_timeout + self.monitor_timeout: int = monitor_timeout + self.noise: int = noise + self.delay: int = delay + self.record: dict = record + self.created_at: str = created_at + self.updated_at: str = updated_at + + +class ListBmCsProjectsStreamsResponseRecord: + """Project record in base proects response.""" + + def __init__( + self, + stream_id: str, + uid: int, + mcp_id: int, + stream_type: str, + name: str, + state: str, + code: int, + stream_urls: list[str], + current_url: str, + region: str, + pitch_shift: int, + check_pitch_shift: int, + remark: str, + created_at: str, + updated_at: str, + record_video: int, + stream_rec_type: int, + epg: str, + add_recordings: int, + config: dict, + timemap: int, + ucf: int, + *args, + user_defined: Optional[str] = None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.stream_id: str = stream_id + self.uid: int = uid + self.mcp_id: int = mcp_id + self.stream_type: str = stream_type + self.name: str = name + self.state: str = state + self.code: int = code + self.stream_urls: list[str] = stream_urls + self.current_url: str = current_url + self.region: str = region + self.pitch_shift: int = pitch_shift + self.check_pitch_shift: int = check_pitch_shift + self.remark: str = remark + self.created_at: str = created_at + self.updated_at: str = updated_at + self.record_video: int = record_video + self.stream_rec_type: int = stream_rec_type + self.epg: str = epg + self.add_recordings: int = add_recordings + self.timemap: int = timemap + self.ucf: int = ucf + self.user_defined: Optional[str] = user_defined + self.config: BmCsProjectsStreamsConfig = BmCsProjectsStreamsConfig(**config) + + +ListBmCsProjectsStreamsResponse = list[ListBmCsProjectsStreamsResponseRecord] + + +class GetBmCsProjectsResultsParams(TypedDict): + """Parameters for getting BM projects custom streams results.""" + + type: str + """last: get the last results, day: get the day results, Default is day""" + date: str + """Get all the results on this date. The format is YYYYmmdd (E.g. 20210201)""" + min_duration: int + """Return the results of played_duration >= min_duration seconds (default: 0)""" + max_duration: int + """Return the results with played_duration <= max_duration seconds (default: 3600)""" + isrc_country: str + """Only return results that match the isrc country code (E.g. DE, FR, IT, US)""" + + +class GetBmCsProjectsResultsResponseRecordStatus: + """Status part of a BM Project's custom stream result record.""" + + def __init__(self, msg: str, version: str, code: int, *args, **kwargs): + assert (len(args) + len(kwargs)) == 0 + self.msg: str = msg + self.version: str = version + self.code: int = code + + +class GetBmCsProjectsResultsResponseRecordMetadata: + """Metadata part of a BM Project's custom stream result record.""" + + # pylint: disable=redefined-builtin + def __init__( + self, + type: str, + timestamp_utc: str, + played_duration: int, + *args, + record_timestamp: Optional[str] = None, + music: Optional[list] = None, + custom_files: Optional[list] = None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.type: str = type + self.timestamp_utc: str = timestamp_utc + + self.played_duration: int = played_duration + + self.record_timestamp: Optional[str] = record_timestamp + + self.music: Optional[list[Music]] = None + if music: + self.music = [Music(**m) for m in music] + self.custom_files: Optional[list[CustomFile]] = None + if custom_files: + self.custom_files = [CustomFile(**m) for m in custom_files] + + +class GetBmCsProjectsResultsResponseRecord: + """Record in a BM Project's custom stream results response.""" + + def __init__( + self, + status: dict, + metadata: dict, + *args, + result_type: Optional[int] = None, + service_type: Optional[str] = None, + **kwargs, + ): + assert (len(args) + len(kwargs)) == 0, f"{self.__class__}: {args=} {kwargs=}" + self.status = GetBmCsProjectsResultsResponseRecordStatus(**status) + self.metadata = GetBmCsProjectsResultsResponseRecordMetadata(**metadata) + self.result_type: Optional[int] = result_type + self.service_type: Optional[str] = service_type + + +GetBmCsProjectsResultsResponse = list[GetBmCsProjectsResultsResponseRecord] diff --git a/acrclient/py.typed b/acrclient/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/pyproject.toml b/pyproject.toml index a7b5aac..c078812 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,12 +38,8 @@ pytest-random-order = "^1.1.0" [tool.isort] -line_length = 120 profile = "black" -[tool.pylint.format] -max-line-lenth = 120 - [tool.pylint.messages_control] # C0114 = missing-module-docstring # C0116 = missing-function-docstring diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..e6c98ca --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,51 @@ +from functools import reduce +from operator import concat + +from pytest import fixture, skip + +from acrclient import Client + + +def pytest_addoption(parser): + parser.addoption( + "--acr-bearer-token", + action="store", + default=None, + help="Bearer token for ACRCloud, used for acceptance testing.", + ) + + +@fixture(name="bearer_token", scope="session") +def bearer_token_fixture(request): + """ "ACRCloud API Key used for acceptance testing.""" + bearer_token = request.config.getoption("--acr-bearer-token") + if not bearer_token: + skip( + reason="provide an ACRCloud bearer token with" + + " --acr-bearer-token to run acceptance testing." + ) + return bearer_token + + +@fixture(name="client", scope="session") +def client_fixture(bearer_token): + return Client(bearer_token=bearer_token) + + +@fixture(name="project_ids", scope="session") +def project_ids_fixture(client): + return [p.id for p in client.list_bm_cs_projects()] + + +@fixture(name="project_stream_ids", scope="session") +def project_stream_ids_fixture(client, project_ids): + return reduce( + concat, + [ + [ + (p, s.stream_id, s.created_at) + for s in client.list_bm_cs_projects_streams(p) + ] + for p in project_ids + ], + ) diff --git a/tests/test_acceptance.py b/tests/test_acceptance.py new file mode 100644 index 0000000..d5313d9 --- /dev/null +++ b/tests/test_acceptance.py @@ -0,0 +1,326 @@ +from functools import reduce +from operator import concat +from datetime import datetime, timedelta + +from pytest import mark + +from acrclient import Client +from acrclient.models import ( + Album, + Artist, + BmCsProjectsConfig, + BmCsProjectsStreamsConfig, + Contributors, + ExternalIds, + ExternalMetadata, + Genre, + Lang, + Music, + Work, + WorkCreator, + WorkWork, + WorkWorkAgency, +) + + +def test_list_bm_cs_projects(client): + data = client.list_bm_cs_projects() + assert_is_list_bm_cs_projects(data) + + +def assert_is_list_bm_cs_projects(data): + assert isinstance(data, list) + assert len(data) > 0 + for record in data: + assert isinstance(record.id, int) + assert isinstance(record.name, str) + assert isinstance(record.uid, int) + assert isinstance(record.type, str) + assert isinstance(record.region, str) + assert isinstance(record.state, int) + assert isinstance(record.access_key, str) + assert isinstance(record.bucket_group, str) + assert isinstance(record.noise, int) + assert isinstance(record.created_at, str) + assert isinstance(record.updated_at, str) + assert isinstance(record.iswc, int) + assert isinstance(record.status_check, int) + assert isinstance(record.metadata_template, str) + assert isinstance(record.result_callback_url, str) + assert isinstance(record.result_callback_send_type, str) + assert isinstance(record.result_callback_send_noresult, str) + assert isinstance(record.state_notification_url, str) + assert isinstance(record.state_notification_email, str) + assert isinstance(record.state_notification_email_frequency, str) + assert isinstance(record.state_notification_email_custom_interval, int) + assert isinstance(record.result_callback_retry, int) + assert isinstance(record.monitoring_num, int) + + assert isinstance(record.config, BmCsProjectsConfig) + if record.config: + assert isinstance(record.config.record, dict) + + assert isinstance(record.external_ids, list) + for external_id in record.external_ids: + assert isinstance(external_id, str) + + assert isinstance(record.buckets, list) + for bucket in record.buckets: + assert isinstance(bucket.id, int) + assert isinstance(bucket.uid, int) + assert isinstance(bucket.name, str) + assert isinstance(bucket.type, str) + assert isinstance(bucket.state, int) + assert isinstance(bucket.region, str) + assert isinstance(bucket.metadata_template, str) + assert isinstance(bucket.net_type, int) + assert isinstance(bucket.tracker, int) + assert isinstance(bucket.created_at, str) + assert isinstance(bucket.updated_at, str) + assert isinstance(bucket.num, int) + assert isinstance(bucket.size, (str, int)) + assert isinstance(bucket.access_permission, str) + assert isinstance(bucket.labels, list) + for label in bucket.labels: + assert isinstance(label, str) + + +def test_list_bm_cs_projects_streams(client, project_ids): + for project_id in project_ids: + data = client.list_bm_cs_projects_streams(project_id=project_id) + print("wat") + assert_is_list_bm_cs_projects_streams(data) + + +def assert_is_list_bm_cs_projects_streams(data): + assert isinstance(data, list) + assert len(data) > 0 + for record in data: + print(data) + assert isinstance(record.stream_id, str) + assert isinstance(record.uid, int) + assert isinstance(record.mcp_id, int) + assert isinstance(record.stream_type, str) + assert isinstance(record.name, str) + assert isinstance(record.state, str) + assert isinstance(record.code, int) + assert isinstance(record.current_url, str) + assert isinstance(record.region, str) + assert isinstance(record.pitch_shift, int) + assert isinstance(record.check_pitch_shift, int) + assert isinstance(record.remark, str) + assert isinstance(record.created_at, str) + assert isinstance(record.updated_at, str) + assert isinstance(record.record_video, int) + assert isinstance(record.stream_rec_type, int) + assert isinstance(record.epg, str) + assert isinstance(record.add_recordings, int) + assert isinstance(record.timemap, int) + assert isinstance(record.ucf, int) + assert isinstance(record.stream_urls, list) + assert isinstance(record.config, BmCsProjectsStreamsConfig) + if record.config: + assert isinstance(record.config.id, int) + assert isinstance(record.config.name, str) + assert isinstance(record.config.uid, int) + assert isinstance(record.config.rec_length, int) + assert isinstance(record.config.interval, int) + assert isinstance(record.config.rec_timeout, int) + assert isinstance(record.config.monitor_timeout, int) + assert isinstance(record.config.noise, int) + assert isinstance(record.config.delay, int) + assert isinstance(record.config.record, dict) + assert isinstance(record.config.created_at, str) + assert isinstance(record.config.updated_at, str) + + for stream_url in record.stream_urls: + assert isinstance(stream_url, str) + if record.user_defined: + # maybe it's a string? + assert isinstance(record.user_defined, str) + + +def get_bm_cs_projects_results_params(): + client = Client(bearer_token=bearer_token) + project_ids = [p.id for p in client.list_bm_cs_projects()] + stream_id_and_created_ats = reduce( + concat, + [ + [ + (p, s.stream_id, s.created_at) + for s in client.list_bm_cs_projects_streams(p) + ] + for p in project_ids + ], + ) + tests = [] + for project_id, stream_id, created_at in stream_id_and_created_ats: + now = datetime.now().date() + + date = datetime.strptime(created_at, "%Y-%m-%d %H:%M:%S").date() + while date < now - timedelta(days=1): + date += timedelta(days=1) + tests.append( + ( + project_id, + stream_id, + { + "date": datetime.strftime(date, "%Y%m%d"), + }, + ) + ) + return tests + + +@mark.parametrize("project_id, stream_id, params", get_bm_cs_projects_results_params()) +def test_get_bm_cs_projects_results(project_id, stream_id, params, client): + data = client.get_bm_cs_projects_results( + project_id, + stream_id, + params=params, + ) + assert_is_get_bm_cs_projects_result(data) + + +# pylint: disable=too-many-statements,too-many-branches,too-many-locals,too-many-nested-blocks +def assert_is_get_bm_cs_projects_result(data): + assert isinstance(data, list) + for record in data: + assert record.status.code == 0 + assert record.status.msg == "Success" + assert record.status.version == "1.0" + + if record.result_type: + assert record.result_type == 0 + + assert record.metadata.type == "delay" + assert isinstance(record.metadata.timestamp_utc, str) + assert record.metadata.played_duration >= 0 + + if not record.metadata.music: + # custom record was set so music is not + record.metadata.music = [] + for music in record.metadata.music: + assert isinstance(music, Music) + assert isinstance(music.title, str) + assert isinstance(music.acrid, str) + if music.release_date: + assert isinstance(music.release_date, str) + if music.label: + assert isinstance(music.label, str) + if music.language: + assert isinstance(music.language, str) + if music.exids: + assert isinstance(music.exids, str) + if music.source: + assert isinstance(music.source, str) + if music.ppm: + assert isinstance(music.ppm, str) + + assert isinstance(music.result_from, int) + assert isinstance(music.sample_end_time_offset_ms, int) + assert isinstance(music.duration_ms, int) + assert isinstance(music.score, int) + assert isinstance(music.db_begin_time_offset_ms, int) + assert isinstance(music.db_end_time_offset_ms, int) + assert isinstance(music.play_offset_ms, int) + assert isinstance(music.sample_end_time_offset_ms, int) + + if music.contributors: + assert isinstance(music.contributors, Contributors) + if music.contributors.composers: + assert isinstance(music.contributors.composers, list) + for composer in music.contributors.composers: + assert isinstance(composer, str) + if music.contributors.lyricists: + assert isinstance(music.contributors.lyricists, list) + for lyricist in music.contributors.lyricists: + assert isinstance(lyricist, str) + if music.langs: + assert isinstance(music.langs, list) + for lang in music.langs: + assert isinstance(lang, (Lang, str)) + if isinstance(lang, Lang): + assert isinstance(lang.code, str) + assert isinstance(lang.name, str) + if music.genres: + assert isinstance(music.genres, list) + for genre in music.genres: + assert isinstance(genre, Genre) + assert isinstance(genre.name, str) + if genre.id: + assert isinstance(genre.id, (int, str)) + if music.works: + assert isinstance(music.works, list) + for work in music.works: + assert isinstance(work, Work) + + assert isinstance(work.id, str) + assert isinstance(work.name, str) + assert isinstance(work.iswc, str) + + assert isinstance(work.creators, list) + for creator in work.creators: + assert isinstance(creator, WorkCreator) + assert isinstance(creator.ipi, int) + assert isinstance(creator.name, str) + if creator.affiliation: + assert isinstance(creator.affiliation, dict) + if creator.last_name: + assert isinstance(creator.last_name, str) + if creator.role: + assert isinstance(creator.role, str) + + if work.works: + assert isinstance(work.works, list) + for inner_work in work.works: + assert isinstance(inner_work, WorkWork) + assert isinstance(inner_work.code, str) + assert isinstance(inner_work.agency, WorkWorkAgency) + assert isinstance(inner_work.agency.code, str) + if work.other_names: + assert isinstance(work.other_names, list) + for other_name in work.other_names: + assert isinstance(other_name, str) + + assert isinstance(music.external_ids, ExternalIds) + if music.external_ids.isrc: + assert isinstance(music.external_ids.isrc, (str, list)) + if music.external_ids.upc: + assert isinstance(music.external_ids.upc, (str, list)) + if music.external_ids.iswc: + assert isinstance(music.external_ids.iswc, str) + + assert isinstance(music.external_metadata, (ExternalMetadata, list)) + if isinstance(music.external_metadata, list): + for external_metadata in music.external_metadata: + assert isinstance(external_metadata, ExternalMetadata) + assert isinstance(external_metadata.spotify, str) + assert isinstance(external_metadata.deezer, str) + assert isinstance(external_metadata.youtube, str) + + assert isinstance(music.album, Album) + if music.album.name: + assert isinstance(music.album.name, str) + if music.album.id: + assert isinstance(music.album.id, (str, int)) + if music.album.image: + assert isinstance(music.album.image, str) + + if music.artists: + assert isinstance(music.artists, list) + for artist in music.artists: + assert isinstance(artist, Artist) + assert isinstance(artist.name, str) + if artist.id: + assert isinstance(artist.id, (str, int, dict)) + if artist.role: + assert isinstance(artist.role, str) + if artist.roles: + assert isinstance(artist.roles, list) + if artist.langs: + assert isinstance(artist.langs, list) + if artist.seokey: + assert isinstance(artist.seokey, str) + if artist.isni: + assert isinstance(artist.isni, str) diff --git a/tests/test_client.py b/tests/test_client.py index fd962ba..fa14256 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,27 +1,275 @@ +"""Unittests for ACRCloud client. + +These are mostly trivial tests for coverage's sake. +Most of our real testing effort is in the acceptance +test suite which helps us ensure that the client +works for our real-world use case. Thus you should +not rely on the examples in here, but rather look +at the type hints or acceptance test suite to see +what you can expect from the API. +""" import requests_mock -import acrclient +from acrclient import Client +from acrclient.client import _Auth +from acrclient.models import ( + GetBmCsProjectsResultsParams, + ListBmCsProjectsParams, + ListBmCsProjectsStreamsParams, +) def test_client(): - client = acrclient.Client("bearer-token") - assert isinstance(client, acrclient.Client) - assert client.bearer_token == "bearer-token" - assert client.base_url == "https://eu-api-v2.acrcloud.com" + client = Client(bearer_token="bearer-token") + assert isinstance(client, Client) + assert isinstance(client.auth, _Auth) + assert client.base_url == "https://api-v2.acrcloud.com" + + +def test_client_get(): + client = Client(bearer_token="bearer-token") + with requests_mock.Mocker() as mock: + mock.get("https://api-v2.acrcloud.com/foo", json={}) + client.get("/foo") + + +def test_client_list_bm_cs_projects(): + client = Client(bearer_token="bearer-token") + with requests_mock.Mocker() as mock: + mock.get( + f"{client.base_url}/api/bm-cs-projects", + json={ + "data": [ + { + "id": "", + "name": "", + "uid": "", + "type": "", + "region": "", + "state": "", + "access_key": "", + "bucket_group": "", + "noise": "", + "created_at": "", + "updated_at": "", + "iswc": "", + "buckets": [ + { + "id": "", + "uid": "", + "name": "", + "type": "", + "state": "", + "region": "", + "metadata_template": "", + "labels": "", + "net_type": "", + "tracker": "", + "created_at": "", + "updated_at": "", + "num": "", + "size": "", + "access_permission": "", + }, + ], + "status_check": "", + "external_ids": "", + "metadata_template": "", + "result_callback_url": "", + "result_callback_send_type": "", + "result_callback_send_noresult": "", + "state_notification_url": "", + "state_notification_email": "", + "state_notification_email_frequency": "", + "state_notification_email_custom_interval": "", + "result_callback_retry": "", + "monitoring_num": "", + "config": { + "record": {}, + }, + }, + ], + }, + ) + client.list_bm_cs_projects( + params=ListBmCsProjectsParams(), + ) -def test_request(): - client = acrclient.Client("bearer-token") +def test_client_list_bm_cs_projects_streams(): + client = Client(bearer_token="bearer-token") with requests_mock.Mocker() as mock: - mock.get("https://eu-api-v2.acrcloud.com/foo", json={}) - client.request("/foo") + mock.get( + f"{client.base_url}/api/bm-cs-projects/project-id/streams", + json={ + "data": [ + { + "stream_id": "", + "uid": "", + "mcp_id": "", + "stream_type": "", + "name": "", + "state": "", + "code": "", + "stream_urls": "", + "current_url": "", + "region": "", + "pitch_shift": "", + "check_pitch_shift": "", + "remark": "", + "created_at": "", + "updated_at": "", + "record_video": "", + "stream_rec_type": "", + "epg": "", + "add_recordings": "", + "timemap": "", + "ucf": "", + "config": { + "id": "", + "name": "", + "uid": "", + "rec_length": "", + "interval": "", + "rec_timeout": "", + "monitor_timeout": "", + "noise": "", + "delay": "", + "record": "", + "created_at": "", + "updated_at": "", + }, + }, + ], + }, + ) + client.list_bm_cs_projects_streams( + "project-id", + params=ListBmCsProjectsStreamsParams(), + ) -def test_get_bm_cs_projects_results(): - client = acrclient.Client("bearer-token") +def test_client_get_bm_cs_projects_results(): + client = Client(bearer_token="bearer-token") with requests_mock.Mocker() as mock: mock.get( f"{client.base_url}/api/bm-cs-projects/project-id/streams/stream-id/results", - json={"data": {}}, + json={ + "data": [ + { + "status": { + "msg": "", + "version": "", + "code": "", + }, + "metadata": { + "type": "", + "timestamp_utc": "", + "played_duration": "", + "music": [ + { + "title": "", + "acrid": "", + "score": 0, + "result_from": 0, + "duration_ms": 0, + "sample_begin_time_offset_ms": 0, + "sample_end_time_offset_ms": 0, + "db_begin_time_offset_ms": 0, + "db_end_time_offset_ms": 0, + "play_offset_ms": 0, + "external_ids": {}, + "external_metadata": { + "musicstory": { + "track": { + "id": 0, + }, + "album": { + "id": 0, + }, + "release": { + "id": 0, + }, + }, + "lyricfind": { + "lfid": "", + }, + "musicbrainz": [ + { + "track": "", + } + ], + }, + "artists": [ + { + "name": "", + } + ], + "album": { + "langs": [ + { + "code": "", + "name": "", + } + ] + }, + "contributors": { + "composers": {}, + "lyricists": {}, + }, + "rights_claim": { + "TODO": "shrug", + }, + "lyrics": { + "TODO": "shrug", + }, + "langs": [ + { + "code": "", + "name": "", + } + ], + "genres": [ + { + "name": "", + } + ], + "works": [ + { + "id": "", + "name": "", + "iswc": "", + "creators": [ + { + "ipi": 0, + "name": "", + } + ], + "works": [ + { + "code": "", + "agency": { + "code": "", + }, + } + ], + } + ], + "release_by_territories": [ + { + "territories": [""], + "release_date": "", + } + ], + }, + ], + }, + }, + ] + }, + ) + client.get_bm_cs_projects_results( + "project-id", + "stream-id", + params=GetBmCsProjectsResultsParams(), ) - client.get_bm_cs_projects_results("project-id", "stream-id")