"""
@bp.route("/")
def home():
return render_template_string(html_content)
@bp.route('/search')
def search():
query = request.args.get('q', '').strip()
if not query:
return jsonify([])
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
# Recherche dans les villes (insensible à la casse et aux accents)
cursor.execute("""
SELECT
cities.insee_code,
cities.city_name as name,
'city' as type,
cities.insee_code as id,
cities.longitude as lng,
cities.latitude as lat,
cities.postal_code,
cities.department_code,
departments.department_name,
cities.region_code,
regions.region_name,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions,
(SELECT COUNT(*) FROM districts WHERE districts.insee_code = cities.insee_code) as district_count
FROM cities
LEFT JOIN departments ON cities.department_code = departments.department_code
LEFT JOIN regions ON cities.region_code = regions.region_code
LEFT JOIN prices p ON cities.insee_code = p.insee_code AND p.data_year = 2024
WHERE LOWER(CONVERT(cities.city_name USING utf8mb4)) LIKE LOWER(CONVERT(%s USING utf8mb4))
OR cities.postal_code LIKE %s
LIMIT 5
""", (f'%{query}%', f'%{query}%'))
cities = cursor.fetchall()
# Recherche dans les régions
cursor.execute("""
SELECT
region_name as name,
'region' as type,
region_code as id,
longitude as lng,
latitude as lat
FROM regions
WHERE LOWER(CONVERT(region_name USING utf8mb4)) LIKE LOWER(CONVERT(%s USING utf8mb4))
LIMIT 5
""", (f'%{query}%',))
regions = cursor.fetchall()
# Recherche dans les départements
cursor.execute("""
SELECT
department_name as name,
'department' as type,
department_code as id,
longitude as lng,
latitude as lat
FROM departments
WHERE LOWER(CONVERT(department_name USING utf8mb4)) LIKE LOWER(CONVERT(%s USING utf8mb4))
OR department_code LIKE %s
LIMIT 5
""", (f'%{query}%', f'%{query}%'))
departments = cursor.fetchall()
# Recherche dans les quartiers
cursor.execute("""
SELECT
district_name as name,
'district' as type,
CONCAT(insee_code, '-', district_id) as id,
longitude as lng,
latitude as lat
FROM districts
WHERE LOWER(CONVERT(district_name USING utf8mb4)) LIKE LOWER(CONVERT(%s USING utf8mb4))
LIMIT 5
""", (f'%{query}%',))
districts = cursor.fetchall()
cursor.close()
connection.close()
# Fusionne les résultats
results = cities + regions + departments + districts
return jsonify(results)
@bp.route('/api/countries')
def get_countries():
conn = get_db_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("""
SELECT country_code as id, country_name as name, latitude as lat, longitude as lng
FROM countries
""")
countries = cursor.fetchall()
cursor.close()
conn.close()
return jsonify(countries)
@bp.route('/regions')
def get_regions():
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
query = "SELECT region_code as region_code, region_name as region_name, longitude as lng, latitude as lat FROM regions"
cursor.execute(query)
regions = cursor.fetchall()
cursor.close()
connection.close()
return jsonify(regions)
@bp.route('/departments')
def get_departments():
department_code = request.args.get('department_code')
region_code = request.args.get('region_code')
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
if department_code:
query = """
SELECT department_code, department_name, latitude as lat, longitude as lng
FROM departments
WHERE department_code = %s
"""
cursor.execute(query, (department_code,))
elif region_code:
query = """
SELECT department_code, department_name, latitude as lat, longitude as lng
FROM departments
WHERE region_code = %s
"""
cursor.execute(query, (region_code,))
else:
return jsonify([])
departments = cursor.fetchall()
cursor.close()
connection.close()
return jsonify(departments)
@bp.route('/cities')
def get_cities():
insee_code = request.args.get('insee_code')
department_code = request.args.get('department_code')
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
if insee_code:
query = """
SELECT
cities.insee_code,
cities.city_name as name,
cities.latitude as lat,
cities.longitude as lng,
cities.department_code,
departments.department_name,
cities.region_code,
regions.region_name,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions,
(SELECT COUNT(*) FROM districts WHERE districts.insee_code = cities.insee_code) as district_count
FROM cities
LEFT JOIN departments ON cities.department_code = departments.department_code
LEFT JOIN regions ON cities.region_code = regions.region_code
LEFT JOIN prices p ON cities.insee_code = p.insee_code AND p.data_year = 2024
WHERE cities.insee_code = %s
"""
cursor.execute(query, (insee_code,))
elif department_code:
query = """
SELECT
cities.insee_code,
cities.city_name as name,
cities.latitude as lat,
cities.longitude as lng,
cities.department_code,
departments.department_name,
cities.region_code,
regions.region_name,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions,
(SELECT COUNT(*) FROM districts WHERE districts.insee_code = cities.insee_code) as district_count
FROM cities
LEFT JOIN departments ON cities.department_code = departments.department_code
LEFT JOIN regions ON cities.region_code = regions.region_code
LEFT JOIN prices p ON cities.insee_code = p.insee_code AND p.data_year = 2024
WHERE cities.department_code = %s
"""
cursor.execute(query, (department_code,))
else:
return jsonify([])
cities = cursor.fetchall()
# Initialiser les champs de prix à None si absents
for city in cities:
if city.get("avg_price_per_sqm_apartment") is None:
city["avg_price_per_sqm_apartment"] = None
if city.get("avg_price_per_sqm_house") is None:
city["avg_price_per_sqm_house"] = None
if city.get("number_of_transactions") is None:
city["number_of_transactions"] = 0
cursor.close()
connection.close()
return jsonify(cities)
@bp.route('/districts')
def get_districts():
insee_code = request.args.get('insee_code', None)
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
if insee_code:
query = """
SELECT insee_code, district_name as name, longitude as lng, latitude as lat
FROM districts
WHERE insee_code = %s
"""
cursor.execute(query, (insee_code,))
else:
query = """
SELECT insee_code, district_name as name, longitude as lng, latitude as lat
FROM districts
"""
cursor.execute(query)
districts = cursor.fetchall()
cursor.close()
connection.close()
return jsonify(districts)
@bp.route('/region/')
def show_region(region_code):
# Redirige vers la carte centrée sur la région
return render_template_string(html_content)
@bp.route('/department/')
def show_department(department_code):
# Redirige vers la carte centrée sur le département
return render_template_string(html_content)
@bp.route('/city/-')
def show_city(city_name, insee_code):
# Redirige vers la carte centrée sur la ville
return render_template_string(html_content)
@bp.route('/district/')
def show_district(district_id):
# Redirige vers la carte centrée sur le quartier
return render_template_string(html_content)
@bp.errorhandler(404)
def page_not_found(e):
return render_template('404.html'), 404
@login_manager.user_loader
def load_user(user_id):
# Logique pour charger l'utilisateur depuis la base de données
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT * FROM users WHERE user_id = %s", (user_id,))
user_data = cursor.fetchone()
cursor.close()
connection.close()
if user_data:
return User(**user_data)
return None
@bp.route('/login', methods=['POST'])
def login():
email = request.form.get('email')
password = request.form.get('password')
remember_me = request.form.get('remember_me') == 'on'
if not email or not password:
return jsonify({"success": False, "message": "Email et mot de passe sont obligatoires."}), 400
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("SELECT * FROM users WHERE email = %s", (email,))
user_data = cursor.fetchone()
cursor.close()
connection.close()
if user_data and bcrypt.check_password_hash(user_data['password_hash'], password):
# Crée l'objet User avec tous les champs retournés par la base de données
user = User(
user_id=user_data['user_id'],
email=user_data['email'],
password_hash=user_data['password_hash'],
account_type=user_data['account_type'],
**{k: v for k, v in user_data.items() if k not in ['user_id', 'email', 'password_hash', 'account_type']}
)
login_user(user, remember=remember_me)
return jsonify({"success": True, "message": "Connexion réussie !"})
else:
return jsonify({"success": False, "message": "Email ou mot de passe incorrect."}), 401
@bp.route('/register', methods=['POST'])
def register():
# Récupérer les données du formulaire
email = request.form.get('email')
password = request.form.get('password')
confirm_password = request.form.get('confirm_password')
account_type = request.form.get('account_type')
first_name = request.form.get('first_name')
last_name = request.form.get('last_name')
newsletter_subscription = request.form.get('newsletter_subscription') == 'on'
# Valider les données
if not email or not password or not confirm_password or not account_type or not first_name or not last_name:
return jsonify({"success": False, "message": "Tous les champs sont obligatoires."}), 400
if password != confirm_password:
return jsonify({"success": False, "message": "Les mots de passe ne correspondent pas."}), 400
# Hacher le mot de passe
password_hash = bcrypt.generate_password_hash(password).decode('utf-8')
# Enregistrer l'utilisateur
connection = get_db_connection()
cursor = connection.cursor()
try:
cursor.execute("""
INSERT INTO users
(email, password_hash, account_type, first_name, last_name, newsletter_subscription)
VALUES (%s, %s, %s, %s, %s, %s)
""", (email, password_hash, account_type, first_name, last_name, newsletter_subscription))
connection.commit()
return jsonify({"success": True, "message": "Inscription réussie !"})
except Exception as e:
return jsonify({"success": False, "message": f"Erreur lors de l'inscription : {str(e)}"}), 500
finally:
cursor.close()
connection.close()
@bp.route('/logout', methods=['POST'])
def logout():
logout_user()
return jsonify({"success": True, "message": "Déconnexion réussie."})
@bp.route('/api/account', methods=['GET'])
@login_required
def api_account():
return jsonify({
"logged_in": True,
"email": current_user.email,
"account_type": current_user.account_type,
"first_name": current_user.first_name,
"last_name": current_user.last_name
})
from flask import jsonify, request
from flask_login import current_user, login_required
@bp.route('/api/favorites', methods=['POST'])
@login_required
def add_favorite():
data = request.get_json()
favorite_type = data.get('favorite_type') # 'region', 'department', 'city', ou 'district'
favorite_code = data.get('favorite_code')
favorite_name = data.get('favorite_name')
# Vérifie si l'utilisateur est premium
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("""
SELECT is_premium
FROM user_subscriptions
WHERE user_id = %s
""", (current_user.id,))
subscription = cursor.fetchone()
is_premium = subscription['is_premium'] if subscription else False
# Compte le nombre de favoris existants
cursor.execute("""
SELECT COUNT(*) AS count
FROM favorites
WHERE user_id = %s
""", (current_user.id,))
favorite_count = cursor.fetchone()['count']
# Vérifie la limite
if not is_premium and favorite_count >= 5:
return jsonify({"success": False, "message": "Limite de 5 favoris atteinte. Passez en version premium pour en ajouter plus."}), 403
elif is_premium and favorite_count >= 50:
return jsonify({"success": False, "message": "Limite de 50 favoris atteinte."}), 403
# Ajoute le favori
cursor.execute("""
INSERT INTO favorites (user_id, favorite_type, favorite_code, favorite_name)
VALUES (%s, %s, %s, %s)
""", (current_user.id, favorite_type, favorite_code, favorite_name))
connection.commit()
cursor.close()
connection.close()
return jsonify({"success": True, "message": "Favori ajouté avec succès !"})
@bp.route('/api/favorites', methods=['GET'])
@login_required
def get_favorites():
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("""
SELECT * FROM favorites
WHERE user_id = %s
""", (current_user.id,))
favorites = cursor.fetchall()
cursor.close()
connection.close()
return jsonify({"success": True, "favorites": favorites})
@bp.route('/api/favorites', methods=['DELETE'])
@login_required
def delete_favorite():
data = request.get_json()
favorite_type = data.get('favorite_type')
favorite_code = data.get('favorite_code')
connection = get_db_connection()
cursor = connection.cursor()
cursor.execute("""
DELETE FROM favorites
WHERE user_id = %s AND favorite_type = %s AND favorite_code = %s
""", (current_user.id, favorite_type, favorite_code))
connection.commit()
cursor.close()
connection.close()
return jsonify({"success": True, "message": "Favori supprimé avec succès !"})
@bp.route('/api/favorites/check', methods=['GET'])
@login_required
def check_favorite():
favorite_type = request.args.get('type')
favorite_code = request.args.get('code')
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("""
SELECT 1 FROM favorites
WHERE user_id = %s AND favorite_type = %s AND favorite_code = %s
""", (current_user.id, favorite_type, favorite_code))
is_favorite = cursor.fetchone() is not None
cursor.close()
connection.close()
return jsonify({"is_favorite": is_favorite})
@bp.route('/api/similar_cities')
def get_similar_cities():
insee_code = request.args.get('insee_code')
if not insee_code:
return jsonify([])
connection = None
try:
connection = get_db_connection()
cursor = connection.cursor(dictionary=True, buffered=True)
# 1. Récupère les coordonnées de la ville actuelle + ses prix
cursor.execute("""
SELECT c.city_name, c.latitude as lat, c.longitude as lng,
p.avg_price_per_sqm_apartment, p.avg_price_per_sqm_house
FROM cities c
LEFT JOIN prices p ON c.insee_code = p.insee_code AND p.data_year = 2024
WHERE c.insee_code = %s
""", (insee_code,))
current_city = cursor.fetchone()
if not current_city or current_city['lat'] is None or current_city['lng'] is None:
return jsonify([])
current_coords = (current_city['lat'], current_city['lng'])
# 2. Récupère les villes à moins de 10 km (filtre côté SQL)
cursor.execute("""
SELECT c.city_name, c.latitude as lat, c.longitude as lng,
p.avg_price_per_sqm_apartment, p.avg_price_per_sqm_house
FROM cities c
LEFT JOIN prices p ON c.insee_code = p.insee_code AND p.data_year = 2024
WHERE c.insee_code != %s
AND c.latitude IS NOT NULL
AND c.longitude IS NOT NULL
""", (insee_code,))
all_cities = cursor.fetchall()
similar_cities = []
for city in all_cities:
city_coords = (city['lat'], city['lng'])
distance = geodesic(current_coords, city_coords).kilometers
if distance < 10:
price = city['avg_price_per_sqm_apartment'] or city['avg_price_per_sqm_house']
similar_cities.append({
'name': city['city_name'],
'price': float(price) if price is not None else None,
'distance': round(distance, 1),
'lat': city['lat'],
'lng': city['lng'],
})
# 3. Trie par distance et retourne les 5 plus proches
similar_cities.sort(key=lambda x: x['distance'])
return jsonify(similar_cities[:5])
except Exception as e:
print(f"[DEBUG] Erreur dans get_similar_cities: {e}")
return jsonify([])
finally:
if cursor:
cursor.close()
if connection and connection.is_connected():
connection.close()
@bp.route('/api/prices/')
def get_prices(insee_code):
try:
connection = get_db_connection()
cursor = connection.cursor(dictionary=True, buffered=True)
cursor.execute("""
SELECT
data_year AS year,
avg_price_per_sqm_apartment,
avg_price_per_sqm_house
FROM prices
WHERE insee_code = %s
ORDER BY data_year
""", (insee_code,))
results = cursor.fetchall()
past_prices = []
for row in results:
year = int(row['year']) # data_year est déjà un int, pas besoin de YEAR()
date_obj = date(year, 1, 1) # Crée une date au 1er janvier de l'année
price = float(row['avg_price_per_sqm_apartment']) if row['avg_price_per_sqm_apartment'] is not None else float(row['avg_price_per_sqm_house'])
past_prices.append({"date": date_obj.isoformat(), "price": price})
return jsonify({"pastPrices": past_prices, "futurePrices": []})
except Exception as e:
return jsonify({"error": str(e)}), 500
finally:
cursor.close()
connection.close()
@bp.route('/api/city-data', methods=['GET'])
def city_data():
insee_code = request.args.get('insee_code')
if not insee_code:
return jsonify({"error": "Le paramètre insee_code est requis."}), 400
city_data = CityDataManager.get_city_data(insee_code)
if not city_data:
return jsonify({"error": "Ville non trouvée ou erreur serveur."}), 404
if 'districts' not in city_data:
city_data['districts'] = []
try:
response = {
"city": {
"name": city_data["name"],
"insee_code": city_data["insee_code"],
"latitude": city_data["latitude"],
"longitude": city_data["longitude"],
"avg_price_per_sqm_apartment": city_data["avg_price_per_sqm_apartment"],
"avg_price_per_sqm_house": city_data["avg_price_per_sqm_house"],
"number_of_transactions": city_data["number_of_transactions"],
"district_count": len(city_data["districts"]),
},
"districts": city_data["districts"]
}
return jsonify(response)
except Exception as e:
print(f"[DEBUG] Erreur lors de la préparation de la réponse: {e}")
return jsonify({"error": "Erreur lors de la préparation de la réponse."}), 500
@bp.route('/api/real-estate-sites', methods=['GET'])
def get_real_estate_sites():
try:
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
query = """
SELECT website_id, name, url, logo_url, description, allow_iframe
FROM real_estate_websites
WHERE is_active = 1
ORDER BY name
"""
cursor.execute(query)
sites = cursor.fetchall()
cursor.close()
connection.close()
return jsonify({"success": True, "sites": sites})
except Exception as e:
return jsonify({"success": False, "message": str(e)}), 500
from flask import jsonify, request
@bp.route('/api/schools')
def get_schools():
connection = get_db_connection()
cursor = connection.cursor(dictionary=True, buffered=True)
insee_code = request.args.get('insee_code')
lat = request.args.get('lat', type=float)
lon = request.args.get('lon', type=float)
radius = request.args.get('radius', default=1.0, type=float) # Rayon en km
if insee_code: # Filtre par code INSEE (ville)
cursor.execute("""
SELECT service_name, address, missions, phone, website, latitude, longitude
FROM public_services
WHERE insee_code = %s AND category = 'school'
LIMIT 20
""", (insee_code,))
elif lat and lon: # Filtre par proximité (quartier)
cursor.execute("""
SELECT service_name, address, missions, phone, website, latitude, longitude,
(6371 * ACOS(
COS(RADIANS(%s)) * COS(RADIANS(latitude)) *
COS(RADIANS(longitude) - RADIANS(%s)) +
SIN(RADIANS(%s)) * SIN(RADIANS(latitude))
)) AS distance
FROM public_services
WHERE category = 'school'
HAVING distance < %s
ORDER BY distance
LIMIT 20
""", (lat, lon, lat, radius))
else:
return jsonify({"error": "Paramètres manquants"}), 400
schools = cursor.fetchall()
cursor.close()
return jsonify(schools)