From 190f85ca9e7b9acdb6b38d49f844c2b583df13f5 Mon Sep 17 00:00:00 2001 From: Kieran Eglin Date: Mon, 8 Apr 2024 14:50:05 -0700 Subject: [PATCH] Hooked up apprise notification module --- .../fast_indexing/fast_indexing_helpers.ex | 4 +- .../fast_indexing/fast_indexing_worker.ex | 12 ++- lib/pinchflat/notifications/command_runner.ex | 6 +- .../notifications/source_notifications.ex | 77 ++++++++++++++ .../media_collection_indexing_worker.ex | 16 ++- .../slow_indexing/slow_indexing_helpers.ex | 8 +- .../fast_indexing_helpers_test.exs | 10 +- .../fast_indexing_worker_test.exs | 25 +++++ .../notifications/command_runner_test.exs | 12 ++- .../source_notifications_test.exs | 100 ++++++++++++++++++ .../media_collection_indexing_worker_test.exs | 33 ++++++ 11 files changed, 287 insertions(+), 16 deletions(-) create mode 100644 lib/pinchflat/notifications/source_notifications.ex create mode 100644 test/pinchflat/notifications/source_notifications_test.exs diff --git a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex b/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex index a38ccf7..b9d5a44 100644 --- a/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex +++ b/lib/pinchflat/fast_indexing/fast_indexing_helpers.ex @@ -25,7 +25,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do function starts individual indexing tasks for each new media item. I think it does make sense grammatically, but I could see how that's confusing. - Returns :ok + Returns [binary()] where each binary is the media ID of a new media item. """ def kickoff_indexing_tasks_from_youtube_rss_feed(%Source{} = source) do {:ok, media_ids} = YoutubeRss.get_recent_media_ids_from_rss(source) @@ -37,6 +37,8 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpers do MediaIndexingWorker.kickoff_with_task(source, url) end) + + new_media_ids end @doc """ diff --git a/lib/pinchflat/fast_indexing/fast_indexing_worker.ex b/lib/pinchflat/fast_indexing/fast_indexing_worker.ex index 1e0f700..15d287f 100644 --- a/lib/pinchflat/fast_indexing/fast_indexing_worker.ex +++ b/lib/pinchflat/fast_indexing/fast_indexing_worker.ex @@ -11,8 +11,10 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do alias __MODULE__ alias Pinchflat.Tasks alias Pinchflat.Sources + alias Pinchflat.Settings alias Pinchflat.Sources.Source alias Pinchflat.FastIndexing.FastIndexingHelpers + alias Pinchflat.Notifications.SourceNotifications @doc """ Starts the source fast indexing worker and creates a task for the source. @@ -37,8 +39,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do source = Sources.get_source!(source_id) if source.fast_index do - FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source) - + perform_indexing_and_notification(source) reschedule_indexing(source) else :ok @@ -48,6 +49,13 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorker do Ecto.StaleEntryError -> Logger.info("#{__MODULE__} discarded: source #{source_id} stale") end + defp perform_indexing_and_notification(source) do + apprise_server = Settings.get!(:apprise_server) + new_media_items = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source) + + SourceNotifications.send_new_media_notification(apprise_server, source, length(new_media_items)) + end + defp reschedule_indexing(source) do next_run_in = Source.fast_index_frequency() * 60 diff --git a/lib/pinchflat/notifications/command_runner.ex b/lib/pinchflat/notifications/command_runner.ex index 3a3005f..edea85c 100644 --- a/lib/pinchflat/notifications/command_runner.ex +++ b/lib/pinchflat/notifications/command_runner.ex @@ -16,9 +16,13 @@ defmodule Pinchflat.Notifications.CommandRunner do Can take a single server string or a list of servers as well as additional arguments to pass to the command. - Returns {:ok, binary()} | {:error, binary()}. + Returns {:ok, binary()} | {:error, :no_servers} | {:error, binary()} """ @impl AppriseCommandRunner + def run(nil, _), do: {:error, :no_servers} + def run("", _), do: {:error, :no_servers} + def run([], _), do: {:error, :no_servers} + def run(endpoints, args) do endpoints = List.wrap(endpoints) parsed_args = CliUtils.parse_options(args) diff --git a/lib/pinchflat/notifications/source_notifications.ex b/lib/pinchflat/notifications/source_notifications.ex new file mode 100644 index 0000000..9ecb5c9 --- /dev/null +++ b/lib/pinchflat/notifications/source_notifications.ex @@ -0,0 +1,77 @@ +defmodule Pinchflat.Notifications.SourceNotifications do + @moduledoc """ + Contains utilities for sending notifications about sources + """ + + require Logger + + alias Pinchflat.Repo + alias Pinchflat.Media.MediaQuery + + @doc """ + Wraps a function that may change the number of pending or downloaded + media items for a source, sending an apprise notification if + the count changes. + + Returns the return value of the provided function + """ + def wrap_new_media_notification(servers, source, func) do + before_count = relevant_media_item_count(source) + retval = func.() + after_count = relevant_media_item_count(source) + + send_new_media_notification(servers, source, after_count - before_count) + + retval + end + + @doc """ + Sends a notification if the count of new media items has changed + + Returns :ok + """ + def send_new_media_notification(_, _, count) when count <= 0, do: :ok + + def send_new_media_notification(servers, source, changed_count) do + opts = [ + title: "[Pinchflat] New media found!", + body: "Found #{changed_count} new media item(s) for #{source.custom_name}. Working on downloading them now!" + ] + + case backend_runner().run(servers, opts) do + {:ok, _} -> + Logger.info("Sent new media notification for source #{source.id}") + + {:error, :no_servers} -> + Logger.info("No notification servers provided for source #{source.id}") + + {:error, err} -> + Logger.error("Failed to send new media notification for source #{source.id}: #{err}") + end + + :ok + end + + defp relevant_media_item_count(source) do + pending_media_item_count(source) + downloaded_media_item_count(source) + end + + defp pending_media_item_count(source) do + MediaQuery.new() + |> MediaQuery.for_source(source) + |> MediaQuery.with_media_pending_download() + |> Repo.aggregate(:count) + end + + defp downloaded_media_item_count(source) do + MediaQuery.new() + |> MediaQuery.for_source(source) + |> MediaQuery.with_media_filepath() + |> Repo.aggregate(:count) + end + + defp backend_runner do + # This approach lets us mock the command for testing + Application.get_env(:pinchflat, :apprise_runner) + end +end diff --git a/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex b/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex index 70f9c2e..0e982eb 100644 --- a/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex +++ b/lib/pinchflat/slow_indexing/media_collection_indexing_worker.ex @@ -11,9 +11,11 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do alias __MODULE__ alias Pinchflat.Tasks alias Pinchflat.Sources + alias Pinchflat.Settings alias Pinchflat.Sources.Source alias Pinchflat.FastIndexing.FastIndexingWorker alias Pinchflat.SlowIndexing.SlowIndexingHelpers + alias Pinchflat.Notifications.SourceNotifications @doc """ Starts the source slow indexing worker and creates a task for the source. @@ -78,21 +80,21 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do case {source.index_frequency_minutes, source.last_indexed_at} do {index_freq, _} when index_freq > 0 -> # If the indexing is on a schedule simply run indexing and reschedule - SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source) + perform_indexing_and_notification(source) maybe_enqueue_fast_indexing_task(source) reschedule_indexing(source) {_, nil} -> # If the source has never been indexed, index it once # even if it's not meant to reschedule - SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source) + perform_indexing_and_notification(source) :ok _ -> # If the source HAS been indexed and is not meant to reschedule, # perform a no-op (unless forced) if args["force"] do - SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source) + perform_indexing_and_notification(source) end :ok @@ -102,6 +104,14 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorker do Ecto.StaleEntryError -> Logger.info("#{__MODULE__} discarded: source #{source_id} stale") end + defp perform_indexing_and_notification(source) do + apprise_server = Settings.get!(:apprise_server) + + SourceNotifications.wrap_new_media_notification(apprise_server, source, fn -> + SlowIndexingHelpers.index_and_enqueue_download_for_media_items(source) + end) + end + defp reschedule_indexing(source) do next_run_in = source.index_frequency_minutes * 60 diff --git a/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex b/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex index c95306c..b6b118a 100644 --- a/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex +++ b/lib/pinchflat/slow_indexing/slow_indexing_helpers.ex @@ -60,7 +60,7 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do def index_and_enqueue_download_for_media_items(%Source{} = source) do # See the method definition below for more info on how file watchers work # (important reading if you're not familiar with it) - {:ok, media_attributes} = get_media_attributes_for_collection_and_setup_file_watcher(source) + {:ok, media_attributes} = setup_file_watcher_and_kickoff_indexing(source) # Reload because the source may have been updated during the (long-running) indexing process # and important settings like `download_media` may have changed. source = Repo.reload!(source) @@ -84,15 +84,15 @@ defmodule Pinchflat.SlowIndexing.SlowIndexingHelpers do # lines (ie: you should gracefully fail if you can't parse a line). # # This works in-tandem with the normal (blocking) media indexing behaviour. When - # the `get_media_attributes_for_collection` method completes it'll return the FULL result to - # the caller for parsing. Ideally, every item in the list will have already + # the `setup_file_watcher_and_kickoff_indexing` method completes it'll return the + # FULL result to the caller for parsing. Ideally, every item in the list will have already # been processed by the file follower, but if not, the caller handles creation # of any media items that were missed/initially failed. # # It attempts a graceful shutdown of the file follower after the indexing is done, # but the FileFollowerServer will also stop itself if it doesn't see any activity # for a sufficiently long time. - defp get_media_attributes_for_collection_and_setup_file_watcher(source) do + defp setup_file_watcher_and_kickoff_indexing(source) do {:ok, pid} = FileFollowerServer.start_link() handler = fn filepath -> setup_file_follower_watcher(pid, filepath, source) end diff --git a/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs b/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs index 6c55051..23350c0 100644 --- a/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs +++ b/test/pinchflat/fast_indexing/fast_indexing_helpers_test.exs @@ -24,7 +24,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpersTest do test "enqueues a new worker for each new media_id in the source's RSS feed", %{source: source} do expect(HTTPClientMock, :get, fn _url -> {:ok, "test_1"} end) - assert :ok = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source) + assert [_] = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source) assert [worker] = all_enqueued(worker: MediaIndexingWorker) assert worker.args["id"] == source.id @@ -35,10 +35,16 @@ defmodule Pinchflat.FastIndexing.FastIndexingHelpersTest do expect(HTTPClientMock, :get, fn _url -> {:ok, "test_1"} end) media_item_fixture(source_id: source.id, media_id: "test_1") - assert :ok = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source) + assert [] = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source) refute_enqueued(worker: MediaIndexingWorker) end + + test "returns the IDs of the found media items", %{source: source} do + expect(HTTPClientMock, :get, fn _url -> {:ok, "test_1"} end) + + assert ["test_1"] = FastIndexingHelpers.kickoff_indexing_tasks_from_youtube_rss_feed(source) + end end describe "index_and_enqueue_download_for_media_item/2" do diff --git a/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs b/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs index a69a99d..55007ba 100644 --- a/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs +++ b/test/pinchflat/fast_indexing/fast_indexing_worker_test.exs @@ -4,6 +4,7 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorkerTest do import Mox import Pinchflat.SourcesFixtures + alias Pinchflat.Settings alias Pinchflat.Sources.Source alias Pinchflat.FastIndexing.FastIndexingWorker @@ -74,4 +75,28 @@ defmodule Pinchflat.FastIndexing.FastIndexingWorkerTest do assert :ok = perform_job(FastIndexingWorker, %{id: 0}) end end + + describe "perform/1 when testing notifications" do + setup do + Settings.set(apprise_server: "server_1") + + :ok + end + + test "sends a notification if new media was found" do + source = source_fixture(fast_index: true) + + expect(HTTPClientMock, :get, fn _url -> {:ok, "test_1"} end) + + expect(AppriseRunnerMock, :run, fn servers, opts -> + assert "server_1" = servers + assert is_binary(Keyword.get(opts, :title)) + assert is_binary(Keyword.get(opts, :body)) + + {:ok, ""} + end) + + perform_job(FastIndexingWorker, %{id: source.id}) + end + end end diff --git a/test/pinchflat/notifications/command_runner_test.exs b/test/pinchflat/notifications/command_runner_test.exs index 658dfed..0ed4cc6 100644 --- a/test/pinchflat/notifications/command_runner_test.exs +++ b/test/pinchflat/notifications/command_runner_test.exs @@ -11,7 +11,7 @@ defmodule Pinchflat.Notifications.CommandRunnerTest do describe "run/2" do test "returns :ok when the command succeeds" do - assert {:ok, _} = Runner.run("", []) + assert {:ok, _} = Runner.run("server_1", []) end test "includes the servers as the first argument" do @@ -27,16 +27,22 @@ defmodule Pinchflat.Notifications.CommandRunnerTest do end test "passes all arguments to the command" do - assert {:ok, output} = Runner.run("", ["--dry-run"]) + assert {:ok, output} = Runner.run("server_1", ["--dry-run"]) assert String.contains?(output, "--dry-run") end test "returns the output when the command fails" do wrap_executable("/bin/false", fn -> - assert {:error, ""} = Runner.run("", []) + assert {:error, ""} = Runner.run("server_1", []) end) end + + test "returns a relevant error if no servers are provided" do + assert {:error, :no_servers} = Runner.run(nil, []) + assert {:error, :no_servers} = Runner.run("", []) + assert {:error, :no_servers} = Runner.run([], []) + end end describe "version/0" do diff --git a/test/pinchflat/notifications/source_notifications_test.exs b/test/pinchflat/notifications/source_notifications_test.exs new file mode 100644 index 0000000..788ae7f --- /dev/null +++ b/test/pinchflat/notifications/source_notifications_test.exs @@ -0,0 +1,100 @@ +defmodule Pinchflat.Notifications.SourceNotificationsTest do + use Pinchflat.DataCase + + import Mox + import Pinchflat.MediaFixtures + import Pinchflat.SourcesFixtures + + alias Pinchflat.Notifications.SourceNotifications + + @apprise_servers ["server_1", "server_2"] + + setup :verify_on_exit! + + describe "wrap_new_media_notification/3" do + test "sends a notification when the pending count changes" do + source = source_fixture() + + expect(AppriseRunnerMock, :run, fn servers, opts -> + assert servers == @apprise_servers + + assert opts == [ + title: "[Pinchflat] New media found!", + body: "Found 1 new media item(s) for #{source.custom_name}. Working on downloading them now!" + ] + + {:ok, ""} + end) + + SourceNotifications.wrap_new_media_notification(@apprise_servers, source, fn -> + media_item_fixture(%{source_id: source.id, media_filepath: nil}) + end) + end + + test "sends a notification when the downloaded count changes" do + source = source_fixture() + + expect(AppriseRunnerMock, :run, fn servers, opts -> + assert servers == @apprise_servers + + assert opts == [ + title: "[Pinchflat] New media found!", + body: "Found 1 new media item(s) for #{source.custom_name}. Working on downloading them now!" + ] + + {:ok, ""} + end) + + SourceNotifications.wrap_new_media_notification(@apprise_servers, source, fn -> + media_item_fixture(%{source_id: source.id, media_filepath: "file.mp4"}) + end) + end + + test "does not send a notification when the count does not change" do + source = source_fixture() + + expect(AppriseRunnerMock, :run, 0, fn _, _ -> {:ok, ""} end) + + SourceNotifications.wrap_new_media_notification(@apprise_servers, source, fn -> + media_item_fixture(%{source_id: source.id, prevent_download: true, media_filepath: nil}) + end) + end + + test "returns the value of the function" do + source = source_fixture() + expect(AppriseRunnerMock, :run, 0, fn _, _ -> {:ok, ""} end) + + retval = SourceNotifications.wrap_new_media_notification(@apprise_servers, source, fn -> "value" end) + + assert retval == "value" + end + end + + describe "send_new_media_notification/3" do + test "sends a notification when count is positive" do + source = source_fixture() + + expect(AppriseRunnerMock, :run, fn servers, opts -> + assert servers == @apprise_servers + + assert opts == [ + title: "[Pinchflat] New media found!", + body: "Found 1 new media item(s) for #{source.custom_name}. Working on downloading them now!" + ] + + {:ok, ""} + end) + + :ok = SourceNotifications.send_new_media_notification(@apprise_servers, source, 1) + end + + test "does not send a notification when count not positive" do + source = source_fixture() + + expect(AppriseRunnerMock, :run, 0, fn _, _ -> {:ok, ""} end) + + :ok = SourceNotifications.send_new_media_notification(@apprise_servers, source, 0) + :ok = SourceNotifications.send_new_media_notification(@apprise_servers, source, -1) + end + end +end diff --git a/test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs b/test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs index 6a3c140..0830435 100644 --- a/test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs +++ b/test/pinchflat/slow_indexing/media_collection_indexing_worker_test.exs @@ -7,6 +7,7 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorkerTest do import Pinchflat.SourcesFixtures alias Pinchflat.Tasks + alias Pinchflat.Settings alias Pinchflat.Sources.Source alias Pinchflat.FastIndexing.FastIndexingWorker alias Pinchflat.Downloading.MediaDownloadWorker @@ -51,6 +52,12 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorkerTest do end describe "perform/1" do + setup do + stub(AppriseRunnerMock, :run, fn _, _ -> {:ok, ""} end) + + :ok + end + test "it indexes the source if it should be indexed" do expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end) @@ -210,4 +217,30 @@ defmodule Pinchflat.SlowIndexing.MediaCollectionIndexingWorkerTest do assert :ok = perform_job(MediaCollectionIndexingWorker, %{id: 0}) end end + + describe "perform/1 when testing apprise notifications" do + setup do + Settings.set(apprise_server: "server_1") + + :ok + end + + test "sends a notification if new media was found" do + source = source_fixture() + + expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> + {:ok, source_attributes_return_fixture()} + end) + + expect(AppriseRunnerMock, :run, fn servers, opts -> + assert "server_1" = servers + assert is_binary(Keyword.get(opts, :title)) + assert is_binary(Keyword.get(opts, :body)) + + {:ok, ""} + end) + + perform_job(MediaCollectionIndexingWorker, %{id: source.id}) + end + end end