Files
pinchflat/lib/pinchflat/workers/media_indexing_worker.ex
T
Kieran a5e7c48f70 Download video(s) (first iteration) (#5)
* Updated package.json (and made an excuse to make a branch)

* Video filepath parser (#6)

* Restructured files; Added parser placeholder

* More restructuring

* Added basic parser for hydrating template strings

* Improved docs

* More docs

* Initial implementation of media profiles (#7)

* [WIP] Added basic video download method

* [WIP] Very-WIP first steps at parsing options and downloading

* Made my options safe by default and removed special safe versions

* Ran html generator for mediaprofile model - leaving as-is for now

* Addressed a bunch of TODO comments

* Add "channel" type Media Source (#8)

* [WIP] Working on fetching channel metadata in yt-dlp backend

* Finished first draft of methods to do with querying channels

* Renamed CommandRunnerMock to have a more descriptive name

* Ran the phx generator for the channel model

* Renamed Downloader namespace to MediaClient

* [WIP] saving before attempting LiveView

* LiveView did not work out but here's a working controller how about

* Index a channel (#9)

* Ran a MediaItem generator; Reformatted to my liking

* [WIP] added basic index function

* setup oban

* Added basic Oban job for indexing

* Added in workers for indexing; hooked them into record creation flow

* Added a task model with a phx generator

* Tied together tasks with jobs and channels

* Download indexed videos (#10)

* Clarified documentation

* more comments

* [WIP] hooked up basic video downloading; starting work on metadata

* Added metadata model and parsing

Adding the metadata model made me realize that, in many cases, yt-dlp
returns undesired input in stdout, breaking parsing. In order to get
the metadata model working, I had to change the way in which the app
interacts with yt-dlp. Now, output is written as a file to disk which
is immediately re-read and returned.

* Added tests for video download worker

* Hooked up video downloading to the channel indexing pipeline

* Adds tasks for media items

* Updated video metadata parser to extract the title

* Ran linting
2024-01-30 20:15:59 -08:00

81 lines
2.8 KiB
Elixir

defmodule Pinchflat.Workers.MediaIndexingWorker do
@moduledoc false
use Oban.Worker,
queue: :media_indexing,
unique: [period: :infinity, states: [:available, :scheduled, :retryable]],
tags: ["media_source", "media_indexing"]
alias __MODULE__
alias Pinchflat.Media
alias Pinchflat.Tasks
alias Pinchflat.MediaSource
alias Pinchflat.Workers.VideoDownloadWorker
@impl Oban.Worker
@doc """
The ID is that of a channel _record_, not a YouTube channel ID. Indexes
the provided channel, kicks off downloads for each new MediaItem, and
reschedules the job to run again in the future (as determined by the
channel's `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" => channel_id}}) do
channel = MediaSource.get_channel!(channel_id)
if channel.index_frequency_minutes <= 0 do
:ok
else
index_media_and_reschedule(channel)
end
end
defp index_media_and_reschedule(channel) do
MediaSource.index_media_items(channel)
enqueue_video_downloads(channel)
channel
|> Map.take([:id])
|> MediaIndexingWorker.new(schedule_in: channel.index_frequency_minutes * 60)
|> Tasks.create_job_with_task(channel)
|> case do
{:ok, task} -> {:ok, task}
{:error, :duplicate_job} -> {:ok, :job_exists}
end
end
# NOTE: this starts a download for each media item that is pending,
# not just the ones that were indexed in this job run. This should ensure
# that any stragglers are caught if, for some reason, they weren't enqueued
# or somehow got de-queued.
#
# I'm not sure of a case where this would happen, but it's cheap insurance.
defp enqueue_video_downloads(channel) do
channel
|> Media.list_pending_media_items_for()
|> Enum.each(fn media_item ->
media_item
|> Map.take([:id])
|> VideoDownloadWorker.new()
|> Tasks.create_job_with_task(media_item)
end)
end
end