Merge pull request #773 from CrazyTim71/anilist

AniList: Add Oauth import for private profiles

Author: FuzzyGrim

.gitignore

@@ -7,3 +7,4 @@ src/staticfiles
 src/db/db.sqlite3
 db.sqlite3-shm
 db.sqlite3-wal
+.vscode/launch.json

src/config/settings.py

@@ -363,7 +363,9 @@ def secret(key, default=undefined, **kwargs):
 
 STEAM_API_KEY = config(
     "STEAM_API_KEY",
-    default=secret("STEAM_API_KEY_FILE", ""), # Generate default key https://steamcommunity.com/dev/apikey
+    default=secret(
+        "STEAM_API_KEY_FILE", "",
+    ),  # Generate default key https://steamcommunity.com/dev/apikey
 )
 
 HARDCOVER_API = config(
@@ -405,6 +407,22 @@ def secret(key, default=undefined, **kwargs):
     ),
 )
 
+ANILIST_ID = config(
+    "ANILIST_ID",
+    default=secret(
+        "ANILIST_ID_FILE",
+        "",
+    ),
+)
+
+ANILIST_SECRET = config(
+    "ANILIST_SECRET",
+    default=secret(
+        "ANILIST_SECRET_FILE",
+        "",
+    ),
+)
+
 SIMKL_ID = config(
     "SIMKL_ID",
     default=secret(

src/integrations/imports/anilist.py

@@ -4,38 +4,112 @@
 
 import requests
 from django.apps import apps
+from django.conf import settings
 from django.utils import timezone
 
 import app
 from app.models import MediaTypes, Sources, Status
+from app.providers import services
 from integrations.imports import helpers
 from integrations.imports.helpers import MediaImportError, MediaImportUnexpectedError
 
 logger = logging.getLogger(__name__)
 
 
-def importer(username, user, mode):
+def get_token(request):
+    """View for getting the AniList OAuth2 token."""
+    domain = request.get_host()
+    scheme = request.scheme
+    code = request.GET["code"]
+
+    url = "https://anilist.co/api/v2/oauth/token"
+
+    params = {
+        "client_id": settings.ANILIST_ID,
+        "client_secret": settings.ANILIST_SECRET,
+        "code": code,
+        "grant_type": "authorization_code",
+        "redirect_uri": f"{scheme}://{domain}/import/anilist/private",
+    }
+
+    try:
+        token_response = app.providers.services.api_request(
+            "ANILIST",
+            "POST",
+            url,
+            params=params,
+        )
+    except services.ProviderAPIError as error:
+        if error.status_code == requests.codes.unauthorized:
+            msg = "Invalid Anilist secret key."
+            raise MediaImportError(msg) from error
+        raise
+
+    return {
+        "access_token": token_response["access_token"],
+        "username": get_username_from_oauth(token_response["access_token"]),
+    }
+
+
+def get_username_from_oauth(access_token):
+    """Get AniList username from access token."""
+    query = """
+    query {
+        Viewer {
+            name
+        }
+    }
+    """
+
+    headers = {
+        "Authorization": f"Bearer {access_token}",
+        "Content-Type": "application/json",
+    }
+
+    try:
+        response = app.providers.services.api_request(
+            "ANILIST",
+            "POST",
+            "https://graphql.anilist.co",
+            headers=headers,
+            params={"query": query},
+        )
+    except services.ProviderAPIError as error:
+        if error.status_code == requests.codes.unauthorized:
+            msg = "Invalid AniList access token."
+            raise MediaImportError(msg) from error
+        raise
+
+    return response["data"]["Viewer"]["name"]
+
+
+def importer(token, user, mode, username):
     """Import anime and manga ratings from Anilist."""
-    anilist_importer = AniListImporter(username, user, mode)
+    anilist_importer = AniListImporter(token, user, mode, username)
     return anilist_importer.import_data()
 
 
 class AniListImporter:
     """Class to handle importing user data from AniList."""
 
-    def __init__(self, username, user, mode):
+    def __init__(self, token, user, mode, username):
         """Initialize the importer with username, user, and mode.
 
         Args:
             username (str): AniList username to import from
+            token (str): Encrypted access token for private imports (optional)
             user: Django user object to import data for
             mode (str): Import mode ("new" or "overwrite")
         """
         self.username = username
+        self.token = token
         self.user = user
         self.mode = mode
         self.warnings = []
 
+        if self.token is not None:
+            self.token = helpers.decrypt(self.token)
+
         # Track existing media for "new" mode
         self.existing_media = helpers.get_existing_media(user)
 
@@ -126,6 +200,14 @@ def import_data(self):
         variables = {"userName": self.username}
         url = "https://graphql.anilist.co"
 
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        }
+
+        if self.token:
+            headers["Authorization"] = f"Bearer {self.token}"
+
         logger.info("Fetching anime and manga from AniList account")
 
         try:
@@ -134,6 +216,7 @@ def import_data(self):
                 "POST",
                 url,
                 params={"query": query, "variables": variables},
+                headers=headers,
             )
         except requests.exceptions.HTTPError as error:
             error_message = error.response.json()["errors"][0].get("message")

src/integrations/tasks.py

@@ -95,9 +95,9 @@ def import_mal(username, user_id, mode):
 
 
 @shared_task(name="Import from AniList")
-def import_anilist(username, user_id, mode):
-    """Celery task for importing anime and manga data from AniList."""
-    return import_media(anilist.importer, username, user_id, mode)
+def import_anilist(user_id, mode, token=None, username=None):
+    """Celery task for importing media data from AniList."""
+    return import_media(anilist.importer, token, user_id, mode, username)
 
 
 @shared_task(name="Import from Kitsu")
@@ -117,16 +117,19 @@ def import_hltb(file, user_id, mode):
     """Celery task for importing media data from HowLongToBeat."""
     return import_media(hltb.importer, file, user_id, mode)
 
+
 @shared_task(name="Import from Steam")
 def import_steam(username, user_id, mode):
     """Celery task for importing game data from Steam."""
     return import_media(steam.importer, username, user_id, mode)
 
+
 @shared_task(name="Import from IMDB")
 def import_imdb(file, user_id, mode):
     """Celery task for importing media data from IMDB."""
     return import_media(imdb.importer, file, user_id, mode)
 
+
 @shared_task(name="Import from GoodReads")
 def import_goodreads(file, user_id, mode):
     """Celery task for importing media data from GoodReads."""

src/integrations/tests/test_imports.py

@@ -120,13 +120,47 @@ def setUp(self):
         self.user = get_user_model().objects.create_user(**self.credentials)
 
     @patch("requests.Session.post")
-    def test_import_anilist(self, mock_request):
+    def test_import_anilist_public(self, mock_request):
         """Basic test importing anime and manga from AniList."""
         with Path(mock_path / "import_anilist.json").open() as file:
             anilist_response = json.load(file)
         mock_request.return_value.json.return_value = anilist_response
 
-        anilist.importer("bloodthirstiness", self.user, "new")
+        anilist.importer(None, self.user, "new", "bloodthirstiness")
+
+        self.assertEqual(Anime.objects.filter(user=self.user).count(), 4)
+        self.assertEqual(Manga.objects.filter(user=self.user).count(), 3)
+        self.assertEqual(
+            Anime.objects.get(user=self.user, item__title="FLCL").status,
+            Status.PAUSED.value,
+        )
+        self.assertEqual(
+            Manga.objects.filter(user=self.user, item__title="One Punch-Man")
+            .first()
+            .score,
+            9,
+        )
+        self.assertEqual(
+            Anime.objects.get(user=self.user, item__title="FLCL")
+            .history.first()
+            .history_date,
+            datetime(2025, 6, 4, 10, 11, 17, tzinfo=UTC),
+        )
+
+    @patch("requests.Session.post")
+    def test_import_anilist_private(self, mock_request):
+        """Basic test importing anime and manga from AniList."""
+        with Path(mock_path / "import_anilist.json").open() as file:
+            anilist_response = json.load(file)
+        mock_request.return_value.json.return_value = anilist_response
+
+        anilist.importer(
+            helpers.encrypt("token"),
+            self.user,
+            "new",
+            "username",
+        )
+
         self.assertEqual(Anime.objects.filter(user=self.user).count(), 4)
         self.assertEqual(Manga.objects.filter(user=self.user).count(), 3)
         self.assertEqual(
@@ -151,9 +185,10 @@ def test_user_not_found(self):
         self.assertRaises(
             helpers.MediaImportError,
             anilist.importer,
-            "fhdsufdsu",
+            None,
             self.user,
             "new",
+            "fhdsufdsu",
         )
 
 

src/integrations/urls.py

@@ -8,7 +8,15 @@
     path("simkl-oauth", views.simkl_oauth, name="simkl_oauth"),
     path("import/simkl", views.import_simkl, name="import_simkl"),
     path("import/mal", views.import_mal, name="import_mal"),
-    path("import/anilist", views.import_anilist, name="import_anilist"),
+    path("import/anilist/oauth", views.anilist_oauth, name="import_anilist_oauth"),
+    path("import/anilist/private",
+         views.import_anilist_private,
+         name="import_anilist_private",
+    ),
+    path("import/anilist/public",
+         views.import_anilist_public,
+         name="import_anilist_public",
+    ),
     path("import/kitsu", views.import_kitsu, name="import_kitsu"),
     path("import/yamtrack", views.import_yamtrack, name="import_yamtrack"),
     path("import/hltb", views.import_hltb, name="import_hltb"),

src/integrations/views.py

@@ -17,7 +17,7 @@
 
 import users
 from integrations import exports, tasks
-from integrations.imports import helpers, simkl, trakt
+from integrations.imports import anilist, helpers, simkl, trakt
 from integrations.webhooks import emby, jellyfin, plex
 
 logger = logging.getLogger(__name__)
@@ -150,7 +150,62 @@ def import_mal(request):
 
 
 @require_POST
-def import_anilist(request):
+def anilist_oauth(request):
+    """Initiate AniList OAuth flow."""
+    redirect_uri = request.build_absolute_uri(reverse("import_anilist_private"))
+    url = "https://anilist.co/api/v2/oauth/authorize"
+    state = {
+        "mode": request.POST["mode"],
+        "frequency": request.POST["frequency"],
+        "time": request.POST["time"],
+    }
+
+    state_token = secrets.token_urlsafe(32)
+    request.session[state_token] = state
+
+    return redirect(
+        f"{url}?client_id={settings.ANILIST_ID}&redirect_uri={redirect_uri}&response_type=code&state={state_token}",
+    )
+
+@require_GET
+def import_anilist_private(request):
+    """View for getting the AniList OAuth2 token."""
+    oauth_callback = anilist.get_token(request)
+    enc_token = helpers.encrypt(oauth_callback["access_token"])
+    state_token = request.GET["state"]
+    username = oauth_callback["username"]
+
+    if not username:
+        messages.error(request, "AniList username is required.")
+        return redirect("import_data")
+
+    frequency = request.session[state_token]["frequency"]
+    mode = request.session[state_token]["mode"]
+    import_time = request.session[state_token]["time"]
+
+    if frequency == "once":
+        tasks.import_anilist.delay(
+            user_id=request.user.id,
+            mode=mode,
+            username=username,
+            token=enc_token,
+        )
+        messages.info(request, "AniList import queued.")
+    else:
+        helpers.create_import_schedule(
+            username=username,
+            request=request,
+            mode=mode,
+            frequency=frequency,
+            import_time=import_time,
+            source="AniList",
+            token=enc_token,
+        )
+    return redirect("import_data")
+
+
+@require_POST
+def import_anilist_public(request):
     """View for importing anime and manga data from AniList."""
     username = request.POST.get("user")
     if not username:
@@ -159,23 +214,23 @@ def import_anilist(request):
 
     mode = request.POST["mode"]
     frequency = request.POST["frequency"]
+    import_time = request.POST["time"]
 
     if frequency == "once":
         tasks.import_anilist.delay(
-            username=username,
             user_id=request.user.id,
             mode=mode,
+            username=username,
         )
-        messages.info(request, "The task to import media from AniList has been queued.")
+        messages.info(request, "AniList import queued.")
     else:
-        import_time = request.POST["time"]
         helpers.create_import_schedule(
-            username,
-            request,
-            mode,
-            frequency,
-            import_time,
-            "AniList",
+            username=username,
+            request=request,
+            mode=mode,
+            frequency=frequency,
+            import_time=import_time,
+            source="AniList",
         )
     return redirect("import_data")