from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user from geopy.distance import geodesic from wtforms.validators import DataRequired, Email, Length, EqualTo from mysql.connector import Error from datetime import date, datetime from flask_login import UserMixin from flask_bcrypt import Bcrypt from flask_wtf import FlaskForm from wtforms import StringField, PasswordField, SelectField, BooleanField, validators import mysql.connector from geopy.distance import geodesic bcrypt = Bcrypt() def get_db_connection(): return mysql.connector.connect( host="localhost", user="root", password="_z7b6CJc", database="immo_new" ) # Modèle utilisateur (simplifié) class User(UserMixin): def __init__(self, user_id, email, password_hash, account_type, **kwargs): # Champs obligatoires self.id = user_id self.email = email self.password_hash = password_hash self.account_type = account_type # Liste des propriétés protégées (celles de UserMixin) protected_properties = ['is_active', 'is_authenticated', 'is_anonymous', 'get_id'] # Assigne dynamiquement les autres attributs (sauf les propriétés protégées) for key, value in kwargs.items(): if key not in protected_properties: setattr(self, key, value) # Gère explicitement is_active si nécessaire if 'is_active' in kwargs: self._is_active = kwargs['is_active'] @property def is_active(self): # Retourne la valeur de _is_active si elle existe, sinon True par défaut return getattr(self, '_is_active', True) class RegistrationForm(FlaskForm): email = StringField( 'Email', validators=[ DataRequired(message="L'email est obligatoire."), Email(message="Format d'email invalide.") ], render_kw={"placeholder": "Votre email"} ) password = PasswordField( 'Mot de passe', validators=[ DataRequired(message="Le mot de passe est obligatoire."), Length(min=8, message="Le mot de passe doit faire au moins 8 caractères.") ], render_kw={"placeholder": "Votre mot de passe"} ) confirm_password = PasswordField( 'Confirmer le mot de passe', validators=[ DataRequired(message="Veuillez confirmer votre mot de passe."), EqualTo('password', message="Les mots de passe doivent correspondre.") ], render_kw={"placeholder": "Confirmez votre mot de passe"} ) account_type = SelectField( 'Type de compte', choices=[ ('particulier', 'Particulier'), ('agent_immobilier', 'Agent immobilier'), ('investisseur', 'Investisseur'), ('autre', 'Autre') ], validators=[DataRequired(message="Veuillez sélectionner un type de compte.")] ) first_name = StringField( 'Prénom', validators=[DataRequired(message="Le prénom est obligatoire.")], render_kw={"placeholder": "Votre prénom"} ) last_name = StringField( 'Nom', validators=[DataRequired(message="Le nom est obligatoire.")], render_kw={"placeholder": "Votre nom"} ) newsletter_subscription = BooleanField('S\'abonner à la newsletter') class LoginForm(FlaskForm): email = StringField( 'Email', validators=[ DataRequired(message="L'email est obligatoire."), Email(message="Format d'email invalide.") ], render_kw={"placeholder": "Votre email"} ) password = PasswordField( 'Mot de passe', validators=[DataRequired(message="Le mot de passe est obligatoire.")], render_kw={"placeholder": "Votre mot de passe"} ) remember_me = BooleanField('Se souvenir de moi') class City: @staticmethod def get_by_insee_code(insee_code, connection): """ Récupère une ville par son code INSEE. """ try: cursor = connection.cursor(dictionary=True) cursor.execute(""" SELECT insee_code, city_name AS name, latitude, longitude, avg_price_per_sqm_apartment, avg_price_per_sqm_house, number_of_transactions FROM cities WHERE insee_code = %s """, (insee_code,)) city_data = cursor.fetchone() if not city_data: print(f"[DEBUG] Aucune ville trouvée avec le code INSEE: {insee_code}") return None print(f"[DEBUG] Ville trouvée: {city_data['name']} (INSEE: {city_data['insee_code']})") return city_data except mysql.connector.Error as e: print(f"[DEBUG] Erreur MySQL dans City.get_by_insee_code: {e}") return None class District: @staticmethod def get_by_id(district_id, connection): """ Récupère un quartier par son ID. """ try: cursor = connection.cursor(dictionary=True) cursor.execute(""" SELECT district_id, district_name AS name, latitude, longitude, avg_price_per_sqm_apartment, avg_price_per_sqm_house, number_of_transactions FROM districts WHERE district_id = %s """, (district_id,)) district_data = cursor.fetchone() if not district_data: print(f"[DEBUG] Aucun quartier trouvé avec l'ID: {district_id}") return None print(f"[DEBUG] Quartier trouvé: {district_data['name']} (ID: {district_data['district_id']})") # Récupération des données associées cursor.execute(""" SELECT total_population, population_by_age, number_of_households, median_income_per_consumption_unit, unemployment_rate, number_of_active_establishments, establishment_type, number_of_housings, housing_type, census_year FROM insee_data WHERE insee_code = (SELECT insee_code FROM districts WHERE district_id = %s) """, (district_id,)) district_data["insee_data"] = cursor.fetchone() cursor.execute(""" SELECT year, avg_price_per_sqm_apartment, avg_price_per_sqm_house, yearly_price_evolution, number_of_transactions FROM territory_indicators WHERE insee_code = (SELECT insee_code FROM districts WHERE district_id = %s) """, (district_id,)) district_data["territory_indicators"] = cursor.fetchall() return district_data except mysql.connector.Error as e: print(f"[DEBUG] Erreur MySQL dans District.get_by_id: {e}") return None class CityDataManager: @staticmethod def get_city_data(insee_code): conn = None try: conn = get_db_connection() if not conn: print("[DEBUG] Échec de la connexion à la base de données.") return None try: current_year = datetime.now().year previous_year = current_year - 1 # 1. Récupération des informations de la ville + prix cursor_city = conn.cursor(dictionary=True, buffered=True) cursor_city.execute(""" SELECT c.insee_code, c.city_name AS name, c.latitude, c.longitude, c.department_code, c.region_code, p.avg_price_per_sqm_apartment, p.avg_price_per_sqm_house, p.number_of_transactions, p.data_year FROM cities c LEFT JOIN prices p ON c.insee_code = p.insee_code AND p.data_year = %s WHERE c.insee_code = %s """, (previous_year, insee_code)) city_data = cursor_city.fetchone() print(f"[DEBUG] Données récupérées : {city_data}") # Debug optionnel return city_data except Exception as e: print(f"[DEBUG] Erreur lors de la récupération des données : {e}") return None finally: if 'cursor_city' in locals(): cursor_city.close() if conn: conn.close() if not city_data: print("[DEBUG] Aucune donnée trouvée pour la ville.") return None # 2. Récupération des quartiers cursor_districts = conn.cursor(dictionary=True, buffered=True) cursor_districts.execute(""" SELECT district_id, district_name AS name, latitude, longitude FROM districts WHERE insee_code = %s """, (insee_code,)) districts = cursor_districts.fetchall() cursor_districts.close() print(f"[DEBUG] Nombre de quartiers trouvés: {len(districts)}") # 3. Pour chaque quartier, récupérer les données associées for district in districts: district_cursor = conn.cursor(dictionary=True, buffered=True) district_cursor.execute(""" SELECT total_population, population_by_age, number_of_households, median_income_per_consumption_unit, unemployment_rate, number_of_active_establishments, establishment_type, number_of_housings, housing_type, census_year FROM insee_data WHERE insee_code = %s """, (insee_code,)) insee_data = district_cursor.fetchone() district["insee_data"] = insee_data district_cursor.execute(""" SELECT year, avg_price_per_sqm_apartment, avg_price_per_sqm_house, yearly_price_evolution, number_of_transactions FROM territory_indicators WHERE insee_code = %s """, (insee_code,)) territory_indicators = district_cursor.fetchall() district["territory_indicators"] = territory_indicators district_cursor.close() # 4. Préparation des données result = { "name": city_data["name"], "insee_code": city_data["insee_code"], "latitude": float(city_data["latitude"]) if city_data["latitude"] is not None else None, "longitude": float(city_data["longitude"]) if city_data["longitude"] is not None else None, "department_code": city_data["department_code"], "region_code": city_data["region_code"], "avg_price_per_sqm_apartment": float(city_data["avg_price_per_sqm_apartment"]) if city_data["avg_price_per_sqm_apartment"] is not None else None, "avg_price_per_sqm_house": float(city_data["avg_price_per_sqm_house"]) if city_data["avg_price_per_sqm_house"] is not None else None, "number_of_transactions": city_data.get("number_of_transactions", 0), "districts": districts } print("[DEBUG] Données complètes récupérées avec succès.") return result except mysql.connector.Error as e: print(f"[DEBUG] Erreur MySQL dans CityDataManager.get_city_data: {e}") return None except Exception as e: print(f"[DEBUG] Erreur inattendue dans CityDataManager.get_city_data: {e}") return None finally: if conn and conn.is_connected(): conn.close() print("[DEBUG] Connexion à la base de données fermée.")