Mercurial > hg > index.fcgi > www > www-1

annotate pics3/pics_flask_app.py @ 140:ae6b4032a912

find changesets by author, revision, files, or words in the commit message
myrss2: FEEDS: Add jabberwocking.com, remove danielmiessler.com
author paulo
date Mon, 03 Jul 2023 04:40:23 +0000
parents 06f97e38e1b2
children
rev   line source
paulo@124 1 import csv
paulo@124 2 import datetime
paulo@124 3 import random
paulo@124 4 import os
paulo@133 5 import time
paulo@124 6
paulo@124 7 import flask
paulo@124 8 import google.cloud.storage
paulo@124 9 from html3.html3 import HTML
paulo@124 10
paulo@124 11 app = flask.Flask(__name__)
paulo@124 12
paulo@124 13 GCS_CLIENT = google.cloud.storage.Client()
paulo@124 14 GCS_BUCKET = GCS_CLIENT.get_bucket(os.environ.get("GCS_BUCKET"))
paulo@130 15 PIN = os.environ.get("PIN")
paulo@130 16
paulo@130 17
paulo@130 18 class PinFailError(Exception):
paulo@130 19 def __str__(self):
paulo@130 20 return "PIN FAIL!"
paulo@130 21
paulo@130 22 class PinSetupError(Exception):
paulo@130 23 def __str__(self):
paulo@130 24 return "PIN SETUP ERROR!"
paulo@124 25
paulo@124 26
paulo@124 27 class PicsDialect(csv.Dialect):
paulo@124 28 delimiter = '\t'
paulo@124 29 quoting = csv.QUOTE_NONE
paulo@124 30 lineterminator = '\n'
paulo@124 31
paulo@124 32 PICSDIALECT = PicsDialect()
paulo@124 33
paulo@124 34
paulo@133 35 class BlobCache(object):
paulo@133 36 def __init__(self):
paulo@133 37 self._bloblist_cache = {}
paulo@133 38 self._bloblist_ttl = 60*15 # 15 minutes
paulo@133 39 self._public_url_cache = {}
paulo@133 40
paulo@133 41 def get_list_blobs(self, prefix):
paulo@133 42 if (prefix not in self._bloblist_cache
paulo@133 43 or time.time() > (self._bloblist_cache[prefix][1] + self._bloblist_ttl)):
paulo@133 44 self._bloblist_cache[prefix] = ([i for i in GCS_CLIENT.list_blobs(GCS_BUCKET, prefix=prefix)],
paulo@133 45 time.time())
paulo@133 46 return self._bloblist_cache[prefix][0]
paulo@133 47
paulo@133 48 def cache_public_url(self, blob):
paulo@133 49 if blob.name not in self._public_url_cache:
paulo@133 50 self._public_url_cache[blob.name] = blob.public_url
paulo@133 51
paulo@133 52 def get_public_url(self, blob_name):
paulo@133 53 return self._public_url_cache[blob_name]
paulo@133 54
paulo@133 55 BLOBCACHE = BlobCache()
paulo@133 56
paulo@133 57
paulo@124 58 def _parse_dt(dts):
paulo@124 59 return datetime.datetime.strptime(dts, "%Y%m%d")
paulo@124 60
paulo@124 61
paulo@124 62 def _format_dt(dt):
paulo@124 63 return dt.strftime("%Y-%m-%d")
paulo@124 64
paulo@124 65
paulo@124 66 def _numeric_pad_basename(path, maxdigits=20):
paulo@124 67 return os.path.basename(path).zfill(maxdigits)
paulo@124 68
paulo@124 69
paulo@124 70 def _get_images(d):
paulo@124 71 exts = (".jpg", ".webm")
paulo@124 72 thumb_dir = f"pics/{d}/thumbs"
paulo@124 73 browse_dir = f"pics/{d}/browse"
paulo@124 74
paulo@133 75 thumb_fns = []
paulo@133 76 for i in BLOBCACHE.get_list_blobs(thumb_dir):
paulo@133 77 if i.name.endswith(exts):
paulo@133 78 thumb_fns.append(i.name)
paulo@133 79 BLOBCACHE.cache_public_url(i)
paulo@124 80 thumb_fns = sorted(thumb_fns, key=_numeric_pad_basename)
paulo@124 81
paulo@133 82 browse_contents = set()
paulo@133 83 for i in BLOBCACHE.get_list_blobs(browse_dir):
paulo@133 84 if i.name.endswith(exts):
paulo@133 85 browse_contents.add(i.name)
paulo@133 86 BLOBCACHE.cache_public_url(i)
paulo@124 87 browse_fns = []
paulo@124 88 for i in thumb_fns:
paulo@124 89 i_basename = os.path.splitext(os.path.basename(i))[0]
paulo@124 90 try:
paulo@124 91 for j in exts:
paulo@124 92 browse_fn = browse_dir + "/" + i_basename + j
paulo@124 93 if browse_fn in browse_contents:
paulo@124 94 browse_fns.append(browse_fn)
paulo@124 95 raise StopIteration
paulo@124 96 except StopIteration:
paulo@124 97 pass
paulo@124 98 else:
paulo@124 99 raise RuntimeError(f"Cannot find browse image for {i}")
paulo@124 100
paulo@124 101 return zip(thumb_fns, browse_fns)
paulo@124 102
paulo@124 103
paulo@124 104 def _get_standard_html_doc(title):
paulo@124 105 root = HTML("html")
paulo@124 106
paulo@124 107 header = root.head
paulo@124 108 header.link(rel="stylesheet", type="text/css", href=flask.url_for("static", filename="index.css"))
paulo@124 109 header.title(title)
paulo@124 110
paulo@124 111 body = root.body
paulo@124 112 body.h1(title)
paulo@124 113
paulo@124 114 return (root, header, body)
paulo@124 115
paulo@124 116
paulo@124 117 def _go_thumbnail_links_to_browse_imgs_html_body(body, d, t, b, a_args={}, img_args={}, lazyload=False):
paulo@133 118 thumb_img_url = BLOBCACHE.get_public_url(t)
paulo@124 119 browse_url = flask.url_for("browse", d=d, img=os.path.basename(b))
paulo@124 120
paulo@124 121 a = body.a(href=browse_url, **a_args)
paulo@124 122 if lazyload:
paulo@124 123 img_args = dict(img_args)
paulo@124 124 img_args["data-src"] = thumb_img_url
paulo@124 125 a.img(**img_args)
paulo@124 126 else:
paulo@124 127 a.img(src=thumb_img_url, **img_args)
paulo@124 128
paulo@124 129 body.text(" ")
paulo@124 130
paulo@124 131
paulo@125 132 def _go_thumbnail_links_to_thumbs_html_body(body, d, t, a_args={}, img_args={}):
paulo@133 133 thumb_img_url = BLOBCACHE.get_public_url(t)
paulo@125 134 thumbs_url_args = {"from": os.path.basename(t), "_anchor": "selected"}
paulo@125 135 thumbs_url = flask.url_for("thumbs", d=d, **thumbs_url_args)
paulo@125 136
paulo@125 137 a = body.a(href=thumbs_url, **a_args)
paulo@125 138 a.img(src=thumb_img_url, **img_args)
paulo@125 139
paulo@125 140 body.text(" ")
paulo@125 141
paulo@125 142
paulo@124 143 @app.route("/")
paulo@124 144 def index():
paulo@124 145 n = 5 # number of thumbnails to display per dir
paulo@124 146
paulo@124 147 (root, header, body) = _get_standard_html_doc("Pictures")
paulo@124 148 header.script('', type="text/javascript", src=flask.url_for("static", filename="lazyload.js"))
paulo@124 149
paulo@130 150 if not PIN:
paulo@130 151 raise PinSetupError
paulo@130 152 elif flask.request.cookies.get("lahat") != PIN:
paulo@130 153 raise PinFailError
paulo@130 154
paulo@124 155 pics_dirs = []
paulo@124 156 pics_dirs_index_blob = GCS_BUCKET.get_blob("pics/index.tsv")
paulo@124 157 if pics_dirs_index_blob:
paulo@124 158 pics_dirs_index_strlist = pics_dirs_index_blob.download_as_text().splitlines()
paulo@124 159 pics_dirs_index_reader = csv.reader(pics_dirs_index_strlist, PICSDIALECT)
paulo@125 160 pics_dirs = sorted(pics_dirs_index_reader, key=lambda x: x[0], reverse=True)
paulo@124 161
paulo@124 162 for (dts, d) in pics_dirs:
paulo@124 163 dt = _parse_dt(dts)
paulo@125 164 body.h2.a(d, href=flask.url_for("thumbs", d=d))
paulo@124 165 body.h3(_format_dt(dt))
paulo@124 166
paulo@124 167 imgs = _get_images(d)
paulo@124 168 imgs_idx = [(i, img) for (i, img) in enumerate(imgs)]
paulo@124 169
paulo@124 170 sampled_imgs_idx = random.sample(imgs_idx, min(len(imgs_idx), n))
paulo@124 171 sampled_imgs_idx.sort(key=lambda x: x[0])
paulo@124 172
paulo@124 173 p = body.p
paulo@124 174 for (i, (t, b)) in sampled_imgs_idx:
paulo@124 175 _go_thumbnail_links_to_browse_imgs_html_body(p, d, t, b, lazyload=True)
paulo@124 176
paulo@124 177 return str(root).encode("utf-8")
paulo@124 178
paulo@124 179
paulo@124 180 @app.route("/<d>/browse/<img>")
paulo@124 181 def browse(d, img):
paulo@124 182 browse_img_blob = GCS_BUCKET.get_blob(f"pics/{d}/browse/{img}")
paulo@124 183 if not browse_img_blob:
paulo@124 184 flask.abort(404)
paulo@124 185
paulo@124 186 imgs = list(_get_images(d))
paulo@124 187
paulo@124 188 # thumbnail preview ribbon
paulo@124 189 w = 7 # must be odd
paulo@124 190 v = int(w/2)
paulo@124 191 imgs_circ = [None] * w
paulo@124 192 x = None
paulo@124 193 n = len(imgs)
paulo@124 194 for (i, (t, b)) in enumerate(imgs):
paulo@124 195 if os.path.basename(b) == img:
paulo@124 196 x = i + 1
paulo@124 197 imgs_circ[v] = (t, b)
paulo@124 198 for j in range(1, v + 1):
paulo@124 199 if (i + j) < n: imgs_circ[v + j] = imgs[i + j]
paulo@124 200 if (i - j) >= 0: imgs_circ[v - j] = imgs[i - j]
paulo@124 201 break
paulo@124 202
paulo@124 203 if x is None:
paulo@124 204 raise AssertionError
paulo@124 205
paulo@124 206 (root, header, body) = _get_standard_html_doc(f"{d} \u2014 {x} of {n}")
paulo@124 207 header.script('', type="text/javascript", src=flask.url_for("static", filename="np_keys.js"))
paulo@124 208
paulo@124 209 browse_img_url = browse_img_blob.public_url
paulo@124 210 ext = os.path.splitext(img)[1]
paulo@124 211 p = body.p
paulo@124 212 if ext == ".webm":
paulo@124 213 p.video(src=browse_img_url, autoplay="true", loop="true")
paulo@124 214 else:
paulo@124 215 p.img(src=browse_img_url)
paulo@124 216
paulo@124 217 p = body.p
paulo@124 218 for (i, img_c) in enumerate(imgs_circ):
paulo@124 219 if img_c is not None:
paulo@124 220 (t, b) = img_c
paulo@124 221 a_args = {}
paulo@124 222 img_args = {}
paulo@125 223 if os.path.basename(b) == img:
paulo@125 224 a_args = {"id": "up"}
paulo@125 225 img_args = {"klass": "sel"}
paulo@125 226 b = None
paulo@124 227 elif i == v + 1:
paulo@124 228 a_args = {"id": "next"}
paulo@124 229 elif i == v - 1:
paulo@124 230 a_args = {"id": "prev"}
paulo@124 231
paulo@125 232 if b is None:
paulo@125 233 _go_thumbnail_links_to_thumbs_html_body(p, d, t, a_args, img_args)
paulo@125 234 else:
paulo@125 235 _go_thumbnail_links_to_browse_imgs_html_body(p, d, t, b, a_args, img_args)
paulo@124 236
paulo@124 237 return str(root).encode("utf-8")
paulo@125 238
paulo@125 239
paulo@125 240 @app.route("/<d>")
paulo@125 241 def thumbs(d):
paulo@125 242 args = flask.request.args
paulo@125 243
paulo@125 244 from_img = args.get("from")
paulo@125 245
paulo@125 246 imgs = list(_get_images(d))
paulo@125 247 (root, header, body) = _get_standard_html_doc(d)
paulo@125 248
paulo@125 249 p = body.p
paulo@125 250 for (t, b) in imgs:
paulo@125 251 img_args = {}
paulo@125 252 if os.path.basename(b) == from_img:
paulo@125 253 img_args={"klass": "sel2", "id":"selected"}
paulo@125 254 _go_thumbnail_links_to_browse_imgs_html_body(p, d, t, b, img_args=img_args)
paulo@125 255
paulo@125 256 return str(root).encode("utf-8")