"""
Website Service
Handles website generation and management using DeepSite AI workflow
"""
from typing import Dict, Any
import re
from pathlib import Path
import zipfile
import sys
import asyncio
import aiofiles
# 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, setup_template, _copy_logo_files
from services.ai_service import AIService
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)
# Save final HTML file
# Sanitize third-party resource references (e.g., Font Awesome kit -> cdnjs)
html_content = self._sanitize_third_party_resources(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
"""
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
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}"
):
# 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
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, save final HTML
final_html = self.ai_service._extract_html_from_response(accumulated_content)
if final_html:
# Sanitize third-party resources
final_html = self._sanitize_third_party_resources(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!',
'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
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)