57 lines
1.7 KiB
Python
57 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)
|
|
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)
|
|
|
|
__table_args__ = (UniqueConstraint('msg_id', 'url', name='ix_attachments_msg_id_url_uc'),)
|
|
|
|
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
|
|
|