-
Notifications
You must be signed in to change notification settings - Fork 0
/
image_searcher.py
74 lines (59 loc) · 2.86 KB
/
image_searcher.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
from http.server import *
from http import HTTPStatus
import shutil
import os
import urllib
import psycopg2
import json
SERVER_PORT=55000
SERVER_LISTEN_ADDRESS="127.0.0.1"
DATABASE_NAME="imagedb"
db_connection=psycopg2.connect(dbname=DATABASE_NAME,user="imageuser",password="imageuser")
db_connection.autocommit=True
class WebServer(BaseHTTPRequestHandler):
def __init__(self,request, client_address, server):
super().__init__(request,client_address,server)
def fetch_images(self,sql_query):
try:
with db_connection.cursor() as cur:
cur.execute(f"select distinct on (path) id,url,path,name,tags,(select name from authors where authors.author_id=images.author_id) as author_name from images,unnest(case when tags <> '{{}}' then tags else '{{null}}' end) as tag where {sql_query}")
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type","application/json; charset=UTF-8")
self.send_header("Access-Control-Allow-Origin","*")
self.end_headers()
#カラム名との辞書を作る
name_data_map=[{d.name:f for d,f in zip(cur.description,row)} for row in cur.fetchall()]
self.wfile.write(json.dumps(name_data_map,default=lambda date:date.isoformat()).encode())
except psycopg2.ProgrammingError as e:
self.send_response(HTTPStatus.BAD_REQUEST)
self.send_header("Access-Control-Allow-Origin","*")
self.end_headers()
self.wfile.write(json.dumps(str(e)).encode())
def add_tag(self):
taginfo=json.loads(self.rfile.read(int(self.headers["Content-Length"])))
with db_connection.cursor() as cur:
cur.execute("update images set tags=array_append(tags,%s) where id=%s",(taginfo["tag"],taginfo["id"]) )
self.send_response(HTTPStatus.OK)
self.send_header("Access-Control-Allow-Origin","*")
self.end_headers()
def do_GET(self):
parsed_url=urllib.parse.urlparse(self.path)
path=urllib.parse.unquote_plus(parsed_url.path)
query_dict=urllib.parse.parse_qs(parsed_url.query)
if path=="/images":
self.fetch_images(query_dict["sql"][0])
return
self.send_error(HTTPStatus.BAD_REQUEST)
self.end_headers()
def do_POST(self):
parsed_url=urllib.parse.urlparse(self.path)
path=urllib.parse.unquote_plus(parsed_url.path)
if path=="/add_tag":
self.add_tag()
return
self.send_error(HTTPStatus.BAD_REQUEST)
self.end_headers()
def run(server_class=HTTPServer, handler_class=WebServer):
httpd = server_class((SERVER_LISTEN_ADDRESS,SERVER_PORT), handler_class)
httpd.serve_forever()
run(ThreadingHTTPServer)