Compare commits

..

No commits in common. "4b28396134562e1d7535185a2eb3cc21c68bc038" and "e6161797745cd6c1513e3df560cf981994f3c92f" have entirely different histories.

8 changed files with 33 additions and 390 deletions

14
Cargo.lock generated
View File

@ -1333,12 +1333,8 @@ dependencies = [
"anyhow",
"async-trait",
"futures",
"lazy_static",
"regex",
"thiserror",
"tokio",
"tracing",
"uuid",
]
[[package]]
@ -1874,16 +1870,6 @@ version = "0.7.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
[[package]]
name = "uuid"
version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dd6469f4314d5f1ffec476e05f17cc9a78bc7a27a6a857842170bdf8d6f98d2f"
dependencies = [
"getrandom",
"rand",
]
[[package]]
name = "valuable"
version = "0.1.0"

View File

@ -21,12 +21,9 @@ impl MutationRoot {
let download = app
.download_service
.clone()
.add_download(Download {
id: None,
id: Some("some-id".to_string()),
link: download_link,
progress: None,
file_name: None,
})
.await
.unwrap();

View File

@ -3,27 +3,29 @@ use std::sync::Arc;
use async_graphql::{Context, Object, Result, SimpleObject, ID};
use scel_core::App;
#[derive(SimpleObject, Clone)]
pub struct Download {
pub id: ID,
pub link: String,
pub progress: Option<u32>,
pub file_name: Option<String>,
#[derive(SimpleObject)]
struct Download {
id: ID,
link: String,
progress: i32,
}
pub struct QueryRoot;
#[Object]
impl QueryRoot {
async fn hello_world(&self) -> &str {
"Hello, world!"
}
async fn get_download(&self, ctx: &Context<'_>, id: ID) -> Result<Option<Download>> {
let app = ctx.data_unchecked::<Arc<App>>();
match app.download_service.get_download(id.to_string()).await {
Ok(Some(d)) => Ok(Some(Download {
id: ID::from(d.id.expect("ID could not be found")),
progress: None,
progress: 0,
link: d.link,
file_name: None,
})),
Ok(None) => Ok(None),
Err(e) => Err(e.into()),

View File

@ -5,18 +5,16 @@ use async_graphql::{
};
use scel_core::App;
use super::query::Download;
pub struct SubscriptionRoot;
struct DownloadChanged {
download: Download,
id: ID,
}
#[Object]
impl DownloadChanged {
async fn download(&self) -> Download {
self.download.clone()
async fn id(&self) -> &ID {
&self.id
}
}
@ -32,16 +30,8 @@ impl SubscriptionRoot {
stream! {
while stream.changed().await.is_ok() {
let next_download = (*stream.borrow()).clone();
let id = ID::from(next_download.id.unwrap());
yield DownloadChanged {
download: Download {
id: id,
link: next_download.link,
file_name: next_download.file_name,
progress: next_download.progress,
}
id: id.clone()
}
}
}

View File

@ -11,7 +11,3 @@ anyhow = { version = "*" }
async-trait = { version = "0.1.56" }
futures = "0.3.21"
tracing = "0.1"
lazy_static = "1.4.0"
regex = { version = "1.5.5" }
thiserror = "1.0.31"
uuid = {version = "1.1.2", features = ["v4", "fast-rng"]}

View File

@ -1,19 +1,16 @@
use std::sync::Arc;
use services::InMemoryDownloadService;
pub mod services;
mod youtube;
#[allow(dead_code)]
pub struct App {
pub download_service: Arc<InMemoryDownloadService>,
pub download_service: InMemoryDownloadService,
}
impl App {
pub fn new() -> Self {
Self {
download_service: Arc::new(InMemoryDownloadService::new()),
download_service: InMemoryDownloadService::new(),
}
}
}

View File

@ -1,29 +1,16 @@
use std::{collections::HashMap, path::PathBuf, sync::Arc};
use std::{collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::{watch, Mutex};
use tracing::error;
use uuid::Uuid;
use crate::youtube::{Arg, YoutubeDL};
use tracing::info;
#[derive(Clone)]
pub struct Download {
pub id: Option<String>,
pub link: String,
pub progress: Option<u32>,
pub file_name: Option<String>,
}
pub struct InMemoryDownloadService {
downloads: Mutex<
HashMap<
String,
(
Arc<Mutex<Download>>,
Arc<Mutex<tokio::sync::watch::Sender<Download>>>,
tokio::sync::watch::Receiver<Download>,
),
>,
>,
downloads:
Mutex<HashMap<String, (Arc<Mutex<Download>>, tokio::sync::watch::Receiver<Download>)>>,
}
impl InMemoryDownloadService {
@ -33,86 +20,28 @@ impl InMemoryDownloadService {
}
}
pub async fn add_download(self: Arc<Self>, download: Download) -> anyhow::Result<Download> {
pub async fn add_download(&self, download: Download) -> anyhow::Result<Download> {
let mut downloads = self.downloads.lock().await;
let (tx, rx) = watch::channel(download.clone());
let shared_tx = Arc::new(Mutex::new(tx));
let mut d = download.to_owned();
let id = Uuid::new_v4().to_string();
d.id = Some(id.clone());
downloads.insert(
id.clone(),
(
Arc::new(Mutex::new(d.clone())),
shared_tx.clone(),
rx.clone(),
),
"key".to_string(),
(Arc::new(Mutex::new(download.clone())), rx.clone()),
);
let args = vec![
Arg::new("--progress"),
Arg::new("--newline"),
Arg::new_with_args("--output", "%(title).90s.%(ext)s"),
];
let ytd = YoutubeDL::new(
&PathBuf::from("./data/downloads"),
args,
download.link.as_str(),
)?;
tokio::spawn({
let download_service = self.clone();
let d = download.clone();
async move {
if let Err(e) = ytd
.download(
|percentage| {
let ds = download_service.clone();
let id = id.clone();
async move {
let mut download = ds.get_download(id).await.unwrap().unwrap();
download.progress = Some(percentage);
let _ = ds.update_download(download).await;
}
},
|file_name| {
let ds = download_service.clone();
let id = id.clone();
async move {
let mut download = ds.get_download(id).await.unwrap().unwrap();
download.file_name = Some(file_name);
let _ = ds.update_download(download).await;
}
},
)
.await
{
error!("Download failed: {}", e);
} else {
let download = download_service.get_download(id).await.unwrap().unwrap();
let _ = download_service.update_download(download).await;
loop {
info!("Sending event: {}", d.clone().id.unwrap());
let _ = tx.send(d.clone());
tokio::time::sleep(Duration::from_millis(300)).await;
}
}
});
Ok(d)
}
pub async fn update_download(self: Arc<Self>, download: Download) -> anyhow::Result<()> {
let mut downloads = self.downloads.lock().await;
if let Some(d) = downloads.get_mut(&download.clone().id.unwrap()) {
let mut d_mut = d.0.lock().await;
*d_mut = download.clone();
let _ = d.1.lock().await.send(download);
}
Ok(())
Ok(download)
}
pub async fn get_download(&self, id: String) -> anyhow::Result<Option<Download>> {
@ -129,7 +58,9 @@ impl InMemoryDownloadService {
pub async fn subscribe_download(&self, id: String) -> tokio::sync::watch::Receiver<Download> {
let downloads = self.downloads.lock().await;
let download = downloads.get(&id).unwrap();
download.2.clone()
download.1.clone()
}
}

View File

@ -1,256 +0,0 @@
use std::fmt::{Display, Formatter};
use std::fs::{canonicalize, create_dir_all};
use std::future::Future;
use std::num::ParseIntError;
use std::path::{Path, PathBuf};
use std::process::{Output, Stdio};
use lazy_static::lazy_static;
use regex::Regex;
use thiserror::Error;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
#[derive(Error, Debug)]
pub enum YoutubeDLError {
#[error("failed to execute youtube-dl")]
IOError(#[from] std::io::Error),
#[error("failed to convert path")]
UTF8Error(#[from] std::string::FromUtf8Error),
#[error("youtube-dl exited with: {0}")]
Failure(String),
}
type Result<T> = std::result::Result<T, YoutubeDLError>;
const YOUTUBE_DL_COMMAND: &str = "yt-dlp";
#[derive(Clone, Debug)]
pub struct Arg {
arg: String,
input: Option<String>,
}
impl Arg {
pub fn new(argument: &str) -> Self {
Self {
arg: argument.to_string(),
input: None,
}
}
pub fn new_with_args(argument: &str, input: &str) -> Self {
Self {
arg: argument.to_string(),
input: Option::from(input.to_string()),
}
}
}
impl Display for Arg {
fn fmt(&self, fmt: &mut Formatter<'_>) -> std::fmt::Result {
match &self.input {
Some(input) => write!(fmt, "{} {}", self.arg, input),
None => write!(fmt, "{}", self.arg),
}
}
}
#[derive(Clone, Debug)]
pub struct YoutubeDL {
path: PathBuf,
links: Vec<String>,
args: Vec<Arg>,
}
#[derive(Clone, Debug)]
pub struct YoutubeDLResult {
path: PathBuf,
output: String,
}
impl YoutubeDLResult {
fn new(path: &PathBuf) -> Self {
Self {
path: path.clone(),
output: String::new(),
}
}
pub fn output_dir(&self) -> &PathBuf {
&self.path
}
}
impl YoutubeDL {
pub fn new_multiple_links(
dl_path: &PathBuf,
args: Vec<Arg>,
links: Vec<String>,
) -> Result<YoutubeDL> {
let path = Path::new(dl_path);
if !path.exists() {
create_dir_all(&path)?;
}
if !path.is_dir() {
return Err(YoutubeDLError::IOError(std::io::Error::new(
std::io::ErrorKind::Other,
"path is not a directory",
)));
}
let path = canonicalize(dl_path)?;
Ok(YoutubeDL { path, links, args })
}
pub fn new(dl_path: &PathBuf, args: Vec<Arg>, link: &str) -> Result<YoutubeDL> {
YoutubeDL::new_multiple_links(dl_path, args, vec![link.to_string()])
}
pub async fn download<F, FutAvailable, FAvailable, Fut>(
&self,
progress_update_fn: F,
file_name_available: FAvailable,
) -> Result<YoutubeDLResult>
where
F: Fn(u32) -> Fut,
FAvailable: Fn(String) -> FutAvailable,
Fut: Future<Output = ()>,
FutAvailable: Future<Output = ()>,
{
let output = self
.spawn_youtube_dl(progress_update_fn, file_name_available)
.await?;
let mut result = YoutubeDLResult::new(&self.path);
if !output.status.success() {
return Err(YoutubeDLError::Failure(String::from_utf8(output.stderr)?));
}
result.output = String::from_utf8(output.stdout)?;
Ok(result)
}
async fn spawn_youtube_dl<F, FutAvailable, FAvailable, Fut>(
&self,
progress_update_fn: F,
file_name_available: FAvailable,
) -> Result<Output>
where
F: Fn(u32) -> Fut,
FAvailable: Fn(String) -> FutAvailable,
Fut: Future<Output = ()>,
FutAvailable: Future<Output = ()>,
{
let mut cmd = Command::new(YOUTUBE_DL_COMMAND);
cmd.current_dir(&self.path)
.env("LC_ALL", "en_US.UTF-8")
.stdout(Stdio::piped())
.stderr(Stdio::piped());
for arg in self.args.iter() {
match &arg.input {
Some(input) => cmd.arg(&arg.arg).arg(input),
None => cmd.arg(&arg.arg),
};
}
for link in self.links.iter() {
cmd.arg(&link);
}
let mut pr = cmd.spawn()?;
{
let stdout = pr.stdout.as_mut().unwrap();
let stdout_reader = BufReader::new(stdout);
let mut stdout_lines = stdout_reader.lines();
let mut have_gotten_file_name = false;
while let Ok(Some(line)) = stdout_lines.next_line().await {
println!("{}", line.clone());
if !have_gotten_file_name {
if let Some(file_name) = parse_file_name(line.clone()) {
file_name_available(file_name).await;
have_gotten_file_name = true
}
}
if let Some(Ok(percentage)) = parse_line(line) {
progress_update_fn(percentage).await;
}
}
}
Ok(pr.wait_with_output().await?)
}
}
fn parse_line(line: String) -> Option<core::result::Result<u32, ParseIntError>> {
lazy_static! {
static ref RE: Regex = Regex::new(r"\[download\]\s+(\d+)").unwrap();
}
let capture: regex::Captures = RE.captures(line.as_str())?;
if capture.len() != 2 {
return None;
}
let str = &capture[1];
Some(str.to_string().parse::<u32>())
}
fn parse_file_name(line: String) -> Option<String> {
lazy_static! {
static ref RE: Regex = Regex::new(r"^\[download\] Destination: (.+)$").unwrap();
}
let capture: regex::Captures = RE.captures(line.as_str())?;
if capture.len() != 2 {
return None;
}
let str = &capture[1];
Some(str.to_string())
}
#[cfg(test)]
mod tests {
use crate::youtube::{parse_file_name, parse_line};
#[test]
fn test_parse_line() {
let percentage = parse_line(
"[download] 95.4% of ~215.85MiB at 9.61MiB/s ETA 00:01 (frag 144/151)".into(),
);
assert_eq!(percentage, Some(Ok(95)))
}
#[test]
fn test_parse_line_get_nothing() {
let nothing = parse_line("[download] Got server HTTP error: The read operation timed out. Retrying (attempt 1 of 10) ...".into());
assert_eq!(nothing, None)
}
#[test]
fn test_parse_file_name() {
let file_name = parse_file_name(
"[download] Destination: 10 Design Patterns Explained in 10 Minutes.mp4".into(),
);
assert_eq!(
file_name,
Some("10 Design Patterns Explained in 10 Minutes.mp4".into())
);
}
#[test]
fn test_parse_file_name_get_nothing() {
let nothing = parse_file_name("[download] No fit: something".into());
assert_eq!(nothing, None)
}
}