This repository has been archived by the owner on Jul 27, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 220
/
flask_app.py
134 lines (109 loc) · 4.06 KB
/
flask_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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
#!/usr/bin/env python3
import logging
import os
import re
from flask import Flask, make_response, jsonify
from flask import render_template
from flask_socketio import SocketIO, emit
from playstoredownloader.downloader.out_dir import OutDir
from playstoredownloader.playstore.meta import PackageMeta
from playstoredownloader.playstore.playstore import Playstore
if "LOG_LEVEL" in os.environ:
log_level = os.environ["LOG_LEVEL"]
else:
log_level = logging.INFO
# Logging configuration.
logger = logging.getLogger(__name__)
logging.basicConfig(
format="%(asctime)s> [%(levelname)s][%(name)s][%(funcName)s()] %(message)s",
datefmt="%d/%m/%Y %H:%M:%S",
level=log_level,
)
logging.getLogger("werkzeug").disabled = True
# Credentials file location (make sure to use a valid json file with the credentials).
credentials_location = os.path.join(
os.path.dirname(os.path.realpath(__file__)), "private_credentials.json"
)
if not os.path.isfile(credentials_location):
# Use the default credentials file if no private credentials are present.
credentials_location = os.path.join(
os.path.dirname(os.path.realpath(__file__)), "credentials.json"
)
# Directory where to save the downloaded applications.
downloaded_apk_location = os.path.join(
os.path.dirname(os.path.realpath(__file__)), "Downloads"
)
# https://developer.android.com/guide/topics/manifest/manifest-element#package
package_name_regex = re.compile(
r"^[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)+$", flags=re.IGNORECASE
)
def create_app():
app = Flask(__name__)
# Create the download directory (if not already existing).
if not os.path.isdir(downloaded_apk_location):
os.makedirs(downloaded_apk_location)
return app
application = create_app()
socket = SocketIO(application, ping_timeout=600)
@application.after_request
def add_cache_header(response):
response.headers[
"Cache-Control"
] = "public, max-age=0, no-cache, no-store, must-revalidate"
response.headers["Pragma"] = "no-cache"
response.headers["Expires"] = "0"
return response
@application.errorhandler(400)
@application.errorhandler(500)
def application_error(error):
logger.error(error)
return make_response(jsonify(str(error)), error.code)
@application.route("/", methods=["GET"], strict_slashes=False)
def home():
return render_template("index.html")
@socket.on("start_download")
def on_start_download(package_name):
if package_name_regex.match(package_name):
try:
api = Playstore(credentials_location)
meta = PackageMeta(api, package_name)
try:
app = meta.app_details().docV2
except AttributeError:
emit(
"download_bad_package",
f"Unable to retrieve application with "
f"package name '{package_name}'",
)
return
details = {
"package_name": app.docid,
"title": app.title,
"creator": app.creator,
}
downloaded_apk_file_path = os.path.join(
downloaded_apk_location,
re.sub(
r"[^\w\-_.\s]",
"_",
f"{details['title']} by {details['creator']} - "
f"{details['package_name']}.apk",
),
)
# noinspection PyProtectedMember
for progress in api._download_with_progress(
meta,
OutDir(downloaded_apk_file_path, meta=meta),
):
emit("download_progress", progress)
logger.info(
f"The application was downloaded and "
f"saved to '{downloaded_apk_file_path}'"
)
emit("download_success", "The application was successfully downloaded")
except Exception as e:
emit("download_error", str(e))
else:
emit("download_error", "Please specify a valid package name")
if __name__ == "__main__":
socket.run(application, host="0.0.0.0", port=5000)