Accédez à des prévisions immobilières basées sur l'intelligence artificielle.
Données avancées
Bénéficiez de données exclusives sur les tendances du marché.
Cartes interactives
Visualisez les opportunités immobilières sur des cartes détaillées.
9,99 €/mois
Sans engagement. Annulez à tout moment.
Paiement sécurisé par Stripe (cryptage SSL, conformité PCI).
Vous serez redirigé vers PayPal pour finaliser le paiement.
Effectuez un virement sur le compte suivant :
IBAN : FR76 1695 8000 0110 8713 8721 488
BIC : QNTOFRP1XXX
Titulaire : Hoaviere SAS
Motif (obligatoire) : PRO-{{ current_user.id }}
Votre abonnement sera activé sous 24h après réception du paiement.
"""
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 loc_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 loc_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 loc_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 loc_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,
r.region_name,
c.population,
(
SELECT COUNT(*)
FROM loc_districts
WHERE insee_code = c.insee_code
) AS district_count,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions,
p.data_year
FROM loc_cities c
JOIN loc_regions r ON c.region_code = r.region_code
LEFT JOIN data_prices p ON c.id = p.city_id
AND p.district_id IS NULL
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_loc_districts.execute("""
SELECT district_id, district_name AS name, latitude, longitude
FROM loc_districts
WHERE insee_code = %s
""", (insee_code,))
districts = cursor_loc_districts.fetchall()
cursor_loc_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)
# Récupérer les données INSEE du quartier (si nécessaire)
district_cursor.execute("""
SELECT *
FROM insee_data
WHERE insee_code = %s
""", (insee_code,))
insee_data = district_cursor.fetchone()
district["insee_data"] = insee_data
# Récupérer les prix du quartier
district_cursor.execute("""
SELECT
avg_price_per_sqm_apartment,
avg_price_per_sqm_house,
number_of_transactions,
data_year
FROM data_prices
WHERE district_id = %s AND data_year = %s
""", (district["district_id"], previous_year))
district_prices = district_cursor.fetchone()
district.update({
"avg_price_per_sqm_apartment": float(district_prices["avg_price_per_sqm_apartment"]) if district_prices and district_prices["avg_price_per_sqm_apartment"] is not None else None,
"avg_price_per_sqm_house": float(district_prices["avg_price_per_sqm_house"]) if district_prices and district_prices["avg_price_per_sqm_house"] is not None else None,
"number_of_transactions": district_prices["number_of_transactions"] if district_prices else 0,
})
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.")
def get_db_connection():
return mysql.connector.connect(
host="localhost",
user="root",
password="_z7b6CJc",
database="immo_new"
)
@auth.verify_password
def verify_password(username, password):
if username in users and check_password_hash(users.get(username), password):
return username
# Protège toutes les routes
@app.before_request
@auth.login_required
def before_request():
pass
@app.route("/")
def home():
return render_template_string(html_content)
def check_pro(f):
@wraps(f)
def decorated_function(*args, **kwargs):
# Vérifie si l'utilisateur est connecté et a un abonnement Pro
if not current_user.is_authenticated or not getattr(current_user, 'is_pro', False):
return jsonify({
'success': False,
'message': "Accès réservé aux abonnés Pro. Souscrivez pour débloquer cette fonctionnalité."
}), 403
return f(*args, **kwargs)
return decorated_function
@app.route('/api/update-profile', methods=['POST'])
@login_required
def update_profile():
try:
# Récupère les données du formulaire
data = request.get_json()
first_name = data.get('first_name')
last_name = data.get('last_name')
phone = data.get('phone')
newsletter_subscription = data.get('newsletter_subscription', False)
# Valide les données
if not first_name or not last_name:
return jsonify({
"success": False,
"message": "Le prénom et le nom sont obligatoires."
}), 400
# Met à jour les informations dans la base de données
connection = get_db_connection()
cursor = connection.cursor()
query = """
UPDATE users
SET first_name = %s, last_name = %s, phone = %s, newsletter_subscription = %s
WHERE user_id = %s
"""
cursor.execute(query, (first_name, last_name, phone, newsletter_subscription, current_user.id))
connection.commit()
# Met à jour les informations de l'utilisateur actuel
current_user.first_name = first_name
current_user.last_name = last_name
current_user.phone = phone
current_user.newsletter_subscription = newsletter_subscription
cursor.close()
connection.close()
return jsonify({
"success": True,
"message": "Votre profil a été mis à jour avec succès."
})
except mysql.connector.Error as e:
return jsonify({
"success": False,
"message": f"Erreur de base de données : {str(e)}"
}), 500
except Exception as e:
return jsonify({
"success": False,
"message": f"Une erreur est survenue : {str(e)}"
}), 500
@app.route('/api/check-pro-status', methods=['GET'])
@login_required
def check_pro_status():
return jsonify({'isPro': current_user.is_pro})
@app.route('/api/user-status', methods=['GET'])
def user_status():
if current_user.is_authenticated:
return jsonify({
'isAuthenticated': True,
'isPro': getattr(current_user, 'is_pro', False),
'email': current_user.email,
'firstName': current_user.first_name, # Assure-toi que ces champs existent dans ton modèle User
'lastName': current_user.last_name,
})
else:
return jsonify({'isAuthenticated': False})
@app.route('/subscription')
@login_required
def subscription():
return render_template('subscription.html')
@app.route('/create-payment-intent', methods=['POST'])
@login_required
def create_payment_intent():
try:
data = request.get_json()
payment_method_id = data.get('payment_method_id')
if not payment_method_id:
return jsonify({'success': False, 'message': "ID de méthode de paiement manquant"}), 400
# Crée un client Stripe si ce n'est pas déjà fait
if not hasattr(current_user, 'stripe_customer_id') or not current_user.stripe_customer_id:
customer = stripe.Customer.create(
email=current_user.email,
name=f"{getattr(current_user, 'first_name', '')} {getattr(current_user, 'last_name', '')}".strip(),
payment_method=payment_method_id,
invoice_settings={
'default_payment_method': payment_method_id,
},
)
# Met à jour le stripe_customer_id en base de données
connection = get_db_connection()
cursor = connection.cursor()
cursor.execute("""
UPDATE users
SET stripe_customer_id = %s
WHERE user_id = %s
""", (customer.id, current_user.id))
connection.commit()
cursor.close()
connection.close()
# Met à jour l'objet current_user
current_user.stripe_customer_id = customer.id
# Crée un PaymentIntent
payment_intent = stripe.PaymentIntent.create(
amount=999, # 9,99 € (en centimes)
currency='eur',
customer=current_user.stripe_customer_id,
payment_method=payment_method_id,
off_session=True,
confirm=True,
)
return jsonify({
'success': True,
'clientSecret': payment_intent.client_secret,
})
except StripeError as e:
return jsonify({'success': False, 'message': f"Erreur Stripe : {str(e)}"}), 400
except Exception as e:
return jsonify({'success': False, 'message': f"Erreur serveur : {str(e)}"}), 500
@app.route('/create-subscription', methods=['POST'])
@login_required
def create_subscription():
try:
if not hasattr(current_user, 'stripe_customer_id') or not current_user.stripe_customer_id:
return jsonify({'success': False, 'message': "ID client Stripe manquant"}), 400
# Crée l'abonnement
subscription = stripe.Subscription.create(
customer=current_user.stripe_customer_id,
items=[{'price': 'price_1RFyjCQHKuOhs2O46XGLCdiB'}], # Remplace par ton ID de prix
)
# Met à jour le statut pro et l'ID de souscription en base de données
connection = get_db_connection()
cursor = connection.cursor()
cursor.execute("""
UPDATE users
SET stripe_subscription_id = %s, is_pro = %s
WHERE user_id = %s
""", (subscription.id, True, current_user.id))
connection.commit()
cursor.close()
connection.close()
# Met à jour l'objet current_user
current_user.stripe_subscription_id = subscription.id
current_user.is_pro = True
return jsonify({'success': True, 'message': 'Souscription réussie !'})
except StripeError as e:
return jsonify({'success': False, 'message': f"Erreur Stripe : {str(e)}"}), 400
except Exception as e:
return jsonify({'success': False, 'message': f"Erreur serveur : {str(e)}"}), 500
@app.route('/stripe-webhook', methods=['POST'])
def stripe_webhook():
payload = request.get_data(as_text=True)
sig_header = request.headers.get('Stripe-Signature')
event = None
try:
event = stripe.Webhook.construct_event(
payload, sig_header, app.config['STRIPE_WEBHOOK_SECRET']
)
except ValueError as e:
return 'Invalid payload', 400
except stripe.error.SignatureVerificationError as e:
return 'Invalid signature', 400
# Gérer l'annulation ou l'expiration de l'abonnement
if event['type'] in ('customer.subscription.deleted', 'customer.subscription.updated'):
subscription = event['data']['object']
if subscription['status'] != 'active':
connection = get_db_connection()
cursor = connection.cursor()
cursor.execute("""
UPDATE users
SET is_pro = FALSE, stripe_subscription_id = NULL
WHERE stripe_subscription_id = %s
""", (subscription.id,))
connection.commit()
cursor.close()
connection.close()
return jsonify({'success': True}), 200
# Page À propos
@app.route('/about')
def about():
return render_template('about.html')
# Politique de confidentialité
@app.route('/privacy')
def privacy():
return render_template('privacy.html')
# Conditions d'utilisation
@app.route('/terms')
def terms():
return render_template('terms.html')
@app.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 (avec jointure pour le score)
cursor.execute("""
SELECT
loc_cities.insee_code,
loc_cities.city_name AS name,
'ville' AS type,
loc_cities.insee_code AS id,
loc_cities.longitude AS lng,
loc_cities.latitude AS lat,
loc_cities.postal_code,
loc_cities.department_code,
loc_departments.department_name,
loc_cities.region_code,
loc_regions.region_name,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions,
(SELECT COUNT(*) FROM loc_districts WHERE loc_districts.insee_code = loc_cities.insee_code) AS district_count,
ls.score AS city_score, -- Renommer le score en `city_score`
CASE
WHEN LOWER(TRIM(loc_cities.city_name)) = LOWER(TRIM(%s)) THEN 0
WHEN loc_cities.postal_code LIKE %s THEN 1
ELSE 2
END AS relevance
FROM loc_cities
LEFT JOIN loc_departments ON loc_cities.department_code = loc_departments.department_code
LEFT JOIN loc_regions ON loc_cities.region_code = loc_regions.region_code
LEFT JOIN data_prices p ON loc_cities.id = p.city_id
AND p.district_id IS NULL
AND p.data_year = YEAR(CURDATE()) - 1
LEFT JOIN loc_scores ls ON loc_cities.insee_code = ls.insee_code -- Jointure pour le score
WHERE
LOWER(TRIM(loc_cities.city_name)) COLLATE utf8mb4_general_ci = LOWER(TRIM(%s)) COLLATE utf8mb4_general_ci
OR LOWER(loc_cities.city_name) COLLATE utf8mb4_general_ci LIKE LOWER(%s) COLLATE utf8mb4_general_ci
OR loc_cities.postal_code LIKE %s
ORDER BY relevance, name
LIMIT 5
""", (query, f'%{query}%', query, f'%{query}%', f'%{query}%'))
cities = cursor.fetchall()
# Recherche dans les régions (inchangée)
cursor.execute("""
SELECT
region_name AS name,
'région' AS type,
region_code AS id,
longitude AS lng,
latitude AS lat,
CASE
WHEN LOWER(TRIM(region_name)) = LOWER(TRIM(%s)) THEN 0
ELSE 1
END AS relevance
FROM loc_regions
WHERE LOWER(region_name) COLLATE utf8mb4_general_ci LIKE LOWER(%s) COLLATE utf8mb4_general_ci
ORDER BY relevance, name
LIMIT 5
""", (query, f'%{query}%'))
regions = cursor.fetchall()
# Recherche dans les départements (inchangée)
cursor.execute("""
SELECT
department_name AS name,
'département' AS type,
department_code AS id,
longitude AS lng,
latitude AS lat,
CASE
WHEN LOWER(TRIM(department_name)) = LOWER(TRIM(%s)) THEN 0
WHEN department_code LIKE %s THEN 1
ELSE 2
END AS relevance
FROM loc_departments
WHERE
LOWER(department_name) COLLATE utf8mb4_general_ci LIKE LOWER(%s) COLLATE utf8mb4_general_ci
OR department_code LIKE %s
ORDER BY relevance, name
LIMIT 5
""", (query, f'%{query}%', f'%{query}%', f'%{query}%'))
departments = cursor.fetchall()
# Recherche dans les quartiers (inchangée)
cursor.execute("""
SELECT
loc_districts.district_name AS name,
loc_districts.insee_code,
'quartier' AS type,
CONCAT(loc_districts.insee_code, '-', loc_districts.district_id) AS id,
loc_districts.longitude AS lng,
loc_districts.latitude AS lat,
loc_cities.postal_code AS city_postal_code,
CASE
WHEN LOWER(TRIM(loc_districts.district_name)) = LOWER(TRIM(%s)) THEN 0
ELSE 1
END AS relevance
FROM loc_districts
LEFT JOIN loc_cities ON loc_districts.insee_code = loc_cities.insee_code
WHERE LOWER(loc_districts.district_name) COLLATE utf8mb4_general_ci LIKE LOWER(%s) COLLATE utf8mb4_general_ci
ORDER BY relevance, name
LIMIT 5
""", (query, f'%{query}%'))
districts = cursor.fetchall()
cursor.close()
connection.close()
results = cities + regions + departments + districts
return jsonify(results)
@app.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 loc_countries
""")
countries = cursor.fetchall()
cursor.close()
conn.close()
return jsonify(countries)
@app.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 loc_regions"
cursor.execute(query)
regions = cursor.fetchall()
cursor.close()
connection.close()
return jsonify(regions)
@app.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
d.department_code,
d.department_name,
d.latitude AS lat,
d.longitude AS lng,
r.region_code,
r.region_name
FROM
loc_departments d
JOIN
loc_regions r ON d.region_code = r.region_code
WHERE
d.department_code = %s
"""
cursor.execute(query, (department_code,))
elif region_code:
query = """
SELECT
d.department_code,
d.department_name,
d.latitude AS lat,
d.longitude AS lng,
r.region_code,
r.region_name
FROM
loc_departments d
JOIN
loc_regions r ON d.region_code = r.region_code
WHERE
d.region_code = %s
"""
cursor.execute(query, (region_code,))
else:
return jsonify([])
departments = cursor.fetchall()
cursor.close()
connection.close()
return jsonify(departments)
@app.route('/cities')
def get_cities():
insee_code = request.args.get('insee_code')
department_code = request.args.get('department_code')
city_name = request.args.get('city_name')
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
if insee_code:
query = """
SELECT
loc_cities.insee_code,
loc_cities.city_name as name,
loc_cities.latitude as lat,
loc_cities.longitude as lng,
loc_cities.department_code,
loc_departments.department_name,
loc_cities.region_code,
loc_cities.population,
loc_cities.postal_code,
loc_regions.region_name,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions,
(SELECT COUNT(*) FROM loc_districts WHERE loc_districts.insee_code = loc_cities.insee_code) as district_count,
(
SELECT ls.score
FROM loc_scores ls
WHERE ls.insee_code = loc_cities.insee_code
AND ls.district_id IS NULL
AND ls.data_year = (SELECT MAX(data_year) FROM loc_scores WHERE insee_code = loc_cities.insee_code AND district_id IS NULL)
) as city_score
FROM loc_cities
LEFT JOIN loc_departments ON loc_cities.department_code = loc_departments.department_code
LEFT JOIN loc_regions ON loc_cities.region_code = loc_regions.region_code
LEFT JOIN data_prices p ON loc_cities.insee_code = p.insee_code AND p.data_year = YEAR(CURDATE()) - 1 AND p.district_id IS NULL
WHERE loc_cities.insee_code = %s
"""
cursor.execute(query, (insee_code,))
elif department_code:
query = """
SELECT
loc_cities.id,
loc_cities.insee_code,
loc_cities.city_name as name,
loc_cities.latitude as lat,
loc_cities.longitude as lng,
loc_cities.department_code,
loc_departments.department_name,
loc_cities.region_code,
loc_regions.region_name,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions,
(SELECT COUNT(*) FROM loc_districts WHERE loc_districts.insee_code = loc_cities.insee_code) as district_count,
(
SELECT ls.score
FROM loc_scores ls
WHERE ls.insee_code = loc_cities.insee_code
AND ls.district_id IS NULL
AND ls.data_year = (SELECT MAX(data_year) FROM loc_scores WHERE insee_code = loc_cities.insee_code AND district_id IS NULL)
) as city_score
FROM loc_cities
LEFT JOIN loc_departments ON loc_cities.department_code = loc_departments.department_code
LEFT JOIN loc_regions ON loc_cities.region_code = loc_regions.region_code
LEFT JOIN data_prices p ON loc_cities.insee_code = p.insee_code AND p.data_year = YEAR(CURDATE()) - 1
WHERE loc_cities.department_code = %s
"""
cursor.execute(query, (department_code,))
elif city_name:
query = """
SELECT
loc_cities.insee_code,
loc_cities.city_name as name,
loc_cities.latitude as lat,
loc_cities.longitude as lng,
loc_cities.department_code,
loc_departments.department_name,
loc_cities.region_code,
loc_regions.region_name,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions,
(SELECT COUNT(*) FROM loc_districts WHERE loc_districts.insee_code = loc_cities.insee_code) as district_count,
(
SELECT ls.score
FROM loc_scores ls
WHERE ls.insee_code = loc_cities.insee_code
AND ls.district_id IS NULL
AND ls.data_year = (SELECT MAX(data_year) FROM loc_scores WHERE insee_code = loc_cities.insee_code AND district_id IS NULL)
) as city_score
FROM loc_cities
LEFT JOIN loc_departments ON loc_cities.department_code = loc_departments.department_code
LEFT JOIN loc_regions ON loc_cities.region_code = loc_regions.region_code
LEFT JOIN data_prices p ON loc_cities.insee_code = p.insee_code AND p.data_year = YEAR(CURDATE()) - 1 AND p.district_id IS NULL
WHERE LOWER(loc_cities.city_name) = LOWER(%s)
"""
cursor.execute(query, (city_name,))
else:
return jsonify([])
cities = cursor.fetchall()
# Initialiser les champs manquants à None ou 0
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
if city.get("city_score") is None:
city["city_score"] = None # Score non disponible
cursor.close()
connection.close()
return jsonify(cities)
@app.route('/districts')
def get_districts():
insee_code = request.args.get('insee_code', None)
district_id = request.args.get('district_id', None)
if not insee_code and not district_id:
return jsonify({"error": "Paramètre manquant : insee_code ou district_id requis"}), 400
connection = get_db_connection()
cursor = connection.cursor(dictionary=True, buffered=True)
try:
if district_id:
query = """
SELECT
d.district_id as id,
d.insee_code,
d.district_name as name,
d.longitude as lng,
d.latitude as lat,
c.department_code,
c.region_code,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions
FROM loc_districts d
JOIN loc_cities c ON d.insee_code = c.insee_code
LEFT JOIN data_prices p ON d.district_id = p.district_id AND p.data_year = YEAR(CURDATE()) - 1
WHERE d.district_id = %s
"""
cursor.execute(query, (district_id,)) # <-- Ajout manquant
elif insee_code:
query = """
SELECT
d.district_id as id,
d.insee_code,
d.district_name as name,
d.longitude as lng,
d.latitude as lat,
c.department_code,
c.region_code,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house,
p.number_of_transactions
FROM loc_districts d
JOIN loc_cities c ON d.insee_code = c.insee_code
LEFT JOIN data_prices p ON d.district_id = p.district_id AND p.data_year = YEAR(CURDATE()) - 1
WHERE d.insee_code = %s
"""
cursor.execute(query, (insee_code,))
districts = cursor.fetchall()
return jsonify(districts)
except Exception as e:
print(f"Erreur SQL: {e}")
return jsonify({"error": "Erreur serveur"}), 500
finally:
cursor.close()
connection.close()
@app.route('/api/services', methods=['GET'])
def get_services():
insee_code = request.args.get('insee_code')
subthemes = request.args.get('subthemes', '') # Utilisez "subthemes" au lieu de "categories"
app.logger.debug(f"insee_code: {insee_code}")
app.logger.debug(f"subthemes: {subthemes}")
if not insee_code:
return jsonify({"error": "insee_code manquant"}), 400
try:
with get_db_connection() as conn:
with conn.cursor(dictionary=True) as cur:
if not subthemes:
query = """
SELECT service_id, service_name, category, original_category, address, postal_code, city,
insee_code, phone, website, latitude, longitude,missions
FROM data_services
WHERE insee_code = %s
"""
params = [insee_code]
else:
subtheme_list = [subtheme.strip() for subtheme in subthemes.split(",") if subtheme.strip()]
if not subtheme_list:
return jsonify([])
# Requête pour filtrer par sous-thème (original_category)
query = f"""
SELECT service_id, service_name, category, original_category, address, postal_code, city,
insee_code, phone, website, latitude, longitude,missions
FROM data_services
WHERE insee_code = %s AND original_category IN ({', '.join(['%s'] * len(subtheme_list))})
"""
params = [insee_code] + subtheme_list
app.logger.debug(f"query: {query}")
app.logger.debug(f"params: {params}")
cur.execute(query, params)
services = cur.fetchall()
app.logger.debug(f"services: {services}")
return jsonify(services)
except psycopg2.Error as e:
app.logger.error(f"Database error: {e}")
return jsonify({"error": "Erreur de base de données"}), 500
except Exception as e:
app.logger.error(f"Unexpected error: {e}")
return jsonify({"error": "Erreur interne du serveur"}), 500
@app.route('/address/')
def show_address(full_address):
# Remplacer les tirets par des espaces pour reformer l'adresse
return render_template_string(html_content)
@app.route('/region/')
def show_region(region_code):
# Redirige vers la carte centrée sur la région
return render_template_string(html_content)
@app.route('/department/')
def show_department(department_code):
# Redirige vers la carte centrée sur le département
return render_template_string(html_content)
@app.route('/city/-')
def show_city(city_name, insee_code):
# Redirige vers la carte centrée sur la ville
return render_template_string(html_content)
@app.route('/district/')
def show_district(district_id):
# Redirige vers la carte centrée sur le quartier
return render_template_string(html_content)
@app.errorhandler(404)
def page_not_found(e):
return render_template('404.html'), 404
# 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)
# Gestion spécifique des attributs Stripe
self.stripe_customer_id = kwargs.get('stripe_customer_id', None)
self.stripe_subscription_id = kwargs.get('stripe_subscription_id', None)
self.is_pro = kwargs.get('is_pro', False)
# 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)
@login_manager.user_loader
def load_user(user_id):
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
cursor.execute("""
SELECT user_id, email, password_hash, account_type,
stripe_customer_id, stripe_subscription_id, is_pro,
first_name, last_name, newsletter_subscription
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
@app.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
@app.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()
@app.route('/logout', methods=['POST'])
def logout():
logout_user()
return jsonify({"success": True, "message": "Déconnexion réussie."})
@app.route('/api/account', methods=['GET'])
@login_required
def account_info():
if current_user.is_authenticated:
# Vérifie l'état de l'abonnement Stripe si l'utilisateur est marqué comme pro
if current_user.is_pro and current_user.stripe_subscription_id:
try:
subscription = stripe.Subscription.retrieve(current_user.stripe_subscription_id)
if subscription.status != 'active':
connection = get_db_connection()
cursor = connection.cursor()
cursor.execute("""
UPDATE users
SET is_pro = FALSE
WHERE user_id = %s
""", (current_user.id,))
connection.commit()
cursor.close()
connection.close()
current_user.is_pro = False
except stripe.error.StripeError:
pass # En cas d'erreur, on garde l'état actuel
return jsonify({
'logged_in': True,
'email': current_user.email,
'first_name': current_user.first_name,
'last_name': current_user.last_name,
'account_type': current_user.account_type,
'is_pro': current_user.is_pro,
})
else:
return jsonify({'logged_in': False})
@app.route('/api/account/password', methods=['PUT'])
@login_required
def update_password():
data = request.get_json()
current_password = data.get('current_password')
new_password = data.get('new_password')
if not check_password_hash(current_user.password_hash, current_password):
return jsonify({'success': False, 'message': 'Mot de passe actuel incorrect'}), 400
try:
new_password_hash = generate_password_hash(new_password)
connection = get_db_connection()
cursor = connection.cursor()
cursor.execute("""
UPDATE users
SET password_hash = %s
WHERE user_id = %s
""", (new_password_hash, current_user.id))
connection.commit()
return jsonify({'success': True, 'message': 'Mot de passe mis à jour avec succès'})
except Exception as e:
return jsonify({'success': False, 'message': str(e)}), 500
finally:
cursor.close()
connection.close()
@app.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 !"})
@app.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})
@app.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 !"})
@app.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})
@app.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 loc_cities c
LEFT JOIN data_prices p ON c.insee_code = p.insee_code AND p.data_year = YEAR(CURDATE()) - 1
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 loc_cities c
LEFT JOIN data_prices p ON c.insee_code = p.insee_code AND p.data_year = YEAR(CURDATE()) - 1
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_loc_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_loc_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()
def get_price_history_and_forecast(insee_code):
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
try:
query = """
SELECT
data_year as year,
avg_price_per_sqm_house as house_price,
avg_price_per_sqm_apartment as apartment_price
FROM data_prices
WHERE insee_code = %s AND district_id IS NULL
ORDER BY data_year
"""
cursor.execute(query, (insee_code,))
rows = cursor.fetchall()
# Prépare les données historiques
past_houses = [{"date": f"{row['year']}-01-01", "price": float(row['house_price'])} for row in rows if row['house_price'] is not None]
past_apartments = [{"date": f"{row['year']}-01-01", "price": float(row['apartment_price'])} for row in rows if row['apartment_price'] is not None]
# Génère les prévisions (uniquement si on a des données historiques)
future_houses = []
future_apartments = []
if past_houses and past_apartments:
last_year = rows[-1]['year']
last_house_price = past_houses[-1]['price']
last_apartment_price = past_apartments[-1]['price']
# Prévisions pour les 2 années suivantes
future_houses = [
{"date": f"{last_year + 1}-01-01", "price": round(last_house_price * 1.02, 2)},
{"date": f"{last_year + 2}-01-01", "price": round(last_house_price * 1.04, 2)}
]
future_apartments = [
{"date": f"{last_year + 1}-01-01", "price": round(last_apartment_price * 1.02, 2)},
{"date": f"{last_year + 2}-01-01", "price": round(last_apartment_price * 1.04, 2)}
]
return {
"pastPrices": past_houses,
"futurePrices": future_houses,
"pastPricesApartments": past_apartments,
"futurePricesApartments": future_apartments
}
except Exception as e:
print(f"[DEBUG] Erreur dans get_price_history_and_forecast: {e}")
return {
"pastPrices": [],
"futurePrices": [],
"pastPricesApartments": [],
"futurePricesApartments": []
}
finally:
cursor.close()
connection.close()
@app.route('/api/prices/')
def get_prices(insee_code):
try:
connection = get_db_connection()
cursor = connection.cursor(dictionary=True, buffered=True)
# Prix de la ville
cursor.execute("""
SELECT
data_year AS year,
avg_price_per_sqm_apartment,
avg_price_per_sqm_house
FROM data_prices
WHERE city_id = (SELECT id From loc_cities WHERE insee_code = %s)
AND district_id IS NULL
ORDER BY data_year
""", (insee_code,))
city_prices = cursor.fetchall()
# Prix des quartiers
cursor.execute("""
SELECT
d.district_id,
d.district_name,
p.data_year AS year,
p.avg_price_per_sqm_apartment,
p.avg_price_per_sqm_house
FROM loc_districts d
JOIN data_prices p ON d.district_id = p.district_id
WHERE d.insee_code = %s
ORDER BY d.district_name, p.data_year
""", (insee_code,))
district_prices = cursor.fetchall()
# Formatage des résultats
past_prices = []
for row in city_prices:
year = int(row['year'])
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": f"{year}-01-01", "price": price, "type": "city"})
for row in district_prices:
year = int(row['year'])
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": f"{year}-01-01",
"price": price,
"type": "district",
"district_id": row['district_id'],
"district_name": row['district_name']
})
return jsonify({"pastPrices": past_prices, "futurePrices": []})
except Exception as e:
return jsonify({"error": str(e)}), 500
finally:
cursor.close()
connection.close()
@app.route('/geocode')
def geocode():
address = request.args.get('address')
geocode_url = f"https://nominatim.openstreetmap.org/search?format=json&q={address}&addressdetails=1" # <-- Ajoutez &addressdetails=1
try:
headers = {
'User-Agent': 'VotreNomApplication/1.0 (votre@email.com)'
}
response = requests.get(geocode_url, headers=headers)
if response.status_code == 200:
data = response.json()
if data:
# Extraire le code du département depuis address (ex: "75000 Paris, Île-de-France, 75, France")
display_name = data[0]['display_name']
address_parts = display_name.split(', ')
department_code = None
# Chercher le code du département (ex: "75" dans "75000 Paris, Île-de-France, 75, France")
for part in address_parts:
if part.strip().isdigit() and len(part.strip()) == 2: # Code département = 2 chiffres
department_code = part.strip()
break
return {
'lat': data[0]['lat'],
'lng': data[0]['lon'],
'display_name': data[0]['display_name'],
'department_code': department_code # <-- Ajoutez cette ligne
}
else:
return {'error': 'Aucun résultat trouvé'}, 404
else:
return {'error': f"Erreur {response.status_code} : {response.text}"}, response.status_code
except Exception as e:
return {'error': f"Une erreur s'est produite : {str(e)}"}, 500
@app.route('/scores')
def get_scores():
department_code = request.args.get('department_code')
insee_code = request.args.get('insee_code')
# Utilisation de la connexion directe avec un curseur
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
# Construction de la requête SQL de base
query = """
SELECT insee_code, city_id, district_id, score
FROM loc_scores
"""
# Ajout des conditions si les paramètres sont présents
conditions = []
params = []
if department_code:
conditions.append("insee_code IN (SELECT insee_code From loc_cities WHERE department_code = %s)")
params.append(department_code)
if insee_code:
conditions.append("insee_code = %s")
params.append(insee_code)
# Ajout des conditions à la requête
if conditions:
query += " WHERE " + " AND ".join(conditions)
# Exécution de la requête avec les paramètres
cursor.execute(query, tuple(params))
scores = cursor.fetchall()
# Fermeture du curseur et de la connexion
cursor.close()
connection.close()
# Conversion des résultats en liste de dictionnaires
scores_list = [
{
"insee_code": row["insee_code"],
"city_id": row["city_id"],
"district_id": row["district_id"],
"score": float(row["score"]) if row["score"] is not None else None
}
for row in scores
]
return jsonify(scores_list)
@app.route('/api/city-data', methods=['GET'])
def city_data():
insee_code = request.args.get('insee_code')
if not insee_code:
return jsonify({"success": False, "message": "Le paramètre insee_code est requis."}), 400
try:
# Récupérer les données de la ville
city_data = CityDataManager.get_city_data(insee_code)
if not city_data:
return jsonify({"success": False, "message": "Ville non trouvée."}), 404
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
# Récupérer les districts
query_districts = """
SELECT
d.district_id as id,
d.district_name as name,
d.latitude as lat,
d.longitude as lng,
c.postal_code as postal_code,
c.city_name as city_name
FROM loc_districts d
JOIN loc_cities c ON d.insee_code = c.insee_code
WHERE d.insee_code = %s
"""
cursor.execute(query_districts, (insee_code,))
districts = cursor.fetchall()
# Récupérer les séries temporelles de prix
price_data = get_price_history_and_forecast(insee_code)
# Récupérer les données météo détaillées
weather_query = """
SELECT
forecast_timestamp,
avg_temperature,
humidity,
precipitation,
wind_speed,
solar_radiation
FROM data_weather
WHERE insee_code = %s
ORDER BY forecast_timestamp DESC
LIMIT 30
"""
cursor.execute(weather_query, (insee_code,))
weather_data = cursor.fetchall()
# Récupérer les scores des services
services_query = """
SELECT
alimentation_score,
commerce_score,
services_reparations_score,
sante_bien_etre_score,
loisirs_culture_score,
education_formation_score,
transports_mobilite_score,
services_publics_score,
autres_services_score,
total_score
FROM data_services_scores
WHERE insee_code = %s
ORDER BY data_year DESC
LIMIT 1 -- Récupérer uniquement la dernière année disponible
"""
cursor.execute(services_query, (insee_code,))
services_data = cursor.fetchone() # On récupère une seule ligne (la plus récente)
# Récupérer le taux de délinquance
delinquency_query = """
SELECT delinquency_rate
FROM data_delinquency
WHERE insee_code = %s
ORDER BY data_year DESC
LIMIT 1 -- Récupérer uniquement la dernière année disponible
"""
cursor.execute(delinquency_query, (insee_code,))
delinquency_data = cursor.fetchone() # On récupère une seule ligne (la plus récente)
# Construire la réponse
response = {
"success": True,
"city": {
"name": city_data.get("name", ""),
"insee_code": city_data.get("insee_code", insee_code),
"postal_code": city_data.get("postal_code", ""),
"latitude": city_data.get("latitude", 0.0),
"longitude": city_data.get("longitude", 0.0),
"population": city_data.get("population", 0.0),
"region_name": city_data.get("region_name", ""),
"avg_price_per_sqm_apartment": city_data.get("avg_price_per_sqm_apartment"),
"avg_price_per_sqm_house": city_data.get("avg_price_per_sqm_house"),
"number_of_transactions": city_data.get("number_of_transactions", 0),
},
"districts": districts,
"price_history": price_data,
"weather_data": weather_data,
"services_scores": services_data, # Scores des services
"delinquency_rate": delinquency_data["delinquency_rate"] if delinquency_data else 0 # Taux de délinquance
}
# Si on a des districts, s'assurer que le code postal est cohérent
if districts:
response["city"]["postal_code"] = districts[0]["postal_code"] or city_data.get("postal_code", "")
return jsonify(response)
except Exception as e:
print(f"[ERROR] Erreur lors de la récupération des données pour {insee_code}: {str(e)}")
return jsonify({"success": False, "message": "Erreur serveur lors de la récupération des données."}), 500
finally:
if 'cursor' in locals():
cursor.close()
if 'connection' in locals():
connection.close()
@app.route('/api/real-estate-sites')
def get_real_estate_sites():
postal_code = request.args.get('postal_code', '')
city = request.args.get('city', '')
connection = get_db_connection()
cursor = connection.cursor(dictionary=True)
# 1. Récupérer le code INSEE pour la ville et le code postal donnés
cursor.execute("""
SELECT insee_code
From loc_cities
WHERE postal_code = %s AND LOWER(city_name) = LOWER(%s)
LIMIT 1
""", (postal_code, city))
city_data = cursor.fetchone()
insee_code = city_data['insee_code'] if city_data else None
# 2. Récupérer les sites immobiliers actifs
cursor.execute("""
SELECT website_id, name, url, logo_url, description, is_active, allow_iframe, url_pattern, color
FROM real_estate_websites
WHERE is_active = 1 order by allow_iframe DESC
""")
sites = cursor.fetchall()
cursor.close()
connection.close()
for site in sites:
if site['url_pattern']:
formatted_postal_code = postal_code[:2] if 'seloger.com' in site['url_pattern'] else postal_code
site['final_url'] = site['url_pattern'].format(
postal_code=formatted_postal_code,
city=city.replace(' ', '-').lower(),
insee_code=insee_code # Ajoute le code INSEE au formatage
)
else:
site['final_url'] = site['url']
return jsonify({
'success': True,
'sites': sites,
'params': {
'postal_code': postal_code,
'city': city,
'insee_code': insee_code # Retourne aussi le code INSEE dans la réponse
}
})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)