135 lines
5.6 KiB
Python
135 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
DataForSEO Client - Updated per official docs (2026-03-08)
|
|
Correct endpoints:
|
|
- Keyword suggestions: /v3/dataforseo_labs/google/keyword_suggestions/live
|
|
- SERP data: /v3/serp/google/organic/live/advanced
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import base64
|
|
import requests
|
|
from typing import Dict, List, Optional
|
|
|
|
|
|
class DataForSEOClient:
|
|
"""DataForSEO API v3 client"""
|
|
|
|
def __init__(self, login: str, password: str):
|
|
self.login = login
|
|
self.password = password
|
|
self.base_url = "https://api.dataforseo.com/v3"
|
|
auth_bytes = f"{login}:{password}".encode('utf-8')
|
|
self._auth_header = f"Basic {base64.b64encode(auth_bytes).decode('utf-8')}"
|
|
|
|
def _make_request(self, endpoint: str, data: List[Dict]) -> Dict:
|
|
url = f"{self.base_url}{endpoint}"
|
|
headers = {'Authorization': self._auth_header, 'Content-Type': 'application/json'}
|
|
response = requests.post(url, json=data, headers=headers, timeout=60)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_keyword_suggestions(self, keyword: str, location: str = "Thailand", language: str = "Thai") -> List[Dict]:
|
|
"""Get keyword suggestions from DataForSEO Labs"""
|
|
try:
|
|
data = [{"keywords": [keyword], "location_name": location, "language_name": language, "include_serp_info": True}]
|
|
endpoint = "/dataforseo_labs/google/keyword_suggestions/live"
|
|
response = self._make_request(endpoint, data)
|
|
|
|
if response.get('status_code') == 20000 and response.get('tasks'):
|
|
task = response['tasks'][0]
|
|
if task.get('result'):
|
|
keywords = []
|
|
for kw_item in task['result'][0].get('related_keywords', []):
|
|
keywords.append({
|
|
'keyword': kw_item.get('keyword', ''),
|
|
'search_volume': kw_item.get('search_volume', 0),
|
|
'cpc': kw_item.get('cpc', 0),
|
|
'competition': kw_item.get('competition', 0)
|
|
})
|
|
return keywords
|
|
return []
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
return []
|
|
|
|
def get_serp_data(self, keyword: str, location: str = "Thailand", language: str = "English") -> Dict:
|
|
"""Get Google SERP data"""
|
|
try:
|
|
data = [{"keyword": keyword, "location_name": location, "language_name": language, "depth": 10}]
|
|
endpoint = "/serp/google/organic/live/advanced"
|
|
response = self._make_request(endpoint, data)
|
|
|
|
if response.get('status_code') == 20000 and response.get('tasks'):
|
|
task = response['tasks'][0]
|
|
if task.get('result'):
|
|
result = task['result'][0]
|
|
return {
|
|
'keyword': keyword,
|
|
'total_results': result.get('total_count', 0),
|
|
'items_count': len(result.get('items', [])),
|
|
'items': result.get('items', [])
|
|
}
|
|
return {'error': 'No data found'}
|
|
except Exception as e:
|
|
return {'error': str(e)}
|
|
|
|
def analyze_competitor_gap(self, your_domain: str, competitor_domain: str, keywords: List[str]) -> Dict:
|
|
"""Find keywords competitor ranks for but you don't"""
|
|
gap_keywords = []
|
|
for keyword in keywords[:20]:
|
|
try:
|
|
serp_data = self.get_serp_data(keyword)
|
|
if 'error' not in serp_data:
|
|
competitor_rank = None
|
|
your_rank = None
|
|
for i, item in enumerate(serp_data.get('items', [])[:20], 1):
|
|
domain = item.get('domain', '')
|
|
if competitor_domain in domain:
|
|
competitor_rank = i
|
|
if your_domain in domain:
|
|
your_rank = i
|
|
if competitor_rank and (not your_rank or competitor_rank < your_rank):
|
|
gap_keywords.append({
|
|
'keyword': keyword,
|
|
'your_position': your_rank,
|
|
'competitor_position': competitor_rank,
|
|
'gap': your_rank - competitor_rank if your_rank else competitor_rank
|
|
})
|
|
except:
|
|
continue
|
|
return {'gap_keywords': gap_keywords, 'total_gaps': len(gap_keywords), 'analyzed_keywords': len(keywords)}
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description='Test DataForSEO Client')
|
|
parser.add_argument('--login', required=True)
|
|
parser.add_argument('--password', required=True)
|
|
parser.add_argument('--keyword', default='podcast')
|
|
parser.add_argument('--location', default='Thailand')
|
|
parser.add_argument('--language', default='Thai')
|
|
args = parser.parse_args()
|
|
|
|
print(f"\n🔍 Testing DataForSEO API v3\n")
|
|
|
|
try:
|
|
client = DataForSEOClient(args.login, args.password)
|
|
print("Getting keyword suggestions...")
|
|
keywords = client.get_keyword_suggestions(args.keyword, args.location, args.language)
|
|
|
|
if keywords:
|
|
print(f" ✅ Found {len(keywords)} keywords\n")
|
|
for kw in keywords[:10]:
|
|
print(f" • {kw['keyword']}: {kw['search_volume']:,} searches")
|
|
print(f"\n ✅ DataForSEO working!")
|
|
else:
|
|
print(" ⚠ No keywords returned")
|
|
except Exception as e:
|
|
print(f"\n❌ ERROR: {e}")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|