de proto

artist.py

import httpx
from urllib.parse import urlparse, parse_qs
from prefect import flow, task

@task(
    version="0.2.0",
    cache_key_fn=lambda ctx, args: f"{args['id']} v{ctx.task.version}",
)
def collate_soundcloud_tracks(id):
    path = f"/matt/soundcloud-dev/rest/Artist?id={id}"
    page1 = get_tracks(path)
    tracks = page1["tracks"]

    if page1["next"] is not None:
        page2 = get_tracks(page1["next"])
        tracks.extend(page2["tracks"])

    return tracks


@flow
def get_artist_tracks(locator: str):
    url = urlparse(locator)

    if url.path == "soundcloud/Artist":
        id = parse_qs(url.query)["id"][0]
        return collate_soundcloud_tracks(id)

    if url.path == "spotify/Artist":
        raise Exception("Spotify not yet implemented")

    raise Exception(f"Unknown project type: {url.path}")

completion.py

PROMPT = """
Task: Identify recurring release patterns from a list of track names.

Given the list of tracks provided, detect ...

SOURCE TRACKLIST:

"""


@flow
def identify_shows(artist_locator: str):
    tracks = get_artist_tracks(artist_locator)
    messages = [
        {
            "role": "user",
            "content": PROMPT + json.dumps(tracks),
        }
    ]
    completion = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=messages)
    return completion