прочее В поиске софта для чека трафика

LongNight · 31.05.2025

Сабж, буду признателен если кто подскажет чем можно проверять листы ~1-2к на трафик

Fooll Zero · 01.06.2025

LongNight сказал(а):

Сабж, буду признателен если кто подскажет чем можно проверять листы ~1-2к на трафик

Реализация не плохая и рабочая

LongNight · 02.06.2025

Fooll Zero сказал(а):

Реализация не плохая и рабочая

не совсем то что я ищу, много лишней даты + гугл таблицы задействованы, пробовал переписать, но там проще с нуля делать

так что все еще ищу =(

agregator · 03.06.2025

ТС извиняюсь, дабы не плодить темы тоже ищу такой софт

seven1up · 04.06.2025

agregator LongNight

Палю софт: https://netpeaksoftware.com/checker

Там есть сейчас триал-акция 3 дня теста фулл-софта бесплатно.

В софт просто импортируешь список список сайтов, на выходе получаешь такое:

Есть подгрузка прокси и 200 потоков. 10к сайтов чекает за 5 минут. Можно хоть миллион настроек добавить и по индексации, и DNS и серпстат короче все что душе угодно даже можно кидать админки что-бы отснять Title, короче все что только можно в софте есть

Если будет желание отблагодарить копеечкой за годный совет и софт буду только рад

Или может нужно будет оплатить лицензию которая копейки стоит и у вас нету забугор карты - тоже обращайтесь помогу с радостью пишите в личные сообщения или тг в профиле

Так же могу просто делать триал-доступы с передачей на ваш акк вам вообще ничего делать не нужно будет написали дали мыло я кинул триалку

_C3FaRiR_ · 16.07.2025

seven1up сказал(а):

agregator LongNight

Палю софт: https://netpeaksoftware.com/checker

Там есть сейчас триал-акция 3 дня теста фулл-софта бесплатно.

В софт просто импортируешь список список сайтов, на выходе получаешь такое:

Посмотреть вложение 108012

Есть подгрузка прокси и 200 потоков. 10к сайтов чекает за 5 минут. Можно хоть миллион настроек добавить и по индексации, и DNS и серпстат короче все что душе угодно даже можно кидать админки что-бы отснять Title, короче все что только можно в софте есть

Если будет желание отблагодарить копеечкой за годный совет и софт буду только рад
Или может нужно будет оплатить лицензию которая копейки стоит и у вас нету забугор карты - тоже обращайтесь помогу с радостью пишите в личные сообщения или тг в профиле

Так же могу просто делать триал-доступы с передачей на ваш акк вам вообще ничего делать не нужно будет написали дали мыло я кинул триалку

Братанчики, кто-то пробывал это ? воркает?

Fooll Zero · 16.07.2025

_C3FaRiR_ сказал(а):

Братанчики, кто-то пробывал это ? воркает?

Пробуй, должно работать

Python:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
similarweb_async_curlcffi.py
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Асинхронная массовая проверка доменов через скрытый API SimilarWeb
с браузерной маскировкой TLS/JA3 (curl_cffi / curl-impersonate).

Особенности
-----------
• Асинхронность (curl_cffi.requests.AsyncSession).
• Имитация настоящего Chrome TLS-/JA3-отпечатка (`impersonate="chrome"`).
• Подробное логирование (DEBUG-уровень): попытки, прокси, статус, время.
• Ротация прокси + удаление заблокированных.
• CSV-отчёт с «плоскими» метриками (до 10 колонок) без полного JSON.

Зависимости
-----------
    pip install curl_cffi fake-useragent validators aiofiles
"""

import asyncio
import csv
import json
import logging
import random
import sys
import time
from collections import deque
from pathlib import Path
from typing import Deque, List, Optional, Tuple

import requests
import validators
from curl_cffi.requests import AsyncSession  # type: ignore
from fake_useragent import UserAgent

# ========================================================================= #
#                                  CONFIG                                   #
# ========================================================================= #
CONFIG = {
    "DOMAINS_FILE": "domains.txt",  # список доменов
    # источник прокси (txt, один на строку)
    "PROXY_SOURCE_URL": "",
    "PROXIES_FILE": "",           # если нужен локальный список (приоритет выше)
    "USE_PROXIES": True,
    "CONCURRENT_REQUESTS": 32,     # параллельных задач
    "MAX_RETRIES": 8,             # попыток на домен
    "REQUEST_TIMEOUT": 20,
    "DELAY_BETWEEN_REQUESTS": (0.5, 1.5),
    "OUTPUT_CSV": "results.csv",
    "LOG_FILE": "scraper.log",
}
# ========================================================================= #

# ----------------------------- LOGGING ----------------------------------- #
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(CONFIG["LOG_FILE"], encoding="utf-8"),
        logging.StreamHandler(sys.stdout),
    ],
)
log = logging.getLogger(__name__)

# ----------------------------- USER-AGENT -------------------------------- #
try:
    DEFAULT_UA = UserAgent(cache=True).random
except Exception:
    DEFAULT_UA = (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/138.0.0.0 Safari/537.36"
    )

# ----------------------------- PROXY POOL -------------------------------- #
def load_proxies() -> Deque[str]:
    proxies: List[str] = []
    pf = Path(CONFIG["PROXIES_FILE"])
    if pf.is_file():
        log.debug("Читаем прокси из локального файла %s", pf)
        proxies = [l.strip() for l in pf.read_text().splitlines() if l.strip()]
    elif CONFIG["PROXY_SOURCE_URL"]:
        log.debug("Скачиваем прокси по ссылке %s", CONFIG["PROXY_SOURCE_URL"])
        try:
            r = requests.get(CONFIG["PROXY_SOURCE_URL"], timeout=15)
            r.raise_for_status()
            proxies = [l.strip() for l in r.text.splitlines() if l.strip()]
        except Exception as e:
            log.warning("Не удалось получить прокси: %s", e)
    if not proxies:
        log.warning("Прокси не найдены — будем работать без них")
    else:
        random.shuffle(proxies)
    return deque(proxies)


class ProxyPool:
    def __init__(self, q: Deque[str]):
        self.q = q
        self.lock = asyncio.Lock()

    async def get(self) -> Optional[str]:
        async with self.lock:
            if not self.q:
                return None
            proxy = self.q.popleft()
            self.q.append(proxy)  # кольцевая очередь
            return proxy

    async def remove(self, proxy: str):
        async with self.lock:
            if proxy in self.q:
                self.q.remove(proxy)
                log.debug("Прокси %s удалён из пула", proxy)

    async def size(self) -> int:
        async with self.lock:
            return len(self.q)


def proxy_to_dict(p: str) -> Tuple[dict, str]:
    parts = p.split(":")
    if len(parts) == 2:
        url = f"http://{parts[0]}:{parts[1]}"
        return {"http": url, "https": url}, url
    if len(parts) == 4:
        host, port, user, pwd = parts
        url = f"http://{user}:{pwd}@{host}:{port}"
        return {"http": url, "https": url}, url
    return {"http": p, "https": p}, p


# ------------------------------- HELPERS ---------------------------------- #
HEADERS = {
    "Accept": "application/json",
    "Referer": "https://data.similarweb.com/",
    "Origin": "https://data.similarweb.com",
    "Accept-Language": "en-US,en;q=0.9",
}


def flatten_metrics(data: dict) -> dict:
    """Выдёргиваем ключевые поля (≤10 колонок)."""
    eng = data.get("Engagments", {}) or {}
    g_rank = data.get("GlobalRank", {}) or {}
    c_rank = data.get("CountryRank", {}) or {}
    cat_rank = data.get("CategoryRank", {}) or {}

    return {
        "visits": eng.get("Visits", 0),
        "page_per_visit": eng.get("PagePerVisit", 0),
        "bounce_rate": eng.get("BounceRate", 0),
        "global_rank": g_rank.get("Rank", ""),
        "country_rank": c_rank.get("Rank", ""),
        "category_rank": cat_rank.get("Rank", ""),
        "is_small": data.get("IsSmall", ""),
        "snapshot": data.get("SnapshotDate", ""),
    }


# ------------------------------ CORE -------------------------------------- #
async def fetch_domain(
    session: AsyncSession,
    domain: str,
    proxy_pool: ProxyPool,
    attempt: int,
) -> Tuple[str, Optional[dict], str]:
    proxy_raw = ""
    proxies = None
    if CONFIG["USE_PROXIES"]:
        proxy_raw = await proxy_pool.get() or ""
        if proxy_raw:
            proxies, _ = proxy_to_dict(proxy_raw)

    t0 = time.perf_counter()
    try:
        resp = await session.get(
            f"https://data.similarweb.com/api/v1/data?domain={domain}",
            headers={**HEADERS, "User-Agent": DEFAULT_UA},
            proxies=proxies,
            timeout=CONFIG["REQUEST_TIMEOUT"],
            impersonate="chrome",
        )
        elapsed = time.perf_counter() - t0
        log.debug(
            "Domain=%s attempt=%d status=%s proxy=%s t=%.2fs",
            domain,
            attempt,
            resp.status_code,
            proxy_raw or "NONE",
            elapsed,
        )

        if resp.status_code == 200:
            return "OK", resp.json(), ""
        # Ошибки
        err = f"HTTP {resp.status_code}"
        if resp.status_code in (403, 429) and proxy_raw:
            await proxy_pool.remove(proxy_raw)
        return "ERROR", None, err

    except Exception as e:
        log.debug("Exception domain=%s proxy=%s err=%s", domain, proxy_raw, e)
        if proxy_raw:
            await proxy_pool.remove(proxy_raw)
        return "ERROR", None, str(e)


async def process(
    domain: str,
    session: AsyncSession,
    semaphore: asyncio.Semaphore,
    proxy_pool: ProxyPool,
) -> dict:
    async with semaphore:
        for attempt in range(1, CONFIG["MAX_RETRIES"] + 1):
            status, data, err = await fetch_domain(session, domain, proxy_pool, attempt)
            if status == "OK":
                flat = flatten_metrics(data or {})
                return {
                    "domain": domain,
                    "status": "OK",
                    **flat,
                    "error": "",
                }
            await asyncio.sleep(random.uniform(*CONFIG["DELAY_BETWEEN_REQUESTS"]))
        return {
            "domain": domain,
            "status": "ERROR",
            "visits": "",
            "page_per_visit": "",
            "bounce_rate": "",
            "global_rank": "",
            "country_rank": "",
            "category_rank": "",
            "is_small": "",
            "snapshot": "",
            "error": err,
        }


def read_domains(path: str) -> List[str]:
    lines = Path(path).read_text(encoding="utf-8").splitlines()
    valid = [d.strip().lower() for d in lines if validators.domain(d.strip())]
    invalid = set(lines) - set(valid)
    if invalid:
        log.warning("Некорректные записи пропущены: %s", ", ".join(invalid))
    return valid


async def main() -> None:
    domains = read_domains(CONFIG["DOMAINS_FILE"])
    if not domains:
        log.error("Список доменов пуст — завершение.")
        return

    proxy_pool = ProxyPool(load_proxies())
    sem = asyncio.Semaphore(CONFIG["CONCURRENT_REQUESTS"])

    async with AsyncSession() as session:
        tasks = [process(d, session, sem, proxy_pool) for d in domains]
        results = await asyncio.gather(*tasks)

    # ------------------- CSV ------------------- #
    fields = [
        "domain",
        "status",
        "visits",
        "page_per_visit",
        "bounce_rate",
        "global_rank",
        "country_rank",
        "category_rank",
        "is_small",
        "snapshot",
        "error",
    ]
    with open(CONFIG["OUTPUT_CSV"], "w", newline="", encoding="utf-8") as f:
        # extrasaction='ignore' на всякий случай, если появятся лишние ключи
        writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(results)

    ok = sum(1 for r in results if r["status"] == "OK")
    log.info("Готово: %d/%d успешных. → %s", ok, len(results), CONFIG["OUTPUT_CSV"])


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        log.info("Остановка пользователем")

LongNight · 17.07.2025

_C3FaRiR_ сказал(а):

Братанчики, кто-то пробывал это ? воркает?

Работает, хорошая штука. Но то что скинул фул зеро тоже подходит + фри

прочее В поиске софта для чека трафика

LongNight

Премиум

Fooll Zero

(L1) cache

LongNight

Премиум

agregator

CD-диск

seven1up

ripper

_C3FaRiR_

Братуха

Fooll Zero

(L1) cache

LongNight

Премиум