Import Watcharr JSON

requirements.txt

@@ -25,3 +25,4 @@ redis[hiredis]==7.1.0
 requests==2.32.5
 requests-ratelimiter==0.8.0
 unidecode==1.4.0
+python-dateutil

src/integrations/imports/watcharr.py

@@ -0,0 +1,130 @@
+import json
+import logging
+
+from dateutil import parser
+
+from integrations.imports.yamtrack import YamtrackImporter
+
+logger = logging.getLogger(__name__)
+
+
+class UnknownStateError(Exception):
+    """Custom exception for unexpected state string."""
+
+
+def importer(file, user, mode):
+    """Import media from Watcharr JSON file resuing the YamtrackImporter."""
+    csv_importer = WatcharrImporter(file, user, mode)
+    return csv_importer.import_data()
+
+
+def get_state(state):
+    """Convert the Watcharr status to a Yamtrack status."""
+    match state:
+        case "FINISHED":
+            return "Completed"
+        case "WATCHING":
+            return "In progress"
+        case "PLANNED":
+            return "Planning"
+        case "PAUSED":
+            return "Paused"
+        case "DROPPED":
+            return "Dropped"
+        case _:
+            raise UnknownStateError
+
+
+def to_date(date_str):
+    """Convert the Watcharr date to ISO 8601."""
+    date = parser.parse(date_str)
+    return date.isoformat()
+
+
+class WatcharrImporter(YamtrackImporter):
+    """Class to handle importing user data from JSON files."""
+
+    def __init__(self, file, user, mode):
+        """Initialize the importer with file, user, and mode.
+
+        Args:
+            file: Uploaded CSV file object
+            user: Django user object to import data for
+            mode (str): Import mode ("new" or "overwrite")
+        """
+        super().__init__(file, user, mode)
+        self._rows = []
+
+    def _add_entry(self, media_type, content_entry, state_entry, dict_entry):
+        """Add a single entry to the list of rows."""
+        dict_entry["media_type"] = media_type
+        dict_entry["source"] = "tmdb"
+        # when testing, in integrations/imports/helpers.py::update_season_references()
+        # existing_tv uses strings as keys:
+        dict_entry["media_id"] = str(content_entry["content"]["tmdbId"])
+        dict_entry["title"] = content_entry["content"]["title"]
+
+        dict_entry["score"] = state_entry["rating"]
+        dict_entry["status"] = get_state(state_entry["status"])
+        dict_entry["created_at"] = to_date(state_entry["createdAt"])
+        dict_entry["progressed_at"] = to_date(state_entry["updatedAt"])
+
+        dict_entry["image"] = ""
+        dict_entry["notes"] = ""
+        dict_entry["start_date"] = ""
+        dict_entry["end_date"] = ""
+
+        if "season_number" not in dict_entry:
+            dict_entry["season_number"] = ""
+        if "episode_number" not in dict_entry:
+            dict_entry["episode_number"] = ""
+        if "progress" not in dict_entry:
+            dict_entry["progress"] = ""
+
+        self._rows.append(dict_entry)
+
+    def get_iterator(self):
+        """Process the JSON file and return an array for YamtrackImporter."""
+        self._rows = []
+        json_structure = json.load(self.file)
+
+        for entry in json_structure:
+            try:
+                self._process_entry(entry)
+            except Exception as error:
+                error_msg = f"Error processing entry: {entry}"
+                logger.exception(error_msg)
+                self.warnings.append(f"{error_msg}. Error: {error}")
+        return self._rows
+
+    def _process_entry(self, entry):
+        """Process a single entry from the main array in the JSON file."""
+        match entry["content"]["type"]:
+            case "movie":
+                self._add_entry(
+                    "movie",
+                    entry,
+                    entry,
+                    {"progress": 1 if entry["status"] == "FINISHED" else 0},
+                )
+            case "tv":
+                self._add_entry("tv", entry, entry, {})
+                if "watchedSeasons" in entry:
+                    for season in entry["watchedSeasons"]:
+                        self._add_entry(
+                            "season",
+                            entry,
+                            season,
+                            {"season_number": season["seasonNumber"]},
+                        )
+                if "watchedEpisodes" in entry:
+                    for episode in entry["watchedEpisodes"]:
+                        self._add_entry(
+                            "episode",
+                            entry,
+                            episode,
+                            {
+                                "season_number": episode["seasonNumber"],
+                                "episode_number": episode["episodeNumber"],
+                            },
+                        )

src/integrations/imports/yamtrack.py

@@ -7,6 +7,7 @@
 from django.utils.dateparse import parse_datetime
 
 import app
+import app.forms
 from app import config
 from app.models import MediaTypes, Sources
 from app.providers import services
@@ -54,15 +55,19 @@ def __init__(self, file, user, mode):
             mode,
         )
 
-    def import_data(self):
-        """Import all user data from the CSV file."""
+    def get_iterator(self):
+        """Return an iterator for the CSV file. Exists to be overridden."""
         try:
             decoded_file = self.file.read().decode("utf-8").splitlines()
         except UnicodeDecodeError as e:
             msg = "Invalid file format. Please upload a CSV file."
             raise MediaImportError(msg) from e
 
-        reader = DictReader(decoded_file)
+        return DictReader(decoded_file)
+
+    def import_data(self):
+        """Import all user data from the CSV file."""
+        reader = self.get_iterator()
 
         for row in reader:
             try:

src/integrations/tasks.py

@@ -18,6 +18,7 @@
     simkl,
     steam,
     trakt,
+    watcharr,
     yamtrack,
 )
 
@@ -137,3 +138,9 @@ def import_imdb(file, user_id, mode):
 def import_goodreads(file, user_id, mode):
     """Celery task for importing media data from GoodReads."""
     return import_media(goodreads.importer, file, user_id, mode)
+
+
+@shared_task(name="Import from Watcharr")
+def import_watcharr(file, user_id, mode):
+    """Celery task for importing the Watcharr JSON export."""
+    return import_media(watcharr.importer, file, user_id, mode)

src/integrations/tests/imports/test_watcharr.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+
+from app.models import (
+    TV,
+    Episode,
+    Movie,
+    Season,
+)
+from integrations.imports import (
+    watcharr,
+)
+
+mock_path = Path(__file__).resolve().parent.parent / "mock_data"
+app_mock_path = (
+    Path(__file__).resolve().parent.parent.parent.parent / "app" / "tests" / "mock_data"
+)
+
+
+class ImportWatcharr(TestCase):
+    """Test importing media from Watcharr JSON."""
+
+    def setUp(self):
+        """Create user for the tests."""
+        self.credentials = {"username": "test", "password": "12345"}
+        self.user = get_user_model().objects.create_user(**self.credentials)
+        with Path(mock_path / "import_watcharr.json").open("rb") as file:
+            self.import_results = watcharr.importer(file, self.user, "new")
+
+    def test_import_counts(self):
+        """Test basic counts of imported media."""
+        self.assertEqual(TV.objects.filter(user=self.user).count(), 1)
+        self.assertEqual(Movie.objects.filter(user=self.user).count(), 2)
+        self.assertEqual(Season.objects.filter(user=self.user).count(), 3)
+        self.assertEqual(
+            Episode.objects.filter(related_season__user=self.user).count(),
+            34,
+        )