512 lines
18 KiB
Python
512 lines
18 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.page_breaks import protect_page_breaks
|
|
from app.services.project_settings import (
|
|
clear_cell_overflows,
|
|
get_cell_overflows,
|
|
get_column_mappings_list,
|
|
get_project_params_map,
|
|
get_simple_list_settings,
|
|
resolve_decimal_and_board,
|
|
resolve_param_value,
|
|
resolve_where_used,
|
|
vedomost_column_map,
|
|
)
|
|
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", "cell_stretches",
|
|
},
|
|
"specification_pcb": {
|
|
"format", "zone", "position", "designation", "name", "quantity", "note",
|
|
"is_header", "is_empty", "stretch", "is_underline", "cell_stretches",
|
|
},
|
|
"specification": {
|
|
"format", "zone", "position", "designation", "name", "quantity", "note",
|
|
"is_header", "is_empty", "stretch", "is_underline", "cell_stretches",
|
|
},
|
|
"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", "cell_stretches",
|
|
},
|
|
"simple_list": {"designator", "name", "quantity", "is_empty", "cell_stretches"},
|
|
}
|
|
|
|
|
|
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()
|
|
overflow_map = get_cell_overflows(db, project_id, table_type)
|
|
result = []
|
|
for r in rows:
|
|
data = row_to_dict(r)
|
|
key = str(data.get("id") or data.get("row_index"))
|
|
data["overflow_fields"] = overflow_map.get(key, [])
|
|
result.append(data)
|
|
return result
|
|
|
|
|
|
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}")
|
|
|
|
clear_cell_overflows(db, project.id, table_type)
|
|
mappings = _mappings_dict(db, project.id)
|
|
components = load_components_for_variant(db, project)
|
|
col_mappings = get_column_mappings_list(db, project.id, table_type)
|
|
|
|
if table_type == "perechen":
|
|
nf = name_field or (col_mappings[0] if col_mappings else "Name") or "Name"
|
|
rows = generate_perechen(components, name_field=nf, mappings=mappings)
|
|
elif table_type == "specification_pcb":
|
|
dec, board = resolve_decimal_and_board(
|
|
db, project.id, project.decimal_number, project.board_name, table_type
|
|
)
|
|
name_prop = col_mappings[1] if len(col_mappings) > 1 and col_mappings[1] else "ManufacturerPartNumber"
|
|
rows = generate_specification_pcb(
|
|
components,
|
|
decimal_number=dec,
|
|
board_name=board,
|
|
pcb_doc_name=project.pcb_doc_name or "",
|
|
mappings=mappings,
|
|
name_field=name_prop,
|
|
)
|
|
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,
|
|
)
|
|
)
|
|
dec, _ = resolve_decimal_and_board(
|
|
db, project.id, project.decimal_number, project.board_name, table_type
|
|
)
|
|
rows = generate_specification(materials, layer_count=layer_count, decimal_number=dec)
|
|
elif table_type == "vedomost":
|
|
params = get_project_params_map(db, project.id)
|
|
where_used = resolve_where_used(db, project.id, col_mappings)
|
|
if not where_used:
|
|
inscriptions = {
|
|
i.field_number: i.field_value
|
|
for i in db.scalars(
|
|
select(TitleInscription).where(TitleInscription.project_id == project.id)
|
|
).all()
|
|
}
|
|
raw = inscriptions.get(101) or inscriptions.get(1001) or project.decimal_number or ""
|
|
where_used = resolve_param_value(raw, params)
|
|
rows = generate_vedomost(
|
|
components,
|
|
where_used=where_used,
|
|
mappings=mappings,
|
|
column_mappings=vedomost_column_map(col_mappings),
|
|
)
|
|
else:
|
|
sl = get_simple_list_settings(db, project.id)
|
|
rows = generate_simple_list(
|
|
components,
|
|
name_field=name_field or sl["name_field"],
|
|
tech_reserve_percent=tech_reserve_percent if tech_reserve_percent is not None else sl["tech_reserve_percent"],
|
|
boards_count=boards_count if boards_count is not None else sl["boards_count"],
|
|
)
|
|
|
|
rows = protect_page_breaks(rows, table_type)
|
|
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 _sort_key_for_field(value: Any) -> tuple[int, float | str]:
|
|
"""Числа (в т.ч. quantity) раньше строк; пустые в конце."""
|
|
s = str(value or "").strip().replace(",", ".")
|
|
if not s:
|
|
return (2, 0.0)
|
|
try:
|
|
return (0, float(s))
|
|
except ValueError:
|
|
return (1, s.casefold())
|
|
|
|
|
|
def _reorder_by_ids(rows: list[Any], row_ids: list[int]) -> list[Any]:
|
|
by_id = {r.id: r for r in rows}
|
|
ordered: list[Any] = []
|
|
seen: set[int] = set()
|
|
for rid in row_ids:
|
|
obj = by_id.get(rid)
|
|
if obj is not None and rid not in seen:
|
|
ordered.append(obj)
|
|
seen.add(rid)
|
|
for r in rows:
|
|
if r.id not in seen:
|
|
ordered.append(r)
|
|
return ordered
|
|
|
|
|
|
def _sort_row_objects(
|
|
rows: list[Any],
|
|
field: str,
|
|
order: str = "asc",
|
|
within_groups: bool = True,
|
|
) -> list[Any]:
|
|
reverse = str(order or "asc").lower().startswith("desc")
|
|
|
|
def sort_block(block: list[Any]) -> list[Any]:
|
|
return sorted(
|
|
block,
|
|
key=lambda r: _sort_key_for_field(getattr(r, field, "")),
|
|
reverse=reverse,
|
|
)
|
|
|
|
if not within_groups:
|
|
headers_empty = [r for r in rows if getattr(r, "is_header", False) or getattr(r, "is_empty", False)]
|
|
data = [r for r in rows if not getattr(r, "is_header", False) and not getattr(r, "is_empty", False)]
|
|
# Простая глобальная сортировка данных; служебные строки — в исходном относительном
|
|
# положении не сохраняем: ставим отсортированные data, затем «лишние» empty/header в конец.
|
|
# Для ГОСТ обычно within_groups=True.
|
|
return sort_block(data) + headers_empty
|
|
|
|
result: list[Any] = []
|
|
i = 0
|
|
n = len(rows)
|
|
while i < n:
|
|
r = rows[i]
|
|
if getattr(r, "is_header", False) or getattr(r, "is_empty", False):
|
|
result.append(r)
|
|
i += 1
|
|
continue
|
|
block: list[Any] = []
|
|
while i < n and not getattr(rows[i], "is_header", False) and not getattr(rows[i], "is_empty", False):
|
|
block.append(rows[i])
|
|
i += 1
|
|
result.extend(sort_block(block))
|
|
return result
|
|
|
|
|
|
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]
|
|
inscription_patch: dict[int, str] = {}
|
|
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))
|
|
elif op == "reorder_rows":
|
|
row_ids = edit.get("row_ids") or []
|
|
if not isinstance(row_ids, list) or not row_ids:
|
|
continue
|
|
rows = db.scalars(
|
|
select(model).where(model.project_id == project_id).order_by(model.row_index)
|
|
).all()
|
|
ordered = _reorder_by_ids(list(rows), [int(x) for x in row_ids])
|
|
for i, r in enumerate(ordered):
|
|
r.row_index = i
|
|
db.flush()
|
|
elif op == "sort_rows":
|
|
field = edit.get("field") or (edit.get("fields") or {}).get("field")
|
|
if not field or field not in allowed:
|
|
continue
|
|
order = edit.get("order") or (edit.get("fields") or {}).get("order") or "asc"
|
|
within = edit.get("within_groups")
|
|
if within is None:
|
|
within = True
|
|
rows = db.scalars(
|
|
select(model).where(model.project_id == project_id).order_by(model.row_index)
|
|
).all()
|
|
ordered = _sort_row_objects(list(rows), str(field), str(order), bool(within))
|
|
for i, r in enumerate(ordered):
|
|
r.row_index = i
|
|
db.flush()
|
|
elif op == "set_inscription":
|
|
try:
|
|
num = int(edit.get("field_number"))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
value = edit.get("value")
|
|
if value is None:
|
|
value = (edit.get("fields") or {}).get("value", "")
|
|
inscription_patch[num] = str(value)
|
|
elif op == "upsert_designator":
|
|
prefix = str(edit.get("prefix") or "").strip().upper()
|
|
if not prefix:
|
|
continue
|
|
singular = str(edit.get("singular_name") or "").strip()
|
|
plural = str(edit.get("plural_name") or singular).strip()
|
|
existing = db.scalar(
|
|
select(DesignatorMapping).where(
|
|
DesignatorMapping.project_id == project_id,
|
|
DesignatorMapping.prefix == prefix,
|
|
)
|
|
)
|
|
if existing:
|
|
existing.singular_name = singular
|
|
existing.plural_name = plural or singular
|
|
else:
|
|
db.add(
|
|
DesignatorMapping(
|
|
project_id=project_id,
|
|
prefix=prefix,
|
|
singular_name=singular,
|
|
plural_name=plural or singular,
|
|
)
|
|
)
|
|
db.flush()
|
|
elif op == "delete_designator":
|
|
prefix = str(edit.get("prefix") or "").strip().upper()
|
|
if not prefix:
|
|
continue
|
|
existing = db.scalar(
|
|
select(DesignatorMapping).where(
|
|
DesignatorMapping.project_id == project_id,
|
|
DesignatorMapping.prefix == prefix,
|
|
)
|
|
)
|
|
if existing:
|
|
db.delete(existing)
|
|
db.flush()
|
|
|
|
if inscription_patch:
|
|
# inline upsert without nested commit
|
|
existing = {
|
|
r.field_number: r
|
|
for r in db.scalars(
|
|
select(TitleInscription).where(TitleInscription.project_id == project_id)
|
|
).all()
|
|
}
|
|
for num, val in inscription_patch.items():
|
|
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()
|
|
# reindex table rows
|
|
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)
|
|
|
|
|
|
def apply_llm_proposals(
|
|
db: Session, project_id: int, table_type: str, edits: list[dict[str, Any]]
|
|
) -> dict[str, Any]:
|
|
"""Apply mixed table/frame/designator proposals; return refreshed project bits."""
|
|
rows = apply_llm_edits(db, project_id, table_type, edits)
|
|
inscriptions = get_inscriptions(db, project_id)
|
|
desig_rows = db.scalars(
|
|
select(DesignatorMapping)
|
|
.where(DesignatorMapping.project_id == project_id)
|
|
.order_by(DesignatorMapping.prefix.asc())
|
|
).all()
|
|
designators = [
|
|
{
|
|
"id": r.id,
|
|
"prefix": r.prefix,
|
|
"singular_name": r.singular_name,
|
|
"plural_name": r.plural_name,
|
|
}
|
|
for r in desig_rows
|
|
]
|
|
return {
|
|
"table_type": table_type,
|
|
"rows": rows,
|
|
"inscriptions": {str(k): v for k, v in inscriptions.items()},
|
|
"designators": designators,
|
|
}
|