223 lines
7.7 KiB
Python
223 lines
7.7 KiB
Python
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 os
|
|
import enum as py_enum
|
|
import uuid
|
|
import enum
|
|
|
|
|
|
APPROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../"))
|
|
|
|
|
|
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"}),
|
|
cr3=dict(category=fcategory.raw,
|
|
mimes={"image/x-canon-cr3"}),
|
|
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 = {
|
|
# <default_extension>: {
|
|
# "category": <fcategory key>, # required, fcategory type
|
|
# "extensions": {<*default_extension|optional extensions>}, # optional, defaults to set([<default_extension>])
|
|
# "mimes": { # required, mimes mapped to this type
|
|
# <mime string>
|
|
# },
|
|
# "type": <str(*image|video)>, # 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])
|
|
regular_mimes = set().union(*[ftype["mimes"] for ftype in ftypes.values() if ftype["category"] == fcategory.image])
|
|
# "derived" files, treated as black boxes, we can't open them because proprietary
|
|
# cr2, cr3, 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])
|
|
video_mimes = set().union(*[ftype["mimes"] for ftype in ftypes.values() 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)
|