fc21c25c82
Push-notify search engines (Bing, Yandex, Seznam, Naver) when content changes instead of waiting for sitemap crawls. Especially valuable for batch article publishing and supplier directory updates. - Add INDEXNOW_KEY config var and key verification route - New seo/_indexnow.py: async fire-and-forget POST to IndexNow API - notify_indexnow() wrapper in sitemap.py expands paths to all lang variants - Integrated at all article publish/unpublish/edit and supplier create points - Bulk operations batch all URLs into a single IndexNow request - Skips silently when key is empty (dev environments) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
468 lines
17 KiB
Python
468 lines
17 KiB
Python
"""
|
|
Padelnomics - Application factory and entry point.
|
|
"""
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from quart import Quart, Response, abort, g, redirect, render_template, request, session, url_for
|
|
|
|
from .analytics import close_analytics_db, open_analytics_db
|
|
from .core import (
|
|
close_db,
|
|
config,
|
|
get_csrf_token,
|
|
init_db,
|
|
is_flag_enabled,
|
|
setup_logging,
|
|
setup_request_id,
|
|
)
|
|
|
|
setup_logging()
|
|
from .i18n import LANG_BLUEPRINTS, SUPPORTED_LANGS, get_country_name, get_translations # noqa: E402
|
|
|
|
_ASSET_VERSION = str(int(time.time()))
|
|
|
|
|
|
def _detect_lang() -> str:
|
|
"""Detect preferred language from cookie then Accept-Language header."""
|
|
cookie_lang = request.cookies.get("lang", "")
|
|
if cookie_lang in SUPPORTED_LANGS:
|
|
return cookie_lang
|
|
accept = request.headers.get("Accept-Language", "")
|
|
for part in accept.split(","):
|
|
tag = part.split(";")[0].strip()[:2].lower()
|
|
if tag in SUPPORTED_LANGS:
|
|
return tag
|
|
return "en"
|
|
|
|
|
|
def _fmt_currency(n) -> str:
|
|
"""Format currency using request-context symbol and locale style."""
|
|
sym = getattr(g, "currency_sym", "\u20ac")
|
|
eu_style = getattr(g, "currency_eu_style", True)
|
|
n = round(float(n))
|
|
s = f"{abs(n):,}"
|
|
if eu_style:
|
|
s = s.replace(",", ".")
|
|
return f"-{sym}{s}" if n < 0 else f"{sym}{s}"
|
|
|
|
|
|
def _fmt_k(n) -> str:
|
|
"""Short currency: €50K, €1.2M, or full _fmt_currency."""
|
|
sym = getattr(g, "currency_sym", "\u20ac")
|
|
n = float(n)
|
|
if abs(n) >= 1_000_000:
|
|
return f"{sym}{n/1_000_000:.1f}M"
|
|
if abs(n) >= 1_000:
|
|
return f"{sym}{n/1_000:.0f}K"
|
|
return _fmt_currency(n)
|
|
|
|
|
|
def _fmt_pct(n) -> str:
|
|
"""Format fraction as percentage: 0.152 → '15.2%'."""
|
|
return f"{float(n) * 100:.1f}%"
|
|
|
|
|
|
def _fmt_x(n) -> str:
|
|
"""Format as MOIC multiple: 2.30x."""
|
|
return f"{float(n):.2f}x"
|
|
|
|
|
|
def _fmt_n(n) -> str:
|
|
"""Format integer with locale-aware thousands separator: 1.234 or 1,234."""
|
|
eu_style = getattr(g, "currency_eu_style", True)
|
|
s = f"{round(float(n)):,}"
|
|
return s.replace(",", ".") if eu_style else s
|
|
|
|
|
|
def _tformat(s: str, **kwargs) -> str:
|
|
"""Format a translation string with named placeholders.
|
|
|
|
Usage: {{ t.some_key | tformat(count=total, name=supplier.name) }}
|
|
JSON value: "Browse {count}+ suppliers from {name}"
|
|
"""
|
|
return s.format_map(kwargs)
|
|
|
|
|
|
def create_app() -> Quart:
|
|
"""Create and configure the Quart application."""
|
|
|
|
pkg_dir = Path(__file__).parent
|
|
|
|
app = Quart(
|
|
__name__,
|
|
template_folder=str(pkg_dir / "templates"),
|
|
static_folder=str(pkg_dir / "static"),
|
|
)
|
|
|
|
app.secret_key = config.SECRET_KEY
|
|
|
|
# Jinja2 filters
|
|
app.jinja_env.filters["fmt_currency"] = _fmt_currency
|
|
app.jinja_env.filters["fmt_k"] = _fmt_k
|
|
app.jinja_env.filters["fmt_pct"] = _fmt_pct
|
|
app.jinja_env.filters["fmt_x"] = _fmt_x
|
|
app.jinja_env.filters["fmt_n"] = _fmt_n
|
|
app.jinja_env.filters["tformat"] = _tformat # translate with placeholders: {{ t.key | tformat(count=n) }}
|
|
app.jinja_env.filters["country_name"] = get_country_name # {{ article.country | country_name(lang) }}
|
|
app.jinja_env.filters["fromjson"] = json.loads # {{ job.payload | fromjson }}
|
|
|
|
# Session config
|
|
app.config["SESSION_COOKIE_SECURE"] = not config.DEBUG
|
|
app.config["SESSION_COOKIE_HTTPONLY"] = True
|
|
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
|
|
app.config["PERMANENT_SESSION_LIFETIME"] = 60 * 60 * 24 * config.SESSION_LIFETIME_DAYS
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Language URL routing
|
|
# -------------------------------------------------------------------------
|
|
|
|
@app.url_value_preprocessor
|
|
def pull_lang(endpoint, values):
|
|
"""Pop <lang> from URL values and stash in g.lang."""
|
|
if values and "lang" in values:
|
|
g.lang = values.pop("lang")
|
|
|
|
@app.url_defaults
|
|
def inject_lang(endpoint, values):
|
|
"""Auto-inject g.lang into url_for() calls on lang-prefixed blueprints."""
|
|
if endpoint and endpoint.partition(".")[0] in LANG_BLUEPRINTS:
|
|
values.setdefault("lang", g.get("lang", "en"))
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Database lifecycle
|
|
# -------------------------------------------------------------------------
|
|
|
|
@app.before_serving
|
|
async def startup():
|
|
await init_db()
|
|
open_analytics_db()
|
|
|
|
@app.after_serving
|
|
async def shutdown():
|
|
await close_db()
|
|
close_analytics_db()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Per-request hooks
|
|
# -------------------------------------------------------------------------
|
|
|
|
@app.before_request
|
|
async def set_user_geo():
|
|
"""Stash Cloudflare geo headers in g for proximity sorting.
|
|
|
|
Requires nginx: proxy_set_header CF-IPCountry $http_cf_ipcountry;
|
|
proxy_set_header CF-RegionCode $http_cf_regioncode;
|
|
proxy_set_header CF-IPCity $http_cf_ipcity;
|
|
"""
|
|
g.user_country = request.headers.get("CF-IPCountry", "").upper() or ""
|
|
g.user_region = request.headers.get("CF-RegionCode", "") or ""
|
|
g.user_city = request.headers.get("CF-IPCity", "") or ""
|
|
|
|
@app.before_request
|
|
async def validate_lang():
|
|
"""404 unsupported language prefixes (e.g. /fr/terms)."""
|
|
lang = g.get("lang")
|
|
if lang is not None and lang not in SUPPORTED_LANGS:
|
|
abort(404)
|
|
|
|
@app.before_request
|
|
async def load_user():
|
|
"""Load current user + subscription + roles before each request."""
|
|
g.user = None
|
|
g.subscription = None
|
|
user_id = session.get("user_id")
|
|
if user_id:
|
|
from .core import fetch_one as _fetch_one
|
|
row = await _fetch_one(
|
|
"""SELECT u.*,
|
|
bc.provider_customer_id,
|
|
(SELECT GROUP_CONCAT(role) FROM user_roles WHERE user_id = u.id) AS roles_csv,
|
|
s.id AS sub_id, s.plan, s.status AS sub_status,
|
|
s.provider_subscription_id, s.current_period_end
|
|
FROM users u
|
|
LEFT JOIN billing_customers bc ON bc.user_id = u.id
|
|
LEFT JOIN subscriptions s ON s.id = (
|
|
SELECT id FROM subscriptions
|
|
WHERE user_id = u.id
|
|
ORDER BY created_at DESC LIMIT 1
|
|
)
|
|
WHERE u.id = ? AND u.deleted_at IS NULL""",
|
|
(user_id,),
|
|
)
|
|
if row:
|
|
g.user = dict(row)
|
|
g.user["roles"] = row["roles_csv"].split(",") if row["roles_csv"] else []
|
|
if row["sub_id"]:
|
|
g.subscription = {
|
|
"id": row["sub_id"], "plan": row["plan"],
|
|
"status": row["sub_status"],
|
|
"provider_subscription_id": row["provider_subscription_id"],
|
|
"current_period_end": row["current_period_end"],
|
|
}
|
|
|
|
@app.after_request
|
|
async def add_security_headers(response):
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["X-Frame-Options"] = "DENY"
|
|
response.headers["X-XSS-Protection"] = "1; mode=block"
|
|
if not config.DEBUG:
|
|
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
|
|
return response
|
|
|
|
@app.after_request
|
|
async def set_lang_cookie(response):
|
|
"""Persist detected/current language in a long-lived cookie."""
|
|
lang = g.get("lang")
|
|
if lang and lang in SUPPORTED_LANGS:
|
|
current_cookie = request.cookies.get("lang", "")
|
|
if current_cookie != lang:
|
|
response.set_cookie(
|
|
"lang", lang,
|
|
max_age=60 * 60 * 24 * 365, # 1 year
|
|
samesite="Lax",
|
|
secure=not config.DEBUG,
|
|
httponly=False, # JS may read for analytics
|
|
)
|
|
return response
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Template context globals
|
|
# -------------------------------------------------------------------------
|
|
|
|
@app.context_processor
|
|
def inject_globals():
|
|
from .core import utcnow as _utcnow
|
|
lang = g.get("lang") or _detect_lang()
|
|
g.lang = lang # ensure g.lang is always set (e.g. for dashboard/billing routes)
|
|
effective_lang = lang if lang in SUPPORTED_LANGS else "en"
|
|
return {
|
|
"config": config,
|
|
"user": g.get("user"),
|
|
"subscription": g.get("subscription"),
|
|
"is_admin": "admin" in (g.get("user") or {}).get("roles", []),
|
|
"now": _utcnow(),
|
|
"csrf_token": get_csrf_token,
|
|
"ab_variant": getattr(g, "ab_variant", None),
|
|
"ab_tag": getattr(g, "ab_tag", None),
|
|
"user_country": g.get("user_country", ""),
|
|
"user_city": g.get("user_city", ""),
|
|
"lang": effective_lang,
|
|
"t": get_translations(effective_lang),
|
|
"v": _ASSET_VERSION,
|
|
"flag": is_flag_enabled,
|
|
}
|
|
|
|
# -------------------------------------------------------------------------
|
|
# App-level routes (no lang prefix)
|
|
# -------------------------------------------------------------------------
|
|
|
|
# Root: detect language and redirect
|
|
@app.route("/")
|
|
async def root():
|
|
lang = _detect_lang()
|
|
return redirect(url_for("public.landing", lang=lang), 301)
|
|
|
|
# robots.txt must live at root (not under /<lang>)
|
|
@app.route("/robots.txt")
|
|
async def robots_txt():
|
|
base = config.BASE_URL.rstrip("/")
|
|
body = (
|
|
"User-agent: *\n"
|
|
"Disallow: /admin/\n"
|
|
"Disallow: /auth/\n"
|
|
"Disallow: /dashboard/\n"
|
|
"Disallow: /billing/\n"
|
|
"Disallow: /directory/results\n"
|
|
f"Sitemap: {base}/sitemap.xml\n"
|
|
)
|
|
return Response(body, content_type="text/plain")
|
|
|
|
@app.route("/sitemap.xml")
|
|
async def sitemap():
|
|
from .sitemap import sitemap_response
|
|
return await sitemap_response(config.BASE_URL)
|
|
|
|
# IndexNow key verification — only register if key is configured
|
|
if config.INDEXNOW_KEY:
|
|
@app.route(f"/{config.INDEXNOW_KEY}.txt")
|
|
async def indexnow_key():
|
|
return Response(config.INDEXNOW_KEY, content_type="text/plain")
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Error pages
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _error_lang() -> str:
|
|
"""Best-effort language from URL path prefix (no g.lang in error handlers)."""
|
|
path = request.path
|
|
if path.startswith("/de/"):
|
|
return "de"
|
|
return "en"
|
|
|
|
@app.errorhandler(404)
|
|
async def handle_404(error):
|
|
import re
|
|
lang = _error_lang()
|
|
t = get_translations(lang)
|
|
country_slug = None
|
|
country_name = None
|
|
m = re.match(r"^/(?:en|de)/markets/([^/]+)/[^/]+/?$", request.path)
|
|
if m:
|
|
country_slug = m.group(1)
|
|
country_name = country_slug.replace("-", " ").title()
|
|
return await render_template(
|
|
"404.html", lang=lang, t=t, country_slug=country_slug,
|
|
country_name=country_name or "",
|
|
), 404
|
|
|
|
@app.errorhandler(500)
|
|
async def handle_500(error):
|
|
app.logger.exception("Unhandled 500 error: %s", error)
|
|
lang = _error_lang()
|
|
t = get_translations(lang)
|
|
return await render_template("500.html", lang=lang, t=t), 500
|
|
|
|
# Health check
|
|
@app.route("/health")
|
|
async def health():
|
|
from .core import fetch_one
|
|
try:
|
|
await fetch_one("SELECT 1")
|
|
return {"status": "healthy", "db": "ok"}
|
|
except Exception as e:
|
|
return {"status": "unhealthy", "db": str(e)}, 500
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Affiliate click redirect — language-agnostic, no blueprint prefix
|
|
# -------------------------------------------------------------------------
|
|
|
|
@app.route("/go/<slug>")
|
|
async def affiliate_redirect(slug: str):
|
|
"""302 redirect to affiliate URL, logging the click.
|
|
|
|
Uses 302 (not 301) so every hit is tracked — browsers don't cache 302s.
|
|
Extracts article_slug and lang from Referer header best-effort.
|
|
"""
|
|
from .affiliate import build_affiliate_url, get_product, log_click
|
|
from .core import check_rate_limit
|
|
|
|
# Extract lang from Referer path (e.g. /de/blog/... → "de"), default de
|
|
referer = request.headers.get("Referer", "")
|
|
lang = "de"
|
|
article_slug = None
|
|
if referer:
|
|
try:
|
|
from urllib.parse import urlparse
|
|
ref_path = urlparse(referer).path
|
|
parts = ref_path.strip("/").split("/")
|
|
if parts and len(parts[0]) == 2:
|
|
lang = parts[0]
|
|
if len(parts) > 1:
|
|
article_slug = parts[-1] or None
|
|
except Exception:
|
|
pass
|
|
|
|
product = await get_product(slug, lang)
|
|
if not product:
|
|
abort(404)
|
|
|
|
# Assemble URL from program template; falls back to baked affiliate_url
|
|
url = build_affiliate_url(product, product.get("_program"))
|
|
|
|
ip = request.remote_addr or "unknown"
|
|
allowed, _info = await check_rate_limit(f"aff:{ip}", limit=60, window=60)
|
|
if not allowed:
|
|
# Still redirect even if rate-limited; just don't log the click
|
|
return redirect(url, 302)
|
|
|
|
await log_click(product["id"], ip, article_slug, referer or None)
|
|
return redirect(url, 302)
|
|
|
|
# Legacy 301 redirects — bookmarked/cached URLs before lang prefixes existed
|
|
@app.route("/terms")
|
|
async def legacy_terms():
|
|
return redirect("/en/terms", 301)
|
|
|
|
@app.route("/privacy")
|
|
async def legacy_privacy():
|
|
return redirect("/en/privacy", 301)
|
|
|
|
@app.route("/imprint")
|
|
async def legacy_imprint():
|
|
return redirect("/en/imprint", 301)
|
|
|
|
@app.route("/about")
|
|
async def legacy_about():
|
|
return redirect("/en/about", 301)
|
|
|
|
@app.route("/features")
|
|
async def legacy_features():
|
|
return redirect("/en/features", 301)
|
|
|
|
@app.route("/suppliers")
|
|
async def legacy_suppliers():
|
|
return redirect("/en/suppliers", 301)
|
|
|
|
@app.route("/market-score")
|
|
async def legacy_market_score():
|
|
return redirect("/en/padelnomics-score", 301)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Blueprint registration
|
|
# -------------------------------------------------------------------------
|
|
|
|
from .admin.pipeline_routes import bp as pipeline_bp
|
|
from .admin.pseo_routes import bp as pseo_bp
|
|
from .admin.routes import bp as admin_bp
|
|
from .api import bp as api_bp
|
|
from .auth.routes import bp as auth_bp
|
|
from .billing.routes import bp as billing_bp
|
|
from .content.routes import bp as content_bp
|
|
from .dashboard.routes import bp as dashboard_bp
|
|
from .directory.routes import bp as directory_bp
|
|
from .leads.routes import bp as leads_bp
|
|
from .planner.routes import bp as planner_bp
|
|
from .public.routes import bp as public_bp
|
|
from .reports.routes import bp as reports_bp
|
|
from .suppliers.routes import bp as suppliers_bp
|
|
from .webhooks import bp as webhooks_bp
|
|
|
|
# Lang-prefixed blueprints (SEO-relevant, public-facing)
|
|
app.register_blueprint(public_bp, url_prefix="/<lang>")
|
|
app.register_blueprint(planner_bp, url_prefix="/<lang>/planner")
|
|
app.register_blueprint(directory_bp, url_prefix="/<lang>/directory")
|
|
app.register_blueprint(leads_bp, url_prefix="/<lang>/leads")
|
|
app.register_blueprint(suppliers_bp, url_prefix="/<lang>/suppliers")
|
|
app.register_blueprint(reports_bp, url_prefix="/<lang>/reports")
|
|
|
|
# Non-prefixed blueprints (internal / behind auth)
|
|
app.register_blueprint(auth_bp)
|
|
app.register_blueprint(dashboard_bp)
|
|
app.register_blueprint(billing_bp)
|
|
app.register_blueprint(admin_bp)
|
|
app.register_blueprint(pseo_bp)
|
|
app.register_blueprint(pipeline_bp)
|
|
app.register_blueprint(webhooks_bp)
|
|
|
|
# JSON API for interactive maps (no lang prefix)
|
|
app.register_blueprint(api_bp, url_prefix="/api")
|
|
|
|
# Content catch-all LAST — lives under /<lang> too
|
|
app.register_blueprint(content_bp, url_prefix="/<lang>")
|
|
|
|
# Request ID tracking
|
|
setup_request_id(app)
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import os
|
|
port = int(os.environ.get("PORT", 5000))
|
|
app.run(debug=config.DEBUG, port=port)
|