feat(affiliate): core affiliate module — product lookup, click logging, stats
Pure async functions: get_product(), get_products_by_category(), log_click(), hash_ip() with daily-rotating GDPR salt, get_click_stats() with SQL aggregation. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
224
web/src/padelnomics/affiliate.py
Normal file
224
web/src/padelnomics/affiliate.py
Normal file
@@ -0,0 +1,224 @@
|
||||
"""
|
||||
Affiliate product catalog: product lookup, click logging, and stats queries.
|
||||
|
||||
All functions are plain async procedures — no classes, no state.
|
||||
|
||||
Design decisions:
|
||||
- IP hashing uses a daily salt (date + SECRET_KEY[:16]) for GDPR compliance.
|
||||
Rotating salt prevents re-identification across days without storing PII.
|
||||
- Products are fetched by (slug, language) with a graceful fallback to any
|
||||
language, so DE cards appear in EN articles rather than nothing.
|
||||
- Stats are computed entirely in SQL — no Python aggregation.
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
from datetime import date
|
||||
|
||||
from .core import config, execute, fetch_all, fetch_one
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
VALID_CATEGORIES = ("racket", "ball", "shoe", "bag", "grip", "eyewear", "accessory")
|
||||
VALID_STATUSES = ("draft", "active", "archived")
|
||||
|
||||
|
||||
def hash_ip(ip_address: str) -> str:
|
||||
"""SHA256(ip + YYYY-MM-DD + SECRET_KEY[:16]) with daily salt rotation."""
|
||||
assert ip_address, "ip_address must not be empty"
|
||||
today = date.today().isoformat()
|
||||
salt = config.SECRET_KEY[:16]
|
||||
raw = f"{ip_address}:{today}:{salt}"
|
||||
return hashlib.sha256(raw.encode()).hexdigest()
|
||||
|
||||
|
||||
async def get_product(slug: str, language: str = "de") -> dict | None:
|
||||
"""Return active product by slug+language, falling back to any language."""
|
||||
assert slug, "slug must not be empty"
|
||||
row = await fetch_one(
|
||||
"SELECT * FROM affiliate_products"
|
||||
" WHERE slug = ? AND language = ? AND status = 'active'",
|
||||
(slug, language),
|
||||
)
|
||||
if row:
|
||||
return _parse_product(row)
|
||||
# Graceful fallback: show any language rather than nothing
|
||||
row = await fetch_one(
|
||||
"SELECT * FROM affiliate_products"
|
||||
" WHERE slug = ? AND status = 'active' LIMIT 1",
|
||||
(slug,),
|
||||
)
|
||||
return _parse_product(row) if row else None
|
||||
|
||||
|
||||
async def get_products_by_category(category: str, language: str = "de") -> list[dict]:
|
||||
"""Return active products in category sorted by sort_order, with fallback."""
|
||||
assert category in VALID_CATEGORIES, f"unknown category: {category}"
|
||||
rows = await fetch_all(
|
||||
"SELECT * FROM affiliate_products"
|
||||
" WHERE category = ? AND language = ? AND status = 'active'"
|
||||
" ORDER BY sort_order ASC, id ASC",
|
||||
(category, language),
|
||||
)
|
||||
if rows:
|
||||
return [_parse_product(r) for r in rows]
|
||||
# Fallback: any language for this category
|
||||
rows = await fetch_all(
|
||||
"SELECT * FROM affiliate_products"
|
||||
" WHERE category = ? AND status = 'active'"
|
||||
" ORDER BY sort_order ASC, id ASC",
|
||||
(category,),
|
||||
)
|
||||
return [_parse_product(r) for r in rows]
|
||||
|
||||
|
||||
async def get_all_products(
|
||||
status: str | None = None,
|
||||
retailer: str | None = None,
|
||||
) -> list[dict]:
|
||||
"""Admin listing — all products, optionally filtered by status and/or retailer."""
|
||||
conditions = []
|
||||
params: list = []
|
||||
if status:
|
||||
assert status in VALID_STATUSES, f"unknown status: {status}"
|
||||
conditions.append("status = ?")
|
||||
params.append(status)
|
||||
if retailer:
|
||||
conditions.append("retailer = ?")
|
||||
params.append(retailer)
|
||||
|
||||
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
||||
rows = await fetch_all(
|
||||
f"SELECT * FROM affiliate_products {where} ORDER BY sort_order ASC, id ASC",
|
||||
tuple(params),
|
||||
)
|
||||
return [_parse_product(r) for r in rows]
|
||||
|
||||
|
||||
async def get_click_counts() -> dict[int, int]:
|
||||
"""Return {product_id: click_count} for all products (used in admin list)."""
|
||||
rows = await fetch_all(
|
||||
"SELECT product_id, COUNT(*) AS cnt FROM affiliate_clicks GROUP BY product_id"
|
||||
)
|
||||
return {r["product_id"]: r["cnt"] for r in rows}
|
||||
|
||||
|
||||
async def log_click(
|
||||
product_id: int,
|
||||
ip_address: str,
|
||||
article_slug: str | None,
|
||||
referrer: str | None,
|
||||
) -> None:
|
||||
"""Insert a click event. Hashes IP for GDPR compliance."""
|
||||
assert product_id > 0, "product_id must be positive"
|
||||
assert ip_address, "ip_address must not be empty"
|
||||
ip = hash_ip(ip_address)
|
||||
await execute(
|
||||
"INSERT INTO affiliate_clicks (product_id, article_slug, referrer, ip_hash)"
|
||||
" VALUES (?, ?, ?, ?)",
|
||||
(product_id, article_slug, referrer, ip),
|
||||
)
|
||||
|
||||
|
||||
async def get_click_stats(days_count: int = 30) -> dict:
|
||||
"""Compute click statistics over the last N days, entirely in SQL."""
|
||||
assert 1 <= days_count <= 365, f"days must be 1-365, got {days_count}"
|
||||
|
||||
# Total clicks in window
|
||||
total_row = await fetch_one(
|
||||
"SELECT COUNT(*) AS cnt FROM affiliate_clicks"
|
||||
" WHERE clicked_at >= datetime('now', ?)",
|
||||
(f"-{days_count} days",),
|
||||
)
|
||||
total = total_row["cnt"] if total_row else 0
|
||||
|
||||
# Active product count
|
||||
product_counts = await fetch_one(
|
||||
"SELECT"
|
||||
" SUM(CASE WHEN status='active' THEN 1 ELSE 0 END) AS active_count,"
|
||||
" SUM(CASE WHEN status='draft' THEN 1 ELSE 0 END) AS draft_count"
|
||||
" FROM affiliate_products"
|
||||
)
|
||||
|
||||
# Top products by clicks
|
||||
top_products = await fetch_all(
|
||||
"SELECT p.id, p.name, p.slug, p.retailer, COUNT(c.id) AS click_count"
|
||||
" FROM affiliate_products p"
|
||||
" LEFT JOIN affiliate_clicks c"
|
||||
" ON c.product_id = p.id"
|
||||
" AND c.clicked_at >= datetime('now', ?)"
|
||||
" GROUP BY p.id"
|
||||
" ORDER BY click_count DESC"
|
||||
" LIMIT 10",
|
||||
(f"-{days_count} days",),
|
||||
)
|
||||
|
||||
# Top articles by clicks
|
||||
top_articles = await fetch_all(
|
||||
"SELECT article_slug, COUNT(*) AS click_count"
|
||||
" FROM affiliate_clicks"
|
||||
" WHERE clicked_at >= datetime('now', ?)"
|
||||
" AND article_slug IS NOT NULL"
|
||||
" GROUP BY article_slug"
|
||||
" ORDER BY click_count DESC"
|
||||
" LIMIT 10",
|
||||
(f"-{days_count} days",),
|
||||
)
|
||||
|
||||
# Clicks by retailer
|
||||
by_retailer = await fetch_all(
|
||||
"SELECT p.retailer, COUNT(c.id) AS click_count"
|
||||
" FROM affiliate_products p"
|
||||
" LEFT JOIN affiliate_clicks c"
|
||||
" ON c.product_id = p.id"
|
||||
" AND c.clicked_at >= datetime('now', ?)"
|
||||
" GROUP BY p.retailer"
|
||||
" ORDER BY click_count DESC",
|
||||
(f"-{days_count} days",),
|
||||
)
|
||||
|
||||
# Daily click counts for bar chart
|
||||
daily = await fetch_all(
|
||||
"SELECT date(clicked_at) AS day, COUNT(*) AS click_count"
|
||||
" FROM affiliate_clicks"
|
||||
" WHERE clicked_at >= datetime('now', ?)"
|
||||
" GROUP BY day"
|
||||
" ORDER BY day ASC",
|
||||
(f"-{days_count} days",),
|
||||
)
|
||||
|
||||
# Normalize daily to percentage heights for CSS bar chart
|
||||
max_daily = max((r["click_count"] for r in daily), default=1)
|
||||
daily_bars = [
|
||||
{"day": r["day"], "click_count": r["click_count"],
|
||||
"pct": round(r["click_count"] / max_daily * 100)}
|
||||
for r in daily
|
||||
]
|
||||
|
||||
return {
|
||||
"total_clicks": total,
|
||||
"active_products": product_counts["active_count"] if product_counts else 0,
|
||||
"draft_products": product_counts["draft_count"] if product_counts else 0,
|
||||
"top_products": [dict(r) for r in top_products],
|
||||
"top_articles": [dict(r) for r in top_articles],
|
||||
"by_retailer": [dict(r) for r in by_retailer],
|
||||
"daily_bars": daily_bars,
|
||||
"days": days_count,
|
||||
}
|
||||
|
||||
|
||||
async def get_distinct_retailers() -> list[str]:
|
||||
"""Return sorted list of distinct retailer names for form datalist."""
|
||||
rows = await fetch_all(
|
||||
"SELECT DISTINCT retailer FROM affiliate_products"
|
||||
" WHERE retailer != '' ORDER BY retailer"
|
||||
)
|
||||
return [r["retailer"] for r in rows]
|
||||
|
||||
|
||||
def _parse_product(row) -> dict:
|
||||
"""Convert aiosqlite Row to plain dict, parsing JSON pros/cons arrays."""
|
||||
d = dict(row)
|
||||
d["pros"] = json.loads(d.get("pros") or "[]")
|
||||
d["cons"] = json.loads(d.get("cons") or "[]")
|
||||
return d
|
||||
Reference in New Issue
Block a user