feat(server): serving mimetypes

Signed-off-by: John Andersen <johnandersen777@protonmail.com>

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "atprotobin"
-version = "0.3.1"
+version = "1.0.0"
 description = "ATProto based pastebin"
 readme = {file = "README.md", content-type = "text/markdown"}
 authors = [
@@ -16,6 +16,7 @@ dependencies = [
     "python-magic>=0.4.27",
     "python-multipart>=0.0.19",
     "pygments>=2.18.0",
+    "pydantic>=2.10.2",
     "markdown2>=2.5.1",
 ]
 

src/atprotobin/server.py

@@ -9,6 +9,7 @@
 from typing import Annotated
 
 import markdown2
+from pydantic import BaseModel
 from atproto import AsyncClient, models
 from fastapi import FastAPI, File, UploadFile, Request, Response
 from fastapi.responses import HTMLResponse
@@ -80,7 +81,7 @@ async def create(file: UploadFile):
         None, encode, file_data, file.filename,
     )
     post = await client.send_image(
-        text=data_as_image_hash,
+        text=mimetype,
         image=data_as_image,
         image_alt=data_as_image_hash,
     )
@@ -91,6 +92,12 @@ async def create(file: UploadFile):
         "cid": post.cid,
     }
 
+class Blob(BaseModel):
+    hash_alg: str
+    hash_value: str
+    mimetype: str
+    contents: bytes
+
 async def load_and_decode(post_id: str) -> tuple[str, bytes]:
     post = await client.get_post(post_id)
     blob = await client.com.atproto.sync.get_blob(
@@ -99,14 +106,23 @@ async def load_and_decode(post_id: str) -> tuple[str, bytes]:
             did=did_plcs[atproto_handle],
         ),
     )
-    return await asyncio.get_event_loop().run_in_executor(
+    data_as_image_hash = post.value.embed.images[0].alt
+    hash_alg = data_as_image_hash.split(":", maxsplit=1)[0]
+    hash_value = data_as_image_hash.split(":", maxsplit=1)[1]
+    mimetype, contents = await asyncio.get_event_loop().run_in_executor(
         None, decode, blob,
     )
+    return Blob(
+        hash_alg=hash_alg,
+        hash_value=hash_value,
+        mimetype=mimetype,
+        contents=contents,
+    )
 
 @app.get("/{post_id}")
 async def get(post_id: str):
-    mimetype, output_bytes = await load_and_decode(post_id)
-    return Response(content=output_bytes, media_type=mimetype)
+    blob = await load_and_decode(post_id)
+    return Response(content=blob.contents, media_type=blob.mimetype)
 
 @app.get("/", response_class=HTMLResponse)
 async def root():

src/atprotobin/zip_image.py

@@ -12,7 +12,10 @@ def create_zip_of_files(file_contents, file_name: str = None):
     zip_buffer = BytesIO()
     mimetype = magic.from_buffer(file_contents, mime=True)
     if mimetype == "text/plain" and file_name is not None:
-        mimetype = mimetypes.guess_file_type(file_name)
+        if hasattr(mimetypes, "guess_file_type"):
+            mimetype = mimetypes.guess_file_type(file_name)
+        else:
+            mimetype, _encoding = mimetypes.guess_type(file_name)
     with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zipf:
         zipf.writestr(mimetype, file_contents)
     zip_buffer.seek(0)