"""
Website Service
Handles website generation and management using DeepSite AI workflow
"""
from typing import Dict, Any
import re
import logging
from pathlib import Path
import zipfile
import sys
import asyncio
import aiofiles
logger = logging.getLogger(__name__)
# Add src to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from services.website_data_utils import parse_json_input, format_website_prompt, format_edit_prompt, setup_template, _copy_logo_files
from services.ai_service import AIService
from services.diff_service import DiffService
class WebsiteService:
def __init__(self):
self.generated_path = Path("templates/generated")
self.generated_path.mkdir(parents=True, exist_ok=True)
self.ai_service = AIService()
async def generate_website(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Generate website from data using DeepSite AI workflow
"""
try:
# Parse JSON input
website_data = parse_json_input(data)
# Convert to DeepSite prompt format
deepsite_prompt = format_website_prompt(website_data)
# Setup template (Phase 1 - Automated)
setup_result = await setup_template(website_data)
folder_name = setup_result['safe_folder_name']
website_path = Path(setup_result['destination_path'])
# Copy logo files if present
logo_files = await _copy_logo_files(website_data, website_path)
# Generate AI content (Phase 2 - AI Generation)
ai_response = await self.ai_service.generate_sync_response(
prompt=deepsite_prompt,
conversation_id=f"website_{folder_name}"
)
# Use AI response content directly
if ai_response.get("content"):
html_content = ai_response["content"]
else:
html_content = self._create_basic_html(website_data)
# Validate HTML is complete before saving
is_valid, error_msg = self._validate_html_complete(html_content)
if not is_valid:
raise Exception(f"Generated HTML is incomplete: {error_msg}")
# Save final HTML file
# Sanitize third-party resource references (e.g., Font Awesome kit -> cdnjs)
html_content = self._sanitize_third_party_resources(html_content)
# Inject Eatance footer
html_content = self._inject_eatance_footer(html_content)
index_file = website_path / "index.html"
await self._write_file_async(index_file, html_content)
return {
"html_content": html_content,
"folder_name": folder_name,
"website_path": str(website_path),
"logo_files": logo_files
}
except Exception as e:
raise Exception(f"Website generation failed: {str(e)}")
def _create_basic_html(self, website_data: Dict[str, Any]) -> str:
"""
Create basic HTML as fallback
"""
website_name = website_data.get('websiteName', 'Restaurant Website')
description = website_data.get('websiteDescription', 'Welcome to our restaurant!')
phone = website_data.get('restaurantPhone', '')
email = website_data.get('restaurantEmail', '')
address = website_data.get('restaurantAddress', '')
return f"""
{website_name}
{website_name}
{description}
Contact Information
{f'
Phone: {phone}
' if phone else ''}
{f'
Email: {email}
' if email else ''}
{f'
Address: {address}
' if address else ''}
"""
async def create_zip(self, website_path: Path, folder_name: str) -> str:
"""
Create ZIP file from website folder asynchronously.
Simple behavior: if downloads/.zip exists, reuse it; otherwise create it.
"""
try:
zip_path = Path("downloads") / f"{folder_name}.zip"
zip_path.parent.mkdir(exist_ok=True)
# If zip already exists, reuse it (fast path)
if zip_path.exists():
return str(zip_path)
# Create the ZIP archive asynchronously
await self._create_zip_async(website_path, zip_path)
return str(zip_path)
except Exception as e:
raise Exception(f"ZIP creation failed: {str(e)}")
async def generate_website_stream(self, data: Dict[str, Any]):
"""
Generate website with streaming response for live preview.
Supports both new generation and edit mode with SEARCH/REPLACE diffs.
"""
try:
# Check if this is an edit request
edit_mode = data.get('editMode', False)
# Parse JSON input
website_data = parse_json_input(data)
if edit_mode:
# Edit mode: use existing HTML and folder
original_html = data.get('originalHtml', '')
instructions = data.get('editInstructions', '')
folder_name = data.get('existingFolder', '')
if not original_html or not instructions:
raise Exception("Edit mode requires originalHtml and editInstructions")
if not folder_name:
raise Exception("Edit mode requires existingFolder")
# Use existing folder path
website_path = Path(f"templates/generated/{folder_name}")
if not website_path.exists():
raise Exception(f"Website folder not found: {folder_name}")
# Read the actual HTML from the saved file (most reliable source)
index_file = website_path / "index.html"
if index_file.exists():
async with aiofiles.open(index_file, 'r', encoding='utf-8') as f:
clean_original_html = await f.read()
logger.info(f"Read HTML from file: {index_file}")
else:
# Fallback to frontend's HTML if file doesn't exist
clean_original_html = self.ai_service._extract_html_from_response(original_html)
logger.warning("index.html not found, using frontend's originalHtml")
# Format edit prompt with SEARCH/REPLACE instructions
edit_system_prompt, deepsite_prompt = format_edit_prompt(clean_original_html, instructions)
else:
# Normal generation mode
# Convert to DeepSite prompt format
deepsite_prompt = format_website_prompt(website_data)
edit_system_prompt = None # Use default system prompt
# Setup template (Phase 1 - Automated)
setup_result = await setup_template(website_data)
folder_name = setup_result['safe_folder_name']
website_path = Path(setup_result['destination_path'])
# Copy logo files if present
try:
await _copy_logo_files(website_data, website_path)
except Exception:
pass # Best effort logo copying
# Stream AI response and accumulate content
accumulated_content = ""
last_render_time = 0
last_sent_index = 0
async for chunk in self.ai_service.generate_streaming_response(
prompt=deepsite_prompt,
conversation_id=f"website_{folder_name}",
system_prompt=edit_system_prompt
):
# Check if this is an error message from the AI service
if chunk.startswith("Error") or chunk.startswith("Connection lost") or chunk.startswith("Unable to connect") or chunk.startswith("Request timed out") or chunk.startswith("AI service error"):
yield {
'type': 'error',
'message': chunk
}
return # Stop processing on error
accumulated_content += chunk
# Extract HTML and send chunks for live preview (only in normal mode)
if not edit_mode:
html_content = self.ai_service._extract_html_from_response(accumulated_content)
if html_content and len(html_content) > last_sent_index:
# Only send if we have substantial HTML content
if len(html_content) > 100:
new_part = html_content[last_sent_index:]
# Throttle sending to ~5000ms
import time
now = time.time() * 1000
if now - last_render_time > 5000:
yield {
'type': 'content',
'chunk': new_part
}
last_render_time = now
last_sent_index = len(html_content)
# After streaming complete, process final HTML based on mode
if edit_mode:
# Edit mode: Apply diffs to cleaned original HTML
# Strip everything before first SEARCH block (AI explanation text)
clean_content = accumulated_content
if '<<<<<<< SEARCH' in accumulated_content:
clean_content = accumulated_content[accumulated_content.find('<<<<<<< SEARCH'):]
final_html, success, error_msg = DiffService.apply_diffs(clean_original_html, clean_content)
if not success:
yield {
'type': 'error',
'message': f'Failed to apply changes: {error_msg}'
}
return
# Validate edited HTML is complete
is_valid, validation_error = self._validate_html_complete(final_html)
if not is_valid:
yield {
'type': 'error',
'message': f'Edited HTML is incomplete: {validation_error}'
}
return
# Sanitize and inject footer
final_html = self._sanitize_third_party_resources(final_html)
final_html = self._inject_eatance_footer(final_html)
# Save edited HTML to same folder
index_file = website_path / "index.html"
await self._write_file_async(index_file, final_html)
# Send the final HTML for preview update
yield {
'type': 'content',
'chunk': final_html
}
else:
# Normal mode: Extract HTML from response
final_html = self.ai_service._extract_html_from_response(accumulated_content)
if final_html:
# Validate HTML is complete before saving
is_valid, error_msg = self._validate_html_complete(final_html)
if not is_valid:
yield {
'type': 'error',
'message': f'Generated HTML is incomplete: {error_msg}'
}
return
# Sanitize third-party resources
final_html = self._sanitize_third_party_resources(final_html)
# Inject Eatance footer
final_html = self._inject_eatance_footer(final_html)
# Save final HTML file
index_file = website_path / "index.html"
await self._write_file_async(index_file, final_html)
yield {
'type': 'complete',
'message': 'Website generation complete!' if not edit_mode else 'Changes applied successfully!',
'folder': folder_name
}
except Exception as e:
yield {
'type': 'error',
'message': f"Streaming generation failed: {str(e)}"
}
def _sanitize_third_party_resources(self, html: str) -> str:
"""Replace known external kit/script tags with safer CDN links.
Currently converts Font Awesome kit scripts to the stable cdnjs stylesheet.
"""
if not html:
return html
# Remove any occurrences
html = re.sub(r"]*src=[\"']https?:\\/\\/kit\.fontawesome\.com\/[\w.-]+\.js[\"'][^>]*>\s*", "", html, flags=re.IGNORECASE)
# Replace any fontawesome CDN script/link with cdnjs stylesheet link if not already present
if 'cdnjs.cloudflare.com/ajax/libs/font-awesome' not in html:
# Insert cdnjs stylesheet into the head if a head tag exists
if '' in html.lower():
html = re.sub(r"(?i)(]*>)", r"\1\n ", html, count=1)
return html
def _validate_html_complete(self, html_content: str) -> tuple[bool, str]:
"""
Validate that HTML content is complete and well-formed.
Returns:
tuple: (is_valid, error_message)
"""
import logging
logger = logging.getLogger(__name__)
if not html_content or len(html_content.strip()) < 100:
error_msg = "HTML content is too short or empty"
logger.error(f"HTML validation failed: {error_msg} (length: {len(html_content) if html_content else 0})")
return False, error_msg
# Check for truncation FIRST (check for unclosed tags at the end - common in truncated responses)
last_100_chars = html_content[-100:].strip()
if last_100_chars and not last_100_chars.endswith('>'):
# Check if it ends with an incomplete tag
if '<' in last_100_chars and '>' not in last_100_chars[last_100_chars.rindex('<'):]:
error_msg = "HTML appears to be truncated (incomplete tag at end)"
logger.error(f"HTML validation failed: {error_msg}, last 100 chars: {last_100_chars}")
return False, error_msg
# Check for essential HTML structure
required_tags = ['', '', '']
missing_tags = [tag for tag in required_tags if tag.lower() not in html_content.lower()]
if missing_tags:
error_msg = f"Missing required HTML tags: {', '.join(missing_tags)}"
logger.error(f"HTML validation failed: {error_msg}")
return False, error_msg
logger.info("HTML validation passed")
return True, ""
def _inject_eatance_footer(self, html_content: str) -> str:
"""
Inject 'Powered by Eatance' footer into HTML content
"""
# Check if footer already exists to avoid duplicates
if 'Powered by Eatance' in html_content:
return html_content
# Create the Eatance footer HTML
eatance_footer = '''
'''
# Inject before closing body tag, or at the end if no body tag
if '' in html_content:
return html_content.replace('
', eatance_footer + '\n')
else:
return html_content + eatance_footer
async def _write_file_async(self, file_path: Path, content: str) -> None:
"""Write content to file asynchronously."""
async with aiofiles.open(file_path, 'w', encoding='utf-8') as f:
await f.write(content)
async def _create_zip_async(self, source_path: Path, zip_path: Path) -> None:
"""Create ZIP file asynchronously."""
def _create_zip():
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for file_path in source_path.rglob('*'):
if file_path.is_file():
arcname = file_path.relative_to(source_path)
zipf.write(file_path, arcname)
# Run the ZIP creation in a thread pool to avoid blocking
await asyncio.get_event_loop().run_in_executor(None, _create_zip)