Lays the groundwork for fast indexing
This commit is contained in:
parent
c06487755b
commit
de18b67480
19 changed files with 358 additions and 113 deletions
|
|
@ -40,7 +40,13 @@ config :pinchflat, Oban,
|
|||
# Keep old jobs for 30 days for display in the UI
|
||||
plugins: [{Oban.Plugins.Pruner, max_age: 30 * 24 * 60 * 60}],
|
||||
# TODO: consider making this an env var or something?
|
||||
queues: [default: 10, media_indexing: 2, media_fetching: 2, media_local_metadata: 8]
|
||||
queues: [
|
||||
default: 10,
|
||||
media_indexing: 2,
|
||||
media_collection_indexing: 2,
|
||||
media_fetching: 2,
|
||||
media_local_metadata: 8
|
||||
]
|
||||
|
||||
# Configures the mailer
|
||||
#
|
||||
|
|
|
|||
|
|
@ -156,7 +156,9 @@ defmodule Pinchflat.Media do
|
|||
end
|
||||
|
||||
@doc """
|
||||
Creates a media_item. Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}.
|
||||
Creates a media_item.
|
||||
|
||||
Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}
|
||||
"""
|
||||
def create_media_item(attrs) do
|
||||
%MediaItem{}
|
||||
|
|
@ -165,7 +167,28 @@ defmodule Pinchflat.Media do
|
|||
end
|
||||
|
||||
@doc """
|
||||
Updates a media_item. Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}.
|
||||
Creates a media item from the attributes returned by the video backend
|
||||
(read: yt-dlp)
|
||||
|
||||
Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}
|
||||
"""
|
||||
def create_media_item_from_backend_attrs(source, media_attrs) do
|
||||
attrs = %{
|
||||
source_id: source.id,
|
||||
title: media_attrs["title"],
|
||||
media_id: media_attrs["id"],
|
||||
original_url: media_attrs["original_url"],
|
||||
livestream: media_attrs["was_live"],
|
||||
description: media_attrs["description"]
|
||||
}
|
||||
|
||||
create_media_item(attrs)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Updates a media_item.
|
||||
|
||||
Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}
|
||||
"""
|
||||
def update_media_item(%MediaItem{} = media_item, attrs) do
|
||||
media_item
|
||||
|
|
@ -177,7 +200,7 @@ defmodule Pinchflat.Media do
|
|||
Deletes a media_item and its associated tasks.
|
||||
Can optionally delete the media_item's files.
|
||||
|
||||
Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}.
|
||||
Returns {:ok, %MediaItem{}} | {:error, %Ecto.Changeset{}}
|
||||
"""
|
||||
def delete_media_item(%MediaItem{} = media_item, opts \\ []) do
|
||||
delete_files = Keyword.get(opts, :delete_files, false)
|
||||
|
|
|
|||
|
|
@ -23,7 +23,6 @@ defmodule Pinchflat.Metadata.MetadataParser do
|
|||
|> Map.merge(parse_infojson_metadata(metadata))
|
||||
end
|
||||
|
||||
# TODO: test new
|
||||
defp parse_media_metadata(metadata) do
|
||||
%{
|
||||
media_id: metadata["id"],
|
||||
|
|
|
|||
|
|
@ -190,7 +190,7 @@ defmodule Pinchflat.Sources do
|
|||
%{__meta__: %{state: :loaded}} ->
|
||||
case changeset.changes do
|
||||
%{index_frequency_minutes: mins} when mins > 0 -> SourceTasks.kickoff_indexing_task(source)
|
||||
%{index_frequency_minutes: _} -> Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker")
|
||||
%{index_frequency_minutes: _} -> Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker")
|
||||
_ -> :ok
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -6,6 +6,11 @@ defmodule Pinchflat.Tasks.MediaItemTasks do
|
|||
do is also defined here. Essentially, a one-stop-shop for media-related tasks/workers.
|
||||
"""
|
||||
alias Pinchflat.Media
|
||||
alias Pinchflat.Tasks
|
||||
alias Pinchflat.Sources.Source
|
||||
alias Pinchflat.Workers.MediaDownloadWorker
|
||||
|
||||
alias Pinchflat.YtDlp.Backend.Media, as: YtDlpMedia
|
||||
|
||||
@doc """
|
||||
Fetches the file size of a media item and saves it to the database.
|
||||
|
|
@ -21,4 +26,36 @@ defmodule Pinchflat.Tasks.MediaItemTasks do
|
|||
err
|
||||
end
|
||||
end
|
||||
|
||||
@doc """
|
||||
Indexes a single media item for a source and enqueues a download job if the
|
||||
media should be downloaded. This method creates the media item record so it's
|
||||
the one-stop-shop for adding a media item (and possibly downloading it) just
|
||||
by a URL and source.
|
||||
|
||||
Returns {:ok, media_item} | {:error, any()}
|
||||
"""
|
||||
def index_and_enqueue_download_for_media_item(%Source{} = source, url) do
|
||||
maybe_media_item = create_media_item_from_url(source, url)
|
||||
|
||||
case maybe_media_item do
|
||||
{:ok, media_item} ->
|
||||
if source.download_media && Media.pending_download?(media_item) do
|
||||
%{id: media_item.id}
|
||||
|> MediaDownloadWorker.new()
|
||||
|> Tasks.create_job_with_task(media_item)
|
||||
end
|
||||
|
||||
{:ok, media_item}
|
||||
|
||||
err ->
|
||||
err
|
||||
end
|
||||
end
|
||||
|
||||
defp create_media_item_from_url(source, url) do
|
||||
{:ok, media_attrs} = YtDlpMedia.get_media_attributes(url)
|
||||
|
||||
Media.create_media_item_from_backend_attrs(source, media_attrs)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -13,9 +13,9 @@ defmodule Pinchflat.Tasks.SourceTasks do
|
|||
alias Pinchflat.Sources
|
||||
alias Pinchflat.Sources.Source
|
||||
alias Pinchflat.Media.MediaItem
|
||||
alias Pinchflat.Workers.MediaIndexingWorker
|
||||
alias Pinchflat.Workers.MediaDownloadWorker
|
||||
alias Pinchflat.YtDlp.Backend.MediaCollection
|
||||
alias Pinchflat.Workers.MediaCollectionIndexingWorker
|
||||
alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer
|
||||
|
||||
@doc """
|
||||
|
|
@ -25,12 +25,11 @@ defmodule Pinchflat.Tasks.SourceTasks do
|
|||
Returns {:ok, %Task{}}.
|
||||
"""
|
||||
def kickoff_indexing_task(%Source{} = source) do
|
||||
Tasks.delete_pending_tasks_for(source, "MediaIndexingWorker")
|
||||
Tasks.delete_pending_tasks_for(source, "MediaCollectionIndexingWorker")
|
||||
|
||||
source
|
||||
|> Map.take([:id])
|
||||
%{id: source.id}
|
||||
# Schedule this one immediately, but future ones will be on an interval
|
||||
|> MediaIndexingWorker.new()
|
||||
|> MediaCollectionIndexingWorker.new()
|
||||
|> Tasks.create_job_with_task(source)
|
||||
|> case do
|
||||
# This should never return {:error, :duplicate_job} since we just deleted
|
||||
|
|
@ -89,8 +88,7 @@ defmodule Pinchflat.Tasks.SourceTasks do
|
|||
source
|
||||
|> Media.list_pending_media_items_for()
|
||||
|> Enum.each(fn media_item ->
|
||||
media_item
|
||||
|> Map.take([:id])
|
||||
%{id: media_item.id}
|
||||
|> MediaDownloadWorker.new()
|
||||
|> Tasks.create_job_with_task(media_item)
|
||||
end)
|
||||
|
|
@ -159,8 +157,7 @@ defmodule Pinchflat.Tasks.SourceTasks do
|
|||
if source.download_media && Media.pending_download?(media_item) do
|
||||
Logger.debug("FileFollowerServer Handler: Enqueuing download task for #{inspect(media_attrs)}")
|
||||
|
||||
media_item
|
||||
|> Map.take([:id])
|
||||
%{id: media_item.id}
|
||||
|> MediaDownloadWorker.new()
|
||||
|> Tasks.create_job_with_task(media_item)
|
||||
end
|
||||
|
|
@ -171,16 +168,7 @@ defmodule Pinchflat.Tasks.SourceTasks do
|
|||
end
|
||||
|
||||
defp create_media_item_from_attributes(source, media_attrs) do
|
||||
attrs = %{
|
||||
source_id: source.id,
|
||||
title: media_attrs["title"],
|
||||
media_id: media_attrs["id"],
|
||||
original_url: media_attrs["original_url"],
|
||||
livestream: media_attrs["was_live"],
|
||||
description: media_attrs["description"]
|
||||
}
|
||||
|
||||
case Media.create_media_item(attrs) do
|
||||
case Media.create_media_item_from_backend_attrs(source, media_attrs) do
|
||||
{:ok, media_item} -> media_item
|
||||
{:error, changeset} -> changeset
|
||||
end
|
||||
|
|
|
|||
71
lib/pinchflat/workers/media_collection_indexing_worker.ex
Normal file
71
lib/pinchflat/workers/media_collection_indexing_worker.ex
Normal file
|
|
@ -0,0 +1,71 @@
|
|||
defmodule Pinchflat.Workers.MediaCollectionIndexingWorker do
|
||||
@moduledoc false
|
||||
|
||||
use Oban.Worker,
|
||||
queue: :media_collection_indexing,
|
||||
unique: [period: :infinity, states: [:available, :scheduled, :retryable]],
|
||||
tags: ["media_source", "media_collection_indexing"]
|
||||
|
||||
alias __MODULE__
|
||||
alias Pinchflat.Tasks
|
||||
alias Pinchflat.Sources
|
||||
alias Pinchflat.Tasks.SourceTasks
|
||||
|
||||
@impl Oban.Worker
|
||||
@doc """
|
||||
The ID is that of a source _record_, not a YouTube channel/playlist ID. Indexes
|
||||
the provided source, kicks off downloads for each new MediaItem, and
|
||||
reschedules the job to run again in the future. It will ALWAYS index a source
|
||||
if it's never been indexed before, but rescheduling is determined by the
|
||||
`index_frequency_minutes` field.
|
||||
|
||||
README: Re-scheduling here works a little different than you may expect.
|
||||
The reschedule time is relative to the time the job has actually _completed_.
|
||||
This has some benefits but also side effects to be aware of:
|
||||
|
||||
- Benefit: No chance for jobs to overlap if a job takes longer than the
|
||||
scheduled interval. Less likely to hit API rate limits.
|
||||
- Side effect: Intervals are "soft" and _always_ walk forward. This may cause
|
||||
user confusion since a 30-minute job scheduled for every hour will
|
||||
actually run every 1 hour and 30 minutes. The tradeoff of not inundating
|
||||
the API with requests and also not overlapping jobs is worth it, IMO.
|
||||
|
||||
NOTE: Since indexing can take a LONG time, I should check what happens if an
|
||||
application restart occurs while a job is running. Will the job be lost?
|
||||
|
||||
IDEA: Should I use paging and do indexing in chunks? Is that even faster?
|
||||
|
||||
Returns :ok | {:ok, %Task{}}
|
||||
"""
|
||||
def perform(%Oban.Job{args: %{"id" => source_id}}) do
|
||||
source = Sources.get_source!(source_id)
|
||||
|
||||
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
|
||||
SourceTasks.index_and_enqueue_download_for_media_items(source)
|
||||
reschedule_indexing(source)
|
||||
|
||||
{_, nil} ->
|
||||
# If the source has never been indexed, index it once
|
||||
# even if it's not meant to reschedule
|
||||
SourceTasks.index_and_enqueue_download_for_media_items(source)
|
||||
:ok
|
||||
|
||||
_ ->
|
||||
# If the source HAS been indexed and is not meant to reschedule,
|
||||
# perform a no-op
|
||||
:ok
|
||||
end
|
||||
end
|
||||
|
||||
defp reschedule_indexing(source) do
|
||||
%{id: source.id}
|
||||
|> MediaCollectionIndexingWorker.new(schedule_in: source.index_frequency_minutes * 60)
|
||||
|> Tasks.create_job_with_task(source)
|
||||
|> case do
|
||||
{:ok, task} -> {:ok, task}
|
||||
{:error, :duplicate_job} -> {:ok, :job_exists}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -45,8 +45,7 @@ defmodule Pinchflat.Workers.MediaDownloadWorker do
|
|||
end
|
||||
|
||||
defp schedule_filesystem_data_worker(media_item) do
|
||||
media_item
|
||||
|> Map.take([:id])
|
||||
%{id: media_item.id}
|
||||
|> FilesystemDataWorker.new()
|
||||
|> Tasks.create_job_with_task(media_item)
|
||||
|> case do
|
||||
|
|
|
|||
|
|
@ -4,69 +4,26 @@ defmodule Pinchflat.Workers.MediaIndexingWorker do
|
|||
use Oban.Worker,
|
||||
queue: :media_indexing,
|
||||
unique: [period: :infinity, states: [:available, :scheduled, :retryable]],
|
||||
tags: ["media_source", "media_indexing"]
|
||||
tags: ["media_source", "media_collection_indexing"]
|
||||
|
||||
alias __MODULE__
|
||||
alias Pinchflat.Tasks
|
||||
alias Pinchflat.Sources
|
||||
alias Pinchflat.Tasks.SourceTasks
|
||||
|
||||
@impl Oban.Worker
|
||||
@doc """
|
||||
The ID is that of a source _record_, not a YouTube channel/playlist ID. Indexes
|
||||
the provided source, kicks off downloads for each new MediaItem, and
|
||||
reschedules the job to run again in the future. It will ALWAYS index a source
|
||||
if it's never been indexed before, but rescheduling is determined by the
|
||||
`index_frequency_minutes` field.
|
||||
Similar to `MediaCollectionIndexingWorker`, but for individual media items.
|
||||
Does not reschedule or check anything to do with a source's indexing
|
||||
frequency - only collects initial metadata then kicks off a download.
|
||||
`MediaCollectionIndexingWorker` should be preferred in general, but this is
|
||||
useful for downloading one-off media items based on a URL (like for fast indexing).
|
||||
|
||||
README: Re-scheduling here works a little different than you may expect.
|
||||
The reschedule time is relative to the time the job has actually _completed_.
|
||||
This has some benefits but also side effects to be aware of:
|
||||
Only downloads media that _should_ be downloaded (ie: the source is set to download
|
||||
and the media matches the profile's format preferences). Splits downloading into
|
||||
another job to keep the indexing queue moving quickly.
|
||||
|
||||
- Benefit: No chance for jobs to overlap if a job takes longer than the
|
||||
scheduled interval. Less likely to hit API rate limits.
|
||||
- Side effect: Intervals are "soft" and _always_ walk forward. This may cause
|
||||
user confusion since a 30-minute job scheduled for every hour will
|
||||
actually run every 1 hour and 30 minutes. The tradeoff of not inundating
|
||||
the API with requests and also not overlapping jobs is worth it, IMO.
|
||||
|
||||
NOTE: Since indexing can take a LONG time, I should check what happens if an
|
||||
application restart occurs while a job is running. Will the job be lost?
|
||||
|
||||
IDEA: Should I use paging and do indexing in chunks? Is that even faster?
|
||||
|
||||
Returns :ok | {:ok, %Task{}}
|
||||
Returns :ok
|
||||
"""
|
||||
def perform(%Oban.Job{args: %{"id" => source_id}}) do
|
||||
source = Sources.get_source!(source_id)
|
||||
|
||||
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
|
||||
SourceTasks.index_and_enqueue_download_for_media_items(source)
|
||||
reschedule_indexing(source)
|
||||
|
||||
{_, nil} ->
|
||||
# If the source has never been indexed, index it once
|
||||
# even if it's not meant to reschedule
|
||||
SourceTasks.index_and_enqueue_download_for_media_items(source)
|
||||
:ok
|
||||
|
||||
_ ->
|
||||
# If the source HAS been indexed and is not meant to reschedule,
|
||||
# perform a no-op
|
||||
:ok
|
||||
end
|
||||
end
|
||||
|
||||
defp reschedule_indexing(source) do
|
||||
source
|
||||
|> Map.take([:id])
|
||||
|> MediaIndexingWorker.new(schedule_in: source.index_frequency_minutes * 60)
|
||||
|> Tasks.create_job_with_task(source)
|
||||
|> case do
|
||||
{:ok, task} -> {:ok, task}
|
||||
{:error, :duplicate_job} -> {:ok, :job_exists}
|
||||
end
|
||||
def perform(%Oban.Job{args: %{"id" => source_id, "media_url" => _media_url}}) do
|
||||
_source = Sources.get_source!(source_id)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -20,20 +20,30 @@ defmodule Pinchflat.YtDlp.Backend.Media do
|
|||
end
|
||||
end
|
||||
|
||||
# TODO: test
|
||||
@doc """
|
||||
Returns a map representing the media at the given URL.
|
||||
|
||||
IDEA: should I return a struct here? I want these methods to be agnostic
|
||||
to implementation, but maybe it can specify its own contract by
|
||||
returning a struct with well-known fields. Would make refactoring
|
||||
turbo easy if a field needs to exist but its behavior changes slightly.
|
||||
|
||||
Returns {:ok, [map()]} | {:error, any, ...}.
|
||||
"""
|
||||
def get_media_attributes(url) do
|
||||
runner = Application.get_env(:pinchflat, :yt_dlp_runner)
|
||||
command_opts = [:simulate, :skip_download]
|
||||
output_template = indexing_output_template()
|
||||
|
||||
case runner.run(url, command_opts, output_template) do
|
||||
{:ok, output} -> Phoenix.json_library().decode!(output)
|
||||
{:ok, output} -> Phoenix.json_library().decode(output)
|
||||
res -> res
|
||||
end
|
||||
end
|
||||
|
||||
# TODO: test
|
||||
# TODO: test that media_collection consumes this maybe?
|
||||
@doc """
|
||||
Returns the output template for yt-dlp's indexing command.
|
||||
"""
|
||||
def indexing_output_template do
|
||||
"%(.{id,title,was_live,original_url,description})j"
|
||||
end
|
||||
|
|
|
|||
|
|
@ -368,6 +368,20 @@ defmodule Pinchflat.MediaTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "create_media_item_from_backend_attrs/2" do
|
||||
test "creates a media item for a given source and attributes" do
|
||||
source = source_fixture()
|
||||
media_attrs = Phoenix.json_library().decode!(media_attributes_return_fixture())
|
||||
|
||||
assert {:ok, %MediaItem{} = media_item} = Media.create_media_item_from_backend_attrs(source, media_attrs)
|
||||
assert media_item.source_id == source.id
|
||||
assert media_item.title == media_attrs["title"]
|
||||
assert media_item.media_id == media_attrs["id"]
|
||||
assert media_item.original_url == media_attrs["original_url"]
|
||||
assert media_item.description == media_attrs["description"]
|
||||
end
|
||||
end
|
||||
|
||||
describe "update_media_item/2" do
|
||||
test "updating with valid data updates the media_item" do
|
||||
media_item = media_item_fixture()
|
||||
|
|
|
|||
|
|
@ -33,13 +33,31 @@ defmodule Pinchflat.YtDlp.Backend.MediaParserTest do
|
|||
test "it extracts the title", %{metadata: metadata} do
|
||||
result = Parser.parse_for_media_item(metadata)
|
||||
|
||||
assert result.title == "Trying to Wheelie Without the Rear Brake"
|
||||
assert result.title == metadata["title"]
|
||||
end
|
||||
|
||||
test "it extracts the description", %{metadata: metadata} do
|
||||
result = Parser.parse_for_media_item(metadata)
|
||||
|
||||
assert is_binary(result.description)
|
||||
assert result.description == metadata["description"]
|
||||
end
|
||||
|
||||
test "it extracts the original_url", %{metadata: metadata} do
|
||||
result = Parser.parse_for_media_item(metadata)
|
||||
|
||||
assert result.original_url == metadata["original_url"]
|
||||
end
|
||||
|
||||
test "it extracts the media_id", %{metadata: metadata} do
|
||||
result = Parser.parse_for_media_item(metadata)
|
||||
|
||||
assert result.media_id == metadata["id"]
|
||||
end
|
||||
|
||||
test "it extracts the livestream flag", %{metadata: metadata} do
|
||||
result = Parser.parse_for_media_item(metadata)
|
||||
|
||||
assert result.livestream == metadata["was_live"]
|
||||
end
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ defmodule Pinchflat.SourcesTest do
|
|||
alias Pinchflat.Sources
|
||||
alias Pinchflat.Tasks.SourceTasks
|
||||
alias Pinchflat.Sources.Source
|
||||
alias Pinchflat.Workers.MediaIndexingWorker
|
||||
alias Pinchflat.Workers.MediaCollectionIndexingWorker
|
||||
alias Pinchflat.Workers.MediaDownloadWorker
|
||||
|
||||
@invalid_source_attrs %{name: nil, collection_id: nil}
|
||||
|
|
@ -166,7 +166,7 @@ defmodule Pinchflat.SourcesTest do
|
|||
|
||||
assert {:ok, %Source{} = source} = Sources.create_source(valid_attrs)
|
||||
|
||||
assert_enqueued(worker: MediaIndexingWorker, args: %{"id" => source.id})
|
||||
assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
|
||||
end
|
||||
|
||||
test "creation schedules an index test even if the index frequency is 0" do
|
||||
|
|
@ -180,7 +180,7 @@ defmodule Pinchflat.SourcesTest do
|
|||
|
||||
assert {:ok, %Source{} = source} = Sources.create_source(valid_attrs)
|
||||
|
||||
assert_enqueued(worker: MediaIndexingWorker, args: %{"id" => source.id})
|
||||
assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
|
||||
end
|
||||
end
|
||||
|
||||
|
|
@ -230,7 +230,7 @@ defmodule Pinchflat.SourcesTest do
|
|||
|
||||
assert {:ok, %Source{} = source} = Sources.update_source(source, update_attrs)
|
||||
assert source.index_frequency_minutes == 123
|
||||
assert_enqueued(worker: MediaIndexingWorker, args: %{"id" => source.id})
|
||||
assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
|
||||
end
|
||||
|
||||
test "updating the index frequency to 0 will not re-schedule the indexing task" do
|
||||
|
|
@ -239,12 +239,12 @@ defmodule Pinchflat.SourcesTest do
|
|||
|
||||
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
|
||||
|
||||
refute_enqueued(worker: MediaIndexingWorker, args: %{"id" => source.id})
|
||||
refute_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
|
||||
end
|
||||
|
||||
test "updating the index frequency to 0 will delete any pending tasks" do
|
||||
source = source_fixture()
|
||||
{:ok, job} = Oban.insert(MediaIndexingWorker.new(%{"id" => source.id}))
|
||||
{:ok, job} = Oban.insert(MediaCollectionIndexingWorker.new(%{"id" => source.id}))
|
||||
task = task_fixture(source_id: source.id, job_id: job.id)
|
||||
update_attrs = %{index_frequency_minutes: 0}
|
||||
|
||||
|
|
@ -261,7 +261,7 @@ defmodule Pinchflat.SourcesTest do
|
|||
assert {:ok, %Source{}} = Sources.update_source(source, update_attrs)
|
||||
|
||||
assert Repo.reload!(task)
|
||||
refute_enqueued(worker: MediaIndexingWorker, args: %{"id" => source.id})
|
||||
refute_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
|
||||
end
|
||||
|
||||
test "enabling the download_media attribute will schedule a download task" do
|
||||
|
|
|
|||
|
|
@ -1,9 +1,19 @@
|
|||
defmodule Pinchflat.Tasks.MediaItemTasksTest do
|
||||
use Pinchflat.DataCase
|
||||
|
||||
import Mox
|
||||
import Pinchflat.MediaFixtures
|
||||
import Pinchflat.SourcesFixtures
|
||||
import Pinchflat.ProfilesFixtures
|
||||
|
||||
alias Pinchflat.Tasks
|
||||
alias Pinchflat.Media.MediaItem
|
||||
alias Pinchflat.Tasks.MediaItemTasks
|
||||
alias Pinchflat.Workers.MediaDownloadWorker
|
||||
|
||||
setup :verify_on_exit!
|
||||
|
||||
@media_url "https://www.youtube.com/watch?v=1234"
|
||||
|
||||
describe "compute_and_save_media_filesize/1" do
|
||||
test "updates the media item with the file size" do
|
||||
|
|
@ -22,4 +32,68 @@ defmodule Pinchflat.Tasks.MediaItemTasksTest do
|
|||
assert {:error, _} = MediaItemTasks.compute_and_save_media_filesize(media_item)
|
||||
end
|
||||
end
|
||||
|
||||
describe "index_and_enqueue_download_for_media_item/2" do
|
||||
setup do
|
||||
stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
|
||||
{:ok, media_attributes_return_fixture()}
|
||||
end)
|
||||
|
||||
{:ok, [source: source_fixture()]}
|
||||
end
|
||||
|
||||
test "creates a new media item based on the URL", %{source: source} do
|
||||
assert Repo.aggregate(MediaItem, :count) == 0
|
||||
assert {:ok, _} = MediaItemTasks.index_and_enqueue_download_for_media_item(source, @media_url)
|
||||
assert Repo.aggregate(MediaItem, :count) == 1
|
||||
end
|
||||
|
||||
test "won't duplicate media_items based on media_id and source", %{source: source} do
|
||||
assert {:ok, _} = MediaItemTasks.index_and_enqueue_download_for_media_item(source, @media_url)
|
||||
assert {:error, _} = MediaItemTasks.index_and_enqueue_download_for_media_item(source, @media_url)
|
||||
|
||||
assert Repo.aggregate(MediaItem, :count) == 1
|
||||
end
|
||||
|
||||
test "enqueues a download job", %{source: source} do
|
||||
assert {:ok, media_item} = MediaItemTasks.index_and_enqueue_download_for_media_item(source, @media_url)
|
||||
|
||||
assert_enqueued(worker: MediaDownloadWorker, args: %{"id" => media_item.id})
|
||||
end
|
||||
|
||||
test "creates a download task record", %{source: source} do
|
||||
assert {:ok, media_item} = MediaItemTasks.index_and_enqueue_download_for_media_item(source, @media_url)
|
||||
|
||||
assert [_] = Tasks.list_tasks_for(:media_item_id, media_item.id, "MediaDownloadWorker")
|
||||
end
|
||||
|
||||
test "does not enqueue a download job if the source does not allow it" do
|
||||
source = source_fixture(%{download_media: false})
|
||||
|
||||
assert {:ok, _} = MediaItemTasks.index_and_enqueue_download_for_media_item(source, @media_url)
|
||||
|
||||
refute_enqueued(worker: MediaDownloadWorker)
|
||||
end
|
||||
|
||||
test "does not enqueue a download job if the media item does not match the format rules" do
|
||||
profile = media_profile_fixture(%{shorts_behaviour: :exclude})
|
||||
source = source_fixture(%{media_profile_id: profile.id})
|
||||
|
||||
stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
|
||||
output =
|
||||
Phoenix.json_library().encode!(%{
|
||||
id: "video2",
|
||||
title: "Video 2",
|
||||
original_url: "https://example.com/shorts/video2",
|
||||
was_live: true,
|
||||
description: "desc2"
|
||||
})
|
||||
|
||||
{:ok, output}
|
||||
end)
|
||||
|
||||
assert {:ok, _media_item} = MediaItemTasks.index_and_enqueue_download_for_media_item(source, @media_url)
|
||||
refute_enqueued(worker: MediaDownloadWorker)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
|
|||
alias Pinchflat.Tasks.Task
|
||||
alias Pinchflat.Tasks.SourceTasks
|
||||
alias Pinchflat.Media.MediaItem
|
||||
alias Pinchflat.Workers.MediaIndexingWorker
|
||||
alias Pinchflat.Workers.MediaCollectionIndexingWorker
|
||||
alias Pinchflat.Workers.MediaDownloadWorker
|
||||
|
||||
setup :verify_on_exit!
|
||||
|
|
@ -22,7 +22,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
|
|||
|
||||
assert {:ok, _} = SourceTasks.kickoff_indexing_task(source)
|
||||
|
||||
assert_enqueued(worker: MediaIndexingWorker, args: %{"id" => source.id})
|
||||
assert_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
|
||||
end
|
||||
|
||||
test "it creates and attaches a task" do
|
||||
|
|
@ -35,7 +35,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
|
|||
|
||||
test "it deletes any pending tasks for the source" do
|
||||
source = source_fixture()
|
||||
{:ok, job} = Oban.insert(MediaIndexingWorker.new(%{"id" => source.id}))
|
||||
{:ok, job} = Oban.insert(MediaCollectionIndexingWorker.new(%{"id" => source.id}))
|
||||
task = task_fixture(source_id: source.id, job_id: job.id)
|
||||
|
||||
assert {:ok, _} = SourceTasks.kickoff_indexing_task(source)
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
||||
defmodule Pinchflat.Workers.MediaCollectionIndexingWorkerTest do
|
||||
use Pinchflat.DataCase
|
||||
|
||||
import Mox
|
||||
|
|
@ -6,7 +6,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
import Pinchflat.SourcesFixtures
|
||||
|
||||
alias Pinchflat.Tasks
|
||||
alias Pinchflat.Workers.MediaIndexingWorker
|
||||
alias Pinchflat.Workers.MediaCollectionIndexingWorker
|
||||
alias Pinchflat.Workers.MediaDownloadWorker
|
||||
|
||||
setup :verify_on_exit!
|
||||
|
|
@ -17,7 +17,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
|
||||
source = source_fixture(index_frequency_minutes: 10)
|
||||
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
end
|
||||
|
||||
test "it indexes the source no matter what if the source has never been indexed before" do
|
||||
|
|
@ -25,7 +25,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
|
||||
source = source_fixture(index_frequency_minutes: 0, last_indexed_at: nil)
|
||||
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
end
|
||||
|
||||
test "it does not do any indexing if the source has been indexed and shouldn't be rescheduled" do
|
||||
|
|
@ -33,16 +33,16 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
|
||||
source = source_fixture(index_frequency_minutes: -1, last_indexed_at: DateTime.utc_now())
|
||||
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
end
|
||||
|
||||
test "it does not reschedule if the source shouldn't be indexed" do
|
||||
stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
|
||||
|
||||
source = source_fixture(index_frequency_minutes: -1)
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
|
||||
refute_enqueued(worker: MediaIndexingWorker, args: %{"id" => source.id})
|
||||
refute_enqueued(worker: MediaCollectionIndexingWorker, args: %{"id" => source.id})
|
||||
end
|
||||
|
||||
test "it kicks off a download job for each pending media item" do
|
||||
|
|
@ -51,7 +51,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
end)
|
||||
|
||||
source = source_fixture(index_frequency_minutes: 10)
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
|
||||
assert length(all_enqueued(worker: MediaDownloadWorker)) == 3
|
||||
end
|
||||
|
|
@ -63,7 +63,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
|
||||
source = source_fixture(index_frequency_minutes: 10)
|
||||
media_item_fixture(%{source_id: source.id, media_filepath: nil})
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
|
||||
assert length(all_enqueued(worker: MediaDownloadWorker)) == 4
|
||||
end
|
||||
|
|
@ -75,7 +75,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
|
||||
source = source_fixture(index_frequency_minutes: 10)
|
||||
media_item_fixture(%{source_id: source.id, media_filepath: nil, media_id: "video1"})
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
|
||||
# Only 3 jobs should be enqueued, since the first video is a duplicate
|
||||
assert length(all_enqueued(worker: MediaDownloadWorker))
|
||||
|
|
@ -85,10 +85,10 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
|
||||
|
||||
source = source_fixture(index_frequency_minutes: 10)
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
|
||||
assert_enqueued(
|
||||
worker: MediaIndexingWorker,
|
||||
worker: MediaCollectionIndexingWorker,
|
||||
args: %{"id" => source.id},
|
||||
scheduled_at: now_plus(source.index_frequency_minutes, :minutes)
|
||||
)
|
||||
|
|
@ -101,7 +101,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
task_count_fetcher = fn -> Enum.count(Tasks.list_tasks()) end
|
||||
|
||||
assert_changed([from: 0, to: 1], task_count_fetcher, fn ->
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
end)
|
||||
end
|
||||
|
||||
|
|
@ -118,7 +118,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
|
|||
end
|
||||
|
||||
assert_changed([from: [], to: ["video1", "video2", "video3"]], media_item_fetcher, fn ->
|
||||
perform_job(MediaIndexingWorker, %{id: source.id})
|
||||
perform_job(MediaCollectionIndexingWorker, %{id: source.id})
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
|
@ -3,6 +3,7 @@ defmodule Pinchflat.YtDlp.Backend.MediaCollectionTest do
|
|||
import Mox
|
||||
import Pinchflat.SourcesFixtures
|
||||
|
||||
alias Pinchflat.YtDlp.Backend.Media
|
||||
alias Pinchflat.YtDlp.Backend.MediaCollection
|
||||
|
||||
@channel_url "https://www.youtube.com/c/TheUselessTrials"
|
||||
|
|
@ -22,7 +23,7 @@ defmodule Pinchflat.YtDlp.Backend.MediaCollectionTest do
|
|||
test "it passes the expected default args" do
|
||||
expect(YtDlpRunnerMock, :run, fn _url, opts, ot, _addl_opts ->
|
||||
assert opts == [:simulate, :skip_download]
|
||||
assert ot == "%(.{id,title,was_live,original_url,description})j"
|
||||
assert ot == Media.indexing_output_template()
|
||||
|
||||
{:ok, ""}
|
||||
end)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
defmodule Pinchflat.YtDlp.Backend.MediaTest do
|
||||
use Pinchflat.DataCase
|
||||
import Mox
|
||||
import Pinchflat.MediaFixtures
|
||||
|
||||
alias Pinchflat.YtDlp.Backend.Media
|
||||
|
||||
|
|
@ -47,4 +48,39 @@ defmodule Pinchflat.YtDlp.Backend.MediaTest do
|
|||
assert {:error, "something"} = Media.download(@media_url)
|
||||
end
|
||||
end
|
||||
|
||||
describe "get_media_attributes/1" do
|
||||
test "returns a list of video attributes" do
|
||||
expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
|
||||
{:ok, media_attributes_return_fixture()}
|
||||
end)
|
||||
|
||||
assert {:ok, %{"description" => _, "id" => _, "original_url" => _, "title" => _, "was_live" => _}} =
|
||||
Media.get_media_attributes(@media_url)
|
||||
end
|
||||
|
||||
test "it passes the expected default args" do
|
||||
expect(YtDlpRunnerMock, :run, fn _url, opts, ot ->
|
||||
assert opts == [:simulate, :skip_download]
|
||||
assert ot == Media.indexing_output_template()
|
||||
|
||||
{:ok, "{}"}
|
||||
end)
|
||||
|
||||
assert {:ok, _} = Media.get_media_attributes(@media_url)
|
||||
end
|
||||
|
||||
test "returns the error straight through when the command fails" do
|
||||
expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:error, "Big issue", 1} end)
|
||||
|
||||
assert {:error, "Big issue", 1} = Media.get_media_attributes(@media_url)
|
||||
end
|
||||
end
|
||||
|
||||
describe "indexing_output_template/0" do
|
||||
test "contains all the greatest hits" do
|
||||
assert "%(.{id,title,was_live,original_url,description})j" ==
|
||||
Media.indexing_output_template()
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -65,4 +65,16 @@ defmodule Pinchflat.MediaFixtures do
|
|||
merged_attrs = Map.merge(attrs, %{media_filepath: stored_media_filepath})
|
||||
media_item_fixture(merged_attrs)
|
||||
end
|
||||
|
||||
def media_attributes_return_fixture do
|
||||
media_attributes = %{
|
||||
id: "video1",
|
||||
title: "Video 1",
|
||||
original_url: "https://example.com/video1",
|
||||
was_live: false,
|
||||
description: "desc1"
|
||||
}
|
||||
|
||||
Phoenix.json_library().encode!(media_attributes)
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Reference in a new issue