main.py

"""SimilarWeb Traffic Data Scraper — competitor analysis demo.

Use case: pull traffic stats, rankings, and traffic-source breakdowns for five
SaaS competitors in the project management niche, then save to JSON and CSV.

Run:
    pip install -r requirements.txt
    cp .env.example .env  # add your APIFY_API_TOKEN
    python main.py

See more examples in the examples/ directory.

Actor: https://apify.com/pro100chok/similarweb-scraper?utm_source=github_pro100&utm_medium=readme&utm_campaign=similarweb-traffic-data-python
"""
from __future__ import annotations

import csv
import json
import os
import sys
from pathlib import Path

from apify_client import ApifyClient
from apify_client.errors import ApifyApiError
from dotenv import load_dotenv

# Project-management SaaS competitors — change this list for your own niche
COMPETITORS = [
    "asana.com",
    "monday.com",
    "trello.com",
    "clickup.com",
    "notion.so",
]

ACTOR_ID = "pro100chok/similarweb-scraper"


def main() -> int:
    load_dotenv()
    token = os.environ.get("APIFY_API_TOKEN")
    if not token:
        sys.exit("Set APIFY_API_TOKEN in your environment or .env file.")

    client = ApifyClient(token)

    run_input = {
        "searchType": "similarweb",
        "domains": COMPETITORS,
        "proxyConfiguration": {"useApifyProxy": True, "apifyProxyGroups": ["RESIDENTIAL"]},
    }

    print(f"Starting actor {ACTOR_ID} for {len(COMPETITORS)} domains...")
    try:
        run = client.actor(ACTOR_ID).call(run_input=run_input, timeout_secs=600)
    except ApifyApiError as exc:
        sys.exit(f"Actor call failed: {exc}")

    if run is None or run.get("status") != "SUCCEEDED":
        sys.exit(f"Run did not succeed: {run.get('status') if run else 'no run'}")

    dataset_id = run["defaultDatasetId"]
    items = list(client.dataset(dataset_id).iterate_items())
    if not items:
        sys.exit("Run succeeded but returned no items — check the actor logs in Apify console.")

    out_dir = Path(__file__).parent
    json_path = out_dir / "output.json"
    csv_path = out_dir / "output.csv"

    json_path.write_text(json.dumps(items, indent=2, ensure_ascii=False), encoding="utf-8")

    flat_rows = [flatten(item) for item in items]
    fieldnames = sorted({k for row in flat_rows for k in row.keys()})
    with csv_path.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(flat_rows)

    print(f"\nSaved {len(items)} items")
    print(f"  JSON: {json_path}")
    print(f"  CSV : {csv_path}")
    summarize(items)
    return 0


def flatten(item: dict, prefix: str = "") -> dict:
    """Flatten one level of nested dicts so the result is CSV-friendly."""
    flat: dict = {}
    for key, value in item.items():
        full_key = f"{prefix}{key}"
        if isinstance(value, dict):
            for inner_k, inner_v in value.items():
                flat[f"{full_key}.{inner_k}"] = stringify(inner_v)
        else:
            flat[full_key] = stringify(value)
    return flat


def stringify(value: object) -> str:
    if isinstance(value, (dict, list)):
        return json.dumps(value, ensure_ascii=False)
    return "" if value is None else str(value)


def summarize(items: list[dict]) -> None:
    print("\nTraffic summary (SimilarWeb mode):")
    for item in items:
        name = item.get("SiteName", "?")
        visits = item.get("Engagments", {}).get("VisitsFormatted", "n/a")
        rank = item.get("GlobalRank", {}).get("Rank", "n/a")
        print(f"  {name:<20} visits/mo: {visits:<10} global rank: {rank}")


if __name__ == "__main__":
    raise SystemExit(main())