Mercurial > hg > index.fcgi > www > www-1
view laterlinks3/laterlinks_flask_app.py @ 121:26cc1a16e7a3
laterlinks3: fix gcs_upload() function
| author | paulo |
|---|---|
| date | Sat, 16 Jan 2021 10:14:02 +0000 |
| parents | 65db090a697e |
| children |
line source
1 import csv
2 import datetime
3 import io
4 import os
5 import tempfile
7 import flask
8 import google.cloud.storage
9 from html3.html3 import HTML
11 app = flask.Flask(__name__)
13 GCS_BUCKET = google.cloud.storage.Client().get_bucket(os.environ.get("GCS_BUCKET"))
14 PIN = os.environ.get("LLPIN")
15 STRTIME_FMT = "%Y-%m-%d %H:%M:%S"
18 class LLDialect(csv.Dialect):
19 delimiter = '\t'
20 quoting = csv.QUOTE_NONE
21 lineterminator = '\n'
24 LLDIALECT = LLDialect()
25 LLDB_FN = "lldb.tsv"
26 LLDB_UNREAD_FN = "lldb_unread.tsv"
29 class LLError(Exception):
30 pass
32 class PinFailError(LLError):
33 def __str__(self):
34 return "PIN FAIL!"
36 class PinSetupError(LLError):
37 def __str__(self):
38 return "PIN SETUP ERROR!"
40 class MissingFieldsError(LLError):
41 def __str__(self):
42 return "MISSING FIELD(s)!"
45 def gcs_download(fn):
46 tmp_f = tempfile.TemporaryFile("w+")
47 blob = GCS_BUCKET.get_blob(fn)
48 if blob:
49 tmp_f.write(str(blob.download_as_string(), encoding="utf-8"))
50 return tmp_f
53 def gcs_upload(fn, tmp_f):
54 blob = GCS_BUCKET.blob(fn)
55 tmp_f.seek(0)
56 blob.upload_from_string(tmp_f.read())
59 def lldb_unread_load(lldb_unread_tmp_f):
60 lldb_unread_tmp_f.seek(0)
61 return csv.reader(lldb_unread_tmp_f, LLDIALECT)
64 def lldb_add(inp, lldb_tmp_f, lldb_unread_tmp_f):
65 title = inp.get("title")
66 url = inp.get("url")
67 if not (title and url):
68 raise MissingFieldsError()
70 dt_str = datetime.datetime.now().strftime(STRTIME_FMT)
72 lldb_tmp_f.seek(0, io.SEEK_END)
73 csv.writer(lldb_tmp_f, LLDIALECT).writerow([title, url, dt_str])
74 gcs_upload(LLDB_FN, lldb_tmp_f)
76 lldb_unread_tmp_f.seek(0, io.SEEK_END)
77 csv.writer(lldb_unread_tmp_f, LLDIALECT).writerow([title, url, dt_str])
78 gcs_upload(LLDB_UNREAD_FN, lldb_unread_tmp_f)
81 def lldb_unread_delete(inp, lldb_unread_tmp_f):
82 delete = inp.getlist("delete")
83 if not delete:
84 raise MissingFieldsError()
86 lldb_unread = [i for i in lldb_unread_load(lldb_unread_tmp_f)]
88 try:
89 for i in delete:
90 for j in lldb_unread:
91 dt_str = j[2]
92 if i == dt_str:
93 lldb_unread.remove(j)
94 finally:
95 lldb_unread_tmp_f.truncate(0)
96 lldb_unread_tmp_f.seek(0)
97 csv.writer(lldb_unread_tmp_f, LLDIALECT).writerows(lldb_unread)
98 gcs_upload(LLDB_UNREAD_FN, lldb_unread_tmp_f)
101 @app.route("/", methods=["GET", "POST"])
102 def index():
103 is_post = (flask.request.method == "POST")
104 inp = flask.request.form
105 cookies = flask.request.cookies
107 with gcs_download(LLDB_UNREAD_FN) as lldb_unread_tmp_f:
108 if is_post:
109 if not PIN:
110 raise PinSetupError
111 elif cookies.get("llpin") != PIN:
112 raise PinFailError
114 if inp["submit"] == "Add":
115 with gcs_download(LLDB_FN) as lldb_tmp_f:
116 lldb_add(inp, lldb_tmp_f, lldb_unread_tmp_f)
117 elif inp["submit"] == "Delete":
118 lldb_unread_delete(inp, lldb_unread_tmp_f)
120 title = "later links..."
121 root = HTML("html")
123 header = root.head
124 header.link(rel="stylesheet", type="text/css", href=flask.url_for("static", filename="index.css"))
125 header.title(title)
127 body = root.body
128 body.h1(title)
130 form = body.form(action="/", method="post")
132 table = form.table
133 hrow = table.tr
134 hrow.th("Link")
135 hrow.th("Created")
136 hrow.th.input(type="submit", name="submit", value="Delete")
138 for (title, url, dt_str) in lldb_unread_load(lldb_unread_tmp_f):
139 row = table.tr
140 row.td.a(title, href=url)
141 row.td(dt_str)
142 row.td.input(type="checkbox", name="delete", value=dt_str)
144 p1 = form.p
145 p1.label("Title").input(type="text", name="title", size="64")
146 p1.br
147 p1.label("URL").input(type="text", name="url", size="64")
148 p1.br
149 p1.input(type="submit", name="submit", value="Add")
151 return str(root).encode("utf-8")