Spaces:
Running
Running
File size: 7,306 Bytes
185c377 |
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 |
import mysql.connector
from mysql.connector import Error
import json
from datetime import datetime
import logging
from src.config import Config
class DatabaseManager:
def __init__(self, config_dict=None):
self.config = config_dict or Config().get_mysql_config()
def get_connection(self):
try:
connection = mysql.connector.connect(**self.config)
return connection
except Error as e:
logging.error(f"Error connecting to MySQL: {e}")
return None
def execute_query(self, query, params=None, fetch=False):
connection = self.get_connection()
if not connection:
return None
try:
cursor = connection.cursor(dictionary=True)
cursor.execute(query, params or ())
if fetch:
return cursor.fetchall()
else:
connection.commit()
return cursor.rowcount
except Error as e:
logging.error(f"Error executing query: {e}")
connection.rollback()
return None
finally:
if connection and connection.is_connected():
cursor.close()
connection.close()
def execute_query_one(self, query, params=None):
connection = self.get_connection()
if not connection:
return None
try:
cursor = connection.cursor(dictionary=True)
cursor.execute(query, params or ())
return cursor.fetchone()
except Error as e:
logging.error(f"Error executing query: {e}")
return None
finally:
if connection and connection.is_connected():
cursor.close()
connection.close()
def save_questionnaire(self, user_id, questionnaire_data):
"""Save questionnaire data to DB and return the inserted ID"""
try:
connection = self.get_connection()
if not connection:
return None
cursor = connection.cursor()
query = """
INSERT INTO questionnaires (
user_id, patient_name, patient_age, patient_gender,
wound_location, wound_duration, pain_level,
previous_treatment, medical_history, medications,
allergies, additional_notes, created_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
params = (
user_id,
questionnaire_data.get("patient_name"),
questionnaire_data.get("patient_age"),
questionnaire_data.get("patient_gender"),
questionnaire_data.get("wound_location"),
questionnaire_data.get("wound_duration"),
questionnaire_data.get("pain_level"),
questionnaire_data.get("previous_treatment"),
questionnaire_data.get("medical_history"),
questionnaire_data.get("medications"),
questionnaire_data.get("allergies"),
questionnaire_data.get("additional_notes"),
datetime.now()
)
cursor.execute(query, params)
connection.commit()
return cursor.lastrowid
except Error as e:
logging.error(f"Error saving questionnaire: {e}")
connection.rollback()
return None
finally:
if connection and connection.is_connected():
cursor.close()
connection.close()
def save_analysis_session(self, user_id, questionnaire_data, image_url, ai_analysis):
try:
connection = self.get_connection()
if not connection:
return None
cursor = connection.cursor()
# Save questionnaire
questionnaire_id = self.save_questionnaire(user_id, questionnaire_data)
# Save image
image_id = None
if image_url:
image_query = """
INSERT INTO wound_images (questionnaire_id, image_url, created_at)
VALUES (%s, %s, %s)
"""
cursor.execute(image_query, (questionnaire_id, image_url, datetime.now()))
image_id = cursor.lastrowid
# Save AI analysis
analysis_query = """
INSERT INTO ai_analyses (questionnaire_id, image_id, analysis_data,
summary, recommendations, created_at)
VALUES (%s, %s, %s, %s, %s, %s)
"""
cursor.execute(analysis_query, (
questionnaire_id,
image_id,
json.dumps(ai_analysis),
ai_analysis.get("summary", ""),
ai_analysis.get("recommendations", ""),
datetime.now()
))
analysis_id = cursor.lastrowid
# Save session
session_query = """
INSERT INTO analysis_sessions (user_id, questionnaire_id, image_id,
analysis_id, created_at)
VALUES (%s, %s, %s, %s, %s)
"""
cursor.execute(session_query, (
user_id, questionnaire_id, image_id, analysis_id, datetime.now()
))
connection.commit()
return cursor.lastrowid
except Error as e:
logging.error(f"Error saving analysis session: {e}")
connection.rollback()
return None
finally:
if connection and connection.is_connected():
cursor.close()
connection.close()
def get_user_analysis_history(self, user_id, limit=10):
query = """
SELECT
s.id as session_id,
s.created_at,
q.patient_name,
q.wound_location,
a.summary as ai_summary,
a.recommendations,
wi.image_url
FROM analysis_sessions s
JOIN questionnaires q ON s.questionnaire_id = q.id
LEFT JOIN ai_analyses a ON s.analysis_id = a.id
LEFT JOIN wound_images wi ON s.image_id = wi.id
WHERE s.user_id = %s
ORDER BY s.created_at DESC
LIMIT %s
"""
return self.execute_query(query, (user_id, limit), fetch=True)
def get_organization_stats(self, organization_id):
query = """
SELECT
COUNT(DISTINCT u.id) as total_practitioners,
COUNT(DISTINCT s.id) as total_analyses,
COUNT(DISTINCT q.patient_name) as unique_patients
FROM users u
LEFT JOIN analysis_sessions s ON u.id = s.user_id
LEFT JOIN questionnaires q ON s.questionnaire_id = q.id
WHERE u.org = %s AND u.role = 'practitioner'
"""
return self.execute_query_one(query, (organization_id,))
|