import base64
import importlib
import json
import logging
import os
import random
import re
import time
import uuid
from collections.abc import Generator
from typing import Any
import httpx
from dotenv import load_dotenv
from flask import Flask, Request, Response, request, stream_with_context
lxml_html: Any = importlib.import_module("lxml.html")
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)
load_dotenv()
app = Flask(__name__)
API_URL = os.getenv("OPENAI_API_URL", "https://api.openai.com/v1/chat/completions")
API_KEY = os.getenv("OPENAI_API_KEY", "")
MODEL = os.getenv("MODEL", "gpt-4o-mini")
logger.info(f"Using model: {MODEL}")
logger.info(f"API URL: {API_URL}")
logger.info(f"API key present: {bool(API_KEY)}")
LOADING_PAGE = """
Generating your page...
Crafting your page...
The AI is designing something unique for you
Characters: 0
"""
def call_llm(prompt: str, system: str = "You are a helpful assistant.") -> str:
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
payload: dict[str, Any] = {
"model": MODEL,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
"temperature": 0.7,
"stream": False,
}
try:
with httpx.Client(timeout=300) as client:
resp = client.post(API_URL, headers=headers, json=payload)
resp.raise_for_status()
data: dict[str, Any] = resp.json()
return data["choices"][0]["message"]["content"]
except BaseException as e: # noqa: BLE001
logger.error(f"call_llm failed: {e}")
return ""
def validate_html(html: str) -> list[str]:
errors: list[str] = []
if not html.strip().lower().startswith(" tag, found {len(html_tags)}")
if len(body_tags) != 1:
errors.append(f"Expected 1 tag, found {len(body_tags)}")
if len(head_tags) != 1:
errors.append(f"Expected 1 tag, found {len(head_tags)}")
except BaseException as e: # noqa: BLE001
errors.append(f"HTML parse error: {e}")
style_blocks = re.findall(r"]*>(.*?)", html, re.DOTALL | re.IGNORECASE)
for i, block in enumerate(style_blocks):
if block.count("{") != block.count("}"):
errors.append(f"CSS block {i + 1}: mismatched braces")
script_blocks = re.findall(r"]*>(.*?)", html, re.DOTALL | re.IGNORECASE)
for i, block in enumerate(script_blocks):
if block.count("{") != block.count("}"):
errors.append(f"JS block {i + 1}: mismatched braces")
if block.count("(") != block.count(")"):
errors.append(f"JS block {i + 1}: mismatched parentheses")
if block.count("[") != block.count("]"):
errors.append(f"JS block {i + 1}: mismatched brackets")
if script_blocks:
js_text = "\n".join(script_blocks)
review = call_llm(
f"Review this JavaScript for runtime errors (undefined variables, wrong selectors, syntax issues). List each issue concisely. If no issues, say 'No issues found.'\n\n\n{js_text}\n"
)
if review and "no issue" not in review.lower():
for line in review.strip().split("\n"):
line = line.strip()
if line and not line.startswith("<"):
errors.append(f"JS review: {line}")
return errors
def fix_html(html: str, errors: list[str]) -> str:
error_text = "\n".join(f"- {e}" for e in errors)
prompt = f"""Fix this HTML document. Issues found:
{error_text}
Rules:
- Return ONLY the complete corrected HTML document ( ... )
- Do NOT wrap in markdown code blocks
- Do NOT add explanations
- Preserve the design and content, only fix errors
HTML to fix:
{html}"""
fixed = call_llm(prompt, system="You are an expert HTML/CSS/JS developer. You fix broken code.")
fixed = re.sub(r"^```html?\s*", "", fixed.strip(), flags=re.MULTILINE)
fixed = re.sub(r"\s*```$", "", fixed.strip(), flags=re.MULTILINE)
if fixed and " str:
seen: list[str] = []
for e in errors:
short = e[:60]
if short not in seen:
seen.append(short)
if len(seen) >= 3:
break
return ", ".join(seen)
def get_user_context(req: Request) -> dict[str, Any]:
user_agent = req.headers.get("User-Agent", "")
lang = req.headers.get("Accept-Language", "")
ip = req.remote_addr or ""
path = req.path
referer = req.headers.get("Referer", "")
cookies = dict(req.cookies)
return {
"user_agent": user_agent,
"language": lang,
"ip": ip,
"path": path,
"referer": referer,
"cookies": cookies,
}
THEMES = [
("fake_store", "A quirky online store selling unusual products (e.g., bottled laughter, cloud-shaped pillows)"),
("fake_store", "A vintage record store homepage with vinyl listings and album reviews"),
("fake_store", "A plant shop selling exotic and rare houseplants"),
("fake_store", "A retro gaming store selling refurbished consoles and cartridges"),
("fake_store", "A boutique selling handmade candles with weird scents"),
("fake_store", "A pet supply store for unusual pets (lizards, tarantulas, etc.)"),
("fake_homepage", "A personal portfolio site for a fictional photographer"),
("fake_homepage", "A musician's homepage with tour dates and album art"),
("fake_homepage", "A food blogger's site with recipe cards and photos"),
("fake_homepage", "A travel vlogger's site showcasing weird destinations"),
("fake_homepage", "A developer's portfolio with fun project descriptions"),
("fake_homepage", "An artist's gallery site with abstract paintings"),
("fake_social", "A social media profile page for a fictional influencer"),
("fake_social", "A Reddit-style thread about a bizarre conspiracy theory"),
("fake_social", "A Twitter/X feed from a fictional account"),
("fake_social", "A dating app profile page with funny bios"),
("fake_social", "A forum discussion about the best fictional pizza toppings"),
("fake_wiki", "A Wikipedia article about a fictional historical event"),
("fake_wiki", "A Wikipedia article about a fictional animal species"),
("fake_wiki", "A Wikipedia article about a fictional technology"),
("fake_wiki", "A wiki page for a fictional video game"),
("fake_wiki", "A wiki page for a fictional band or music group"),
("fake_forum", "A 2000s-era forum page with threads about tech support"),
("fake_forum", "A gaming forum with strategy discussions"),
("fake_forum", "A cooking forum with recipe debates and tips"),
("fake_forum", "A conspiracy forum with wild theories"),
("fake_blog", "A tech blog reviewing fictional gadgets"),
("fake_blog", "A lifestyle blog about minimalist living"),
("fake_blog", "A horror story blog with creepy tales"),
("fake_blog", "A science blog explaining weird phenomena"),
("fake_news", "A local news article about a quirky town event"),
("fake_news", "A sports article about a fictional team"),
("fake_news", "A celebrity gossip article"),
("fake_app", "A landing page for a fictional mobile app"),
("fake_app", "A SaaS dashboard for a fake productivity tool"),
("fake_app", "A streaming service page with fictional shows"),
("fake_other", "A fake email inbox from the year 2000"),
("fake_other", "A fake search engine results page for a weird query"),
("fake_other", "A fake 404 error page with an interactive game"),
("fake_other", "A fake loading screen for a fictional video game"),
("fake_other", "A fake terminal emulator with ASCII art and commands"),
("web_game", "A browser-based snake game with score tracking and increasing difficulty"),
("web_game", "A memory card matching game with colorful icons and timer"),
("web_game", "A clicker/idle game where you build a fictional empire"),
("web_game", "A whack-a-mole style arcade game with combos and high scores"),
("web_game", "A platformer game with a cute character and multiple levels"),
("web_game", "A typing speed test game with word races and accuracy stats"),
("web_game", "A 2048-style number sliding puzzle game"),
("web_game", "A reaction time test game with random color flashes"),
("web_game", "A virtual pet game where you feed, play, and care for a creature"),
("web_game", "A maze runner game with procedurally generated levels"),
]
def build_prompt(context: dict[str, Any]) -> str:
site_type, theme = random.choice(THEMES)
mood = random.choice(["playful", "mysterious", "cozy", "epic", "dreamy", "energetic", "melancholic", "whimsical"])
return f"""You are a web designer creating a realistic-looking fake website.
Here is information about the visitor:
- Preferred language: {context["language"]}
Site type: {site_type}
Theme: {theme}
Mood: {mood}
Generate a COMPLETE HTML document ( ... ) that looks like a REAL website of this type:
- Make it look authentic and believable as if it actually existed on the internet
- Include realistic details: fake names, fake content, fake comments, fake usernames, fake timestamps
- Use appropriate layout and conventions for the site type (e.g., a wiki should look like Wikipedia, a forum should have threads, etc.)
- Use inline CSS for styling - make it polished and visually appealing
- Include subtle animations or interactive elements where appropriate
- Make the content fun, interesting, and creative - not boring lorem ipsum
- Use the visitor's language for text when possible
- Responsive design that works on mobile
Examples:
- fake_store: product listings, prices, cart button, reviews
- fake_social: profile pic, followers count, posts, likes, comments
- fake_wiki: infobox, table of contents, citations, edit links
- fake_forum: thread list, usernames, post counts, timestamps
- fake_blog: article with author, date, comments section
- fake_homepage: about section, portfolio/work, contact info
Return ONLY the HTML, no markdown, no explanations."""
def stream_llm(context: dict[str, Any]) -> Generator[str]:
prompt = build_prompt(context)
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
}
payload: dict[str, Any] = {
"model": MODEL,
"messages": [
{"role": "system", "content": "You are a creative web designer."},
{"role": "user", "content": prompt},
],
"temperature": 0.9,
"stream": True,
}
req_id = uuid.uuid4().hex[:8]
start = time.time()
logger.info(f"[{req_id}] Streaming LLM call for IP={context['ip']}, lang={context['language']}")
try:
with httpx.Client(timeout=300) as client, client.stream("POST", API_URL, headers=headers, json=payload) as resp:
elapsed = time.time() - start
resp.raise_for_status()
buffer = ""
for chunk in resp.iter_text():
buffer += chunk
while "\n" in buffer:
line, buffer = buffer.split("\n", 1)
line = line.strip()
if not line or line == "data: [DONE]":
continue
if line.startswith("data: "):
data_str = line[6:]
try:
data: dict[str, Any] = json.loads(data_str)
delta = data.get("choices", [{}])[0].get("delta", {}).get("content", "")
if delta:
yield delta
except json.JSONDecodeError:
pass
elapsed = time.time() - start
logger.info(f"[{req_id}] Stream complete in {elapsed:.1f}s")
except httpx.HTTPStatusError as e:
elapsed = time.time() - start
logger.error(f"[{req_id}] HTTP error in {elapsed:.1f}s: {e.response.status_code} {e.response.text[:200]}")
yield f"\n\n"
except BaseException as e: # noqa: BLE001
elapsed = time.time() - start
logger.error(f"[{req_id}] Stream failed in {elapsed:.1f}s: {e}")
yield f"\n\n"
def sse_stream(context: dict[str, Any]) -> Generator[str]:
buffer: list[str] = []
def status(msg: str) -> str:
return f"data: [STATUS:{msg}]\n\n"
yield status("Generating HTML...")
for chunk in stream_llm(context):
buffer.append(chunk)
encoded = base64.b64encode(chunk.encode("utf-8")).decode("ascii")
yield f"data: {encoded}\n\n"
html = "".join(buffer)
yield "data: [DONE]\n\n"
yield status("Validating HTML structure...")
errors = validate_html(html)
if errors:
short = summarize_errors(errors)
yield status(f"Found {len(errors)} issue(s): {short}. Fixing (attempt 1/2)...")
html = fix_html(html, errors)
errors = validate_html(html)
if errors:
short = summarize_errors(errors)
yield status("Still invalid. Fixing (attempt 2/2)...")
html = fix_html(html, errors)
errors = validate_html(html)
if errors:
short = summarize_errors(errors)
yield status(f"Still has {len(errors)} issue(s): {short}. Serving as-is.")
else:
yield status("Fixed successfully.")
yield "data: [CLEAR]\n\n"
for i in range(0, len(html), 1000):
chunk = html[i : i + 1000]
encoded = base64.b64encode(chunk.encode("utf-8")).decode("ascii")
yield f"data: {encoded}\n\n"
else:
yield status("Validation passed.")
yield "data: [RENDER]\n\n"
@app.route("/", defaults={"path": ""})
@app.route("/")
def index(path: str) -> Response:
logger.info(f"Request from {request.remote_addr}: {request.path} (UA: {request.headers.get('User-Agent', '')[:80]})")
return Response(LOADING_PAGE, mimetype="text/html")
@app.route("/stream")
def stream() -> Response:
context = get_user_context(request)
logger.info(f"Stream request from {request.remote_addr}")
def generate() -> Generator[str]:
yield from sse_stream(context)
return Response(
stream_with_context(generate()),
mimetype="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
"Connection": "keep-alive",
},
)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=True)