feat(seo): add IndexNow integration for instant Bing/Yandex URL submission

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>
This commit is contained in:
Deeman
2026-03-10 15:52:45 +01:00
parent 608f16f578
commit fc21c25c82
5 changed files with 127 additions and 7 deletions

View File

@@ -1105,6 +1105,10 @@ async def supplier_new():
category, tier, contact_name, contact_email, contact_role, category, tier, contact_name, contact_email, contact_role,
services_offered, linkedin_url, instagram_url, youtube_url, now), services_offered, linkedin_url, instagram_url, youtube_url, now),
) )
from ..sitemap import invalidate_sitemap_cache, notify_indexnow
invalidate_sitemap_cache()
await notify_indexnow([f"/directory/{slug}"])
await flash(f"Supplier '{name}' created.", "success") await flash(f"Supplier '{name}' created.", "success")
return redirect(url_for("admin.supplier_detail", supplier_id=supplier_id)) return redirect(url_for("admin.supplier_detail", supplier_id=supplier_id))
@@ -2625,20 +2629,28 @@ async def articles_bulk():
) )
if action == "publish": if action == "publish":
affected = await fetch_all(
f"SELECT DISTINCT url_path FROM articles WHERE {where}", tuple(where_params)
)
await execute( await execute(
f"UPDATE articles SET status = 'published', updated_at = ? WHERE {where}", f"UPDATE articles SET status = 'published', updated_at = ? WHERE {where}",
(now, *where_params), (now, *where_params),
) )
from ..sitemap import invalidate_sitemap_cache from ..sitemap import invalidate_sitemap_cache, notify_indexnow
invalidate_sitemap_cache() invalidate_sitemap_cache()
await notify_indexnow([r["url_path"] for r in affected])
elif action == "unpublish": elif action == "unpublish":
affected = await fetch_all(
f"SELECT DISTINCT url_path FROM articles WHERE {where}", tuple(where_params)
)
await execute( await execute(
f"UPDATE articles SET status = 'draft', updated_at = ? WHERE {where}", f"UPDATE articles SET status = 'draft', updated_at = ? WHERE {where}",
(now, *where_params), (now, *where_params),
) )
from ..sitemap import invalidate_sitemap_cache from ..sitemap import invalidate_sitemap_cache, notify_indexnow
invalidate_sitemap_cache() invalidate_sitemap_cache()
await notify_indexnow([r["url_path"] for r in affected])
elif action == "toggle_noindex": elif action == "toggle_noindex":
await execute( await execute(
@@ -2685,16 +2697,26 @@ async def articles_bulk():
f"UPDATE articles SET status = 'published', updated_at = ? WHERE id IN ({placeholders})", f"UPDATE articles SET status = 'published', updated_at = ? WHERE id IN ({placeholders})",
(now, *article_ids), (now, *article_ids),
) )
from ..sitemap import invalidate_sitemap_cache from ..sitemap import invalidate_sitemap_cache, notify_indexnow
invalidate_sitemap_cache() invalidate_sitemap_cache()
affected = await fetch_all(
f"SELECT DISTINCT url_path FROM articles WHERE id IN ({placeholders})",
tuple(article_ids),
)
await notify_indexnow([r["url_path"] for r in affected])
elif action == "unpublish": elif action == "unpublish":
await execute( await execute(
f"UPDATE articles SET status = 'draft', updated_at = ? WHERE id IN ({placeholders})", f"UPDATE articles SET status = 'draft', updated_at = ? WHERE id IN ({placeholders})",
(now, *article_ids), (now, *article_ids),
) )
from ..sitemap import invalidate_sitemap_cache from ..sitemap import invalidate_sitemap_cache, notify_indexnow
invalidate_sitemap_cache() invalidate_sitemap_cache()
affected = await fetch_all(
f"SELECT DISTINCT url_path FROM articles WHERE id IN ({placeholders})",
tuple(article_ids),
)
await notify_indexnow([r["url_path"] for r in affected])
elif action == "toggle_noindex": elif action == "toggle_noindex":
await execute( await execute(
@@ -2808,8 +2830,10 @@ async def article_new():
(url_path, article_slug, title, meta_description, og_image_url, (url_path, article_slug, title, meta_description, og_image_url,
country, region, language, status, pub_dt, seo_head, article_type), country, region, language, status, pub_dt, seo_head, article_type),
) )
from ..sitemap import invalidate_sitemap_cache from ..sitemap import invalidate_sitemap_cache, notify_indexnow
invalidate_sitemap_cache() invalidate_sitemap_cache()
if status == "published":
await notify_indexnow([url_path])
await flash(f"Article '{title}' created.", "success") await flash(f"Article '{title}' created.", "success")
return redirect(url_for("admin.articles")) return redirect(url_for("admin.articles"))
@@ -2881,6 +2905,9 @@ async def article_edit(article_id: int):
(title, url_path, meta_description, og_image_url, (title, url_path, meta_description, og_image_url,
country, region, language, status, pub_dt, seo_head, article_type, now, article_id), country, region, language, status, pub_dt, seo_head, article_type, now, article_id),
) )
if status == "published":
from ..sitemap import notify_indexnow
await notify_indexnow([url_path])
await flash("Article updated.", "success") await flash("Article updated.", "success")
return redirect(url_for("admin.articles")) return redirect(url_for("admin.articles"))
@@ -2976,8 +3003,11 @@ async def article_publish(article_id: int):
(new_status, now, article_id), (new_status, now, article_id),
) )
from ..sitemap import invalidate_sitemap_cache from ..sitemap import invalidate_sitemap_cache, notify_indexnow
invalidate_sitemap_cache() invalidate_sitemap_cache()
toggled = await fetch_one("SELECT url_path FROM articles WHERE id = ?", (article_id,))
if toggled:
await notify_indexnow([toggled["url_path"]])
if request.headers.get("HX-Request"): if request.headers.get("HX-Request"):
updated = await fetch_one( updated = await fetch_one(

View File

@@ -284,6 +284,12 @@ def create_app() -> Quart:
from .sitemap import sitemap_response from .sitemap import sitemap_response
return await sitemap_response(config.BASE_URL) 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 # Error pages
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------

View File

@@ -72,6 +72,7 @@ class Config:
GSC_SITE_URL: str = os.getenv("GSC_SITE_URL", "") GSC_SITE_URL: str = os.getenv("GSC_SITE_URL", "")
BING_WEBMASTER_API_KEY: str = os.getenv("BING_WEBMASTER_API_KEY", "") BING_WEBMASTER_API_KEY: str = os.getenv("BING_WEBMASTER_API_KEY", "")
BING_SITE_URL: str = os.getenv("BING_SITE_URL", "") BING_SITE_URL: str = os.getenv("BING_SITE_URL", "")
INDEXNOW_KEY: str = os.getenv("INDEXNOW_KEY", "")
RESEND_API_KEY: str = os.getenv("RESEND_API_KEY", "") RESEND_API_KEY: str = os.getenv("RESEND_API_KEY", "")
EMAIL_FROM: str = _env("EMAIL_FROM", "hello@padelnomics.io") EMAIL_FROM: str = _env("EMAIL_FROM", "hello@padelnomics.io")

View File

@@ -0,0 +1,60 @@
"""IndexNow push notifications — instant URL submission to Bing, Yandex, Seznam, Naver.
Fire-and-forget: logs errors but never raises. Content publishing must not fail
because IndexNow is down.
"""
import logging
import httpx
logger = logging.getLogger(__name__)
_INDEXNOW_ENDPOINT = "https://api.indexnow.org/IndexNow"
_BATCH_LIMIT = 10_000 # IndexNow max URLs per request
async def notify_urls(
urls: list[str], host: str, key: str, timeout_seconds: int = 10
) -> bool:
"""POST changed URLs to IndexNow. Returns True on success, False on failure.
Skips silently if key is empty (dev environments).
Batches into chunks of 10,000 URLs per IndexNow spec.
"""
assert 1 <= timeout_seconds <= 60, "timeout_seconds must be 1-60"
if not key:
return False
if not urls:
return True
key_location = f"https://{host}/{key}.txt"
try:
async with httpx.AsyncClient(timeout=timeout_seconds) as client:
for offset in range(0, len(urls), _BATCH_LIMIT):
batch = urls[offset : offset + _BATCH_LIMIT]
response = await client.post(
_INDEXNOW_ENDPOINT,
json={
"host": host,
"key": key,
"keyLocation": key_location,
"urlList": batch,
},
)
# IndexNow returns 200 or 202 on success
if response.status_code not in (200, 202):
logger.warning(
"IndexNow returned %d for %d URLs: %s",
response.status_code,
len(batch),
response.text[:200],
)
return False
logger.info("IndexNow accepted %d URLs", len(batch))
return True
except Exception:
logger.warning("IndexNow notification failed", exc_info=True)
return False

View File

@@ -1,10 +1,14 @@
"""Sitemap generation with in-memory TTL cache and hreflang alternates.""" """Sitemap generation with in-memory TTL cache and hreflang alternates."""
import logging
import time import time
from urllib.parse import urlparse
from quart import Response from quart import Response
from .core import fetch_all from .core import config, fetch_all
logger = logging.getLogger(__name__)
# Process-local cache — valid for the current single-Hypercorn-worker deployment # Process-local cache — valid for the current single-Hypercorn-worker deployment
# (Dockerfile: `--workers 1`). If worker count increases, replace with a # (Dockerfile: `--workers 1`). If worker count increases, replace with a
@@ -128,3 +132,22 @@ async def sitemap_response(base_url: str) -> Response:
content_type="application/xml", content_type="application/xml",
headers={"Cache-Control": f"public, max-age={CACHE_TTL_SECONDS}"}, headers={"Cache-Control": f"public, max-age={CACHE_TTL_SECONDS}"},
) )
async def notify_indexnow(paths: list[str]) -> None:
"""Ping IndexNow with full URLs for the given paths (both lang variants).
Paths should be lang-agnostic (e.g. "/blog/my-article"). Each path gets
expanded to all supported language variants.
"""
if not config.INDEXNOW_KEY or not paths:
return
base = config.BASE_URL.rstrip("/")
urls = []
for path in paths:
for lang in LANGS:
urls.append(f"{base}/{lang}{path}")
host = urlparse(config.BASE_URL).hostname or "padelnomics.io"
from .seo._indexnow import notify_urls
await notify_urls(urls, host=host, key=config.INDEXNOW_KEY)