無塊排名跟蹤器API：IPFLY如何實現99.9%的準確監控

如果你建立或依賴排名跟蹤基礎設施，你就知道這種模式：最初的成功，然後逐漸下降。幾個關鍵字完美跟蹤，然後驗證碼出現。地理定位適用於流行市場，然後開始返回扭曲的結果。規模增加，突然間整個知識產權範圍面臨永久封鎖。

排名跟蹤器api領域充斥着部分解決方案——工具工作到不行，平臺承諾規模但限制性能，以及在檢測壓力下崩潰的DIY實現。根本問題不是技術能力；這是基礎設施的真實性。當您的排名跟蹤請求來自可識別的數據中心IP、商業VPN範圍或已知代理網絡時，複雜的搜索引擎保護系統會以不斷升級的對策做出響應。

這就是爲什麼專業的搜索引擎優化操作——管理數百名客戶的機構、跟蹤數百萬個關鍵字的企業和服務於數千名用戶的情報平臺——需要建立在真正住宅網絡存在之上的排名跟蹤器api基礎設施。不是作爲優化，而是作爲可靠運行的先決條件。

有效排名跟蹤器API系統的架構

核心功能需求

生產級排名跟蹤器api必須提供：

地理精度：排名因位置而異。準確的跟蹤需要城市級甚至社區級的查詢來源，而不是近似的區域存在。

設備和平臺覆蓋範圍：桌面和移動結果通常存在顯着差異。全面監控需要兩種環境的真實表示。

SERP特徵提取：現代結果遠遠超出了十個藍色鏈接。特色片段、本地包、知識面板、人們還問和視頻輪播都會影響可見性並需要捕獲。

時間一致性：排名全天波動。專業跟蹤需要預定的、可靠的收集，保持一致的時間，而不觸發基於速度的檢測。

規模和併發：企業運營跟蹤數千到數百萬個關鍵字。基礎設施必須支持大規模並行查詢執行，而不會降低性能或檢測風險。

檢測挑戰

搜索引擎部署了專門針對排名跟蹤器api操作的複雜保護：

IP聲譽分析：數據中心範圍、託管服務提供商IP和已知的自動化基礎設施填充實時封鎖列表。來自這些來源的查詢面臨立即限制或誤導性結果。

行爲指紋：請求計時、標頭簽名、TLS特徵和導航模式揭示了自動化。即使是“隱身”配置也經常表現出可檢測到的規律。

地理一致性驗證：不可能的旅行場景-幾分鐘內來自多個大陸的查詢-觸發安全響應。分佈式收集需要真實的本地存在，而不是VPN風格的路由。

漸進式響應升級：初始速率限制升級爲驗證碼挑戰、臨時阻止和永久黑名單。從高級限制中恢復需要更換基礎架構，而不是調整配置。

IPFLY的解決方案：Rank Tracker API卓越的住宅基礎設施

正宗網絡基金會

IPFLY爲排名跟蹤器api開發人員提供關鍵基礎設施層：190多個國家/地區的9000多萬個住宅IP地址，代表與真實消費者和商業地點的真正ISP分配連接。

這個住宅基礎改變了排名跟蹤能力：

無法檢測到的查詢發起：請求顯示爲來自真實家庭和辦公室互聯網連接的合法用戶搜索。搜索引擎保護系統無法將專業排名跟蹤與普通消費者行爲區分開來。

地理真實性：城市和州級定位確保排名跟蹤器api查詢捕獲真正的本地搜索結果，而不是VPN近似或data-center-distorted響應。

大規模分發能力：數以百萬計的可用IP支持查詢分發，將每個地址的頻率保持在遠低於檢測閾值的水平，同時實現企業規模所需的聚合收集速度。

企業級操作標準

專業等級跟蹤器api操作要求可靠性：

99.9%正常運行時間SLA：持續監控不能容忍基礎設施中斷。IPFLY的冗餘網絡架構確保了一致的查詢能力。

無限併發處理：從數億到數百萬的同時關鍵字檢查，基礎設施可以在不限制、排隊或性能下降的情況下擴展。

毫秒響應優化：高速骨幹網連接可最大限度地減少查詢啓動和結果捕獲之間的延遲，從而實現實時或接近實時的智能交付。

24/7技術支持：集成優化、故障排除和擴展指導的專家協助-而不是自動響應或社區論壇。

數據質量保證

IPFLY通過以下方式提高排名跟蹤器api的準確性：

IP信譽管理：嚴格的過濾確保查詢僅使用具有乾淨搜索歷史的高純度住宅地址，防止先前標記的IP受到污染。

一致的結果交付：真實的住宅訪問消除了損害基於數據中心的跟蹤的個性化扭曲、阻塞響應和誤導性結果。

會話穩定性選項：靜態住宅分配支持縱向跟蹤的持久本地存在，而動態輪換提供最大分佈以實現廣泛覆蓋。

構建您的Rank Tracker API：技術實現

Python核心實現

具有IPFLY集成的基本排名跟蹤器：

蟒蛇

import requests
from bs4 import BeautifulSoup
from urllib.parse import quote_plus, urlencode
from typing import List, Dict, Optional, Union
from dataclasses import dataclass
from datetime import datetime
import time
import random
import json

@dataclassclassRankingResult:
    keyword:str
    position:int
    url:str
    title:str
    description:str
    location:str
    device_type:str
    serp_features: Dict
    timestamp: datetime

classIPFLYRankTracker:"""
    Production-grade rank tracker with IPFLY residential proxy integration.
    """
    
    SEARCH_ENGINES ={'google':'https://www.google.com/search','bing':'https://www.bing.com/search'}def__init__(self, ipfly_config: Dict, engine:str='google'):
        self.ipfly_config = ipfly_config
        self.engine = engine
        self.session = requests.Session()# Configure IPFLY residential proxy
        proxy_url =(f"http://{ipfly_config['username']}:{ipfly_config['password']}"f"@{ipfly_config['host']}:{ipfly_config['port']}")
        self.session.proxies ={'http': proxy_url,'https': proxy_url
        }# Rotate realistic user agents
        self.user_agents =['Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36','Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36','Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36','Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0']# Mobile user agents for device-specific tracking
        self.mobile_agents =['Mozilla/5.0 (iPhone; CPU iPhone OS 17_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1','Mozilla/5.0 (Linux; Android 14; SM-S918B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36']defconstruct_search_url(
        self,
        keyword:str,
        location:str='us',
        language:str='en',
        start:int=0,
        device_type:str='desktop')->str:"""Build search URL with localization parameters."""
        base_url = self.SEARCH_ENGINES.get(self.engine, self.SEARCH_ENGINES['google'])
        
        params ={'q': quote_plus(keyword),'hl': language,'gl': location,'start': start,'num':100}# Device-specific parametersif device_type =='mobile':
            params['uiv']='mb'# Mobile view indicator for some engines
        
        query_string = urlencode(params)returnf"{base_url}?{query_string}"deftrack_keyword(
        self,
        keyword:str,
        target_domain: Optional[str]=None,
        location:str='us',
        language:str='en',
        device_type:str='desktop',
        pages:int=3)-> List[RankingResult]:"""
        Track keyword rankings with comprehensive result extraction.
        """
        all_results =[]
        target_found =Falsefor page inrange(pages):
            offset = page *100
            
            url = self.construct_search_url(
                keyword, location, language, offset, device_type
            )# Select appropriate user agent
            user_agent =(
                random.choice(self.mobile_agents)if device_type =='mobile'else random.choice(self.user_agents))
            
            headers ={'User-Agent': user_agent,'Accept':'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8','Accept-Language':f'{language}-{location},{language};q=0.9','Accept-Encoding':'gzip, deflate, br','DNT':'1','Connection':'keep-alive','Upgrade-Insecure-Requests':'1'}try:# Human-like delay with jitter
                time.sleep(random.uniform(3,7))
                
                response = self.session.get(
                    url,
                    headers=headers,
                    timeout=45,
                    allow_redirects=True)
                response.raise_for_status()# Parse results with engine-specific logicif self.engine =='google':
                    page_results = self._parse_google_results(
                        response.text, keyword, location, device_type, offset
                    )else:
                    page_results = self._parse_bing_results(
                        response.text, keyword, location, device_type, offset
                    )# Check for target domainif target_domain:for result in page_results:if target_domain in result.url:
                            target_found =True
                            result.is_target =True
                
                all_results.extend(page_results)# Early termination if target found or no more resultsif target_found orlen(page_results)<10:breakexcept requests.exceptions.RequestException as e:print(f"Request failed for '{keyword}' page {page}: {e}")# IPFLY's rotating proxy automatically handles IP refresh# Implement retry logic with exponential backoff
                time.sleep(random.uniform(10,20))continuereturn all_results
    
    def_parse_google_results(
        self,
        html:str,
        keyword:str,
        location:str,
        device_type:str,
        offset:int)-> List[RankingResult]:"""Parse Google SERP results with feature extraction."""
        soup = BeautifulSoup(html,'html.parser')
        results =[]# Organic results
        result_containers = soup.select('div.g, div[data-header-feature]')for idx, container inenumerate(result_containers):
            position = offset + idx +1try:
                title_elem = container.select_one('h3')
                url_elem = container.select_one('a[href]')
                desc_elem = container.select_one('div.VwiC3b, span.aCOpRe')# Extract SERP features
                features = self._extract_google_features(container)if title_elem and url_elem:
                    result = RankingResult(
                        keyword=keyword,
                        position=position,
                        url=url_elem.get('href',''),
                        title=title_elem.get_text(strip=True),
                        description=desc_elem.get_text(strip=True)if desc_elem else'',
                        location=location,
                        device_type=device_type,
                        serp_features=features,
                        timestamp=datetime.utcnow())
                    results.append(result)except Exception as e:print(f"Parsing error at position {position}: {e}")continue# Extract additional SERP features (PAA, related searches, etc.)
        page_features = self._extract_page_level_features(soup)for result in results:
            result.serp_features.update(page_features)return results
    
    def_extract_google_features(self, container)-> Dict:"""Extract Google-specific SERP features from result container."""
        features ={'type':'organic','has_sitelinks':len(container.select('div.osl a'))>0,'has_image':len(container.select('img.XNo5Ab'))>0,'is_featured_snippet':'xpdopen'in container.get('class',[]),'is_local_pack':False,'is_knowledge_panel':False}# Detect rich result typesif container.select('div.xpdopen'):
            features['type']='featured_snippet'elif container.select('div.g-blk'):
            features['type']='knowledge_panel'elif container.select('div.dbg0pd'):
            features['type']='local_result'
            features['is_local_pack']=Truereturn features
    
    def_extract_page_level_features(self, soup)-> Dict:"""Extract page-level SERP features."""return{'people_also_ask':[
                elem.get_text(strip=True)for elem in soup.select('div.related-question-pair span')],'related_searches':[
                elem.get_text(strip=True)for elem in soup.select('div.AJLUJb a')],'has_local_pack':len(soup.select('div#lclbox'))>0,'has_knowledge_panel':len(soup.select('div.knowledge-panel'))>0,'has_top_stories':len(soup.select('g-section-with-header'))>0}def_parse_bing_results(self, html, keyword, location, device_type, offset):"""Parse Bing SERP results."""# Implementation similar to Google parsing# with Bing-specific selectorspass# Production usage exampleif __name__ =="__main__":
    ipfly_config ={'host':'proxy.ipfly.com','port':'3128','username':'your_ipfly_username','password':'your_ipfly_password'}
    
    tracker = IPFLYRankTracker(ipfly_config, engine='google')# Track keyword with geographic and device precision
    results = tracker.track_keyword(
        keyword="enterprise seo software",
        target_domain="example.com",
        location="us",
        device_type="desktop",
        pages=2)print(f"Found {len(results)} results")for r in results[:10]:
        marker =" [TARGET]"ifgetattr(r,'is_target',False)else""print(f"{r.position}. {r.title[:60]}{marker}")print(f"   {r.url[:70]}...")

基於瀏覽器的高級實現

對於JavaScript密集型SERP和綜合特徵提取：

蟒蛇

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.action_chains import ActionChains
from webdriver_manager.chrome import ChromeDriverManager
import time
import random

classBrowserBasedRankTracker:"""
    Browser automation rank tracker with IPFLY SOCKS5 proxy and stealth configuration.
    """def__init__(self, ipfly_config: Dict, headless:bool=True):
        self.ipfly_config = ipfly_config
        self.headless = headless
        self.driver =Nonedefinitialize_driver(self):"""Initialize stealth Chrome with IPFLY residential proxy."""
        chrome_options = Options()if self.headless:
            chrome_options.add_argument('--headless')# Essential stability arguments
        chrome_options.add_argument('--no-sandbox')
        chrome_options.add_argument('--disable-dev-shm-usage')
        chrome_options.add_argument('--disable-blink-features=AutomationControlled')
        chrome_options.add_argument('--disable-web-security')
        chrome_options.add_argument('--disable-features=IsolateOrigins,site-per-process')
        chrome_options.add_argument('--disable-site-isolation-trials')# IPFLY SOCKS5 proxy configuration
        socks_proxy =(f"{self.ipfly_config['host']}:{self.ipfly_config.get('socks_port','1080')}")
        chrome_options.add_argument(f'--proxy-server=socks5://{socks_proxy}')# Window size and user agent
        chrome_options.add_argument('--window-size=1920,1080')
        chrome_options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36')# Exclude automation indicators
        chrome_options.add_experimental_option("excludeSwitches",["enable-automation"])
        chrome_options.add_experimental_option('useAutomationExtension',False)# Initialize driver
        service = Service(ChromeDriverManager().install())
        self.driver = webdriver.Chrome(service=service, options=chrome_options)# Execute CDP commands for stealth
        self.driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument',{'source':'''
                Object.defineProperty(navigator, 'webdriver', {
                    get: () => undefined
                });
                Object.defineProperty(navigator, 'plugins', {
                    get: () => [1, 2, 3, 4, 5]
                });
                window.chrome = {
                    runtime: {}
                };
            '''})# Set additional headers via CDP
        self.driver.execute_cdp_cmd('Network.setExtraHTTPHeaders',{'headers':{'Accept-Language':'en-US,en;q=0.9','DNT':'1'}})deftrack_with_visual_verification(
        self,
        keyword:str,
        location:str='United States',
        device_type:str='desktop')-> Dict:"""
        Track rankings with full browser rendering and visual feature detection.
        """ifnot self.driver:
            self.initialize_driver()try:# Construct search URL with localization
            search_url =(f"https://www.google.com/search?"f"q={keyword.replace(' ','+')}&"f"hl=en&"f"gl={location[:2].lower()iflen(location)==2else'us'}")# Add mobile parameter if neededif device_type =='mobile':
                search_url +="&uiv=mb"
                
            self.driver.get(search_url)# Wait for results with realistic timing
            wait = WebDriverWait(self.driver,15)
            wait.until(EC.presence_of_element_located((By.ID,"search")))# Random scroll behavior to mimic human interaction
            self._human_like_scroll()# Extract comprehensive results
            organic_results = self._extract_organic_with_selenium()
            serp_features = self._extract_all_serp_features()return{'keyword': keyword,'location': location,'device_type': device_type,'organic_results': organic_results,'serp_features': serp_features,'page_title': self.driver.title,'timestamp': time.time()}except Exception as e:print(f"Tracking failed: {e}")return{'error':str(e)}def_human_like_scroll(self):"""Simulate human scrolling behavior."""try:# Initial pause
            time.sleep(random.uniform(2,4))# Random scroll downfor _ inrange(random.randint(2,5)):
                scroll_amount = random.randint(300,800)
                self.driver.execute_script(f"window.scrollBy(0, {scroll_amount});")
                time.sleep(random.uniform(1,3))# Occasional scroll upif random.random()>0.7:
                self.driver.execute_script("window.scrollBy(0, -400);")
                time.sleep(random.uniform(1,2))except:passdef_extract_organic_with_selenium(self)-> List[Dict]:"""Extract organic results using Selenium for JavaScript-rendered content."""
        results =[]try:
            containers = self.driver.find_elements(By.CSS_SELECTOR,"div.g")for position, container inenumerate(containers,1):try:
                    title = container.find_element(By.CSS_SELECTOR,"h3").text
                    url = container.find_element(By.CSS_SELECTOR,"a").get_attribute("href")# Multiple attempts for description
                    description =""for selector in["div.VwiC3b","span.aCOpRe","div.s3v94d"]:try:
                            description = container.find_element(By.CSS_SELECTOR, selector).text
                            breakexcept:continue
                    
                    results.append({'position': position,'title': title,'url': url,'description': description[:200],'is_featured_snippet':'xpdopen'in container.get_attribute("class")})except Exception as e:continueexcept Exception as e:print(f"Extraction error: {e}")return results
    
    def_extract_all_serp_features(self)-> Dict:"""Extract comprehensive SERP feature set."""
        features ={'featured_snippet': self._extract_featured_snippet(),'people_also_ask': self._extract_paa(),'local_pack': self._extract_local_pack(),'knowledge_panel': self._extract_knowledge_panel(),'top_stories': self._extract_top_stories(),'video_carousel': self._extract_video_carousel(),'image_pack': self._extract_image_pack(),'related_searches': self._extract_related_searches()}return features
    
    def_extract_featured_snippet(self)-> Optional[Dict]:"""Extract featured snippet content."""try:
            snippet = self.driver.find_element(By.CSS_SELECTOR,"div.xpdopen")return{'type':'paragraph',# Simplified - detect actual type'content': snippet.text[:500],'source': snippet.find_element(By.CSS_SELECTOR,"a").get_attribute("href")}except:returnNonedef_extract_paa(self)-> List[Dict]:"""Extract People Also Ask questions and answers."""
        paa_data =[]try:
            questions = self.driver.find_elements(By.CSS_SELECTOR,"div.related-question-pair")for q in questions:try:
                    question = q.find_element(By.CSS_SELECTOR,"span").text
                    # Click to expand and get answer
                    q.click()
                    time.sleep(0.5)
                    answer = q.find_element(By.CSS_SELECTOR,"div").text
                    paa_data.append({'question': question,'answer': answer[:300]})except:continueexcept:passreturn paa_data
    
    def_extract_local_pack(self)-> Optional[List[Dict]]:"""Extract local pack results."""try:
            local_results =[]
            pack = self.driver.find_elements(By.CSS_SELECTOR,"div.dbg0pd")for item in pack:
                local_results.append({'name': item.text,'details':'Extracted from local pack'})return local_results if local_results elseNoneexcept:returnNonedef_extract_knowledge_panel(self)-> Optional[Dict]:"""Extract knowledge panel information."""try:
            panel = self.driver.find_element(By.CSS_SELECTOR,"div.knowledge-panel, div#kp-wp-tab-overview")return{'title': panel.find_element(By.CSS_SELECTOR,"h2, div[role='heading']").text,'entity_type':'detected'}except:returnNonedef_extract_top_stories(self)-> List[Dict]:"""Extract top stories/news results."""
        stories =[]try:
            story_elements = self.driver.find_elements(By.CSS_SELECTOR,"g-section-with-header div[role='listitem']")for elem in story_elements[:3]:
                stories.append({'title': elem.text[:100],'source':'news'})except:passreturn stories
    
    def_extract_video_carousel(self)-> List[Dict]:"""Extract video carousel results."""
        videos =[]try:
            video_elements = self.driver.find_elements(By.CSS_SELECTOR,"g-scrolling-carousel div[role='listitem']")for elem in video_elements[:3]:
                videos.append({'title': elem.text[:100]})except:passreturn videos
    
    def_extract_image_pack(self)-> Optional[Dict]:"""Extract image pack results."""try:
            pack = self.driver.find_element(By.CSS_SELECTOR,"div#imagebox_bigimages")return{'present':True,'count':len(pack.find_elements(By.CSS_SELECTOR,"img"))}except:returnNonedef_extract_related_searches(self)-> List[str]:"""Extract related search queries."""
        related =[]try:
            elements = self.driver.find_elements(By.CSS_SELECTOR,"div.AJLUJb a")for elem in elements:
                related.append(elem.text)except:passreturn related
    
    defclose(self):"""Clean up resources."""if self.driver:
            self.driver.quit()# Production proxy rotation managerclassIPFLYRotationManager:"""
    Manages IPFLY proxy rotation for high-volume rank tracking operations.
    """def__init__(self, ipfly_accounts: List[Dict]):
        self.accounts = ipfly_accounts
        self.current_idx =0
        self.failure_counts ={i:0for i inrange(len(accounts))}
        self.success_counts ={i:0for i inrange(len(accounts))}defget_optimal_proxy(self, target_location:str='us')-> Dict:"""Select best proxy based on performance history and location match."""
        candidates =[]for idx, account inenumerate(self.accounts):# Skip high-failure IPsif self.failure_counts[idx]>10:continue# Prioritize location match
            location_match = account.get('location','us')== target_location
            score = self.success_counts[idx]-(self.failure_counts[idx]*2)if location_match:
                score +=5
                
            candidates.append((idx, score, account))ifnot candidates:# Reset and use all
            self.failure_counts ={i:0for i inrange(len(self.accounts))}
            candidates =[(i,0, acc)for i, acc inenumerate(self.accounts)]# Select highest scoring candidate
        best =max(candidates, key=lambda x: x[1])return best[2]defreport_result(self, proxy_idx:int, success:bool):"""Update performance tracking."""if success:
            self.success_counts[proxy_idx]+=1
            self.failure_counts[proxy_idx]=max(0, self.failure_counts[proxy_idx]-1)else:
            self.failure_counts[proxy_idx]+=1

FastAPI生產服務

將您的排名跟蹤器api部署爲可擴展的Web服務：

蟒蛇

from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks, Query
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field, validator
from typing import List, Optional, Dict, Union
from datetime import datetime, timedelta
from enum import Enum
import asyncio
import aioredis
import json
import os
import time

app = FastAPI(
    title="Rank Tracker API",
    description="Production-grade SEO rank tracking with IPFLY residential proxies",
    version="2.0.0",
    docs_url="/docs",
    redoc_url="/redoc")

security = HTTPBearer()# Redis connection for caching and queuesasyncdefget_redis():
    redis =await aioredis.from_url(
        os.getenv("REDIS_URL","redis://localhost:6379"),
        decode_responses=True)return redis

classSearchEngine(str, Enum):
    google ="google"
    bing ="bing"classDeviceType(str, Enum):
    desktop ="desktop"
    mobile ="mobile"
    tablet ="tablet"classTrackingRequest(BaseModel):
    keywords: List[str]= Field(..., min_items=1, max_items=1000)
    target_domain: Optional[str]= Field(None, description="Domain to highlight in results")
    locations: List[str]= Field(default=["us"], max_items=50)
    device_types: List[DeviceType]= Field(default=[DeviceType.desktop])
    search_engine: SearchEngine = Field(default=SearchEngine.google)
    pages_per_keyword:int= Field(default=2, ge=1, le=10)
    priority:int= Field(default=1, ge=1, le=5)@validator('keywords')defvalidate_keywords(cls, v):ifany(len(k)>500for k in v):raise ValueError("Individual keywords must be under 500 characters")return v

classRankingData(BaseModel):
    keyword:str
    position:int
    url:str
    title:str
    description:str
    location:str
    device_type:str
    serp_features: Dict
    is_target:bool=False
    timestamp: datetime

classTrackingResponse(BaseModel):
    job_id:str
    status:str
    keywords_submitted:int
    estimated_completion:str
    check_status_url:strclassJobStatus(BaseModel):
    job_id:str
    status:str# queued, processing, completed, failed
    progress_percent:float
    results_count:int
    errors: List[str]
    completed_at: Optional[datetime]defverify_api_key(credentials: HTTPAuthorizationCredentials = Depends(security)):"""Verify API authentication."""# Implement proper key validationif credentials.credentials != os.getenv("API_KEY"):raise HTTPException(status_code=401, detail="Invalid API key")return credentials.credentials

defget_ipfly_pool():"""Load IPFLY proxy pool configuration."""# Load from secure configurationreturn[{'host': os.getenv(f'IPFLY_HOST_{i}','proxy.ipfly.com'),'port': os.getenv(f'IPFLY_PORT_{i}','3128'),'username': os.getenv(f'IPFLY_USER_{i}'),'password': os.getenv(f'IPFLY_PASS_{i}'),'location': os.getenv(f'IPFLY_LOC_{i}','us')}for i inrange(int(os.getenv('IPFLY_POOL_SIZE','10')))]@app.post("/track", response_model=TrackingResponse)asyncdefsubmit_tracking_job(
    request: TrackingRequest,
    background_tasks: BackgroundTasks,
    api_key:str= Depends(verify_api_key)):"""
    Submit keywords for rank tracking across specified locations and devices.
    """
    job_id =f"track_{int(time.time())}_{hash(str(request.keywords))}"# Calculate estimated completion
    total_queries =(len(request.keywords)*len(request.locations)*len(request.device_types)* 
        request.pages_per_keyword
    )
    estimated_minutes =max(1, total_queries /100)# Approximate rate# Queue job for processing
    redis =await get_redis()await redis.setex(f"job:{job_id}",86400,# 24 hour TTL
        json.dumps({'status':'queued','request': request.dict(),'submitted_at': datetime.utcnow().isoformat(),'total_queries': total_queries
        }))# Trigger background processing
    background_tasks.add_task(process_tracking_job, job_id, request)return TrackingResponse(
        job_id=job_id,
        status="queued",
        keywords_submitted=len(request.keywords),
        estimated_completion=f"{estimated_minutes:.1f} minutes",
        check_status_url=f"/status/{job_id}")asyncdefprocess_tracking_job(job_id:str, request: TrackingRequest):"""Background processing of rank tracking job."""
    redis =await get_redis()
    ipfly_pool = get_ipfly_pool()try:await redis.hset(f"job:{job_id}","status","processing")
        
        tracker = IPFLYRankTracker(ipfly_pool[0], engine=request.search_engine.value)
        
        all_results =[]
        completed =0
        total =len(request.keywords)*len(request.locations)*len(request.device_types)for keyword in request.keywords:for location in request.locations:for device in request.device_types:try:
                        results = tracker.track_keyword(
                            keyword=keyword,
                            target_domain=request.target_domain,
                            location=location,
                            device_type=device.value,
                            pages=request.pages_per_keyword
                        )
                        
                        all_results.extend([r.dict()for r in results])
                        completed +=1# Update progress
                        progress =(completed / total)*100await redis.hset(f"job:{job_id}","progress",str(progress))await redis.hset(f"job:{job_id}","results_count",str(len(all_results)))# Brief delay between queriesawait asyncio.sleep(random.uniform(2,5))except Exception as e:await redis.lpush(f"job:{job_id}:errors",str(e))# Store final resultsawait redis.setex(f"results:{job_id}",604800,# 7 day retention
            json.dumps(all_results))await redis.hset(f"job:{job_id}","status","completed")await redis.hset(f"job:{job_id}","completed_at", datetime.utcnow().isoformat())except Exception as e:await redis.hset(f"job:{job_id}","status","failed")await redis.hset(f"job:{job_id}","error",str(e))@app.get("/status/{job_id}", response_model=JobStatus)asyncdefcheck_job_status(
    job_id:str,
    api_key:str= Depends(verify_api_key)):"""Check status of submitted tracking job."""
    redis =await get_redis()
    
    status_data =await redis.hgetall(f"job:{job_id}")ifnot status_data:raise HTTPException(status_code=404, detail="Job not found")
    
    errors =await redis.lrange(f"job:{job_id}:errors",0,-1)return JobStatus(
        job_id=job_id,
        status=status_data.get('status','unknown'),
        progress_percent=float(status_data.get('progress',0)),
        results_count=int(status_data.get('results_count',0)),
        errors=errors or[],
        completed_at=status_data.get('completed_at'))@app.get("/results/{job_id}")asyncdefget_job_results(
    job_id:str,format:str= Query(default="json", regex="^(json|csv)$"),
    api_key:str= Depends(verify_api_key)):"""Retrieve completed tracking results."""
    redis =await get_redis()
    
    results_json =await redis.get(f"results:{job_id}")ifnot results_json:raise HTTPException(status_code=404, detail="Results not found or expired")
    
    results = json.loads(results_json)ifformat=="csv":# Convert to CSVimport csv
        import io
        
        output = io.StringIO()if results:
            writer = csv.DictWriter(output, fieldnames=results[0].keys())
            writer.writeheader()
            writer.writerows(results)from fastapi.responses import PlainTextResponse
        return PlainTextResponse(
            content=output.getvalue(),
            media_type="text/csv",
            headers={"Content-Disposition":f"attachment; filename=rankings_{job_id}.csv"})return{"job_id": job_id,"results": results,"count":len(results)}@app.get("/health")asyncdefhealth_check():"""Service health status."""
    redis =await get_redis()
    redis_ok =await redis.ping()return{"status":"healthy","redis":"connected"if redis_ok else"disconnected","ipfly_pool_size":len(get_ipfly_pool()),"timestamp": datetime.utcnow()}if __name__ =="__main__":import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

IPFLY集成：確保Rank Tracker API成功

爲什麼住宅代理是不可談判的

搜索引擎保護系統專門針對排名跟蹤器api基礎設施：

配置優化

蟒蛇

# IPFLY configuration for rank tracking excellenceclassIPFLYRankConfig:"""
    Optimized IPFLY configurations for different rank tracking scenarios.
    """# High-frequency tracking: Maximum distribution
    HIGH_VOLUME ={'type':'rotating','rotation_policy':'per_request','countries':['us','gb','ca','au','de','fr'],'session_duration':0}# Local SEO tracking: Geographic precision
    LOCAL_SEO ={'type':'static','rotation_policy':'daily','city_targeting':True,'session_duration':86400# 24 hours}# Competitive monitoring: Persistent identity
    COMPETITIVE_INTEL ={'type':'static','rotation_policy':'weekly','sticky_sessions':True,'session_duration':604800# 7 days}# Mobile tracking: Device-appropriate networks
    MOBILE_TRACKING ={'type':'rotating','mobile_isp_pool':True,'carrier_simulation':True}

生產級排名跟蹤器API基礎設施

構建提供可靠、準確和可擴展的搜索引擎優化智能的排名跟蹤器應用編程接口需要將卓越的技術實現與確保一致、不可檢測訪問的基礎設施相結合。IPFLY的住宅代理網絡提供了這一基礎——真實的互聯網服務提供商分配的地址、大規模的全球規模和企業級可靠性，將排名跟蹤從脆弱的實驗轉變爲強大的操作能力。

對於致力於數據驅動的卓越搜索引擎優化的組織，IPFLY支持符合專業要求的排名跟蹤器應用編程接口開發：地理精度、全面的特徵提取、無限規模和檢測阻力，無論目標複雜程度如何，都能保持數據完整性。