Spaces:

Ilirt
/

dataflow-server

Sleeping

upload

5aaca54 verified 6 months ago

11.8 kB

"""
DataFlow Server — Main FastAPI Application
"""
import uuid
import json
from datetime import datetime
from typing import List

from fastapi import FastAPI, Depends, HTTPException, status, UploadFile, File, Form
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from database import get_db, init_db, pwd_context
from models import User, Project, Task, Expense, Document, Notification, AIForecast
import schemas
import ai_service

app = FastAPI(title="DataFlow API", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.on_event("startup")
async def startup_event():
    await init_db()

# ─── Auth Routes ───
@app.post("/auth/login", response_model=schemas.Token)
async def login(credentials: schemas.UserLogin, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.email == credentials.email))
    user = result.scalar_one_or_none()
    
    if not user or not pwd_context.verify(credentials.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Incorrect email or password")
        
    return {
        "access_token": "fake-jwt-token-for-demo",
        "token_type": "bearer",
        "user_id": user.id,
        "full_name": user.full_name,
        "role": user.role
    }

@app.post("/auth/register", response_model=schemas.Token)
async def register(user_data: schemas.UserRegister, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.email == user_data.email))
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="Email already registered")
        
    new_user = User(
        id=str(uuid.uuid4()),
        full_name=user_data.full_name,
        email=user_data.email,
        hashed_password=pwd_context.hash(user_data.password),
        position=user_data.position,
        role=user_data.role
    )
    db.add(new_user)
    await db.commit()
    
    return {
        "access_token": "fake-jwt-token-for-demo",
        "token_type": "bearer",
        "user_id": new_user.id,
        "full_name": new_user.full_name,
        "role": new_user.role
    }

# ─── Project Routes ───
@app.get("/projects", response_model=List[schemas.ProjectOut])
async def get_projects(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Project))
    return result.scalars().all()

@app.post("/projects", response_model=schemas.ProjectOut)
async def create_project(project: schemas.ProjectCreate, db: AsyncSession = Depends(get_db)):
    new_project = Project(
        id=str(uuid.uuid4()),
        **project.model_dump()
    )
    db.add(new_project)
    await db.commit()
    await db.refresh(new_project)
    return new_project

@app.patch("/projects/{project_id}", response_model=schemas.ProjectOut)
async def update_project(project_id: str, updates: schemas.ProjectUpdate, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Project).where(Project.id == project_id))
    project = result.scalar_one_or_none()
    if not project:
        raise HTTPException(status_code=404, detail="Project not found")
        
    for key, value in updates.model_dump(exclude_unset=True).items():
        setattr(project, key, value)
        
    await db.commit()
    await db.refresh(project)
    return project

# ─── Task Routes ───
@app.get("/tasks", response_model=List[schemas.TaskOut])
async def get_tasks(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Task))
    return result.scalars().all()

@app.post("/tasks", response_model=schemas.TaskOut)
async def create_task(task: schemas.TaskCreate, db: AsyncSession = Depends(get_db)):
    new_task = Task(id=str(uuid.uuid4()), **task.model_dump())
    db.add(new_task)
    await db.commit()
    await db.refresh(new_task)
    return new_task

@app.patch("/tasks/{task_id}", response_model=schemas.TaskOut)
async def update_task(task_id: str, updates: schemas.TaskUpdate, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Task).where(Task.id == task_id))
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
        
    for key, value in updates.model_dump(exclude_unset=True).items():
        setattr(task, key, value)
        
    await db.commit()
    await db.refresh(task)
    return task

# ─── Expense Routes ───
@app.get("/expenses", response_model=List[schemas.ExpenseOut])
async def get_expenses(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Expense))
    return result.scalars().all()

@app.post("/expenses", response_model=schemas.ExpenseOut)
async def create_expense(expense: schemas.ExpenseCreate, db: AsyncSession = Depends(get_db)):
    new_expense = Expense(
        id=str(uuid.uuid4()),
        **expense.model_dump(exclude={"date"}),
        date=expense.date or datetime.utcnow()
    )
    db.add(new_expense)
    
    # Update project spent amount
    project_result = await db.execute(select(Project).where(Project.id == expense.project_id))
    project = project_result.scalar_one_or_none()
    if project:
        project.spent += expense.amount
        
    await db.commit()
    await db.refresh(new_expense)
    return new_expense

@app.patch("/expenses/{expense_id}/status")
async def update_expense_status(expense_id: str, status: str, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Expense).where(Expense.id == expense_id))
    expense = result.scalar_one_or_none()
    if not expense:
        raise HTTPException(status_code=404, detail="Expense not found")
        
    expense.status = status
    await db.commit()
    return {"status": "success"}

# ─── Document & OCR Routes ───
@app.get("/documents", response_model=List[schemas.DocumentOut])
async def get_documents(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Document))
    return result.scalars().all()

@app.post("/documents/upload", response_model=schemas.DocumentOut)
async def upload_document(
    project_id: str = Form(...),
    file: UploadFile = File(...),
    db: AsyncSession = Depends(get_db)
):
    content = await file.read()
    file_ext = file.filename.split('.')[-1].lower() if '.' in file.filename else 'unknown'
    
    doc = Document(
        id=str(uuid.uuid4()),
        name=file.filename,
        project_id=project_id,
        type=file_ext,
        size_bytes=len(content)
    )
    
    # Trigger OCR if image or pdf
    if file_ext in ['jpg', 'jpeg', 'png', 'pdf']:
        ocr_result = await ai_service.ocr_extract(content, file.filename)
        if ocr_result.get("success"):
            doc.extracted_text = ocr_result.get("extracted_text")
            doc.recognition_confidence = ocr_result.get("confidence")
            doc.detected_type = ocr_result.get("detected_type")
            doc.extracted_amount = ocr_result.get("extracted_amount")
            
            # Create notification for OCR result
            notif = Notification(
                id=str(uuid.uuid4()),
                title="Документ распознан",
                body=f"Документ {file.filename} успешно обработан AI.",
                type="document"
            )
            db.add(notif)
            
    db.add(doc)
    await db.commit()
    await db.refresh(doc)
    return doc

# ─── AI Forecast Routes ───
@app.get("/forecasts", response_model=List[schemas.ForecastOut])
async def get_forecasts(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(AIForecast))
    return result.scalars().all()

@app.post("/forecasts/generate", response_model=schemas.ForecastOut)
async def generate_forecast(req: schemas.ForecastRequest, db: AsyncSession = Depends(get_db)):
    # 1. Get project
    proj_res = await db.execute(select(Project).where(Project.id == req.project_id))
    project = proj_res.scalar_one_or_none()
    if not project:
        raise HTTPException(status_code=404, detail="Project not found")
        
    # 2. Get tasks to calc completion
    task_res = await db.execute(select(Task).where(Task.project_id == project.id))
    tasks = task_res.scalars().all()
    completed_tasks = sum(1 for t in tasks if t.status == "done")
    
    # 3. Get expenses for categories
    exp_res = await db.execute(select(Expense).where(Expense.project_id == project.id))
    expenses = exp_res.scalars().all()
    cats = {}
    for e in expenses:
        cats[e.category] = cats.get(e.category, 0) + e.amount
        
    # 4. Calc days
    now = datetime.utcnow()
    total_days = (project.end_date - project.start_date).days
    days_elapsed = (now - project.start_date).days
    if total_days <= 0: total_days = 1
    if days_elapsed < 0: days_elapsed = 0
    
    # 5. Call AI
    ai_result = await ai_service.generate_forecast(
        project_name=project.name,
        budget=project.budget,
        spent=project.spent,
        days_elapsed=days_elapsed,
        total_days=total_days,
        expense_categories=cats,
        completed_tasks=completed_tasks,
        total_tasks=len(tasks)
    )
    
    if not ai_result.get("success"):
        raise HTTPException(status_code=500, detail="Failed to generate AI forecast")
        
    # 6. Save or update forecast
    fc_res = await db.execute(select(AIForecast).where(AIForecast.project_id == project.id))
    forecast = fc_res.scalar_one_or_none()
    
    if forecast:
        forecast.current_spent = project.spent
        forecast.planned_budget = project.budget
        forecast.risk_level = ai_result.get("risk_level", "green")
        forecast.overrun_probability = ai_result.get("overrun_probability", 0)
        forecast.optimistic_forecast = ai_result.get("optimistic_forecast", 0)
        forecast.realistic_forecast = ai_result.get("realistic_forecast", 0)
        forecast.pessimistic_forecast = ai_result.get("pessimistic_forecast", 0)
        forecast.recommendations = json.dumps(ai_result.get("recommendations", []))
        forecast.generated_at = datetime.utcnow()
    else:
        forecast = AIForecast(
            id=str(uuid.uuid4()),
            project_id=project.id,
            current_spent=project.spent,
            planned_budget=project.budget,
            risk_level=ai_result.get("risk_level", "green"),
            overrun_probability=ai_result.get("overrun_probability", 0),
            optimistic_forecast=ai_result.get("optimistic_forecast", 0),
            realistic_forecast=ai_result.get("realistic_forecast", 0),
            pessimistic_forecast=ai_result.get("pessimistic_forecast", 0),
            recommendations=json.dumps(ai_result.get("recommendations", []))
        )
        db.add(forecast)
        
    await db.commit()
    await db.refresh(forecast)
    return forecast

# ─── Notifications Routes ───
@app.get("/notifications", response_model=List[schemas.NotificationOut])
async def get_notifications(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Notification).order_by(Notification.created_at.desc()))
    return result.scalars().all()

@app.patch("/notifications/{notif_id}/read")
async def mark_notification_read(notif_id: str, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Notification).where(Notification.id == notif_id))
    notif = result.scalar_one_or_none()
    if notif:
        notif.is_read = True
        await db.commit()
    return {"status": "success"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

	"""
	DataFlow Server — Main FastAPI Application
	"""
	import uuid
	import json
	from datetime import datetime
	from typing import List

	from fastapi import FastAPI, Depends, HTTPException, status, UploadFile, File, Form
	from fastapi.middleware.cors import CORSMiddleware
	from sqlalchemy.ext.asyncio import AsyncSession
	from sqlalchemy import select

	from database import get_db, init_db, pwd_context
	from models import User, Project, Task, Expense, Document, Notification, AIForecast
	import schemas
	import ai_service

	app = FastAPI(title="DataFlow API", version="1.0.0")

	app.add_middleware(
	CORSMiddleware,
	allow_origins=["*"],
	allow_credentials=True,
	allow_methods=["*"],
	allow_headers=["*"],
	)

	@app.on_event("startup")
	async def startup_event():
	await init_db()

	# ─── Auth Routes ───
	@app.post("/auth/login", response_model=schemas.Token)
	async def login(credentials: schemas.UserLogin, db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(User).where(User.email == credentials.email))
	user = result.scalar_one_or_none()

	if not user or not pwd_context.verify(credentials.password, user.hashed_password):
	raise HTTPException(status_code=401, detail="Incorrect email or password")

	return {
	"access_token": "fake-jwt-token-for-demo",
	"token_type": "bearer",
	"user_id": user.id,
	"full_name": user.full_name,
	"role": user.role
	}

	@app.post("/auth/register", response_model=schemas.Token)
	async def register(user_data: schemas.UserRegister, db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(User).where(User.email == user_data.email))
	if result.scalar_one_or_none():
	raise HTTPException(status_code=400, detail="Email already registered")

	new_user = User(
	id=str(uuid.uuid4()),
	full_name=user_data.full_name,
	email=user_data.email,
	hashed_password=pwd_context.hash(user_data.password),
	position=user_data.position,
	role=user_data.role
	)
	db.add(new_user)
	await db.commit()

	return {
	"access_token": "fake-jwt-token-for-demo",
	"token_type": "bearer",
	"user_id": new_user.id,
	"full_name": new_user.full_name,
	"role": new_user.role
	}

	# ─── Project Routes ───
	@app.get("/projects", response_model=List[schemas.ProjectOut])
	async def get_projects(db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(Project))
	return result.scalars().all()

	@app.post("/projects", response_model=schemas.ProjectOut)
	async def create_project(project: schemas.ProjectCreate, db: AsyncSession = Depends(get_db)):
	new_project = Project(
	id=str(uuid.uuid4()),
	**project.model_dump()
	)
	db.add(new_project)
	await db.commit()
	await db.refresh(new_project)
	return new_project

	@app.patch("/projects/{project_id}", response_model=schemas.ProjectOut)
	async def update_project(project_id: str, updates: schemas.ProjectUpdate, db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(Project).where(Project.id == project_id))
	project = result.scalar_one_or_none()
	if not project:
	raise HTTPException(status_code=404, detail="Project not found")

	for key, value in updates.model_dump(exclude_unset=True).items():
	setattr(project, key, value)

	await db.commit()
	await db.refresh(project)
	return project

	# ─── Task Routes ───
	@app.get("/tasks", response_model=List[schemas.TaskOut])
	async def get_tasks(db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(Task))
	return result.scalars().all()

	@app.post("/tasks", response_model=schemas.TaskOut)
	async def create_task(task: schemas.TaskCreate, db: AsyncSession = Depends(get_db)):
	new_task = Task(id=str(uuid.uuid4()), **task.model_dump())
	db.add(new_task)
	await db.commit()
	await db.refresh(new_task)
	return new_task

	@app.patch("/tasks/{task_id}", response_model=schemas.TaskOut)
	async def update_task(task_id: str, updates: schemas.TaskUpdate, db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(Task).where(Task.id == task_id))
	task = result.scalar_one_or_none()
	if not task:
	raise HTTPException(status_code=404, detail="Task not found")

	for key, value in updates.model_dump(exclude_unset=True).items():
	setattr(task, key, value)

	await db.commit()
	await db.refresh(task)
	return task

	# ─── Expense Routes ───
	@app.get("/expenses", response_model=List[schemas.ExpenseOut])
	async def get_expenses(db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(Expense))
	return result.scalars().all()

	@app.post("/expenses", response_model=schemas.ExpenseOut)
	async def create_expense(expense: schemas.ExpenseCreate, db: AsyncSession = Depends(get_db)):
	new_expense = Expense(
	id=str(uuid.uuid4()),
	**expense.model_dump(exclude={"date"}),
	date=expense.date or datetime.utcnow()
	)
	db.add(new_expense)

	# Update project spent amount
	project_result = await db.execute(select(Project).where(Project.id == expense.project_id))
	project = project_result.scalar_one_or_none()
	if project:
	project.spent += expense.amount

	await db.commit()
	await db.refresh(new_expense)
	return new_expense

	@app.patch("/expenses/{expense_id}/status")
	async def update_expense_status(expense_id: str, status: str, db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(Expense).where(Expense.id == expense_id))
	expense = result.scalar_one_or_none()
	if not expense:
	raise HTTPException(status_code=404, detail="Expense not found")

	expense.status = status
	await db.commit()
	return {"status": "success"}

	# ─── Document & OCR Routes ───
	@app.get("/documents", response_model=List[schemas.DocumentOut])
	async def get_documents(db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(Document))
	return result.scalars().all()

	@app.post("/documents/upload", response_model=schemas.DocumentOut)
	async def upload_document(
	project_id: str = Form(...),
	file: UploadFile = File(...),
	db: AsyncSession = Depends(get_db)
	):
	content = await file.read()
	file_ext = file.filename.split('.')[-1].lower() if '.' in file.filename else 'unknown'

	doc = Document(
	id=str(uuid.uuid4()),
	name=file.filename,
	project_id=project_id,
	type=file_ext,
	size_bytes=len(content)
	)

	# Trigger OCR if image or pdf
	if file_ext in ['jpg', 'jpeg', 'png', 'pdf']:
	ocr_result = await ai_service.ocr_extract(content, file.filename)
	if ocr_result.get("success"):
	doc.extracted_text = ocr_result.get("extracted_text")
	doc.recognition_confidence = ocr_result.get("confidence")
	doc.detected_type = ocr_result.get("detected_type")
	doc.extracted_amount = ocr_result.get("extracted_amount")

	# Create notification for OCR result
	notif = Notification(
	id=str(uuid.uuid4()),
	title="Документ распознан",
	body=f"Документ {file.filename} успешно обработан AI.",
	type="document"
	)
	db.add(notif)

	db.add(doc)
	await db.commit()
	await db.refresh(doc)
	return doc

	# ─── AI Forecast Routes ───
	@app.get("/forecasts", response_model=List[schemas.ForecastOut])
	async def get_forecasts(db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(AIForecast))
	return result.scalars().all()

	@app.post("/forecasts/generate", response_model=schemas.ForecastOut)
	async def generate_forecast(req: schemas.ForecastRequest, db: AsyncSession = Depends(get_db)):
	# 1. Get project
	proj_res = await db.execute(select(Project).where(Project.id == req.project_id))
	project = proj_res.scalar_one_or_none()
	if not project:
	raise HTTPException(status_code=404, detail="Project not found")

	# 2. Get tasks to calc completion
	task_res = await db.execute(select(Task).where(Task.project_id == project.id))
	tasks = task_res.scalars().all()
	completed_tasks = sum(1 for t in tasks if t.status == "done")

	# 3. Get expenses for categories
	exp_res = await db.execute(select(Expense).where(Expense.project_id == project.id))
	expenses = exp_res.scalars().all()
	cats = {}
	for e in expenses:
	cats[e.category] = cats.get(e.category, 0) + e.amount

	# 4. Calc days
	now = datetime.utcnow()
	total_days = (project.end_date - project.start_date).days
	days_elapsed = (now - project.start_date).days
	if total_days <= 0: total_days = 1
	if days_elapsed < 0: days_elapsed = 0

	# 5. Call AI
	ai_result = await ai_service.generate_forecast(
	project_name=project.name,
	budget=project.budget,
	spent=project.spent,
	days_elapsed=days_elapsed,
	total_days=total_days,
	expense_categories=cats,
	completed_tasks=completed_tasks,
	total_tasks=len(tasks)
	)

	if not ai_result.get("success"):
	raise HTTPException(status_code=500, detail="Failed to generate AI forecast")

	# 6. Save or update forecast
	fc_res = await db.execute(select(AIForecast).where(AIForecast.project_id == project.id))
	forecast = fc_res.scalar_one_or_none()

	if forecast:
	forecast.current_spent = project.spent
	forecast.planned_budget = project.budget
	forecast.risk_level = ai_result.get("risk_level", "green")
	forecast.overrun_probability = ai_result.get("overrun_probability", 0)
	forecast.optimistic_forecast = ai_result.get("optimistic_forecast", 0)
	forecast.realistic_forecast = ai_result.get("realistic_forecast", 0)
	forecast.pessimistic_forecast = ai_result.get("pessimistic_forecast", 0)
	forecast.recommendations = json.dumps(ai_result.get("recommendations", []))
	forecast.generated_at = datetime.utcnow()
	else:
	forecast = AIForecast(
	id=str(uuid.uuid4()),
	project_id=project.id,
	current_spent=project.spent,
	planned_budget=project.budget,
	risk_level=ai_result.get("risk_level", "green"),
	overrun_probability=ai_result.get("overrun_probability", 0),
	optimistic_forecast=ai_result.get("optimistic_forecast", 0),
	realistic_forecast=ai_result.get("realistic_forecast", 0),
	pessimistic_forecast=ai_result.get("pessimistic_forecast", 0),
	recommendations=json.dumps(ai_result.get("recommendations", []))
	)
	db.add(forecast)

	await db.commit()
	await db.refresh(forecast)
	return forecast

	# ─── Notifications Routes ───
	@app.get("/notifications", response_model=List[schemas.NotificationOut])
	async def get_notifications(db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(Notification).order_by(Notification.created_at.desc()))
	return result.scalars().all()

	@app.patch("/notifications/{notif_id}/read")
	async def mark_notification_read(notif_id: str, db: AsyncSession = Depends(get_db)):
	result = await db.execute(select(Notification).where(Notification.id == notif_id))
	notif = result.scalar_one_or_none()
	if notif:
	notif.is_read = True
	await db.commit()
	return {"status": "success"}

	if __name__ == "__main__":
	import uvicorn
	uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)