patrickandClaude Sonnet 5
7a3fd7acda
Wichtigster Fund: mein eigener CI-Check hat nur den zuletzt aktualisierten Task angeschaut statt beide Jobs (backend-tests/frontend-build) einzeln zu prüfen - dadurch ist die tatsächlich rote backend-tests-Pipeline seit mehreren Commits unbemerkt geblieben (auch schon deployed). Echter Bug (von mir eingeführt in SEARCH-002): /suche filterte auf Objektposition.status, das Feld heißt aber ist_status - AttributeError bei jeder Materialsuche. 5 veraltete Testerwartungen korrigiert (Logik in dashboard.py ist seit längerem korrekt und dokumentiert, Tests wurden nie nachgezogen): "noch_nie_kontrolliert" gilt laut _kategorie_und_gruende() IMMER zusätzlich zu jedem anderen Grund - test_dashboard.py/test_mangel.py/ test_fahrzeugdetails.py prüften teils das Gegenteil. 1 Typo-Bug in test_kontrolle.py (UUID != str-Vergleich, immer falsch). Offen/nicht geklärt: test_stammdaten.py::test_fach_verwendung_und_ersetzen_ fuehrt_zusammen (KeyError 'id' bei Fach-Merge-Response) - ohne lokale Testausführung nicht sicher diagnostizierbar, braucht echten pytest-Lauf. Nebenbei: FOUND-009/FOUND-011 im Backlog korrigiert (CI existiert bereits unter .gitea/, vorheriger Abgleich hatte nur .github/ geprüft). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
78 lines
3.1 KiB
Python
78 lines
3.1 KiB
Python
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.api.deps import get_current_user
|
|
from app.db.session import get_db
|
|
from app.models.objekt import Objekt
|
|
from app.models.objektposition import Objektposition, ObjektpositionStatus
|
|
from app.models.stammdaten import Material
|
|
from app.schemas.suche import SuchTreffer
|
|
|
|
router = APIRouter()
|
|
|
|
_TREFFER_LIMIT = 8
|
|
|
|
|
|
@router.get("/suche", response_model=list[SuchTreffer])
|
|
async def suche(
|
|
q: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
_=Depends(get_current_user),
|
|
) -> list[SuchTreffer]:
|
|
"""Globale Suche (Command Palette) über Objektname/-code UND Material-
|
|
Namen (SEARCH-002: wer "Kompressen" sucht, will Objekte finden, die
|
|
welche führen) - per pg_trgm-Index statt vollständig geladener
|
|
Objektliste + Client-Filter (SEARCH-001). Kurze Eingaben (<2 Zeichen)
|
|
liefern bewusst nichts - sonst zu viele Zufallstreffer über den
|
|
Ähnlichkeits-Schwellenwert von pg_trgm."""
|
|
q = q.strip()
|
|
if len(q) < 2:
|
|
return []
|
|
|
|
muster = f"%{q}%"
|
|
|
|
aehnlichkeit_objekt = func.greatest(func.similarity(Objekt.name, q), func.similarity(Objekt.code, q))
|
|
direkt_stmt = (
|
|
select(Objekt.id, Objekt.code, Objekt.name, aehnlichkeit_objekt.label("score"))
|
|
.where(or_(Objekt.name.ilike(muster), Objekt.code.ilike(muster), aehnlichkeit_objekt > 0.2))
|
|
.order_by(aehnlichkeit_objekt.desc())
|
|
.limit(_TREFFER_LIMIT)
|
|
)
|
|
direkt_ergebnis = (await db.execute(direkt_stmt)).all()
|
|
|
|
aehnlichkeit_material = func.similarity(Material.name, q)
|
|
material_stmt = (
|
|
select(Objekt.id, Objekt.code, Objekt.name, Material.name.label("material_name"), aehnlichkeit_material.label("score"))
|
|
.join(Objektposition, Objektposition.objekt_id == Objekt.id)
|
|
.join(Material, Material.id == Objektposition.material_id)
|
|
.where(
|
|
Objektposition.ist_status == ObjektpositionStatus.aktiv,
|
|
or_(Material.name.ilike(muster), aehnlichkeit_material > 0.2),
|
|
)
|
|
.order_by(aehnlichkeit_material.desc())
|
|
.limit(_TREFFER_LIMIT)
|
|
)
|
|
material_ergebnis = (await db.execute(material_stmt)).all()
|
|
|
|
# Zusammenführen: direkter Objekt-Treffer schlägt Material-Treffer für
|
|
# dasselbe Objekt (kein "über Material XY" anzeigen, wenn der Objektname
|
|
# selbst schon passt), sonst höherer Score gewinnt bei Mehrfachtreffer.
|
|
kandidaten: dict[int, SuchTreffer] = {}
|
|
scores: dict[int, float] = {}
|
|
for row in direkt_ergebnis:
|
|
kandidaten[row.id] = SuchTreffer(id=row.id, code=row.code, name=row.name)
|
|
scores[row.id] = float(row.score or 0)
|
|
for row in material_ergebnis:
|
|
score = float(row.score or 0)
|
|
if row.id in kandidaten:
|
|
continue
|
|
if row.id not in scores or score > scores[row.id]:
|
|
kandidaten[row.id] = SuchTreffer(
|
|
id=row.id, code=row.code, name=row.name, ueber_material=row.material_name
|
|
)
|
|
scores[row.id] = score
|
|
|
|
treffer = sorted(kandidaten.values(), key=lambda t: scores[t.id], reverse=True)
|
|
return treffer[:_TREFFER_LIMIT]
|