1 files changed, 110 insertions(+), 6 deletions(-)

M src/bin/aqua_watch.rs

@@ 22,6 22,7 @@
 //
 
 #[macro_use] extern crate log;
+#[macro_use] extern crate serde_derive;
 
 extern crate aqua;
 extern crate clap;


@@ 30,6 31,8 @@ extern crate dotenv;
 extern crate env_logger;
 extern crate image;
 extern crate notify;
+extern crate serde;
+extern crate serde_json;
 
 use aqua::controllers::prelude::hash_file;
 use aqua::models::{Entry, NewEntry};


@@ 39,7 42,8 @@ use diesel::prelude::*;
 use diesel::pg::PgConnection;
 use dotenv::dotenv;
 use notify::{DebouncedEvent, Watcher, RecursiveMode, watcher};
-use std::{env, fmt};
+use std::{env, fmt, process};
+use std::collections::HashMap;
 use std::convert::From;
 use std::error::Error;
 use std::fs::{self, File, OpenOptions};


@@ 53,10 57,12 @@ enum ProcessingError {
     DigestFailed,
     DetectionFailed,
     ThumbnailFailed,
+
     DbConnErr(diesel::ConnectionError),
     DbQueryErr(diesel::result::Error),
     IoErr(io::Error),
     ImageErr(image::ImageError),
+    Misc(Box<Error>),
 }
 
 impl Error for ProcessingError {


@@ 72,6 78,7 @@ impl Error for ProcessingError {
             ProcessingError::DbQueryErr(ref inner) => inner.description(),
             ProcessingError::IoErr(ref inner)      => inner.description(),
             ProcessingError::ImageErr(ref inner)   => inner.description(),
+            ProcessingError::Misc(ref inner)       => inner.description(),
         }
     }
 


@@ 81,6 88,7 @@ impl Error for ProcessingError {
             ProcessingError::DbQueryErr(ref err) => Some(err),
             ProcessingError::IoErr(ref err)      => Some(err),
             ProcessingError::ImageErr(ref err)   => Some(err),
+            ProcessingError::Misc(ref err)       => Some(err.as_ref()),
             _ => None,
         }
     }


@@ 111,6 119,14 @@ impl From<io::Error> for ProcessingError {
     fn from(err: io::Error) -> Self { ProcessingError::IoErr(err) }
 }
 
+impl From<serde_json::Error> for ProcessingError {
+    fn from(err: serde_json::Error) -> Self { ProcessingError::Misc(Box::new(err)) }
+}
+
+impl From<std::string::FromUtf8Error> for ProcessingError {
+    fn from(err: std::string::FromUtf8Error) -> Self { ProcessingError::Misc(Box::new(err)) }
+}
+
 fn main() {
     dotenv().expect("must provide .env file, see README (TODO: haha jk)");
     env_logger::init().expect("could not initialize console logging");


@@ 149,7 165,7 @@ fn main() {
                 if path.is_file() { 
                     match handle_new_file(path, content_store) {
                         Ok(_res) => info!("file processed successfully ..."),
-                        Err(msg) => warn!("could not process file: {}", msg.description()),
+                        Err(msg) => warn!("could not process file: {:?} (inner: {:?}", msg, msg.cause()),
                     };
                 }
                 else { info!("directory created, ignoring ..."); }


@@ 183,15 199,72 @@ fn handle_new_file(path: PathBuf, content_store: &str) -> Result<(), ProcessingE
         info!("inserted: {:?} into database", db_entry);
 
         Ok(())
-    } else if let Some(_nil) = ffmpeg_detect(path.as_path()) {
-        unreachable!()
+    } else if let Some(ffmpeg_metadata) = ffmpeg_detect(path.as_path())? {
+        info!("got an video ...");
+        process_video(content_store, &digest, &path)?;
+        move_file(path.as_path(), content_store, &digest, ffmpeg_metadata.ext)?;
+
+        let db_entry = create_db_entry(&digest, ffmpeg_metadata.mime)?;
+        info!("inserted: {:?} into database", db_entry);
+
+        Ok(())
     } else {
         Err(ProcessingError::DetectionFailed)
     }
 }
 
-fn ffmpeg_detect(path: &Path) -> Option<()> {
-    None
+#[derive(Debug, Serialize, Deserialize)]
+struct FFProbeResult {
+    format: Option<FFProbeFormat>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct FFProbeFormat {
+    filename:    String,
+    nb_streams:  i32,
+    format_name: String,
+    start_time:  String, // NOTE: these appear to be fixed decimal
+    duration:    String, // NOTE: these appear to be fixed decimal
+    size:        String, // NOTE: appears to be an integer
+    bit_rate:    String, // NOTE: appears to be an integer
+    probe_score: i32,    // NOTE: accuracy of the detection? (???)
+
+    tags: HashMap<String, String>, // NOTE: not sure this is correct type
+}
+
+struct FFProbeMeta {
+    pub mime: &'static str,
+    pub ext:  &'static str,
+}
+
+fn ffmpeg_detect(path: &Path) -> Result<Option<FFProbeMeta>, ProcessingError> {
+    let ffprobe_cmd = process::Command::new("ffprobe")
+        .arg("-v").arg("quiet")            // silence debug output
+        .arg("-hide_banner")               // don't print ffmpeg configuration
+        .arg("-print_format").arg("json") // serialize to json
+        .arg("-show_format")              // use the serializer
+        .arg("-i").arg(path.as_os_str())  // set the input to current file
+        .output()?;
+
+    let json_str = String::from_utf8(ffprobe_cmd.stdout)?;
+    let probe_result: FFProbeResult = serde_json::from_str(&json_str)?;
+    info!("got result: {:?}", probe_result);
+
+    // see if ffprobe was able to determine the file type ...
+    let probe_result = match probe_result.format {
+        Some(format_data) => format_data,
+        None => return Ok(None),
+    };
+
+    // TODO: handle alternate streams
+    if probe_result.nb_streams != 1 { return Err(ProcessingError::DetectionFailed) }
+
+    let meta_data = match probe_result.format_name.as_ref() {
+        "matroska,webm" => Some(FFProbeMeta { mime: "video/webm", ext: "webm" }),
+        _ => None,
+    };
+
+    Ok(meta_data) // return the ffprobe results, if we know how to handle them ...
 }
 
 fn establish_connection() -> Result<PgConnection, ProcessingError> {


@@ 253,3 326,34 @@ fn process_image(content_store: &str, digest: &str, buf: &[u8]) -> Result<(), Pr
     thumb.save(&mut dest_file, image::ImageFormat::JPEG)?;
     Ok(dest_file.flush()?)
 }
+
+fn process_video(content_store: &str, digest: &str, src: &Path) -> Result<(), ProcessingError> {
+    let thumb_bucket   = format!("t{}", &digest[0..2]);
+    let thumb_filename = format!("{}.thumbnail", &digest);
+
+    // store them in content store
+    let dest = PathBuf::from(content_store)
+        .join(thumb_bucket)
+        .join(thumb_filename);
+
+    // write thumbnail file to disk
+    fs::create_dir_all(dest.parent().unwrap())?;
+    let ffmpeg_cmd = process::Command::new("ffmpeg")
+        .arg("-i").arg(src.as_os_str())            // the input file
+        .arg("-vf").arg("thumbnail,scale=200:200") // have ffmpeg seek for a "thumbnail"
+        .arg("-frames:v").arg("1")                 // take a single frame
+        .arg("-f").arg("mjpeg")                    // save it as jpeg
+        .arg(dest.as_path().as_os_str())           // into the content store
+        .output()?;
+
+    debug!("ffmpeg stderr: {}", String::from_utf8_lossy(&ffmpeg_cmd.stderr));
+    debug!("ffmpeg stdout: {}", String::from_utf8_lossy(&ffmpeg_cmd.stdout));
+
+    info!("digest? {}", digest);
+    info!("dest exists? {:?} => {}", dest, dest.is_file());
+
+    match dest.is_file() {
+        true  => Ok(()),
+        false => Err(ProcessingError::ThumbnailFailed),
+    }
+}