417 lines
14 KiB
Python
417 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
|
|
from fastapi.responses import Response
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.auth import require_token
|
|
from app.core.database import get_db
|
|
from app.models import Component, LlmMessage, Project, ProjectParam
|
|
from app.schemas import (
|
|
ExportRequest,
|
|
GenerateTableRequest,
|
|
InscriptionsUpdate,
|
|
LlmApplyRequest,
|
|
LlmChatRequest,
|
|
LlmChatResponse,
|
|
LlmEdit,
|
|
ProjectCreate,
|
|
ProjectOut,
|
|
ProjectUpdate,
|
|
TableRowsPatch,
|
|
TableRowsResponse,
|
|
TableSettingsUpdate,
|
|
)
|
|
from app.services.excel_export import export_xlsx
|
|
from app.services.llm import chat_edit_table
|
|
from app.services.pdf_export import export_pdf
|
|
from app.services.project_service import get_project_full, ingest_zip, project_dir
|
|
from app.services.project_settings import (
|
|
TABLE_SETTINGS_META,
|
|
get_font_stretch,
|
|
get_project_params_map,
|
|
get_property_names,
|
|
get_table_settings,
|
|
resolve_inscriptions,
|
|
save_cell_overflows,
|
|
update_table_settings,
|
|
)
|
|
from app.services.table_service import (
|
|
TABLE_TYPES,
|
|
apply_llm_edits,
|
|
generate_table,
|
|
get_inscriptions,
|
|
get_rows,
|
|
patch_rows,
|
|
set_inscriptions,
|
|
)
|
|
from app.services import table_service
|
|
|
|
router = APIRouter(prefix="/api")
|
|
secured = APIRouter(dependencies=[Depends(require_token)])
|
|
|
|
|
|
def _project_out(db: Session, project: Project) -> ProjectOut:
|
|
variants = [v.name for v in project.variants] if project.variants else []
|
|
component_count = db.scalar(
|
|
select(func.count()).select_from(Component).where(Component.project_id == project.id)
|
|
) or 0
|
|
layer_count = project.pcb_data.layer_count if project.pcb_data else 0
|
|
return ProjectOut(
|
|
id=project.id,
|
|
name=project.name,
|
|
status=project.status,
|
|
current_variant=project.current_variant,
|
|
zip_path=project.zip_path,
|
|
pcb_doc_name=project.pcb_doc_name,
|
|
decimal_number=project.decimal_number,
|
|
board_name=project.board_name,
|
|
error_message=project.error_message,
|
|
created_at=project.created_at,
|
|
updated_at=project.updated_at,
|
|
variants=variants,
|
|
component_count=component_count,
|
|
layer_count=layer_count,
|
|
)
|
|
|
|
|
|
def _get_project(db: Session, project_id: int) -> Project:
|
|
project = get_project_full(db, project_id)
|
|
if not project:
|
|
raise HTTPException(404, "Project not found")
|
|
return project
|
|
|
|
|
|
@router.get("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
@secured.get("/projects", response_model=list[ProjectOut])
|
|
def list_projects(db: Session = Depends(get_db)):
|
|
projects = db.scalars(select(Project).order_by(Project.id.desc())).all()
|
|
result = []
|
|
for p in projects:
|
|
full = get_project_full(db, p.id) or p
|
|
result.append(_project_out(db, full))
|
|
return result
|
|
|
|
|
|
@secured.post("/projects", response_model=ProjectOut)
|
|
def create_project(body: ProjectCreate, db: Session = Depends(get_db)):
|
|
project = Project(name=body.name, status="created")
|
|
db.add(project)
|
|
db.commit()
|
|
db.refresh(project)
|
|
project_dir(project.id)
|
|
return _project_out(db, project)
|
|
|
|
|
|
@secured.get("/projects/{project_id}", response_model=ProjectOut)
|
|
def get_project(project_id: int, db: Session = Depends(get_db)):
|
|
return _project_out(db, _get_project(db, project_id))
|
|
|
|
|
|
@secured.patch("/projects/{project_id}", response_model=ProjectOut)
|
|
def update_project(project_id: int, body: ProjectUpdate, db: Session = Depends(get_db)):
|
|
project = _get_project(db, project_id)
|
|
for field, value in body.model_dump(exclude_unset=True).items():
|
|
setattr(project, field, value)
|
|
db.commit()
|
|
return _project_out(db, _get_project(db, project_id))
|
|
|
|
|
|
@secured.delete("/projects/{project_id}")
|
|
def delete_project(project_id: int, db: Session = Depends(get_db)):
|
|
project = db.get(Project, project_id)
|
|
if not project:
|
|
raise HTTPException(404, "Project not found")
|
|
db.delete(project)
|
|
db.commit()
|
|
return {"ok": True}
|
|
|
|
|
|
@secured.post("/projects/{project_id}/upload", response_model=ProjectOut)
|
|
async def upload_zip(
|
|
project_id: int,
|
|
file: UploadFile = File(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
project = _get_project(db, project_id)
|
|
data = await file.read()
|
|
try:
|
|
ingest_zip(db, project, data, filename=file.filename or "project.zip", keep_tables=False)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e)) from e
|
|
return _project_out(db, _get_project(db, project_id))
|
|
|
|
|
|
@secured.put("/projects/{project_id}/upload", response_model=ProjectOut)
|
|
async def reupload_zip(
|
|
project_id: int,
|
|
file: UploadFile = File(...),
|
|
keep_tables: bool = True,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
project = _get_project(db, project_id)
|
|
data = await file.read()
|
|
try:
|
|
ingest_zip(
|
|
db,
|
|
project,
|
|
data,
|
|
filename=file.filename or "project.zip",
|
|
keep_tables=keep_tables,
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e)) from e
|
|
return _project_out(db, _get_project(db, project_id))
|
|
|
|
|
|
@secured.get("/projects/{project_id}/components")
|
|
def list_components(project_id: int, db: Session = Depends(get_db)):
|
|
project = _get_project(db, project_id)
|
|
views = table_service.load_components_for_variant(db, project)
|
|
return [
|
|
{
|
|
"designator": v.designator,
|
|
"properties": v.properties,
|
|
"is_fitted": v.is_fitted,
|
|
}
|
|
for v in views
|
|
]
|
|
|
|
|
|
@secured.get("/projects/{project_id}/params")
|
|
def list_params(project_id: int, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
rows = db.scalars(select(ProjectParam).where(ProjectParam.project_id == project_id)).all()
|
|
return [
|
|
{"id": r.id, "name": r.name, "value": r.value, "variant_name": r.variant_name}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
@secured.get("/projects/{project_id}/pcb")
|
|
def get_pcb(project_id: int, db: Session = Depends(get_db)):
|
|
project = _get_project(db, project_id)
|
|
if not project.pcb_data:
|
|
return {"layer_count": 0, "materials": []}
|
|
return {
|
|
"layer_count": project.pcb_data.layer_count,
|
|
"materials": [
|
|
{
|
|
"name": m.name,
|
|
"value": m.value,
|
|
"height": m.height,
|
|
"diel_type": m.diel_type,
|
|
"layer_number": m.layer_number,
|
|
}
|
|
for m in project.pcb_data.diel_materials
|
|
],
|
|
}
|
|
|
|
|
|
@secured.get("/projects/{project_id}/property-names")
|
|
def read_property_names(project_id: int, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
return {"property_names": get_property_names(db, project_id)}
|
|
|
|
|
|
@secured.get("/projects/{project_id}/table-settings/{table_type}")
|
|
def read_table_settings(project_id: int, table_type: str, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
if table_type not in TABLE_TYPES:
|
|
raise HTTPException(400, f"Unknown table type. Use one of: {TABLE_TYPES}")
|
|
return get_table_settings(db, project_id, table_type)
|
|
|
|
|
|
@secured.patch("/projects/{project_id}/table-settings/{table_type}")
|
|
def patch_table_settings(
|
|
project_id: int,
|
|
table_type: str,
|
|
body: TableSettingsUpdate,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
_get_project(db, project_id)
|
|
if table_type not in TABLE_TYPES:
|
|
raise HTTPException(400, f"Unknown table type")
|
|
return update_table_settings(db, project_id, table_type, body.model_dump(exclude_unset=True))
|
|
|
|
|
|
@secured.get("/projects/{project_id}/table-settings-meta")
|
|
def read_table_settings_meta(project_id: int, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
return TABLE_SETTINGS_META
|
|
|
|
|
|
@secured.get("/projects/{project_id}/inscriptions")
|
|
def read_inscriptions(project_id: int, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
return get_inscriptions(db, project_id)
|
|
|
|
|
|
@secured.patch("/projects/{project_id}/inscriptions")
|
|
def update_inscriptions(project_id: int, body: InscriptionsUpdate, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
return set_inscriptions(db, project_id, body.inscriptions)
|
|
|
|
|
|
@secured.get("/projects/{project_id}/tables/{table_type}", response_model=TableRowsResponse)
|
|
def read_table(project_id: int, table_type: str, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
if table_type not in TABLE_TYPES:
|
|
raise HTTPException(400, f"Unknown table type. Use one of: {TABLE_TYPES}")
|
|
return TableRowsResponse(table_type=table_type, rows=get_rows(db, project_id, table_type))
|
|
|
|
|
|
@secured.patch("/projects/{project_id}/tables/{table_type}", response_model=TableRowsResponse)
|
|
def update_table(
|
|
project_id: int,
|
|
table_type: str,
|
|
body: TableRowsPatch,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
_get_project(db, project_id)
|
|
if table_type not in TABLE_TYPES:
|
|
raise HTTPException(400, f"Unknown table type")
|
|
rows = patch_rows(db, project_id, table_type, body.rows)
|
|
return TableRowsResponse(table_type=table_type, rows=rows)
|
|
|
|
|
|
@secured.post("/projects/{project_id}/tables/{table_type}/generate", response_model=TableRowsResponse)
|
|
def generate(
|
|
project_id: int,
|
|
table_type: str,
|
|
body: GenerateTableRequest | None = None,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
project = _get_project(db, project_id)
|
|
if table_type not in TABLE_TYPES:
|
|
raise HTTPException(400, f"Unknown table type")
|
|
body = body or GenerateTableRequest()
|
|
try:
|
|
rows = generate_table(
|
|
db,
|
|
project,
|
|
table_type,
|
|
name_field=body.name_field,
|
|
tech_reserve_percent=body.tech_reserve_percent,
|
|
boards_count=body.boards_count,
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e)) from e
|
|
return TableRowsResponse(table_type=table_type, rows=rows)
|
|
|
|
|
|
@secured.post("/projects/{project_id}/export")
|
|
def export_document(project_id: int, body: ExportRequest, db: Session = Depends(get_db)):
|
|
project = _get_project(db, project_id)
|
|
if body.table_type not in TABLE_TYPES:
|
|
raise HTTPException(400, "Unknown table type")
|
|
rows = get_rows(db, project_id, body.table_type)
|
|
if not rows:
|
|
raise HTTPException(400, "Table is empty. Generate it first.")
|
|
inscriptions = get_inscriptions(db, project_id)
|
|
params = get_project_params_map(db, project_id)
|
|
resolved_inscriptions = resolve_inscriptions(inscriptions, params)
|
|
|
|
if body.format == "xlsx":
|
|
data = export_xlsx(body.table_type, rows, title=project.name)
|
|
return Response(
|
|
content=data,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={
|
|
"Content-Disposition": f'attachment; filename="{project.name}_{body.table_type}.xlsx"'
|
|
},
|
|
)
|
|
if body.format == "pdf":
|
|
try:
|
|
stretch = get_font_stretch(db, project_id, body.table_type)
|
|
result = export_pdf(body.table_type, rows, resolved_inscriptions, font_stretch=stretch)
|
|
save_cell_overflows(db, project_id, body.table_type, result.overflows)
|
|
except ValueError as e:
|
|
raise HTTPException(400, str(e)) from e
|
|
return Response(
|
|
content=result.data,
|
|
media_type="application/pdf",
|
|
headers={
|
|
"Content-Disposition": f'attachment; filename="{project.name}_{body.table_type}.pdf"'
|
|
},
|
|
)
|
|
raise HTTPException(400, "format must be pdf or xlsx")
|
|
|
|
|
|
@secured.post("/projects/{project_id}/llm/chat", response_model=LlmChatResponse)
|
|
async def llm_chat(project_id: int, body: LlmChatRequest, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
if body.table_type not in TABLE_TYPES:
|
|
raise HTTPException(400, "Unknown table type")
|
|
rows = get_rows(db, project_id, body.table_type)
|
|
db.add(
|
|
LlmMessage(
|
|
project_id=project_id,
|
|
role="user",
|
|
content=body.message,
|
|
table_type=body.table_type,
|
|
)
|
|
)
|
|
db.commit()
|
|
try:
|
|
result = await chat_edit_table(body.message, body.table_type, rows)
|
|
except Exception as e:
|
|
raise HTTPException(502, f"LLM request failed: {e}") from e
|
|
|
|
msg = LlmMessage(
|
|
project_id=project_id,
|
|
role="assistant",
|
|
content=result["reply"],
|
|
table_type=body.table_type,
|
|
proposed_edits=json.dumps(result["edits"], ensure_ascii=False),
|
|
)
|
|
db.add(msg)
|
|
db.commit()
|
|
db.refresh(msg)
|
|
return LlmChatResponse(
|
|
reply=result["reply"],
|
|
edits=[LlmEdit(**e) for e in result["edits"]],
|
|
message_id=msg.id,
|
|
)
|
|
|
|
|
|
@secured.post("/projects/{project_id}/llm/apply", response_model=TableRowsResponse)
|
|
def llm_apply(project_id: int, body: LlmApplyRequest, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
if body.table_type not in TABLE_TYPES:
|
|
raise HTTPException(400, "Unknown table type")
|
|
edits = [e.model_dump() for e in body.edits]
|
|
rows = apply_llm_edits(db, project_id, body.table_type, edits)
|
|
return TableRowsResponse(table_type=body.table_type, rows=rows)
|
|
|
|
|
|
@secured.get("/projects/{project_id}/llm/history")
|
|
def llm_history(project_id: int, db: Session = Depends(get_db)):
|
|
_get_project(db, project_id)
|
|
msgs = db.scalars(
|
|
select(LlmMessage)
|
|
.where(LlmMessage.project_id == project_id)
|
|
.order_by(LlmMessage.id.asc())
|
|
).all()
|
|
return [
|
|
{
|
|
"id": m.id,
|
|
"role": m.role,
|
|
"content": m.content,
|
|
"table_type": m.table_type,
|
|
"proposed_edits": json.loads(m.proposed_edits) if m.proposed_edits else None,
|
|
"created_at": m.created_at,
|
|
}
|
|
for m in msgs
|
|
]
|
|
|
|
router.include_router(secured)
|
|
|