Merge pull request #35 from HakierGrzonzo/2-optymalizacja-załączników

Optymalizacja załączników

.gitignore

@@ -1,5 +1,6 @@
 /local/
 /ionic/
+/redis/
 /files/*
 *.pyc
 /db
@@ -17,4 +18,4 @@ api-keys.env
 .idea
 /ionic/editor/android
 /ionic/editor/ios
-node_modules
+node_modules

backend/Dockerfile.worker

@@ -1,5 +1,5 @@
 FROM python:3.10-bullseye
-RUN apt update -y && apt install ffmpeg -y
+RUN apt update -y && apt install ffmpeg imagemagick -y
 
 WORKDIR /app
 

backend/alembic/versions/9fd1f8185c46_add_optimized_mime.py

@@ -0,0 +1,30 @@
+"""Add optimized mime
+
+Revision ID: 9fd1f8185c46
+Revises: c9cf9b80f022
+Create Date: 2022-04-08 22:43:35.395754
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '9fd1f8185c46'
+down_revision = 'c9cf9b80f022'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('Files', sa.Column('optimized_mime', sa.String(length=64), nullable=True))
+    op.add_column('Files', sa.Column('optimized_name', sa.String(length=256), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('Files', 'optimized_name')
+    op.drop_column('Files', 'optimized_mime')
+    # ### end Alembic commands ###

backend/backend/app.py

@@ -89,10 +89,16 @@ async def startup_event():
     tags=["data"],
 )
 
+file_router = FileRouter(fastapi_users, FILE_PREFIX)
 app.include_router(
-    FileRouter(fastapi_users, FILE_PREFIX).get_router(),
+    file_router.get_router(),
     prefix=FILE_PREFIX,
     tags=["files"],
 )
+app.include_router(
+    file_router.get_admin_router(),
+    prefix="/local",
+    tags=["local", "files"]
+)
 
 app.include_router(tea, prefix="/api/auth", tags=["auth"])

backend/backend/database.py

@@ -7,6 +7,7 @@
     SQLAlchemyAccessTokenDatabase,
     SQLAlchemyBaseAccessTokenTable,
 )
+from sqlalchemy.engine.create import create_engine
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 from sqlalchemy.ext.declarative import DeclarativeMeta
 from sqlalchemy.orm import relationship, sessionmaker, declarative_base
@@ -38,7 +39,9 @@ class Files(Base):
         UUID(as_uuid=True), ForeignKey("user.id", use_alter=True), index=True
     )
     mime = Column(String(64))
+    optimized_mime = Column(String(64))
     original_name = Column(String(256))
+    optimized_name = Column(String(256))
     measurement_id = Column(
         Integer, ForeignKey("Measurements.id", use_alter=True), index=True
     )
@@ -73,11 +76,17 @@ class Measurements(Base):
     engine, class_=AsyncSession, expire_on_commit=False
 )
 
+sync_engine = create_engine(DATABASE_URL.replace("+asyncpg", ""))
+sync_session_maker = sessionmaker(sync_engine)
+
 
 async def create_db_and_tables():
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
 
+def get_sync_session():
+    with sync_session_maker() as session:
+        yield session
 
 async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
     async with async_session_maker() as session:

backend/backend/files.py

@@ -7,8 +7,10 @@
 from sqlalchemy.ext.asyncio.session import AsyncSession
 from starlette.responses import FileResponse, Response
 from typing import Tuple
-from fastapi_redis_cache import cache_one_day, cache
-from .models import FileReference, User
+from fastapi_redis_cache import cache
+
+from backend.tasks import send_file_to_be_optimized
+from .models import AdminPanelMsg, FileReference, User
 from .database import get_async_session, Files, Measurements
 from fastapi.routing import APIRouter
 from os import environ, unlink
@@ -32,6 +34,7 @@ def _table_to_file_refrence(self, source: Files) -> FileReference:
             original_name=source.original_name,
             link="{}/file/{}".format(self.prefix, source.id),
             measurement=source.measurement_id,
+            optimized_mime=source.optimized_mime
         )
 
     async def get_all_files(self, session: AsyncSession):
@@ -59,17 +62,18 @@ async def insert_file_to_db(
         return self._table_to_file_refrence(file)
 
     async def get_filename_mime(
-        self, session: AsyncSession, file_id: UUID4
-    ) -> Tuple[str, str]:
+            self, session: AsyncSession, file_id: UUID4, optimized: bool = False
+    ) -> Tuple[str, str, bool]:
         result = await session.execute(
             select(Files).filter(Files.id == file_id)
         )
         res = result.scalars().all()
         if len(res) == 1:
-            return [res[0].mime, res[0].original_name]
+            if optimized and res[0].optimized_mime is not None:
+                return [res[0].optimized_mime, res[0].optimized_name, True]
+            return [res[0].mime, res[0].original_name, False]
         else:
             raise HTTPException(status_code=404, detail=errors.FILE_ERROR)
-        pass
 
     async def delete_file(
         self, session: AsyncSession, user: User, file_id: UUID4
@@ -141,6 +145,7 @@ async def upload_new_file(
                     while content := await uploaded_file.read(8192):
                         await f.write(content)
                 await session.commit()
+                send_file_to_be_optimized(file_refrence)
                 return file_refrence
             except Exception as e:
                 raise e
@@ -177,16 +182,17 @@ async def return_file(
             will most likely differ!**
             """
             try:
-                mime, original_name = await self.get_filename_mime(session, id)
+                mime, original_name, is_optimized = await self.get_filename_mime(session, id, optimized)
                 if isDownload:
                     return FileResponse(
-                        join(FILE_PATH_PREFIX, str(id)),
+                        join(FILE_PATH_PREFIX, str(id) + ("_opt" if is_optimized else "")),
                         media_type=mime,
                         filename=original_name,
                     )
                 else:
                     return FileResponse(
-                        join(FILE_PATH_PREFIX, str(id)), media_type=mime
+                        join(FILE_PATH_PREFIX, str(id) + ("_opt" if is_optimized else "")),
+                        media_type=mime
                     )
             except Exception as e:
                 raise e
@@ -227,3 +233,16 @@ async def delete_file(
                 raise e
 
         return router
+
+    def get_admin_router(self):
+        router = APIRouter()
+
+        @router.get("/reoptimize", response_model=AdminPanelMsg)
+        async def reoptimize_all_files(session: AsyncSession = Depends(get_async_session)):
+            files_query = await session.execute(select(Files))
+            files = [self._table_to_file_refrence(x) for x in files_query.scalars().all()]
+            count = sum(send_file_to_be_optimized(file) for file in files)
+            return AdminPanelMsg(msg=f"Sent {count} files to be optimized")
+
+        return router
+

backend/backend/measurements.py

@@ -48,6 +48,7 @@ def _table_to_model(self, source: Measurements) -> Measurement:
                         original_name=x.original_name,
                         link="{}/file/{}".format(self.file_prefix, x.id),
                         measurement=x.measurement_id,
+                        optimized_mime=x.optimized_mime,
                     )
                     for x in source.files
                 ]

backend/backend/models.py

@@ -43,6 +43,7 @@ class FileEntry(BaseModel):
     file_id: UUID4
     original_name: str
     mime: str
+    optimized_mime: Optional[str]
     measurement: int
     owner: UUID4
 
@@ -65,6 +66,9 @@ class Measurement(_protoMeasurement):
     files: list[FileReference]
     weather: Optional[Weather]
 
+class AdminPanelMsg(BaseModel):
+    msg: str
+
 
 class CreateMeasurement(_protoMeasurement):
     pass

backend/backend/tasks.py

@@ -4,7 +4,13 @@
 from dramatiq.brokers.redis import RedisBroker
 from sqlalchemy.sql.expression import select
 from .utils import run_as_sync
-from .database import get_async_session, Measurements
+from .database import Files, Measurements, get_sync_session
+from backend.models import FileReference
+import ffmpeg
+from wand.image import Image
+
+
+FILE_PATH_PREFIX = os.environ["FILE_PATH"]
 
 owm_token = os.environ.get("OPEN_WEATHER_MAP")
 if owm_token:
@@ -15,25 +21,22 @@
 redis_broker = RedisBroker(host="redis")
 dramatiq.set_broker(redis_broker)
 
-
 @dramatiq.actor
 def hello_queue():
     os.system("ffmpeg -version")
     print("hello world")
 
 
 @dramatiq.actor
-@run_as_sync
-async def on_new_location(measurement):
+def on_new_location(measurement):
     # TODO add redis time lat long cache
     mgr = owm.weather_manager()
-    async for session in get_async_session():
-        old = await session.execute(
+    for session in get_sync_session():
+        old = session.execute(
             select(Measurements).filter(
                 Measurements.id == measurement["measurement_id"]
-            )
-        )
-        old = old.unique().scalars().all()
+            ) 
+        ).unique().scalars().all()
         if len(old) != 1:
             raise Exception("Failed to get measurement from db to set weather")
         target = old[0]
@@ -48,5 +51,59 @@ async def on_new_location(measurement):
         target.humidity = w.weather.humidity
         target.weather_status = w.weather.detailed_status
         target.wind_speed = w.weather.wind()["speed"]
-        await session.commit()
+        session.commit()
     print(f"Got weather for {measurement['measurement_id']}")
+
+
+@dramatiq.actor
+def ffmpeg_compress_audio(audio_uuid):
+    ffmpeg.input(os.path.join(FILE_PATH_PREFIX, audio_uuid)).output(
+        os.path.join(FILE_PATH_PREFIX, audio_uuid + "_opt"),
+        f="adts",
+        acodec="aac",
+        audio_bitrate=128 * 1024,
+    ).overwrite_output().run()
+    for session in get_sync_session():
+        f_query = session.execute(select(Files).filter(Files.id == audio_uuid))
+        fs = f_query.scalars().all()
+        if len(fs) != 1:
+            raise Exception("Failed to process file, not found in db")
+        f = fs[0]
+        f.optimized_name = f.original_name + ".aac"
+        f.optimized_mime = "audio/aac"
+        session.commit()
+    print(f"Transcoded {audio_uuid} successfully")
+
+
+@dramatiq.actor
+def magick_compress_picture(picture_uuid):
+    try:
+        img = Image(filename=os.path.join(FILE_PATH_PREFIX, picture_uuid))
+    except:
+        print(f"Picture {picture_uuid} is unopenable!")
+        return
+    img.resize(img.width // 10, img.height // 10)
+    img_webp = img.convert('webp')
+    img_webp.save(filename=os.path.join(FILE_PATH_PREFIX, picture_uuid + "_opt"))
+    for session in get_sync_session():
+        f_query = session.execute(select(Files).filter(Files.id == picture_uuid))
+        fs = f_query.scalars().all()
+        if len(fs) != 1:
+            raise Exception("Failed to process file, not found in db")
+        f = fs[0]
+        f.optimized_name = f.original_name + (
+            ".webp" if not f.original_name.endswith(".webp") else ""
+        )
+        f.optimized_mime = "image/webp"
+        session.commit()
+    print(f"Converted {picture_uuid} successfully")
+
+def send_file_to_be_optimized(file: FileReference) -> bool:
+    if "audio" in file.mime.lower():
+        ffmpeg_compress_audio.send(str(file.file_id))
+        return True
+    elif "image" in file.mime.lower():
+        magick_compress_picture.send(str(file.file_id))
+        return True
+    return False
+

backend/backend/utils.py

@@ -2,6 +2,10 @@
 
 def run_as_sync(f):
     def inner(*args, **kwargs):
-        loop = asyncio.new_event_loop()
+        try: 
+            loop = asyncio.get_event_loop()
+        except:
+            loop = asyncio.new_event_loop()
         return loop.run_until_complete(f(*args, **kwargs))
     return inner
+

backend/requirements.txt

@@ -11,3 +11,5 @@ psycopg2
 fastapi-redis-cache-hakiergrzonzo==0.2.6
 dramatiq
 pyowm
+ffmpeg-python
+wand

docker-compose.yml

@@ -44,6 +44,9 @@ services:
 
   redis:
     image: redis
+    command: redis-server --save 60 1 --loglevel warning
+    volumes: 
+      - "./redis:/data"
 
   db:
     image: postgres:latest