1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
use std::{
fs::{self, canonicalize},
io::{stderr, stdout, Read},
mem,
os::unix::fs::symlink,
path::PathBuf,
process::Command,
sync::mpsc::{self, Receiver, Sender},
thread::{self, JoinHandle},
};
use anyhow::{bail, Context, Result};
use log::{debug, warn};
use url::Url;
use crate::constants::{status_path, CONCURRENT, DOWNLOAD_DIR, MPV_FLAGS, YT_DLP_FLAGS};
#[derive(Debug)]
pub struct Downloadable {
pub url: Url,
pub id: Option<u32>,
}
pub struct Downloader {
sent: usize,
download_thread: JoinHandle<Result<()>>,
orx: Receiver<(PathBuf, Option<u32>)>,
itx: Option<Sender<Downloadable>>,
playspec: Vec<Downloadable>,
}
impl Downloader {
pub fn new(mut playspec: Vec<Downloadable>) -> anyhow::Result<Downloader> {
let (itx, irx): (Sender<Downloadable>, Receiver<Downloadable>) = mpsc::channel();
let (otx, orx) = mpsc::channel();
let jh = thread::spawn(move || -> Result<()> {
while let Some(pt) = irx.recv().ok() {
debug!("Got '{}|{}' to be downloaded", pt.url, pt.id.unwrap_or(0));
let path = download_url(&pt.url)
.with_context(|| format!("Failed to download url: '{}'", &pt.url))?;
otx.send((path, pt.id)).expect("Should not be dropped");
}
debug!("Finished Downloading everything");
Ok(())
});
playspec.reverse();
let mut output = Downloader {
sent: 0,
download_thread: jh,
orx,
itx: Some(itx),
playspec,
};
if output.playspec.len() <= CONCURRENT as usize {
output.add(output.playspec.len() as u32)?;
} else {
output.add(CONCURRENT)?;
}
Ok(output)
}
pub fn add(&mut self, number_to_add: u32) -> Result<()> {
debug!("Adding {} to be downloaded concurrently", number_to_add);
for _ in 0..number_to_add {
let pt = self.playspec.pop().context("No more playthings to pop")?;
self.itx.as_ref().expect("Should still be valid").send(pt)?;
}
Ok(())
}
/// Return the next video already downloaded, will block until the download is complete
pub fn next(&mut self) -> Option<(PathBuf, Option<u32>)> {
debug!("Requesting next output");
match self.orx.recv() {
Ok(ok) => {
debug!("Output downloaded to: {}", ok.0.display());
self.sent += 1;
if self.sent < self.playspec.len() {
debug!("Will add 1");
self.add(1).ok()?;
} else {
debug!(
"Done sending videos to be downloaded, downoladed: {} videos",
self.sent
);
let itx = mem::take(&mut self.itx);
drop(itx)
}
debug!("Returning: {}|{}", ok.0.display(), ok.1.unwrap_or(0));
Some(ok)
}
Err(err) => {
debug!("Received error while listening: {}", err);
None
}
}
}
pub fn drop(self) -> anyhow::Result<()> {
match self.download_thread.join() {
Ok(ok) => ok,
Err(err) => panic!("Can't join thread: '{:#?}'", err),
}
}
pub fn consume(mut self) -> anyhow::Result<()> {
while let Some((path, id)) = self.next() {
debug!("Next path to play is: '{}'", path.display());
let mut info_json = canonicalize(&path).context("Failed to canoncialize path")?;
info_json.set_extension("info.json");
if status_path()?.is_symlink() {
fs::remove_file(status_path()?).context("Failed to delete old status file")?;
} else if !status_path()?.exists() {
debug!(
"The status path at '{}' does not exists",
status_path()?.display()
);
} else {
bail!(
"The status path ('{}') is not a symlink but exists!",
status_path()?.display()
);
}
symlink(info_json, status_path()?).context("Failed to symlink")?;
let mut mpv = Command::new("mpv");
mpv.stdout(stdout());
mpv.stderr(stderr());
mpv.args(MPV_FLAGS);
mpv.arg(&path);
let status = mpv.status().context("Failed to run mpv")?;
if status.success() {
fs::remove_file(&path)?;
if let Some(id) = id {
println!("\x1b[32;1mMarking {} as watched!\x1b[0m", id);
let mut ytcc = std::process::Command::new("ytcc");
ytcc.stdout(stdout());
ytcc.stderr(stderr());
ytcc.args(["mark"]);
ytcc.arg(id.to_string());
let status = ytcc.status().context("Failed to run ytcc")?;
if let Some(code) = status.code() {
if code != 0 {
bail!("Ytcc failed with status: {}", code);
}
}
}
debug!("mpv exited with: '{}'", status);
} else {
warn!("mpv exited with: '{}'", status);
}
}
self.drop()?;
Ok(())
}
}
fn download_url(url: &Url) -> Result<PathBuf> {
let output_file = tempfile::NamedTempFile::new().context("Failed to create tempfile")?;
output_file
.as_file()
.set_len(0)
.context("Failed to truncate temp-file")?;
if !Into::<PathBuf>::into(DOWNLOAD_DIR).exists() {
fs::create_dir_all(DOWNLOAD_DIR)
.with_context(|| format!("Failed to create download dir at: {}", DOWNLOAD_DIR))?
}
let mut yt_dlp = Command::new("yt-dlp");
yt_dlp.current_dir(DOWNLOAD_DIR);
yt_dlp.stdout(stdout());
yt_dlp.stderr(stderr());
yt_dlp.args(YT_DLP_FLAGS);
yt_dlp.args([
"--output",
"%(channel)s/%(title)s.%(ext)s",
url.as_str(),
"--print-to-file",
"after_move:filepath",
]);
yt_dlp.arg(output_file.path().as_os_str());
let status = yt_dlp.status().context("Failed to run yt-dlp")?;
if !status.success() {
bail!("yt_dlp execution failed with error: '{}'", status);
}
let mut path = String::new();
output_file
.as_file()
.read_to_string(&mut path)
.context("Failed to read output file temp file")?;
let path = path.trim();
Ok(path.into())
}
|
