Files
memestream/db/attachment.py
T
2024-04-08 23:06:42 +10:00

56 lines
1.7 KiB
Python

from urllib.parse import urlparse
from uuid import UUID
import os
from typing import Any, Optional
from urllib3.util import parse_url
from urllib3.util import Url
from sqlalchemy import ForeignKey
from sqlalchemy import UniqueConstraint
from sqlalchemy import BigInteger
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import relationship
from db.base import Base
# Define a model with a JSON column
class Attachment(Base):
__tablename__ = 'attachments'
# can be multiple attachments on a msg
# so we make our own ids for attachments
id: Mapped[UUID] = mapped_column(primary_key=True)
attachment_id: Mapped[int] = mapped_column(BigInteger, nullable=False, unique=True)
msg_id: Mapped[int] = mapped_column(BigInteger, nullable=False, index=True)
url: Mapped[str]
message: Mapped[dict[str, Any]]
guild: Mapped[dict[str, Any]]
channel: Mapped[dict[str, Any]]
vision: Mapped[Optional[dict[str, Any]]] = mapped_column(default=None)
def file_details(self) -> tuple[str, str]:
u = urlparse(self.url)
if not u.path:
raise RuntimeError(f"attachment url missing path {self.url}")
return os.path.splitext(os.path.basename(u.path))
def object_name(self) -> str:
print(self.id)
(filename, ext) = self.file_details()
if self.convert_to_webp():
return f"{self.id}.webp"
return f"{self.id}{ext}"
def convert_to_webp(self) -> bool:
(filename, ext) = self.file_details()
convert_extensions = ['jpg', 'png', 'jpeg', 'webp']
return ext.lower()[1:] in convert_extensions