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.
331 lines
12 KiB
Python
331 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Data Validation Script for AMO CRM Import
|
||
Validates the integrity and completeness of imported data
|
||
"""
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
from typing import Dict, Any, List
|
||
|
||
# Add project root to Python path
|
||
project_root = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
from adapters.postgres.database import SessionLocal
|
||
from adapters.postgres.models import (
|
||
User, Pipeline, PipelineStage, Company, Contact, Deal, Event, CustomField
|
||
)
|
||
from sqlalchemy import func, text
|
||
|
||
|
||
class DataValidator:
|
||
"""Validates imported AMO CRM data"""
|
||
|
||
def __init__(self):
|
||
self.db = SessionLocal()
|
||
self.issues: List[str] = []
|
||
self.warnings: List[str] = []
|
||
|
||
def log_issue(self, message: str) -> None:
|
||
"""Log a data integrity issue"""
|
||
self.issues.append(message)
|
||
print(f"❌ ISSUE: {message}")
|
||
|
||
def log_warning(self, message: str) -> None:
|
||
"""Log a data warning"""
|
||
self.warnings.append(message)
|
||
print(f"⚠️ WARNING: {message}")
|
||
|
||
def log_info(self, message: str) -> None:
|
||
"""Log informational message"""
|
||
print(f"ℹ️ {message}")
|
||
|
||
def log_success(self, message: str) -> None:
|
||
"""Log success message"""
|
||
print(f"✅ {message}")
|
||
|
||
def validate_all(self) -> bool:
|
||
"""Run all validation checks"""
|
||
print("🔍 Validating AMO CRM Data")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
# Basic counts
|
||
self.validate_basic_counts()
|
||
|
||
# Data integrity
|
||
self.validate_foreign_keys()
|
||
self.validate_required_fields()
|
||
self.validate_data_consistency()
|
||
|
||
# Relationships
|
||
self.validate_relationships()
|
||
|
||
# Custom fields
|
||
self.validate_custom_fields()
|
||
|
||
# Summary
|
||
self.print_summary()
|
||
|
||
return len(self.issues) == 0
|
||
|
||
except Exception as e:
|
||
self.log_issue(f"Validation failed with error: {str(e)}")
|
||
return False
|
||
finally:
|
||
self.db.close()
|
||
|
||
def validate_basic_counts(self) -> None:
|
||
"""Validate basic record counts"""
|
||
self.log_info("Checking record counts...")
|
||
|
||
counts = {
|
||
'Users': self.db.query(User).count(),
|
||
'Pipelines': self.db.query(Pipeline).count(),
|
||
'Pipeline Stages': self.db.query(PipelineStage).count(),
|
||
'Companies': self.db.query(Company).count(),
|
||
'Contacts': self.db.query(Contact).count(),
|
||
'Deals': self.db.query(Deal).count(),
|
||
'Events': self.db.query(Event).count(),
|
||
'Custom Fields': self.db.query(CustomField).count(),
|
||
}
|
||
|
||
for entity, count in counts.items():
|
||
if count > 0:
|
||
self.log_success(f"{entity}: {count:,} records")
|
||
else:
|
||
self.log_warning(f"{entity}: No records found")
|
||
|
||
# Check for reasonable ratios
|
||
if counts['Users'] == 0 and counts['Deals'] > 0:
|
||
self.log_issue("Deals exist but no users found - foreign key issues likely")
|
||
|
||
if counts['Pipelines'] == 0 and counts['Deals'] > 0:
|
||
self.log_issue("Deals exist but no pipelines found - foreign key issues likely")
|
||
|
||
def validate_foreign_keys(self) -> None:
|
||
"""Validate foreign key relationships"""
|
||
self.log_info("Checking foreign key integrity...")
|
||
|
||
# Users references
|
||
queries = [
|
||
("Companies with invalid responsible_user_id", """
|
||
SELECT COUNT(*) FROM amo_companies c
|
||
WHERE c.responsible_user_id IS NOT NULL
|
||
AND c.responsible_user_id NOT IN (SELECT id FROM amo_users)
|
||
"""),
|
||
("Contacts with invalid responsible_user_id", """
|
||
SELECT COUNT(*) FROM amo_contacts c
|
||
WHERE c.responsible_user_id IS NOT NULL
|
||
AND c.responsible_user_id NOT IN (SELECT id FROM amo_users)
|
||
"""),
|
||
("Deals with invalid responsible_user_id", """
|
||
SELECT COUNT(*) FROM amo_deals d
|
||
WHERE d.responsible_user_id IS NOT NULL
|
||
AND d.responsible_user_id NOT IN (SELECT id FROM amo_users)
|
||
"""),
|
||
("Deals with invalid pipeline_id", """
|
||
SELECT COUNT(*) FROM amo_deals d
|
||
WHERE d.pipeline_id IS NOT NULL
|
||
AND d.pipeline_id NOT IN (SELECT id FROM amo_pipelines)
|
||
"""),
|
||
("Deals with invalid status_id", """
|
||
SELECT COUNT(*) FROM amo_deals d
|
||
WHERE d.status_id IS NOT NULL
|
||
AND d.status_id NOT IN (SELECT id FROM amo_pipeline_stages)
|
||
"""),
|
||
]
|
||
|
||
for description, query in queries:
|
||
result = self.db.execute(text(query)).scalar()
|
||
if result > 0:
|
||
self.log_issue(f"{description}: {result}")
|
||
else:
|
||
self.log_success(f"{description}: OK")
|
||
|
||
def validate_required_fields(self) -> None:
|
||
"""Validate required fields are not empty"""
|
||
self.log_info("Checking required fields...")
|
||
|
||
# Check for empty names
|
||
entities_with_names = [
|
||
(User, "Users"),
|
||
(Pipeline, "Pipelines"),
|
||
(Company, "Companies"),
|
||
(Contact, "Contacts"),
|
||
(Deal, "Deals")
|
||
]
|
||
|
||
for model, name in entities_with_names:
|
||
empty_names = self.db.query(model).filter(
|
||
(model.name == '') | (model.name.is_(None))
|
||
).count()
|
||
|
||
if empty_names > 0:
|
||
self.log_warning(f"{name} with empty names: {empty_names}")
|
||
else:
|
||
self.log_success(f"{name} names: OK")
|
||
|
||
def validate_data_consistency(self) -> None:
|
||
"""Validate data consistency"""
|
||
self.log_info("Checking data consistency...")
|
||
|
||
# Check for future dates
|
||
future_deals = self.db.query(Deal).filter(
|
||
Deal.created_at > int(datetime.now().timestamp())
|
||
).count()
|
||
|
||
if future_deals > 0:
|
||
self.log_warning(f"Deals with future created_at dates: {future_deals}")
|
||
|
||
# Check for very old dates (before 2010)
|
||
old_deals = self.db.query(Deal).filter(
|
||
Deal.created_at < 1262304000 # 2010-01-01
|
||
).count()
|
||
|
||
if old_deals > 0:
|
||
self.log_warning(f"Deals with very old dates (before 2010): {old_deals}")
|
||
|
||
# Check for negative prices
|
||
negative_prices = self.db.query(Deal).filter(Deal.price < 0).count()
|
||
|
||
if negative_prices > 0:
|
||
self.log_warning(f"Deals with negative prices: {negative_prices}")
|
||
|
||
self.log_success("Data consistency checks completed")
|
||
|
||
def validate_relationships(self) -> None:
|
||
"""Validate relationship tables"""
|
||
self.log_info("Checking relationship integrity...")
|
||
|
||
# Check deal-contact relationships
|
||
invalid_deal_contacts = self.db.execute(text("""
|
||
SELECT COUNT(*) FROM amo_deal_contacts dc
|
||
WHERE dc.deal_id NOT IN (SELECT id FROM amo_deals)
|
||
OR dc.contact_id NOT IN (SELECT id FROM amo_contacts)
|
||
""")).scalar()
|
||
|
||
if invalid_deal_contacts > 0:
|
||
self.log_issue(f"Invalid deal-contact relationships: {invalid_deal_contacts}")
|
||
else:
|
||
self.log_success("Deal-contact relationships: OK")
|
||
|
||
# Check deal-company relationships
|
||
invalid_deal_companies = self.db.execute(text("""
|
||
SELECT COUNT(*) FROM amo_deal_companies dc
|
||
WHERE dc.deal_id NOT IN (SELECT id FROM amo_deals)
|
||
OR dc.company_id NOT IN (SELECT id FROM amo_companies)
|
||
""")).scalar()
|
||
|
||
if invalid_deal_companies > 0:
|
||
self.log_issue(f"Invalid deal-company relationships: {invalid_deal_companies}")
|
||
else:
|
||
self.log_success("Deal-company relationships: OK")
|
||
|
||
# Check contact-company relationships
|
||
invalid_contact_companies = self.db.execute(text("""
|
||
SELECT COUNT(*) FROM amo_contact_companies cc
|
||
WHERE cc.contact_id NOT IN (SELECT id FROM amo_contacts)
|
||
OR cc.company_id NOT IN (SELECT id FROM amo_companies)
|
||
""")).scalar()
|
||
|
||
if invalid_contact_companies > 0:
|
||
self.log_issue(f"Invalid contact-company relationships: {invalid_contact_companies}")
|
||
else:
|
||
self.log_success("Contact-company relationships: OK")
|
||
|
||
def validate_custom_fields(self) -> None:
|
||
"""Validate custom fields"""
|
||
self.log_info("Checking custom fields...")
|
||
|
||
# Check for custom fields with missing entities
|
||
invalid_custom_fields = self.db.execute(text("""
|
||
SELECT cf.entity_type, COUNT(*) as count
|
||
FROM amo_custom_fields cf
|
||
LEFT JOIN amo_deals d ON cf.entity_type = 'deals' AND cf.entity_id = d.id
|
||
LEFT JOIN amo_contacts c ON cf.entity_type = 'contacts' AND cf.entity_id = c.id
|
||
LEFT JOIN amo_companies co ON cf.entity_type = 'companies' AND cf.entity_id = co.id
|
||
WHERE d.id IS NULL AND c.id IS NULL AND co.id IS NULL
|
||
GROUP BY cf.entity_type
|
||
""")).fetchall()
|
||
|
||
if invalid_custom_fields:
|
||
for entity_type, count in invalid_custom_fields:
|
||
self.log_issue(f"Custom fields for non-existent {entity_type}: {count}")
|
||
else:
|
||
self.log_success("Custom fields entity references: OK")
|
||
|
||
# Check custom field types distribution
|
||
field_types = self.db.execute(text("""
|
||
SELECT field_type, COUNT(*) as count
|
||
FROM amo_custom_fields
|
||
GROUP BY field_type
|
||
ORDER BY count DESC
|
||
""")).fetchall()
|
||
|
||
if field_types:
|
||
self.log_info("Custom field types distribution:")
|
||
for field_type, count in field_types:
|
||
print(f" {field_type}: {count:,} fields")
|
||
else:
|
||
self.log_warning("No custom fields found")
|
||
|
||
def print_summary(self) -> None:
|
||
"""Print validation summary"""
|
||
print("\n" + "=" * 50)
|
||
print("VALIDATION SUMMARY")
|
||
print("=" * 50)
|
||
|
||
if not self.issues and not self.warnings:
|
||
self.log_success("All validation checks passed! ✨")
|
||
else:
|
||
if self.issues:
|
||
print(f"❌ {len(self.issues)} critical issues found:")
|
||
for issue in self.issues:
|
||
print(f" • {issue}")
|
||
|
||
if self.warnings:
|
||
print(f"⚠️ {len(self.warnings)} warnings:")
|
||
for warning in self.warnings:
|
||
print(f" • {warning}")
|
||
|
||
# Additional statistics
|
||
print("\nDatabase Statistics:")
|
||
total_records = (
|
||
self.db.query(User).count() +
|
||
self.db.query(Company).count() +
|
||
self.db.query(Contact).count() +
|
||
self.db.query(Deal).count() +
|
||
self.db.query(Event).count()
|
||
)
|
||
|
||
custom_fields_count = self.db.query(CustomField).count()
|
||
|
||
print(f" Total entity records: {total_records:,}")
|
||
print(f" Custom field values: {custom_fields_count:,}")
|
||
|
||
if total_records > 0:
|
||
print(f" Custom fields per entity: {custom_fields_count/total_records:.1f}")
|
||
|
||
|
||
def main() -> int:
|
||
"""Main function"""
|
||
try:
|
||
validator = DataValidator()
|
||
success = validator.validate_all()
|
||
return 0 if success else 1
|
||
|
||
except Exception as e:
|
||
print(f"❌ Validation script failed: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
exit_code = main()
|
||
sys.exit(exit_code)
|
||
|