feat: rebuild bare metal experience
This commit is contained in:
Gemini AI
342
serve10000.py
Normal file
342
serve10000.py
Normal file
@@ -0,0 +1,342 @@
|
||||
from flask import Flask, jsonify, send_from_directory
|
||||
from pathlib import Path
|
||||
import time
|
||||
import json
|
||||
from urllib.parse import urljoin
|
||||
import re
|
||||
|
||||
try:
|
||||
import requests
|
||||
except ImportError: # pragma: no cover
|
||||
requests = None
|
||||
|
||||
try:
|
||||
from bs4 import BeautifulSoup
|
||||
except ImportError: # pragma: no cover
|
||||
BeautifulSoup = None
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent
|
||||
CUSTOM_OFFERS_PATH = BASE_DIR / "custom_offers.json"
|
||||
REMOTE_URL = "https://www.dedicatednodes.io/solana-nodes/"
|
||||
CACHE_TTL = 5 * 60
|
||||
OPTION_CACHE_TTL = 5 * 60
|
||||
HEADERS = {
|
||||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
||||
}
|
||||
|
||||
_cache = {"ts": 0, "data": None}
|
||||
_remote_cache = {"ts": 0, "soup": None}
|
||||
_instant_cache = {"ts": 0, "data": None}
|
||||
_option_cache = {}
|
||||
PRICE_PATTERN = re.compile(r"([+-])?\s*([€$£]|[A-Za-z]{1,3})?\s*([0-9.,]+)")
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
|
||||
def normalize_text(value: str) -> str:
|
||||
if not value:
|
||||
return ""
|
||||
return " ".join(value.replace("\xa0", " ").split())
|
||||
|
||||
|
||||
def text_from_element(el):
|
||||
if not el:
|
||||
return ""
|
||||
return normalize_text(el.get_text(separator=" ", strip=True))
|
||||
|
||||
|
||||
def find_card_root(node):
|
||||
current = node
|
||||
depth = 0
|
||||
while current and depth < 6:
|
||||
current = current.parent
|
||||
depth += 1
|
||||
if not current:
|
||||
break
|
||||
classes = current.get("class") or []
|
||||
if isinstance(classes, str):
|
||||
classes = classes.split()
|
||||
if "card" in classes:
|
||||
return current
|
||||
return node.parent
|
||||
|
||||
|
||||
def collect_features(card):
|
||||
features = []
|
||||
for li in card.select("li"):
|
||||
component = li.select_one(".component")
|
||||
if component:
|
||||
label = normalize_text(component.get_text())
|
||||
value = text_from_element(li.select_one(".component-value"))
|
||||
description = text_from_element(li.select_one(".component-description"))
|
||||
parts = [part for part in (value, description) if part]
|
||||
if label and parts:
|
||||
features.append("|".join([label] + parts))
|
||||
continue
|
||||
price_band = li.select_one(".price-band")
|
||||
if price_band:
|
||||
label_text = normalize_text(price_band.select_one(".label").get_text()) if price_band.select_one(".label") else "Starting from"
|
||||
amount_text = normalize_text(price_band.select_one(".amount").get_text()) if price_band.select_one(".amount") else ""
|
||||
per_text = normalize_text(price_band.select_one(".per").get_text()) if price_band.select_one(".per") else ""
|
||||
parts = [part for part in (amount_text, per_text) if part]
|
||||
if label_text and parts:
|
||||
features.append("|".join([label_text] + parts))
|
||||
return features
|
||||
|
||||
|
||||
def parse_custom_offers(soup):
|
||||
if not BeautifulSoup or not soup:
|
||||
return []
|
||||
offers = []
|
||||
seen_links = set()
|
||||
for anchor in soup.find_all("a", string=lambda text: text and "configure now" in text.lower()):
|
||||
href = anchor.get("href")
|
||||
if not href or href in seen_links:
|
||||
continue
|
||||
seen_links.add(href)
|
||||
card = find_card_root(anchor)
|
||||
if not card:
|
||||
continue
|
||||
title_el = card.find(class_="card-title") or card.find(["h2", "h3", "h4"])
|
||||
title = normalize_text(title_el.get_text()) if title_el else "Custom Server"
|
||||
features = collect_features(card)
|
||||
if not features:
|
||||
continue
|
||||
link = urljoin(REMOTE_URL, href)
|
||||
offer = {
|
||||
"title": title,
|
||||
"link": link,
|
||||
"features": features,
|
||||
}
|
||||
options = parse_product_options(link)
|
||||
if options:
|
||||
offer["options"] = options
|
||||
offers.append(offer)
|
||||
return offers
|
||||
|
||||
|
||||
def fetch_remote_html():
|
||||
if not requests:
|
||||
return None
|
||||
try:
|
||||
response = requests.get(REMOTE_URL, headers=HEADERS, timeout=10)
|
||||
response.raise_for_status()
|
||||
return response.text
|
||||
except Exception as exc: # pragma: no cover
|
||||
print(f"[serve10000] Unable to reach {REMOTE_URL}: {exc}")
|
||||
return None
|
||||
|
||||
|
||||
def load_fallback_offers():
|
||||
if not CUSTOM_OFFERS_PATH.exists():
|
||||
return []
|
||||
try:
|
||||
data = json.loads(CUSTOM_OFFERS_PATH.read_text(encoding="utf-8"))
|
||||
if isinstance(data, list):
|
||||
return data
|
||||
except json.JSONDecodeError:
|
||||
print("[serve10000] Invalid fallback JSON, returning empty list")
|
||||
return []
|
||||
|
||||
|
||||
def fetch_remote_soup():
|
||||
if not BeautifulSoup:
|
||||
return None
|
||||
now = time.time()
|
||||
cached = _remote_cache.get("soup")
|
||||
if cached and now - _remote_cache["ts"] < CACHE_TTL:
|
||||
return cached
|
||||
html = fetch_remote_html()
|
||||
if not html:
|
||||
return None
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
_remote_cache["soup"] = soup
|
||||
_remote_cache["ts"] = now
|
||||
return soup
|
||||
|
||||
|
||||
def parse_instant_offers(soup):
|
||||
results = []
|
||||
if not soup:
|
||||
return results
|
||||
table = soup.select_one("#instant-servers-table tbody")
|
||||
if not table:
|
||||
return results
|
||||
for row in table.select("tr"):
|
||||
cells = row.select("td")
|
||||
if len(cells) < 8:
|
||||
continue
|
||||
cpu = normalize_text(cells[0].get_text())
|
||||
location = normalize_text(cells[1].get_text())
|
||||
cores = normalize_text(cells[2].get_text())
|
||||
memory = normalize_text(cells[3].get_text())
|
||||
storage = normalize_text(cells[4].get_text())
|
||||
network = normalize_text(cells[5].get_text())
|
||||
bandwidth = normalize_text(cells[6].get_text())
|
||||
price_text = normalize_text(cells[7].get_text())
|
||||
price_delta, currency_symbol = parse_price_delta(price_text)
|
||||
order_link = ""
|
||||
action = row.select_one(".action-cell a")
|
||||
if action:
|
||||
order_link = urljoin(REMOTE_URL, action.get("href", ""))
|
||||
results.append(
|
||||
{
|
||||
"cpu": cpu,
|
||||
"location": location,
|
||||
"cores": cores,
|
||||
"memory": memory,
|
||||
"storage": storage,
|
||||
"network": network,
|
||||
"bandwidth": bandwidth,
|
||||
"priceText": price_text,
|
||||
"price": price_delta,
|
||||
"currencySymbol": currency_symbol or "€",
|
||||
"orderUrl": order_link,
|
||||
}
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
def get_instant_offers():
|
||||
now = time.time()
|
||||
if _instant_cache["data"] and now - _instant_cache["ts"] < CACHE_TTL:
|
||||
return _instant_cache["data"]
|
||||
soup = fetch_remote_soup()
|
||||
offers = parse_instant_offers(soup)
|
||||
_instant_cache["data"] = offers
|
||||
_instant_cache["ts"] = now
|
||||
return offers
|
||||
|
||||
|
||||
def format_price_delta(delta: float, symbol: str) -> str:
|
||||
if not delta:
|
||||
return ""
|
||||
sign = "+" if delta > 0 else "-"
|
||||
abs_value = abs(delta)
|
||||
formatted = f"{abs_value:,.2f}".replace(",", "_").replace(".", ",").replace("_", ".")
|
||||
return f"{sign} {symbol}{formatted}".strip()
|
||||
|
||||
|
||||
def strip_price_from_label(text: str) -> str:
|
||||
cleaned = re.sub(
|
||||
r"\s*[+-]?\s*([€$£]|[A-Za-z]{1,3}|ƒ,ª)?\s*[0-9.,]+\s*(EUR|USD|GBP)?",
|
||||
"",
|
||||
text,
|
||||
flags=re.IGNORECASE,
|
||||
).strip()
|
||||
return cleaned or text
|
||||
|
||||
|
||||
def parse_price_delta(text: str) -> tuple[float, str]:
|
||||
match = PRICE_PATTERN.search(text)
|
||||
if not match:
|
||||
return 0.0, ""
|
||||
sign = match.group(1)
|
||||
symbol = match.group(2) or ""
|
||||
raw_amount = match.group(3).replace(" ", "")
|
||||
if raw_amount.count(",") and raw_amount.count("."):
|
||||
normalized = raw_amount.replace(".", "").replace(",", ".")
|
||||
elif "," in raw_amount:
|
||||
normalized = raw_amount.replace(",", ".")
|
||||
else:
|
||||
normalized = raw_amount
|
||||
try:
|
||||
amount = float(normalized)
|
||||
except ValueError:
|
||||
amount = 0.0
|
||||
if sign == "-":
|
||||
amount = -amount
|
||||
return amount, symbol
|
||||
|
||||
|
||||
def parse_product_options(url: str):
|
||||
now = time.time()
|
||||
cached = _option_cache.get(url)
|
||||
if cached and now - cached["ts"] < OPTION_CACHE_TTL:
|
||||
return cached["data"]
|
||||
if not requests or not BeautifulSoup:
|
||||
return cached["data"] if cached else []
|
||||
try:
|
||||
response = requests.get(url, headers=HEADERS, timeout=10)
|
||||
response.raise_for_status()
|
||||
except Exception as exc:
|
||||
print(f"[serve10000] Failed to fetch options from {url}: {exc}")
|
||||
return cached["data"] if cached else []
|
||||
soup = BeautifulSoup(response.text, "html.parser")
|
||||
selects = []
|
||||
for select in soup.select('select[name^="configoption"]'):
|
||||
label_text = ""
|
||||
select_id = select.get("id")
|
||||
if select_id:
|
||||
label_el = soup.find("label", attrs={"for": select_id})
|
||||
if label_el:
|
||||
label_text = normalize_text(label_el.get_text())
|
||||
if not label_text:
|
||||
label_text = normalize_text(
|
||||
select.get("data-option-name")
|
||||
or select.get("data-custom-option-name")
|
||||
or select.get("name")
|
||||
)
|
||||
choices = []
|
||||
for opt in select.select("option"):
|
||||
option_text = normalize_text(opt.get_text())
|
||||
delta, currency = parse_price_delta(option_text)
|
||||
choices.append(
|
||||
{
|
||||
"value": opt.get("value"),
|
||||
"label": strip_price_from_label(option_text),
|
||||
"rawLabel": option_text,
|
||||
"priceDelta": delta,
|
||||
"priceSymbol": currency,
|
||||
"priceDisplay": format_price_delta(delta, currency or ""),
|
||||
}
|
||||
)
|
||||
selects.append(
|
||||
{
|
||||
"name": select.get("name"),
|
||||
"label": label_text,
|
||||
"choices": choices,
|
||||
}
|
||||
)
|
||||
_option_cache[url] = {"ts": now, "data": selects}
|
||||
return selects
|
||||
|
||||
|
||||
def get_custom_offers():
|
||||
now = time.time()
|
||||
if _cache["data"] and now - _cache["ts"] < CACHE_TTL:
|
||||
return _cache["data"]
|
||||
soup = fetch_remote_soup()
|
||||
offers = parse_custom_offers(soup) if soup else []
|
||||
if not offers:
|
||||
offers = load_fallback_offers()
|
||||
_cache["data"] = offers
|
||||
_cache["ts"] = now
|
||||
return offers
|
||||
|
||||
|
||||
@app.route("/")
|
||||
@app.route("/new-baremetal.html")
|
||||
def serve_baremetal():
|
||||
return send_from_directory(BASE_DIR, "new-baremetal.html")
|
||||
|
||||
|
||||
@app.route("/custom_offers.json")
|
||||
def custom_offers_file():
|
||||
return send_from_directory(BASE_DIR, "custom_offers.json")
|
||||
|
||||
|
||||
@app.route("/api/custom-offers")
|
||||
def custom_offers_api():
|
||||
return jsonify({"offers": get_custom_offers()})
|
||||
|
||||
|
||||
@app.route("/api/instant-offers")
|
||||
def instant_offers_api():
|
||||
return jsonify({"offers": get_instant_offers()})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Starting DedicatedNodes mock site at http://127.0.0.1:10000/new-baremetal.html")
|
||||
app.run(host="127.0.0.1", port=10000)
|
||||
Reference in New Issue
Block a user