Maxim Snesarev
33d6bb7ebd
- Updated database configuration to switch from SQLite to PostgreSQL, including changes to alembic.ini, Docker Compose, and environment settings. - Refactored application code to utilize PostgreSQL database adapters, ensuring compatibility with the new database structure. - Enhanced API routes and data handling to support the new database, including adjustments in data models and query logic. - Introduced new job processing mechanisms for full synchronization of AMO CRM entities, leveraging FastStream for background tasks. - Improved logging and error handling across the application to facilitate better monitoring and debugging. - Removed obsolete SQLite adapter files and migrations, streamlining the project structure for PostgreSQL integration.
724 lines
27 KiB
Python
724 lines
27 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
AMO CRM Data Import Script
|
||
Comprehensive script to import all AMO CRM data into SQLite database
|
||
|
||
Usage:
|
||
python scripts/import_amocrm_data.py [options]
|
||
|
||
Options:
|
||
--entities: Comma-separated list of entities to import (default: all)
|
||
--limit: Limit per entity (default: 1000)
|
||
--batch-size: Batch size for processing (default: 100)
|
||
--dry-run: Preview what would be imported without saving
|
||
--force: Overwrite existing data
|
||
--skip-relationships: Skip relationship processing
|
||
--verbose: Detailed output
|
||
|
||
Examples:
|
||
python scripts/import_amocrm_data.py --entities=users,deals --limit=500
|
||
python scripts/import_amocrm_data.py --dry-run --verbose
|
||
python scripts/import_amocrm_data.py --force
|
||
"""
|
||
|
||
import asyncio
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Dict, Any, List, Optional, Tuple, Union
|
||
from dataclasses import dataclass
|
||
import time
|
||
|
||
# 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 adapters.postgres.database import SessionLocal, init_db
|
||
from adapters.postgres.models import (
|
||
User, Pipeline, PipelineStage, Company, Contact, Deal, Event, CustomField,
|
||
deal_contacts, deal_companies, contact_companies
|
||
)
|
||
from utils.config import settings
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
from sqlalchemy import text
|
||
|
||
|
||
@dataclass
|
||
class ImportStats:
|
||
"""Statistics for import operations"""
|
||
entity_type: str
|
||
fetched: int = 0
|
||
processed: int = 0
|
||
created: int = 0
|
||
updated: int = 0
|
||
errors: int = 0
|
||
skipped: int = 0
|
||
start_time: float = 0
|
||
end_time: float = 0
|
||
|
||
@property
|
||
def duration(self) -> float:
|
||
return self.end_time - self.start_time if self.end_time else time.time() - self.start_time
|
||
|
||
def __str__(self) -> str:
|
||
return (f"{self.entity_type}: {self.processed}/{self.fetched} processed, "
|
||
f"{self.created} created, {self.updated} updated, "
|
||
f"{self.errors} errors, {self.skipped} skipped "
|
||
f"({self.duration:.1f}s)")
|
||
|
||
|
||
class AMOCRMImporter:
|
||
"""Main importer class for AMO CRM data"""
|
||
|
||
SUPPORTED_ENTITIES = [
|
||
'users', 'pipelines', 'companies', 'contacts', 'deals', 'events'
|
||
]
|
||
|
||
ENTITY_DEPENDENCIES = {
|
||
'users': [],
|
||
'pipelines': [],
|
||
'companies': ['users'],
|
||
'contacts': ['users'],
|
||
'deals': ['users', 'pipelines', 'companies', 'contacts'],
|
||
'events': ['users']
|
||
}
|
||
|
||
def __init__(self,
|
||
limit_per_entity: int = 1000,
|
||
batch_size: int = 100,
|
||
dry_run: bool = False,
|
||
force_overwrite: bool = False,
|
||
skip_relationships: bool = False,
|
||
verbose: bool = False):
|
||
|
||
self.client = AmoCRMClient()
|
||
self.limit_per_entity = limit_per_entity
|
||
self.batch_size = batch_size
|
||
self.dry_run = dry_run
|
||
self.force_overwrite = force_overwrite
|
||
self.skip_relationships = skip_relationships
|
||
self.verbose = verbose
|
||
|
||
self.stats: Dict[str, ImportStats] = {}
|
||
self.custom_fields_cache: Dict[str, Dict[int, Dict]] = {}
|
||
self._db_initialized = False
|
||
|
||
def log(self, message: str, level: str = "INFO") -> None:
|
||
"""Log message with timestamp"""
|
||
timestamp = datetime.now().strftime("%H:%M:%S")
|
||
prefix = {"INFO": "ℹ️", "SUCCESS": "✅", "WARNING": "⚠️", "ERROR": "❌"}
|
||
print(f"[{timestamp}] {prefix.get(level, 'ℹ️')} {message}")
|
||
|
||
def log_verbose(self, message: str) -> None:
|
||
"""Log verbose message"""
|
||
if self.verbose:
|
||
self.log(message)
|
||
|
||
async def import_all_data(self, entities: Optional[List[str]] = None) -> Dict[str, ImportStats]:
|
||
"""Import all or specified entities"""
|
||
# Initialize database if not already done
|
||
if not self._db_initialized:
|
||
await init_db()
|
||
self._db_initialized = True
|
||
|
||
entities_to_import = entities or self.SUPPORTED_ENTITIES
|
||
|
||
# Validate entities
|
||
invalid_entities = [e for e in entities_to_import if e not in self.SUPPORTED_ENTITIES]
|
||
if invalid_entities:
|
||
raise ValueError(f"Invalid entities: {invalid_entities}")
|
||
|
||
# Order entities by dependencies
|
||
ordered_entities = self._order_entities_by_dependencies(entities_to_import)
|
||
|
||
self.log(f"Starting import for entities: {', '.join(ordered_entities)}")
|
||
if self.dry_run:
|
||
self.log("DRY RUN MODE - No data will be saved", "WARNING")
|
||
|
||
# Import each entity in order
|
||
for entity_type in ordered_entities:
|
||
try:
|
||
await self._import_entity(entity_type)
|
||
except Exception as e:
|
||
self.log(f"Failed to import {entity_type}: {str(e)}", "ERROR")
|
||
self.stats[entity_type] = ImportStats(entity_type)
|
||
self.stats[entity_type].errors = 1
|
||
|
||
# Print summary
|
||
self._print_summary()
|
||
|
||
return self.stats
|
||
|
||
def _order_entities_by_dependencies(self, entities: List[str]) -> List[str]:
|
||
"""Order entities based on their dependencies"""
|
||
ordered = []
|
||
remaining = entities.copy()
|
||
|
||
while remaining:
|
||
# Find entities with no unresolved dependencies
|
||
ready = []
|
||
for entity in remaining:
|
||
deps = self.ENTITY_DEPENDENCIES[entity]
|
||
if all(dep in ordered or dep not in entities for dep in deps):
|
||
ready.append(entity)
|
||
|
||
if not ready:
|
||
# Circular dependency or missing dependency
|
||
self.log(f"Cannot resolve dependencies for: {remaining}", "WARNING")
|
||
ready = remaining # Import anyway
|
||
|
||
ordered.extend(ready)
|
||
for entity in ready:
|
||
remaining.remove(entity)
|
||
|
||
return ordered
|
||
|
||
async def _import_entity(self, entity_type: str) -> None:
|
||
"""Import specific entity type"""
|
||
self.log(f"Importing {entity_type}...")
|
||
|
||
stats = ImportStats(entity_type)
|
||
stats.start_time = time.time()
|
||
self.stats[entity_type] = stats
|
||
|
||
try:
|
||
# Fetch data from AMO CRM
|
||
data = await self._fetch_entity_data(entity_type)
|
||
|
||
if not data:
|
||
self.log(f"No data found for {entity_type}", "WARNING")
|
||
return
|
||
|
||
stats.fetched = len(data)
|
||
self.log_verbose(f"Fetched {stats.fetched} {entity_type} records")
|
||
|
||
# Process data in batches
|
||
if not self.dry_run:
|
||
await self._process_entity_data(entity_type, data, stats)
|
||
else:
|
||
self._preview_entity_data(entity_type, data, stats)
|
||
|
||
except Exception as e:
|
||
self.log(f"Error importing {entity_type}: {str(e)}", "ERROR")
|
||
stats.errors += 1
|
||
finally:
|
||
stats.end_time = time.time()
|
||
self.log(f"Completed {entity_type}: {stats}")
|
||
|
||
async def _fetch_entity_data(self, entity_type: str) -> List[Dict[str, Any]]:
|
||
"""Fetch data for entity type from AMO CRM"""
|
||
all_data: List[Dict[str, Any]] = []
|
||
page = 1
|
||
|
||
while len(all_data) < self.limit_per_entity:
|
||
try:
|
||
if entity_type == 'users':
|
||
response = await self.client.get_users(limit=min(250, self.limit_per_entity - len(all_data)))
|
||
elif entity_type == 'pipelines':
|
||
response = await self.client.get_pipelines()
|
||
elif entity_type == 'companies':
|
||
response = await self.client.get_companies(
|
||
limit=min(250, self.limit_per_entity - len(all_data)),
|
||
page=page
|
||
)
|
||
elif entity_type == 'contacts':
|
||
response = await self.client.get_contacts(
|
||
limit=min(250, self.limit_per_entity - len(all_data)),
|
||
page=page
|
||
)
|
||
elif entity_type == 'deals':
|
||
response = await self.client.get_deals(
|
||
limit=min(250, self.limit_per_entity - len(all_data)),
|
||
page=page
|
||
)
|
||
elif entity_type == 'events':
|
||
response = await self.client.get_events(
|
||
limit=min(250, self.limit_per_entity - len(all_data)),
|
||
page=page
|
||
)
|
||
else:
|
||
raise ValueError(f"Unsupported entity type: {entity_type}")
|
||
|
||
# Extract embedded data
|
||
embedded_key = self._get_embedded_key(entity_type)
|
||
embedded_data = response.get('_embedded', {}).get(embedded_key, [])
|
||
|
||
if not embedded_data:
|
||
break
|
||
|
||
all_data.extend(embedded_data)
|
||
|
||
# Handle pipelines special case (no pagination)
|
||
if entity_type == 'pipelines':
|
||
break
|
||
|
||
page += 1
|
||
|
||
except Exception as e:
|
||
self.log(f"Error fetching {entity_type} page {page}: {str(e)}", "ERROR")
|
||
break
|
||
|
||
return all_data[:self.limit_per_entity]
|
||
|
||
def _get_embedded_key(self, entity_type: str) -> str:
|
||
"""Get the embedded key for entity type"""
|
||
mapping = {
|
||
'users': 'users',
|
||
'pipelines': 'pipelines',
|
||
'companies': 'companies',
|
||
'contacts': 'contacts',
|
||
'deals': 'leads', # AMO CRM uses 'leads' for deals
|
||
'events': 'events'
|
||
}
|
||
return mapping[entity_type]
|
||
|
||
def _preview_entity_data(self, entity_type: str, data: List[Dict], stats: ImportStats) -> None:
|
||
"""Preview data without saving (dry run)"""
|
||
stats.processed = len(data)
|
||
|
||
if data:
|
||
sample = data[0]
|
||
self.log(f"Sample {entity_type} record:")
|
||
self.log(f" ID: {sample.get('id')}")
|
||
self.log(f" Name: {sample.get('name', 'N/A')}")
|
||
|
||
if 'custom_fields_values' in sample:
|
||
custom_fields = len(sample['custom_fields_values'])
|
||
self.log(f" Custom fields: {custom_fields}")
|
||
|
||
if '_embedded' in sample:
|
||
embedded = sample['_embedded']
|
||
for key, value in embedded.items():
|
||
self.log(f" Embedded {key}: {len(value) if isinstance(value, list) else 1}")
|
||
|
||
async def _process_entity_data(self, entity_type: str, data: List[Dict], stats: ImportStats) -> None:
|
||
"""Process and save entity data to database"""
|
||
db = SessionLocal()
|
||
try:
|
||
# Load custom fields metadata if needed
|
||
if entity_type in ['companies', 'contacts', 'deals']:
|
||
await self._load_custom_fields_metadata(entity_type)
|
||
|
||
# Process in batches
|
||
for i in range(0, len(data), self.batch_size):
|
||
batch = data[i:i + self.batch_size]
|
||
await self._process_batch(db, entity_type, batch, stats)
|
||
|
||
if self.verbose and i > 0:
|
||
self.log_verbose(f"Processed {min(i + self.batch_size, len(data))}/{len(data)} {entity_type}")
|
||
|
||
db.commit()
|
||
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise e
|
||
finally:
|
||
db.close()
|
||
|
||
async def _process_batch(self, db: Session, entity_type: str, batch: List[Dict], stats: ImportStats) -> None:
|
||
"""Process a batch of entity data"""
|
||
for item_data in batch:
|
||
try:
|
||
await self._process_single_item(db, entity_type, item_data, stats)
|
||
except Exception as e:
|
||
stats.errors += 1
|
||
self.log_verbose(f"Error processing {entity_type} ID {item_data.get('id')}: {str(e)}")
|
||
|
||
async def _process_single_item(self, db: Session, entity_type: str, item_data: Dict, stats: ImportStats) -> None:
|
||
"""Process single entity item"""
|
||
entity_id = item_data.get('id')
|
||
if not entity_id:
|
||
stats.skipped += 1
|
||
return
|
||
|
||
# Check if entity exists
|
||
model_class = self._get_model_class(entity_type)
|
||
existing = db.query(model_class).filter(model_class.id == entity_id).first()
|
||
|
||
if existing and not self.force_overwrite:
|
||
stats.skipped += 1
|
||
return
|
||
|
||
# Create or update entity
|
||
if existing:
|
||
entity = existing
|
||
stats.updated += 1
|
||
else:
|
||
entity = model_class()
|
||
stats.created += 1
|
||
|
||
# Map data to entity
|
||
self._map_entity_data(entity, entity_type, item_data)
|
||
|
||
if not existing:
|
||
db.add(entity)
|
||
|
||
stats.processed += 1
|
||
|
||
# Process custom fields
|
||
if entity_type in ['companies', 'contacts', 'deals']:
|
||
await self._process_custom_fields(db, entity_type, entity_id, item_data)
|
||
|
||
# Process relationships
|
||
if not self.skip_relationships and entity_type == 'deals':
|
||
await self._process_deal_relationships(db, entity, item_data)
|
||
elif not self.skip_relationships and entity_type == 'pipelines':
|
||
await self._process_pipeline_stages(db, entity, item_data)
|
||
|
||
def _get_model_class(self, entity_type: str) -> Any:
|
||
"""Get SQLAlchemy model class for entity type"""
|
||
mapping = {
|
||
'users': User,
|
||
'pipelines': Pipeline,
|
||
'companies': Company,
|
||
'contacts': Contact,
|
||
'deals': Deal,
|
||
'events': Event
|
||
}
|
||
return mapping[entity_type]
|
||
|
||
def _map_entity_data(self, entity: Any, entity_type: str, data: Dict) -> None:
|
||
"""Map AMO CRM data to entity model"""
|
||
entity.id = data.get('id')
|
||
entity.raw_data = data
|
||
|
||
if entity_type == 'users':
|
||
entity.name = data.get('name', '')
|
||
entity.email = data.get('email')
|
||
entity.is_active = not data.get('is_deleted', False)
|
||
entity.created_at = data.get('created_at')
|
||
entity.updated_at = data.get('updated_at')
|
||
|
||
elif entity_type == 'pipelines':
|
||
entity.name = data.get('name', '')
|
||
entity.sort = data.get('sort')
|
||
entity.is_main = data.get('is_main', False)
|
||
entity.is_unsorted = data.get('is_unsorted', False)
|
||
entity.is_archive = data.get('is_archive', False)
|
||
entity.account_id = data.get('account_id')
|
||
entity.created_at = data.get('created_at')
|
||
entity.updated_at = data.get('updated_at')
|
||
|
||
elif entity_type in ['companies', 'contacts']:
|
||
entity.name = data.get('name', '')
|
||
entity.responsible_user_id = data.get('responsible_user_id')
|
||
entity.group_id = data.get('group_id')
|
||
entity.created_by = data.get('created_by')
|
||
entity.updated_by = data.get('updated_by')
|
||
entity.created_at = data.get('created_at')
|
||
entity.updated_at = data.get('updated_at')
|
||
entity.closest_task_at = data.get('closest_task_at')
|
||
entity.is_deleted = data.get('is_deleted', False)
|
||
|
||
if entity_type == 'contacts':
|
||
entity.first_name = data.get('first_name')
|
||
entity.last_name = data.get('last_name')
|
||
|
||
elif entity_type == 'deals':
|
||
entity.name = data.get('name', '')
|
||
entity.price = data.get('price', 0)
|
||
entity.responsible_user_id = data.get('responsible_user_id')
|
||
entity.group_id = data.get('group_id')
|
||
entity.status_id = data.get('status_id')
|
||
entity.pipeline_id = data.get('pipeline_id')
|
||
entity.loss_reason_id = data.get('loss_reason_id')
|
||
entity.created_by = data.get('created_by')
|
||
entity.updated_by = data.get('updated_by')
|
||
entity.closed_at = data.get('closed_at')
|
||
entity.created_at = data.get('created_at')
|
||
entity.updated_at = data.get('updated_at')
|
||
entity.closest_task_at = data.get('closest_task_at')
|
||
entity.is_deleted = data.get('is_deleted', False)
|
||
|
||
elif entity_type == 'events':
|
||
entity.type = data.get('type', '')
|
||
entity.entity_id = data.get('entity_id')
|
||
entity.entity_type = data.get('entity_type')
|
||
entity.created_by = data.get('created_by')
|
||
entity.created_at = data.get('created_at')
|
||
entity.value_after = data.get('value_after')
|
||
entity.value_before = data.get('value_before')
|
||
entity.account_id = data.get('account_id')
|
||
|
||
async def _load_custom_fields_metadata(self, entity_type: str) -> None:
|
||
"""Load custom fields metadata for entity type"""
|
||
if entity_type in self.custom_fields_cache:
|
||
return
|
||
|
||
try:
|
||
# Map entity type to AMO CRM API endpoint
|
||
api_entity_type = 'leads' if entity_type == 'deals' else entity_type
|
||
response = await self.client.get_custom_fields(api_entity_type)
|
||
|
||
fields_data = response.get('_embedded', {}).get('custom_fields', [])
|
||
fields_dict = {field['id']: field for field in fields_data}
|
||
|
||
self.custom_fields_cache[entity_type] = fields_dict
|
||
self.log_verbose(f"Loaded {len(fields_dict)} custom fields for {entity_type}")
|
||
|
||
except Exception as e:
|
||
self.log(f"Warning: Could not load custom fields for {entity_type}: {str(e)}", "WARNING")
|
||
self.custom_fields_cache[entity_type] = {}
|
||
|
||
async def _process_custom_fields(self, db: Session, entity_type: str, entity_id: int, data: Dict) -> None:
|
||
"""Process custom fields for entity"""
|
||
custom_fields_values = data.get('custom_fields_values', [])
|
||
if not custom_fields_values:
|
||
return
|
||
|
||
# Delete existing custom fields if force overwrite
|
||
if self.force_overwrite:
|
||
db.query(CustomField).filter(
|
||
CustomField.entity_type == entity_type,
|
||
CustomField.entity_id == entity_id
|
||
).delete()
|
||
|
||
fields_metadata = self.custom_fields_cache.get(entity_type, {})
|
||
|
||
for field_data in custom_fields_values:
|
||
field_id = field_data.get('field_id')
|
||
if not field_id:
|
||
continue
|
||
|
||
field_meta = fields_metadata.get(field_id, {})
|
||
field_name = field_meta.get('name', f'field_{field_id}')
|
||
field_type = field_meta.get('type', 'unknown')
|
||
|
||
# Check if field already exists
|
||
existing_field = db.query(CustomField).filter(
|
||
CustomField.entity_type == entity_type,
|
||
CustomField.entity_id == entity_id,
|
||
CustomField.field_id == field_id
|
||
).first()
|
||
|
||
if existing_field and not self.force_overwrite:
|
||
continue
|
||
|
||
# Create custom field record
|
||
custom_field = existing_field or CustomField()
|
||
custom_field.entity_type = entity_type
|
||
custom_field.entity_id = entity_id
|
||
custom_field.field_id = field_id
|
||
custom_field.field_name = field_name
|
||
custom_field.field_type = field_type
|
||
custom_field.is_custom = True
|
||
custom_field.created_at = int(time.time())
|
||
custom_field.updated_at = int(time.time())
|
||
|
||
# Process field values
|
||
values = field_data.get('values', [])
|
||
if values:
|
||
first_value = values[0]
|
||
value_str = str(first_value.get('value', ''))
|
||
|
||
custom_field.field_value = value_str
|
||
|
||
# Type-specific processing
|
||
if field_type == 'numeric' and value_str.replace('.', '').replace('-', '').isdigit():
|
||
custom_field.field_value_numeric = float(value_str)
|
||
elif field_type == 'date' and value_str.isdigit():
|
||
custom_field.field_value_date = int(value_str)
|
||
elif field_type in ['select', 'multiselect']:
|
||
# Handle multiple values
|
||
all_values = [str(v.get('value', '')) for v in values]
|
||
custom_field.field_value = ', '.join(all_values)
|
||
|
||
if not existing_field:
|
||
db.add(custom_field)
|
||
|
||
async def _process_deal_relationships(self, db: Session, deal: Deal, data: Dict) -> None:
|
||
"""Process deal relationships (contacts and companies)"""
|
||
embedded = data.get('_embedded', {})
|
||
|
||
# Process contacts
|
||
contacts_data = embedded.get('contacts', [])
|
||
for contact_data in contacts_data:
|
||
contact_id = contact_data.get('id')
|
||
is_main = contact_data.get('is_main', False)
|
||
|
||
if contact_id:
|
||
# Check if relationship already exists
|
||
existing = db.execute(
|
||
text("SELECT 1 FROM amo_deal_contacts WHERE deal_id = :deal_id AND contact_id = :contact_id"),
|
||
{"deal_id": deal.id, "contact_id": contact_id}
|
||
).first()
|
||
|
||
if not existing:
|
||
db.execute(
|
||
text("INSERT INTO amo_deal_contacts (deal_id, contact_id, is_main) VALUES (:deal_id, :contact_id, :is_main)"),
|
||
{"deal_id": deal.id, "contact_id": contact_id, "is_main": is_main}
|
||
)
|
||
|
||
# Process companies
|
||
companies_data = embedded.get('companies', [])
|
||
for company_data in companies_data:
|
||
company_id = company_data.get('id')
|
||
is_main = company_data.get('is_main', False)
|
||
|
||
if company_id:
|
||
# Check if relationship already exists
|
||
existing = db.execute(
|
||
text("SELECT 1 FROM amo_deal_companies WHERE deal_id = :deal_id AND company_id = :company_id"),
|
||
{"deal_id": deal.id, "company_id": company_id}
|
||
).first()
|
||
|
||
if not existing:
|
||
db.execute(
|
||
text("INSERT INTO amo_deal_companies (deal_id, company_id, is_main) VALUES (:deal_id, :company_id, :is_main)"),
|
||
{"deal_id": deal.id, "company_id": company_id, "is_main": is_main}
|
||
)
|
||
|
||
async def _process_pipeline_stages(self, db: Session, pipeline: Pipeline, data: Dict) -> None:
|
||
"""Process pipeline stages"""
|
||
embedded = data.get('_embedded', {})
|
||
stages_data = embedded.get('statuses', [])
|
||
|
||
for stage_data in stages_data:
|
||
stage_id = stage_data.get('id')
|
||
if not stage_id:
|
||
continue
|
||
|
||
# Check if stage exists
|
||
existing_stage = db.query(PipelineStage).filter(PipelineStage.id == stage_id).first()
|
||
|
||
if existing_stage and not self.force_overwrite:
|
||
continue
|
||
|
||
stage = existing_stage or PipelineStage()
|
||
stage.id = stage_id
|
||
stage.pipeline_id = pipeline.id
|
||
stage.name = stage_data.get('name', '')
|
||
stage.sort = stage_data.get('sort')
|
||
stage.is_editable = stage_data.get('is_editable', True)
|
||
stage.color = stage_data.get('color')
|
||
stage.created_at = stage_data.get('created_at')
|
||
stage.updated_at = stage_data.get('updated_at')
|
||
stage.raw_data = stage_data
|
||
|
||
if not existing_stage:
|
||
db.add(stage)
|
||
|
||
def _print_summary(self) -> None:
|
||
"""Print import summary"""
|
||
self.log("\n" + "="*60)
|
||
self.log("IMPORT SUMMARY", "SUCCESS")
|
||
self.log("="*60)
|
||
|
||
total_stats = ImportStats("TOTAL")
|
||
|
||
for entity_type, stats in self.stats.items():
|
||
self.log(f"{stats}")
|
||
|
||
total_stats.fetched += stats.fetched
|
||
total_stats.processed += stats.processed
|
||
total_stats.created += stats.created
|
||
total_stats.updated += stats.updated
|
||
total_stats.errors += stats.errors
|
||
total_stats.skipped += stats.skipped
|
||
|
||
self.log("-" * 60)
|
||
self.log(f"TOTAL: {total_stats.processed}/{total_stats.fetched} processed, "
|
||
f"{total_stats.created} created, {total_stats.updated} updated, "
|
||
f"{total_stats.errors} errors, {total_stats.skipped} skipped")
|
||
|
||
if total_stats.errors > 0:
|
||
self.log(f"⚠️ {total_stats.errors} errors occurred during import", "WARNING")
|
||
|
||
if self.dry_run:
|
||
self.log("🔍 This was a DRY RUN - no data was actually saved", "WARNING")
|
||
|
||
|
||
async def main() -> int:
|
||
"""Main function"""
|
||
parser = argparse.ArgumentParser(description="Import AMO CRM data into database")
|
||
parser.add_argument(
|
||
'--entities',
|
||
type=str,
|
||
help='Comma-separated list of entities to import (default: all)',
|
||
default=None
|
||
)
|
||
parser.add_argument(
|
||
'--limit',
|
||
type=int,
|
||
help='Limit per entity (default: 1000)',
|
||
default=1000
|
||
)
|
||
parser.add_argument(
|
||
'--batch-size',
|
||
type=int,
|
||
help='Batch size for processing (default: 100)',
|
||
default=100
|
||
)
|
||
parser.add_argument(
|
||
'--dry-run',
|
||
action='store_true',
|
||
help='Preview what would be imported without saving'
|
||
)
|
||
parser.add_argument(
|
||
'--force',
|
||
action='store_true',
|
||
help='Overwrite existing data'
|
||
)
|
||
parser.add_argument(
|
||
'--skip-relationships',
|
||
action='store_true',
|
||
help='Skip relationship processing'
|
||
)
|
||
parser.add_argument(
|
||
'--verbose',
|
||
action='store_true',
|
||
help='Detailed output'
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
# Parse entities
|
||
entities = None
|
||
if args.entities:
|
||
entities = [e.strip() for e in args.entities.split(',')]
|
||
invalid = [e for e in entities if e not in AMOCRMImporter.SUPPORTED_ENTITIES]
|
||
if invalid:
|
||
print(f"❌ Invalid entities: {invalid}")
|
||
print(f"Supported entities: {', '.join(AMOCRMImporter.SUPPORTED_ENTITIES)}")
|
||
return 1
|
||
|
||
# Check configuration
|
||
if not settings.AMO_CRM_ACCESS_TOKEN:
|
||
print("❌ Error: AMO_CRM_ACCESS_TOKEN not set!")
|
||
print("Please set it in your .env file")
|
||
return 1
|
||
|
||
try:
|
||
# Create importer
|
||
importer = AMOCRMImporter(
|
||
limit_per_entity=args.limit,
|
||
batch_size=args.batch_size,
|
||
dry_run=args.dry_run,
|
||
force_overwrite=args.force,
|
||
skip_relationships=args.skip_relationships,
|
||
verbose=args.verbose
|
||
)
|
||
|
||
# Run import
|
||
stats = await importer.import_all_data(entities)
|
||
|
||
# Check for errors
|
||
total_errors = sum(s.errors for s in stats.values())
|
||
return 1 if total_errors > 0 else 0
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n❌ Import interrupted by user")
|
||
return 1
|
||
except Exception as e:
|
||
print(f"❌ Import failed: {str(e)}")
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
exit_code = asyncio.run(main())
|
||
sys.exit(exit_code)
|