283 lines
9.9 KiB
Python
283 lines
9.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from app.models import (
|
|
Component,
|
|
ComponentVariant,
|
|
DesignatorMapping,
|
|
PerechenRow,
|
|
Project,
|
|
SimpleListRow,
|
|
SpecificationPcbRow,
|
|
SpecificationRow,
|
|
TitleInscription,
|
|
Variant,
|
|
VariantProperty,
|
|
VedomostRow,
|
|
)
|
|
from app.services.table_generators import (
|
|
TABLE_TYPES,
|
|
ComponentView,
|
|
MaterialView,
|
|
generate_perechen,
|
|
generate_simple_list,
|
|
generate_specification,
|
|
generate_specification_pcb,
|
|
generate_vedomost,
|
|
)
|
|
|
|
ROW_MODELS = {
|
|
"perechen": PerechenRow,
|
|
"specification_pcb": SpecificationPcbRow,
|
|
"specification": SpecificationRow,
|
|
"vedomost": VedomostRow,
|
|
"simple_list": SimpleListRow,
|
|
}
|
|
|
|
EDITABLE_FIELDS = {
|
|
"perechen": {"position", "designation", "quantity", "note", "is_header", "is_empty", "stretch", "is_underline"},
|
|
"specification_pcb": {
|
|
"format", "zone", "position", "designation", "name", "quantity", "note",
|
|
"is_header", "is_empty", "stretch", "is_underline",
|
|
},
|
|
"specification": {
|
|
"format", "zone", "position", "designation", "name", "quantity", "note",
|
|
"is_header", "is_empty", "stretch", "is_underline",
|
|
},
|
|
"vedomost": {
|
|
"name", "product_code", "document_code", "supplier", "where_used",
|
|
"quantity_per_item", "quantity_in_set", "quantity_for_reg", "total_quantity", "note",
|
|
"is_header", "is_empty", "stretch", "is_underline",
|
|
},
|
|
"simple_list": {"designator", "name", "quantity", "is_empty"},
|
|
}
|
|
|
|
|
|
def _mappings_dict(db: Session, project_id: int) -> dict[str, tuple[str, str]]:
|
|
rows = db.scalars(
|
|
select(DesignatorMapping).where(DesignatorMapping.project_id == project_id)
|
|
).all()
|
|
return {r.prefix: (r.singular_name, r.plural_name) for r in rows}
|
|
|
|
|
|
def load_components_for_variant(db: Session, project: Project) -> list[ComponentView]:
|
|
variant = db.scalar(
|
|
select(Variant).where(
|
|
Variant.project_id == project.id,
|
|
Variant.name == project.current_variant,
|
|
)
|
|
)
|
|
components = db.scalars(
|
|
select(Component)
|
|
.where(Component.project_id == project.id)
|
|
.options(selectinload(Component.properties))
|
|
).all()
|
|
|
|
views: list[ComponentView] = []
|
|
for comp in components:
|
|
props = {p.key: p.value for p in comp.properties}
|
|
is_fitted = True
|
|
if variant:
|
|
link = db.scalar(
|
|
select(ComponentVariant).where(
|
|
ComponentVariant.component_id == comp.id,
|
|
ComponentVariant.variant_id == variant.id,
|
|
)
|
|
)
|
|
if link is not None:
|
|
is_fitted = link.is_fitted
|
|
overrides = db.scalars(
|
|
select(VariantProperty).where(
|
|
VariantProperty.component_id == comp.id,
|
|
VariantProperty.variant_id == variant.id,
|
|
)
|
|
).all()
|
|
for o in overrides:
|
|
props[o.key] = o.value
|
|
views.append(ComponentView(designator=comp.designator, properties=props, is_fitted=is_fitted))
|
|
return views
|
|
|
|
|
|
def row_to_dict(row: Any) -> dict[str, Any]:
|
|
data = {}
|
|
for col in row.__table__.columns:
|
|
data[col.name] = getattr(row, col.name)
|
|
return data
|
|
|
|
|
|
def get_rows(db: Session, project_id: int, table_type: str) -> list[dict[str, Any]]:
|
|
model = ROW_MODELS[table_type]
|
|
rows = db.scalars(
|
|
select(model).where(model.project_id == project_id).order_by(model.row_index)
|
|
).all()
|
|
return [row_to_dict(r) for r in rows]
|
|
|
|
|
|
def replace_rows(db: Session, project_id: int, table_type: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
model = ROW_MODELS[table_type]
|
|
db.execute(delete(model).where(model.project_id == project_id))
|
|
allowed = EDITABLE_FIELDS[table_type] | {"row_index", "page_number", "is_auto_generated"}
|
|
result = []
|
|
for i, raw in enumerate(rows):
|
|
payload = {k: v for k, v in raw.items() if k in allowed and k != "id"}
|
|
payload["project_id"] = project_id
|
|
payload["row_index"] = payload.get("row_index", i)
|
|
obj = model(**payload)
|
|
db.add(obj)
|
|
result.append(obj)
|
|
db.commit()
|
|
return get_rows(db, project_id, table_type)
|
|
|
|
|
|
def patch_rows(db: Session, project_id: int, table_type: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Upsert by id when present; otherwise full replace if no ids."""
|
|
if not rows:
|
|
return replace_rows(db, project_id, table_type, [])
|
|
if all("id" in r and r["id"] for r in rows):
|
|
model = ROW_MODELS[table_type]
|
|
allowed = EDITABLE_FIELDS[table_type] | {"row_index", "page_number", "is_auto_generated"}
|
|
for raw in rows:
|
|
obj = db.get(model, raw["id"])
|
|
if not obj or obj.project_id != project_id:
|
|
continue
|
|
for k, v in raw.items():
|
|
if k in allowed:
|
|
setattr(obj, k, v)
|
|
db.commit()
|
|
return get_rows(db, project_id, table_type)
|
|
return replace_rows(db, project_id, table_type, rows)
|
|
|
|
|
|
def generate_table(
|
|
db: Session,
|
|
project: Project,
|
|
table_type: str,
|
|
name_field: str | None = None,
|
|
tech_reserve_percent: float | None = None,
|
|
boards_count: int | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
if table_type not in TABLE_TYPES:
|
|
raise ValueError(f"Unknown table type: {table_type}")
|
|
|
|
mappings = _mappings_dict(db, project.id)
|
|
components = load_components_for_variant(db, project)
|
|
|
|
if table_type == "perechen":
|
|
rows = generate_perechen(components, name_field=name_field or "Name", mappings=mappings)
|
|
elif table_type == "specification_pcb":
|
|
rows = generate_specification_pcb(
|
|
components,
|
|
decimal_number=project.decimal_number,
|
|
board_name=project.board_name,
|
|
pcb_doc_name=project.pcb_doc_name or "",
|
|
mappings=mappings,
|
|
)
|
|
elif table_type == "specification":
|
|
materials: list[MaterialView] = []
|
|
layer_count = 0
|
|
if project.pcb_data:
|
|
layer_count = project.pcb_data.layer_count
|
|
for m in project.pcb_data.diel_materials:
|
|
materials.append(
|
|
MaterialView(
|
|
name=m.name,
|
|
value=m.value,
|
|
height=m.height,
|
|
diel_type=m.diel_type,
|
|
layer_number=m.layer_number,
|
|
)
|
|
)
|
|
rows = generate_specification(materials, layer_count=layer_count, decimal_number=project.decimal_number)
|
|
elif table_type == "vedomost":
|
|
inscriptions = {
|
|
i.field_number: i.field_value
|
|
for i in db.scalars(
|
|
select(TitleInscription).where(TitleInscription.project_id == project.id)
|
|
).all()
|
|
}
|
|
where_used = inscriptions.get(101) or inscriptions.get(1001) or project.decimal_number
|
|
rows = generate_vedomost(components, where_used=where_used, mappings=mappings)
|
|
else:
|
|
rows = generate_simple_list(
|
|
components,
|
|
name_field=name_field or "Name",
|
|
tech_reserve_percent=tech_reserve_percent if tech_reserve_percent is not None else 10.0,
|
|
boards_count=boards_count if boards_count is not None else 1,
|
|
)
|
|
|
|
return replace_rows(db, project.id, table_type, rows)
|
|
|
|
|
|
def get_inscriptions(db: Session, project_id: int) -> dict[int, str]:
|
|
rows = db.scalars(
|
|
select(TitleInscription).where(TitleInscription.project_id == project_id)
|
|
).all()
|
|
return {r.field_number: r.field_value for r in rows}
|
|
|
|
|
|
def set_inscriptions(db: Session, project_id: int, values: dict[int, str]) -> dict[int, str]:
|
|
existing = {
|
|
r.field_number: r
|
|
for r in db.scalars(
|
|
select(TitleInscription).where(TitleInscription.project_id == project_id)
|
|
).all()
|
|
}
|
|
for num, val in values.items():
|
|
num = int(num)
|
|
if num in existing:
|
|
existing[num].field_value = val
|
|
else:
|
|
db.add(TitleInscription(project_id=project_id, field_number=num, field_value=val))
|
|
db.commit()
|
|
return get_inscriptions(db, project_id)
|
|
|
|
|
|
def apply_llm_edits(
|
|
db: Session, project_id: int, table_type: str, edits: list[dict[str, Any]]
|
|
) -> list[dict[str, Any]]:
|
|
model = ROW_MODELS[table_type]
|
|
allowed = EDITABLE_FIELDS[table_type]
|
|
for edit in edits:
|
|
op = edit.get("op", "update_row")
|
|
if op == "update_row":
|
|
obj = None
|
|
if edit.get("row_id"):
|
|
obj = db.get(model, edit["row_id"])
|
|
elif edit.get("row_index") is not None:
|
|
obj = db.scalar(
|
|
select(model).where(
|
|
model.project_id == project_id,
|
|
model.row_index == edit["row_index"],
|
|
)
|
|
)
|
|
if not obj or obj.project_id != project_id:
|
|
continue
|
|
for k, v in (edit.get("fields") or {}).items():
|
|
if k in allowed:
|
|
setattr(obj, k, v)
|
|
elif op == "delete_row":
|
|
obj = None
|
|
if edit.get("row_id"):
|
|
obj = db.get(model, edit["row_id"])
|
|
if obj and obj.project_id == project_id:
|
|
db.delete(obj)
|
|
elif op == "add_row":
|
|
payload = {k: v for k, v in (edit.get("fields") or {}).items() if k in allowed}
|
|
payload["project_id"] = project_id
|
|
payload["row_index"] = edit.get("row_index", 9999)
|
|
payload["is_auto_generated"] = False
|
|
db.add(model(**payload))
|
|
db.commit()
|
|
# reindex
|
|
rows = db.scalars(
|
|
select(model).where(model.project_id == project_id).order_by(model.row_index)
|
|
).all()
|
|
for i, r in enumerate(rows):
|
|
r.row_index = i
|
|
db.commit()
|
|
return get_rows(db, project_id, table_type)
|