In the three entries below, the
first field is a space-separated list of users who can
access the {item} (use a single*
for anyone, or add a * after your
username so that only logged-in users can see the {item}).
Second field is a space-separated list of IPs/subnets
which can access the {item} (use * for
anywhere). Note that subnets can be specified as
xxx.yyy (no trailing dot). Third field
is the release date. To block a user or IP/subnet prepend a
! as in !Julia
Remember to open access once you are ready to make the
{item} public. All photos are individually subject
to the same ACL as the gallery and cannot be accessed
otherwise.
"""
%
% import random
% import os
% import copy
% import time
% import datetime
% import json
% import socket
% import re
% import imghdr
%
% no_pil = False
% try:
% from PIL import Image, ImageDraw, ImageOps
% import PIL.ExifTags as PILX
% except:
% logging.error(f'{log_id}: could not import PIL')
% no_pil = True
% ERR = 'you need to install PIL'
% end
% no_bleach = False
% try:
% import bleach
% except:
% logging.error(f'{log_id}: could not import bleach')
% no_bleach = True
% ERR = 'you need to install bleach'
% end
% no_ns = False
% try:
% import natsort
% except:
% logging.warning(f'{log_id}: could not import natsort')
% no_ns = True
% end
%
% pages = CONFIG['dir_pages']
% assets = CONFIG['page_assets']
% valid_slug = re.compile(CONFIG['slug_re'])
% hst_ext = '.hst.png'
% tn_ext = '.tn.jpg'
% showcase = PAGE.get('showcase', '')
% sc_data = {}
% if showcase:
% with open(os.path.join(showcase, 'showcase.koi'), "r", encoding='utf-8') as fd:
% sc_koi = json.load(fd)
% end
% for real_id, meta in sc_koi['library'].items():
% sc_data[meta['orig_id']] = (real_id, meta['hidden'])
% end
% end
%
% #####################################################################################
%
% def check_slug(slug):
% """
check_slug(slug[str]) -> (slug[str], ERR[str])
Check the validity of the slug.
"""
% logging.debug(f'{log_id}: executing "check_slug({slug=})"')
% ERR = ''
% if not valid_slug.match(slug):
% ERR = 'slug does not match slug_re'
% logging.debug(f'{log_id}: {ERR}')
% return (slug, ERR)
% end
% if os.path.isdir(os.path.join(os.path.dirname(ME['path']), slug)):
% ERR = 'slug already exists'
% logging.debug(f'{log_id}: {ERR}')
% end
% return (slug, ERR)
% end
%
% #####################################################################################
%
% def check_editor(must=True):
% """
check_editor(must[bool]) -> editor[bool]
Check editor credentials. Unless input has been tampered with this
function should always return, hence the finality of raising a SystemExit
if "must" is "True".
"""
% logging.debug(f'{log_id}: executing "check_editor({must=})"')
% editor = False
% if user and ('*' in PAGE['editors'] or user.lower() in PAGE['editors']):
% editor = True
% end
% if must and not editor:
% logging.error(f'{log_id}: user "{user}" is not a editor (tampering?)')
% raise SystemExit
% end
% return editor
% end
%
% #####################################################################################
%
% def get_img_id():
% """
get_img_id() -> img_id[str]
Get an image from QUERY and verify its validity. Unless input has been tampered
with this function should always return, hence the finality of raising a SystemExit.
"""
% logging.debug(f'{log_id}: executing "get_img_id()"')
% img_id = QUERY.get('img_id', '')
% if img_id not in PAGE['library']:
% logging.error(f'{log_id}: invalid image "{img_id}" (tampering?)')
% raise SystemExit
% end
% return img_id
% end
%
% #####################################################################################
%
% def get_exif(img_fp):
% """
get_exif(img_fp[str]) -> exif_data[dict]
Given an image's full path return its EXIF data.
"""
% logging.debug(f'{log_id}: executing "get_exif({img_fp=})"')
% im = Image.open(img_fp)
% exif_data = {}
% exif2label = {'DateTimeOriginal': 'date taken', 'ExposureTime': 'exposure time', \
% 'FocalLength': 'focal length (mm)', 'FNumber': 'aperture', \
% 'ISOSpeedRatings': 'ISO', 'ExposureProgram': 'exposure', \
% 'Flash': 'flash', 'MeteringMode': 'metering mode', \
% 'WhiteBalance': 'white balance', 'FocalLengthIn35mmFilm': '35mm equiv', \
% 'Model': 'camera model', 'Flash': 'flash'}
% # http://www.awaresystems.be/imaging/tiff/tifftags/privateifd/exif.html
% ep = {0: 'not defined', 1: 'manual', 2: 'normal program', \
% 3: 'aperture priority', 4: 'shutter priority', 5: 'creative program', \
% 6: 'action program', 7: 'portrait mode', 8: 'landscape mode'}
% mm = {0: 'unknown', 1: 'average', 2: 'center weighted average', 3: 'spot', \
% 4: 'multi spot', 5: 'pattern', 6: 'partial', 255: 'other', 65535: 'unknown'}
% wb = {0: 'auto', 1: 'manual'}
% # Full flash codes at:
% # http://www.awaresystems.be/imaging/tiff/tifftags/privateifd/exif/flash.html
% # Real way of doing this:
% # (hex code) & (bit value) == (bit value)
% # e.g. for strobe light (bits 1 and 2 on => 000110):
% # 0x001D & int('000110', 2) == int('000110', 2) => False
% # 0x001F & int('000110', 2) == int('000110', 2) => True
% # Flash is off for: 0x000 0x010 0x018 0x020
% try:
% exif_it = im._getexif().items()
% except Exception as e:
% logging.debug(f'{log_id}: unable to extract EXIF data')
% return exif_data
% end
% if exif_it:
% exif_dict = dict((PILX.TAGS[k], v) for (k, v) in exif_it if k in PILX.TAGS)
% end
% for k, v in exif_dict.items():
% if k in exif2label:
% if k == 'ExposureProgram':
% exif_data[exif2label[k]] = ep[v]
% elif k == 'MeteringMode':
% exif_data[exif2label[k]] = mm[v]
% elif k == 'WhiteBalance':
% exif_data[exif2label[k]] = wb[v]
% elif k == 'FocalLength':
% exif_data[exif2label[k]] = f'{int(v)}'
% elif k == 'FNumber':
% exif_data[exif2label[k]] = f'f/{v}'
% elif k == 'Model':
% exif_data[exif2label[k]] = v.lower()
% elif k == 'ExposureTime':
% exif_data[exif2label[k]] = f'{v*100}/100s'
% elif k == 'Flash':
% if v in [0, 16, 24, 32]:
% exif_data[exif2label[k]] = 'off'
% else:
% exif_data[exif2label[k]] = 'on'
% end
% else:
% exif_data[exif2label[k]] = v
% end
% end
% end
% (xpix, ypix) = im.size
% exif_data['geometry'] = f'{xpix}x{ypix}px'
% exif_data['size'] = f'{os.stat(img_fp).st_size/1000}kB'
% return exif_data
% end
%
% #####################################################################################
%
% def make_tn(img_fp):
% """
make_fn(img_fp[str]) -> tn_fp[str]
Given an image's full path generate a thumbnail and retun its full path.
"""
% logging.debug(f'{log_id}: executing "make_tn({img_fp=})"')
% tn_fp = ''
% tn_fp = img_fp+tn_ext
% is_tn = os.path.exists(tn_fp)
% img_ts = os.stat(img_fp).st_mtime
% if is_tn and os.stat(tn_fp).st_mtime >= img_ts:
% return tn_fp
% end
% im = Image.open(img_fp)
% try:
% if is_tn:
% os.remove(tn_fp)
% end
% tn_dict = PAGE['thumbnail']
% if tn_dict['square']:
% TN = ImageOps.fit(im, tn_dict['size'], Image.ANTIALIAS)
% else:
% TN = im.copy()
% TN.thumbnail(tn_dict['size'], Image.ANTIALIAS)
% end
% if TN.mode != 'RGB':
% TN = TN.convert('RGB')
% end
% TN.save(tn_fp, tn_dict['type'], quality=tn_dict['quality'], \
% optimize=tn_dict['optimize'], progressive=tn_dict['progressive'])
% os.chmod(tn_fp, CONFIG['wwwfperms'])
% except Exception as e:
% logging.debug(f'{log_id}: error creating thumbnail {tn_fp} [{e}]')
% tn_fp = ''
% end
% return tn_fp
% end
%
% #####################################################################################
%
% # http://tophattaylor.blogspot.ca/2009/05/python-rgb-histogram.html
% # http://www.cambridgeincolour.com/tutorials/histograms2.htm
% # http://stackoverflow.com/questions/596216/formula-to-determine-brightness-of-rgb-color
% # http://alienryderflex.com/hsp.html
% def make_hst(img_fp):
% """
make_hst(img_fp[str]) -> hst_fp[str]
Given an image's full path generate a histogram (if PAGE['histogram']['create']
is True) and retun its full path.
"""
% logging.debug(f'{log_id}: executing "make_hst({img_fp=})"')
% hst_fp = ''
% if not PAGE['histogram']['create']:
% return hst_fp
% end
% hst_fp = img_fp+hst_ext
% is_hst = os.path.exists(hst_fp)
% img_ts = os.stat(img_fp).st_mtime
% if is_hst and os.stat(hst_fp).st_mtime >= img_ts:
% return hst_fp
% end
% im = Image.open(img_fp)
% try:
% if is_hst:
% os.remove(hst_fp)
% end
% rgb_hst = im.histogram()
% depth = len(rgb_hst)
% bdr_col = tuple(PAGE['histogram']['border_col'])
% fg_col = tuple(PAGE['histogram']['fg_col'])
% bg_col = tuple(PAGE['histogram']['bg_col'])
% lin_col = tuple(PAGE['histogram']['lines_col'])
% height = PAGE['histogram']['height']
% n_lines = PAGE['histogram']['n_lines']
% if depth in [768, 1024]:
% r_hst = rgb_hst[0:256]
% g_hst = rgb_hst[256:512]
% b_hst = rgb_hst[512:768]
% rw = 0.299; gw = 0.587; bw = 0.114
% hst_data = [(rw*r_hst[i]**2 + gw*g_hst[i]**2 + bw*b_hst[i]**2)**0.5 for \
% i in range(0,256)]
% elif depth == 256:
% hst_data = rgb_hst
% else:
% raise ValueError
% end
% width = len(hst_data)
% y_scale = height/max(hst_data)
% canvas = Image.new("RGBA", (width, height), bg_col)
% draw = ImageDraw.Draw(canvas)
% if n_lines:
% xmarker = width/n_lines
% x = 0
% for i in range(1, n_lines+1):
% draw.line((x, 0, x, height), fill=lin_col)
% x+=xmarker
% end
% end
% x=0
% for i in hst_data:
% if int(i)==0:
% pass
% else:
% draw.line((x, height, x, height-(i*y_scale)), fill=fg_col)
% end
% x+=1
% end
% # Top
% draw.line((0, 0, width, 0), fill=bdr_col)
% # Right side
% draw.line((width-1, 0, width-1, height), fill=bdr_col)
% # Bottom
% draw.line((0, height-1, width, height-1), fill=bdr_col)
% # Left side
% draw.line((0, 0, 0, height), fill=bdr_col)
% canvas.save(hst_fp, 'PNG')
% os.chmod(hst_fp, CONFIG['wwwfperms'])
% except Exception as e:
% logging.debug(f'{log_id}: error creating histogram {hst_fp} [{e}]')
% hst_fp = ''
% end
% return hst_fp
% end
%
% #####################################################################################
%
% def get_slide(slides):
% """
make_fn(slides[list]) -> (img_id[str], nxt[str], prev[str])
Get an img_id (using QUERY data if present) and return the prior
and net slides (again as an img_id of each). "img_id" is the name of
the file, btw.
"""
% logging.debug(f'{log_id}: executing "get_slide({slides=})"')
% tot = len(slides)
% img_id = QUERY.get('img_id', '')
% # The following also sanitizes img_id:
% if not img_id:
% img_id = slides[random.randint(0, tot-1)]
% elif img_id not in slides:
% img_id = slides[0]
% end
% current = slides.index(img_id)
% if img_id == slides[-1]:
% nxt = slides[0]
% else:
% nxt = slides[min(current+1, tot-1)]
% end
% if img_id == slides[0]:
% prev = slides[-1]
% else:
% prev = slides[max(current-1, 0)]
% end
% return (img_id, nxt, prev, tot, current+1)
% end
%
% #####################################################################################
%
% def save_upload():
% """
save_upload() -> (files[dict], ERR[str])
Save all files from an upload returning a dictionary with
information about each file.
"""
% logging.debug(f'{log_id}: executing "save_upload()"')
% files = {}
% ERR = ''
% max_msg = f'max: {CONFIG["upload_max_size"]/(1024*1024)}MB'
% if not UPLOAD:
% ERR = 'no upload file found'
% logging.debug(f'{log_id}: "{ERR}"')
% else:
% for key, up in UPLOAD.items():
% img_id = up['safe_name']
% if os.path.splitext(img_id)[1].lower() not in PAGE['image']['formats']:
% logging.error(f'{log_id}: attempted to upload a non-valid image "{img_id}" (tampering?)')
% raise SystemExit
% end
% try:
% if not up['OK']:
% logging.debug(f'{log_id}: file "{img_id}" is too large ({max_msg})')
% raise IOError
% end
% magic = imghdr.what('', h=up['file_data'])
% if not magic or '.'+magic not in PAGE['image']['formats']:
% logging.error(f'{log_id}: attempted to upload a non-valid image "{img_id}" of type {magic}')
% raise SystemExit
% end
% with open(os.path.join(ME['path'], img_id), 'wb') as fd:
% fd.write(up['file_data'])
% end
% ftype = up['content_type']
% fsize = len(up['file_data'])
% try:
% update_library(img_id, action='add')
% OK = True
% status = 'success'
% except Exception as e:
% logging.debug(f'{log_id}: unable to update image library [{e}]')
% OK = False
% status = 'unable to update library'
% end
% files[img_id] = {'OK': OK, 'status': status, 'type': ftype, 'size': fsize}
% logging.debug(f'{log_id}: saved file "{img_id}"')
% except IOError:
% files[img_id] = {'OK': False, 'status': f'file too large ({max_msg})'}
% except Exception as e:
% logging.error(f'{log_id}: unable to save file "{img_id}" [{e}]')
% files[img_id] = {'OK': False, 'status': 'unable to save file'}
% end
% end
% end
% return (files, ERR)
% end
%
% #####################################################################################
%
% def process_acl():
% """
process_acl() -> (acl_users[str/list], acl_ips[str/list], acl_time[int], editors[list])
Process an ACL from QUERY parameters.
"""
% logging.debug(f'{log_id}: executing "process_acl()"')
% users_lc = [i.lower() for i in USERS]
% if QUERY['who'].strip() != '*':
% allowed_users = [i.lower() for i in QUERY['who'].split() if i[0] != '!']
% blocked_users = [i[1:].lower() for i in QUERY['who'].split() if i[0] == '!']
% end
% if QUERY['where'].strip() != '*':
% allowed_ips = [i for i in QUERY['where'].split() if i[0] != '!']
% blocked_ips = [i[1:] for i in QUERY['where'].split() if i[0] == '!']
% end
% if QUERY['who'].strip() == '*':
% acl_users = '*'
% else:
% acl_allowed = [i for i in allowed_users if i in users_lc or i == '*']
% acl_blocked = ['!'+i for i in blocked_users if i in users_lc]
% acl_users = acl_allowed + acl_blocked
% end
% # Subnets must be of the form xxx.yyy (no trailing ".")
% if QUERY['where'].strip() == '*':
% acl_ips = '*'
% else:
% acl_allowed = [i for i in allowed_ips if socket.inet_aton(i)]
% acl_blocked = ['!'+i for i in blocked_ips if socket.inet_aton(i)]
% acl_ips = acl_allowed + acl_blocked
% end
% editors = [i.lower() for i in QUERY['editors'].split()]
% editors = [i for i in editors if i in users_lc or i == '*']
% if user.lower() not in editors:
% editors = [user.lower()] + editors
% end
% acl_time = int(time.mktime(time.strptime(QUERY['when'], clock_format)))
% return (acl_users, acl_ips, acl_time, editors)
% end
%
% #####################################################################################
%
% def update_gallery():
% """
update_gallery() -> (page[str], ERR[str])
Update the gallery ACL in "gallery.koi".
"""
% logging.debug(f'{log_id}: executing "update_gallery()"')
% ERR = ''
% page = ME['page']
% try:
% title = bleach.clean(QUERY['title'], strip=True)
% keywords = bleach.clean(QUERY['keywords'], strip=True)
% except Exception as e:
% # This CANNOT happen unless input has been tampered with.
% logging.error(f'{log_id}: unexpected QUERY [{e}] (tampering?)')
% raise SystemExit
% end
% koi_data = copy.deepcopy(PAGE)
% koi_file_fp = os.path.join(ME['path'], 'gallery.koi')
% (acl_users, acl_ips, acl_time, editors) = process_acl()
% koi_data['editors'] = editors
% koi_data['title'] = title
% koi_data['keywords'] = keywords
% koi_data['acl']['gallery.koi'] = {'users': acl_users, 'groups': [], 'ips': acl_ips, \
% 'time': acl_time}
% try:
% with open(koi_file_fp, "w", encoding='utf-8') as fd:
% json.dump(koi_data, fd, ensure_ascii=False)
% end
% os.chmod(koi_file_fp, CONFIG['wwwfperms'])
% except Exception as e:
% ERR = 'unable to update gallery'
% logging.error(f'{log_id}: {ERR} "{koi_file_fp}" [{e}]')
% return (page, ERR)
% end
% slug = QUERY.get('slug', '')
% if slug != page:
% (slug, ERR) = check_slug(slug)
% if not ERR:
% try:
% slug_fp = os.path.join(os.path.dirname(ME['path']), slug)
% os.rename(ME['path'], slug_fp)
% logging.debug(f'{log_id}: renamed slug "{page}" to "{slug}"')
% page = slug
% except Exception as e:
% ERR = 'unable to rename slug'
% logging.error(f'{log_id}: {ERR} "{slug}" [{e}]')
% end
% end
% end
% return (page, ERR)
% end
%
% #####################################################################################
%
% def update_library(img_id, action):
% """
update_library(img_id[str], action[str])
Update the image library/ACL in "gallery.koi". "action" can be "add", "delete", or
"update". For "add" the img_id input is sanitized by the upload routine (or added
through the back-end). For the latter two the img_id must already be present in the
library.
"""
% logging.debug(f'{log_id}: executing "update_library({img_id=}, {action=})"')
% koi_file_fp = os.path.join(ME['path'], 'gallery.koi')
% # We need a fresh re-read if we're going to run this function in a loop
% # (the PAGE info becomes stale after the first loop).
% with open(koi_file_fp, "r", encoding='utf-8') as fd:
% koi_data = json.load(fd)
% end
% img_fp = os.path.join(ME['path'], img_id)
% if action == 'delete':
% if os.path.isfile(img_fp):
% os.remove(img_fp)
% end
% if os.path.isfile(img_fp+tn_ext):
% os.remove(img_fp+tn_ext)
% end
% if os.path.isfile(img_fp+hst_ext):
% os.remove(img_fp+hst_ext)
% end
% del koi_data['library'][img_id]
% if img_id in koi_data['acl']:
% del koi_data['acl'][img_id]
% end
% with open(koi_file_fp, "w", encoding='utf-8') as fd:
% json.dump(koi_data, fd, ensure_ascii=False)
% end
% os.chmod(koi_file_fp, CONFIG['wwwfperms'])
% return
% end
% if action == 'add':
% hide = PAGE['image']['hide_new']
% elif action == 'update':
% if QUERY.get('hide', '') == 'show':
% hide = False
% else:
% hide = True
% end
% end
% koi_data['library'][img_id] = {'watermark': '', 'tags': [], 'caption': '', \
% 'comments': {}, 'marked': '', 'data': {}, \
% 'hidden': hide, 'talent': {}, 'www': '', 'keywords': '', \
% 'email': '', 'photographer': '', 'credits': {}, \
% 'social': {}, 'location': '', 'copyright': ''}
% with open(koi_file_fp, "w", encoding='utf-8') as fd:
% json.dump(koi_data, fd, ensure_ascii=False)
% end
% os.chmod(koi_file_fp, CONFIG['wwwfperms'])
% return
% end
%
% #####################################################################################
%
% def get_photos(editor):
% """
get_photos(editor[bool]) -> (slides[list], show[str])
Return a list of images (photo file names) and a "show" hidden-input string to
display hidden images (empty if proper tag is not in QUERY).
"""
% logging.debug(f'{log_id}: executing "get_photos(, {editor=})"')
% images = [i for i in PAGE['library']]
% if not no_ns:
% images = natsort.natsorted(images, alg=natsort.ns.IGNORECASE)
% end
% if editor:
% slides = images
% show = ''
% elif PAGE['image']['unhide_tag'] in QUERY:
% slides = images
% show = f''
% else:
% slides = [i for i in images if not PAGE['library'][i]['hidden']]
% show = ''
% end
% return (slides, show)
% end
%
% #####################################################################################
%
% def sync_db():
% """
sync_db() -> (refresh[str], ERR[str])
Synchronize images on disk with those in the gallery.koi library and return a
refresh HTML string (empty if not needed).
"""
% logging.debug(f'{log_id}: executing "sync_db()"')
% ERR = ''
% refresh = ''
% images = [i for i in ME['files'] if os.path.splitext(i)[1].lower() in PAGE['image']['formats']]
% images = [i for i in images if not (i.endswith(tn_ext) or i.endswith(hst_ext))]
% # Sync database (in case of back-end additions/deletions).
% for img_id in images:
% if img_id not in PAGE['library']:
% try:
% update_library(img_id, action='add')
% refresh = ''
% except Exception as e:
% logging.debug(f'{log_id}: unable to update image library [{e}]')
% ERR = 'unable to update library'
% end
% end
% end
% for img_id in PAGE['library']:
% if img_id not in images:
% try:
% update_library(img_id, action='delete')
% refresh = ''
% except Exception as e:
% logging.debug(f'{log_id}: unable to update image library [{e}]')
% ERR = 'unable to update library'
% end
% end
% end
% return (refresh, ERR)
% end
%
% #####################################################################################
%
% (refresh, ERR) = sync_db()
% if not (refresh or ERR):
% editor = check_editor(must=False)
% (slides, show) = get_photos(editor)
% tot = len(slides)
% end
% logging.info(f'{log_id}: processing DOCTYPE')
{{!refresh}}
% include('head.tpl')
{{PAGE['title']}}
% include('header.tpl', show_search=True, show_login=True, hr=False)
% #####################################################################################
% if ERR:
% # Looping over slides keeps thumbnails hidden too.
% for img_id in slides:
% tn_fp = make_tn(os.path.join(ME['path'], img_id))
% if editor and PAGE['library'][img_id]['hidden']:
% h = ' hidden'
% else:
% h = ''
% end
% if img_id in sc_data:
% s = ' showcase'
% else:
% s = ''
% end
% end
% if editor:
gallery ACL
Gallery curators (space-separated list of user names):
gallery: {{PAGE['title']}} ({{idx}}/{{tot}}) {{!showcase_url}}
% for i in ['name', 'date taken', 'geometry', 'size', 'camera model', 'flash', 'white balance', 'metering mode', 'exposure']:
% if i not in exif:
% continue
% end
{{i}}: {{exif[i]}}
% end
% for i in ['focal length (mm)', 'exposure time', 'aperture', 'ISO']:
% if i not in exif:
% continue
% end
{{i}}: {{exif[i]}}
% end
% if hst_fp:
% end
% if editor:
% end
% #####################################################################################
% else:
% logging.debug(f'{log_id}: unknown request')
could not process request
% end
% #####################################################################################