Init

Modules/Server/src/worker.cpp

@@ -0,0 +1,145 @@
+//
+// Created by s-har on 26.05.2021.
+//
+#include "worker.hpp"
+
+#include <chrono>
+#include <functional>
+#include <vector>
+
+#include <boost/process.hpp>
+#include <ctre.hpp>
+#include <fmt/ostream.h>
+#include <spdlog/spdlog.h>
+
+using namespace std::chrono_literals;
+
+void operator+=(std::vector<std::string>& vec, std::string const& s)
+{
+    vec.push_back(s);
+}
+
+namespace bp = boost::process;
+
+Worker::Worker(soci::session& sql) : thread(std::bind_front(&Worker::loop, this)), sql(sql)
+{
+    spdlog::info("Worker started.");
+}
+
+[[noreturn]] void Worker::loop()
+{
+    while (true)
+    {
+        File task;
+        if (!getTask(task))
+        {
+            std::this_thread::sleep_for(10s);
+            continue;
+        }
+
+        spdlog::info("Worker Task: {}", task);
+
+        spdlog::debug("Begin Download {}", task.id);
+        task.status = FileStatus::DOWNLOADING;
+
+        sql << "UPDATE Files SET status = :status WHERE id = :id", soci::use(task);
+
+        if (download(task))
+        {
+            spdlog::debug("Download end {}", task.id);
+            task.status = FileStatus::COMPLETE;
+        } else
+        {
+            // Error
+            task.status = FileStatus::DOWNLOAD_ERROR;
+            spdlog::error("Download Error {}: ", task);
+
+        }
+
+        sql << "UPDATE Files SET status = :status WHERE id = :id", soci::use(task);
+        std::this_thread::sleep_for(10s);
+    }
+}
+
+bool Worker::getTask(File& task)
+{
+    try
+    {
+        soci::rowset<File> data =
+          (sql.prepare << "SELECT * FROM Files WHERE status = 0 ORDER BY id LIMIT 1;");
+
+        for (auto const& el : data)
+        {
+            task = el;
+            return true;
+        }
+    }
+    catch (std::exception const& e)
+    {
+        spdlog::error(e.what());
+        fmt::format("{}\n", e.what());
+    }
+    return false;
+}
+
+bool Worker::download(File& task)
+{
+    bp::ipstream out;
+    bp::ipstream err;
+
+    std::vector<std::string> args;
+
+    args += "-o";
+    args += fmt::format("{}/%(title)s.%(ext)s", output_path);
+
+    args += "--extract-audio";
+
+    args += task.url;
+
+    spdlog::debug("youtube-dl {}", fmt::join(args, " "));
+
+    bp::child c("youtube-dl.exe", bp::args(args), bp::std_out > out, bp::std_err > err);
+
+    std::string destination_file;
+
+    std::string line;
+    while (c.running() || !out.eof() || !err.eof())
+    {
+        if (std::getline(out, line) && !line.empty())
+        {
+            spdlog::info("[youtube-dl] {}", line);
+
+            if (auto [match, info, destination] =
+                  ctre::match<R"(\[(.*)\] Destination: (.*))">(line);
+                match)
+            {
+                destination_file = destination;
+            }
+
+        } else if (std::getline(err, line) && !line.empty())
+        {
+            if (auto [match, type, messages] = ctre::match<R"((.*): (.*))">(line); match)
+            {
+                if (type == "WARNING")
+                    spdlog::warn("[youtube-dl] {}", messages);
+            } else
+            {
+                spdlog::error("[youtube-dl] {}", line);
+            }
+        }
+    }
+
+    c.wait();
+    auto exit_code = c.exit_code();
+
+    spdlog::debug("File: {}", destination_file);
+
+    Download dwn;
+    dwn.id = task.id;
+    dwn.path = destination_file;
+
+    sql << "INSERT INTO Downloads (id, path) values(:id, :path)", soci::use(dwn);
+
+    spdlog::debug("youtube-dl exit code {}", exit_code);
+    return exit_code == 0;
+}