103 lines
2.8 KiB
Python
103 lines
2.8 KiB
Python
|
"""
|
||
|
|
One-shot script : supprime toutes les notes 10/10 de ton compte Trakt.
|
||
|
|
Usage : python remove_10s.py
|
||
|
|
"""
|
||
|
|
import httpx
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
from dotenv import load_dotenv
|
||
|
|
|
||
|
|
load_dotenv()
|
||
|
|
|
||
|
|
CLIENT_ID = os.environ["TRAKT_CLIENT_ID"]
|
||
|
|
CLIENT_SECRET = os.environ["TRAKT_CLIENT_SECRET"]
|
||
|
|
REDIRECT_URI = "urn:ietf:wg:oauth:2.0:oob" # mode "device" — pas besoin de serveur
|
||
|
|
BASE = "https://api.trakt.tv"
|
||
|
|
|
||
|
|
HEADERS = {
|
||
|
|
"Content-Type": "application/json",
|
||
|
|
"trakt-api-version": "2",
|
||
|
|
"trakt-api-key": CLIENT_ID,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def get_token() -> str:
|
||
|
|
# 1. Demander un code device
|
||
|
|
r = httpx.post(f"{BASE}/oauth/device/code", json={"client_id": CLIENT_ID})
|
||
|
|
r.raise_for_status()
|
||
|
|
data = r.json()
|
||
|
|
|
||
|
|
print(f"\nOuvre cette URL dans ton navigateur :\n {data['verification_url']}")
|
||
|
|
print(f"\nSaisis ce code : {data['user_code']}")
|
||
|
|
print("\nAttente de l'autorisation…")
|
||
|
|
|
||
|
|
# 2. Polling automatique jusqu'à autorisation ou expiration
|
||
|
|
import time
|
||
|
|
interval = data.get("interval", 5)
|
||
|
|
expires_in = data.get("expires_in", 600)
|
||
|
|
deadline = time.time() + expires_in
|
||
|
|
|
||
|
|
while time.time() < deadline:
|
||
|
|
time.sleep(interval)
|
||
|
|
r = httpx.post(f"{BASE}/oauth/device/token", json={
|
||
|
|
"code": data["device_code"],
|
||
|
|
"client_id": CLIENT_ID,
|
||
|
|
"client_secret": CLIENT_SECRET,
|
||
|
|
})
|
||
|
|
if r.status_code == 200:
|
||
|
|
print("Autorisé !")
|
||
|
|
return r.json()["access_token"]
|
||
|
|
if r.status_code == 400:
|
||
|
|
continue # pending
|
||
|
|
if r.status_code == 409:
|
||
|
|
continue # slow down
|
||
|
|
break
|
||
|
|
|
||
|
|
print("Autorisation échouée ou expirée.")
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
token = get_token()
|
||
|
|
auth_headers = {**HEADERS, "Authorization": f"Bearer {token}"}
|
||
|
|
|
||
|
|
# 3. Récupérer toutes les notes
|
||
|
|
print("Récupération des notes…")
|
||
|
|
r = httpx.get(f"{BASE}/sync/ratings/movies", headers=auth_headers, timeout=60)
|
||
|
|
r.raise_for_status()
|
||
|
|
all_ratings = r.json()
|
||
|
|
|
||
|
|
movies_10 = [
|
||
|
|
{"ids": m["movie"]["ids"]}
|
||
|
|
for m in all_ratings
|
||
|
|
if m["rating"] == 10
|
||
|
|
]
|
||
|
|
|
||
|
|
if not movies_10:
|
||
|
|
print("Aucun film noté 10/10.")
|
||
|
|
return
|
||
|
|
|
||
|
|
print(f"{len(movies_10)} films notés 10/10 trouvés. Suppression en cours…")
|
||
|
|
|
||
|
|
# 4. Supprimer par batch de 100
|
||
|
|
BATCH = 100
|
||
|
|
total = 0
|
||
|
|
for i in range(0, len(movies_10), BATCH):
|
||
|
|
batch = movies_10[i:i + BATCH]
|
||
|
|
r = httpx.post(
|
||
|
|
f"{BASE}/sync/ratings/remove",
|
||
|
|
headers=auth_headers,
|
||
|
|
json={"movies": batch},
|
||
|
|
timeout=30,
|
||
|
|
)
|
||
|
|
r.raise_for_status()
|
||
|
|
deleted = r.json().get("deleted", {}).get("movies", 0)
|
||
|
|
total += deleted
|
||
|
|
print(f" batch {i // BATCH + 1} — {deleted} supprimés")
|
||
|
|
|
||
|
|
print(f"\nTerminé. {total} notes 10/10 supprimées.")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|