from sqlalchemy import Column, Integer, String, DateTime, Unicode, DECIMAL, ForeignKey, Boolean, Enum from sqlalchemy.orm import relationship from sqlalchemy.schema import UniqueConstraint from photoapp.dbutils import Base from datetime import datetime import enum as py_enum import uuid import enum class fcategory(py_enum.Enum): image = 0 raw = 1 video = 2 ftypes = dict(jpg=dict(category=fcategory.image, extensions={"jpeg"}, mimes={"image/jpeg"}), png=dict(category=fcategory.image, mimes={"image/png"}), gif=dict(category=fcategory.image, mimes={"image/gif"}), cr2=dict(category=fcategory.raw, mimes={"image/x-canon-cr2"}), xmp=dict(category=fcategory.raw, mimes={"application/octet-stream-xmp"}), psd=dict(category=fcategory.raw, mimes={"image/vnd.adobe.photoshop"}), mp4=dict(category=fcategory.video, mimes={"audio/mp4", "video/mp4"}), mov=dict(category=fcategory.video, mimes={"video/quicktime"})) # set various defaults in ftypes # it should look like: # ftypes = { # : { # "category": , # required, fcategory type # "extensions": {<*default_extension|optional extensions>}, # optional, defaults to set([]) # "mimes": { # required, mimes mapped to this type # # }, # "type": , # optional, defaults to str("image") # }, # } for extension, entry in ftypes.items(): entry["extensions"] = entry.get("extensions", set()) | set([extension]) entry["type"] = entry.get("type", "image") # file extensions we allow known_extensions = set.union(*[i["extensions"] for i in ftypes.values()]) # allowed file types (based on magic identification) known_mimes = set.union(*[i["mimes"] for i in ftypes.values()]) # categorizaiton of media type based on extension # we can pull metadata out of these # jpg, png, gif etc regular_images = set([extension for extension, ftype in ftypes.items() if ftype["category"] == fcategory.image]) # "derived" files, treated as black boxes, we can't open them because proprietary # cr2, xmp, etc files_raw = set([extension for extension, ftype in ftypes.items() if ftype["category"] == fcategory.raw]) # video types # mp4, mov, etc files_video = set([extension for extension, ftype in ftypes.items() if ftype["category"] == fcategory.video]) def mime2ext(mime): """ Given a mime type return the canonical file extension """ for ext, ftype in ftypes.items(): if mime in ftype["mimes"]: return ext raise Exception(f"Unknown mime: '{mime}'") def genuuid(): return str(uuid.uuid4()) def map_extension(ext): for known_ext, ftype in ftypes.items(): if ext in ftype["extensions"]: return known_ext raise Exception(f"Unknown extension: '{ext}'") def generate_storage_path(timestamp, sha, extension): basepath = timestamp.strftime("%Y/%m/%d/%Y-%m-%d_%H.%M.%S") return f"{basepath}_{sha[0:8]}.{extension.lower()}" class PhotoStatus(enum.Enum): private = 0 public = 1 hidden = 2 class PhotoSet(Base): __tablename__ = 'photos' id = Column(Integer, primary_key=True) uuid = Column(Unicode(length=36), unique=True, default=lambda: str(uuid.uuid4())) date = Column(DateTime) date_real = Column(DateTime) date_offset = Column(Integer, default=0) # minutes lat = Column(DECIMAL(precision=11, scale=8)) lon = Column(DECIMAL(precision=11, scale=8)) files = relationship("Photo", back_populates="set") tags = relationship("TagItem", back_populates="set") title = Column(String(length=128)) description = Column(String(length=1024)) slug = Column(String(length=128)) status = Column(Enum(PhotoStatus), default=PhotoStatus.private) def to_json(self, files=True): s = {attr: getattr(self, attr) for attr in {"uuid", "title", "description"}} s["lat"] = str(self.lat) if self.lat else None s["lon"] = str(self.lon) if self.lon else None s["date"] = self.date.isoformat() if files: s["files"] = {i.uuid: i.to_json() for i in self.files} s["tags"] = [t.tag.name for t in self.tags] s["status"] = self.status.name if self.status else PhotoStatus.private.name # duplicate default definition return s class Photo(Base): __tablename__ = 'files' id = Column(Integer, primary_key=True) set_id = Column(Integer, ForeignKey("photos.id")) uuid = Column(Unicode(length=36), unique=True, default=genuuid) set = relationship("PhotoSet", back_populates="files", foreign_keys=[set_id]) size = Column(Integer) width = Column(Integer) height = Column(Integer) orientation = Column(Integer, default=0) hash = Column(String(length=64), unique=True) path = Column(Unicode(length=64)) format = Column(String(length=32)) # TODO how long can a mime string be fname = Column(String(length=64)) # seems generous enough def to_json(self): j = {attr: getattr(self, attr) for attr in {"uuid", "size", "width", "height", "orientation", "format", "hash", "fname"}} j["set"] = self.set.uuid if self.set else None return j class Tag(Base): __tablename__ = 'tags' id = Column(Integer, primary_key=True) uuid = Column(Unicode(length=36), unique=True, default=lambda: str(uuid.uuid4())) created = Column(DateTime, default=lambda: datetime.now()) modified = Column(DateTime, default=lambda: datetime.now()) is_album = Column(Boolean, default=False) # slug-like short name such as "iomtrip" name = Column(String(length=32), unique=True) # longer human-format title like "Isle of Man trip" title = Column(String(length=32)) # url slug like "isle-of-man-trip" slug = Column(String(length=32), unique=True) # fulltext description description = Column(String(length=256)) entries = relationship("TagItem", back_populates="tag") def to_json(self): return {attr: getattr(self, attr) for attr in {"uuid", "is_album", "name", "title", "description"}} class TagItem(Base): __tablename__ = 'tag_items' id = Column(Integer, primary_key=True) tag_id = Column(Integer, ForeignKey("tags.id")) set_id = Column(Integer, ForeignKey("photos.id")) order = Column(Integer, default=0) tag = relationship("Tag", back_populates="entries", foreign_keys=[tag_id]) set = relationship("PhotoSet", back_populates="tags", foreign_keys=[set_id]) UniqueConstraint(tag_id, set_id) class UserStatus(enum.Enum): banned = -1 guest = 0 normal = 1 admin = 2 class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(String(length=64), unique=True) password = Column(String(length=64)) # sha256 status = Column(Enum(UserStatus), default=UserStatus.normal) def to_json(self): return dict(name=self.name, status=self.status.name)