-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
75 lines (60 loc) · 2.7 KB
/
app.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
75
from io import BytesIO
import base64
from flask import Flask, request, jsonify
from flask_cors import CORS
from PIL import Image
from environs import Env
from vision import GCPVisionAPI
from storage import GCPStorageAPI
from signed_url import generate_signed_url
from visual import draw, write_to_buffer, read_from_btyes
from resize import get_resized_byte_string
from recaptcha import verify
env = Env()
env.read_env()
key_file = env('GCP_SERVICE_ACCOUNT_CREDENTIALS_PATH')
recaptcha_action_name = env('RECAPTCHA_ACTION_NAME')
gcp_bucket_name = env('GCP_BUCKET_NAME')
recaptcha_pass_threshold = env.float('RECAPTCHA_PASS_THRESHOLD')
app = Flask(__name__)
CORS(app)
vision_api = GCPVisionAPI(key_file)
storage_api = GCPStorageAPI(gcp_bucket_name, key_file)
def annotate_and_upload(image_byte_string, criteria, vision_api, storage_api):
response = vision_api.get_text_annotations(image_byte_string)
paragraphs = vision_api.parse_response_to_paragraphs(response)
image = read_from_btyes(image_byte_string)
annotated_image = draw(image, paragraphs, criteria)
buffer = write_to_buffer(annotated_image)
blob_name, uploaded_link = storage_api.upload(buffer, 'image/png')
return blob_name, uploaded_link
def process_request(content):
base64_encoded_image = content['image']
criteria = content['criteria']
byte_string = base64.b64decode(base64_encoded_image)
buffer = BytesIO(byte_string)
image = Image.open(buffer)
resized_byte_string = get_resized_byte_string(image)
blob_name, uploaded_link = annotate_and_upload(resized_byte_string, criteria, vision_api, storage_api)
signed_url = generate_signed_url(service_account_file=key_file,
bucket_name=storage_api.bucket_name,
object_name=blob_name, subresource=None, expiration=3600,
http_method='GET',
query_parameters=None, headers=None)
return {"annotatedImage": signed_url}
@app.route('/process', methods=['POST'])
def process_image():
content = request.get_json()
recaptcha_token = content.get("token")
if recaptcha_token:
assessment = verify(recaptcha_token)
print("assessment results: {}".format(assessment))
if 'error' not in assessment and assessment.get('tokenProperties').get("action") == recaptcha_action_name and \
assessment.get('score') >= recaptcha_pass_threshold:
response = jsonify(process_request(content))
return response
response = jsonify({"error": "recaptcha assessment failed"})
response.status_code = 403
return response
if __name__ == '__main__':
app.run(port=8080, host="0.0.0.0")