142 lines
4.6 KiB
Python
142 lines
4.6 KiB
Python
"""Serveur local AutoMood : interface web + API + stockage CSV."""
|
|
|
|
import csv
|
|
import os
|
|
import threading
|
|
import webbrowser
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, jsonify, request, send_file
|
|
|
|
import extractor
|
|
import scraper
|
|
|
|
DOSSIER = Path(__file__).parent
|
|
CSV_PATH = DOSSIER / "prospects.csv"
|
|
COLONNES = [
|
|
"Nom du prospect", "Département", "Ville", "Code Postal", "Adresse",
|
|
"Date d'ajout", "Date de contact", "Nom de contact",
|
|
"Téléphone", "Email", "Infos du lieu", "Type", "Lien Facebook",
|
|
]
|
|
|
|
app = Flask(__name__, static_folder="static")
|
|
verrou_scrape = threading.Lock()
|
|
verrou_csv = threading.Lock()
|
|
|
|
|
|
def lire_prospects():
|
|
if not CSV_PATH.exists():
|
|
ecrire_prospects([])
|
|
return []
|
|
with open(CSV_PATH, encoding="utf-8-sig", newline="") as f:
|
|
lecteur = csv.DictReader(f, delimiter=";")
|
|
return [{col: (ligne.get(col) or "") for col in COLONNES} for ligne in lecteur]
|
|
|
|
|
|
def ecrire_prospects(lignes):
|
|
tmp = CSV_PATH.with_suffix(".csv.tmp")
|
|
with open(tmp, "w", encoding="utf-8-sig", newline="") as f:
|
|
ecrivain = csv.DictWriter(f, fieldnames=COLONNES, delimiter=";", extrasaction="ignore")
|
|
ecrivain.writeheader()
|
|
ecrivain.writerows(lignes)
|
|
os.replace(tmp, CSV_PATH)
|
|
|
|
|
|
@app.get("/")
|
|
def accueil():
|
|
return app.send_static_file("index.html")
|
|
|
|
|
|
@app.post("/api/login")
|
|
def api_login():
|
|
if not verrou_scrape.acquire(blocking=False):
|
|
return jsonify({"error": "occupe", "message": "Une autre opération est en cours, patientez."}), 429
|
|
try:
|
|
if scraper.connexion():
|
|
return jsonify({"ok": True, "message": "Connexion Facebook enregistrée."})
|
|
return jsonify({
|
|
"error": "login_timeout",
|
|
"message": "Connexion non détectée dans le temps imparti. Réessayez.",
|
|
}), 408
|
|
except Exception as e:
|
|
return jsonify({"error": "erreur", "message": f"Échec de la connexion : {e}"}), 500
|
|
finally:
|
|
verrou_scrape.release()
|
|
|
|
|
|
@app.post("/api/scrape")
|
|
def api_scrape():
|
|
url = (request.get_json(silent=True) or {}).get("url", "").strip()
|
|
if not url:
|
|
return jsonify({"error": "url_manquante", "message": "Collez un lien Facebook."}), 400
|
|
if not verrou_scrape.acquire(blocking=False):
|
|
return jsonify({"error": "occupe", "message": "Une analyse est déjà en cours, patientez."}), 429
|
|
try:
|
|
resultat = scraper.scrape(url)
|
|
champs = extractor.extraire(resultat["titre"], resultat["texte"], resultat["url"])
|
|
return jsonify(champs)
|
|
except scraper.ErreurScrape as e:
|
|
statuts = {"login_required": 409, "page_introuvable": 404, "url_invalide": 400, "redirection": 422}
|
|
return jsonify({"error": e.code, "message": str(e)}), statuts.get(e.code, 500)
|
|
except Exception as e:
|
|
return jsonify({"error": "erreur", "message": f"Échec de l'analyse : {e}"}), 500
|
|
finally:
|
|
verrou_scrape.release()
|
|
|
|
|
|
@app.get("/api/prospects")
|
|
def api_lister():
|
|
lignes = lire_prospects()
|
|
return jsonify([{"index": i, **ligne} for i, ligne in enumerate(lignes)])
|
|
|
|
|
|
@app.post("/api/prospects")
|
|
def api_ajouter():
|
|
donnees = request.get_json(silent=True) or {}
|
|
ligne = {col: str(donnees.get(col, "")).strip() for col in COLONNES}
|
|
if not ligne["Date d'ajout"]:
|
|
ligne["Date d'ajout"] = date.today().strftime("%d/%m/%Y")
|
|
with verrou_csv:
|
|
lignes = lire_prospects()
|
|
lignes.append(ligne)
|
|
ecrire_prospects(lignes)
|
|
return jsonify({"ok": True, "index": len(lignes) - 1})
|
|
|
|
|
|
@app.put("/api/prospects/<int:idx>")
|
|
def api_modifier(idx):
|
|
donnees = request.get_json(silent=True) or {}
|
|
with verrou_csv:
|
|
lignes = lire_prospects()
|
|
if not 0 <= idx < len(lignes):
|
|
return jsonify({"error": "introuvable", "message": "Ligne inexistante."}), 404
|
|
for col in COLONNES:
|
|
if col in donnees:
|
|
lignes[idx][col] = str(donnees[col]).strip()
|
|
ecrire_prospects(lignes)
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.delete("/api/prospects/<int:idx>")
|
|
def api_supprimer(idx):
|
|
with verrou_csv:
|
|
lignes = lire_prospects()
|
|
if not 0 <= idx < len(lignes):
|
|
return jsonify({"error": "introuvable", "message": "Ligne inexistante."}), 404
|
|
lignes.pop(idx)
|
|
ecrire_prospects(lignes)
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@app.get("/api/export")
|
|
def api_export():
|
|
lire_prospects() # crée le fichier si absent
|
|
return send_file(CSV_PATH, as_attachment=True, download_name="prospects.csv")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
lire_prospects()
|
|
threading.Timer(1.0, webbrowser.open, args=["http://127.0.0.1:5000"]).start()
|
|
app.run(host="127.0.0.1", port=5000)
|