4 changed files with 2590 additions and 3 deletions
+15
View File
@@ -0,0 +1,15 @@
# Python
__pycache__/
*.py[cod]
*$py.class
venv/
# Uploads
uploads/
# IDE
.vscode/
.idea/
# OS
.DS_Store
-3
View File
@@ -1,3 +0,0 @@
# site-survey-ai
IT Site Survey AI Application
+980
View File
@@ -0,0 +1,980 @@
#!/usr/bin/env python3
"""
IT Site Survey AI Flask Application
Creates surveys for client sites and uses AI to recommend network setups.
Enhanced with PDF export, photo uploads, and selectable Ollama models.
"""
import os
import json
import uuid
import base64
import requests
from datetime import datetime
from flask import Flask, jsonify, request, send_from_directory, Response, stream_with_context, send_file
from flask_cors import CORS
from reportlab.lib import colors
from reportlab.lib.pagesizes import letter
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import inch
import io
import time
app = Flask(__name__, static_url_path='', static_folder='.')
CORS(app)
OLLAMA_BASE = os.environ.get('OLLAMA_URL', 'http://192.168.19.25:11434')
UPLOAD_FOLDER = os.environ.get('UPLOAD_FOLDER', '/tmp/site-survey-uploads')
# Ensure upload folder exists
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
# In-memory storage for surveys
surveys_db = {}
survey_responses_db = {}
DEFAULT_MODEL = 'kimi-k2.6:cloud'
# Default survey template
DEFAULT_SURVEY_TEMPLATE = {
"id": str(uuid.uuid4()),
"name": "Comprehensive Network Infrastructure Site Survey",
"description": "Detailed survey to assess network infrastructure, cabling, and IT requirements for accurate implementation planning",
"questions": [
# Site Topology & Layout
{
"id": "q1",
"category": "Site Topology & Layout",
"question": "What is the total square footage of the site?",
"type": "text",
"placeholder": "e.g., 5000 sq ft"
},
{
"id": "q2",
"category": "Site Topology & Layout",
"question": "How many floors does the building have?",
"type": "number",
"placeholder": "e.g., 3"
},
{
"id": "q3",
"category": "Site Topology & Layout",
"question": "Provide a detailed description of the site layout (open plan, offices, cubicles, etc.)",
"type": "textarea",
"placeholder": "e.g., Open floor plan with 20 cubicles, 5 private offices, 2 conference rooms..."
},
{
"id": "q4",
"category": "Site Topology & Layout",
"question": "Are there any physical barriers that may affect cable runs?",
"type": "multiselect",
"options": ["Concrete walls", "Elevator shafts", "Stairwells", "Firewalls", "Historic building restrictions", "Underground levels", "None"]
},
# Network Closet & Equipment Locations
{
"id": "q5",
"category": "Network Closet & Equipment",
"question": "How many network closets (MDF/IDF) are planned or exist?",
"type": "number",
"placeholder": "e.g., 2"
},
{
"id": "q6",
"category": "Network Closet & Equipment",
"question": "Describe the current or planned network closet locations",
"type": "textarea",
"placeholder": "e.g., MDF in basement, IDF on floors 2 and 4..."
},
{
"id": "q7",
"category": "Network Closet & Equipment",
"question": "What is the power capacity available in network closets (UPS, circuits, etc.)?",
"type": "text",
"placeholder": "e.g., 2x 20A circuits, UPS backup planned"
},
{
"id": "q8",
"category": "Network Closet & Equipment",
"question": "Are there equipment racks installed or planned?",
"type": "select",
"options": ["Yes, fully equipped racks", "Yes, partial racks", "Racks needed", "Wall-mounted only", "No racks planned"]
},
# Cabling Infrastructure
{
"id": "q9",
"category": "Cabling Infrastructure",
"question": "What type of network cabling is currently installed?",
"type": "select",
"options": ["Cat6a (10GbE capable)", "Cat6 (1GbE)", "Cat5e (100Mb-1GbE)", "Cat5 or older", "Fiber only", "Mixed/Unknown", "No existing cabling"]
},
{
"id": "q10",
"category": "Cabling Infrastructure",
"question": "How many total network drops (ports) are needed?",
"type": "number",
"placeholder": "e.g., 150"
},
{
"id": "q11",
"category": "Cabling Infrastructure",
"question": "What is the estimated average cable run length?",
"type": "text",
"placeholder": "e.g., 150 feet average, longest run 300 feet"
},
{
"id": "q12",
"category": "Cabling Infrastructure",
"question": "Are there specific cabling requirements?",
"type": "multiselect",
"options": ["Plenum-rated cable required", "Outdoor/rated cable needed", "Shielded cable (STP)", "Fiber backbone needed", "Conduit/tray already installed", "None"]
},
# Network Equipment & Hardware
{
"id": "q13",
"category": "Network Equipment",
"question": "What network equipment is currently installed?",
"type": "multiselect",
"options": ["Router/Firewall", "Managed switches", "Unmanaged switches", "Wireless access points", "Server(s)", "UPS/Backup power", "Patch panels", "None - greenfield"]
},
{
"id": "q14",
"category": "Network Equipment",
"question": "What type of network switches are needed?",
"type": "select",
"options": ["Fully managed (Layer 3)", "Managed (Layer 2)", "Smart/Web-managed", "Unmanaged", "PoE required", "Not sure"]
},
{
"id": "q15",
"category": "Network Equipment",
"question": "How many wireless access points are needed?",
"type": "number",
"placeholder": "e.g., 8"
},
# Power Requirements
{
"id": "q16",
"category": "Power Requirements",
"question": "What is the total estimated power consumption for network equipment?",
"type": "text",
"placeholder": "e.g., 2000W estimated"
},
{
"id": "q17",
"category": "Power Requirements",
"question": "Are PoE (Power over Ethernet) devices planned?",
"type": "multiselect",
"options": ["IP Phones", "Wireless APs", "IP Cameras", "Access control", "Digital signage", "IoT sensors", "None"]
},
{
"id": "q18",
"category": "Power Requirements",
"question": "What is the UPS/backup power requirement?",
"type": "select",
"options": ["Full backup (4+ hours)", "Limited backup (1-2 hours)", "Graceful shutdown only", "None", "Not sure"]
},
# User & Device Requirements
{
"id": "q19",
"category": "User & Device Requirements",
"question": "How many employees will be onsite?",
"type": "number",
"placeholder": "e.g., 50"
},
{
"id": "q20",
"category": "User & Device Requirements",
"question": "How many concurrent devices do you expect (including phones, laptops, tablets, IoT)?",
"type": "number",
"placeholder": "e.g., 150"
},
{
"id": "q21",
"category": "User & Device Requirements",
"question": "What types of network activities will users be doing?",
"type": "multiselect",
"options": ["Email/Web Browsing", "Video conferencing", "File sharing", "Cloud applications", "VoIP/Phone calls", "Streaming/Media", "Remote desktop", "Development/Testing", "Guest WiFi"]
},
# Security & Compliance
{
"id": "q22",
"category": "Security & Compliance",
"question": "What security compliance needs do you have?",
"type": "multiselect",
"options": ["HIPAA", "PCI-DSS", "SOC 2", "ISO 27001", "GDPR", "NIST", "None/General security"]
},
{
"id": "q23",
"category": "Security & Compliance",
"question": "Do you need network segmentation (VLANs)?",
"type": "multiselect",
"options": ["Corporate network", "Guest network", "IoT network", "Management network", "Voice/VLAN", "Video surveillance VLAN", "No segmentation needed"]
},
{
"id": "q24",
"category": "Security & Compliance",
"question": "What security features are required?",
"type": "multiselect",
"options": ["Firewall", "Intrusion Detection/Prevention", "Content filtering", "VPN access", "Network access control (NAC)", "Endpoint security", "None"]
},
# Internet & WAN
{
"id": "q25",
"category": "Internet & WAN",
"question": "What internet connectivity is available or planned?",
"type": "multiselect",
"options": ["Fiber", "Cable/DSL", "Fixed wireless", "5G/Cellular backup", "MPLS", "SD-WAN", "Multiple ISPs", "Not sure"]
},
{
"id": "q26",
"category": "Internet & WAN",
"question": "What is the required internet bandwidth?",
"type": "text",
"placeholder": "e.g., 1Gbps download / 500Mbps upload"
},
{
"id": "q27",
"category": "Internet & WAN",
"question": "Do you have remote sites that need to connect?",
"type": "select",
"options": ["Yes - Site-to-site VPN needed", "Yes - SD-WAN preferred", "Yes - MPLS", "No - Single site only", "Maybe in future"]
},
# Existing Infrastructure Assessment
{
"id": "q28",
"category": "Existing Infrastructure",
"question": "Is there existing network infrastructure to assess?",
"type": "select",
"options": ["Yes - Full assessment needed", "Yes - Partial upgrade", "No - Complete new build", "Not sure"]
},
{
"id": "q29",
"category": "Existing Infrastructure",
"question": "Describe any existing network infrastructure issues or concerns",
"type": "textarea",
"placeholder": "e.g., Slow WiFi in conference rooms, dead zones, outdated equipment..."
},
{
"id": "q30",
"category": "Existing Infrastructure",
"question": "Are there any known performance issues?",
"type": "multiselect",
"options": ["Slow internet", "WiFi dead zones", "Network congestion", "Frequent disconnections", "Latency issues", "None known"]
},
# Budget & Timeline
{
"id": "q31",
"category": "Budget & Timeline",
"question": "What is your approximate budget range?",
"type": "select",
"options": ["Under $10,000", "$10,000 - $25,000", "$25,000 - $50,000", "$50,000 - $100,000", "$100,000+", "Not established yet"]
},
{
"id": "q32",
"category": "Budget & Timeline",
"question": "When do you need this completed?",
"type": "select",
"options": ["ASAP - Emergency", "Within 2 weeks", "Within 1 month", "Within 3 months", "Within 6 months", "Flexible"]
},
{
"id": "q33",
"category": "Budget & Timeline",
"question": "Who will be responsible for ongoing network management?",
"type": "select",
"options": ["Internal IT team", "Managed Service Provider (MSP)", "Hybrid approach", "Not decided", "Outsourced"]
},
# Additional Details
{
"id": "q34",
"category": "Additional Details",
"question": "Upload photos of the site (network closets, cable runs, equipment locations)",
"type": "file",
"placeholder": "Upload images for visual assessment"
},
{
"id": "q35",
"category": "Additional Details",
"question": "Any other specific requirements, concerns, or notes?",
"type": "textarea",
"placeholder": "e.g., Need outdoor WiFi coverage, specific vendor preferences, regulatory requirements, future expansion plans..."
}
]
}
def call_ollama(prompt, model=None, stream=False):
"""Call Ollama API with the given prompt."""
model = model or DEFAULT_MODEL
try:
if stream:
resp = requests.post(
f'{OLLAMA_BASE}/api/generate',
json={'model': model, 'prompt': prompt, 'stream': True},
stream=True,
timeout=120
)
return resp
else:
resp = requests.post(
f'{OLLAMA_BASE}/api/generate',
json={'model': model, 'prompt': prompt, 'stream': False},
timeout=120
)
return resp.json().get('response', '')
except Exception as e:
return f'Error: {str(e)}'
def get_ollama_models():
"""Fetch list of available models from Ollama."""
try:
resp = requests.get(f'{OLLAMA_BASE}/api/tags', timeout=5)
if resp.status_code == 200:
models = resp.json().get('models', [])
return [m.get('name', '') for m in models if m.get('name')]
except:
pass
return [DEFAULT_MODEL]
@app.route('/')
def index():
return send_from_directory('.', 'index.html')
@app.route('/<path:path>')
def serve_static(path):
return send_from_directory('.', path)
@app.route('/api/health')
def health_check():
try:
r = requests.get(f'{OLLAMA_BASE}/', timeout=3)
ollama_online = r.status_code == 200
except:
ollama_online = False
return jsonify({
'status': 'ok',
'service': 'site-survey-ai',
'ollama_connected': ollama_online,
'timestamp': datetime.utcnow().isoformat()
})
@app.route('/api/ollama-status')
def ollama_status():
try:
r = requests.get(f'{OLLAMA_BASE}/', timeout=3)
return jsonify({'online': r.status_code == 200})
except:
return jsonify({'online': False})
@app.route('/api/models')
def list_models():
"""Return list of available Ollama models."""
models = get_ollama_models()
return jsonify({
'models': models,
'default': DEFAULT_MODEL
})
@app.route('/api/surveys/template')
def get_template():
return jsonify({'template': DEFAULT_SURVEY_TEMPLATE})
# Delete implementation functions
def delete_survey_impl(survey_id):
"""Delete a specific survey by ID"""
if survey_id in surveys_db:
# Clean up uploaded photos if they exist
survey = surveys_db[survey_id]
if 'responses' in survey:
for response in survey['responses']:
if 'photos' in response:
for photo in response['photos']:
photo_path = os.path.join(UPLOAD_FOLDER, photo['filename'])
if os.path.exists(photo_path):
try:
os.remove(photo_path)
except:
pass
del surveys_db[survey_id]
return jsonify({'success': True, 'message': 'Survey deleted successfully'})
return jsonify({'success': False, 'error': 'Survey not found'}), 404
def clear_all_surveys_impl():
"""Delete all surveys"""
global surveys_db
count = len(surveys_db)
# Clean up uploaded photos
for survey_id, survey in surveys_db.items():
if 'responses' in survey:
for response in survey['responses']:
if 'photos' in response:
for photo in response['photos']:
photo_path = os.path.join(UPLOAD_FOLDER, photo['filename'])
if os.path.exists(photo_path):
try:
os.remove(photo_path)
except:
pass
surveys_db = {}
return jsonify({'success': True, 'message': f'{count} surveys deleted successfully'})
@app.route('/api/surveys', methods=['GET', 'DELETE'])
def get_surveys():
if request.method == 'DELETE':
return clear_all_surveys_impl()
return jsonify({
'surveys': list(surveys_db.values())
})
@app.route('/api/surveys', methods=['POST'])
def create_survey():
data = request.get_json()
survey_id = str(uuid.uuid4())
survey = {
'id': survey_id,
'name': data.get('name', 'New Survey'),
'description': data.get('description', ''),
'client_name': data.get('client_name', ''),
'site_name': data.get('site_name', ''),
'questions': data.get('questions', DEFAULT_SURVEY_TEMPLATE['questions']),
'created_at': datetime.utcnow().isoformat(),
'status': 'active',
'responses': []
}
# If responses are provided, add them as the first response
responses_data = data.get('responses')
if responses_data and isinstance(responses_data, dict) and len(responses_data) > 0:
response_entry = {
'id': str(uuid.uuid4()),
'survey_id': survey_id,
'responses': responses_data,
'photos': [],
'analyzed': False,
'analysis': None,
'submitter_name': 'Survey Creator',
'created_at': datetime.utcnow().isoformat()
}
survey['responses'].append(response_entry)
surveys_db[survey_id] = survey
return jsonify({'survey': survey}), 201
@app.route('/api/surveys/<survey_id>', methods=['GET', 'DELETE'])
def get_survey(survey_id):
if request.method == 'DELETE':
return delete_survey_impl(survey_id)
survey = surveys_db.get(survey_id)
if not survey:
return jsonify({'error': 'Survey not found'}), 404
return jsonify({'survey': survey})
@app.route('/api/surveys/<survey_id>/responses', methods=['POST'])
def submit_response(survey_id):
survey = surveys_db.get(survey_id)
if not survey:
return jsonify({'error': 'Survey not found'}), 404
data = request.get_json()
# Handle photo uploads (base64 encoded)
photos = data.get('photos', [])
photo_filenames = []
for i, photo_data in enumerate(photos):
if ',' in photo_data:
# Handle data URL format: data:image/jpeg;base64,/9j/4AAQ...
header, encoded = photo_data.split(',', 1)
try:
image_bytes = base64.b64decode(encoded)
ext = 'jpg'
if 'png' in header.lower():
ext = 'png'
elif 'webp' in header.lower():
ext = 'webp'
filename = f"{survey_id}_{uuid.uuid4().hex[:8]}.{ext}"
filepath = os.path.join(UPLOAD_FOLDER, filename)
with open(filepath, 'wb') as f:
f.write(image_bytes)
photo_filenames.append(filename)
except Exception as e:
print(f"Photo decode error: {e}")
response_id = str(uuid.uuid4())
response = {
'id': response_id,
'survey_id': survey_id,
'submitted_by': data.get('submitted_by', 'Anonymous'),
'responses': data.get('responses', {}),
'model_used': data.get('model', DEFAULT_MODEL),
'photos': photo_filenames,
'submitted_at': datetime.utcnow().isoformat()
}
survey['responses'].append(response)
survey_responses_db[response_id] = response
return jsonify({'response': response}), 201
@app.route('/api/photos/<survey_id>/<filename>')
def serve_photo(survey_id, filename):
"""Serve an uploaded photo."""
# Security: only serve files that belong to this survey
safe_filename = os.path.basename(filename)
filepath = os.path.join(UPLOAD_FOLDER, safe_filename)
if not os.path.exists(filepath):
return jsonify({'error': 'Photo not found'}), 404
return send_file(filepath)
@app.route('/api/surveys/<survey_id>/analyze', methods=['POST'])
def analyze_survey(survey_id):
"""Analyze survey responses with AI to generate recommendations."""
survey = surveys_db.get(survey_id)
if not survey:
return jsonify({'error': 'Survey not found'}), 404
if not survey['responses']:
return jsonify({'error': 'No responses to analyze'}), 400
data = request.get_json() or {}
model = data.get('model', survey['responses'][-1].get('model_used', DEFAULT_MODEL))
latest_response = survey['responses'][-1]
responses = latest_response['responses']
photos = latest_response.get('photos', [])
context = f"""
IT Infrastructure Site Survey Results:
Client: {survey.get('client_name', 'Unknown')}
Site: {survey.get('site_name', 'Unknown')}
AI Model Used: {model}
Survey Responses:
"""
for q in survey['questions']:
qid = q['id']
answer = responses.get(qid, 'Not answered')
context += f"\n{q['category']} - {q['question']}\nAnswer: {answer}\n"
# Include photo descriptions if available
if photos:
photo_context = f"\n\nPHOTOS ATTACHED ({len(photos)} total):\n"
for i, photo in enumerate(photos):
photo_context += f"Photo {i+1}: [see uploaded image - {photo}]\n"
context += photo_context
prompt = f"""You are a senior network architect. Based on this site survey, provide a concise IT infrastructure recommendation.
{context}
Provide:
1. **EXECUTIVE SUMMARY** - Brief overview
2. **NETWORK DESIGN** - Recommended topology and VLANs
3. **HARDWARE NEEDED** - Router, switches, access points
4. **SECURITY** - Key security measures
5. **ESTIMATED COSTS** - Rough budget estimate
6. **TIMELINE** - Implementation phases
7. **PHOTO NOTES** - What the photos show (if any)
8. **NEXT STEPS** - Priority actions
Keep responses concise and actionable.
- Top 3 immediate actions needed
Format as a professional proposal suitable for client presentation.
"""
def generate():
try:
resp = requests.post(
f'{OLLAMA_BASE}/api/generate',
json={'model': model, 'prompt': prompt, 'stream': True},
stream=True,
timeout=180
)
for line in resp.iter_lines():
if line:
yield line.decode('utf-8') + '\n'
except Exception as e:
error_chunk = json.dumps({'error': str(e), 'done': True})
yield error_chunk + '\n'
return Response(stream_with_context(generate()), mimetype='application/x-ndjson')
@app.route('/api/surveys/<survey_id>/analyze-with-photos', methods=['POST'])
def analyze_with_photos(survey_id):
"""Analyze survey responses WITH photos sent directly (for vision-capable models)."""
survey = surveys_db.get(survey_id)
if not survey:
return jsonify({'error': 'Survey not found'}), 404
if not survey['responses']:
return jsonify({'error': 'No responses to analyze'}), 400
data = request.get_json() or {}
model = data.get('model', survey['responses'][-1].get('model_used', DEFAULT_MODEL))
latest_response = survey['responses'][-1]
responses = latest_response['responses']
photo_urls = latest_response.get('photos', [])
context = f"""
IT Infrastructure Site Survey Results:
Client: {survey.get('client_name', 'Unknown')}
Site: {survey.get('site_name', 'Unknown')}
AI Model Used: {model}
Survey Responses:
"""
for q in survey['questions']:
qid = q['id']
answer = responses.get(qid, 'Not answered')
context += f"\n{q['category']} - {q['question']}\nAnswer: {answer}\n"
if photo_urls:
context += f"\n\nPHOTOS ATTACHED ({len(photo_urls)} total):\n"
for i, photo in enumerate(photo_urls):
context += f"Photo {i+1}: {photo}\n"
prompt = f"""You are a senior network architect and IT infrastructure consultant. Based on the following site survey AND any attached photos, provide a comprehensive network infrastructure recommendation.
{context}
Generate a detailed IT infrastructure recommendation report including:
1. **EXECUTIVE SUMMARY**
2. **NETWORK TOPOLOGY RECOMMENDATION**
3. **HARDWARE RECOMMENDATIONS**
4. **INTERNET CONNECTIVITY**
5. **SECURITY RECOMMENDATIONS**
6. **ESTIMATED COSTS**
7. **IMPLEMENTATION TIMELINE**
8. **PHOTO ASSESSMENT** - Describe what you observe in the photos relevant to infrastructure. Note physical constraints, wiring, equipment, room layouts.
9. **PRIORITY ACTIONS**
Format as a professional proposal suitable for client presentation.
"""
def generate():
try:
resp = requests.post(
f'{OLLAMA_BASE}/api/generate',
json={'model': model, 'prompt': prompt, 'stream': True},
stream=True,
timeout=180
)
for line in resp.iter_lines():
if line:
yield line.decode('utf-8') + '\n'
except Exception as e:
error_chunk = json.dumps({'error': str(e), 'done': True})
yield error_chunk + '\n'
return Response(stream_with_context(generate()), mimetype='application/x-ndjson')
@app.route('/api/surveys/<survey_id>/generate-quote', methods=['POST'])
def generate_quote(survey_id):
"""Generate a service quote based on survey responses."""
survey = surveys_db.get(survey_id)
if not survey or not survey['responses']:
return jsonify({'error': 'No survey data available'}), 400
latest_response = survey['responses'][-1]
responses = latest_response['responses']
context = f"""
Site Survey Data for Quote Generation:
Client: {survey.get('client_name', 'Unknown')}
Site: {survey.get('site_name', 'Unknown')}
"""
for q in survey['questions']:
qid = q['id']
answer = responses.get(qid, 'Not answered')
context += f"\n{q['category']}: {answer}"
prompt = f"""You are an IT services sales consultant. Based on the following site survey, create a professional service quote for network infrastructure setup.
{context}
Generate a professional service quote in the following format:
**IT INFRASTRUCTURE SERVICES QUOTE**
**For: {survey.get('client_name', 'Client')}**
**Site: {survey.get('site_name', 'Site')}**
**Date: {datetime.now().strftime('%B %d, %Y')}**
**SCOPE OF WORK:**
[List the specific services needed based on the survey - network design, equipment procurement, installation, configuration, testing]
**RECOMMENDED SOLUTION:**
1. [Category - e.g., Network Design & Planning] - $[Price]
- [Detailed description of what's included]
2. [Category - e.g., Equipment & Hardware] - $[Price]
- [List of equipment and costs]
3. [Category - e.g., Installation & Configuration] - $[Price]
- [Labor hours and rates]
4. [Category - e.g., Security Implementation] - $[Price]
- [Security services included]
5. [Category - e.g., Testing & Documentation] - $[Price]
- [Testing procedures and documentation]
**INVESTMENT SUMMARY:**
- Initial Assessment & Design: $[Amount]
- Equipment/Hardware: $[Amount]
- Labor/Installation: $[Amount]
- Training & Documentation: $[Amount]
- **Total Project Investment: $[Amount]**
**OPTIONAL ONGOING SERVICES:**
- Monthly Management & Monitoring: $[Amount]/month
- Annual Maintenance: $[Amount]/year
- Support Contract: $[Amount]/month
**TIMELINE:** [Timeframe]
- Week 1-2: Design & Procurement
- Week 3-4: Installation
- Week 5: Testing & Training
**TERMS:**
- 50% deposit required to begin
- Net 15 on completion
- Quote valid for 30 days
This is an estimate based on the initial site survey. Final pricing will be confirmed after detailed site assessment.
"""
response = call_ollama(prompt, stream=False)
return jsonify({'quote': response})
@app.route('/api/surveys/<survey_id>/export/pdf', methods=['POST'])
def export_pdf(survey_id):
"""Export survey report to PDF."""
survey = surveys_db.get(survey_id)
if not survey or not survey['responses']:
return jsonify({'error': 'No survey data available'}), 400
latest_response = survey['responses'][-1]
responses = latest_response['responses']
buffer = io.BytesIO()
doc = SimpleDocTemplate(buffer, pagesize=letter,
rightMargin=72, leftMargin=72,
topMargin=72, bottomMargin=18)
elements = []
styles = getSampleStyleSheet()
title_style = ParagraphStyle(
'CustomTitle',
parent=styles['Heading1'],
fontSize=24,
textColor=colors.HexColor('#1f6feb'),
spaceAfter=30
)
heading_style = ParagraphStyle(
'CustomHeading',
parent=styles['Heading2'],
fontSize=16,
textColor=colors.HexColor('#58a6ff'),
spaceAfter=12,
spaceBefore=12
)
# Title
elements.append(Paragraph(f"IT Site Survey Report", title_style))
elements.append(Paragraph(f"Client: {survey.get('client_name', 'Unknown')}", styles['Normal']))
elements.append(Paragraph(f"Site: {survey.get('site_name', 'Unknown')}", styles['Normal']))
elements.append(Paragraph(f"Date: {datetime.now().strftime('%B %d, %Y')}", styles['Normal']))
elements.append(Spacer(1, 0.3*inch))
# Photos section
photos = latest_response.get('photos', [])
if photos:
elements.append(Paragraph("Attached Photos", heading_style))
elements.append(Paragraph(f"{len(photos)} photo(s) uploaded with this survey.", styles['Normal']))
elements.append(Spacer(1, 0.2*inch))
# Survey Responses
elements.append(Paragraph("Survey Responses", heading_style))
for q in survey['questions']:
qid = q['id']
answer = responses.get(qid, 'Not answered')
if isinstance(answer, list):
answer = ', '.join(answer)
elements.append(Paragraph(f"<b>{q['category']}</b>", styles['Heading3']))
elements.append(Paragraph(q['question'], styles['Normal']))
elements.append(Paragraph(f"Answer: {answer}", styles['Normal']))
elements.append(Spacer(1, 0.1*inch))
doc.build(elements)
buffer.seek(0)
return send_file(
buffer,
as_attachment=True,
download_name=f'Site_Survey_{survey.get("client_name", "Client").replace(" ", "_")}_{datetime.now().strftime("%Y%m%d")}.pdf',
mimetype='application/pdf'
)
@app.route('/api/surveys/<survey_id>/export/text', methods=['POST'])
def export_text(survey_id):
"""Export survey report to formatted text."""
survey = surveys_db.get(survey_id)
if not survey or not survey['responses']:
return jsonify({'error': 'No survey data available'}), 400
latest_response = survey['responses'][-1]
responses = latest_response['responses']
photos = latest_response.get('photos', [])
text_report = f"""IT SITE SURVEY REPORT
{'=' * 60}
Client: {survey.get('client_name', 'Unknown')}
Site: {survey.get('site_name', 'Unknown')}
Date: {datetime.now().strftime('%B %d, %Y')}
Photos: {len(photos)} attached
{'=' * 60}
SURVEY RESPONSES:
{'=' * 60}
"""
for q in survey['questions']:
qid = q['id']
answer = responses.get(qid, 'Not answered')
if isinstance(answer, list):
answer = ', '.join(answer)
text_report += f"""
{q['category']}:
Question: {q['question']}
Answer: {answer}
"""
text_report += f"""
{'=' * 60}
Report generated by IT Site Survey AI
{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
"""
buffer = io.BytesIO(text_report.encode('utf-8'))
return send_file(
buffer,
as_attachment=True,
download_name=f'Site_Survey_{survey.get("client_name", "Client").replace(" ", "_")}_{datetime.now().strftime("%Y%m%d")}.txt',
mimetype='text/plain'
)
@app.route('/api/surveys/<survey_id>/export/email', methods=['POST'])
def export_email(survey_id):
"""Export survey report formatted for email."""
survey = surveys_db.get(survey_id)
if not survey or not survey['responses']:
return jsonify({'error': 'No survey data available'}), 400
latest_response = survey['responses'][-1]
responses = latest_response['responses']
email_body = f"""
<h2>IT Site Survey Report - {survey.get('client_name', 'Client')}</h2>
<p><strong>Site:</strong> {survey.get('site_name', 'Unknown')}</p>
<p><strong>Date:</strong> {datetime.now().strftime('%B %d, %Y')}</p>
<p><strong>Photos:</strong> {len(latest_response.get('photos', []))} attached</p>
<hr>
<h3>Survey Responses:</h3>
<table border="1" cellpadding="5" cellspacing="0" style="border-collapse: collapse;">
<tr style="background-color: #f0f0f0;">
<th>Category</th>
<th>Question</th>
<th>Answer</th>
</tr>
"""
for q in survey['questions']:
qid = q['id']
answer = responses.get(qid, 'Not answered')
if isinstance(answer, list):
answer = ', '.join(answer)
email_body += f"""
<tr>
<td>{q['category']}</td>
<td>{q['question']}</td>
<td>{answer}</td>
</tr>
"""
email_body += f"""
</table>
<p>---<br>
Report generated by IT Site Survey AI<br>
{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
"""
return jsonify({'email_body': email_body})
if __name__ == '__main__':
port = int(os.environ.get('PORT', 3003))
print("───────────────────────────────────────")
print(" 🏢 IT Site Survey AI")
print(f" Running on http://0.0.0.0:{port}")
print(f" Ollama: {OLLAMA_BASE}")
print(f" Upload folder: {UPLOAD_FOLDER}")
print("───────────────────────────────────────")
app.run(host='0.0.0.0', port=port, debug=False, use_reloader=False)
# Delete endpoints
+1595
View File
File diff suppressed because it is too large Load Diff