# app.py — Neovia Licensing API (Flask/WSGI, cPanel-friendly) import os, hashlib, time, sqlite3 from datetime import datetime, timedelta, timezone from flask import Flask, request, jsonify, Response from dotenv import load_dotenv load_dotenv() # charge /home2/oopromo/api.neovia.club/.env dans os.environ DB_PATH = os.environ.get("DB_PATH", "saas.db") API_SECRET = os.environ.get("API_SECRET", "changeme") STRIPE_WEBHOOK_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET", "whsec_xxx") app = Flask(__name__) # ---------- DB helpers ---------- def db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): conn = db(); cur = conn.cursor() cur.executescript(""" CREATE TABLE IF NOT EXISTS licenses( key TEXT PRIMARY KEY, plan TEXT NOT NULL, account_bound INTEGER, machine_hash TEXT, expires_at TEXT NOT NULL, active INTEGER NOT NULL DEFAULT 1, meta TEXT ); CREATE TABLE IF NOT EXISTS events( ts TEXT, event TEXT, license_key TEXT, note TEXT ); """) conn.commit(); conn.close() def now_utc(): return datetime.now(timezone.utc) def add_event(ev, lic, note=""): conn=db(); cur=conn.cursor() cur.execute("INSERT INTO events(ts,event,license_key,note) VALUES(?,?,?,?)", (now_utc().isoformat(), ev, lic, note)) conn.commit(); conn.close() init_db() # ---------- Simple landing & health (for cPanel checks) ---------- @app.route("/", methods=["GET"]) def root(): html = "Neovia API

Neovia API is running ✅ (Flask)

" return Response(html, mimetype="text/html; charset=UTF-8") @app.route("/health", methods=["GET"]) def health(): return jsonify(ok=True, time=now_utc().isoformat()) # ---------- Endpoints ---------- @app.route("/admin/create_license", methods=["POST"]) def create_license(): if request.headers.get("x-api-key") != API_SECRET: return jsonify(error="Unauthorized"), 401 body = request.get_json(force=True, silent=True) or {} plan = body.get("plan") if plan not in ("trial","monthly","annual"): return jsonify(error="invalid plan"), 400 days = body.get("days") if days is None: days = {"trial":7, "monthly":31, "annual":366}[plan] key = hashlib.sha256(f"{time.time_ns()}".encode()).hexdigest()[:24].upper() expires_at = (now_utc() + timedelta(days=int(days))).isoformat() account = body.get("account") machine_hash = body.get("machine_hash") conn=db(); cur=conn.cursor() cur.execute( "INSERT INTO licenses(key,plan,account_bound,machine_hash,expires_at,active,meta) VALUES (?,?,?,?,?,?,?)", (key, plan, account, machine_hash, expires_at, 1, "{}"), ) conn.commit(); conn.close() add_event("create", key, f"plan={plan},days={days}") return jsonify(key=key, expires_at=expires_at) @app.route("/license/activate", methods=["POST"]) def activate(): body = request.get_json(force=True, silent=True) or {} key = body.get("key") account = body.get("account") if not key or account is None: return jsonify(error="key and account required"), 400 conn=db(); cur=conn.cursor() row = cur.execute("SELECT * FROM licenses WHERE key=?", (key,)).fetchone() if not row: return jsonify(error="license not found"), 404 if not row["active"]: return jsonify(error="inactive license"), 403 if datetime.fromisoformat(row["expires_at"]) < now_utc(): return jsonify(error="expired"), 403 if row["account_bound"] and row["account_bound"] != account: return jsonify(error="bound to different account"), 403 if not row["account_bound"]: cur.execute("UPDATE licenses SET account_bound=? WHERE key=?", (account, key)) conn.commit() add_event("activate", key, f"acct={account};ver={body.get('version')}") return jsonify( ok=True, expires_at=row["expires_at"], config={ "risk_pct_total": 2.0, "alloc": [0.25, 0.25, 0.25, 0.25], "runner_rr": 15.0, "features": {"move_to_be_on_tp1": True, "market_entry": True}, }, ) @app.route("/license/heartbeat", methods=["POST"]) def heartbeat(): body = request.get_json(force=True, silent=True) or {} key = body.get("key") account = body.get("account") if not key or account is None: return jsonify(error="key and account required"), 400 conn=db(); cur=conn.cursor() row = cur.execute("SELECT * FROM licenses WHERE key=?", (key,)).fetchone() if not row: return jsonify(error="license not found"), 404 if not row["active"]: return jsonify(error="inactive"), 403 if datetime.fromisoformat(row["expires_at"]) < now_utc(): return jsonify(error="expired"), 403 if row["account_bound"] and row["account_bound"] != account: return jsonify(error="wrong account"), 403 add_event("hb", key, f"acct={account};eq={body.get('equity')}") return jsonify(ok=True, server_time=now_utc().isoformat()) @app.route("/admin/deactivate", methods=["POST"]) def deactivate(): if request.headers.get("x-api-key") != API_SECRET: return jsonify(error="Unauthorized"), 401 key = request.args.get("key") or request.form.get("key") if not key: return jsonify(error="key required"), 400 conn=db(); cur=conn.cursor() cur.execute("UPDATE licenses SET active=0 WHERE key=?", (key,)) conn.commit(); conn.close() add_event("deactivate", key) return jsonify(ok=True) @app.route("/stripe/webhook", methods=["POST"]) def stripe_webhook(): payload = request.get_data(as_text=True)[:200] add_event("stripe_webhook", "-", payload) # TODO: verify signature STRIPE_WEBHOOK_SECRET return jsonify(received=True) # --- DEBUG: expose env var --- @app.get("/debug/env") def debug_env(): import os return {"API_SECRET_seen_by_server": os.environ.get("API_SECRET")} # Passenger WSGI looks for 'application' by default if you import this module directly application = app # --- AJOUTS TOP FICHIER (import + helpers) --- import json, re SIGNAL_CSV_PATH = os.environ.get("SIGNAL_CSV_PATH", "data/signal.csv") TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "") # bot de @BotFather TELEGRAM_WEBHOOK_SECRET = os.environ.get("TELEGRAM_WEBHOOK_SECRET", "hooksecret123") # chemin secret os.makedirs(os.path.dirname(SIGNAL_CSV_PATH), exist_ok=True) NUM = r"([-+]?\d+(?:[.,]\d+)?)" def parse_signal_text(text: str): """Retourne dict ou None si non reconnu""" t = text.replace(",", ".") m1 = re.search(r"\b(Buy|Sell)\b\s+([A-Za-z0-9\-\._]+)", t, re.I) if not m1: return None direction = "Buy" if m1.group(1).lower()=="buy" else "Sell" base_symbol = m1.group(2).upper() # mapping broker symbol_map = {"XAUUSD": os.environ.get("XAUUSD_MAP", "XAUUSD-ECN")} symbol = symbol_map.get(base_symbol, base_symbol) m_zone = re.search(r"(Zone d[’']entr[eé]e|Entry zone)\s*:\s*"+NUM+r"\s*-\s*"+NUM, t, re.I) if not m_zone: return None e1 = float(m_zone.group(2)); e2 = float(m_zone.group(3)) m_tp1 = re.search(r"TP1\s*:\s*"+NUM, t, re.I) m_tp2 = re.search(r"TP2\s*:\s*"+NUM, t, re.I) m_tp3 = re.search(r"TP3\s*:\s*"+NUM, t, re.I) m_sl = re.search(r"\bSL\b\s*:\s*"+NUM, t, re.I) if not (m_tp1 and m_tp2 and m_tp3 and m_sl): return None tp1=float(m_tp1.group(1)); tp2=float(m_tp2.group(1)); tp3=float(m_tp3.group(1)); sl=float(m_sl.group(1)) return { "direction": direction, "symbol": symbol, "entry1": e1, "entry2": e2, "sl": sl, "tp1": tp1, "tp2": tp2, "tp3": tp3 } def fmt_num(x: float): s = f"{x:.5f}".rstrip("0").rstrip(".") return s def write_signal_csv(sig: dict): line = f"{sig['direction']};{sig['symbol']};{fmt_num(sig['entry1'])}-{fmt_num(sig['entry2'])};{fmt_num(sig['sl'])};{fmt_num(sig['tp1'])};{fmt_num(sig['tp2'])};{fmt_num(sig['tp3'])}\n" with open(SIGNAL_CSV_PATH, "w", encoding="utf-8") as f: f.write(line) add_event("signal_csv", "-", line.strip()) return line