google-ads-ver-2/scripts/reports/fetch_shoper_data.py

#!/usr/bin/env python3
"""
Pobiera dane e-commerce ze Shoper API (zamówienia, przychody, AOV).

Użycie:
    python scripts/reports/fetch_shoper_data.py --domain innsi.pl --month 2026-02
"""

import argparse
import json
import os
import sys
import io
from collections import defaultdict
from pathlib import Path

if __name__ == "__main__":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")

import requests
ROOT = Path(__file__).parent.parent.parent
sys.path.insert(0, str(ROOT))
from src.gads_v2.config import load_env

load_env(ROOT / ".env")


def _shoper_auth(domain):
    """Authenticate to Shoper API, return (base_url, headers)."""
    base = os.environ[f"SHOPER_API_URL_{domain}"].rstrip("/")
    login = os.environ[f"SHOPER_API_LOGIN_{domain}"]
    password = os.environ[f"SHOPER_API_PASSWORD_{domain}"]

    r = requests.post(f"{base}/auth", auth=(login, password), timeout=15)
    r.raise_for_status()
    token = r.json()["access_token"]
    return base, {"Authorization": f"Bearer {token}"}


def _collect_orders_for_month(base, headers, target_month):
    """Paginate orders (newest first) and collect all for target_month."""
    orders = []
    page = 1
    found = False

    while page < 100:
        r = requests.get(
            f"{base}/orders?limit=50&page={page}&order=date+desc",
            headers=headers, timeout=20,
        )
        r.raise_for_status()
        data = r.json()
        page_orders = data.get("list", [])
        if not page_orders:
            break

        for o in page_orders:
            month = o["date"][:7]
            if month == target_month:
                found = True
                orders.append(o)
            elif found and month < target_month:
                return orders

        page += 1

    return orders


def fetch_shoper_ecommerce(domain, month, prev_month):
    """Fetch Shoper e-commerce data for month and previous month.

    Returns dict compatible with ga4.ecommerce structure.
    """
    base, headers = _shoper_auth(domain)

    # Collect orders for both months in one pass
    current_orders = []
    prev_orders = []
    page = 1
    found_current = False
    found_prev = False
    passed_prev = False

    while page < 200 and not passed_prev:
        r = requests.get(
            f"{base}/orders?limit=50&page={page}&order=date+desc",
            headers=headers, timeout=20,
        )
        r.raise_for_status()
        data = r.json()
        page_orders = data.get("list", [])
        if not page_orders:
            break

        for o in page_orders:
            m = o["date"][:7]
            if m == month:
                found_current = True
                current_orders.append(o)
            elif m == prev_month:
                found_prev = True
                prev_orders.append(o)
            elif found_prev and m < prev_month:
                passed_prev = True
                break

        page += 1

    def _summarize(orders):
        if not orders:
            return {"transactions": 0, "revenue": 0.0, "aov": 0.0}
        total = sum(float(o["sum"]) for o in orders)
        count = len(orders)
        return {
            "transactions": count,
            "revenue": round(total, 2),
            "aov": round(total / count, 2),
        }

    current = _summarize(current_orders)
    previous = _summarize(prev_orders)

    # Daily breakdown
    daily_map = defaultdict(lambda: {"revenue": 0.0, "transactions": 0})
    for o in current_orders:
        day = o["date"][:10]
        daily_map[day]["revenue"] += float(o["sum"])
        daily_map[day]["transactions"] += 1

    daily = []
    for day in sorted(daily_map.keys()):
        daily.append({
            "date": day,
            "revenue": round(daily_map[day]["revenue"], 2),
            "transactions": daily_map[day]["transactions"],
        })

    # MoM change
    def _pct(cur, prev):
        if prev == 0:
            return 100.0 if cur > 0 else 0.0
        return round(((cur - prev) / prev) * 100, 1)

    mom = {
        "transactions_pct": _pct(current["transactions"], previous["transactions"]),
        "revenue_pct": _pct(current["revenue"], previous["revenue"]),
        "aov_pct": _pct(current["aov"], previous["aov"]),
    }

    return {
        "source": "shoper",
        "current": current,
        "previous": previous,
        "mom_change": mom,
        "daily": daily,
    }


def main():
    parser = argparse.ArgumentParser(description="Pobierz dane e-commerce ze Shoper")
    parser.add_argument("--domain", required=True)
    parser.add_argument("--month", required=True, help="YYYY-MM")
    args = parser.parse_args()

    # Calculate prev month
    y, m = map(int, args.month.split("-"))
    if m == 1:
        prev = f"{y-1}-12"
    else:
        prev = f"{y}-{m-1:02d}"

    data = fetch_shoper_ecommerce(args.domain, args.month, prev)
    print(json.dumps(data, indent=2, ensure_ascii=False))
    cur = data["current"]
    print(f"\n{args.month}: {cur['transactions']} zamówień, {cur['revenue']:.2f} PLN, AOV {cur['aov']:.2f} PLN")


if __name__ == "__main__":
    main()