119 lines
4.2 KiB
Python
119 lines
4.2 KiB
Python
from datetime import datetime
|
|
from uuid import uuid4
|
|
import os
|
|
import json
|
|
import logging
|
|
from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import Optional, Literal, Dict, Any
|
|
|
|
from core.celery_app import celery_app
|
|
from core.config import settings
|
|
from core.database import get_core_db
|
|
from core.security import get_current_user, validate_access_to_resource
|
|
|
|
from .tasks import scan_file, insert_valid_rows
|
|
from .schemas import ImportJobResponse, ImportJobStatus, CommitRequest
|
|
|
|
router = APIRouter()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@router.post("/upload/{model_target}", response_model=ImportJobResponse)
|
|
async def upload_import_file(
|
|
model_target: Literal["invoice_header", "invoice_details"],
|
|
file: UploadFile = File(...),
|
|
footer_config: Optional[str] = Form(None), # JSON string with settings
|
|
company_id: int = Query(..., description="Company ID"), # Required for context
|
|
operation_type: Optional[str] = Query("imp"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Step 1: Upload CSV, save to temp, trigger scan task.
|
|
"""
|
|
# 1. Validate Access & Get Tenant
|
|
try:
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
except Exception as e:
|
|
logger.error(f"Access validation failed: {e}")
|
|
raise HTTPException(status_code=403, detail="Invalid company access")
|
|
|
|
if not file.filename.endswith(".csv"):
|
|
raise HTTPException(status_code=400, detail="Only .csv files allowed")
|
|
|
|
job_id = str(uuid4())
|
|
|
|
# Ensure directory exists (Safety check)
|
|
upload_dir = os.path.join(os.getcwd(), "uploads", "temp")
|
|
os.makedirs(upload_dir, exist_ok=True)
|
|
|
|
file_path = os.path.join(upload_dir, f"{job_id}.csv")
|
|
meta_path = os.path.join(upload_dir, f"{job_id}.meta.json")
|
|
|
|
try:
|
|
# Save CSV
|
|
contents = await file.read()
|
|
with open(file_path, "wb") as f:
|
|
f.write(contents)
|
|
|
|
# Save Metadata (Context)
|
|
meta_data = {
|
|
"tenant_id": tenant_id,
|
|
"company_id": company_id,
|
|
"user_id": current_user.get("id"),
|
|
"footer_config": footer_config,
|
|
"operation_type": operation_type,
|
|
}
|
|
with open(meta_path, "w") as f:
|
|
json.dump(meta_data, f)
|
|
|
|
except Exception as e:
|
|
logger.error(f"File save error: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to save file: {str(e)}")
|
|
|
|
# Trigger Celery Task (Async)
|
|
# Use our job_id as the Celery task_id for easier tracking
|
|
scan_file.apply_async(args=[job_id, file_path, model_target, footer_config], task_id=job_id)
|
|
|
|
return ImportJobResponse(
|
|
job_id=job_id,
|
|
status="queued",
|
|
message="File uploaded. Scanning started."
|
|
)
|
|
|
|
@router.get("/{job_id}/status")
|
|
async def get_import_status(job_id: str):
|
|
"""
|
|
Poll this endpoint to get % progress or final report.
|
|
"""
|
|
# In a real app, query Redis or DB.
|
|
# For MVP, we might mock or use Celery AsyncResult if backend shares Redis.
|
|
task_result = celery_app.AsyncResult(job_id)
|
|
|
|
if task_result.state == 'PENDING':
|
|
return {"status": "processing", "progress": 0}
|
|
elif task_result.state == 'PROGRESS':
|
|
return {
|
|
"status": "processing",
|
|
"progress": task_result.info.get('current', 0),
|
|
"total": task_result.info.get('total', 0)
|
|
}
|
|
elif task_result.state == 'SUCCESS':
|
|
return task_result.result # Should return the report
|
|
else:
|
|
return {"status": task_result.state, "error": str(task_result.info)}
|
|
|
|
|
|
@router.post("/{job_id}/commit")
|
|
async def commit_import_job(job_id: str, body: CommitRequest):
|
|
"""
|
|
Step 2: User confirms import. Trigger bulk insert.
|
|
"""
|
|
task = insert_valid_rows.delay(job_id, body.model_target)
|
|
|
|
return {
|
|
"status": "committing",
|
|
"message": "Bulk insert started.",
|
|
"commit_job_id": task.id
|
|
}
|