file.py

Go to the documentation of this file.

#!/usr/bin/python
# -*- coding: utf-8 -*- 
 
import subprocess
import re
import os
 
import pafera.db
import pafera.modelbase
 
from pafera.utils import *
from pafera.validators import *
 
# File types
FILE_DOCUMENT      =  1;
FILE_IMAGE         =  2;
FILE_AUDIO         =  3;
FILE_VIDEO         =  4;
FILE_ARCHIVE       =  5;
FILE_PROGRAM       =  6;
FILE_SPREADSHEET   =  7;
FILE_PRESENTATION  =  8;
FILE_BOOK          =  9;
FILE_TEXT          =  10;
FILE_DISK_IMAGE    =  11;
FILE_WEBPAGE       =  12;
 
FILE_DELETED       =  126;
FILE_OTHER         =  127;
 
# Flags
FILE_PUBLIC       = 0x01;
FILE_PROTECTED    = 0x02;
FILE_PRIVATE      = 0x04;
 
# *********************************************************************
class system_file(pafera.modelbase.ModelBase):
  """Class for managing all uploaded files. 
  
  This is a system-wide repository for user uploaded files that allows
  easy search and discovery while retaining security and ease of
  management. Files can be made searchable, requires a direct link to 
  access, or locked down to the point where a timed access token is 
  needed to access the file. 
  
  For efficiency, all uploaded images are converted to the webp format,
  audio to mp3, and video to h264 MP4s. If you don't want this behavior,
  modify the system/upload handler in system/__init__.py
  """
  
  _dbfields     = {
    'rid':          ('INTEGER PRIMARY KEY', 'NOT NULL',),
    'filepath':     ('TEXT', 'NOT NULL', BlankValidator()),
    'filename':     ('TEXT', 'NOT NULL', BlankValidator()),
    'extension':    ('TEXT', 'NOT NULL', BlankValidator()),
    'descriptions': ('TRANSLATION', "NOT NULL DEFAULT 0",),
    'modified':     ('DATETIME', 'NOT NULL',),
    'size':         ('BIGINT', 'NOT NULL',),
    'hash':         ('TEXT', 'NOT NULL',),
    'filetype':     ('INT16', 'NOT NULL',),
    'width':        ('INT16', 'NOT NULL DEFAULT 0',),
    'height':       ('INT16', 'NOT NULL DEFAULT 0',),
    'length':       ('INT16', 'NOT NULL DEFAULT 0',),
    'flags':        ('INT', 'NOT NULL DEFAULT 0',),
    'accesstokens': ('DICT', "NOT NULL DEFAULT ''",),
  }
  _dbindexes  = ()
  _dblinks    = []
  _dbdisplay  = ['filename', 'description', 'modified', 'size']
  _dbflags    = pafera.db.DB_SECURE
  
  # Generic file types for faster searches
  TYPES = {
    'document':     1,
    'image':        2,
    'sound':        3,
    'video':        4,
    'archive':      5,
    'program':      6,
    'spreadsheet':  7,
    'presentation': 8,
    'book':         9,
    'text':         10,
    'disk image':   11,
    'webpage':      12,
    'source code':  13,
    'deleted':      126,
    'other':        127,
  }
  
  # -------------------------------------------------------------------
  def __init__(self):
    super().__init__()
    
    self.fullpath = ''
 
  # -------------------------------------------------------------------
  def FullPath(self):
    """Returns the filesystem path for this file.
    """
    
    if self.fullpath:
      return self.fullpath
    
    dirpath = CodeDir(ToShortCode(self.rid))
    
    if self.flags & FILE_PRIVATE:
      return 'private/system/files/' + dirpath + '.' + self.extension
    
    return 'static/system/files/' + dirpath + '.' + self.extension
  
  # -------------------------------------------------------------------
  def ThumbnailFile(self):
    """Returns the filesystem path for an appropiate thumbnail.
    """
    path  = self.MakeThumbnail()
    
    if path:
      return path
    
    if self.filetype == FILE_DOCUMENT:
      return "static/system/icons/document.webp"
    elif self.filetype == FILE_SPREADSHEET:
      return "static/system/icons/spreadsheet.webp"
    elif self.filetype == FILE_PRESENTATION:
      return "static/system/icons/presentation.webp"
    elif self.filetype == FILE_IMAGE:
      return "static/system/icons/image.webp"
    elif self.filetype == FILE_AUDIO:
      return "static/system/icons/audio.webp"
    elif self.filetype == FILE_VIDEO:
      return "static/system/icons/video.webp"
    elif self.filetype == FILE_ARCHIVE:
      return "static/system/icons/archive.webp"
    elif self.filetype == FILE_PROGRAM:
      return "static/system/icons/program.webp"
    elif self.filetype == FILE_BOOK:
      return "static/system/icons/book.webp"
    elif self.filetype == FILE_TEXT:
      return "static/system/icons/text.webp"
    elif self.filetype == FILE_DISK_IMAGE:
      return "static/system/icons/iso.webp"
    elif self.filetype == FILE_WEBPAGE:
      return "static/system/icons/webpage.webp"
    
    return "static/system/icons/unknown.webp"
  
  # -------------------------------------------------------------------
  def ThumbnailURL(self):
    """Returns a URL to display the thumbnail image.
    """
    if self.flags & FILE_PRIVATE:
      return '/system/thumbnailer/' + ToShortCode(self.rid)
    
    return self.ThumbnailFile()[6:]
  
  # -------------------------------------------------------------------
  def Process(self, g):
    """Analyzes the file and inserts it into the system registry, or 
    update any changed files.
    """
    if not self.filename:
      raise Exception('Cannot process empty templateclass.')
 
    fullpath  = self.FullPath()
    stats     = os.stat(fullpath)
    
    size      = stats.st_size 
    modified  = stats.st_mtime;
 
    # Only process updated or different sized files
    if self.size == size and self.modified == modified:
      return;
    
    changed = 1
 
    extension = os.path.splitext(this.filename)[1]
    
    if extension in [
      'jpg',
      'jpeg',
      'gif',
      'png',
      'tif',
      'tiff',
      'svg',
      'psd',
      'bmp',
      'pcd',
      'pcx',
      'pct',
      'pgm',
      'ppm',
      'tga',
      'img',
      'raw',
      'webp',
      'wbmp',
      'eps',
      'cdr',
      'ai',
      'dwg',
      'indd',
      'dss',
      'fla',
      'ss',
    ]:
      filetype =FILE_IMAGE
    elif extension in [
      'doc',
      'docx',
      'odt',
      'rtf',
      'pub',
      'wpd',
      'qxd',
      'cdl',
      'eml',
    ]:
      filetype  = FILE_DOCUMENT
    elif extension in [
      'chm',
      'pdf',
      'epub',
      'mobi',
      'cbr',
      'cbz',
      'cb7',
      'cbt',
      'cba',
      'ibooks',
      'kf8',
      'pdg',
      'azw',
      'prc',
    ]:
      filetype = FILE_BOOK
    elif extension in [
      'mp3',
      'aac',
      'wma',
      'oga',
      'ogg',
      'm4a',
      'wav',
      'aif',
      'aiff',
      'dvf',
      'm4b',
      'm4p',
      'mid',
      'midi',
      'ram',
      'mp2',
    ]:
      filetype  = FILE_AUDIO
    elif extension in [
      'mp4',
      'mpg',
      'avi',
      'wmv',
      'rm',
      'rmvb',
      'ogv',
      'ogm',
      'm4v',
      'mov',
      'flv',
      'f4v',
      '3gp',
      '3gpp',
      'vob',
      'asf',
      'divx',
      'mswmm',
      'asx',
      'amr',
      'mkv',
      'vp8',
      'webm',
    ]:
      filetype  = FILE_VIDEO
    elif extension in [
      'zip',
      'gz',
      'bz2',
      'rar',
      'tar',
      'lz',
      'lzma',
      'lzo',
      'rz',
      'sfark',
      'xz',
      'z',
      '7z',
      's7z',
      'ace',
      'cab',
      'pak',
      'rk',
      'sfx',
      'sit',
      'sitx',
      'qbb',
      'jar',
      'pst',
      'hqx',
      'sea',
    ]:
      filetype  = FILE_ARCHIVE
    elif extension in [
      'exe',
      'msi',
      'bat',
      'cmd',
    ]:
      filetype  = FILE_PROGRAM
    elif extension in [
      'xls',
      'xlsx',
      'ods',
      'qbw',
      'wps',
    ]:
      filetype  = FILE_SPREADSHEET
    elif extension in [
      'ppt',
      'pptx',
      'odp',
    ]:
      filetype  = FILE_PRESENTATION
    elif extension in [
      'txt',
      'ini',
      'cfg',
      'log',
    ]:
      filetype  = FILE_TEXT
    elif extension in [
      'htm',
      'html',
      'mhtml',
    ]:
      filetype  = FILE_WEBPAGE
    elif extension in [
      'deleted',
    ]:
      filetype  = FILE_DELETED
    else:
      filetype  = FILE_OTHER
    
    self.Set(
      size        = size,
      modified    = modified,
      extension   = extension,
      filetype    = filetype,
    );
    
    if self.rid:
      g.db.Update(self)
    else:
      g.db.Insert(self)
      
  # -------------------------------------------------------------------
  def MakeThumbnail(self, thumbsize = '256'):
    """Creates or updates a thumbnail if needed. 
    
    The default thumbnail size is 256x256, but if you are working with 
    high resolution screens, feel free to use 512x512 or higher. 
    
    You'll need to install imagemagick, ffmpeg, and ffmpegthumbnailer
    to use this functionality.
    """
    if self.flags & FILE_PRIVATE:
      thumbfile = 'private/system/thumbs/' + CodeDir(ToShortCode(self.rid)) + '.webp'
    else:
      thumbfile = 'static/system/thumbs/' + CodeDir(ToShortCode(self.rid)) + '.webp'
      
    os.makedirs(os.path.dirname(thumbfile), exist_ok = True)
    
    if os.path.exists(thumbfile):
      stats = os.stat(thumbfile)
      
      if stats.st_mtime > self.modified.timestamp():
        return thumbfile
    
    fullpath  = self.FullPath()
    
    if self.filetype == FILE_IMAGE:
      # No need to make thumbnails for small images
      if self.size < (20 * 1024):
        if self.flags & FILE_PRIVATE:
          thumbfile = 'private/system/files/' + CodeDir(ToShortCode(self.rid)) + '.webp'
        else:
          thumbfile = 'static/system/files/' + CodeDir(ToShortCode(self.rid)) + '.webp'
        
        return thumbfile
      
      result = subprocess.run(['identify', fullpath], stdout=subprocess.PIPE).stdout.decode('utf-8')
      
      if not self.width:
        matches = re.findall('([0-9]+)x([0-9]+)', result, re.MULTILINE)
        
        self.Set(
          width   = matches[0][0],
          height  = matches[0][1],
        )
        
        subprocess.run([
          'convert',
          fullpath,
          '-thumbnail',
          thumbsize + 'x' + thumbsize + '^',
          '-gravity',
          'center',
          '-extent',
          thumbsize + 'x' + thumbsize,
          thumbfile
        ])
    elif self.filetype == FILE_VIDEO:
      if not self.width:
        result = subprocess.run(['ffmpeg', '-i', fullpath], stdout=subprocess.PIPE).stdout.decode('utf-8')
        
        matches = re.findall('Duration: ([0-9]+):([0-9]+):([0-9]+)', result, re.MULTILINE)
        
        self.Set(
          length  = (
            (int(matches[0][0]) * 3600)
            + (int(matches[0][1]) * 60)
            + (int(matches[0][2]))
          )
        )
        
        matches = re.findall('([0-9]+)x([0-9]+)', result, re.MULTILINE)
        
        self.Set(
          width   = matches[0][0],
          height  = matches[0][1],
        )
        
        subprocess.run(['ffmpegthumbnailer', '-i', 'fullpath', '-o', thumbfile, '-s', '512', '-a'])
    
    if os.path.exists(thumbfile):
      return thumbfile
    
    return ''