247 lines
11 KiB
Python
247 lines
11 KiB
Python
import hashlib
|
|
import mimetypes
|
|
import re
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from fastapi import HTTPException, UploadFile, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import settings
|
|
from app.models.knowledge import KnowledgeDevice, KnowledgeDocument, KnowledgeManufacturer, KnowledgeNote
|
|
from app.repositories.knowledge_repository import KnowledgeRepository
|
|
from app.schemas.knowledge import (
|
|
KnowledgeDeviceCreate,
|
|
KnowledgeDeviceUpdate,
|
|
KnowledgeDocumentCreate,
|
|
KnowledgeDocumentUpdate,
|
|
KnowledgeManufacturerCreate,
|
|
KnowledgeManufacturerUpdate,
|
|
KnowledgeNoteCreate,
|
|
KnowledgeNoteUpdate,
|
|
)
|
|
|
|
ALLOWED_EXTENSIONS = {".pdf", ".jpg", ".jpeg", ".png", ".webp", ".txt", ".zip"}
|
|
ALLOWED_MIME_TYPES = {
|
|
"application/pdf",
|
|
"image/jpeg",
|
|
"image/png",
|
|
"image/webp",
|
|
"text/plain",
|
|
"application/zip",
|
|
"application/x-zip-compressed",
|
|
}
|
|
|
|
|
|
def slugify(value: str) -> str:
|
|
normalized = value.strip().lower()
|
|
normalized = normalized.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("ß", "ss")
|
|
normalized = re.sub(r"[^a-z0-9]+", "-", normalized)
|
|
normalized = normalized.strip("-")
|
|
return normalized or uuid.uuid4().hex[:10]
|
|
|
|
|
|
def unique_slug(db: Session, base: str, exists) -> str:
|
|
slug = slugify(base)
|
|
candidate = slug
|
|
index = 2
|
|
while exists(db, candidate):
|
|
candidate = f"{slug}-{index}"
|
|
index += 1
|
|
return candidate
|
|
|
|
|
|
def storage_root() -> Path:
|
|
root = Path(settings.knowledge_storage_path).resolve()
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
return root
|
|
|
|
|
|
def assert_safe_path(path: Path) -> Path:
|
|
root = storage_root()
|
|
resolved = path.resolve()
|
|
if root != resolved and root not in resolved.parents:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Ungültiger Dateipfad")
|
|
return resolved
|
|
|
|
|
|
def safe_file_name(file_name: str) -> str:
|
|
name = Path(file_name).name.strip()
|
|
stem = slugify(Path(name).stem)
|
|
suffix = Path(name).suffix.lower()
|
|
return f"{stem}{suffix}" if suffix else stem
|
|
|
|
|
|
def parse_tags(value: str) -> list[str]:
|
|
seen: set[str] = set()
|
|
tags: list[str] = []
|
|
for item in value.split(","):
|
|
tag = item.strip()
|
|
key = tag.lower()
|
|
if tag and key not in seen:
|
|
seen.add(key)
|
|
tags.append(tag)
|
|
return tags
|
|
|
|
|
|
async def read_upload(file: UploadFile) -> tuple[bytes, str, str]:
|
|
original_name = file.filename or ""
|
|
file_name = safe_file_name(original_name)
|
|
extension = Path(file_name).suffix.lower()
|
|
if extension not in ALLOWED_EXTENSIONS:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Dateityp ist nicht erlaubt")
|
|
|
|
max_bytes = max(1, settings.knowledge_max_upload_mb) * 1024 * 1024
|
|
content = await file.read(max_bytes + 1)
|
|
if not content:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Upload-Datei ist leer")
|
|
if len(content) > max_bytes:
|
|
raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="Upload-Datei ist zu groß")
|
|
|
|
mime_type = file.content_type or mimetypes.guess_type(file_name)[0] or "application/octet-stream"
|
|
if mime_type not in ALLOWED_MIME_TYPES:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="MIME-Type ist nicht erlaubt")
|
|
|
|
return content, file_name, mime_type
|
|
|
|
|
|
class KnowledgeService:
|
|
@staticmethod
|
|
def create_manufacturer(db: Session, payload: KnowledgeManufacturerCreate) -> KnowledgeManufacturer:
|
|
if KnowledgeRepository.get_manufacturer_by_name(db, payload.name):
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Herstellername ist bereits vergeben")
|
|
slug = unique_slug(db, payload.name, lambda session, value: KnowledgeRepository.get_manufacturer_by_slug(session, value) is not None)
|
|
manufacturer = KnowledgeManufacturer(name=payload.name, slug=slug, website=str(payload.website or ""), notes=payload.notes)
|
|
db.add(manufacturer)
|
|
db.commit()
|
|
db.refresh(manufacturer)
|
|
return manufacturer
|
|
|
|
@staticmethod
|
|
def update_manufacturer(db: Session, manufacturer: KnowledgeManufacturer, payload: KnowledgeManufacturerUpdate) -> KnowledgeManufacturer:
|
|
if KnowledgeRepository.get_manufacturer_by_name(db, payload.name, manufacturer.id):
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Herstellername ist bereits vergeben")
|
|
manufacturer.name = payload.name
|
|
manufacturer.slug = unique_slug(
|
|
db,
|
|
payload.name,
|
|
lambda session, value: KnowledgeRepository.get_manufacturer_by_slug(session, value, manufacturer.id) is not None,
|
|
)
|
|
manufacturer.website = str(payload.website or "")
|
|
manufacturer.notes = payload.notes
|
|
db.commit()
|
|
db.refresh(manufacturer)
|
|
return manufacturer
|
|
|
|
@staticmethod
|
|
def create_device(db: Session, payload: KnowledgeDeviceCreate) -> KnowledgeDevice:
|
|
if KnowledgeRepository.get_manufacturer(db, payload.manufacturer_id) is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Hersteller nicht gefunden")
|
|
slug = slugify(payload.model_number or payload.name)
|
|
if KnowledgeRepository.get_device_by_slug(db, manufacturer_id=payload.manufacturer_id, slug=slug):
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Geräte-Slug ist für diesen Hersteller bereits vergeben")
|
|
device = KnowledgeDevice(slug=slug, **payload.model_dump())
|
|
db.add(device)
|
|
db.commit()
|
|
return KnowledgeRepository.get_device(db, device.id) or device
|
|
|
|
@staticmethod
|
|
def update_device(db: Session, device: KnowledgeDevice, payload: KnowledgeDeviceUpdate) -> KnowledgeDevice:
|
|
if KnowledgeRepository.get_manufacturer(db, payload.manufacturer_id) is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Hersteller nicht gefunden")
|
|
slug = slugify(payload.model_number or payload.name)
|
|
if KnowledgeRepository.get_device_by_slug(db, manufacturer_id=payload.manufacturer_id, slug=slug, exclude_id=device.id):
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Geräte-Slug ist für diesen Hersteller bereits vergeben")
|
|
for key, value in payload.model_dump().items():
|
|
setattr(device, key, value)
|
|
device.slug = slug
|
|
db.commit()
|
|
return KnowledgeRepository.get_device(db, device.id) or device
|
|
|
|
@staticmethod
|
|
def create_document(db: Session, payload: KnowledgeDocumentCreate) -> KnowledgeDocument:
|
|
KnowledgeService._validate_document_links(db, payload.manufacturer_id, payload.device_id)
|
|
slug = unique_slug(db, payload.title, lambda session, value: KnowledgeRepository.get_document_by_slug(session, value) is not None)
|
|
document = KnowledgeDocument(slug=slug, external_url=str(payload.external_url or ""), **payload.model_dump(exclude={"external_url"}))
|
|
db.add(document)
|
|
db.commit()
|
|
return KnowledgeRepository.get_document(db, document.id) or document
|
|
|
|
@staticmethod
|
|
async def upload_document(
|
|
db: Session,
|
|
*,
|
|
file: UploadFile,
|
|
payload: KnowledgeDocumentCreate,
|
|
) -> KnowledgeDocument:
|
|
KnowledgeService._validate_document_links(db, payload.manufacturer_id, payload.device_id)
|
|
content, file_name, mime_type = await read_upload(file)
|
|
checksum = hashlib.sha256(content).hexdigest()
|
|
slug = unique_slug(db, payload.title, lambda session, value: KnowledgeRepository.get_document_by_slug(session, value) is not None)
|
|
target_dir = storage_root() / str(payload.manufacturer_id)
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
stored_name = f"{uuid.uuid4().hex}-{file_name}"
|
|
file_path = assert_safe_path(target_dir / stored_name)
|
|
file_path.write_bytes(content)
|
|
document = KnowledgeDocument(
|
|
slug=slug,
|
|
external_url=str(payload.external_url or ""),
|
|
file_name=file_name,
|
|
file_path=str(file_path),
|
|
mime_type=mime_type,
|
|
file_size=len(content),
|
|
checksum_sha256=checksum,
|
|
**payload.model_dump(exclude={"external_url"}),
|
|
)
|
|
db.add(document)
|
|
db.commit()
|
|
return KnowledgeRepository.get_document(db, document.id) or document
|
|
|
|
@staticmethod
|
|
def update_document(db: Session, document: KnowledgeDocument, payload: KnowledgeDocumentUpdate) -> KnowledgeDocument:
|
|
KnowledgeService._validate_document_links(db, payload.manufacturer_id, payload.device_id)
|
|
slug = unique_slug(db, payload.title, lambda session, value: KnowledgeRepository.get_document_by_slug(session, value, document.id) is not None)
|
|
for key, value in payload.model_dump(exclude={"external_url"}).items():
|
|
setattr(document, key, value)
|
|
document.external_url = str(payload.external_url or "")
|
|
document.slug = slug
|
|
db.commit()
|
|
return KnowledgeRepository.get_document(db, document.id) or document
|
|
|
|
@staticmethod
|
|
def create_note(db: Session, payload: KnowledgeNoteCreate) -> KnowledgeNote:
|
|
KnowledgeService._validate_optional_links(db, payload.manufacturer_id, payload.device_id)
|
|
note = KnowledgeNote(**payload.model_dump())
|
|
db.add(note)
|
|
db.commit()
|
|
return KnowledgeRepository.get_note(db, note.id) or note
|
|
|
|
@staticmethod
|
|
def update_note(db: Session, note: KnowledgeNote, payload: KnowledgeNoteUpdate) -> KnowledgeNote:
|
|
KnowledgeService._validate_optional_links(db, payload.manufacturer_id, payload.device_id)
|
|
for key, value in payload.model_dump().items():
|
|
setattr(note, key, value)
|
|
db.commit()
|
|
return KnowledgeRepository.get_note(db, note.id) or note
|
|
|
|
@staticmethod
|
|
def _validate_document_links(db: Session, manufacturer_id: int, device_id: int | None) -> None:
|
|
if KnowledgeRepository.get_manufacturer(db, manufacturer_id) is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Hersteller nicht gefunden")
|
|
if device_id is not None:
|
|
device = KnowledgeRepository.get_device(db, device_id)
|
|
if device is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Gerät nicht gefunden")
|
|
if device.manufacturer_id != manufacturer_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Gerät gehört nicht zum Hersteller")
|
|
|
|
@staticmethod
|
|
def _validate_optional_links(db: Session, manufacturer_id: int | None, device_id: int | None) -> None:
|
|
if manufacturer_id is not None and KnowledgeRepository.get_manufacturer(db, manufacturer_id) is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Hersteller nicht gefunden")
|
|
if device_id is not None:
|
|
device = KnowledgeRepository.get_device(db, device_id)
|
|
if device is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Gerät nicht gefunden")
|
|
if manufacturer_id is not None and device.manufacturer_id != manufacturer_id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Gerät gehört nicht zum Hersteller")
|