feat: affiliates programs — AffiliateProgram, registry, eBay EPN + Amazon Associates builders

circuitforge_core/affiliates/programs.py

@@ -0,0 +1,103 @@
+"""Affiliate program definitions and URL builders.
+
+Each ``AffiliateProgram`` knows how to append its affiliate parameters to a
+plain product URL.  Built-in programs (eBay EPN, Amazon Associates) are
+registered at module import time.  Products can register additional programs
+with ``register_program()``.
+
+Affiliate IDs are read from environment variables at call time so they pick
+up values set after process startup (useful in tests).
+"""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Callable
+from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
+
+
+@dataclass(frozen=True)
+class AffiliateProgram:
+    """One affiliate program and its URL building logic.
+
+    Attributes:
+        name:         Human-readable program name.
+        retailer_key: Matches the ``retailer=`` argument in ``wrap_url()``.
+        env_var:      Environment variable holding CF's affiliate ID.
+        build_url:    ``(plain_url, affiliate_id) -> affiliate_url`` callable.
+    """
+
+    name: str
+    retailer_key: str
+    env_var: str
+    build_url: Callable[[str, str], str]
+
+    def cf_affiliate_id(self) -> str | None:
+        """Return CF's configured affiliate ID, or None if the env var is unset/blank."""
+        val = os.environ.get(self.env_var, "").strip()
+        return val or None
+
+
+# ---------------------------------------------------------------------------
+# URL builders
+# ---------------------------------------------------------------------------
+
+def _build_ebay_url(url: str, affiliate_id: str) -> str:
+    """Append eBay Partner Network parameters to a listing URL."""
+    sep = "&" if "?" in url else "?"
+    params = urlencode({
+        "mkcid": "1",
+        "mkrid": "711-53200-19255-0",
+        "siteid": "0",
+        "campid": affiliate_id,
+        "toolid": "10001",
+        "mkevt": "1",
+    })
+    return f"{url}{sep}{params}"
+
+
+def _build_amazon_url(url: str, affiliate_id: str) -> str:
+    """Merge an Amazon Associates tag into a product URL's query string."""
+    parsed = urlparse(url)
+    qs = parse_qs(parsed.query, keep_blank_values=True)
+    qs["tag"] = [affiliate_id]
+    new_query = urlencode({k: v[0] for k, v in qs.items()})
+    return urlunparse(parsed._replace(query=new_query))
+
+
+# ---------------------------------------------------------------------------
+# Registry
+# ---------------------------------------------------------------------------
+
+_REGISTRY: dict[str, AffiliateProgram] = {}
+
+
+def register_program(program: AffiliateProgram) -> None:
+    """Register an affiliate program (overwrites any existing entry for the same key)."""
+    _REGISTRY[program.retailer_key] = program
+
+
+def get_program(retailer_key: str) -> AffiliateProgram | None:
+    """Return the registered program for *retailer_key*, or None."""
+    return _REGISTRY.get(retailer_key)
+
+
+def registered_keys() -> list[str]:
+    """Return all currently registered retailer keys."""
+    return list(_REGISTRY.keys())
+
+
+# Register built-ins
+register_program(AffiliateProgram(
+    name="eBay Partner Network",
+    retailer_key="ebay",
+    env_var="EBAY_AFFILIATE_CAMPAIGN_ID",
+    build_url=_build_ebay_url,
+))
+
+register_program(AffiliateProgram(
+    name="Amazon Associates",
+    retailer_key="amazon",
+    env_var="AMAZON_ASSOCIATES_TAG",
+    build_url=_build_amazon_url,
+))

tests/test_affiliates/test_programs.py

@@ -0,0 +1,99 @@
+"""Tests for affiliate program registry and URL builders."""
+import pytest
+from circuitforge_core.affiliates.programs import (
+    AffiliateProgram,
+    get_program,
+    register_program,
+    registered_keys,
+    _build_ebay_url,
+    _build_amazon_url,
+)
+
+
+class TestAffiliateProgram:
+    def test_cf_affiliate_id_returns_env_value(self, monkeypatch):
+        monkeypatch.setenv("TEST_AFF_ID", "my-id")
+        prog = AffiliateProgram(
+            name="Test", retailer_key="test",
+            env_var="TEST_AFF_ID", build_url=lambda u, i: u
+        )
+        assert prog.cf_affiliate_id() == "my-id"
+
+    def test_cf_affiliate_id_returns_none_when_unset(self, monkeypatch):
+        monkeypatch.delenv("TEST_AFF_ID", raising=False)
+        prog = AffiliateProgram(
+            name="Test", retailer_key="test",
+            env_var="TEST_AFF_ID", build_url=lambda u, i: u
+        )
+        assert prog.cf_affiliate_id() is None
+
+    def test_cf_affiliate_id_returns_none_when_blank(self, monkeypatch):
+        monkeypatch.setenv("TEST_AFF_ID", "   ")
+        prog = AffiliateProgram(
+            name="Test", retailer_key="test",
+            env_var="TEST_AFF_ID", build_url=lambda u, i: u
+        )
+        assert prog.cf_affiliate_id() is None
+
+
+class TestRegistry:
+    def test_builtin_ebay_registered(self):
+        assert get_program("ebay") is not None
+        assert get_program("ebay").name == "eBay Partner Network"
+
+    def test_builtin_amazon_registered(self):
+        assert get_program("amazon") is not None
+        assert get_program("amazon").name == "Amazon Associates"
+
+    def test_unknown_key_returns_none(self):
+        assert get_program("not_a_retailer") is None
+
+    def test_register_custom_program(self):
+        prog = AffiliateProgram(
+            name="Custom Shop", retailer_key="customshop",
+            env_var="CUSTOM_ID", build_url=lambda u, i: f"{u}?ref={i}"
+        )
+        register_program(prog)
+        assert get_program("customshop") is prog
+        assert "customshop" in registered_keys()
+
+    def test_register_overwrites_existing(self):
+        prog1 = AffiliateProgram("A", "overwrite_test", "X", lambda u, i: u)
+        prog2 = AffiliateProgram("B", "overwrite_test", "Y", lambda u, i: u)
+        register_program(prog1)
+        register_program(prog2)
+        assert get_program("overwrite_test").name == "B"
+
+
+class TestEbayUrlBuilder:
+    def test_appends_params_to_plain_url(self):
+        url = _build_ebay_url("https://www.ebay.com/itm/123", "my-campaign")
+        assert "campid=my-campaign" in url
+        assert "mkcid=1" in url
+        assert "mkevt=1" in url
+        assert url.startswith("https://www.ebay.com/itm/123?")
+
+    def test_uses_ampersand_when_query_already_present(self):
+        url = _build_ebay_url("https://www.ebay.com/itm/123?existing=1", "c1")
+        assert url.startswith("https://www.ebay.com/itm/123?existing=1&")
+        assert "campid=c1" in url
+
+    def test_does_not_double_encode(self):
+        url = _build_ebay_url("https://www.ebay.com/itm/123", "camp-id-1")
+        assert "camp-id-1" in url
+
+
+class TestAmazonUrlBuilder:
+    def test_appends_tag_to_plain_url(self):
+        url = _build_amazon_url("https://www.amazon.com/dp/B001234567", "cf-kiwi-20")
+        assert "tag=cf-kiwi-20" in url
+
+    def test_merges_tag_into_existing_query(self):
+        url = _build_amazon_url("https://www.amazon.com/dp/B001234567?ref=sr_1_1", "cf-kiwi-20")
+        assert "tag=cf-kiwi-20" in url
+        assert "ref=sr_1_1" in url
+
+    def test_replaces_existing_tag(self):
+        url = _build_amazon_url("https://www.amazon.com/dp/B001?tag=old-tag-20", "new-tag-20")
+        assert "tag=new-tag-20" in url
+        assert "old-tag-20" not in url