369 lines
11 KiB
Python
369 lines
11 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from typing import Dict, Any, List, Optional
|
|
from pydantic import BaseModel
|
|
from datetime import datetime
|
|
import time
|
|
import logging
|
|
|
|
from adapters.sqlite.database import get_db
|
|
from adapters.sqlite.models import ExportConfiguration, ExportEntityMapping, ExportJob
|
|
from servers.job_server import JobServer
|
|
from workers.broker import broker
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class FieldMapping(BaseModel):
|
|
field_name: str
|
|
column: str
|
|
order: int
|
|
|
|
|
|
class EntityMappingConfig(BaseModel):
|
|
sheet_name: str
|
|
is_enabled: bool = True
|
|
field_mapping: List[FieldMapping]
|
|
|
|
|
|
class ExportConfigurationRequest(BaseModel):
|
|
name: str
|
|
sheet_id: str
|
|
date_range_start: Optional[str] = None
|
|
date_range_end: Optional[str] = None
|
|
entity_mappings: Dict[str, EntityMappingConfig]
|
|
|
|
|
|
class ExportJobStart(BaseModel):
|
|
configuration_id: int
|
|
|
|
|
|
class SyncJobRequest(BaseModel):
|
|
entity_type: str
|
|
limit: Optional[int] = None
|
|
page: Optional[int] = None
|
|
|
|
|
|
@router.post("/configure")
|
|
async def create_export_configuration(
|
|
config: ExportConfigurationRequest,
|
|
db: Session = Depends(get_db)
|
|
) -> Dict[str, Any]:
|
|
"""Create or update unified export configuration for all entities"""
|
|
|
|
# Convert date strings to timestamps
|
|
date_start = None
|
|
date_end = None
|
|
|
|
if config.date_range_start:
|
|
try:
|
|
date_start = int(datetime.fromisoformat(config.date_range_start.replace('Z', '+00:00')).timestamp())
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="Invalid date_range_start format")
|
|
|
|
if config.date_range_end:
|
|
try:
|
|
date_end = int(datetime.fromisoformat(config.date_range_end.replace('Z', '+00:00')).timestamp())
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="Invalid date_range_end format")
|
|
|
|
# Create configuration
|
|
db_config = ExportConfiguration(
|
|
name=config.name,
|
|
sheet_id=config.sheet_id,
|
|
date_range_start=date_start,
|
|
date_range_end=date_end,
|
|
is_active=True,
|
|
created_at=int(time.time()),
|
|
updated_at=int(time.time())
|
|
)
|
|
|
|
db.add(db_config)
|
|
db.flush() # Get the ID
|
|
|
|
# Create entity mappings
|
|
valid_entities = ["deals", "contacts", "companies", "pipelines", "users", "events"]
|
|
|
|
for entity_type, mapping_config in config.entity_mappings.items():
|
|
if entity_type not in valid_entities:
|
|
raise HTTPException(status_code=400, detail=f"Invalid entity type: {entity_type}")
|
|
|
|
# Convert field mapping to JSON
|
|
field_mapping_json = [
|
|
{
|
|
"field_name": fm.field_name,
|
|
"column": fm.column,
|
|
"order": fm.order
|
|
}
|
|
for fm in mapping_config.field_mapping
|
|
]
|
|
|
|
db_mapping = ExportEntityMapping(
|
|
configuration_id=db_config.id,
|
|
entity_type=entity_type,
|
|
sheet_name=mapping_config.sheet_name,
|
|
field_mapping=field_mapping_json,
|
|
is_enabled=mapping_config.is_enabled,
|
|
created_at=int(time.time()),
|
|
updated_at=int(time.time())
|
|
)
|
|
|
|
db.add(db_mapping)
|
|
|
|
db.commit()
|
|
|
|
return {
|
|
"configuration_id": db_config.id,
|
|
"message": "Export configuration created successfully"
|
|
}
|
|
|
|
|
|
@router.get("/configurations")
|
|
async def list_export_configurations(db: Session = Depends(get_db)) -> Dict[str, Any]:
|
|
"""List all export configurations"""
|
|
|
|
configs = db.query(ExportConfiguration).filter(
|
|
ExportConfiguration.is_active == True
|
|
).all()
|
|
|
|
result = []
|
|
for config in configs:
|
|
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
|
|
}
|
|
|
|
result.append({
|
|
"id": config.id,
|
|
"name": config.name,
|
|
"sheet_id": config.sheet_id,
|
|
"date_range_start": datetime.fromtimestamp(config.date_range_start).isoformat() + "Z" if config.date_range_start else None,
|
|
"date_range_end": datetime.fromtimestamp(config.date_range_end).isoformat() + "Z" if config.date_range_end else None,
|
|
"entity_mappings": entity_mappings,
|
|
"created_at": datetime.fromtimestamp(config.created_at).isoformat() + "Z" if config.created_at else None
|
|
})
|
|
|
|
return {"configurations": result}
|
|
|
|
|
|
@router.post("/start")
|
|
async def start_export_job(
|
|
job_request: ExportJobStart,
|
|
db: Session = Depends(get_db)
|
|
) -> Dict[str, Any]:
|
|
"""Start async export job"""
|
|
|
|
# Check if configuration exists
|
|
config = db.query(ExportConfiguration).filter(
|
|
ExportConfiguration.id == job_request.configuration_id,
|
|
ExportConfiguration.is_active == True
|
|
).first()
|
|
|
|
if not config:
|
|
raise HTTPException(status_code=404, detail="Export configuration not found")
|
|
|
|
# Create export job
|
|
job = ExportJob(
|
|
configuration_id=config.id,
|
|
status="pending",
|
|
records_processed=0,
|
|
total_records=0,
|
|
created_at=int(time.time())
|
|
)
|
|
|
|
db.add(job)
|
|
db.commit()
|
|
|
|
# Queue job for background processing with FastStream
|
|
try:
|
|
job_server = JobServer(broker)
|
|
job_uuid = await job_server.queue_export_job(config.id)
|
|
|
|
logger.info(f"Export job {job.id} queued with UUID {job_uuid}")
|
|
|
|
return {
|
|
"job_id": job.id,
|
|
"job_uuid": job_uuid,
|
|
"status": "pending",
|
|
"message": "Export job queued successfully"
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to queue export job {job.id}: {str(e)}")
|
|
|
|
# Update job status to failed
|
|
job.status = "failed"
|
|
job.error_message = f"Failed to queue job: {str(e)}"
|
|
db.commit()
|
|
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to queue export job: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/status/{job_id}")
|
|
async def get_export_status(job_id: int, db: Session = Depends(get_db)) -> Dict[str, Any]:
|
|
"""Get export job status"""
|
|
|
|
job = db.query(ExportJob).filter(ExportJob.id == job_id).first()
|
|
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Export job not found")
|
|
|
|
result = {
|
|
"job_id": job.id,
|
|
"status": job.status,
|
|
"progress": {
|
|
"records_processed": job.records_processed,
|
|
"total_records": job.total_records,
|
|
"percentage": round((job.records_processed / job.total_records * 100), 2) if job.total_records > 0 else 0
|
|
}
|
|
}
|
|
|
|
if job.started_at:
|
|
result["started_at"] = datetime.fromtimestamp(job.started_at).isoformat() + "Z"
|
|
|
|
if job.completed_at:
|
|
result["completed_at"] = datetime.fromtimestamp(job.completed_at).isoformat() + "Z"
|
|
|
|
if job.error_message:
|
|
result["error_message"] = job.error_message
|
|
|
|
# TODO: Add detailed entity progress when implementing worker
|
|
if job.status == "running":
|
|
result["progress"]["entities"] = {
|
|
"deals": {"processed": 1500, "total": 1500, "status": "completed"},
|
|
"contacts": {"processed": 600, "total": 2200, "status": "running"},
|
|
"companies": {"processed": 0, "total": 300, "status": "pending"},
|
|
"events": {"processed": 0, "total": 200, "status": "pending"}
|
|
}
|
|
result["estimated_completion"] = "2024-01-15T10:35:00Z" # TODO: Calculate based on progress
|
|
|
|
return result
|
|
|
|
|
|
@router.get("/jobs")
|
|
async def list_export_jobs(
|
|
page: int = 1,
|
|
per_page: int = 20,
|
|
db: Session = Depends(get_db)
|
|
) -> Dict[str, Any]:
|
|
"""List all export jobs"""
|
|
|
|
offset = (page - 1) * per_page
|
|
|
|
jobs_query = db.query(ExportJob).order_by(ExportJob.created_at.desc())
|
|
total = jobs_query.count()
|
|
jobs = jobs_query.offset(offset).limit(per_page).all()
|
|
|
|
result_jobs = []
|
|
for job in jobs:
|
|
job_data = {
|
|
"job_id": job.id,
|
|
"configuration_name": job.configuration.name,
|
|
"status": job.status,
|
|
"records_processed": job.records_processed,
|
|
"total_records": job.total_records
|
|
}
|
|
|
|
if job.started_at:
|
|
job_data["started_at"] = datetime.fromtimestamp(job.started_at).isoformat() + "Z"
|
|
|
|
if job.completed_at:
|
|
job_data["completed_at"] = datetime.fromtimestamp(job.completed_at).isoformat() + "Z"
|
|
|
|
# TODO: Add entities_processed breakdown when implementing worker
|
|
if job.status == "completed":
|
|
job_data["entities_processed"] = {
|
|
"deals": 1500,
|
|
"contacts": 2200,
|
|
"companies": 300,
|
|
"events": 200
|
|
}
|
|
|
|
result_jobs.append(job_data)
|
|
|
|
return {
|
|
"jobs": result_jobs,
|
|
"total": total,
|
|
"page": page,
|
|
"per_page": per_page
|
|
}
|
|
|
|
|
|
@router.post("/sync")
|
|
async def queue_sync_job(
|
|
sync_request: SyncJobRequest,
|
|
db: Session = Depends(get_db)
|
|
) -> Dict[str, Any]:
|
|
"""Queue a data synchronization job from AMO CRM"""
|
|
|
|
# Validate entity type
|
|
valid_entities = ["deals", "contacts", "companies", "pipelines", "users", "events"]
|
|
if sync_request.entity_type not in valid_entities:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Invalid entity type. Must be one of: {', '.join(valid_entities)}"
|
|
)
|
|
|
|
try:
|
|
job_server = JobServer(broker)
|
|
job_uuid = await job_server.queue_sync_job(
|
|
entity_type=sync_request.entity_type,
|
|
limit=sync_request.limit,
|
|
page=sync_request.page
|
|
)
|
|
|
|
logger.info(f"Sync job queued for {sync_request.entity_type} with UUID {job_uuid}")
|
|
|
|
return {
|
|
"job_uuid": job_uuid,
|
|
"entity_type": sync_request.entity_type,
|
|
"status": "queued",
|
|
"message": f"Sync job for {sync_request.entity_type} queued successfully"
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to queue sync job for {sync_request.entity_type}: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to queue sync job: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.post("/refresh/{entity_type}")
|
|
async def schedule_refresh_job(entity_type: str) -> Dict[str, Any]:
|
|
"""Schedule a data refresh job for an entity type"""
|
|
|
|
# Validate entity type
|
|
valid_entities = ["deals", "contacts", "companies", "pipelines", "users", "events"]
|
|
if entity_type not in valid_entities:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Invalid entity type. Must be one of: {', '.join(valid_entities)}"
|
|
)
|
|
|
|
try:
|
|
job_server = JobServer(broker)
|
|
await job_server.schedule_refresh_job(entity_type)
|
|
|
|
logger.info(f"Refresh job scheduled for {entity_type}")
|
|
|
|
return {
|
|
"entity_type": entity_type,
|
|
"message": f"Refresh job for {entity_type} scheduled successfully"
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to schedule refresh job for {entity_type}: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to schedule refresh job: {str(e)}"
|
|
)
|