132 lines
4.5 KiB
Python
132 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Script to fetch real AMO CRM responses for testing
|
|
Usage: python scripts/fetch_amocrm_data.py
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
# Add project root to Python path
|
|
project_root = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
from adapters.amocrm_client import AmoCRMClient
|
|
from utils.config import settings
|
|
|
|
|
|
def save_response_to_file(data: dict, filename: str, output_dir: Path = None):
|
|
"""Save response data to a JSON file"""
|
|
if output_dir is None:
|
|
output_dir = project_root / "tests" / "fixtures" / "real_responses"
|
|
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
filepath = output_dir / f"{filename}.json"
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
print(f"💾 Saved {filename} to {filepath}")
|
|
|
|
|
|
def generate_python_fixtures(all_data: dict, output_file: Path = None):
|
|
"""Generate Python fixtures file from real responses"""
|
|
if output_file is None:
|
|
output_file = project_root / "tests" / "fixtures" / "real_amocrm_responses.py"
|
|
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
content = f'''"""
|
|
Real AMO CRM API responses from wecheap.amocrm.ru
|
|
Generated on: {timestamp}
|
|
"""
|
|
|
|
# Users (Пользователи) response
|
|
USERS_RESPONSE = {json.dumps(all_data.get("users", {}), ensure_ascii=False, indent=4)}
|
|
|
|
# Pipelines (Воронки) response
|
|
PIPELINES_RESPONSE = {json.dumps(all_data.get("pipelines", {}), ensure_ascii=False, indent=4)}
|
|
|
|
# Companies (Компании) response
|
|
COMPANIES_RESPONSE = {json.dumps(all_data.get("companies", {}), ensure_ascii=False, indent=4)}
|
|
|
|
# Contacts (Контакты) response
|
|
CONTACTS_RESPONSE = {json.dumps(all_data.get("contacts", {}), ensure_ascii=False, indent=4)}
|
|
|
|
# Deals (Сделки) response
|
|
DEALS_RESPONSE = {json.dumps(all_data.get("deals", {}), ensure_ascii=False, indent=4)}
|
|
|
|
# Events (События) response
|
|
EVENTS_RESPONSE = {json.dumps(all_data.get("events", {}), ensure_ascii=False, indent=4)}
|
|
|
|
# Custom fields metadata for deals
|
|
CUSTOM_FIELDS_DEALS_RESPONSE = {json.dumps(all_data.get("custom_fields_deals", {}), ensure_ascii=False, indent=4)}
|
|
|
|
# Custom fields metadata for contacts
|
|
CUSTOM_FIELDS_CONTACTS_RESPONSE = {json.dumps(all_data.get("custom_fields_contacts", {}), ensure_ascii=False, indent=4)}
|
|
|
|
# Custom fields metadata for companies
|
|
CUSTOM_FIELDS_COMPANIES_RESPONSE = {json.dumps(all_data.get("custom_fields_companies", {}), ensure_ascii=False, indent=4)}
|
|
'''
|
|
|
|
with open(output_file, 'w', encoding='utf-8') as f:
|
|
f.write(content)
|
|
|
|
print(f"🐍 Generated Python fixtures at {output_file}")
|
|
|
|
|
|
async def main():
|
|
"""Main function to fetch and save AMO CRM data"""
|
|
print("🚀 Fetching AMO CRM data from wecheap.amocrm.ru")
|
|
print(f"📡 Domain: {settings.AMO_CRM_DOMAIN}")
|
|
|
|
if not settings.AMO_CRM_ACCESS_TOKEN:
|
|
print("❌ Error: AMO_CRM_ACCESS_TOKEN not set!")
|
|
print("Please set it in your .env file")
|
|
return
|
|
|
|
try:
|
|
# Create client and fetch data
|
|
client = AmoCRMClient()
|
|
all_data = await client.fetch_all_data()
|
|
|
|
print("\n📁 Saving responses...")
|
|
|
|
# Save individual JSON files
|
|
for entity_type, data in all_data.items():
|
|
if data: # Only save non-empty responses
|
|
save_response_to_file(data, entity_type)
|
|
|
|
# Generate Python fixtures
|
|
generate_python_fixtures(all_data)
|
|
|
|
print("\n✅ All done! Real AMO CRM responses saved for testing")
|
|
print("\nFiles created:")
|
|
print("- tests/fixtures/real_responses/*.json (individual JSON files)")
|
|
print("- tests/fixtures/real_amocrm_responses.py (Python fixtures)")
|
|
|
|
# Show summary
|
|
print("\n📊 Data Summary:")
|
|
for entity_type, data in all_data.items():
|
|
if "_embedded" in data:
|
|
embedded_key = list(data["_embedded"].keys())[0] if data["_embedded"] else "items"
|
|
count = len(data["_embedded"].get(embedded_key, []))
|
|
print(f" - {entity_type}: {count} items")
|
|
else:
|
|
print(f" - {entity_type}: metadata")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
print("\nTroubleshooting:")
|
|
print("1. Check your AMO_CRM_ACCESS_TOKEN in .env file")
|
|
print("2. Make sure the token is valid and not expired")
|
|
print("3. Verify you have access to wecheap.amocrm.ru")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|