#views.py
from django.shortcuts import render, redirect, get_object_or_404
from .models import AssetData, AssetType,AssetImage,Asset_movement_data
from .forms import AssetDataForm, AssetTypeForm,AssetMovementForm
from django.urls import reverse
from .utils import generate_qr_code
from authentication.models import Branches
from django.http import JsonResponse
from django.contrib.auth.decorators import login_required
def asset_list(request):
user = request.user
if user.is_superuser:
assets = AssetData.objects.all()
asset_type = AssetType.objects.all()
elif user.is_superadmin and user.head_office:
branches = Branches.objects.filter(head_office=user.head_office)
assets = AssetData.objects.filter(branch__in=branches)
asset_type = AssetType.objects.filter(assets__branch__in=branches).distinct()
elif user.is_admin and user.branch:
assets = AssetData.objects.filter(branch=user.branch)
asset_type = AssetType.objects.filter(assets__branch=user.branch).distinct()
else:
assets = AssetData.objects.none()
asset_type = AssetType.objects.none()
print(assets)
asset_data=[]
for asset in assets:
vessel_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id]))
qr_code_url = generate_qr_code(asset.asset_name, vessel_url)
asset_data.append({'asset': asset, 'qr_code_url': qr_code_url})
return render(request, 'asset/asset_list.html', {'assets': asset_data,'asset_type':asset_type})
def asset_create(request):
if request.method == 'POST':
if 'asset_type_submit' in request.POST:
return handle_asset_type_submit(request)
elif 'asset_submit' in request.POST:
return handle_asset_submit(request)
else:
asset_type_form = AssetTypeForm(prefix='asset_type')
asset_form = AssetDataForm(prefix='asset')
return render(request, 'asset/asset_form.html', {
'asset_type_form': asset_type_form,
'asset_form': asset_form,
})
def handle_asset_type_submit(request):
asset_type_form = AssetTypeForm(request.POST, prefix='asset_type')
asset_form = AssetDataForm(prefix='asset')
if asset_type_form.is_valid():
asset_type_form.save()
return redirect('asset_create')
return render(request, 'asset/asset_form.html', {
'asset_type_form': asset_type_form,
'asset_form': asset_form,
})
def handle_asset_submit(request):
asset_form = AssetDataForm(request.POST, request.FILES, prefix='asset')
print(f"POST data: {request.POST}")
print(f"Form errors: {asset_form.errors}")
asset_type_form = AssetTypeForm(prefix='asset_type')
if asset_form.is_valid():
asset = asset_form.save(commit=False)
asset.save()
for image in asset_form.cleaned_data['images']:
AssetImage.objects.create(asset=asset, image=image)
asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id]))
asset.qr_code_base64 = generate_qr_code(asset.asset_name, asset_url)
asset.save()
return redirect('asset_list')
return render(request, 'asset/asset_form.html', {
'asset_type_form': asset_type_form,
'asset_form': asset_form,
})
def get_branches(request):
office_id = request.GET.get('office_id')
print(office_id)
branches = Branches.objects.filter(head_office_id=office_id).values('id', 'branch_name')
print(branches)
return JsonResponse(list(branches), safe=False)
def asset_detail(request, pk):
asset = get_object_or_404(AssetData, pk=pk)
asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id]))
qr_code_base64 = generate_qr_code(asset.asset_name, asset_url)
return render(request, 'asset/asset_detail.html', {'asset': asset,'qr_code_base64': qr_code_base64,})
def asset_type_edit(request, pk):
asset = get_object_or_404(AssetType, pk=pk)
if request.method == 'POST':
asset_type_form = AssetTypeForm(request.POST, instance=asset, prefix='asset_type')
if 'asset_type_submit' in request.POST and asset_type_form.is_valid():
asset_type_form.save()
return redirect('asset_list')
else:
asset_type_form = AssetTypeForm(instance=asset, prefix='asset_type')
return render(request, 'asset/edit_assettype.html', {'asset_type_form': asset_type_form})
def asset_edit(request, pk):
asset = get_object_or_404(AssetData, pk=pk)
if request.method == 'POST':
asset_form = AssetDataForm(request.POST, request.FILES, instance=asset, prefix='asset')
if 'asset_submit' in request.POST and asset_form.is_valid():
asset = asset_form.save(commit=False)
asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id]))
asset.qr_code_base64 = generate_qr_code(asset.asset_name, asset_url)
asset.save()
# Clear existing images if necessary
asset.images.all().delete()
# Handle image uploads
for image in asset_form.cleaned_data['images']:
AssetImage.objects.create(asset=asset, image=image)
return redirect('asset_list')
else:
asset_form = AssetDataForm(instance=asset, prefix='asset')
return render(request, 'asset/asset_edit.html', {
'asset_form': asset_form,
})
def asset_delete(request, pk):
asset = get_object_or_404(AssetData, pk=pk)
asset.delete()
return redirect('asset_list')
def asset_link(request, asset_id):
asset = get_object_or_404(AssetData, id=asset_id)
main_details = [
{'key': 'Name', 'value': asset.asset_name},
{'key': 'Asset Type', 'value': asset.asset_type},
{'key': 'Head Office', 'value': asset.office},
{'key': 'Branch', 'value': asset.branch.branch_name},
{'key': 'Model', 'value': asset.model},
{'key': 'Purchase Date', 'value': asset.purchase_date},
{'key': 'Warranty Info', 'value': asset.warranty_info},
]
images = asset.images.all()
context = {
'asset': asset,
'main_details': main_details,
'images': images,
}
return render(request, 'asset/asset_link.html', context)
from django.http import HttpResponse
from django.conf import settings
def download_qr_code(request, vessel_id):
vessel = get_object_or_404(AssetData, id=vessel_id)
qr_code_path = f'{settings.MEDIA_ROOT}/qr_codes/{vessel.name}.png'
try:
with open(qr_code_path, 'rb') as f:
response = HttpResponse(f.read(), content_type='image/png')
response['Content-Disposition'] = f'attachment; filename="{vessel.name}_qr.png"'
return response
except FileNotFoundError:
return HttpResponse("QR code not found", status=404)
from django.db.models import Q
def asset_movement_page(request):
if request.method == 'POST':
form = AssetMovementForm(request.POST)
if form.is_valid():
form.save()
return redirect('asset_movement_list')
else:
print(form.errors)
else:
form = AssetMovementForm()
# movements = Asset_movement_data.objects.all().order_by('-movement_date')
user = request.user
if user.is_superuser:
movements = Asset_movement_data.objects.all()
elif user.is_superadmin and user.head_office:
branches = Branches.objects.filter(head_office=user.head_office)
movements = Asset_movement_data.objects.filter(
Q(from_branch__in=branches) | Q(to_branch__in=branches)
)
elif user.is_admin and user.branch:
movements = Asset_movement_data.objects.filter(
Q(from_branch=user.branch) | Q(to_branch=user.branch)
)
else:
movements = Asset_movement_data.objects.none()
return render(request, 'asset/asset_movement_list.html', {'form': form, 'movements': movements})
def edit_asset_movement(request, movement_id):
movement = get_object_or_404(Asset_movement_data, id=movement_id)
if request.method == 'POST':
form = AssetMovementForm(request.POST, instance=movement)
if form.is_valid():
form.save()
return redirect('asset_movement_list')
else:
form = AssetMovementForm(instance=movement)
return render(request, 'asset/edit_asset_movement.html', {'form': form, 'movement': movement})
def asset_movement_success(request):
return render(request, 'asset/asset_movement_success.html')
def asset_filter_by_branches(request, branch_id):
branch = Branches.objects.get(id=branch_id)
# Assets currently in the branch, excluding those that have been moved
moved_out_assets = Asset_movement_data.objects.filter(from_branch=branch).values_list('asset_id', flat=True)
assets = AssetData.objects.filter(branch=branch).exclude(id__in=moved_out_assets)
# Assets moved from the branch
moved_assets = Asset_movement_data.objects.filter(from_branch=branch).order_by('-movement_date')
# Assets received at the branch
received_assets = Asset_movement_data.objects.filter(to_branch=branch).order_by('-movement_date')
context = {
'branch': branch,
'assets': assets,
'moved_assets': moved_assets,
'received_assets': received_assets,
}
return render(request, 'asset/assets_by_branch.html', context)
@login_required
def asset_movement_list(request):
user = request.user
if user.is_superuser:
movements = Asset_movement_data.objects.all()
elif user.is_superadmin and user.head_office:
branches = Branches.objects.filter(head_office=user.head_office)
movements = Asset_movement_data.objects.filter(
Q(from_branch__in=branches) | Q(to_branch__in=branches)
)
else:
movements = Asset_movement_data.objects.none()
form = AssetMovementForm() # Assuming you have this form
return render(request, 'authentication/asset_movement_list.html', {
'movements': movements,
'form': form
})
#views
#views.py
from django.shortcuts import render, redirect, get_object_or_404
from .models import AssetData, AssetType,AssetImage,Asset_movement_data
from .forms import AssetDataForm, AssetTypeForm,AssetMovementForm
from django.urls import reverse
from .utils import generate_qr_code
from authentication.models import Branches
from django.http import JsonResponse
from django.contrib.auth.decorators import login_required
def asset_list(request):
user = request.user
if user.is_superuser:
assets = AssetData.objects.all()
asset_type = AssetType.objects.all()
elif user.is_superadmin and user.head_office:
branches = Branches.objects.filter(head_office=user.head_office)
assets = AssetData.objects.filter(branch__in=branches)
asset_type = AssetType.objects.filter(assets__branch__in=branches).distinct()
elif user.is_admin and user.branch:
assets = AssetData.objects.filter(branch=user.branch)
asset_type = AssetType.objects.filter(assets__branch=user.branch).distinct()
else:
assets = AssetData.objects.none()
asset_type = AssetType.objects.none()
print(assets)
asset_data=[]
for asset in assets:
vessel_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id]))
qr_code_url = generate_qr_code(asset.asset_name, vessel_url)
asset_data.append({'asset': asset, 'qr_code_url': qr_code_url})
return render(request, 'asset/asset_list.html', {'assets': asset_data,'asset_type':asset_type})
def asset_create(request):
if request.method == 'POST':
if 'asset_type_submit' in request.POST:
return handle_asset_type_submit(request)
elif 'asset_submit' in request.POST:
return handle_asset_submit(request)
else:
asset_type_form = AssetTypeForm(prefix='asset_type')
asset_form = AssetDataForm(prefix='asset')
return render(request, 'asset/asset_form.html', {
'asset_type_form': asset_type_form,
'asset_form': asset_form,
})
def handle_asset_type_submit(request):
asset_type_form = AssetTypeForm(request.POST, prefix='asset_type')
asset_form = AssetDataForm(prefix='asset')
if asset_type_form.is_valid():
asset_type_form.save()
return redirect('asset_create')
return render(request, 'asset/asset_form.html', {
'asset_type_form': asset_type_form,
'asset_form': asset_form,
})
def handle_asset_submit(request):
asset_form = AssetDataForm(request.POST, request.FILES, prefix='asset')
print(f"POST data: {request.POST}")
print(f"Form errors: {asset_form.errors}")
asset_type_form = AssetTypeForm(prefix='asset_type')
if asset_form.is_valid():
asset = asset_form.save(commit=False)
asset.save()
for image in asset_form.cleaned_data['images']:
AssetImage.objects.create(asset=asset, image=image)
asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id]))
asset.qr_code_base64 = generate_qr_code(asset.asset_name, asset_url)
asset.save()
return redirect('asset_list')
return render(request, 'asset/asset_form.html', {
'asset_type_form': asset_type_form,
'asset_form': asset_form,
})
def get_branches(request):
office_id = request.GET.get('office_id')
print(office_id)
branches = Branches.objects.filter(head_office_id=office_id).values('id', 'branch_name')
print(branches)
return JsonResponse(list(branches), safe=False)
def asset_detail(request, pk):
asset = get_object_or_404(AssetData, pk=pk)
asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id]))
qr_code_base64 = generate_qr_code(asset.asset_name, asset_url)
return render(request, 'asset/asset_detail.html', {'asset': asset,'qr_code_base64': qr_code_base64,})
def asset_type_edit(request, pk):
asset = get_object_or_404(AssetType, pk=pk)
if request.method == 'POST':
asset_type_form = AssetTypeForm(request.POST, instance=asset, prefix='asset_type')
if 'asset_type_submit' in request.POST and asset_type_form.is_valid():
asset_type_form.save()
return redirect('asset_list')
else:
asset_type_form = AssetTypeForm(instance=asset, prefix='asset_type')
return render(request, 'asset/edit_assettype.html', {'asset_type_form': asset_type_form})
def asset_edit(request, pk):
asset = get_object_or_404(AssetData, pk=pk)
if request.method == 'POST':
asset_form = AssetDataForm(request.POST, request.FILES, instance=asset, prefix='asset')
if 'asset_submit' in request.POST and asset_form.is_valid():
asset = asset_form.save(commit=False)
asset_url = request.build_absolute_uri(reverse('asset_link', args=[asset.id]))
asset.qr_code_base64 = generate_qr_code(asset.asset_name, asset_url)
asset.save()
# Clear existing images if necessary
asset.images.all().delete()
# Handle image uploads
for image in asset_form.cleaned_data['images']:
AssetImage.objects.create(asset=asset, image=image)
return redirect('asset_list')
else:
asset_form = AssetDataForm(instance=asset, prefix='asset')
return render(request, 'asset/asset_edit.html', {
'asset_form': asset_form,
})
def asset_delete(request, pk):
asset = get_object_or_404(AssetData, pk=pk)
asset.delete()
return redirect('asset_list')
def asset_link(request, asset_id):
asset = get_object_or_404(AssetData, id=asset_id)
main_details = [
{'key': 'Name', 'value': asset.asset_name},
{'key': 'Asset Type', 'value': asset.asset_type},
{'key': 'Head Office', 'value': asset.office},
{'key': 'Branch', 'value': asset.branch.branch_name},
{'key': 'Model', 'value': asset.model},
{'key': 'Purchase Date', 'value': asset.purchase_date},
{'key': 'Warranty Info', 'value': asset.warranty_info},
]
images = asset.images.all()
context = {
'asset': asset,
'main_details': main_details,
'images': images,
}
return render(request, 'asset/asset_link.html', context)
from django.http import HttpResponse
from django.conf import settings
def download_qr_code(request, vessel_id):
vessel = get_object_or_404(AssetData, id=vessel_id)
qr_code_path = f'{settings.MEDIA_ROOT}/qr_codes/{vessel.name}.png'
try:
with open(qr_code_path, 'rb') as f:
response = HttpResponse(f.read(), content_type='image/png')
response['Content-Disposition'] = f'attachment; filename="{vessel.name}_qr.png"'
return response
except FileNotFoundError:
return HttpResponse("QR code not found", status=404)
from django.db.models import Q
def asset_movement_page(request):
if request.method == 'POST':
form = AssetMovementForm(request.POST)
if form.is_valid():
form.save()
return redirect('asset_movement_list')
else:
print(form.errors)
else:
form = AssetMovementForm()
# movements = Asset_movement_data.objects.all().order_by('-movement_date')
user = request.user
if user.is_superuser:
movements = Asset_movement_data.objects.all()
elif user.is_superadmin and user.head_office:
branches = Branches.objects.filter(head_office=user.head_office)
movements = Asset_movement_data.objects.filter(
Q(from_branch__in=branches) | Q(to_branch__in=branches)
)
elif user.is_admin and user.branch:
movements = Asset_movement_data.objects.filter(
Q(from_branch=user.branch) | Q(to_branch=user.branch)
)
else:
movements = Asset_movement_data.objects.none()
return render(request, 'asset/asset_movement_list.html', {'form': form, 'movements': movements})
def edit_asset_movement(request, movement_id):
movement = get_object_or_404(Asset_movement_data, id=movement_id)
if request.method == 'POST':
form = AssetMovementForm(request.POST, instance=movement)
if form.is_valid():
form.save()
return redirect('asset_movement_list')
else:
form = AssetMovementForm(instance=movement)
return render(request, 'asset/edit_asset_movement.html', {'form': form, 'movement': movement})
def asset_movement_success(request):
return render(request, 'asset/asset_movement_success.html')
def asset_filter_by_branches(request, branch_id):
branch = Branches.objects.get(id=branch_id)
# Assets currently in the branch, excluding those that have been moved
moved_out_assets = Asset_movement_data.objects.filter(from_branch=branch).values_list('asset_id', flat=True)
assets = AssetData.objects.filter(branch=branch).exclude(id__in=moved_out_assets)
# Assets moved from the branch
moved_assets = Asset_movement_data.objects.filter(from_branch=branch).order_by('-movement_date')
# Assets received at the branch
received_assets = Asset_movement_data.objects.filter(to_branch=branch).order_by('-movement_date')
context = {
'branch': branch,
'assets': assets,
'moved_assets': moved_assets,
'received_assets': received_assets,
}
return render(request, 'asset/assets_by_branch.html', context)
@login_required
def asset_movement_list(request):
user = request.user
if user.is_superuser:
movements = Asset_movement_data.objects.all()
elif user.is_superadmin and user.head_office:
branches = Branches.objects.filter(head_office=user.head_office)
movements = Asset_movement_data.objects.filter(
Q(from_branch__in=branches) | Q(to_branch__in=branches)
)
else:
movements = Asset_movement_data.objects.none()
form = AssetMovementForm() # Assuming you have this form
return render(request, 'authentication/asset_movement_list.html', {
'movements': movements,
'form': form
})
#views
import os
import sys
import re
import json
import logging
from pathlib import Path
from dotenv import load_dotenv
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from django.views.decorators.csrf import csrf_exempt
from django.db import connection
from langchain_community.agent_toolkits import create_sql_agent
from langchain_community.utilities import SQLDatabase
from langchain_groq import ChatGroq
from langchain.tools import BaseTool
from langchain.chains import LLMChain
from langchain.agents import ZeroShotAgent, Tool, AgentExecutor
from langchain.schema import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from .models import AssetData
from langchain.agents import load_tools, initialize_agent, AgentType
# Setup Django environment
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(project_root)
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "asset_management_system.settings")
import django
django.setup()
# Setup logging
logger = logging.getLogger(__name__)
# Load environment variables and set up API keys
load_dotenv()
api_key = os.getenv('OPENAI_API_KEY')
groq_api_key = os.getenv("GROQ_API_KEY")
os.environ["GROQ_API_KEY"] = groq_api_key
# Initialize LLM
llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
# Constants
WEBAPP_INFO = """
The Asset Management System features a login system with three roles: Master Admin, Super Admin, and Admin, each with specific permissions. The system supports various asset types, with QR codes generated for each asset for easy tracking and management. Assets can be transferred between branches, with movement history tracked for audits.
"""
offense_words = ["Ninde ama, pati!", "Aaana kunna", "aan-vedi", "achante andi", "Achinga Kunnan", "adichu poli", "Ajoli ka Thajoli", "ammaudi pootil poocha", "Ammente Pooru", "amminhnha", "Andi pidiyan", "anna kunna", "Appi", "appikunna", "Arjun shambu", "arraykku chuttum poorruu ullaval", "avaraathi", "avarathi mone", "Beejam", "chalam nakki", "Chandi", "Chathavante andi oombiya", "Cheekoothi mon", "Cheekoothi mone", "Chokka lingam", "CHotta kunnan", "Chotta Tha", "coondi", "Da patti", "Dushtan", "eli kunna", "Ettukali Patti Pooran", "inchi mola", "johninte andi", "kaatu poori", "kallel oouuki", "kandara oli", "kandi", "kandi theeni", "Kandu", "Kanni", "Kanni mola", "kara vedi", "karim kunna", "karim pooran", "katta thayoli", "kazhappu perutha mairan", "Keepu", "Kettal", "kodam nakiii", "Kolekkeri", "Koothara", "Koothi", "Koothichi", "kotham", "kotham kalakki", "kotham nakku", "Koyimani", "kuch", "kulamavuka", "kundan", "kundaroli poori mone", "Kundi", "Kundi mon", "Kundi oota", "kundimol", "kunji kunnan", "kunna", "Kunna", "kunna chappu", "Kunna Oli", "Kunna paal", "Kunna thayoli", "kunna urunji", "kunnappal", "kushukk", "Lick Me", "malayalam", "Mandu", "Maratel ooki", "Masa", "Mattanga Poore", "matti", "Mayiradi mon", "Mlechan", "mola", "moonchikko", "Mula Adi", "mula chappu", "mula kashakku", "mula mol", "Mundachi", "Myir", "Myre", "Myrru", "Naayi meedan", "naayinte mone", "nayinte mone", "Nayinte Mone", "Nayinte Monne", "ni urru myre", "ninde andi maire", "Ninte ama koondi ishtima", "Ninte ammakku vettu nayinte mone", "Ninte ammaku vetu", "ninte ammeda tar", "Ninte Ammede Kothil.", "ninte ammede pooru", "Ninte ammede thenga mairu", "ninte appante andi", "Odu myre", "ookki", "oomban", "oombi mon", "Oooomb", "Ootan", "paara pooru", "paareel ookki", "Pacha tayolli", "Pachila Pooran", "Paik keriko myra", "paiku vetti", "pala thanthakkundaya thaoli", "pallinedayil kanthu", "pambara kutichi", "pampara thayoli", "panchavarna kunna"]
# Create SQLDatabase instance
db_engine = connection.settings_dict['ENGINE']
db_name = connection.settings_dict['NAME']
if isinstance(db_name, (str, Path)):
db_name = str(db_name)
else:
raise ValueError(f"Unexpected type for db_name: {type(db_name)}")
db_url = f"sqlite:///{db_name}" if 'sqlite' in db_engine else f"{db_engine}://{db_name}"
db = SQLDatabase.from_uri(db_url)
# Create SQL agent
sql_agent = create_sql_agent(llm, db=db, agent_type="openai-tools", verbose=True)
class GreetingTool(BaseTool):
name = "greeting_tool"
description = "Use this to respond to greetings or simple queries that don't require complex reasoning."
def _run(self, query: str) -> str:
greetings = ["hello", "hi", "hey", "greetings", "good morning", "good afternoon", "good evening"]
if any(greeting in query.lower() for greeting in greetings):
return "Hello! I'm the Asset Trace AI assistant. How can I help you with asset management today?"
return "I'm here to help with asset management. What specific information or assistance do you need?"
async def _arun(self, query: str):
raise NotImplementedError("This tool does not support async")
# Tool classes
class QueryValidationTool(BaseTool):
name = "query_validation_tool"
description = f"""Evaluate the given query to ensure it is appropriate, respectful, and suitable for general audiences. Check for offensive language, hate speech, explicit content, harassment, sensitive topics, and respectfulness."""
def _run(self, query: str) -> str:
return "VALID" if len(query) > 5 and all(word.lower() not in query.lower() for word in offense_words) else "INVALID"
async def _arun(self, query: str):
raise NotImplementedError("This tool does not support async")
class CybersecurityTool(BaseTool):
name = "cybersecurity_tool"
description = """Checks if the query contains any potentially malicious content, including injection attacks, XSS, malware links, obfuscation techniques, and known vulnerabilities."""
def _run(self, query: str) -> str:
return "SAFE" if all(word.lower() not in query.lower() for word in ["hack", "exploit", "vulnerability", "attack", "'; DROP TABLE", "<script>"]) else "UNSAFE"
async def _arun(self, query: str):
raise NotImplementedError("This tool does not support async")
class AssetAdditionTool(BaseTool):
name = "asset_addition_tool"
description = "Provides instructions on how to add an asset to the Asset Management System."
def _run(self, query: str) -> str:
return """
To add an asset to the Asset Management System:
1. Log in to the system.
2. Navigate to 'Assets' section.
3. Click 'Add New Asset'.
4. Fill in asset details.
5. Click 'Save'.
6. A QR code will be generated for tracking.
Contact support for further assistance.
"""
async def _arun(self, query: str):
raise NotImplementedError("This tool does not support async")
# Create instances of specialized tools
query_validation_tool = QueryValidationTool()
cybersecurity_tool = CybersecurityTool()
asset_addition_tool = AssetAdditionTool()
greeting_tool = GreetingTool()
# Define the tools
tools = [
Tool(name="Greeting", func=greeting_tool.run, description="Use this to handle greetings and simple queries"),
Tool(name="Query Validation", func=query_validation_tool.run, description="Use this to validate user queries"),
Tool(name="Cybersecurity Check", func=cybersecurity_tool.run, description="Use this to perform security checks on user queries"),
Tool(name="Asset Addition", func=asset_addition_tool.run, description="Use this to get instructions on adding an asset"),
Tool(name="Database Query", func=sql_agent.run, description="Use this to query the database for asset information and should not give the table details")
]
# Define the agent
prefix = """You are an AI assistant for the Asset Trace application. Your role is to provide accurate and helpful information about the application, assist users with queries, and guide them in using the app effectively.
Application Details:
{WEBAPP_INFO}
Available Tools: [Greeting, Query Validation, Cybersecurity Check, Asset Addition, Database Query]
Key Instructions:
- For simple greetings or queries, use the Greeting tool directly without further reasoning.
- For more complex queries related to asset management, use the appropriate tools.
- Respond in the same language as the user's query.
- Provide concise answers for simple questions, offering to elaborate if needed.
- Maintain a professional and friendly tone.
- If you're unsure about any information, state that clearly and suggest where the user might find accurate details.
- For complex queries, break down your response into clear, manageable steps.
- Respect user privacy and data security at all times.
Please respond to user queries based on these guidelines and the information provided about the Asset Trace application.
"""
suffix = """Begin! Remember to focus on the current query and not be overly influenced by past interactions"""
prompt = ZeroShotAgent.create_prompt(
tools,
prefix=prefix,
suffix=suffix,
input_variables=["input", "agent_scratchpad", "WEBAPP_INFO"]
)
llm_chain = LLMChain(llm=llm, prompt=prompt)
tool_names = [tool.name for tool in tools]
agent = ZeroShotAgent(llm_chain=llm_chain, allowed_tools=tool_names)
# Create the agent with a maximum number of iterations
agent = initialize_agent(
tools,
llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
max_iterations=3, # Limit the number of iterations
early_stopping_method="generate",
agent_kwargs={
"prefix": prefix,
"format_instructions": """Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question"""
}
)
def process_query(query: str) -> str:
try:
response = agent.run(input=query, WEBAPP_INFO=WEBAPP_INFO)
return response
except Exception as e:
logger.error(f"Error processing query: {str(e)}", exc_info=True)
return "I apologize, but I encountered an issue while processing your request. Could you please rephrase your question or try again later?"
@csrf_exempt
@require_http_methods(["POST"])
def chatbot_response(request):
logger.info("Chatbot response view called")
try:
data = json.loads(request.body)
user_message = data.get('message', '').strip()
conversation_history = data.get('conversation_history', [])
logger.info(f"Received message: {user_message}")
system_message = f"""
You are an AI assistant for an asset management system. Here's the current context:
1. User's latest message: "{user_message}"
2. System overview: {WEBAPP_INFO}
Respond naturally and conversationally. If the user asks about an asset or the system's capabilities, provide relevant information.
If they ask about something we don't have info on, politely explain that and offer to help with something else.
Always stay focused on asset management and the system's capabilities.
"""
messages = [SystemMessage(content=system_message)]
for msg in conversation_history[-5:]:
messages.append(HumanMessage(content=msg['user_message']))
messages.append(SystemMessage(content=msg['ai_response']))
messages.append(HumanMessage(content=user_message))
response = process_query(user_message)
conversation_history.append({
'user_message': user_message,
'ai_response': response
})
return JsonResponse({
'response': response,
'conversation_history': conversation_history
})
except json.JSONDecodeError:
logger.error("Invalid JSON received")
return JsonResponse({'error': 'We encountered an issue processing your request. Please try again.'}, status=400)
except Exception as e:
logger.error(f"An error occurred: {str(e)}", exc_info=True)
return JsonResponse({'error': 'We apologize, but we experienced a temporary issue. Please try again later.'}, status=500)
#image_views
from django.views.decorators.csrf import csrf_exempt
from django.http import JsonResponse
from django.core.files.storage import default_storage
from django.core.files.base import ContentFile
from django.conf import settings
import os
import torch
from PIL import Image
from transformers import CLIPProcessor, CLIPModel
from .models import AssetData, AssetImage
# Load the CLIP model and processor
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
def extract_features(image_path):
image = Image.open(image_path)
inputs = processor(images=image, return_tensors="pt", padding=True)
with torch.no_grad():
image_features = model.get_image_features(**inputs)
return image_features
@csrf_exempt
def chatbot_image_response(request):
if request.method == 'POST':
try:
# Get the uploaded image from the request
image_file = request.FILES.get('image')
if not image_file:
return JsonResponse({'error': 'No image file provided'}, status=400)
# Save the image temporarily
temp_path = default_storage.save('temp_image.jpg', ContentFile(image_file.read()))
full_temp_path = os.path.join(settings.MEDIA_ROOT, temp_path)
# Extract features from the uploaded image
uploaded_features = extract_features(full_temp_path)
# Find the top 3 matching images in the database
matches = []
for asset_image in AssetImage.objects.all():
db_image_path = os.path.join(settings.MEDIA_ROOT, str(asset_image.image))
db_features = extract_features(db_image_path)
similarity = torch.nn.functional.cosine_similarity(uploaded_features, db_features).item()
matches.append((asset_image, similarity))
# Sort matches by similarity (descending) and get top 3
top_matches = sorted(matches, key=lambda x: x[1], reverse=True)[:3]
if top_matches and top_matches[0][1] > 0.8:
response = "I've identified the top 3 assets that match the uploaded image. Here's the information:\n\n"
for i, (asset_image, similarity) in enumerate(top_matches, 1):
asset = asset_image.asset
image_url = request.build_absolute_uri(asset_image.image.url)
response += f"{i}. Match {i} (Similarity: {similarity:.2f})\n"
response += f" • Asset Name: {asset.asset_name}\n"
response += f" • Asset Type: {asset.asset_type}\n"
response += f" • Model: {asset.model}\n"
response += f" • Serial Number: {asset.serial_number}\n"
response += f" • Purchase Date: {asset.purchase_date}\n"
response += f" • Warranty Info: {asset.warranty_info}\n"
response += f" • Image: [View Image]({image_url})\n\n"
# Include the uploaded image URL in the response
uploaded_image_url = request.build_absolute_uri(settings.MEDIA_URL + temp_path)
response += f"Uploaded Image: [View Uploaded Image]({uploaded_image_url})\n"
else:
response = "I couldn't find any assets that closely match the uploaded image. Please make sure the image is clear and matches an asset in our database."
# Clean up the temporary file
os.remove(full_temp_path)
return JsonResponse({'response': response})
except Exception as e:
return JsonResponse({'error': str(e)}, status=500)
return JsonResponse({'error': 'Invalid request method'}, status=405)
