1 | 1 | from urllib.parse import unquote |
|
2 | 1 | ||
3 | |||
4 | 1 | import log |
|
5 | 1 | import requests |
|
0 ignored issues
–
show
introduced
by
![]() |
|||
6 | import background |
||
0 ignored issues
–
show
|
|||
7 | from flask import (Response, url_for, render_template, send_file, |
||
0 ignored issues
–
show
|
|||
8 | current_app, request) |
||
9 | 1 | ||
10 | |||
11 | def samples(blank=False): |
||
12 | 1 | """Generate dictionaries of sample image data for template rendering.""" |
|
13 | for template in sorted(current_app.template_service.all()): |
||
14 | 1 | path = "_" if blank else template.sample_path |
|
15 | 1 | url = route('image.get', key=template.key, path=path, |
|
16 | 1 | preview=True, watermark='none') |
|
17 | yield { |
||
18 | 1 | 'key': template.key, |
|
19 | 'name': template.name, |
||
20 | 'url': url, |
||
21 | 1 | } |
|
22 | |||
23 | 1 | ||
24 | 1 | def route(*args, **kwargs): |
|
25 | """Unquoted version of Flask's `url_for`.""" |
||
26 | for key, value in sorted(kwargs.items()): |
||
27 | if value is True: |
||
28 | kwargs[key] = 'true' |
||
29 | 1 | ||
30 | return _secure(unquote(url_for(*args, **kwargs))) |
||
31 | |||
32 | |||
33 | def display(title, path, share=False, raw=False, mimetype='image/jpeg'): |
||
34 | 1 | """Render a webpage or raw image based on request.""" |
|
35 | img = _format_url(request, 'share') |
||
36 | 1 | img_twitter = _format_url( |
|
37 | 1 | request, 'share', |
|
38 | width=current_app.config['TWITTER_IMAGE_WIDTH'], |
||
39 | 1 | height=current_app.config['TWITTER_IMAGE_HEIGHT'], |
|
40 | ) |
||
41 | img_facebook = _format_url( |
||
42 | request, 'share', |
||
43 | width=current_app.config['FACEBOOK_IMAGE_WIDTH'], |
||
44 | height=current_app.config['FACEBOOK_IMAGE_HEIGHT'], |
||
45 | ) |
||
46 | img_copy = _format_url(request, 'share', 'watermark') |
||
47 | url = _format_url(request, 'width', 'height') |
||
48 | 1 | ||
49 | if share: |
||
50 | log.info("Sharing image on page: %s", img) |
||
51 | 1 | ||
52 | 1 | html = render_template( |
|
53 | 'share.html', |
||
54 | title=title, |
||
55 | 1 | img=_secure(img), |
|
56 | img_twitter=_secure(img_twitter), |
||
57 | 1 | img_facebook=_secure(img_facebook), |
|
58 | 1 | img_copy=_secure(img_copy), |
|
59 | 1 | url=_secure(url), |
|
60 | config=current_app.config, |
||
61 | ) |
||
62 | return html if raw else Response(html) |
||
63 | |||
64 | else: |
||
65 | log.info("Sending image: %s", path) |
||
66 | return send_file(path, mimetype=mimetype) |
||
67 | |||
68 | |||
69 | def track(title): |
||
70 | """Log the requested content, server-side.""" |
||
71 | google_url = current_app.config['GOOGLE_ANALYTICS_URL'] |
||
72 | 1 | google_tid = current_app.config['GOOGLE_ANALYTICS_TID'] |
|
73 | google_data = dict( |
||
74 | v=1, |
||
75 | tid=google_tid, |
||
76 | cid=request.remote_addr, |
||
77 | 1 | ||
78 | 1 | t='pageview', |
|
79 | dh='memegen.link', |
||
80 | dp=request.full_path, |
||
81 | dt=str(title), |
||
82 | |||
83 | 1 | uip=request.remote_addr, |
|
84 | ua=request.user_agent.string, |
||
85 | ) |
||
86 | |||
87 | remote_url = current_app.config['REMOTE_TRACKING_URL'] |
||
88 | 1 | remote_data = dict( |
|
89 | text=str(title), |
||
90 | 1 | source='memegen.link', |
|
91 | context=unquote(request.url), |
||
92 | ) |
||
93 | 1 | ||
94 | @background.task |
||
95 | def run(): |
||
96 | 1 | if google_tid != 'localhost': |
|
97 | response = requests.post(google_url, data=google_data) |
||
98 | 1 | params = _format_query(google_data, as_string=True) |
|
99 | log.debug("Tracking POST: %s %s", response.url, params) |
||
100 | 1 | ||
101 | 1 | if remote_url: |
|
102 | response = requests.get(remote_url, params=remote_data) |
||
103 | 1 | log.debug("Tracking GET: %s", response.url) |
|
104 | run() |
||
105 | 1 | ||
106 | 1 | ||
107 | def _secure(url): |
||
108 | 1 | """Ensure HTTPS is used in production.""" |
|
109 | if current_app.config['ENV'] == 'prod': |
||
110 | if url.startswith('http:'): |
||
111 | 1 | url = url.replace('http:', 'https:', 1) |
|
112 | 1 | return url |
|
113 | 1 | ||
114 | 1 | ||
115 | def _format_url(req, *skip, **add): |
||
116 | 1 | """Get a formatted URL with sanitized query parameters.""" |
|
117 | base = req.base_url |
||
118 | |||
119 | 1 | options = {} |
|
120 | for key, value in dict(req.args).items(): |
||
121 | if key in skip: |
||
122 | continue |
||
123 | if isinstance(value, str): |
||
124 | options[key] = value |
||
125 | else: |
||
126 | options[key] = value[0] |
||
127 | |||
128 | options.update(add) |
||
129 | |||
130 | params = _format_query(options) |
||
131 | |||
132 | if params: |
||
133 | return base + "?{}".format("&".join(params)) |
||
134 | else: |
||
135 | return base |
||
136 | |||
137 | |||
138 | def _format_query(options, *, as_string=False): |
||
139 | pattern = "{}={!r}" if as_string else "{}={}" |
||
140 | pairs = sorted(pattern.format(k, v) for k, v in options.items()) |
||
141 | if as_string: |
||
142 | return ' '.join(pairs) |
||
143 | return pairs |
||
144 |