YoutubeDL command

PublishHelperBot/Environment.fs

@@ -9,6 +9,7 @@ type public BotConfig = {
   relayUrl: string
   chanelId: int64
   adminChatId: int64
+  YoutubeDlUrl: string
 }
 
 let private ReadConfig =

PublishHelperBot/Handlers.fs

@@ -1,7 +1,9 @@
 module PublishHelperBot.Handlers
 
 open System.Threading.Tasks
+open Microsoft.FSharp.Core
 open PublishHelperBot.Environment
+open PublishHelperBot.YoutubeDl
 open Telegram.Bot
 open Telegram.Bot.Types
 open Telegram.Bot.Types.Enums
@@ -10,6 +12,7 @@ type BaseHandlerArgs = Update * BotConfig
 type HandlerArgs = Update * BotConfig * ITelegramBotClient
 type HandlerRequirements = BaseHandlerArgs -> bool
 type Handler = HandlerArgs -> Task
+type Handler<'deps> = 'deps * HandlerArgs -> Task
 
 // Utils
 let UpdateIsAMessage (x: Update) = x.Type = UpdateType.Message
@@ -67,11 +70,20 @@ let public RelayHandler: Handler = fun (u, c, tg) ->
   | _ -> Task.CompletedTask
 
 
-// Youtube repost
+// YoutubeDL repost
 
+let YoutubeRepostMatchCmd = "\\ytdl"
 let public YoutubeRepostMatch: HandlerRequirements = fun (u, c) ->
   UpdateIsAMessage u &&
   FromAdminChat <| (u.Message, c) &&
-  u.Message.Text.StartsWith("\\yt") &&
+  u.Message.Text.StartsWith YoutubeRepostMatchCmd &&
   u.Message.Text.Split(' ').Length = 2
+
+let public YoutubeRepostHandler: Handler<YoutubeDlBackgroundService> = fun (yt, (u, c, tg)) ->
+  task {
+      let trim (x: string) = x.Trim()
+      let! id = YoutubeRepostMatchCmd |> u.Message.Text.Split |> Array.last |> trim |> yt.EnqueueJob
+      do! tg.SendTextMessageAsync(c.adminChatId, id.ToString()) |> Async.AwaitTask |> Async.Ignore
+  }
+
   

PublishHelperBot/Program.fs

@@ -6,6 +6,7 @@ open System.Threading
 open System.Threading.Tasks
 open PublishHelperBot.Handlers
 open PublishHelperBot.Environment
+open PublishHelperBot.YoutubeDl
 open Telegram.Bot
 open Telegram.Bot.Polling
 open Telegram.Bot.Types
@@ -14,13 +15,18 @@ open Telegram.Bot.Types.Enums
 let CreateBot (config: BotConfig, http: HttpClient) = TelegramBotClient(config.token, http)
 let config = CreateConfig <| "SBPB_CONFIG_PATH";
 let botClient = CreateBot <| (config, new HttpClient())
+let YtService = YoutubeDlBackgroundService <|
+                (new HttpClient(), config.YoutubeDlUrl, botClient, config.chanelId, CancellationToken.None)
+
 let startDate = DateTime.UtcNow
 let isObsoleteUpdate (u: Update) = u.Type = UpdateType.Message && u.Message.Date < startDate;
 
 let updateHandle (bc: ITelegramBotClient) (u: Update) (ct: CancellationToken): Task =
+  let tgCtx = (u, config, bc)
   match u with
   | _ when isObsoleteUpdate u -> Task.CompletedTask
-  | _ when RelayMatch <| (u, config) -> RelayHandler <| (u, config, bc)
+  | _ when RelayMatch <| (u, config) -> RelayHandler <| tgCtx
+  | _ when YoutubeRepostMatch <| (u, config) -> YoutubeRepostHandler <| (YtService, tgCtx)
   | _ -> Task.CompletedTask
 
 let handlePollingError (bc: ITelegramBotClient) (e: Exception) (t: CancellationToken) =
@@ -30,6 +36,7 @@ let handlePollingError (bc: ITelegramBotClient) (e: Exception) (t: CancellationT
 let receiverOptions = ReceiverOptions(AllowedUpdates = Array.zeroCreate<UpdateType> 0) 
 
 botClient.StartReceiving(updateHandle,handlePollingError,receiverOptions)
+YtService.StartYoutubeDlService()
 
 printf "Я родился"
 Console.ReadKey() |> ignore

PublishHelperBot/PublishHelperBot.fsproj

@@ -7,10 +7,12 @@
 
     <ItemGroup>
         <PackageReference Include="Telegram.Bot" Version="18.0.0" />
+        <PackageReference Include="Nito.AsyncEx" Version="5.1.2" />
     </ItemGroup>
     
     <ItemGroup>
         <Compile Include="Environment.fs" />
+        <Compile Include="YoutubeDl.fs" />
         <Compile Include="Handlers.fs" />
         <Compile Include="Program.fs" />
         <Content Include="config.example.json" />

PublishHelperBot/YoutubeDl.fs

@@ -0,0 +1,176 @@
+module PublishHelperBot.YoutubeDl
+
+open System
+open System.Collections.Generic
+open System.IO
+open System.Net.Http
+open System.Text
+open System.Threading
+open Microsoft.FSharp.Core
+open Newtonsoft.Json
+open Nito.AsyncEx
+open Telegram.Bot
+open Telegram.Bot.Types.InputFiles
+
+type public CreateYoutubeDLUrl = string
+type public ChatId = int64
+type public CreateYoutubeDLJob = {
+      url: string
+      savePath: string
+}
+type public CreateYoutubeDLJobSuccess = {
+  task: Guid
+}
+
+type public YoutubeDlStateResponse = {
+  state: string
+}
+
+type public YoutubeDlError = {
+  message: string
+}
+type public YoutubeDlJob<'A> = {
+  internalId: Guid
+  externalId: 'A
+  url: string
+  state: string
+  savePath: string
+}
+
+type YoutubeDlClientActions = Create | Check | Delete
+type HttpMethods = GET | POST | DELETE
+type CreateJobResult = Result<CreateYoutubeDLJobSuccess, YoutubeDlError> Async
+type CheckJobResult = Result<YoutubeDlStateResponse, YoutubeDlError> Async
+type CleanJobResult = Result<YoutubeDlStateResponse, YoutubeDlError> Async
+type StartYoutubeDlServiceArgs = HttpClient * CreateYoutubeDLUrl * ITelegramBotClient * ChatId * CancellationToken
+type YoutubeDlJobWithId = YoutubeDlJob<Guid>
+type YoutubeDlJobWithoutId = YoutubeDlJob<unit>
+type YoutubeDlCurrentJob =
+  | Created of YoutubeDlJobWithoutId
+  | Awaiting of YoutubeDlJobWithId
+  | Downloaded of YoutubeDlJobWithId
+  | Done of YoutubeDlJobWithId
+  | None of unit
+  
+let inline (<!>) (lck: AsyncLock) f = async {
+  use! __ = Async.AwaitTask <| lck.LockAsync().AsTask()
+  return! f
+}
+
+type YoutubeDlClient(baseUrl: string, client: HttpClient) =
+  let lock = AsyncLock()
+  let apiPrefix = $"{baseUrl}api/";
+  let ResolvePath(action: YoutubeDlClientActions) =
+    match action with
+    | Create -> $"{apiPrefix}download"
+    | Check -> $"{apiPrefix}status"
+    | Delete -> $"{apiPrefix}clear"
+  
+  let doHttp (url: string, method: HttpMethods, content: HttpContent): Result<'TRes, YoutubeDlError> Async = async {
+    try
+      let! res =
+        match method with
+        | POST -> client.PostAsync(url, content) |> Async.AwaitTask
+        | GET -> client.GetAsync(url) |> Async.AwaitTask
+        | DELETE -> client.DeleteAsync(url) |> Async.AwaitTask
+        
+      let! content = res.Content.ReadAsStringAsync() |> Async.AwaitTask
+      return
+        match res.IsSuccessStatusCode with
+        | true -> Ok (JsonConvert.DeserializeObject<'TRes> <| content)
+        | false -> Error { message = "Unknown network error" }
+    with
+      | ex -> return Error { message = ex.Message  }
+  }
+  
+  member this.CreateJob(model: CreateYoutubeDLJob): CreateJobResult = lock <!> async {
+      use content = new StringContent(JsonConvert.SerializeObject <| model, Encoding.UTF8, "application/json")
+      return! doHttp <| (ResolvePath Create, POST, content)
+  }
+  
+  member this.CheckJob(id: Guid): CheckJobResult = lock <!> async {
+    let arg = [KeyValuePair("id", id.ToString())]
+    use content = new FormUrlEncodedContent(arg)
+    let! query = content.ReadAsStringAsync() |> Async.AwaitTask
+    return! doHttp <| ($"{ResolvePath Check}?{query}", GET, content)
+  }
+  
+  member this.CleanJob(id: Guid): CleanJobResult = lock <!> async {
+    let arg = [KeyValuePair("id", id.ToString())]
+    use content = new FormUrlEncodedContent(arg)
+    let! query = content.ReadAsStringAsync() |> Async.AwaitTask
+    return! doHttp <| ($"{ResolvePath Delete}?{query}", DELETE, content)
+  }
+
+type YoutubeDlBackgroundService(requirements: StartYoutubeDlServiceArgs) =
+  let (http, url, tg, chatId, ct) = requirements
+  let lock = AsyncLock()
+  let mutable currentJob: YoutubeDlCurrentJob = None ()
+  let jobPool = Queue<YoutubeDlJobWithoutId>()
+  let ytClient = YoutubeDlClient <| (url, http)
+  let mapJobToApi (job: YoutubeDlJob<_>): CreateYoutubeDLJob = {
+    url = job.url
+    savePath = job.savePath
+  }
+  let attachExternalId (id: Guid, job: YoutubeDlJobWithoutId): YoutubeDlCurrentJob =
+    Awaiting { internalId = job.internalId; state = job.state; url = job.state; externalId = id; savePath = job.savePath }
+  
+  let tryAssignNewJob() = async {
+    let (result, job) = jobPool.TryDequeue()
+    match result with
+    | true -> currentJob <- Created job
+    | false -> currentJob <- None ()
+  }
+    
+  let uploadToYtDl(job: YoutubeDlJobWithoutId) = async {
+        match! ytClient.CreateJob <| mapJobToApi job with
+        | Ok x -> currentJob <- attachExternalId <| (x.task, job)
+        // TODO: Logging!
+        | Error _ -> currentJob <- None ()
+  }
+
+  let checkJob(job: YoutubeDlJobWithId) = async {
+    match! ytClient.CheckJob <| job.externalId with
+    | Ok x when x.state.Equals("Finished", StringComparison.OrdinalIgnoreCase) -> currentJob <- Downloaded job
+    | Error _ -> currentJob <- None ()
+    | _ -> ()
+    // That's take a while
+    do! Async.Sleep 5000
+  }
+  
+  let postVideo(job: YoutubeDlJobWithId) = async {
+    use file = File.OpenRead <| job.savePath
+    let input = InputOnlineFile(file, job.savePath)
+    let caption = $"Source: {job.url}"
+    do! tg.SendVideoAsync(chatId, input, caption = caption) |> Async.AwaitTask |> Async.Ignore
+    currentJob <- Done job
+  }
+  
+  let cleanUp(job: YoutubeDlJobWithId) = async {
+    match! ytClient.CleanJob <| job.externalId with
+    | Ok _ -> currentJob <- None ()
+    | Error _ -> currentJob <- None ()
+  }
+  
+  let chooseAction() = lock <!> async {
+      do! match currentJob with
+          | None _ -> tryAssignNewJob()
+          | Created x -> x |> uploadToYtDl
+          | Awaiting x -> x |> checkJob
+          | Downloaded x -> x |> postVideo
+          | Done x -> x |> cleanUp
+  }
+
+  let rec loop () = async {
+    do! match ct.IsCancellationRequested with
+        | false -> chooseAction()
+        | true -> async { () }
+    do! Async.Sleep 150
+    return! loop()
+  }
+  member public this.StartYoutubeDlService() = loop() |> Async.Start
+  member public this.EnqueueJob(url: string) = lock <!> async {
+    let id = Guid.NewGuid()
+    jobPool.Enqueue({ internalId = id; externalId = (); state = "new"; url = url; savePath = Path.GetTempFileName() })
+    return id
+  }