3d7fb1aa48
- Created migration scripts for AnyDesk sessions and hardware assets. - Implemented apply_migration_115.py to execute migration for AnyDesk sessions. - Added set_customer_wiki_slugs.py script to update customer wiki slugs based on a predefined folder list. - Developed run_migration.py to apply AnyDesk migration schema. - Added tests for Service Contract Wizard to ensure functionality and dry-run mode.
589 lines
21 KiB
Python
589 lines
21 KiB
Python
import logging
|
|
from typing import List, Optional
|
|
from fastapi import APIRouter, HTTPException, Query, UploadFile, File
|
|
from app.core.database import execute_query
|
|
from datetime import datetime, date
|
|
import os
|
|
import uuid
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
# ============================================================================
|
|
# CRUD Endpoints for Hardware Assets
|
|
# ============================================================================
|
|
|
|
@router.get("/hardware", response_model=List[dict])
|
|
async def list_hardware(
|
|
customer_id: Optional[int] = None,
|
|
status: Optional[str] = None,
|
|
asset_type: Optional[str] = None,
|
|
q: Optional[str] = None
|
|
):
|
|
"""List all hardware with optional filters."""
|
|
query = "SELECT * FROM hardware_assets WHERE deleted_at IS NULL"
|
|
params = []
|
|
|
|
if customer_id:
|
|
query += " AND current_owner_customer_id = %s"
|
|
params.append(customer_id)
|
|
if status:
|
|
query += " AND status = %s"
|
|
params.append(status)
|
|
if asset_type:
|
|
query += " AND asset_type = %s"
|
|
params.append(asset_type)
|
|
if q:
|
|
query += " AND (serial_number ILIKE %s OR model ILIKE %s OR brand ILIKE %s)"
|
|
search_param = f"%{q}%"
|
|
params.extend([search_param, search_param, search_param])
|
|
|
|
query += " ORDER BY created_at DESC"
|
|
|
|
result = execute_query(query, tuple(params))
|
|
logger.info(f"✅ Listed {len(result) if result else 0} hardware assets")
|
|
return result or []
|
|
|
|
|
|
@router.get("/hardware/by-customer/{customer_id}", response_model=List[dict])
|
|
async def list_hardware_by_customer(customer_id: int):
|
|
"""List hardware assets owned by a customer."""
|
|
query = """
|
|
SELECT * FROM hardware_assets
|
|
WHERE deleted_at IS NULL AND current_owner_customer_id = %s
|
|
ORDER BY created_at DESC
|
|
"""
|
|
result = execute_query(query, (customer_id,))
|
|
return result or []
|
|
|
|
|
|
@router.get("/hardware/by-contact/{contact_id}", response_model=List[dict])
|
|
async def list_hardware_by_contact(contact_id: int):
|
|
"""List hardware assets linked to a contact via company relations."""
|
|
query = """
|
|
SELECT h.*
|
|
FROM hardware_assets h
|
|
JOIN contact_companies cc ON cc.customer_id = h.current_owner_customer_id
|
|
WHERE cc.contact_id = %s AND h.deleted_at IS NULL
|
|
ORDER BY h.created_at DESC
|
|
"""
|
|
result = execute_query(query, (contact_id,))
|
|
return result or []
|
|
|
|
|
|
@router.post("/hardware", response_model=dict)
|
|
async def create_hardware(data: dict):
|
|
"""Create a new hardware asset."""
|
|
try:
|
|
query = """
|
|
INSERT INTO hardware_assets (
|
|
asset_type, brand, model, serial_number, customer_asset_id,
|
|
internal_asset_id, notes, current_owner_type, current_owner_customer_id,
|
|
status, status_reason, warranty_until, end_of_life,
|
|
anydesk_id, anydesk_link
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING *
|
|
"""
|
|
params = (
|
|
data.get("asset_type"),
|
|
data.get("brand"),
|
|
data.get("model"),
|
|
data.get("serial_number"),
|
|
data.get("customer_asset_id"),
|
|
data.get("internal_asset_id"),
|
|
data.get("notes"),
|
|
data.get("current_owner_type", "bmc"),
|
|
data.get("current_owner_customer_id"),
|
|
data.get("status", "active"),
|
|
data.get("status_reason"),
|
|
data.get("warranty_until"),
|
|
data.get("end_of_life"),
|
|
data.get("anydesk_id"),
|
|
data.get("anydesk_link"),
|
|
)
|
|
result = execute_query(query, params)
|
|
if not result:
|
|
raise HTTPException(status_code=500, detail="Failed to create hardware")
|
|
|
|
hardware = result[0]
|
|
logger.info(f"✅ Created hardware asset: {hardware['id']} - {hardware['brand']} {hardware['model']}")
|
|
|
|
# Create initial ownership record if owner specified
|
|
if data.get("current_owner_type"):
|
|
ownership_query = """
|
|
INSERT INTO hardware_ownership_history (
|
|
hardware_id, owner_type, owner_customer_id, start_date, notes
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
"""
|
|
ownership_params = (
|
|
hardware['id'],
|
|
data.get("current_owner_type"),
|
|
data.get("current_owner_customer_id"),
|
|
date.today(),
|
|
"Initial ownership record"
|
|
)
|
|
execute_query(ownership_query, ownership_params)
|
|
|
|
return hardware
|
|
except Exception as e:
|
|
logger.error(f"❌ Failed to create hardware: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/hardware/quick", response_model=dict)
|
|
async def quick_create_hardware(data: dict):
|
|
"""Quick create hardware with minimal fields (name + AnyDesk info)."""
|
|
try:
|
|
name = (data.get("name") or "").strip()
|
|
customer_id = data.get("customer_id")
|
|
anydesk_id = (data.get("anydesk_id") or "").strip() or None
|
|
anydesk_link = (data.get("anydesk_link") or "").strip() or None
|
|
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="Name is required")
|
|
if not customer_id:
|
|
raise HTTPException(status_code=400, detail="Customer ID is required")
|
|
|
|
query = """
|
|
INSERT INTO hardware_assets (
|
|
asset_type, model, current_owner_type, current_owner_customer_id,
|
|
status, anydesk_id, anydesk_link, notes
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING *
|
|
"""
|
|
params = (
|
|
"andet",
|
|
name,
|
|
"customer",
|
|
customer_id,
|
|
"active",
|
|
anydesk_id,
|
|
anydesk_link,
|
|
"Quick created from case/ticket flow",
|
|
)
|
|
result = execute_query(query, params)
|
|
if not result:
|
|
raise HTTPException(status_code=500, detail="Failed to create hardware")
|
|
|
|
return result[0]
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"❌ Failed to quick-create hardware: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/hardware/{hardware_id}", response_model=dict)
|
|
async def get_hardware(hardware_id: int):
|
|
"""Get hardware details by ID."""
|
|
query = "SELECT * FROM hardware_assets WHERE id = %s AND deleted_at IS NULL"
|
|
result = execute_query(query, (hardware_id,))
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Hardware not found")
|
|
|
|
logger.info(f"✅ Retrieved hardware: {hardware_id}")
|
|
return result[0]
|
|
|
|
|
|
@router.patch("/hardware/{hardware_id}", response_model=dict)
|
|
async def update_hardware(hardware_id: int, data: dict):
|
|
"""Update hardware asset."""
|
|
try:
|
|
# Build dynamic update query
|
|
update_fields = []
|
|
params = []
|
|
|
|
allowed_fields = [
|
|
"asset_type", "brand", "model", "serial_number", "customer_asset_id",
|
|
"internal_asset_id", "notes", "current_owner_type", "current_owner_customer_id",
|
|
"status", "status_reason", "warranty_until", "end_of_life",
|
|
"follow_up_date", "follow_up_owner_user_id", "anydesk_id", "anydesk_link"
|
|
]
|
|
|
|
for field in allowed_fields:
|
|
if field in data:
|
|
update_fields.append(f"{field} = %s")
|
|
params.append(data[field])
|
|
|
|
if not update_fields:
|
|
raise HTTPException(status_code=400, detail="No valid fields to update")
|
|
|
|
update_fields.append("updated_at = NOW()")
|
|
params.append(hardware_id)
|
|
|
|
query = f"""
|
|
UPDATE hardware_assets
|
|
SET {', '.join(update_fields)}
|
|
WHERE id = %s AND deleted_at IS NULL
|
|
RETURNING *
|
|
"""
|
|
|
|
result = execute_query(query, tuple(params))
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Hardware not found")
|
|
|
|
logger.info(f"✅ Updated hardware: {hardware_id}")
|
|
return result[0]
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"❌ Failed to update hardware {hardware_id}: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.delete("/hardware/{hardware_id}")
|
|
async def delete_hardware(hardware_id: int):
|
|
"""Soft-delete hardware asset."""
|
|
query = """
|
|
UPDATE hardware_assets
|
|
SET deleted_at = NOW()
|
|
WHERE id = %s AND deleted_at IS NULL
|
|
RETURNING id
|
|
"""
|
|
result = execute_query(query, (hardware_id,))
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Hardware not found")
|
|
|
|
logger.info(f"✅ Deleted hardware: {hardware_id}")
|
|
return {"message": "Hardware deleted successfully"}
|
|
|
|
|
|
# ============================================================================
|
|
# Ownership History Endpoints
|
|
# ============================================================================
|
|
|
|
@router.get("/hardware/{hardware_id}/ownership", response_model=List[dict])
|
|
async def get_ownership_history(hardware_id: int):
|
|
"""Get ownership history for hardware."""
|
|
query = """
|
|
SELECT * FROM hardware_ownership_history
|
|
WHERE hardware_id = %s AND deleted_at IS NULL
|
|
ORDER BY start_date DESC
|
|
"""
|
|
result = execute_query(query, (hardware_id,))
|
|
logger.info(f"✅ Retrieved ownership history for hardware: {hardware_id}")
|
|
return result or []
|
|
|
|
|
|
@router.post("/hardware/{hardware_id}/ownership", response_model=dict)
|
|
async def add_ownership_record(hardware_id: int, data: dict):
|
|
"""Add ownership record (auto-closes previous active ownership)."""
|
|
try:
|
|
# Close any active ownership records
|
|
close_query = """
|
|
UPDATE hardware_ownership_history
|
|
SET end_date = %s
|
|
WHERE hardware_id = %s AND end_date IS NULL AND deleted_at IS NULL
|
|
"""
|
|
execute_query(close_query, (date.today(), hardware_id))
|
|
|
|
# Create new ownership record
|
|
insert_query = """
|
|
INSERT INTO hardware_ownership_history (
|
|
hardware_id, owner_type, owner_customer_id, start_date, notes
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
RETURNING *
|
|
"""
|
|
params = (
|
|
hardware_id,
|
|
data.get("owner_type"),
|
|
data.get("owner_customer_id"),
|
|
data.get("start_date", date.today()),
|
|
data.get("notes")
|
|
)
|
|
result = execute_query(insert_query, params)
|
|
|
|
# Update current owner in hardware_assets
|
|
update_query = """
|
|
UPDATE hardware_assets
|
|
SET current_owner_type = %s, current_owner_customer_id = %s, updated_at = NOW()
|
|
WHERE id = %s
|
|
"""
|
|
execute_query(update_query, (data.get("owner_type"), data.get("owner_customer_id"), hardware_id))
|
|
|
|
logger.info(f"✅ Added ownership record for hardware: {hardware_id}")
|
|
return result[0]
|
|
except Exception as e:
|
|
logger.error(f"❌ Failed to add ownership record: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# Location History Endpoints
|
|
# ============================================================================
|
|
|
|
@router.get("/hardware/{hardware_id}/locations", response_model=List[dict])
|
|
async def get_location_history(hardware_id: int):
|
|
"""Get location history for hardware."""
|
|
query = """
|
|
SELECT * FROM hardware_location_history
|
|
WHERE hardware_id = %s AND deleted_at IS NULL
|
|
ORDER BY start_date DESC
|
|
"""
|
|
result = execute_query(query, (hardware_id,))
|
|
logger.info(f"✅ Retrieved location history for hardware: {hardware_id}")
|
|
return result or []
|
|
|
|
|
|
@router.post("/hardware/{hardware_id}/locations", response_model=dict)
|
|
async def add_location_record(hardware_id: int, data: dict):
|
|
"""Add location record (auto-closes previous active location)."""
|
|
try:
|
|
# Close any active location records
|
|
close_query = """
|
|
UPDATE hardware_location_history
|
|
SET end_date = %s
|
|
WHERE hardware_id = %s AND end_date IS NULL AND deleted_at IS NULL
|
|
"""
|
|
execute_query(close_query, (date.today(), hardware_id))
|
|
|
|
# Create new location record
|
|
insert_query = """
|
|
INSERT INTO hardware_location_history (
|
|
hardware_id, location_id, location_name, start_date, notes
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
RETURNING *
|
|
"""
|
|
params = (
|
|
hardware_id,
|
|
data.get("location_id"),
|
|
data.get("location_name"),
|
|
data.get("start_date", date.today()),
|
|
data.get("notes")
|
|
)
|
|
result = execute_query(insert_query, params)
|
|
|
|
# Update current location in hardware_assets
|
|
update_query = """
|
|
UPDATE hardware_assets
|
|
SET current_location_id = %s, updated_at = NOW()
|
|
WHERE id = %s
|
|
"""
|
|
execute_query(update_query, (data.get("location_id"), hardware_id))
|
|
|
|
logger.info(f"✅ Added location record for hardware: {hardware_id}")
|
|
return result[0]
|
|
except Exception as e:
|
|
logger.error(f"❌ Failed to add location record: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
# ============================================================================
|
|
# Attachment Endpoints
|
|
# ============================================================================
|
|
|
|
@router.get("/hardware/{hardware_id}/attachments", response_model=List[dict])
|
|
async def get_attachments(hardware_id: int):
|
|
"""Get all attachments for hardware."""
|
|
query = """
|
|
SELECT * FROM hardware_attachments
|
|
WHERE hardware_id = %s AND deleted_at IS NULL
|
|
ORDER BY uploaded_at DESC
|
|
"""
|
|
result = execute_query(query, (hardware_id,))
|
|
logger.info(f"✅ Retrieved {len(result) if result else 0} attachments for hardware: {hardware_id}")
|
|
return result or []
|
|
|
|
|
|
@router.post("/hardware/{hardware_id}/attachments", response_model=dict)
|
|
async def upload_attachment(hardware_id: int, data: dict):
|
|
"""Upload attachment for hardware."""
|
|
try:
|
|
# Generate storage reference (in production, this would upload to cloud storage)
|
|
storage_ref = f"hardware/{hardware_id}/{uuid.uuid4()}_{data.get('file_name')}"
|
|
|
|
query = """
|
|
INSERT INTO hardware_attachments (
|
|
hardware_id, file_type, file_name, storage_ref,
|
|
file_size_bytes, mime_type, description, uploaded_by_user_id
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING *
|
|
"""
|
|
params = (
|
|
hardware_id,
|
|
data.get("file_type", "other"),
|
|
data.get("file_name"),
|
|
storage_ref,
|
|
data.get("file_size_bytes"),
|
|
data.get("mime_type"),
|
|
data.get("description"),
|
|
data.get("uploaded_by_user_id")
|
|
)
|
|
result = execute_query(query, params)
|
|
|
|
logger.info(f"✅ Uploaded attachment for hardware: {hardware_id}")
|
|
return result[0]
|
|
except Exception as e:
|
|
logger.error(f"❌ Failed to upload attachment: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.delete("/hardware/{hardware_id}/attachments/{attachment_id}")
|
|
async def delete_attachment(hardware_id: int, attachment_id: int):
|
|
"""Soft-delete attachment."""
|
|
query = """
|
|
UPDATE hardware_attachments
|
|
SET deleted_at = NOW()
|
|
WHERE id = %s AND hardware_id = %s AND deleted_at IS NULL
|
|
RETURNING id
|
|
"""
|
|
result = execute_query(query, (attachment_id, hardware_id))
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Attachment not found")
|
|
|
|
logger.info(f"✅ Deleted attachment {attachment_id} for hardware: {hardware_id}")
|
|
return {"message": "Attachment deleted successfully"}
|
|
|
|
|
|
# ============================================================================
|
|
# Case Relations Endpoints
|
|
# ============================================================================
|
|
|
|
@router.get("/hardware/{hardware_id}/cases", response_model=List[dict])
|
|
async def get_related_cases(hardware_id: int):
|
|
"""Get all cases related to this hardware."""
|
|
query = """
|
|
SELECT hcr.*, s.titel, s.status, s.customer_id
|
|
FROM hardware_case_relations hcr
|
|
LEFT JOIN sag_sager s ON hcr.case_id = s.id
|
|
WHERE hcr.hardware_id = %s AND hcr.deleted_at IS NULL AND s.deleted_at IS NULL
|
|
ORDER BY hcr.created_at DESC
|
|
"""
|
|
result = execute_query(query, (hardware_id,))
|
|
logger.info(f"✅ Retrieved {len(result) if result else 0} related cases for hardware: {hardware_id}")
|
|
return result or []
|
|
|
|
|
|
@router.post("/hardware/{hardware_id}/cases", response_model=dict)
|
|
async def link_case(hardware_id: int, data: dict):
|
|
"""Link hardware to a case."""
|
|
try:
|
|
query = """
|
|
INSERT INTO hardware_case_relations (
|
|
hardware_id, case_id, relation_type
|
|
)
|
|
VALUES (%s, %s, %s)
|
|
RETURNING *
|
|
"""
|
|
params = (
|
|
hardware_id,
|
|
data.get("case_id"),
|
|
data.get("relation_type", "related")
|
|
)
|
|
result = execute_query(query, params)
|
|
|
|
logger.info(f"✅ Linked hardware {hardware_id} to case {data.get('case_id')}")
|
|
return result[0]
|
|
except Exception as e:
|
|
logger.error(f"❌ Failed to link case: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.delete("/hardware/{hardware_id}/cases/{case_id}")
|
|
async def unlink_case(hardware_id: int, case_id: int):
|
|
"""Unlink hardware from a case."""
|
|
query = """
|
|
UPDATE hardware_case_relations
|
|
SET deleted_at = NOW()
|
|
WHERE hardware_id = %s AND case_id = %s AND deleted_at IS NULL
|
|
RETURNING id
|
|
"""
|
|
result = execute_query(query, (hardware_id, case_id))
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Case relation not found")
|
|
|
|
logger.info(f"✅ Unlinked hardware {hardware_id} from case {case_id}")
|
|
return {"message": "Case unlinked successfully"}
|
|
|
|
|
|
# ============================================================================
|
|
# Tag Endpoints
|
|
# ============================================================================
|
|
|
|
@router.get("/hardware/{hardware_id}/tags", response_model=List[dict])
|
|
async def get_tags(hardware_id: int):
|
|
"""Get all tags for hardware."""
|
|
query = """
|
|
SELECT * FROM hardware_tags
|
|
WHERE hardware_id = %s AND deleted_at IS NULL
|
|
ORDER BY created_at DESC
|
|
"""
|
|
result = execute_query(query, (hardware_id,))
|
|
logger.info(f"✅ Retrieved {len(result) if result else 0} tags for hardware: {hardware_id}")
|
|
return result or []
|
|
|
|
|
|
@router.post("/hardware/{hardware_id}/tags", response_model=dict)
|
|
async def add_tag(hardware_id: int, data: dict):
|
|
"""Add tag to hardware."""
|
|
try:
|
|
query = """
|
|
INSERT INTO hardware_tags (
|
|
hardware_id, tag_name, tag_type
|
|
)
|
|
VALUES (%s, %s, %s)
|
|
RETURNING *
|
|
"""
|
|
params = (
|
|
hardware_id,
|
|
data.get("tag_name"),
|
|
data.get("tag_type", "manual")
|
|
)
|
|
result = execute_query(query, params)
|
|
|
|
logger.info(f"✅ Added tag '{data.get('tag_name')}' to hardware: {hardware_id}")
|
|
return result[0]
|
|
except Exception as e:
|
|
logger.error(f"❌ Failed to add tag: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.delete("/hardware/{hardware_id}/tags/{tag_id}")
|
|
async def delete_tag(hardware_id: int, tag_id: int):
|
|
"""Delete tag from hardware."""
|
|
query = """
|
|
UPDATE hardware_tags
|
|
SET deleted_at = NOW()
|
|
WHERE id = %s AND hardware_id = %s AND deleted_at IS NULL
|
|
RETURNING id
|
|
"""
|
|
result = execute_query(query, (tag_id, hardware_id))
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Tag not found")
|
|
|
|
logger.info(f"✅ Deleted tag {tag_id} from hardware: {hardware_id}")
|
|
return {"message": "Tag deleted successfully"}
|
|
|
|
|
|
# ============================================================================
|
|
# Search Endpoint
|
|
# ============================================================================
|
|
|
|
@router.get("/search/hardware", response_model=List[dict])
|
|
async def search_hardware(q: str = Query(..., min_length=1)):
|
|
"""Search hardware by serial number, model, or brand."""
|
|
query = """
|
|
SELECT * FROM hardware_assets
|
|
WHERE deleted_at IS NULL
|
|
AND (
|
|
serial_number ILIKE %s
|
|
OR model ILIKE %s
|
|
OR brand ILIKE %s
|
|
OR customer_asset_id ILIKE %s
|
|
OR internal_asset_id ILIKE %s
|
|
)
|
|
ORDER BY created_at DESC
|
|
LIMIT 50
|
|
"""
|
|
search_param = f"%{q}%"
|
|
result = execute_query(query, (search_param, search_param, search_param, search_param, search_param))
|
|
|
|
logger.info(f"✅ Search for '{q}' returned {len(result) if result else 0} results")
|
|
return result or []
|