added web service

This commit is contained in:
2026-09-02 12:48:03 +03:00
parent 84b7eceb9f
commit 02c43072c4
60 changed files with 6148 additions and 22 deletions
View File
View File
+368
View File
@@ -0,0 +1,368 @@
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,
)
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.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}/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)
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:
data = export_pdf(body.table_type, rows, inscriptions)
except ValueError as e:
raise HTTPException(400, str(e)) from e
return Response(
content=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)
View File
+30
View File
@@ -0,0 +1,30 @@
from fastapi import Depends, HTTPException, Security, status
from fastapi.security import APIKeyHeader, HTTPAuthorizationCredentials, HTTPBearer
from app.core.config import get_settings
_bearer = HTTPBearer(auto_error=False)
_api_key = APIKeyHeader(name="X-API-Token", auto_error=False)
def require_token(
bearer: HTTPAuthorizationCredentials | None = Security(_bearer),
api_key: str | None = Security(_api_key),
) -> None:
settings = get_settings()
expected = settings.api_token
if not expected:
return
token = None
if bearer and bearer.credentials:
token = bearer.credentials
elif api_key:
token = api_key
if token != expected:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or missing API token",
)
AuthDep = Depends(require_token)
+33
View File
@@ -0,0 +1,33 @@
from functools import lru_cache
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
database_url: str = "postgresql+psycopg://gost:gost@localhost:5432/gost"
api_token: str = "changeme"
data_dir: str = "./data"
font_path: str = str(Path(__file__).resolve().parents[2] / "fonts" / "GOST_A.TTF")
openrouter_base_url: str = "https://openrouter.ai/api/v1"
openrouter_api_key: str = ""
openrouter_model: str = "deepseek/deepseek-chat"
# SOCKS5 proxy for OpenRouter, e.g. socks5://user:pass@host:1080
openrouter_proxy: str = ""
llm_max_context_rows: int = 80
cors_origins: str = "*"
@property
def cors_origin_list(self) -> list[str]:
if self.cors_origins.strip() == "*":
return ["*"]
return [o.strip() for o in self.cors_origins.split(",") if o.strip()]
@lru_cache
def get_settings() -> Settings:
return Settings()
+23
View File
@@ -0,0 +1,23 @@
from collections.abc import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from app.core.config import get_settings
class Base(DeclarativeBase):
pass
settings = get_settings()
engine = create_engine(settings.database_url, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
+35
View File
@@ -0,0 +1,35 @@
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.routes import router
from app.core.config import get_settings
from app.core.database import Base, engine
import app.models # noqa: F401 — register models
@asynccontextmanager
async def lifespan(_: FastAPI):
settings = get_settings()
Path(settings.data_dir).mkdir(parents=True, exist_ok=True)
Base.metadata.create_all(bind=engine)
yield
app = FastAPI(title="GostGenerator Web API", version="1.0.0", lifespan=lifespan)
settings = get_settings()
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origin_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(router)
@app.get("/")
def root():
return {"service": "GostGenerator Web API", "docs": "/docs"}
+318
View File
@@ -0,0 +1,318 @@
from __future__ import annotations
from datetime import datetime
from typing import Optional
from sqlalchemy import (
Boolean,
DateTime,
Float,
ForeignKey,
Integer,
String,
Text,
UniqueConstraint,
func,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base
class Project(Base):
__tablename__ = "projects"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(255), nullable=False)
status: Mapped[str] = mapped_column(String(64), default="created")
current_variant: Mapped[str] = mapped_column(String(255), default="No Variations")
zip_path: Mapped[Optional[str]] = mapped_column(String(1024), nullable=True)
extract_path: Mapped[Optional[str]] = mapped_column(String(1024), nullable=True)
pcb_doc_name: Mapped[Optional[str]] = mapped_column(String(512), nullable=True)
decimal_number: Mapped[str] = mapped_column(String(255), default="")
board_name: Mapped[str] = mapped_column(String(255), default="")
error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)
variants: Mapped[list[Variant]] = relationship(back_populates="project", cascade="all, delete-orphan")
components: Mapped[list[Component]] = relationship(back_populates="project", cascade="all, delete-orphan")
project_params: Mapped[list[ProjectParam]] = relationship(
back_populates="project", cascade="all, delete-orphan"
)
pcb_data: Mapped[Optional[PcbData]] = relationship(
back_populates="project", uselist=False, cascade="all, delete-orphan"
)
settings: Mapped[list[ProjectSetting]] = relationship(
back_populates="project", cascade="all, delete-orphan"
)
inscriptions: Mapped[list[TitleInscription]] = relationship(
back_populates="project", cascade="all, delete-orphan"
)
designator_mappings: Mapped[list[DesignatorMapping]] = relationship(
back_populates="project", cascade="all, delete-orphan"
)
llm_messages: Mapped[list[LlmMessage]] = relationship(
back_populates="project", cascade="all, delete-orphan"
)
class Variant(Base):
__tablename__ = "variants"
__table_args__ = (UniqueConstraint("project_id", "name", name="uq_variant_project_name"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
name: Mapped[str] = mapped_column(String(255), nullable=False)
project: Mapped[Project] = relationship(back_populates="variants")
class Component(Base):
__tablename__ = "components"
__table_args__ = (UniqueConstraint("project_id", "designator", name="uq_component_designator"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
designator: Mapped[str] = mapped_column(String(128), nullable=False)
project: Mapped[Project] = relationship(back_populates="components")
properties: Mapped[list[ComponentProperty]] = relationship(
back_populates="component", cascade="all, delete-orphan"
)
variant_links: Mapped[list[ComponentVariant]] = relationship(
back_populates="component", cascade="all, delete-orphan"
)
variant_properties: Mapped[list[VariantProperty]] = relationship(
back_populates="component", cascade="all, delete-orphan"
)
class ComponentProperty(Base):
__tablename__ = "component_properties"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
component_id: Mapped[int] = mapped_column(ForeignKey("components.id", ondelete="CASCADE"), index=True)
key: Mapped[str] = mapped_column(String(255), nullable=False)
value: Mapped[str] = mapped_column(Text, default="")
component: Mapped[Component] = relationship(back_populates="properties")
class ComponentVariant(Base):
__tablename__ = "component_variants"
__table_args__ = (
UniqueConstraint("component_id", "variant_id", name="uq_component_variant"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
component_id: Mapped[int] = mapped_column(ForeignKey("components.id", ondelete="CASCADE"), index=True)
variant_id: Mapped[int] = mapped_column(ForeignKey("variants.id", ondelete="CASCADE"), index=True)
is_fitted: Mapped[bool] = mapped_column(Boolean, default=True)
component: Mapped[Component] = relationship(back_populates="variant_links")
variant: Mapped[Variant] = relationship()
class VariantProperty(Base):
__tablename__ = "variant_properties"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
component_id: Mapped[int] = mapped_column(ForeignKey("components.id", ondelete="CASCADE"), index=True)
variant_id: Mapped[int] = mapped_column(ForeignKey("variants.id", ondelete="CASCADE"), index=True)
key: Mapped[str] = mapped_column(String(255), nullable=False)
value: Mapped[str] = mapped_column(Text, default="")
component: Mapped[Component] = relationship(back_populates="variant_properties")
variant: Mapped[Variant] = relationship()
class ProjectParam(Base):
__tablename__ = "project_params"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
name: Mapped[str] = mapped_column(String(255), nullable=False)
value: Mapped[str] = mapped_column(Text, default="")
variant_name: Mapped[str] = mapped_column(String(255), default="")
project: Mapped[Project] = relationship(back_populates="project_params")
class ProjectSetting(Base):
__tablename__ = "project_settings"
__table_args__ = (UniqueConstraint("project_id", "key", name="uq_project_setting"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
key: Mapped[str] = mapped_column(String(255), nullable=False)
value: Mapped[str] = mapped_column(Text, default="")
project: Mapped[Project] = relationship(back_populates="settings")
class DesignatorMapping(Base):
__tablename__ = "designator_mappings"
__table_args__ = (UniqueConstraint("project_id", "prefix", name="uq_designator_prefix"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
prefix: Mapped[str] = mapped_column(String(16), nullable=False)
singular_name: Mapped[str] = mapped_column(String(255), default="")
plural_name: Mapped[str] = mapped_column(String(255), default="")
project: Mapped[Project] = relationship(back_populates="designator_mappings")
class TitleInscription(Base):
__tablename__ = "title_inscriptions"
__table_args__ = (UniqueConstraint("project_id", "field_number", name="uq_inscription_field"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
field_number: Mapped[int] = mapped_column(Integer, nullable=False)
field_value: Mapped[str] = mapped_column(Text, default="")
project: Mapped[Project] = relationship(back_populates="inscriptions")
class PcbData(Base):
__tablename__ = "pcb_data"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(
ForeignKey("projects.id", ondelete="CASCADE"), unique=True, index=True
)
layer_count: Mapped[int] = mapped_column(Integer, default=0)
project: Mapped[Project] = relationship(back_populates="pcb_data")
diel_materials: Mapped[list[DielMaterial]] = relationship(
back_populates="pcb_data", cascade="all, delete-orphan"
)
class DielMaterial(Base):
__tablename__ = "diel_materials"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
pcb_data_id: Mapped[int] = mapped_column(ForeignKey("pcb_data.id", ondelete="CASCADE"), index=True)
name: Mapped[str] = mapped_column(String(255), default="DielMaterial")
value: Mapped[str] = mapped_column(String(255), default="")
height: Mapped[float] = mapped_column(Float, default=0.0)
diel_type: Mapped[int] = mapped_column(Integer, default=1)
layer_number: Mapped[int] = mapped_column(Integer, default=0)
pcb_data: Mapped[PcbData] = relationship(back_populates="diel_materials")
class PerechenRow(Base):
__tablename__ = "perechen_rows"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
row_index: Mapped[int] = mapped_column(Integer, default=0)
page_number: Mapped[int] = mapped_column(Integer, default=1)
position: Mapped[str] = mapped_column(Text, default="")
designation: Mapped[str] = mapped_column(Text, default="")
quantity: Mapped[str] = mapped_column(String(64), default="")
note: Mapped[str] = mapped_column(Text, default="")
is_header: Mapped[bool] = mapped_column(Boolean, default=False)
is_empty: Mapped[bool] = mapped_column(Boolean, default=False)
is_auto_generated: Mapped[bool] = mapped_column(Boolean, default=True)
stretch: Mapped[bool] = mapped_column(Boolean, default=False)
is_underline: Mapped[bool] = mapped_column(Boolean, default=False)
class SpecificationPcbRow(Base):
__tablename__ = "specification_pcb_rows"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
row_index: Mapped[int] = mapped_column(Integer, default=0)
page_number: Mapped[int] = mapped_column(Integer, default=1)
format: Mapped[str] = mapped_column(String(64), default="")
zone: Mapped[str] = mapped_column(String(64), default="")
position: Mapped[str] = mapped_column(String(64), default="")
designation: Mapped[str] = mapped_column(Text, default="")
name: Mapped[str] = mapped_column(Text, default="")
quantity: Mapped[str] = mapped_column(String(64), default="")
note: Mapped[str] = mapped_column(Text, default="")
is_header: Mapped[bool] = mapped_column(Boolean, default=False)
is_empty: Mapped[bool] = mapped_column(Boolean, default=False)
is_auto_generated: Mapped[bool] = mapped_column(Boolean, default=True)
stretch: Mapped[bool] = mapped_column(Boolean, default=False)
is_underline: Mapped[bool] = mapped_column(Boolean, default=False)
class SpecificationRow(Base):
__tablename__ = "specification_rows"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
row_index: Mapped[int] = mapped_column(Integer, default=0)
page_number: Mapped[int] = mapped_column(Integer, default=1)
format: Mapped[str] = mapped_column(String(64), default="")
zone: Mapped[str] = mapped_column(String(64), default="")
position: Mapped[str] = mapped_column(String(64), default="")
designation: Mapped[str] = mapped_column(Text, default="")
name: Mapped[str] = mapped_column(Text, default="")
quantity: Mapped[str] = mapped_column(String(64), default="")
note: Mapped[str] = mapped_column(Text, default="")
is_header: Mapped[bool] = mapped_column(Boolean, default=False)
is_empty: Mapped[bool] = mapped_column(Boolean, default=False)
is_auto_generated: Mapped[bool] = mapped_column(Boolean, default=True)
stretch: Mapped[bool] = mapped_column(Boolean, default=False)
is_underline: Mapped[bool] = mapped_column(Boolean, default=False)
class VedomostRow(Base):
__tablename__ = "vedomost_rows"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
row_index: Mapped[int] = mapped_column(Integer, default=0)
page_number: Mapped[int] = mapped_column(Integer, default=1)
name: Mapped[str] = mapped_column(Text, default="")
product_code: Mapped[str] = mapped_column(Text, default="")
document_code: Mapped[str] = mapped_column(Text, default="")
supplier: Mapped[str] = mapped_column(Text, default="")
where_used: Mapped[str] = mapped_column(Text, default="")
quantity_per_item: Mapped[str] = mapped_column(String(64), default="")
quantity_in_set: Mapped[str] = mapped_column(String(64), default="")
quantity_for_reg: Mapped[str] = mapped_column(String(64), default="")
total_quantity: Mapped[str] = mapped_column(String(64), default="")
note: Mapped[str] = mapped_column(Text, default="")
is_header: Mapped[bool] = mapped_column(Boolean, default=False)
is_empty: Mapped[bool] = mapped_column(Boolean, default=False)
is_auto_generated: Mapped[bool] = mapped_column(Boolean, default=True)
stretch: Mapped[bool] = mapped_column(Boolean, default=False)
is_underline: Mapped[bool] = mapped_column(Boolean, default=False)
class SimpleListRow(Base):
__tablename__ = "simple_list_rows"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
row_index: Mapped[int] = mapped_column(Integer, default=0)
designator: Mapped[str] = mapped_column(Text, default="")
name: Mapped[str] = mapped_column(Text, default="")
quantity: Mapped[str] = mapped_column(String(64), default="")
is_auto_generated: Mapped[bool] = mapped_column(Boolean, default=True)
is_empty: Mapped[bool] = mapped_column(Boolean, default=False)
class LlmMessage(Base):
__tablename__ = "llm_messages"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
project_id: Mapped[int] = mapped_column(ForeignKey("projects.id", ondelete="CASCADE"), index=True)
role: Mapped[str] = mapped_column(String(32), nullable=False)
content: Mapped[str] = mapped_column(Text, default="")
table_type: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
proposed_edits: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
project: Mapped[Project] = relationship(back_populates="llm_messages")
+193
View File
@@ -0,0 +1,193 @@
from __future__ import annotations
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel, Field
class ProjectCreate(BaseModel):
name: str = Field(min_length=1, max_length=255)
class ProjectUpdate(BaseModel):
name: Optional[str] = None
current_variant: Optional[str] = None
decimal_number: Optional[str] = None
board_name: Optional[str] = None
class ProjectOut(BaseModel):
id: int
name: str
status: str
current_variant: str
zip_path: Optional[str] = None
pcb_doc_name: Optional[str] = None
decimal_number: str
board_name: str
error_message: Optional[str] = None
created_at: datetime
updated_at: datetime
variants: list[str] = []
component_count: int = 0
layer_count: int = 0
model_config = {"from_attributes": True}
class ComponentPropertyOut(BaseModel):
key: str
value: str
class ComponentOut(BaseModel):
id: int
designator: str
properties: list[ComponentPropertyOut] = []
is_fitted: bool = True
model_config = {"from_attributes": True}
class ProjectParamOut(BaseModel):
id: int
name: str
value: str
variant_name: str = ""
model_config = {"from_attributes": True}
class DielMaterialOut(BaseModel):
name: str
value: str
height: float
diel_type: int
layer_number: int
model_config = {"from_attributes": True}
class PcbInfoOut(BaseModel):
layer_count: int = 0
materials: list[DielMaterialOut] = []
class InscriptionOut(BaseModel):
field_number: int
field_value: str
class InscriptionsUpdate(BaseModel):
inscriptions: dict[int, str]
class TableRowBase(BaseModel):
id: Optional[int] = None
row_index: int = 0
page_number: int = 1
is_header: bool = False
is_empty: bool = False
is_auto_generated: bool = True
stretch: bool = False
is_underline: bool = False
class PerechenRowOut(TableRowBase):
position: str = ""
designation: str = ""
quantity: str = ""
note: str = ""
class SpecPcbRowOut(TableRowBase):
format: str = ""
zone: str = ""
position: str = ""
designation: str = ""
name: str = ""
quantity: str = ""
note: str = ""
class SpecRowOut(TableRowBase):
format: str = ""
zone: str = ""
position: str = ""
designation: str = ""
name: str = ""
quantity: str = ""
note: str = ""
class VedomostRowOut(TableRowBase):
name: str = ""
product_code: str = ""
document_code: str = ""
supplier: str = ""
where_used: str = ""
quantity_per_item: str = ""
quantity_in_set: str = ""
quantity_for_reg: str = ""
total_quantity: str = ""
note: str = ""
class SimpleListRowOut(BaseModel):
id: Optional[int] = None
row_index: int = 0
designator: str = ""
name: str = ""
quantity: str = ""
is_auto_generated: bool = True
is_empty: bool = False
class TableRowsResponse(BaseModel):
table_type: str
rows: list[dict[str, Any]]
class TableRowsPatch(BaseModel):
rows: list[dict[str, Any]]
class GenerateTableRequest(BaseModel):
name_field: Optional[str] = None
tech_reserve_percent: Optional[float] = None
boards_count: Optional[int] = None
class LlmChatRequest(BaseModel):
message: str = Field(min_length=1)
table_type: str
class LlmEdit(BaseModel):
op: str
row_id: Optional[int] = None
row_index: Optional[int] = None
fields: dict[str, Any] = {}
class LlmChatResponse(BaseModel):
reply: str
edits: list[LlmEdit] = []
message_id: Optional[int] = None
class LlmApplyRequest(BaseModel):
table_type: str
edits: list[LlmEdit]
class ExportRequest(BaseModel):
table_type: str
format: str = "pdf" # pdf | xlsx
class DesignatorMappingOut(BaseModel):
prefix: str
singular_name: str
plural_name: str
+359
View File
@@ -0,0 +1,359 @@
"""Altium .PrjPcb / .SchDoc / .PcbDoc parser (ported from desktop AltiumParser)."""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
@dataclass
class ComponentProperty:
name: str
text: str
@dataclass
class DielProperty:
name: str
value: str
height: float
diel_type: int
layer_number: int
@dataclass
class ProjectData:
variant_names: list[str] = field(default_factory=lambda: ["No Variations"])
components_list: list[list[ComponentProperty]] = field(default_factory=list)
components_variant_list: list[list[list[ComponentProperty]]] = field(default_factory=list)
components_prop_variant_list: list[list[ComponentProperty]] = field(default_factory=list)
component_variant_prop_list: list[ComponentProperty] = field(default_factory=list)
dnf_designators_list: list[str] = field(default_factory=list)
dnf_variant_designators_list: list[list[str]] = field(default_factory=list)
fitted_designators_list: list[str] = field(default_factory=list)
fitted_variant_designators_list: list[list[str]] = field(default_factory=list)
prj_params_variant_list: list[list[list[str]]] = field(default_factory=list)
pcb_layer_count: int = 0
pcb_diel_materials: list[DielProperty] = field(default_factory=list)
pcb_doc_file_name: str = ""
is_waiting_variant_description: bool = False
current_variant_number: int = 0
def _read_cp1251_lines(path: Path) -> list[str]:
raw = path.read_bytes()
text = raw.decode("cp1251", errors="replace")
return [line.rstrip("\r\n") for line in text.splitlines()]
def _read_pipe_chunks(path: Path) -> list[str]:
"""Read Altium binary-ish docs and extract pipe-separated ASCII chunks per line."""
raw = path.read_bytes()
text = raw.decode("cp1251", errors="replace")
# Also try latin-1 overlay for pipe records that may have been mangled
return [line for line in text.splitlines() if line.strip()]
class AltiumParser:
def __init__(self) -> None:
self.last_error = ""
self.pcb_doc_file_name = ""
def parse_prjpcb(self, filename: str | Path) -> ProjectData:
path = Path(filename)
if not path.exists():
self.last_error = f"File not found: {path}"
raise FileNotFoundError(self.last_error)
data = ProjectData()
data.prj_params_variant_list = [[]] # index 0 = No Variations
lines = _read_cp1251_lines(path)
prev = ""
prev_prev = ""
for line in lines:
prj = line.strip()
self._parse_project_variant_section(prj, data)
self._parse_component_variations(prj, data)
self._parse_project_parameters(prev_prev, prev, prj, data)
self._parse_schdoc_files(prj, path, data)
self._parse_pcbdoc_files(prj, path, data)
prev_prev, prev = prev, prj
# finalize last variant buffers if any
if data.components_prop_variant_list or data.dnf_designators_list or data.fitted_designators_list:
self._finalize_current_variant(data)
self.pcb_doc_file_name = data.pcb_doc_file_name
return data
def _finalize_current_variant(self, data: ProjectData) -> None:
if data.component_variant_prop_list:
data.components_prop_variant_list.append(list(data.component_variant_prop_list))
data.component_variant_prop_list.clear()
if data.components_prop_variant_list:
data.components_variant_list.append(list(data.components_prop_variant_list))
data.components_prop_variant_list.clear()
data.dnf_variant_designators_list.append(list(data.dnf_designators_list))
data.fitted_variant_designators_list.append(list(data.fitted_designators_list))
data.dnf_designators_list.clear()
data.fitted_designators_list.clear()
def _parse_project_variant_section(self, prj: str, data: ProjectData) -> None:
upper = prj.upper()
if len(prj) >= 15 and upper.startswith("[PROJECTVARIANT"):
data.is_waiting_variant_description = True
# Extract number if present
m = re.search(r"(\d+)", prj)
data.current_variant_number = int(m.group(1)) if m else len(data.variant_names)
while len(data.prj_params_variant_list) <= data.current_variant_number:
data.prj_params_variant_list.append([])
return
if data.is_waiting_variant_description and upper.startswith("DESCRIPTION="):
name = prj[12:].strip()
if name and name not in data.variant_names:
data.variant_names.append(name)
data.is_waiting_variant_description = False
return
if len(prj) >= 19 and upper.startswith("PARAMVARIATIONCOUNT"):
self._finalize_current_variant(data)
def _parse_component_variations(self, prj: str, data: ProjectData) -> None:
upper = prj.upper()
parts = prj.split("|")
if not parts:
return
values = parts[0].split("=")
if len(values) <= 2:
return
if "VARIATION" in values[0].upper() and "DESIGNATOR" in values[1].upper():
designator = values[2]
prop_list = [ComponentProperty("Designator", designator)]
kind = "0"
if len(parts) > 2:
kind_part = parts[2]
if "=" in kind_part:
kind = kind_part.split("=", 1)[1]
prop_list.append(ComponentProperty("Kind", kind))
if kind == "1":
data.dnf_designators_list.append(designator)
else:
data.fitted_designators_list.append(designator)
# flush previous component props into list
if data.component_variant_prop_list:
# keep designator props as start of new component
pass
data.components_prop_variant_list.append(prop_list)
return
if "PARAMVARIATION" in values[0].upper() and "PARAMETERNAME" in values[1].upper():
prop_name = values[2]
prop_text = ""
if len(parts) > 1 and "=" in parts[1]:
prop_text = parts[1].split("=", 1)[1]
data.component_variant_prop_list.append(ComponentProperty(prop_name, prop_text))
# attach to last variation component if present
if data.components_prop_variant_list:
data.components_prop_variant_list[-1].append(ComponentProperty(prop_name, prop_text))
def _parse_project_parameters(
self, prev_prev: str, prev: str, prj: str, data: ProjectData
) -> None:
if not prev_prev.upper().startswith("[PARAMETER"):
return
if not prev.upper().startswith("NAME="):
return
if not prj.upper().startswith("VALUE="):
return
name = prev[5:]
value = prj[6:]
idx = data.current_variant_number if "_" in prev_prev else 0
while len(data.prj_params_variant_list) <= idx:
data.prj_params_variant_list.append([])
data.prj_params_variant_list[idx].append([name, value])
def _parse_schdoc_files(self, prj: str, prj_path: Path, data: ProjectData) -> None:
if len(prj) < 19:
return
if prj[:13].upper() != "DOCUMENTPATH=":
return
if prj[-6:].upper() != "SCHDOC":
return
rel = prj[13:]
sch_path = (prj_path.parent / rel).resolve()
if sch_path.exists():
self._parse_schdoc_file(sch_path, data.components_list)
def _parse_pcbdoc_files(self, prj: str, prj_path: Path, data: ProjectData) -> None:
if len(prj) < 19:
return
if prj[:13].upper() != "DOCUMENTPATH=":
return
if prj[-6:].upper() != "PCBDOC":
return
rel = prj[13:]
pcb_path = (prj_path.parent / rel).resolve()
if pcb_path.exists():
if self._parse_pcbdoc_file(pcb_path, data):
data.pcb_doc_file_name = str(pcb_path)
def _parse_schdoc_file(self, path: Path, components_list: list[list[ComponentProperty]]) -> None:
lines = _read_pipe_chunks(path)
is_component = False
is_no_bom = False
is_first_part = True
component_props: list[ComponentProperty] = []
def flush():
nonlocal component_props, is_no_bom, is_first_part, is_component
if not is_no_bom and component_props and is_first_part:
components_list.append(list(component_props))
component_props = []
is_no_bom = False
is_first_part = True
for line in lines:
parts = line.split("|")
i = 0
while i < len(parts):
part = parts[i]
up = part.upper()
if len(part) == 8 and up == "RECORD=1":
if is_component:
flush()
is_component = True
i += 1
continue
if is_component:
if up.startswith("COMPONENTKINDVERSION2=5"):
is_no_bom = True
elif up.startswith("CURRENTPARTID="):
if up != "CURRENTPARTID=1" or len(part) > 15:
is_first_part = False
elif (
i + 1 < len(parts)
and up.startswith("TEXT=")
and parts[i + 1].upper().startswith("NAME=")
):
text = part[5:]
name = parts[i + 1][5:]
if not is_no_bom:
component_props.append(ComponentProperty(name, text))
i += 2
continue
elif up.startswith("HEADER="):
flush()
is_component = False
i += 1
if is_component:
flush()
def _parse_pcbdoc_file(self, path: Path, data: ProjectData) -> bool:
lines = _read_pipe_chunks(path)
signal_layerset: Optional[str] = None
layer_count = 0
materials: list[DielProperty] = []
layerset_re = re.compile(r"LAYERSET(\d+)NAME=&Signal Layers", re.I)
v9_name_re = re.compile(r"V9_STACK_LAYER(\d+)_NAME=", re.I)
v9_diel_re = re.compile(r"V9_STACK_LAYER(\d+)_DIELTYPE=(\d+)", re.I)
v8_diel_re = re.compile(r"LAYER_V8_(\d+)_DIELTYPE=(\d+)", re.I)
found_layerset = False
for line in lines:
m = layerset_re.search(line)
if m:
signal_layerset = m.group(1)
found_layerset = True
for line in lines:
parts = line.split("|")
joined = line
if signal_layerset:
key = f"LAYERSET{signal_layerset}LAYERS="
for part in parts:
if part.upper().startswith(key.upper()):
layers = part.split("=", 1)[1].split(",")
layer_count = sum(1 for x in layers if x.strip() and x.strip() != "MultiLayer")
for m in v9_name_re.finditer(joined):
n = int(m.group(1))
layer_count = max(layer_count, n + 1)
for m in v9_diel_re.finditer(joined):
n = int(m.group(1))
diel_type = int(m.group(2))
height = 0.0
value = ""
hkey = f"V9_STACK_LAYER{n}_DIELHEIGHT="
mkey = f"V9_STACK_LAYER{n}_DIELMATERIAL="
for part in parts:
pu = part.upper()
if pu.startswith(hkey.upper()):
hraw = part.split("=", 1)[1]
hraw = re.sub(r"mil", "", hraw, flags=re.I).strip()
try:
height = float(hraw) * 0.0254
except ValueError:
height = 0.0
if pu.startswith(mkey.upper()):
value = part.split("=", 1)[1]
if value and height > 0:
materials.append(
DielProperty("DielMaterial", value, height, diel_type, n)
)
for m in v8_diel_re.finditer(joined):
n = int(m.group(1))
diel_type = int(m.group(2))
height = 0.0
value = ""
hkey = f"LAYER_V8_{n}_DIELHEIGHT="
mkey = f"LAYER_V8_{n}_DIELMATERIAL="
for part in parts:
pu = part.upper()
if pu.startswith(hkey.upper()):
hraw = part.split("=", 1)[1]
hraw = re.sub(r"mil", "", hraw, flags=re.I).strip()
try:
height = float(hraw) * 0.0254
except ValueError:
height = 0.0
if pu.startswith(mkey.upper()):
value = part.split("=", 1)[1]
if value and height > 0:
materials.append(
DielProperty("DielMaterial", value, height, diel_type, n)
)
if not found_layerset and layer_count > 0:
layer_count -= 1
data.pcb_layer_count = max(layer_count, 0)
data.pcb_diel_materials = materials
return True
def make_complex_string(text: str, props: list[ComponentProperty]) -> str:
if not text.startswith("="):
return text
expr = text[1:]
mapping = {p.name: p.text for p in props}
def repl(m: re.Match) -> str:
key = m.group(1)
return mapping.get(key, "")
return re.sub(r"['\"]([^'\"]+)['\"]", repl, expr)
def find_prjpcb(extract_dir: Path) -> Optional[Path]:
candidates = list(extract_dir.rglob("*.PrjPcb")) + list(extract_dir.rglob("*.prjpcb"))
return candidates[0] if candidates else None
+178
View File
@@ -0,0 +1,178 @@
"""Default designator type names (singular/plural) by letter prefix."""
from __future__ import annotations
import re
# Minimal useful defaults; projects can override via DB
DEFAULT_MAPPINGS: dict[str, tuple[str, str]] = {
"C": ("Конденсатор", "Конденсаторы"),
"R": ("Резистор", "Резисторы"),
"L": ("Катушка индуктивности", "Катушки индуктивности"),
"D": ("Диод", "Диоды"),
"VD": ("Диод", "Диоды"),
"VT": ("Транзистор", "Транзисторы"),
"Q": ("Транзистор", "Транзисторы"),
"DA": ("Микросхема", "Микросхемы"),
"DD": ("Микросхема", "Микросхемы"),
"U": ("Микросхема", "Микросхемы"),
"X": ("Соединитель", "Соединители"),
"XP": ("Соединитель", "Соединители"),
"XS": ("Соединитель", "Соединители"),
"FU": ("Предохранитель", "Предохранители"),
"F": ("Предохранитель", "Предохранители"),
"SA": ("Переключатель", "Переключатели"),
"SB": ("Кнопка", "Кнопки"),
"HL": ("Индикатор", "Индикаторы"),
"HG": ("Индикатор", "Индикаторы"),
"G": ("Генератор", "Генераторы"),
"T": ("Трансформатор", "Трансформаторы"),
"TV": ("Трансформатор", "Трансформаторы"),
"K": ("Реле", "Реле"),
"B": ("Пьезоэлемент", "Пьезоэлементы"),
"Z": ("Фильтр", "Фильтры"),
}
def split_designator(designator: str) -> tuple[str, str]:
m = re.match(r"^([A-Za-z]+)(\d*)$", designator.strip())
if m:
return m.group(1), m.group(2)
for i, ch in enumerate(designator):
if ch.isdigit():
return designator[:i], designator[i:]
return designator, ""
def designator_sort_key(designator: str) -> tuple[str, int, str]:
letter, num = split_designator(designator)
try:
n = int(num) if num else 0
except ValueError:
n = 0
return letter, n, num
return letter.upper(), n, num
def prefix_from_designator(designator: str) -> str:
letter, _ = split_designator(designator)
return letter.upper()
def get_type_names(
designator: str, mappings: dict[str, tuple[str, str]] | None = None
) -> tuple[str, str]:
maps = {**DEFAULT_MAPPINGS, **(mappings or {})}
prefix = prefix_from_designator(designator)
if prefix in maps:
return maps[prefix]
# try 2-letter then 1-letter
if len(prefix) >= 2 and prefix[:2] in maps:
return maps[prefix[:2]]
if prefix[:1] in maps:
return maps[prefix[:1]]
return (prefix, prefix)
def format_designator_range(designators: list[str]) -> str:
if not designators:
return ""
if len(designators) == 1:
return designators[0]
if len(designators) == 2:
return f"{designators[0]}, {designators[1]}"
return f"{designators[0]}-{designators[-1]}"
def format_consecutive_ranges(designators: list[str]) -> str:
"""Group consecutive numbered designators into ranges."""
if not designators:
return ""
parsed = []
for d in designators:
letter, num = split_designator(d)
try:
n = int(num) if num else None
except ValueError:
n = None
parsed.append((d, letter, n))
segments: list[list[str]] = []
current: list[str] = []
last_letter = None
last_num = None
for d, letter, n in parsed:
if not current:
current = [d]
last_letter, last_num = letter, n
continue
if letter == last_letter and n is not None and last_num is not None and n == last_num + 1:
current.append(d)
last_num = n
else:
segments.append(current)
current = [d]
last_letter, last_num = letter, n
if current:
segments.append(current)
parts = []
for seg in segments:
if len(seg) == 1:
parts.append(seg[0])
elif len(seg) == 2:
parts.append(f"{seg[0]}, {seg[1]}")
else:
parts.append(f"{seg[0]}-{seg[-1]}")
return ", ".join(parts)
def split_long_text(text: str, max_length: int) -> list[str]:
if not text or len(text) <= max_length:
return [text] if text is not None else [""]
parts: list[str] = []
remaining = text
while len(remaining) > max_length:
chunk = remaining[:max_length]
split_at = chunk.rfind(",")
if split_at < max_length // 3:
split_at = chunk.rfind(" ")
if split_at < max_length // 3:
split_at = max_length
parts.append(remaining[:split_at].rstrip(", ").strip())
remaining = remaining[split_at:].lstrip(", ").strip()
if remaining:
parts.append(remaining)
return parts or [""]
def resolve_expression(value: str, props: dict[str, str]) -> str:
if not value.startswith("="):
return value
expr = value[1:]
result = []
i = 0
while i < len(expr):
ch = expr[i]
if ch in "'\"":
quote = ch
i += 1
start = i
while i < len(expr) and expr[i] != quote:
i += 1
result.append(expr[start:i])
i += 1
elif ch == "+":
i += 1
elif ch.isspace():
i += 1
else:
start = i
while i < len(expr) and expr[i] not in "+'\"":
i += 1
key = expr[start:i].strip()
val = props.get(key, "")
if val.startswith("="):
val = resolve_expression(val, props)
result.append(val)
return "".join(result)
+73
View File
@@ -0,0 +1,73 @@
from __future__ import annotations
from io import BytesIO
from typing import Any
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font
HEADERS = {
"perechen": ["Поз. обозначение", "Наименование", "Кол.", "Примечание"],
"specification_pcb": ["Формат", "Зона", "Поз.", "Обозначение", "Наименование", "Кол.", "Примечание"],
"specification": ["Формат", "Зона", "Поз.", "Обозначение", "Наименование", "Кол.", "Примечание"],
"vedomost": [
"Наименование",
"Код продукции",
"Обозн. документа на поставку",
"Поставщик",
"Куда входит",
"Кол. на изделие",
"Кол. в комплекте",
"Кол. на регулир.",
"Всего",
"Примечание",
],
"simple_list": ["Поз. обозначение", "Наименование", "Кол."],
}
FIELDS = {
"perechen": ["position", "designation", "quantity", "note"],
"specification_pcb": ["format", "zone", "position", "designation", "name", "quantity", "note"],
"specification": ["format", "zone", "position", "designation", "name", "quantity", "note"],
"vedomost": [
"name",
"product_code",
"document_code",
"supplier",
"where_used",
"quantity_per_item",
"quantity_in_set",
"quantity_for_reg",
"total_quantity",
"note",
],
"simple_list": ["designator", "name", "quantity"],
}
def export_xlsx(table_type: str, rows: list[dict[str, Any]], title: str = "") -> bytes:
wb = Workbook()
ws = wb.active
ws.title = table_type[:31]
headers = HEADERS[table_type]
fields = FIELDS[table_type]
if title:
ws.append([title])
ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=len(headers))
ws["A1"].font = Font(bold=True, size=14)
ws.append(headers)
for cell in ws[ws.max_row]:
cell.font = Font(bold=True)
cell.alignment = Alignment(wrap_text=True)
for row in rows:
if row.get("is_empty"):
ws.append([""] * len(fields))
continue
values = [row.get(f, "") or "" for f in fields]
ws.append(values)
if row.get("is_header"):
for cell in ws[ws.max_row]:
cell.font = Font(bold=True, underline="single")
buf = BytesIO()
wb.save(buf)
return buf.getvalue()
+124
View File
@@ -0,0 +1,124 @@
from __future__ import annotations
import json
import re
from typing import Any
import httpx
from app.core.config import get_settings
SYSTEM_PROMPT = """Ты помощник инженера по редактированию ГОСТ-таблиц (перечень, спецификация, ведомость).
Пользователь даёт команды на русском. Ты отвечаешь ТОЛЬКО валидным JSON без markdown:
{
"reply": "краткий ответ пользователю",
"edits": [
{"op": "update_row", "row_id": 123, "fields": {"note": "..."}},
{"op": "update_row", "row_index": 5, "fields": {"designation": "..."}},
{"op": "add_row", "row_index": 10, "fields": {...}},
{"op": "delete_row", "row_id": 123}
]
}
Правила:
- Меняй только то, о чём просят.
- Используй row_id из снимка таблицы, если есть.
- Не выдумывай поля вне списка колонок.
- Если правок нет — edits: [].
"""
def _extract_json(text: str) -> dict[str, Any]:
text = text.strip()
if text.startswith("```"):
text = re.sub(r"^```(?:json)?\s*", "", text)
text = re.sub(r"\s*```$", "", text)
try:
return json.loads(text)
except json.JSONDecodeError:
m = re.search(r"\{.*\}", text, re.S)
if m:
return json.loads(m.group(0))
return {"reply": text, "edits": []}
def compress_rows(rows: list[dict[str, Any]], max_rows: int) -> list[dict[str, Any]]:
slim = []
for r in rows[:max_rows]:
item = {"row_id": r.get("id"), "row_index": r.get("row_index")}
for k, v in r.items():
if k in ("id", "project_id", "is_auto_generated", "stretch"):
continue
if v not in ("", None, False, 0) or k in ("is_header", "is_empty"):
item[k] = v
slim.append(item)
return slim
async def chat_edit_table(
message: str,
table_type: str,
rows: list[dict[str, Any]],
) -> dict[str, Any]:
settings = get_settings()
if not settings.openrouter_api_key:
return {
"reply": "OpenRouter API key не настроен (OPENROUTER_API_KEY).",
"edits": [],
}
snapshot = compress_rows(rows, settings.llm_max_context_rows)
user_content = (
f"Тип таблицы: {table_type}\n"
f"Снимок строк (до {settings.llm_max_context_rows}):\n"
f"{json.dumps(snapshot, ensure_ascii=False)}\n\n"
f"Команда пользователя: {message}"
)
payload = {
"model": settings.openrouter_model,
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_content},
],
"temperature": 0.2,
"max_tokens": 2000,
}
headers = {
"Authorization": f"Bearer {settings.openrouter_api_key}",
"Content-Type": "application/json",
"HTTP-Referer": "https://gostgenerator.local",
"X-Title": "GostGenerator Web",
}
url = settings.openrouter_base_url.rstrip("/") + "/chat/completions"
proxy = (settings.openrouter_proxy or "").strip() or None
if proxy and "://" not in proxy:
proxy = f"socks5://{proxy}"
client_kwargs: dict[str, Any] = {"timeout": 90.0}
if proxy:
client_kwargs["proxy"] = proxy
async with httpx.AsyncClient(**client_kwargs) as client:
resp = await client.post(url, headers=headers, json=payload)
resp.raise_for_status()
data = resp.json()
content = data["choices"][0]["message"]["content"]
parsed = _extract_json(content)
edits = parsed.get("edits") or []
# normalize
norm_edits = []
for e in edits:
if not isinstance(e, dict):
continue
norm_edits.append(
{
"op": e.get("op", "update_row"),
"row_id": e.get("row_id"),
"row_index": e.get("row_index"),
"fields": e.get("fields") or {},
}
)
return {"reply": parsed.get("reply") or "", "edits": norm_edits}
+182
View File
@@ -0,0 +1,182 @@
"""Simplified GOST A4 PDF export with frame and table."""
from __future__ import annotations
from io import BytesIO
from pathlib import Path
from typing import Any
from reportlab.lib.pagesizes import A4
from reportlab.lib.units import mm
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
from reportlab.pdfgen import canvas
from app.core.config import get_settings
_font_registered = False
FONT_NAME = "GOST_A"
def _ensure_font() -> str:
global _font_registered
if not _font_registered:
path = Path(get_settings().font_path)
if path.exists():
pdfmetrics.registerFont(TTFont(FONT_NAME, str(path)))
_font_registered = True
return FONT_NAME
return "Helvetica"
return FONT_NAME if Path(get_settings().font_path).exists() else "Helvetica"
DOC_TITLES = {
"perechen": "Перечень элементов",
"specification_pcb": "Спецификация",
"specification": "Спецификация",
"vedomost": "Ведомость покупных изделий",
}
COLUMNS = {
"perechen": [
("position", 30 * mm, "Поз."),
("designation", 100 * mm, "Наименование"),
("quantity", 15 * mm, "Кол."),
("note", 35 * mm, "Прим."),
],
"specification_pcb": [
("format", 12 * mm, "Форм."),
("zone", 10 * mm, "Зона"),
("position", 10 * mm, "Поз."),
("designation", 40 * mm, "Обозн."),
("name", 55 * mm, "Наименование"),
("quantity", 12 * mm, "Кол."),
("note", 30 * mm, "Прим."),
],
"specification": [
("format", 12 * mm, "Форм."),
("zone", 10 * mm, "Зона"),
("position", 10 * mm, "Поз."),
("designation", 40 * mm, "Обозн."),
("name", 55 * mm, "Наименование"),
("quantity", 12 * mm, "Кол."),
("note", 30 * mm, "Прим."),
],
"vedomost": [
("name", 40 * mm, "Наименование"),
("product_code", 20 * mm, "Код"),
("document_code", 25 * mm, "Док."),
("supplier", 20 * mm, "Пост."),
("where_used", 20 * mm, "Куда"),
("quantity_per_item", 12 * mm, "На изд."),
("total_quantity", 12 * mm, "Всего"),
("note", 20 * mm, "Прим."),
],
}
ROWS_FIRST = {"perechen": 25, "specification_pcb": 26, "specification": 27, "vedomost": 24}
ROWS_OTHER = {"perechen": 32, "specification_pcb": 32, "specification": 33, "vedomost": 29}
def _draw_frame(c: canvas.Canvas, w: float, h: float, page: int, total: int, inscriptions: dict[int, str], doc_type: str):
font = _ensure_font()
margin_left = 20 * mm
margin_right = 5 * mm
margin_top = 5 * mm
margin_bottom = 5 * mm
c.setLineWidth(0.8)
c.rect(margin_left, margin_bottom, w - margin_left - margin_right, h - margin_top - margin_bottom)
# Title block (simplified bottom-right)
block_h = 55 * mm if page == 1 else 15 * mm
block_w = 185 * mm
x0 = w - margin_right - block_w
y0 = margin_bottom
c.setLineWidth(0.5)
c.rect(x0, y0, block_w, block_h)
c.setFont(font, 8)
name = inscriptions.get(1, "")
designation = inscriptions.get(2, "")
org = inscriptions.get(9, "")
title = DOC_TITLES.get(doc_type, "")
c.drawString(x0 + 2 * mm, y0 + block_h - 6 * mm, f"{title}")
c.drawString(x0 + 2 * mm, y0 + block_h - 12 * mm, designation[:60])
c.drawString(x0 + 2 * mm, y0 + block_h - 18 * mm, name[:60])
if page == 1:
c.drawString(x0 + 2 * mm, y0 + 8 * mm, org[:40])
c.drawString(x0 + 2 * mm, y0 + 3 * mm, f"Разраб. {inscriptions.get(111, '')}")
c.drawString(x0 + 50 * mm, y0 + 3 * mm, f"Пров. {inscriptions.get(112, '')}")
c.drawRightString(x0 + block_w - 2 * mm, y0 + 3 * mm, f"Лист {page}/{total}")
def _chunk_rows(rows: list[dict[str, Any]], first: int, other: int) -> list[list[dict[str, Any]]]:
if not rows:
return [[]]
pages: list[list[dict[str, Any]]] = []
i = 0
limit = first
while i < len(rows):
pages.append(rows[i : i + limit])
i += limit
limit = other
return pages
def export_pdf(
table_type: str,
rows: list[dict[str, Any]],
inscriptions: dict[int, str] | None = None,
) -> bytes:
if table_type == "simple_list":
raise ValueError("PDF export is not available for simple_list")
inscriptions = inscriptions or {}
font = _ensure_font()
cols = COLUMNS[table_type]
first = ROWS_FIRST[table_type]
other = ROWS_OTHER[table_type]
pages = _chunk_rows(rows, first, other)
total = max(len(pages), 1)
buf = BytesIO()
c = canvas.Canvas(buf, pagesize=A4)
w, h = A4
for page_idx, page_rows in enumerate(pages, start=1):
_draw_frame(c, w, h, page_idx, total, inscriptions, table_type)
# table area
left = 20 * mm
top = h - 10 * mm
row_h = 6 * mm
header_y = top - 8 * mm
# column headers
x = left + 2 * mm
c.setFont(font, 7)
for field, width, label in cols:
c.drawString(x, header_y, label)
x += width
c.line(left, header_y - 2 * mm, left + sum(w for _, w, _ in cols) + 4 * mm, header_y - 2 * mm)
y = header_y - row_h
for row in page_rows:
if y < 65 * mm and page_idx == 1:
break
if y < 25 * mm:
break
x = left + 2 * mm
style_size = 8 if row.get("is_header") else 7
c.setFont(font, style_size)
for field, width, _ in cols:
text = "" if row.get("is_empty") else str(row.get(field, "") or "")
# truncate to fit roughly
max_chars = max(int(width / mm), 1)
c.drawString(x, y, text[: max_chars + 5])
x += width
y -= row_h
c.showPage()
c.save()
return buf.getvalue()
+268
View File
@@ -0,0 +1,268 @@
from __future__ import annotations
import shutil
import zipfile
from pathlib import Path
from sqlalchemy import delete, select
from sqlalchemy.orm import Session, selectinload
from app.core.config import get_settings
from app.models import (
Component,
ComponentProperty,
ComponentVariant,
DesignatorMapping,
DielMaterial,
PcbData,
Project,
ProjectParam,
Variant,
VariantProperty,
)
from app.services.altium_parser import AltiumParser, ProjectData, find_prjpcb, make_complex_string
from app.services.designators import DEFAULT_MAPPINGS
def project_dir(project_id: int) -> Path:
root = Path(get_settings().data_dir) / "projects" / str(project_id)
root.mkdir(parents=True, exist_ok=True)
return root
def clear_parsed_data(db: Session, project: Project, keep_tables: bool = True) -> None:
"""Remove components/variants/params/pcb; optionally keep edited tables."""
comps = db.scalars(select(Component).where(Component.project_id == project.id)).all()
comp_ids = [c.id for c in comps]
if comp_ids:
db.execute(delete(VariantProperty).where(VariantProperty.component_id.in_(comp_ids)))
db.execute(delete(ComponentVariant).where(ComponentVariant.component_id.in_(comp_ids)))
db.execute(delete(ComponentProperty).where(ComponentProperty.component_id.in_(comp_ids)))
db.execute(delete(Component).where(Component.id.in_(comp_ids)))
db.execute(delete(Variant).where(Variant.project_id == project.id))
db.execute(delete(ProjectParam).where(ProjectParam.project_id == project.id))
pcb = db.scalar(select(PcbData).where(PcbData.project_id == project.id))
if pcb:
db.execute(delete(DielMaterial).where(DielMaterial.pcb_data_id == pcb.id))
db.delete(pcb)
if not keep_tables:
from app.models import (
PerechenRow,
SimpleListRow,
SpecificationPcbRow,
SpecificationRow,
VedomostRow,
)
for model in (PerechenRow, SpecificationPcbRow, SpecificationRow, VedomostRow, SimpleListRow):
db.execute(delete(model).where(model.project_id == project.id))
db.flush()
def ensure_default_mappings(db: Session, project: Project) -> None:
existing = db.scalars(
select(DesignatorMapping).where(DesignatorMapping.project_id == project.id)
).all()
if existing:
return
for prefix, (sing, plur) in DEFAULT_MAPPINGS.items():
db.add(
DesignatorMapping(
project_id=project.id,
prefix=prefix,
singular_name=sing,
plural_name=plur,
)
)
def save_project_data(db: Session, project: Project, data: ProjectData) -> None:
variant_objs: list[Variant] = []
for name in data.variant_names:
v = Variant(project_id=project.id, name=name)
db.add(v)
variant_objs.append(v)
db.flush()
# Flat props for expression resolution of project params
flat_props = []
for comp_props in data.components_list:
flat_props.extend(comp_props)
# Regular components
designator_to_component: dict[str, Component] = {}
for comp_props in data.components_list:
designator = ""
for p in comp_props:
if p.name.lower() == "designator":
designator = p.text
break
if not designator:
continue
if designator in designator_to_component:
continue
comp = Component(project_id=project.id, designator=designator)
db.add(comp)
db.flush()
for p in comp_props:
db.add(ComponentProperty(component_id=comp.id, key=p.name, value=p.text))
designator_to_component[designator] = comp
# link to all variants as fitted by default
for v in variant_objs:
db.add(ComponentVariant(component_id=comp.id, variant_id=v.id, is_fitted=True))
# Variant overrides (skip No Variations at index 0 of variant_names)
for i, variant_comps in enumerate(data.components_variant_list):
variant_index = i + 1
if variant_index >= len(variant_objs):
continue
variant = variant_objs[variant_index]
dnf_list = (
data.dnf_variant_designators_list[i]
if i < len(data.dnf_variant_designators_list)
else []
)
for comp_props in variant_comps:
designator = ""
for p in comp_props:
if p.name == "Designator":
designator = p.text
break
if not designator:
continue
comp = designator_to_component.get(designator)
if not comp:
continue
is_dnf = designator in dnf_list
link = db.scalar(
select(ComponentVariant).where(
ComponentVariant.component_id == comp.id,
ComponentVariant.variant_id == variant.id,
)
)
if link:
link.is_fitted = not is_dnf
for p in comp_props:
if p.name in ("Designator", "Kind"):
continue
db.add(
VariantProperty(
component_id=comp.id,
variant_id=variant.id,
key=p.name,
value=p.text,
)
)
# Project params
for variant_idx, params in enumerate(data.prj_params_variant_list):
for item in params:
if len(item) < 2:
continue
name, value = item[0], item[1]
value = make_complex_string(value, flat_props)
db.add(
ProjectParam(
project_id=project.id,
name=name,
value=value,
variant_name=data.variant_names[variant_idx]
if variant_idx < len(data.variant_names)
else "",
)
)
# PCB
if data.pcb_layer_count or data.pcb_diel_materials:
pcb = PcbData(project_id=project.id, layer_count=data.pcb_layer_count)
db.add(pcb)
db.flush()
for m in data.pcb_diel_materials:
db.add(
DielMaterial(
pcb_data_id=pcb.id,
name=m.name,
value=m.value,
height=m.height,
diel_type=m.diel_type,
layer_number=m.layer_number,
)
)
project.pcb_doc_name = Path(data.pcb_doc_file_name).name if data.pcb_doc_file_name else None
if "No Variations" in data.variant_names:
project.current_variant = "No Variations"
elif data.variant_names:
project.current_variant = data.variant_names[0]
def ingest_zip(
db: Session,
project: Project,
zip_bytes: bytes,
filename: str = "project.zip",
keep_tables: bool = True,
) -> Project:
settings = get_settings()
Path(settings.data_dir).mkdir(parents=True, exist_ok=True)
pdir = project_dir(project.id)
zip_path = pdir / filename
extract_path = pdir / "extract"
if extract_path.exists():
shutil.rmtree(extract_path)
extract_path.mkdir(parents=True, exist_ok=True)
zip_path.write_bytes(zip_bytes)
try:
with zipfile.ZipFile(zip_path, "r") as zf:
zf.extractall(extract_path)
except zipfile.BadZipFile as e:
project.status = "error"
project.error_message = f"Invalid zip: {e}"
db.commit()
raise ValueError(project.error_message) from e
prj = find_prjpcb(extract_path)
if not prj:
project.status = "error"
project.error_message = "No .PrjPcb found in archive"
db.commit()
raise ValueError(project.error_message)
project.status = "parsing"
project.zip_path = str(zip_path)
project.extract_path = str(extract_path)
project.error_message = None
db.commit()
try:
clear_parsed_data(db, project, keep_tables=keep_tables)
parser = AltiumParser()
data = parser.parse_prjpcb(prj)
save_project_data(db, project, data)
ensure_default_mappings(db, project)
project.status = "ready"
db.commit()
db.refresh(project)
return project
except Exception as e:
project.status = "error"
project.error_message = str(e)
db.commit()
raise
def get_project_full(db: Session, project_id: int) -> Project | None:
return db.scalar(
select(Project)
.where(Project.id == project_id)
.options(
selectinload(Project.variants),
selectinload(Project.components).selectinload(Component.properties),
selectinload(Project.project_params),
selectinload(Project.pcb_data).selectinload(PcbData.diel_materials),
selectinload(Project.inscriptions),
selectinload(Project.designator_mappings),
)
)
@@ -0,0 +1,619 @@
"""GOST table generators (ported from desktop table controllers)."""
from __future__ import annotations
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Any, Optional
from app.services.designators import (
designator_sort_key,
format_consecutive_ranges,
format_designator_range,
get_type_names,
resolve_expression,
split_designator,
split_long_text,
)
@dataclass
class ComponentView:
designator: str
properties: dict[str, str] = field(default_factory=dict)
is_fitted: bool = True
def get(self, key: str, default: str = "") -> str:
raw = self.properties.get(key, default)
return resolve_expression(raw, self.properties) if raw else default
@dataclass
class MaterialView:
name: str
value: str
height: float
diel_type: int
layer_number: int
def _empty_flags(**extra: Any) -> dict[str, Any]:
base = {
"is_header": False,
"is_empty": False,
"is_auto_generated": True,
"stretch": False,
"is_underline": False,
"page_number": 1,
}
base.update(extra)
return base
def _paginate(rows: list[dict[str, Any]], first: int, other: int) -> list[dict[str, Any]]:
page = 1
used = 0
limit = first
for i, row in enumerate(rows):
if used >= limit:
page += 1
used = 0
limit = other
row["row_index"] = i
row["page_number"] = page
used += 1
return rows
def _group_by_letter(components: list[ComponentView]) -> OrderedDict[str, list[ComponentView]]:
groups: OrderedDict[str, list[ComponentView]] = OrderedDict()
for comp in sorted(components, key=lambda c: designator_sort_key(c.designator)):
letter, _ = split_designator(comp.designator)
letter = letter.upper()
groups.setdefault(letter, []).append(comp)
return groups
def generate_perechen(
components: list[ComponentView],
name_field: str = "Name",
mappings: dict[str, tuple[str, str]] | None = None,
) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
fitted = [c for c in components if c.is_fitted]
groups = _group_by_letter(fitted)
for letter, comps in groups.items():
singular, plural = get_type_names(comps[0].designator, mappings)
if len(comps) > 1:
rows.append(_empty_flags(is_empty=True, position="", designation="", quantity="", note=""))
rows.append(
_empty_flags(
is_header=True,
position="",
designation=plural,
quantity="",
note="",
is_underline=True,
)
)
rows.append(_empty_flags(is_empty=True, position="", designation="", quantity="", note=""))
# subgroup by name value, preserving order
subgroups: OrderedDict[str, list[ComponentView]] = OrderedDict()
for c in comps:
key = c.get(name_field, "")
subgroups.setdefault(key, []).append(c)
for value, group in subgroups.items():
desigs = [c.designator for c in group]
desig_text = format_designator_range(desigs)
desig_parts = split_long_text(desig_text, 60)
name_parts = split_long_text(value, 60)
max_rows = max(len(desig_parts), len(name_parts), 1)
for i in range(max_rows):
rows.append(
_empty_flags(
position=desig_parts[i] if i < len(desig_parts) else "",
designation=name_parts[i] if i < len(name_parts) else "",
quantity=str(len(group)) if i == 0 else "",
note="",
)
)
else:
c = comps[0]
value = c.get(name_field, "")
designation = f"{singular} {value}".strip()
rows.append(_empty_flags(is_empty=True, position="", designation="", quantity="", note=""))
rows.append(
_empty_flags(
position=c.designator,
designation=designation,
quantity="1",
note="",
)
)
return _paginate(rows, 25, 32)
def generate_specification_pcb(
components: list[ComponentView],
decimal_number: str = "",
board_name: str = "",
pcb_doc_name: str = "",
mappings: dict[str, tuple[str, str]] | None = None,
) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
dec = decimal_number or ""
def add_empty():
rows.append(
_empty_flags(
format="", zone="", position="", designation="", name="", quantity="", note="", is_empty=True
)
)
def add_header(name: str):
rows.append(
_empty_flags(
format="",
zone="",
position="",
designation="",
name=name,
quantity="",
note="",
is_header=True,
is_underline=True,
)
)
def add_doc(fmt: str, designation: str, name: str, note: str = "", pos: str = ""):
rows.append(
_empty_flags(
format=fmt,
zone="",
position=pos,
designation=designation,
name=name,
quantity="",
note=note,
)
)
add_empty()
add_header("Документация")
add_empty()
add_doc("A1", f"{dec} СБ", "Сборочный чертеж")
add_doc("A3", f"{dec} Э3", "Схема электрическая принципиальная")
add_doc("A4", f"{dec} ПЭ3", "Перечень элементов")
add_doc("A3", f"{dec} ВП", "Ведомость покупных изделий")
add_doc("*)", f"{dec} Д33", "Данные результатов проектирования", "DVD диск")
add_empty()
add_doc("А4", f"{dec} Д10-УЛ", "Удостоверяющий лист", "Размножать")
add_doc("", "", "Данные результатов проектирования", "по указанию")
add_empty()
add_header("Сборочные единицы")
add_empty()
rows.append(
_empty_flags(
format="",
zone="",
position="1",
designation=f"{dec} платы" if dec else "платы",
name=board_name,
quantity="",
note="",
)
)
add_empty()
add_header("Детали")
add_empty()
add_header("Стандартные изделия")
add_empty()
add_header("Прочие изделия")
add_empty()
pos = 2
fitted = [c for c in components if c.is_fitted]
groups = _group_by_letter(fitted)
for letter, comps in groups.items():
subgroups: OrderedDict[str, list[ComponentView]] = OrderedDict()
for c in comps:
key = c.get("ManufacturerPartNumber", "") or c.get("Name", "")
subgroups.setdefault(key, []).append(c)
for part_number, group in subgroups.items():
singular, plural = get_type_names(group[0].designator, mappings)
qty = len(group)
type_name = plural if qty > 1 else singular
desigs = format_consecutive_ranges([c.designator for c in group])
note_parts = split_long_text(desigs, 11)
name_full = f"{type_name} {part_number}".strip()
name_parts = split_long_text(name_full, 34)
# if type + first part too long, put type on its own line
if name_parts and len(f"{type_name} {name_parts[0]}") > 34 and part_number:
name_parts = [type_name] + split_long_text(part_number, 34)
max_rows = max(len(name_parts), len(note_parts), 1)
for i in range(max_rows):
rows.append(
_empty_flags(
format="",
zone="",
position=str(pos) if i == 0 else "",
designation="",
name=name_parts[i] if i < len(name_parts) else "",
quantity=str(qty) if i == 0 else "",
note=note_parts[i] if i < len(note_parts) else "",
)
)
pos += 1
add_empty()
add_header("Примечание")
rows.append(
_empty_flags(
format="", zone="", position="", designation="", name="Изготовить плату печатную", quantity="", note=""
)
)
pcb_short = Path_name(pcb_doc_name)
rows.append(
_empty_flags(
format="",
zone="",
position="",
designation="",
name=f"По файлу {pcb_short}" if pcb_short else "По файлу",
quantity="",
note="",
)
)
rows.append(
_empty_flags(
format="",
zone="",
position="",
designation="",
name=f"из состава {dec} Д10" if dec else "из состава Д10",
quantity="",
note="",
)
)
return _paginate(rows, 26, 32)
def Path_name(path: str) -> str:
if not path:
return ""
return path.replace("\\", "/").split("/")[-1]
def generate_specification(
materials: list[MaterialView],
layer_count: int = 0,
decimal_number: str = "",
) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
dec = decimal_number or ""
def add_empty():
rows.append(
_empty_flags(
format="", zone="", position="", designation="", name="", quantity="", note="", is_empty=True
)
)
def add_header(name: str):
rows.append(
_empty_flags(
format="",
zone="",
position="",
designation="",
name=name,
quantity="",
note="",
is_header=True,
is_underline=True,
)
)
add_empty()
add_header("Документация")
add_empty()
rows.append(
_empty_flags(
format="A1",
zone="",
position="",
designation=f"{dec}Э3 СБ",
name="Сборочный чертеж",
quantity="",
note="",
)
)
rows.append(
_empty_flags(
format="*)",
zone="",
position="",
designation=f"{dec} Э3 Т5М",
name="Данные проектирования",
quantity="",
note="",
)
)
add_empty()
add_header("Материалы")
add_empty()
sorted_mats = sorted(materials, key=lambda m: m.layer_number)
by_name: OrderedDict[str, list[MaterialView]] = OrderedDict()
for m in sorted_mats:
by_name.setdefault(m.name, []).append(m)
core_count = sum(1 for m in materials if m.diel_type == 1)
foil_count = max(layer_count - core_count * 2, 0)
pos = 1
for name, group in by_name.items():
if name == "DielMaterial":
by_value: OrderedDict[str, list[MaterialView]] = OrderedDict()
for m in group:
by_value.setdefault(m.value, []).append(m)
for value, subgroup in by_value.items():
display = {
"FR4 PR": "Препрег FR4 PR",
"FR4 Tg150": "Стеклотекстолит FR4 Tg150",
"Solder Resist": "Паяльная маска",
}.get(value, value)
thickness = subgroup[0].height
display = f"{display} {thickness:.3f} мм"
rows.append(
_empty_flags(
format="",
zone="",
position=str(pos),
designation="",
name=display,
quantity=str(len(subgroup)),
note="",
)
)
pos += 1
else:
rows.append(
_empty_flags(
format="",
zone="",
position=str(pos),
designation="",
name=name,
quantity=str(len(group)),
note="",
)
)
pos += 1
if foil_count > 0:
rows.append(
_empty_flags(
format="",
zone="",
position=str(pos),
designation="",
name="Фольга медная толщиной 18 мкм",
quantity=str(foil_count),
note="",
)
)
return _paginate(rows, 27, 33)
def generate_vedomost(
components: list[ComponentView],
where_used: str = "",
mappings: dict[str, tuple[str, str]] | None = None,
column_mappings: Optional[dict[str, str]] = None,
) -> list[dict[str, Any]]:
cmap = column_mappings or {
"Name": "Name",
"ProductCode": "ProductCode",
"DocumentCode": "DocumentCode",
"Supplier": "Supplier",
"Note": "Note",
}
rows: list[dict[str, Any]] = []
fitted = [c for c in components if c.is_fitted]
groups = _group_by_letter(fitted)
for letter, comps in groups.items():
subgroups: OrderedDict[str, list[ComponentView]] = OrderedDict()
for c in comps:
key = c.get(cmap.get("Name", "Name"), "")
subgroups.setdefault(key, []).append(c)
multiple_names = len(subgroups) > 1
if len(comps) > 1 and multiple_names:
_, plural = get_type_names(comps[0].designator, mappings)
rows.append(
_empty_flags(
is_empty=True,
name="",
product_code="",
document_code="",
supplier="",
where_used="",
quantity_per_item="",
quantity_in_set="",
quantity_for_reg="",
total_quantity="",
note="",
)
)
rows.append(
_empty_flags(
is_header=True,
is_underline=True,
name=plural,
product_code="",
document_code="",
supplier="",
where_used="",
quantity_per_item="",
quantity_in_set="",
quantity_for_reg="",
total_quantity="",
note="",
)
)
rows.append(
_empty_flags(
is_empty=True,
name="",
product_code="",
document_code="",
supplier="",
where_used="",
quantity_per_item="",
quantity_in_set="",
quantity_for_reg="",
total_quantity="",
note="",
)
)
for value, group in subgroups.items():
singular, plural = get_type_names(group[0].designator, mappings)
qty = len(group)
type_name = ""
if len(comps) == 1 or (len(comps) > 1 and not multiple_names):
type_name = plural if qty > 1 else singular
name = f"{type_name} {value}".strip() if type_name else value
product = group[0].get(cmap.get("ProductCode", "ProductCode"), "")
document = group[0].get(cmap.get("DocumentCode", "DocumentCode"), "")
supplier = group[0].get(cmap.get("Supplier", "Supplier"), "")
note = group[0].get(cmap.get("Note", "Note"), "")
name_parts = split_long_text(name, 32)
product_parts = split_long_text(product, 25)
document_parts = split_long_text(document, 35)
supplier_parts = split_long_text(supplier, 25)
where_parts = split_long_text(where_used, 35)
note_parts = split_long_text(note, 16)
max_rows = max(
len(name_parts),
len(product_parts),
len(document_parts),
len(supplier_parts),
len(where_parts),
len(note_parts),
1,
)
if len(comps) == 1 or (len(comps) > 1 and not multiple_names):
rows.append(
_empty_flags(
is_empty=True,
name="",
product_code="",
document_code="",
supplier="",
where_used="",
quantity_per_item="",
quantity_in_set="",
quantity_for_reg="",
total_quantity="",
note="",
)
)
for i in range(max_rows):
rows.append(
_empty_flags(
name=name_parts[i] if i < len(name_parts) else "",
product_code=product_parts[i] if i < len(product_parts) else "",
document_code=document_parts[i] if i < len(document_parts) else "",
supplier=supplier_parts[i] if i < len(supplier_parts) else "",
where_used=where_parts[i] if i < len(where_parts) else "",
quantity_per_item=str(qty) if i == 0 else "",
quantity_in_set="",
quantity_for_reg="",
total_quantity=str(qty) if i == 0 else "",
note=note_parts[i] if i < len(note_parts) else "",
)
)
if len(comps) == 1 or (len(comps) > 1 and not multiple_names):
rows.append(
_empty_flags(
is_empty=True,
name="",
product_code="",
document_code="",
supplier="",
where_used="",
quantity_per_item="",
quantity_in_set="",
quantity_for_reg="",
total_quantity="",
note="",
)
)
return _paginate(rows, 24, 29)
def generate_simple_list(
components: list[ComponentView],
name_field: str = "Name",
tech_reserve_percent: float = 10.0,
boards_count: int = 1,
) -> list[dict[str, Any]]:
rows: list[dict[str, Any]] = []
fitted = sorted(
[c for c in components if c.is_fitted],
key=lambda c: designator_sort_key(c.designator),
)
# group consecutive same name
groups: list[list[ComponentView]] = []
current: list[ComponentView] = []
current_name = None
for c in fitted:
name = c.get(name_field, "")
if current and name != current_name:
groups.append(current)
current = []
current.append(c)
current_name = name
if current:
groups.append(current)
for group in groups:
base = len(group)
reserve = max(1, round(base * boards_count * tech_reserve_percent / 100))
qty = boards_count * base + reserve
desigs = format_designator_range([c.designator for c in group])
rows.append(
{
"row_index": 0,
"designator": desigs,
"name": group[0].get(name_field, ""),
"quantity": str(qty),
"is_auto_generated": True,
"is_empty": False,
}
)
for i, row in enumerate(rows):
row["row_index"] = i
return rows
TABLE_TYPES = ("perechen", "specification_pcb", "specification", "vedomost", "simple_list")
+282
View File
@@ -0,0 +1,282 @@
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)