-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
185 lines (149 loc) · 6.67 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
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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
import json
import os
import random
import time
from datetime import datetime
from flask import Flask, render_template
from flask_login import LoginManager
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy
from flask_apscheduler import APScheduler
from configuration.config import Config as app_config
from utils import NFT_collections, core_url_NFT, collection_to_min_max_price
db = SQLAlchemy()
def create_app():
# ===== Flask app
app = Flask(__name__, template_folder='templates', static_folder='assets')
app.config.from_object(app_config)
# Initialisation de l'APScheduler
scheduler = APScheduler()
scheduler.init_app(app)
# ===== Blueprint
from auth.auth import BLP_auth
from general.general import BLP_general
from API.api import BLP_api
app.register_blueprint(BLP_auth)
app.register_blueprint(BLP_general)
app.register_blueprint(BLP_api)
# ===== Login manager
login_manager = LoginManager()
login_manager.login_view = 'BLP_auth.login'
login_manager.init_app(app)
with app.app_context():
db.init_app(app)
if not os.path.exists("instance/db.sqlite"):
# ===== init SQLAlchemy
db.create_all()
time.sleep(1)
# ===== Init NFT marketplace
# Go trhough all NFTs and add them to the database
# Add a random price between 0.1 and 100 in crypto among :
# [Litecoin, Cardano, XRP, Dogecoin, Qtum, Basic Attention Token, NEO]
# Lower price is more probable than higher price
# Add a name that is the same as the collection and the index with #
# add the collection name
# add the image path
from models import NFT, NFTPriceOwnerHistory
print("[INFO] Adding NFTs to the database")
for collection in NFT_collections:
collection_path = core_url_NFT + collection.lower() + '/'
# Add as many NFTs as there is in the folder with the same name and _index (starting at 1)
# get the number of file in collection_path (just img files that end with .png or .jpg)
files_ = os.listdir('assets' + collection_path)
nb_files = len([f for f in files_ if f.endswith('.png') or f.endswith('.jpg')])
for i in range(1, nb_files + 1):
# Init the NFT
name = f"{collection} #{i}"
path = f"{collection_path}{collection.lower()}_{i}.png"
price = round(random.uniform(collection_to_min_max_price[collection][0],
collection_to_min_max_price[collection][1]), 3)
nft = NFT(name=name, collection=collection, image_path=path, price=price, owner_id=None)
db.session.add(nft)
db.session.flush()
db.session.commit()
# ===== Init Mining server
from models import MiningServer
path_to_mining_server_config = 'configuration/mining_servers.json'
# Iterate over all the json in the document, and add the mining server to the database
with open(path_to_mining_server_config) as json_file:
data = json.load(json_file)
for mining_server in data:
mining_server = MiningServer(
name=mining_server['Name'],
symbol=mining_server['Symbol'],
buy_amount=mining_server['BuyAmount'],
power=mining_server['Power'],
logo_path=mining_server['Logo'],
category=mining_server['Category']
)
db.session.add(mining_server)
db.session.commit()
# ===== Init the first user (admin)
from init_fonctions import init_admin
init_admin()
from models import User
@login_manager.user_loader
def load_user(user_id):
# since the user_id is just the primary key of our user table, use it in the query for the user
return User.query.get(int(user_id))
# ===== error page
@app.errorhandler(404)
def forbidden(error):
return render_template('errors/error_404.html')
@app.errorhandler(500)
def forbidden(error):
return render_template('errors/error_500.html')
# ===== Scheduler
from crypto_manager import CryptoDataManager
from mining_server_manager import Mining_server_manager
from nft_manager import NFT_manager
from wallet_manager import wallet_manager
def schedule_update():
# Update crypto data
manager = CryptoDataManager()
manager.update_crypto_data()
# Update NFT prices if needed
nft_manager = NFT_manager()
nft_manager.update_NFT_price()
# Update trading data (positions, etc)
users = User.query.all()
for user in users:
wallet_manager().update_positions(user.id)
def server_payment_process():
users = User.query.all()
mining_server_manager = Mining_server_manager()
for user in users:
mining_server_manager.check_for_server_payment(user.id)
# Start the first update
with app.app_context():
schedule_update()
# Clean all cron tasks before adding them to avoid duplicates
scheduler.remove_all_jobs()
def cron_debug_crypto_update():
with app.app_context():
schedule_update()
server_payment_process()
def cron_crypto_update():
with app.app_context():
schedule_update()
def cron_server_payment():
with app.app_context():
server_payment_process()
frequency_crypto_update = os.getenv('FREQUENCY_UPDATE_CRYPTO_MINUTES')
# Add the task to the scheduler based on the DEBUG_MODE environment variable
if os.getenv('DEBUG_MODE') == 'True':
print("Debug mode is on, updating every hour")
scheduler.add_job(func=cron_debug_crypto_update, trigger='cron',
hour='*', id='debug_crypto_update')
else:
print(f"Regular mode, updating every {frequency_crypto_update} minutes and payment process at 1 AM")
scheduler.add_job(func=cron_crypto_update, trigger='cron',
minute=f'*/{frequency_crypto_update}', id='crypto_update')
scheduler.add_job(func=cron_server_payment, trigger='cron',
hour='1', minute='0', id='server_payment')
# Check if scheduler is not already running before starting
if not scheduler.running:
scheduler.start()
return app
app = create_app()
migrate = Migrate(app, db)