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.
428 lines
15 KiB
Python
428 lines
15 KiB
Python
"""
|
|
Export server for processing Google Sheets export jobs.
|
|
|
|
This module handles the actual processing of export jobs queued through
|
|
the FastStream message broker, including data retrieval, formatting,
|
|
and Google Sheets API integration.
|
|
"""
|
|
|
|
import logging
|
|
import time
|
|
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime
|
|
from sqlalchemy import and_
|
|
from adapters.postgres.database import SessionLocal
|
|
from adapters.postgres.models import (
|
|
Deal, Contact, Company, User, Pipeline, Event,
|
|
ExportConfiguration, ExportEntityMapping, ExportJob,
|
|
CustomField
|
|
)
|
|
from adapters.google_sheets_client import GoogleSheetsClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ExportServer:
|
|
"""Server for processing Google Sheets export operations."""
|
|
|
|
def __init__(self):
|
|
"""Initialize ExportServer with database and Google Sheets client."""
|
|
try:
|
|
self.sheets_client = GoogleSheetsClient()
|
|
logger.info("Google Sheets client initialized successfully")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to initialize Google Sheets client: {str(e)}")
|
|
self.sheets_client = None
|
|
|
|
async def process_export_job(self, job_data: Dict[str, Any]) -> None:
|
|
"""
|
|
Process a Google Sheets export job.
|
|
|
|
Args:
|
|
job_data: Dictionary containing job_id, configuration_id, and metadata
|
|
"""
|
|
job_id = job_data["job_id"]
|
|
configuration_id = job_data["configuration_id"]
|
|
|
|
try:
|
|
logger.info(f"Starting export job {job_id} for configuration {configuration_id}")
|
|
|
|
# Update job status to running
|
|
await self._update_job_status(job_id, "running", started_at=datetime.utcnow())
|
|
|
|
# Get export configuration
|
|
config = await self._get_export_configuration(configuration_id)
|
|
if not config:
|
|
raise ValueError(f"Export configuration {configuration_id} not found")
|
|
|
|
# Process each enabled entity
|
|
total_records = 0
|
|
processed_entities = {}
|
|
|
|
for entity_type, mapping in config["entity_mappings"].items():
|
|
if mapping.get("is_enabled", False):
|
|
logger.info(f"Exporting {entity_type} data")
|
|
|
|
records_count = await self._export_entity(
|
|
entity_type,
|
|
config["sheet_id"],
|
|
mapping,
|
|
config.get("date_range_start"),
|
|
config.get("date_range_end")
|
|
)
|
|
|
|
total_records += records_count
|
|
processed_entities[entity_type] = {
|
|
"processed": records_count,
|
|
"total": records_count,
|
|
"status": "completed"
|
|
}
|
|
|
|
logger.info(f"Exported {records_count} {entity_type} records")
|
|
|
|
# Update job as completed
|
|
await self._update_job_status(
|
|
job_id,
|
|
"completed",
|
|
completed_at=datetime.utcnow(),
|
|
records_processed=total_records,
|
|
total_records=total_records,
|
|
entities_processed=processed_entities
|
|
)
|
|
|
|
logger.info(f"Export job {job_id} completed successfully with {total_records} records")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Export job {job_id} failed: {str(e)}")
|
|
await self._update_job_status(
|
|
job_id,
|
|
"failed",
|
|
completed_at=datetime.utcnow(),
|
|
error_message=str(e)
|
|
)
|
|
raise
|
|
|
|
async def _export_entity(
|
|
self,
|
|
entity_type: str,
|
|
sheet_id: str,
|
|
mapping: Dict[str, Any],
|
|
date_range_start: Optional[str] = None,
|
|
date_range_end: Optional[str] = None
|
|
) -> int:
|
|
"""
|
|
Export specific entity type to Google Sheets.
|
|
|
|
Args:
|
|
entity_type: Type of entity to export (deals, contacts, companies, etc.)
|
|
sheet_id: Google Sheets document ID
|
|
mapping: Field mapping configuration for this entity
|
|
date_range_start: Optional start date filter
|
|
date_range_end: Optional end date filter
|
|
|
|
Returns:
|
|
Number of records exported
|
|
"""
|
|
try:
|
|
# Get entity data from database
|
|
data = await self._get_entity_data(entity_type, date_range_start, date_range_end)
|
|
|
|
if not data:
|
|
logger.warning(f"No data found for {entity_type}")
|
|
return 0
|
|
|
|
# Format data according to mapping
|
|
formatted_data = await self._format_data_for_export(data, mapping)
|
|
|
|
# Export to Google Sheets
|
|
await self._write_to_google_sheets(sheet_id, mapping["sheet_name"], formatted_data)
|
|
|
|
return len(data)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to export {entity_type}: {str(e)}")
|
|
raise
|
|
|
|
async def _get_entity_data(
|
|
self,
|
|
entity_type: str,
|
|
date_range_start: Optional[int] = None,
|
|
date_range_end: Optional[int] = None
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Retrieve entity data from the database.
|
|
|
|
Args:
|
|
entity_type: Type of entity to retrieve
|
|
date_range_start: Optional start date filter (unix timestamp)
|
|
date_range_end: Optional end date filter (unix timestamp)
|
|
|
|
Returns:
|
|
List of entity records
|
|
"""
|
|
try:
|
|
# Map entity types to models
|
|
model_map = {
|
|
"deals": Deal,
|
|
"contacts": Contact,
|
|
"companies": Company,
|
|
"users": User,
|
|
"pipelines": Pipeline,
|
|
"events": Event
|
|
}
|
|
|
|
model = model_map.get(entity_type)
|
|
if not model:
|
|
logger.error(f"Unknown entity type: {entity_type}")
|
|
return []
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
# Build query with date filtering
|
|
query = db.query(model)
|
|
|
|
if date_range_start and hasattr(model, 'created_at'):
|
|
query = query.filter(model.created_at >= date_range_start)
|
|
|
|
if date_range_end and hasattr(model, 'created_at'):
|
|
query = query.filter(model.created_at <= date_range_end)
|
|
|
|
# Execute query
|
|
results = query.all()
|
|
|
|
# Convert to dictionaries
|
|
data = []
|
|
for row in results:
|
|
row_dict = {}
|
|
|
|
# Get all column values
|
|
for column in row.__table__.columns:
|
|
value = getattr(row, column.name)
|
|
row_dict[column.name] = value
|
|
|
|
# Add custom fields if available
|
|
if entity_type in ["deals", "contacts", "companies"]:
|
|
custom_fields = db.query(CustomField).filter(
|
|
CustomField.entity_type == entity_type,
|
|
CustomField.entity_id == row.id
|
|
).all()
|
|
|
|
for cf in custom_fields:
|
|
# Use field_name as key, or fallback to field_id
|
|
field_key = cf.field_name or f"field_{cf.field_id}"
|
|
row_dict[field_key] = cf.field_value
|
|
|
|
data.append(row_dict)
|
|
|
|
logger.info(f"Retrieved {len(data)} {entity_type} records from database")
|
|
return data
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to retrieve {entity_type} data: {str(e)}")
|
|
return []
|
|
|
|
async def _format_data_for_export(
|
|
self,
|
|
data: List[Dict[str, Any]],
|
|
mapping: Dict[str, Any]
|
|
) -> List[List[Any]]:
|
|
"""
|
|
Format data according to the export mapping configuration.
|
|
|
|
Args:
|
|
data: Raw entity data from database
|
|
mapping: Field mapping configuration
|
|
|
|
Returns:
|
|
Formatted data ready for Google Sheets export
|
|
"""
|
|
field_mapping = mapping.get("field_mapping", [])
|
|
|
|
# Sort field mapping by order
|
|
sorted_fields = sorted(field_mapping, key=lambda x: x.get("order", 0))
|
|
|
|
# Create header row
|
|
headers = [field["field_name"] for field in sorted_fields]
|
|
formatted_data = [headers]
|
|
|
|
# Format data rows
|
|
for record in data:
|
|
row = []
|
|
for field in sorted_fields:
|
|
field_name = field["field_name"]
|
|
value = record.get(field_name, "")
|
|
|
|
# Handle different data types and formatting
|
|
if isinstance(value, datetime):
|
|
value = value.isoformat()
|
|
elif value is None:
|
|
value = ""
|
|
|
|
row.append(value)
|
|
|
|
formatted_data.append(row)
|
|
|
|
logger.info(f"Formatted {len(data)} records with {len(headers)} columns")
|
|
return formatted_data
|
|
|
|
async def _write_to_google_sheets(
|
|
self,
|
|
sheet_id: str,
|
|
sheet_name: str,
|
|
data: List[List[Any]]
|
|
) -> None:
|
|
"""
|
|
Write formatted data to Google Sheets.
|
|
|
|
Args:
|
|
sheet_id: Google Sheets document ID
|
|
sheet_name: Name of the sheet tab
|
|
data: Formatted data to write
|
|
"""
|
|
if not self.sheets_client:
|
|
logger.warning("Google Sheets client not available, skipping export")
|
|
return
|
|
|
|
try:
|
|
logger.info(f"Writing {len(data)} rows to sheet '{sheet_name}' in document {sheet_id}")
|
|
|
|
# Write data to Google Sheets
|
|
result = await self.sheets_client.write_data(
|
|
spreadsheet_id=sheet_id,
|
|
sheet_name=sheet_name,
|
|
data=data,
|
|
clear_existing=True
|
|
)
|
|
|
|
logger.info(
|
|
f"Successfully wrote {result['updated_rows']} rows "
|
|
f"({result['updated_cells']} cells) to Google Sheets"
|
|
)
|
|
|
|
# Format header row if there's data
|
|
if data:
|
|
internal_sheet_id = self.sheets_client.get_sheet_id(sheet_id, sheet_name)
|
|
if internal_sheet_id is not None:
|
|
await self.sheets_client.format_header_row(
|
|
spreadsheet_id=sheet_id,
|
|
sheet_name=sheet_name,
|
|
sheet_id=internal_sheet_id
|
|
)
|
|
logger.info(f"Formatted header row for sheet '{sheet_name}'")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to write to Google Sheets: {str(e)}")
|
|
raise
|
|
|
|
async def _get_export_configuration(self, configuration_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get export configuration from database.
|
|
|
|
Args:
|
|
configuration_id: ID of the export configuration
|
|
|
|
Returns:
|
|
Export configuration data or None if not found
|
|
"""
|
|
try:
|
|
db = SessionLocal()
|
|
try:
|
|
# Query configuration
|
|
config = db.query(ExportConfiguration).filter(
|
|
ExportConfiguration.id == configuration_id,
|
|
ExportConfiguration.is_active == True
|
|
).first()
|
|
|
|
if not config:
|
|
logger.warning(f"Export configuration {configuration_id} not found")
|
|
return None
|
|
|
|
# Build entity mappings
|
|
entity_mappings = {}
|
|
for mapping in config.entity_mappings:
|
|
entity_mappings[mapping.entity_type] = {
|
|
"sheet_name": mapping.sheet_name,
|
|
"is_enabled": mapping.is_enabled,
|
|
"field_mapping": mapping.field_mapping # Already stored as JSON
|
|
}
|
|
|
|
result = {
|
|
"id": config.id,
|
|
"name": config.name,
|
|
"sheet_id": config.sheet_id,
|
|
"date_range_start": config.date_range_start,
|
|
"date_range_end": config.date_range_end,
|
|
"entity_mappings": entity_mappings
|
|
}
|
|
|
|
logger.info(f"Retrieved export configuration {configuration_id}")
|
|
return result
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get export configuration {configuration_id}: {str(e)}")
|
|
return None
|
|
|
|
async def _update_job_status(
|
|
self,
|
|
job_id: str,
|
|
status: str,
|
|
**kwargs
|
|
) -> None:
|
|
"""
|
|
Update job status in the database.
|
|
|
|
Args:
|
|
job_id: Unique identifier of the job (UUID string)
|
|
status: New status (pending, running, completed, failed)
|
|
**kwargs: Additional fields to update
|
|
"""
|
|
try:
|
|
db = SessionLocal()
|
|
try:
|
|
# Find job - job_id from the message is the UUID, need to map to DB ID
|
|
# The job_id in the message corresponds to ExportJob.id
|
|
job = None
|
|
if isinstance(job_id, int) or (isinstance(job_id, str) and job_id.isdigit()):
|
|
job = db.query(ExportJob).filter(
|
|
ExportJob.id == int(job_id) if isinstance(job_id, str) else job_id
|
|
).first()
|
|
|
|
if not job:
|
|
logger.warning(f"Job {job_id} not found for status update")
|
|
return
|
|
|
|
# Update status
|
|
job.status = status
|
|
|
|
# Handle datetime objects
|
|
for key, value in kwargs.items():
|
|
if isinstance(value, datetime):
|
|
value = int(value.timestamp())
|
|
|
|
if key == "started_at":
|
|
job.started_at = value
|
|
elif key == "completed_at":
|
|
job.completed_at = value
|
|
elif key == "error_message":
|
|
job.error_message = value
|
|
elif key == "records_processed":
|
|
job.records_processed = value
|
|
elif key == "total_records":
|
|
job.total_records = value
|
|
|
|
db.commit()
|
|
logger.info(f"Updated job {job_id} status to {status}")
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to update job status for {job_id}: {str(e)}")
|
|
raise
|