-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapp.py
More file actions
197 lines (153 loc) · 6.2 KB
/
app.py
File metadata and controls
197 lines (153 loc) · 6.2 KB
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
186
187
188
189
190
191
192
193
194
195
196
197
"""
Flask Application for Malaria Detection & Tracking System.
This module sets up and configures the Flask application,
including database connections, user authentication,
and route handling for the Malaria detection project.
"""
from datetime import datetime
from flask import Flask, current_app
from flask_login import LoginManager
from flask_bcrypt import Bcrypt
from flask_migrate import Migrate
from flask_mail import Mail
from models import db, Users, User_Report, AuditLog
from config import Config
from routes.auth_routes import auth_bp
from routes.detection_routes import detection_bp
from routes.admin_routes import admin_bp
from routes.general_routes import general_bp
from routes.user_routes import user_bp
from routes.donation_routes import donation_bp, calculate_allocations
from routes.openai_routes import openai_bp
from routes.search_paper_routes import search_paper_bp
from flask_apscheduler import APScheduler
from routes.wise_routes import create_wise_transfer_quote, BUSINESS_PROFILE_ID, execute_wise_transfer
# Initialize extensions
bcrypt = Bcrypt()
login_manager = LoginManager()
migrate = Migrate()
mail = Mail()
def configure_login_manager(login_manager):
"""Configures the Flask-Login manager."""
login_manager.session_protection = "strong"
login_manager.login_view = "auth.login"
login_manager.login_message_category = "info"
@login_manager.user_loader
def load_user(user_id):
"""Loads a user from the database using their ID."""
return Users.query.get(int(user_id))
def init_extensions(app):
"""Initializes Flask extensions."""
db.init_app(app)
migrate.init_app(app, db)
bcrypt.init_app(app)
login_manager.init_app(app)
mail.init_app(app)
def register_blueprints(app):
"""Registers the Flask blueprints."""
app.register_blueprint(auth_bp)
app.register_blueprint(detection_bp)
app.register_blueprint(admin_bp)
app.register_blueprint(general_bp)
app.register_blueprint(user_bp)
app.register_blueprint(donation_bp)
app.register_blueprint(openai_bp, url_prefix="/openai")
app.register_blueprint(search_paper_bp)
from routes.wise_routes import wise_bp
app.register_blueprint(wise_bp)
def create_medical_resources_list(app):
"""Creates a list of medical resources from the database."""
medical_resources = []
with app.app_context():
users = Users.query.filter_by(is_resource=True).all()
for index, user in enumerate(users):
medical_resources.append(
{
user.username: [
index,
user.currency_needed,
user.bank_account,
user.first_name,
user.last_name,
]
}
)
return medical_resources
def inject_user_reports(app):
"""Makes user reports available to html templates"""
@app.context_processor
def inject_user_reports():
user_reports = User_Report.query.all()
return dict(user_reports=user_reports)
def create_app():
"""Creates and configures the Flask application instance."""
app = Flask(__name__)
# Use the updated Config class with Supabase connection string
app.config.from_object(Config)
# Initialize Flask extensions
init_extensions(app)
configure_login_manager(login_manager)
register_blueprints(app)
inject_user_reports(app)
# Ensure tables are created in your Supabase instance
with app.app_context():
db.create_all()
# Initialize and add medical resources to app.config
medical_resources = create_medical_resources_list(app)
app.config['MEDICAL_RESOURCES'] = medical_resources
# Scheduler for automatic distributions
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()
# Register the job here
scheduler.add_job(
id='daily_donation_distribution',
func=automatic_donation_distribution,
trigger='interval',
hours=24 # change to seconds=60 for testing
)
return app
def automatic_donation_distribution():
with app.app_context():
data = calculate_allocations()
total_incidence = data['IncidenceRate'].sum()
donation_pool = current_app.config.get('DAILY_DONATION_POOL', 1000)
users = Users.query.filter(Users.wise_recipient_id.isnot(None)).all()
users_map = {u.country: u for u in users}
for _, row in data.iterrows():
country = row['name']
incidence_rate = row['IncidenceRate']
allocation_amount = round((incidence_rate / total_incidence) * donation_pool, 2)
user = users_map.get(country)
if not user:
current_app.logger.warning(f"No recipient found for {country}, skipping.")
continue
quote = create_wise_transfer_quote({
"profile_id": BUSINESS_PROFILE_ID,
"source_currency": "GBP",
"target_currency": user.currency_needed,
"amount": allocation_amount
})
if quote and "id" in quote:
transfer = execute_wise_transfer({
"recipient_id": user.wise_recipient_id,
"quote_id": quote["id"]
})
if "id" in transfer and transfer["status"] == "incoming_payment_waiting":
AuditLog.log_action(
user_id=None, # system user
action="Wise Transfer (Automated)",
amount=allocation_amount,
details=f"Transfer ID: {transfer['id']} | Recipient: {user.wise_recipient_id} | Country: {country} | Status: {transfer['status']} | Date: {datetime.utcnow()}"
)
print("Automated Distribution Completed.")
if __name__ == "__main__":
app = create_app()
@app.template_filter('currency')
def currency_filter(amount):
# Convert from cents to dollars and format as $100.00
return "${:,.2f}".format(amount / 100)
@app.template_filter('datetimeformat')
def datetimeformat(value, format='%Y-%m-%d %H:%M:%S'):
return datetime.utcfromtimestamp(value).strftime(format)
app.run(debug=True)