added streaming

App/Streaming/StreamingRoutes.py

@@ -0,0 +1,12 @@
+from fastapi import APIRouter, Request
+from .Utils.Handler import Handler
+from App import bot
+
+streaming_router = APIRouter(tags=["streaming"])
+
+
+@streaming_router.get("/stream/")
+async def create_file(id: int, req: Request):
+    hander = Handler(req=req, client=bot, id=id)
+    hander.route = "inline"
+    return await hander.process_request()

App/Streaming/Utils/Download.py

@@ -0,0 +1,79 @@
+import math, asyncio, subprocess
+from telethon import TelegramClient
+from fastapi.responses import StreamingResponse
+
+import logging
+
+logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
+
+class Download:
+    client: TelegramClient
+    route: str
+    offset: int
+    handler: None
+    file: None
+    limit: int
+    file_size: float
+
+    def __init__(self, handler):
+        self.handler = handler
+        self.file = handler.message.media
+        self.file_size = handler.message.file.size
+        self.limit = handler.sanity.limit
+        self.offset = handler.sanity.offset
+        self.client = handler.client
+        self.mime_type = handler.message.file.mime_type
+
+    async def download(self):
+        part_size = int(512 * 1024) * 2
+        first_part_cut = self.offset % part_size
+        first_part = math.floor(self.offset / part_size)
+        last_part_cut = part_size - (self.limit % part_size)
+        last_part = math.ceil(self.limit / part_size)
+        part_count = math.ceil(self.file_size / part_size)
+        part = first_part
+        try:
+            async for chunk in self.client.iter_download(
+                self.file, offset=first_part * part_size, request_size=part_size
+            ):
+                if part == first_part:
+                    yield bytes(chunk[first_part_cut:])
+                elif part == last_part:
+                    yield bytes(chunk[:last_part_cut])
+                else:
+                    yield bytes(chunk)
+                logging.debug(f"Part {part}/{last_part} (total {part_count}) served!")
+                part += 1
+            logging.debug("serving finished")
+        except (GeneratorExit, StopAsyncIteration, asyncio.CancelledError):
+            logging.debug("file serve interrupted")
+
+            raise
+        except Exception as e:
+            print(e)
+            logging.debug("file serve errored", exc_info=True)
+
+    async def handle_request(self):
+        headers = {
+            "content-type": self.mime_type,
+            "content-range": f"bytes {self.offset}-{self.limit-1}/{self.file_size}",
+            "content-length": str(self.limit - self.offset),
+            "accept-ranges": "bytes",
+            "content-transfer-encoding": "Binary",
+            "content-disposition": f'{self.handler.route}; filename="{self.handler.message.file.name}"',
+        }
+        logging.info(
+            f"Serving file in {self.handler.message.file.name}) ; Range: {self.offset} - {self.limit}"
+        )
+        if self.handler.head:
+            body = None
+        else:
+            body = self.download()
+        return StreamingResponse(
+            media_type=self.mime_type,
+            content=body,
+            headers=headers,
+            status_code=206 if self.offset else 200,
+        )

App/Streaming/Utils/Handler.py

@@ -0,0 +1,56 @@
+from .Sanity import Sanity
+from fastapi import Request
+from telethon import TelegramClient
+from telethon.tl.types import Message
+from .Download import Download
+from fastapi.responses import JSONResponse
+
+
+class Handler:
+    req: Request
+    client: TelegramClient
+    chat_id = -1001925049183
+    message: Message
+    route: str
+    head = False
+    sanity: Sanity
+
+    def __init__(self, id, req: Request, client, route=None, head=False):
+        self.head = head
+        self.req = req
+        self.client = client
+        self.sanity = Sanity()
+        self.sanity.client = self.client
+        self.sanity.chat_id = self.chat_id
+        self.sanity.req = self.req
+        self.sanity.file_id = id
+
+    async def sanity_checks(self):
+        self.message = await self.sanity.file_exists()
+
+        try:
+            if not self.message.media:
+                return JSONResponse(
+                    status_code=404,
+                    content={"Error": "File Does not Exist"},
+                )
+        except:
+            return JSONResponse(
+                status=404,
+                content={"Error": "File Does not Exist", "route": self.route},
+            )
+
+        if self.sanity.check_ranges() == False:
+            return JSONResponse(
+                status=416,
+                content={"Error": "416: Range Not Satisfiable"},
+                headers={"Content-Range": f"bytes */{self.message.file.size}"},
+            )
+
+    async def process_request(self):
+        response = await self.sanity_checks()
+        if type(response) is JSONResponse:
+            return response
+
+        # download/stream
+        return await Download(self).handle_request()

App/Streaming/Utils/Sanity.py

@@ -0,0 +1,41 @@
+from telethon import TelegramClient
+from fastapi import Request
+
+
+class Sanity:
+    client: TelegramClient
+    media = None
+    chat_id: int = -1001925049183
+    file_id: int
+    req: Request
+    limit: int
+    offset: int
+
+    async def file_exists(self):
+        try:
+            self.media = await self.client.get_messages(
+                entity=self.chat_id, ids=self.file_id
+            )
+            return self.media
+        except Exception as e:
+            pass
+
+    def check_ranges(self):
+        range_header = self.req.headers.get("Range")
+        if range_header:
+            offset = range_header.split("=")[1].split("-")[0] or 0
+            limit = range_header.split("=")[1].split("-")[1] or self.media.file.size
+        else:
+            offset = 0
+            limit = self.media.file.size
+        self.offset = int(offset)
+        self.limit = int(limit)
+
+        if (
+            (limit > self.media.file.size)
+            or (self.offset < 0)
+            or (self.limit < self.offset)
+        ):
+            return False
+        else:
+            return True

App/Transcription/Model.py

@@ -1,10 +1,5 @@
-import asyncio
 import orm
-import psycopg2
 import datetime
-import ujson
-import pydantic
-from passlib.context import CryptContext
 from App.modelInit import database, models
 
 

App/Transcription/Schemas.py

@@ -1,7 +1,4 @@
-from typing import List, Optional
-from pydantic import EmailStr, BaseModel
-from passlib.context import CryptContext
-from fastapi import UploadFile
+from pydantic import BaseModel
 
 
 class TranscriptionMetadata(BaseModel):

App/Transcription/TranscriptionRoutes.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter, status, Form, UploadFile, File, Query
 from typing_extensions import Annotated
 from .Schemas import UserDetails
+from App import bot
 import aiofiles
 from celery.result import AsyncResult
 from App.Worker import transcription_task
@@ -27,15 +28,28 @@ async def create_file(
         async with aiofiles.open(file.filename, "wb") as f:
             while contents := await file.read(1024 * 1024):
                 await f.write(contents)
+                data = await bot.send_file(
+                    -1001925049183,
+                    file_size=file.size,
+                    caption=file.filename,
+                    file=contents,
+                )
     except Exception:
         return {"message": "There was an error uploading the file"}
     finally:
         await file.close()
 
+    # telegram
+
     # celery task
-    task = transcription_task.delay(file.filename, model)
+    # task = transcription_task.delay(file.filename, model)
 
-    return {"file_size": file.size, "file_name": file.filename, "task_id": task.id}
+    return {
+        "file_size": file.size,
+        "file_name": file.filename,
+        "task_id": task.id,
+        "message_id": data.id,
+    }
 
 
 @transcription_router.get("/tasks/{task_id}")

App/__init__.py

@@ -0,0 +1,9 @@
+import uuid, contextvars
+from telethon import TelegramClient
+
+
+bot = TelegramClient(
+    "transcription_bot",
+    api_id=870972,
+    api_hash="ce2efaca02dfcd110941be6025e9ac0d",
+)

App/app.py

@@ -1,15 +1,17 @@
 from fastapi import FastAPI
+from App import bot
 from .Users.UserRoutes import user_router
 from .modelInit import models, database
 from .Transcription.TranscriptionRoutes import transcription_router
+from .Streaming.StreamingRoutes import streaming_router
 from fastapi.middleware.cors import CORSMiddleware
 import logging
 
 # Configure logging
 logging.basicConfig(
     level=logging.DEBUG,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    datefmt='%Y-%m-%d %H:%M:%S'
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
 )
 
 app = FastAPI()
@@ -27,6 +29,7 @@ app.add_middleware(
 
 @app.on_event("startup")
 async def startup_event():
+    await bot.start()
     await models.create_all()
     if not database.is_connected:
         await database.connect()
@@ -35,6 +38,7 @@ async def startup_event():
 
 @app.on_event("shutdown")
 async def shutdown_event():
+    await bot.stop()
     if not database.is_connected:
         await database.disconnect()
         print("shutting down!")
@@ -47,3 +51,4 @@ async def landing_page():
 
 app.include_router(user_router)
 app.include_router(transcription_router)
+app.include_router(streaming_router)

App/celery_config.py

@@ -8,8 +8,6 @@ timezone = "Europe/Oslo"
 enable_utc = True
 
 broker_url = f"rediss://default:59c9ffda43f61cc18b44a3407a2a7793@master.transcription--legal-stuff--a96n-7tyr.addon.code.run:6379?ssl_cert_reqs=none"
-
-
 result_backend = f"rediss://default:59c9ffda43f61cc18b44a3407a2a7793@master.transcription--legal-stuff--a96n-7tyr.addon.code.run:6379?ssl_cert_reqs=none"
 
 # SSL/TLS and SNI configuration