Initial commit : working script with API functionnality and blacklist

musik/__main__.py

@@ -0,0 +1,194 @@
+import argparse
+import itertools
+import pathlib
+import random
+import re
+
+from datetime import date, datetime
+from pathlib import Path
+from urllib import parse, request
+
+
+ROOT_PATH = Path("./lists")
+BLACKLIST = Path("./blacklists")
+RESULTS = Path("./results")
+NUM_MUS = 2
+
+MUSIK = []
+USERS = []
+
+
+def bl_path(user):
+    return Path(BLACKLIST, user).with_suffix(".txt")
+
+
+# Lecture arguments console
+parser = argparse.ArgumentParser(
+    prog="python -m musik", description="Lancer une partie de Musik"
+)
+parser.add_argument(
+    "-a",
+    "--no-api",
+    action="store_true",
+    help="Désactiver l'API Youtube (affiche la liste des liens)",
+)
+parser.add_argument(
+    "-b",
+    "--no-blacklist",
+    action="store_true",
+    help="Désactiver le méchanisme de blacklist (en lecture et écriture)",
+)
+parser.add_argument(
+    "-n",
+    "--number",
+    type=int,
+    default=NUM_MUS,
+    help=f"Modifier le nombre de musiques par joueur (défaut : {NUM_MUS})",
+)
+parser.add_argument(
+    "--lists",
+    type=Path,
+    default=ROOT_PATH,
+    help="Sélectionner le dossier contenant les listes de musiques",
+)
+parser.add_argument(
+    "--blacklists",
+    type=Path,
+    default=BLACKLIST,
+    help="Sélectionner le dossier contenant les blacklist",
+)
+parser.add_argument(
+    "--results",
+    type=Path,
+    default=RESULTS,
+    help="Sélectionner le dossier pour stocker les résultats",
+)
+
+args = parser.parse_args()
+NO_API = args.no_api
+NO_BLACKLIST = args.no_blacklist
+NUM_MUS = args.number
+ROOT_PATH = args.lists
+BLACKLIST = args.blacklists
+RESULTS = args.results
+
+print("--- DÉBUT ---")
+
+print("> Vérification")
+assert ROOT_PATH.is_dir(), f"ROOT_PATH={ROOT_PATH} n'est pas un dossier"
+if not BLACKLIST.is_dir():
+    BLACKLIST.mkdir()
+if not RESULTS.is_dir():
+    RESULTS.mkdir()
+
+assert NUM_MUS >= 1, "NUM_MUS < 1"
+
+# Lecture des fichiers musique dans ROOT_PATH
+# Faire un dossier différent pour les gens qui ne jouent pas
+print("> Génération de la liste de musiques")
+for q in ROOT_PATH.iterdir():
+    _u = q.stem
+    print(f"> > Musiques de {_u}")
+    if (not NO_BLACKLIST) and bl_path(_u).exists():
+        print("> > > Blacklist")
+        with bl_path(_u).open("r") as blf:
+            blacklist = blf.read().splitlines()
+    else:
+        blacklist = []
+
+    print("> > > Lecture de la liste")
+    with q.open() as f:
+        _raw_musiks = [
+            re.compile(r"[\?\&]v=(\w+)").search(_musik)[1]
+            for _musik in f.read().splitlines()
+        ]
+        _musiks = list(filter(lambda _m: not _m in blacklist, _raw_musiks))
+
+    assert (
+        len(_musiks) >= NUM_MUS
+    ), f"{_u} a {len(_musiks)} musique(s) non black-listée au lieu de {NUM_MUS}"
+
+    print("> > > Ajout des musiques à la liste")
+    MUSIK += random.sample(_musiks, NUM_MUS)
+    USERS += [_u] * NUM_MUS
+
+# Shuffle musics
+print("> Classement aléatoire des musiques")
+UM = list(zip(USERS, MUSIK))
+random.shuffle(UM)
+USERS, MUSIK = zip(*UM)
+
+if not NO_API:
+    import google_auth_oauthlib.flow
+    import googleapiclient.discovery
+    import googleapiclient.errors
+
+    # Connexion à l'API youtube, obtention d'un jeton OAuth
+    print("> Connexion à l'API Youtube")
+    flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_secrets_file(
+        "./secret.json", ["https://www.googleapis.com/auth/youtube.force-ssl"]
+    )
+    credentials = flow.run_local_server(port=0)
+    youtube = googleapiclient.discovery.build("youtube", "v3", credentials=credentials)
+
+    # Création d'une playlist
+    print("> Création de la playlist")
+    pl_request = youtube.playlists().insert(
+        part="snippet,status",
+        body={
+            "snippet": {
+                "title": f"Musik {date.today().strftime('%x')}",
+            },
+            "status": {
+                "privacyStatus": "private",
+            },
+        },
+    )
+    pl_response = pl_request.execute()
+    print(
+        f"> > Playlist créée : https://www.youtube.com/playlist?list={pl_response['id']}"
+    )
+
+    # Insertion des musiques dans la playlist
+    print("> Insertion des musiques dans la playlist")
+    print(f"> > {'_'*len(MUSIK)}")
+    print("> > ", end="")
+    for musik in MUSIK:
+        print("#", end="")
+        request = youtube.playlistItems().insert(
+            part="snippet",
+            body={
+                "snippet": {
+                    "playlistId": pl_response.get("id"),
+                    "position": 0,
+                    "resourceId": {
+                        "kind": "youtube#video",
+                        "videoId": musik,
+                    },
+                }
+            },
+        )
+        response = request.execute()
+    print()
+else:
+    print("> Liste des musiques :")
+    for musik in MUSIK:
+        print(f"> # https://www.youtube.com/watch?v={musik}")
+
+# Écriture des résultats
+print("> Écriture des résultats")
+with RESULTS.joinpath(datetime.now().strftime("%Y%m%d %H%M%S")).with_suffix(
+    ".txt"
+).open("a") as f:
+    f.write(f"Résultats {datetime.now()}\n\n")
+    f.write("\n".join("\t".join(um) for um in zip(USERS, MUSIK)))
+
+# Écriture de la blacklist
+if not NO_BLACKLIST:
+    print("> Écriture de la blacklist")
+    for user, musik in zip(USERS, MUSIK):
+        with bl_path(user).open("a") as f:
+            f.write("\n")
+            f.write(musik)
+
+print("--- FIN ---")