patrickandClaude Sonnet 5
69e7d0c8d0
objektposition.code (nullable, unique) für Geräte-Etiketten, separat von
seriennummer. Lookup-Endpunkt für Geräte-Scan (GET /objektpositionen/code/{code})
sowie Auto-Vorschlag PRAEFIX-NNN für Objekt- und Objektposition-Codes
(GET /objekte/naechster-code, GET /objektpositionen/naechster-code).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
124 lines
4.3 KiB
Python
124 lines
4.3 KiB
Python
import re
|
||
from decimal import Decimal
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models.objekt import Objekt
|
||
from app.models.objektposition import Objektposition, ObjektpositionStatus
|
||
from app.models.vorlage import Beladungsvorlage, Vorlagenposition
|
||
|
||
_CODE_MUSTER = re.compile(r"^-(\d+)$")
|
||
|
||
|
||
async def naechster_freier_code(db: AsyncSession, spalte, praefix: str) -> str:
|
||
"""Konzept-Ergänzung Karte 10/Prompt 09: Code-Vorschlag PRAEFIX-NNN mit
|
||
fortlaufender Nummer je Präfix. Zahlenbreite folgt der bisher größten
|
||
vergebenen Nummer (mind. 5 Stellen bei Erstvergabe), damit die Sortierung im
|
||
Regal stabil bleibt, solange nicht mehr als 10^Breite Objekte existieren."""
|
||
result = await db.execute(select(spalte).where(spalte.like(f"{praefix}-%")))
|
||
hoechste = 0
|
||
breite = 5
|
||
for (code,) in result.all():
|
||
treffer = _CODE_MUSTER.match(code[len(praefix) :])
|
||
if treffer is None:
|
||
continue
|
||
zahl = treffer.group(1)
|
||
breite = max(breite, len(zahl))
|
||
hoechste = max(hoechste, int(zahl))
|
||
return f"{praefix}-{hoechste + 1:0{breite}d}"
|
||
|
||
|
||
async def sollmenge_map(db: AsyncSession, vorlage_id: int | None) -> dict[int, Decimal]:
|
||
"""material_id -> Sollmenge laut Vorlage (Prompt 10: Basis für die dynamische
|
||
Auflösung, wenn eine Objektposition keinen Override hat)."""
|
||
if vorlage_id is None:
|
||
return {}
|
||
result = await db.execute(
|
||
select(Vorlagenposition.material_id, Vorlagenposition.sollmenge).where(
|
||
Vorlagenposition.vorlage_id == vorlage_id
|
||
)
|
||
)
|
||
return {material_id: sollmenge for material_id, sollmenge in result.all()}
|
||
|
||
|
||
def sollmenge_effektiv(position: Objektposition, vorlagen_sollmengen: dict[int, Decimal]) -> Decimal:
|
||
if position.sollmenge_override is not None:
|
||
return position.sollmenge_override
|
||
return vorlagen_sollmengen.get(position.material_id, Decimal(0))
|
||
|
||
|
||
async def objekt_aus_vorlage(
|
||
db: AsyncSession,
|
||
*,
|
||
code: str,
|
||
name: str,
|
||
standort_id: int,
|
||
vorlage: Beladungsvorlage,
|
||
zustaendiger_server_id: int,
|
||
) -> Objekt:
|
||
"""Prompt 09 (Neuanlage, nicht Duplizieren): Objektpositionen werden aus den
|
||
Vorlagenpositionen initialisiert, Ist-Menge startet bei 0 ("nicht kontrolliert")."""
|
||
objekt = Objekt(
|
||
code=code,
|
||
name=name,
|
||
objekttyp_id=vorlage.objekttyp_id,
|
||
vorlage_id=vorlage.id,
|
||
standort_id=standort_id,
|
||
zustaendiger_server_id=zustaendiger_server_id,
|
||
)
|
||
db.add(objekt)
|
||
await db.flush()
|
||
|
||
result = await db.execute(select(Vorlagenposition).where(Vorlagenposition.vorlage_id == vorlage.id))
|
||
for vorlagenposition in result.scalars().all():
|
||
db.add(
|
||
Objektposition(
|
||
objekt_id=objekt.id,
|
||
material_id=vorlagenposition.material_id,
|
||
ist_status=ObjektpositionStatus.aktiv,
|
||
istmenge=Decimal(0),
|
||
)
|
||
)
|
||
await db.flush()
|
||
return objekt
|
||
|
||
|
||
async def dupliziere_objekt(
|
||
db: AsyncSession,
|
||
*,
|
||
quelle: Objekt,
|
||
neuer_code: str,
|
||
neuer_name: str,
|
||
neuer_standort_id: int,
|
||
zustaendiger_server_id: int,
|
||
) -> Objekt:
|
||
"""Prompt 09: Struktur/Sollmengen (inkl. individueller Overrides) werden kopiert,
|
||
Zustandsdaten (Ist-Menge, SN, Ablauf, Charge) NICHT – neues Objekt startet
|
||
"nicht kontrolliert". Fehlbestände/Historie existieren für ein neues Objekt
|
||
ohnehin nicht (kein Kopierschritt nötig)."""
|
||
neues_objekt = Objekt(
|
||
code=neuer_code,
|
||
name=neuer_name,
|
||
objekttyp_id=quelle.objekttyp_id,
|
||
vorlage_id=quelle.vorlage_id,
|
||
standort_id=neuer_standort_id,
|
||
zustaendiger_server_id=zustaendiger_server_id,
|
||
)
|
||
db.add(neues_objekt)
|
||
await db.flush()
|
||
|
||
result = await db.execute(select(Objektposition).where(Objektposition.objekt_id == quelle.id))
|
||
for quellposition in result.scalars().all():
|
||
db.add(
|
||
Objektposition(
|
||
objekt_id=neues_objekt.id,
|
||
material_id=quellposition.material_id,
|
||
sollmenge_override=quellposition.sollmenge_override,
|
||
ist_status=quellposition.ist_status,
|
||
istmenge=Decimal(0),
|
||
)
|
||
)
|
||
await db.flush()
|
||
return neues_objekt
|