کد ایندکس سریع گوگل + کد پایتون indexing API

خب همونطور که در مرحله قبل به خوبی تونستیم تنظیم api index google رو انجام بدیم میرسیم به این مرحله که به سادگی با استفاده از کد ایندکس سریع گوگل + کد پایتون indexing API ساده می تونیم به راحتی دستور ارسال ایندکس سریع صفحات رو به گوگل بفرستیم فقط در نظر داشته باشید که هر دفعه بیشتر از 200 ارسال نفرستید که فایروال هاست شما ای پی رو بلاک نکنه.

مراحل رو به صورت کامل تو کلیپ زیر توضیح دادم

مرحله اول دسترسی به کد پایتون ایندکس سریع گوگل

به سادگی از طریق لینک زیر بدون اینکه نیاز به دانش برنامه نویسی یا برنامه های اجرا و شبیه سازی پایتون داشته باشید با خود سرویس گوگل به کد مورد نظر دسترسی پیدا کنید.

لینک اجرا و نمایش کد ایندکس سریع گوگل

خود کد رو هم اینجا قرار میدم که خواستید بتونید استفاده کنید ولی بهترین راه استفاده از لینک بالا هست چون خود گوگل به صورت انلاین کد پایتون شما رو اجرا میکنه و نیاز به ابزار خاصی ندارید.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Fast Indexer - Batch URL submission tool for Google Indexing API

Optimized version with:
  - Multiple Service Account support (round-robin) to increase total quota
  - Rate limiting (control requests per second/day)
  - Retry with Exponential Backoff for transient errors (429 / 5xx)
  - URL validation
  - Full CSV output report (success/failure + error details)
  - Non-interactive mode via argparse (good for cron / automation)
  - Interactive mode (prompts step by step if arguments are missing)
  - Progress bar (tqdm)

Usage:
    python fast_indexer.py --sa-dir ./service_accounts --csv urls.csv --type PUBLISH --output report.csv

    Or with a single Service Account:
    python fast_indexer.py --sa-file account1.json --csv urls.csv --type DELETE

    Or just run with no arguments and answer the prompts:
    python fast_indexer.py
"""

import argparse
import csv
import glob
import os
import sys
import time
import random
import itertools
from urllib.parse import urlparse
from datetime import datetime

import requests
from google.oauth2 import service_account
from google.auth.transport.requests import AuthorizedSession

try:
    from tqdm import tqdm
except ImportError:
    # Fallback if tqdm is not installed, so the script doesn't crash
    def tqdm(iterable, **kwargs):
        return iterable

API_URL = "https://indexing.googleapis.com/v3/urlNotifications:publish"
SCOPES = ["https://www.googleapis.com/auth/indexing"]

# Each Service Account usually gets a 200 requests/day quota by default from Google
DEFAULT_DAILY_QUOTA_PER_ACCOUNT = 200
DEFAULT_MIN_DELAY_SECONDS = 1.0  # minimum delay between consecutive requests on the same account
MAX_RETRIES = 5
BACKOFF_BASE = 2  # base for exponential backoff (seconds)


class ServiceAccountPool:
    """
    Manages multiple Service Accounts in round-robin fashion.
    When one account's quota is exhausted, moves on to the next one.
    """

    def __init__(self, sa_paths, daily_quota=DEFAULT_DAILY_QUOTA_PER_ACCOUNT):
        if not sa_paths:
            raise ValueError("No valid Service Account file was found.")

        self.accounts = []
        for path in sa_paths:
            try:
                credentials = service_account.Credentials.from_service_account_file(
                    path, scopes=SCOPES
                )
                session = AuthorizedSession(credentials)
                self.accounts.append({
                    "path": path,
                    "session": session,
                    "used_today": 0,
                    "quota": daily_quota,
                    "last_request_time": 0.0,
                })
            except Exception as e:
                print(f"WARNING: failed to load Service Account '{path}': {e}")

        if not self.accounts:
            raise ValueError("No valid Service Account could be loaded.")

        self._cycle = itertools.cycle(range(len(self.accounts)))
        print(f"Loaded {len(self.accounts)} Service Account(s) successfully. "
              f"Estimated total daily quota: {len(self.accounts) * daily_quota} requests")

    def get_next_available(self):
        """
        Returns an account whose daily quota is not yet exhausted (round-robin).
        Returns None if every account's quota is exhausted.
        """
        for _ in range(len(self.accounts)):
            idx = next(self._cycle)
            acc = self.accounts[idx]
            if acc["used_today"] < acc["quota"]:
                return acc
        return None

    def mark_used(self, account):
        account["used_today"] += 1


def is_valid_url(url):
    """Simple URL validity check"""
    try:
        result = urlparse(url.strip())
        return all([result.scheme in ("http", "https"), result.netloc])
    except Exception:
        return False


def load_service_account_paths(args):
    """
    Builds the list of Service Account JSON file paths from --sa-file or --sa-dir.
    """
    paths = []

    if args.sa_file:
        for p in args.sa_file:
            if not os.path.exists(p):
                print(f"WARNING: file '{p}' not found, skipping.")
                continue
            paths.append(p)

    if args.sa_dir:
        if not os.path.isdir(args.sa_dir):
            raise FileNotFoundError(f"Folder '{args.sa_dir}' not found.")
        json_files = sorted(glob.glob(os.path.join(args.sa_dir, "*.json")))
        paths.extend(json_files)

    # de-duplicate while preserving order
    seen = set()
    unique_paths = []
    for p in paths:
        if p not in seen:
            seen.add(p)
            unique_paths.append(p)

    return unique_paths


def load_urls_from_csv(csv_file_path, encoding="utf-8"):
    """Reads and validates URLs from a CSV file (first column of each row)"""
    if not os.path.exists(csv_file_path):
        raise FileNotFoundError(f"CSV file not found at '{csv_file_path}'.")

    valid_urls = []
    invalid_rows = []

    try:
        with open(csv_file_path, mode="r", encoding=encoding, errors="replace") as f:
            reader = csv.reader(f)
            for i, row in enumerate(reader, start=1):
                if not row or not row[0].strip():
                    continue
                url = row[0].strip()
                # skip a possible header row like "url"
                if i == 1 and url.lower() in ("url", "urls", "link"):
                    continue
                if is_valid_url(url):
                    valid_urls.append(url)
                else:
                    invalid_rows.append((i, url))
    except UnicodeDecodeError as e:
        raise ValueError(f"Error reading CSV file with encoding '{encoding}': {e}")

    if not valid_urls:
        raise ValueError("No valid URL found in the CSV file.")

    if invalid_rows:
        print(f"WARNING: {len(invalid_rows)} invalid row(s) skipped (sample: {invalid_rows[:3]})")

    return valid_urls


def send_with_retry(pool, url, notification_type, min_delay, max_retries=MAX_RETRIES):
    """
    Sends a single request to the API with:
      - selecting a free account from the pool
      - respecting the minimum delay between requests on the same account
      - retry with exponential backoff on transient errors (429 / 5xx)
    """
    last_error = None

    for attempt in range(1, max_retries + 1):
        account = pool.get_next_available()
        if account is None:
            return {
                "url": url,
                "status": "QUOTA_EXCEEDED",
                "type": "N/A",
                "notify_time": "N/A",
                "error": "All Service Accounts have exhausted their quota.",
                "attempts": attempt,
            }

        # respect minimum delay for this account (rate limiting)
        elapsed = time.time() - account["last_request_time"]
        if elapsed < min_delay:
            time.sleep(min_delay - elapsed)

        content = {"url": url, "type": notification_type}

        try:
            response = account["session"].post(API_URL, json=content, timeout=15)
            account["last_request_time"] = time.time()
            pool.mark_used(account)

            if response.status_code == 200:
                response_json = response.json()
                latest_update = response_json.get("urlNotificationMetadata", {}).get("latestUpdate", {})
                return {
                    "url": url,
                    "status": response.status_code,
                    "type": latest_update.get("type", "N/A"),
                    "notify_time": latest_update.get("notifyTime", "N/A"),
                    "error": "",
                    "attempts": attempt,
                }

            # transient errors -> retry with backoff
            if response.status_code == 429 or 500 <= response.status_code < 600:
                last_error = f"HTTP {response.status_code}: {response.text[:200]}"
                sleep_time = (BACKOFF_BASE ** attempt) + random.uniform(0, 1)
                time.sleep(sleep_time)
                continue

            # permanent errors (400, 401, 403, 404, ...) -> return without retry
            return {
                "url": url,
                "status": response.status_code,
                "type": "N/A",
                "notify_time": "N/A",
                "error": response.text[:300],
                "attempts": attempt,
            }

        except requests.exceptions.RequestException as e:
            last_error = str(e)
            account["last_request_time"] = time.time()
            sleep_time = (BACKOFF_BASE ** attempt) + random.uniform(0, 1)
            time.sleep(sleep_time)
            continue

    return {
        "url": url,
        "status": "FAILED_AFTER_RETRIES",
        "type": "N/A",
        "notify_time": "N/A",
        "error": last_error or "Unknown error",
        "attempts": max_retries,
    }


def write_report(results, output_path):
    """Writes the final report to a CSV file"""
    fieldnames = ["url", "status", "type", "notify_time", "error", "attempts"]
    with open(output_path, mode="w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for r in results:
            writer.writerow(r)


def parse_args():
    parser = argparse.ArgumentParser(
        description="Batch-submit URLs to the Google Indexing API with multi-Service-Account support, "
                    "rate limiting, retry, and an output report."
    )
    sa_group = parser.add_argument_group("Service Account")
    sa_group.add_argument("--sa-file", action="append",
                           help="Path to a single Service Account JSON file (repeat flag for multiple files)")
    sa_group.add_argument("--sa-dir",
                           help="Path to a folder containing multiple Service Account JSON files")

    parser.add_argument("--csv", default=None, help="Path to the CSV file containing URLs (first column)")
    parser.add_argument("--type", choices=["PUBLISH", "DELETE"], default=None,
                         help="Operation type: PUBLISH or DELETE")
    parser.add_argument("--output", default=None,
                         help="Path to the output CSV report (default: report_<timestamp>.csv)")
    parser.add_argument("--encoding", default="utf-8", help="Encoding of the input CSV file (default utf-8)")
    parser.add_argument("--min-delay", type=float, default=DEFAULT_MIN_DELAY_SECONDS,
                         help="Minimum delay between two requests on the same account (seconds)")
    parser.add_argument("--daily-quota", type=int, default=DEFAULT_DAILY_QUOTA_PER_ACCOUNT,
                         help="Daily quota per Service Account (default 200)")
    parser.add_argument("--max-retries", type=int, default=MAX_RETRIES,
                         help="Max retry attempts per URL on transient errors")

    return parser.parse_args()


def prompt_missing_args(args):
    """
    If required arguments weren't passed on the command line, ask for them
    interactively (input), one at a time.
    """
    # --- Service Account(s) ---
    if not args.sa_file and not args.sa_dir:
        print("\n--- Service Account setup ---")
        mode = input("Single file or a folder with multiple files? (file/dir) [file]: ").strip().lower() or "file"
        if mode == "dir":
            sa_dir = input("Enter the path to the folder containing the Service Account JSON files: ").strip()
            args.sa_dir = sa_dir
        else:
            paths = []
            while True:
                p = input("Enter the path to a Service Account JSON file "
                           "(press Enter with no input to stop adding files): ").strip()
                if not p:
                    break
                paths.append(p)
            if not paths:
                print("ERROR: at least one Service Account is required.")
                sys.exit(1)
            args.sa_file = paths

    # --- operation type ---
    if not args.type:
        op = input("\nEnter operation type (PUBLISH or DELETE): ").strip().upper()
        while op not in ("PUBLISH", "DELETE"):
            op = input("Invalid value. Please enter PUBLISH or DELETE: ").strip().upper()
        args.type = op

    # --- input CSV file ---
    if not args.csv:
        args.csv = input("\nEnter the path to the CSV file containing the URLs: ").strip()

    return args


def main():
    args = parse_args()
    args = prompt_missing_args(args)
    notification_type = "URL_UPDATED" if args.type == "PUBLISH" else "URL_DELETED"

    # 1. Load Service Account(s)
    try:
        sa_paths = load_service_account_paths(args)
        if not sa_paths:
            print("ERROR: no Service Account specified. Use --sa-file or --sa-dir.")
            sys.exit(1)
        pool = ServiceAccountPool(sa_paths, daily_quota=args.daily_quota)
    except Exception as e:
        print(f"ERROR loading Service Account(s): {e}")
        sys.exit(1)

    # 2. Load and validate URLs
    try:
        urls = load_urls_from_csv(args.csv, encoding=args.encoding)
        print(f"Found {len(urls)} valid URL(s) to process.")
    except Exception as e:
        print(f"ERROR reading CSV file: {e}")
        sys.exit(1)

    # warn if quota looks insufficient
    total_quota = len(pool.accounts) * args.daily_quota
    if len(urls) > total_quota:
        print(f"WARNING: number of URLs ({len(urls)}) exceeds the estimated total quota ({total_quota}). "
              f"Some URLs may come back with QUOTA_EXCEEDED status.")

    # 3. Process
    results = []
    success_count = 0
    fail_count = 0

    for url in tqdm(urls, desc="Submitting", unit="url"):
        result = send_with_retry(pool, url, notification_type, args.min_delay, args.max_retries)
        results.append(result)
        if result["status"] == 200:
            success_count += 1
        else:
            fail_count += 1

    # 4. Write output report
    output_path = args.output or f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
    write_report(results, output_path)

    # 5. Final summary
    print("\n" + "=" * 50)
    print(f"Success: {success_count}")
    print(f"Failed: {fail_count}")
    print(f"Full report saved to: {output_path}")
    print("=" * 50)


if __name__ == "__main__":
    main()

مرحله دوم روش استفاده از کد

خب تا اینجا کد رو در اختیار دارید برای استفاده از کد کافیه به صورت زیر عمل کنید

به 2 چیز لازم داریم برای اجرای کد

1. فایل JSON که در پست تنظیم api index google و ایندکس سریع صفحات بهتون آموزش دادم چطوری بسازیدش.
2. یک فایل با فرمت اکسل یا CSV که به ترتیب لینک های سایت خودتون رو داخلش قرار دادید مثل این فایل که میتونید به عنوان نمونه استفاده کنید.

خب این دوتا رو که داشتید میریم سراغ استفاده از کد..

 روش استفاده از کد

وارد لینک زیر میشید که در بالا هم آوردم

لینک دسترسی به کد 

مطابق شکل زیر کد رو در شبیه ساز خود گوگل مشاهده می کنید

مطابق شکل زیر از منو سمت چپ روی آیکون فولدر می زنیم

فایل های JSON و CSV رو در این قمست مطابق شکل زیر آپلود می کنیم هم به صورت کشیدن و رها کردن و هم با استفاده از دکمه زیر میتونید اپلود کنید.

در ادامه روی run به صورت زیر می زنیم

وقتی کد رو زدیم از ما مسیر فایل JSON رو میخاد که اپلود کردیم به صورت زیر روی گزینه copy path کلیک کرده و داخل کادر مورد نظر پیست می کنیم و اینتر کی بورد رو میزنیم.

در مرحله بعد این متن میاد نوع عملیات را وارد کنید (PUBLISH یا DELETE): اگر مطالب شما تازه پابلیش شدن یا میخواهید که بروزرسانی شوند و ایندکس داخل گوگل گزینه PUBLISH و اگر میخاهید که مطاب پاک شوند DELETE را کپی کرده و داخل کادر به صورت زیر پیست می کنید و بعد اینتر را می زنید.

در مرحله بعد یعنی وقتی مسیر فایل CSV که شامل URLها است را وارد کنید: امد . دوباره مسیر فایل CSV را با استفاده از گزینه copy path کپی کرده و داخل کادر پیست می کنیم.

در نهایت مطابق تصویر زیر تمام لینک های سایت شما شروع میکنن به گوگل فرستاده شدن و در صف ایندکس سریع قرار می گیرند.

امیدوام که مفید بوده باشه اگر سوالی داشتید در کامنت بپرسید.