311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""Per-project document settings (column mappings, param references)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from sqlalchemy import distinct, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import Component, ComponentProperty, ProjectParam, ProjectSetting
|
|
|
|
DOC_TYPE_KEYS = {
|
|
"perechen": "Perechen",
|
|
"specification_pcb": "SpecificationPCB",
|
|
"specification": "Specification",
|
|
"vedomost": "Vedomost",
|
|
"simple_list": "SimpleList",
|
|
}
|
|
|
|
TABLE_SETTINGS_META: dict[str, dict[str, Any]] = {
|
|
"perechen": {
|
|
"columns": [{"label": "Наименование", "key": "name", "mapping_index": 0}],
|
|
"defaults": ["Name"],
|
|
},
|
|
"specification_pcb": {
|
|
"columns": [
|
|
{"label": "Обозначение", "key": "designation", "mapping_index": 0},
|
|
{"label": "Наименование", "key": "name", "mapping_index": 1},
|
|
],
|
|
"defaults": ["Designator", "ManufacturerPartNumber", "", "", "", "", "", ""],
|
|
"project_params": [
|
|
{"key": "board_name", "label": "Название платы", "mapping_index": 6},
|
|
{"key": "decimal_number", "label": "Децимальный №", "mapping_index": 7},
|
|
],
|
|
},
|
|
"specification": {
|
|
"columns": [],
|
|
"defaults": [],
|
|
"project_params": [
|
|
{"key": "decimal_number", "label": "Децимальный №", "mapping_index": 7},
|
|
{"key": "document_name", "label": "Название документа", "mapping_index": 8},
|
|
],
|
|
},
|
|
"vedomost": {
|
|
"columns": [
|
|
{"label": "Наименование", "key": "name", "mapping_index": 0},
|
|
{"label": "Код продукции", "key": "product_code", "mapping_index": 1},
|
|
{"label": "Обозначение документа", "key": "document_code", "mapping_index": 2},
|
|
{"label": "Поставщик", "key": "supplier", "mapping_index": 3},
|
|
{"label": "Примечание", "key": "note", "mapping_index": 5},
|
|
],
|
|
"defaults": ["Name", "ProductCode", "DocumentCode", "Supplier", "", "Note"],
|
|
"project_params": [
|
|
{"key": "where_used", "label": "Куда входит", "mapping_index": 4},
|
|
],
|
|
},
|
|
"simple_list": {
|
|
"columns": [],
|
|
"defaults": [],
|
|
"component_fields": [
|
|
{"key": "name_field", "label": "Поле наименования", "setting_key": "SimpleList_nameField"},
|
|
],
|
|
"extra": [
|
|
{"key": "tech_reserve_percent", "label": "Тех. запас, %", "setting_key": "SimpleList_techReservePercent", "default": "10"},
|
|
{"key": "boards_count", "label": "Кол-во плат", "setting_key": "SimpleList_boardsCount", "default": "1"},
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def _setting_key(table_type: str) -> str:
|
|
doc = DOC_TYPE_KEYS.get(table_type, table_type)
|
|
return f"columnMappings_{doc}"
|
|
|
|
|
|
def _get_setting(db: Session, project_id: int, key: str) -> str | None:
|
|
row = db.scalar(
|
|
select(ProjectSetting).where(
|
|
ProjectSetting.project_id == project_id,
|
|
ProjectSetting.key == key,
|
|
)
|
|
)
|
|
return row.value if row else None
|
|
|
|
|
|
def _set_setting(db: Session, project_id: int, key: str, value: str) -> None:
|
|
row = db.scalar(
|
|
select(ProjectSetting).where(
|
|
ProjectSetting.project_id == project_id,
|
|
ProjectSetting.key == key,
|
|
)
|
|
)
|
|
if row:
|
|
row.value = value
|
|
else:
|
|
db.add(ProjectSetting(project_id=project_id, key=key, value=value))
|
|
|
|
|
|
def get_project_params_map(db: Session, project_id: int) -> dict[str, str]:
|
|
rows = db.scalars(select(ProjectParam).where(ProjectParam.project_id == project_id)).all()
|
|
return {r.name: r.value for r in rows}
|
|
|
|
|
|
def resolve_param_value(stored: str, params: dict[str, str]) -> str:
|
|
if not stored or stored == "-- Не выбрано --":
|
|
return ""
|
|
if stored in params and params[stored]:
|
|
return params[stored]
|
|
return stored
|
|
|
|
|
|
def resolve_inscriptions(
|
|
inscriptions: dict[int, str], params: dict[str, str]
|
|
) -> dict[int, str]:
|
|
resolved: dict[int, str] = {}
|
|
for num, val in inscriptions.items():
|
|
resolved[int(num)] = resolve_param_value(val or "", params)
|
|
return resolved
|
|
|
|
|
|
def get_property_names(db: Session, project_id: int) -> list[str]:
|
|
keys = db.scalars(
|
|
select(distinct(ComponentProperty.key))
|
|
.join(Component, Component.id == ComponentProperty.component_id)
|
|
.where(Component.project_id == project_id)
|
|
.order_by(ComponentProperty.key)
|
|
).all()
|
|
names = sorted({k for k in keys if k})
|
|
if "Designator" not in names:
|
|
names.insert(0, "Designator")
|
|
return names
|
|
|
|
|
|
def get_column_mappings_list(db: Session, project_id: int, table_type: str) -> list[str]:
|
|
meta = TABLE_SETTINGS_META.get(table_type, {})
|
|
defaults = list(meta.get("defaults", []))
|
|
raw = _get_setting(db, project_id, _setting_key(table_type))
|
|
if not raw:
|
|
return defaults
|
|
try:
|
|
loaded = json.loads(raw)
|
|
if isinstance(loaded, list):
|
|
return [str(x) for x in loaded]
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return defaults
|
|
|
|
|
|
def set_column_mappings_list(
|
|
db: Session, project_id: int, table_type: str, mappings: list[str]
|
|
) -> list[str]:
|
|
_set_setting(db, project_id, _setting_key(table_type), json.dumps(mappings, ensure_ascii=False))
|
|
db.commit()
|
|
return get_column_mappings_list(db, project_id, table_type)
|
|
|
|
|
|
def get_simple_list_settings(db: Session, project_id: int) -> dict[str, Any]:
|
|
name_field = _get_setting(db, project_id, "SimpleList_nameField") or "Name"
|
|
tech = _get_setting(db, project_id, "SimpleList_techReservePercent") or "10"
|
|
boards = _get_setting(db, project_id, "SimpleList_boardsCount") or "1"
|
|
return {
|
|
"name_field": name_field,
|
|
"tech_reserve_percent": float(tech),
|
|
"boards_count": int(boards),
|
|
}
|
|
|
|
|
|
def set_simple_list_settings(db: Session, project_id: int, data: dict[str, Any]) -> dict[str, Any]:
|
|
if "name_field" in data and data["name_field"] is not None:
|
|
_set_setting(db, project_id, "SimpleList_nameField", str(data["name_field"]))
|
|
if "tech_reserve_percent" in data and data["tech_reserve_percent"] is not None:
|
|
_set_setting(db, project_id, "SimpleList_techReservePercent", str(data["tech_reserve_percent"]))
|
|
if "boards_count" in data and data["boards_count"] is not None:
|
|
_set_setting(db, project_id, "SimpleList_boardsCount", str(data["boards_count"]))
|
|
db.commit()
|
|
return get_simple_list_settings(db, project_id)
|
|
|
|
|
|
def get_table_settings(db: Session, project_id: int, table_type: str) -> dict[str, Any]:
|
|
meta = TABLE_SETTINGS_META.get(table_type, {})
|
|
mappings = get_column_mappings_list(db, project_id, table_type)
|
|
columns = meta.get("columns", [])
|
|
column_mappings = {}
|
|
for col in columns:
|
|
idx = col.get("mapping_index", columns.index(col) if col in columns else 0)
|
|
column_mappings[col["key"]] = mappings[idx] if idx < len(mappings) else ""
|
|
|
|
result: dict[str, Any] = {
|
|
"table_type": table_type,
|
|
"columns": columns,
|
|
"column_mappings": column_mappings,
|
|
"mappings": mappings,
|
|
"property_names": get_property_names(db, project_id),
|
|
"project_params": [
|
|
{"name": n, "value": v} for n, v in get_project_params_map(db, project_id).items()
|
|
],
|
|
}
|
|
|
|
for pp in meta.get("project_params", []):
|
|
idx = pp["mapping_index"]
|
|
result["column_mappings"][pp["key"]] = mappings[idx] if idx < len(mappings) else ""
|
|
|
|
fields: list[dict[str, str]] = []
|
|
for col in columns:
|
|
fields.append({"key": col["key"], "label": col["label"], "source": "property"})
|
|
for pp in meta.get("project_params", []):
|
|
fields.append({"key": pp["key"], "label": pp["label"], "source": "project_param"})
|
|
for cf in meta.get("component_fields", []):
|
|
fields.append({"key": cf["key"], "label": cf["label"], "source": "component_field"})
|
|
result["fields"] = fields
|
|
|
|
if table_type == "simple_list":
|
|
result["simple_list"] = get_simple_list_settings(db, project_id)
|
|
sl = result["simple_list"]
|
|
result["column_mappings"]["name_field"] = sl["name_field"]
|
|
|
|
return result
|
|
|
|
|
|
def update_table_settings(
|
|
db: Session, project_id: int, table_type: str, body: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
meta = TABLE_SETTINGS_META.get(table_type, {})
|
|
mappings = get_column_mappings_list(db, project_id, table_type)
|
|
|
|
if "mappings" in body and body["mappings"] is not None:
|
|
mappings = [str(x) for x in body["mappings"]]
|
|
elif "column_mappings" in body and body["column_mappings"]:
|
|
cm = body["column_mappings"]
|
|
columns = meta.get("columns", [])
|
|
for col in columns:
|
|
key = col["key"]
|
|
if key not in cm:
|
|
continue
|
|
idx = col.get("mapping_index", 0)
|
|
while len(mappings) <= idx:
|
|
mappings.append("")
|
|
mappings[idx] = cm[key] or ""
|
|
for pp in meta.get("project_params", []):
|
|
key = pp["key"]
|
|
idx = pp["mapping_index"]
|
|
if key in cm:
|
|
while len(mappings) <= idx:
|
|
mappings.append("")
|
|
mappings[idx] = cm[key] or ""
|
|
|
|
set_column_mappings_list(db, project_id, table_type, mappings)
|
|
|
|
sl_patch = body.get("simple_list") or {}
|
|
if table_type == "simple_list":
|
|
cm = body.get("column_mappings") or {}
|
|
if "name_field" in cm:
|
|
sl_patch = {**sl_patch, "name_field": cm["name_field"]}
|
|
if sl_patch:
|
|
set_simple_list_settings(db, project_id, sl_patch)
|
|
|
|
return get_table_settings(db, project_id, table_type)
|
|
|
|
|
|
def vedomost_column_map(mappings: list[str]) -> dict[str, str]:
|
|
field_indices = {
|
|
"Name": 0,
|
|
"ProductCode": 1,
|
|
"DocumentCode": 2,
|
|
"Supplier": 3,
|
|
"Note": 5,
|
|
}
|
|
defaults = {
|
|
"Name": "Name",
|
|
"ProductCode": "ProductCode",
|
|
"DocumentCode": "DocumentCode",
|
|
"Supplier": "Supplier",
|
|
"Note": "Note",
|
|
}
|
|
return {
|
|
k: mappings[idx] if idx < len(mappings) and mappings[idx] else defaults[k]
|
|
for k, idx in field_indices.items()
|
|
}
|
|
|
|
|
|
def resolve_where_used(db: Session, project_id: int, mappings: list[str]) -> str:
|
|
params = get_project_params_map(db, project_id)
|
|
if len(mappings) > 4 and mappings[4]:
|
|
return resolve_param_value(mappings[4], params)
|
|
return ""
|
|
|
|
|
|
def resolve_decimal_and_board(
|
|
db: Session, project_id: int, project_decimal: str, project_board: str, table_type: str
|
|
) -> tuple[str, str]:
|
|
params = get_project_params_map(db, project_id)
|
|
mappings = get_column_mappings_list(db, project_id, table_type)
|
|
|
|
dec = project_decimal
|
|
board = project_board
|
|
|
|
if table_type == "specification_pcb":
|
|
if len(mappings) > 7 and mappings[7]:
|
|
dec = resolve_param_value(mappings[7], params)
|
|
elif dec:
|
|
dec = resolve_param_value(dec, params)
|
|
if len(mappings) > 6 and mappings[6]:
|
|
board = resolve_param_value(mappings[6], params)
|
|
elif board:
|
|
board = resolve_param_value(board, params)
|
|
elif dec:
|
|
dec = resolve_param_value(dec, params)
|
|
|
|
return dec, board
|