"""
def make_httpx_client() -> httpx.Client:
return httpx.Client(timeout=300)
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 make_httpx_client() 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)}") # type: ignore[arg-type]
if len(body_tags) != 1: # type: ignore[arg-type]
errors.append(f"Expected 1 tag, found {len(body_tags)}") # type: ignore[arg-type]
if len(head_tags) != 1: # type: ignore[arg-type]
errors.append(f"Expected 1 tag, found {len(head_tags)}") # type: ignore[arg-type]
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:
try:
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}")
except BaseException as e: # noqa: BLE001
logger.warning(f"JS review failed: {e}")
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 enqueue_request(context: dict[str, Any]) -> tuple[str, int]:
request_id = uuid.uuid4().hex[:12]
ctx_key = f"wotf:req:{request_id}:ctx"
rv.hset(ctx_key, mapping={k: json.dumps(v) for k, v in context.items()})
rv.set(f"wotf:req:{request_id}:status", "pending")
rv.rpush("wotf:queue", request_id)
position = rv.llen("wotf:queue")
return (request_id, position)
def get_queue_info(request_id: str) -> dict[str, Any]:
queue_items = rv.lrange("wotf:queue", 0, -1)
if request_id in queue_items:
position = queue_items.index(request_id) + 1
else:
position = 0
avg_raw = rv.get("wotf:stats:avg_time")
avg_str: str = str(avg_raw) if avg_raw is not None else "15"
avg = float(avg_str)
eta = position * avg
return {"position": position, "avg_time": avg, "eta": eta}
def worker_loop() -> None:
worker_id = f"worker-{os.getpid()}-{threading.current_thread().name}"
hb_key = "wotf:processing:hb"
lock_hb_start = time.time()
while worker_running:
request_id: str | None = None
try:
lock_holder = rv.get("wotf:processing")
if lock_holder is not None:
hb_ttl = rv.ttl(hb_key)
if hb_ttl == -2:
logger.info(f"[worker] Reclaiming stale lock from {lock_holder}")
rv.delete("wotf:processing")
rv.delete(hb_key)
else:
time.sleep(0.5)
continue
acquired = rv.set("wotf:processing", worker_id, ex=PROCESSING_LOCK_TTL_S, nx=True)
if not acquired:
time.sleep(0.5)
continue
rv.set(hb_key, worker_id, ex=HB_KEY_TTL_S)
request_id = rv.lpop("wotf:queue")
if request_id is None:
rv.delete("wotf:processing")
rv.delete(hb_key)
time.sleep(0.5)
continue
rv.set(f"wotf:req:{request_id}:status", "waiting_confirmation")
rv.delete(f"wotf:stream:{request_id}:chunks")
rv.rpush(f"wotf:stream:{request_id}:chunks", f"[ID:{request_id}]")
rv.rpush(f"wotf:stream:{request_id}:chunks", "[STATUS:waiting_confirmation]")
rv.delete("wotf:processing")
rv.delete(hb_key)
confirmed = False
for _ in range(CONFIRMATION_TIMEOUT_ITERATIONS):
time.sleep(CONFIRMATION_POLL_INTERVAL)
if rv.get(f"wotf:req:{request_id}:confirmed"):
confirmed = True
break
if not confirmed:
logger.info(f"[worker] No confirmation for {request_id}, marking as error")
rv.set(f"wotf:req:{request_id}:status", "error")
rv.set(f"wotf:req:{request_id}:error", "Worker did not receive confirmation from client")
rv.rpush(f"wotf:stream:{request_id}:chunks", "[STATUS:Connection to worker failed]")
rv.rpush(f"wotf:stream:{request_id}:chunks", "[ERROR]")
rv.delete(f"wotf:req:{request_id}:confirmed")
rv.expire(f"wotf:req:{request_id}:ctx", REQUEST_CTX_TTL_S)
time.sleep(0.5)
continue
acquired = False
for _ in range(5):
if rv.set("wotf:processing", worker_id, ex=PROCESSING_LOCK_TTL_S, nx=True):
acquired = True
break
time.sleep(0.5)
if not acquired:
logger.error(f"[worker] Failed to acquire lock for {request_id} after retries")
rv.set(f"wotf:req:{request_id}:status", "pending")
rv.delete(f"wotf:stream:{request_id}:chunks")
rv.delete(f"wotf:req:{request_id}:confirmed")
rv.rpush("wotf:queue", request_id)
continue
rv.set(hb_key, worker_id, ex=HB_KEY_TTL_S)
rv.set(f"wotf:req:{request_id}:status", "processing")
rv.set(f"wotf:req:{request_id}:last_heartbeat", "0")
ctx_raw = rv.hgetall(f"wotf:req:{request_id}:ctx")
ctx: dict[str, Any] = {}
for k_str, v_raw in ctx_raw.items():
k: str = k_str
v: Any = v_raw
try:
ctx[k] = json.loads(v) if isinstance(v, str) else v
except json.JSONDecodeError:
ctx[k] = v
start = time.time()
hb_start = time.time()
hb_check_start = time.time()
html_buffer: list[str] = []
aborted = False
try:
for chunk in stream_llm(ctx):
if time.time() - hb_start >= 5:
rv.set(hb_key, worker_id, ex=HB_KEY_TTL_S)
hb_start = time.time()
if time.time() - hb_check_start >= HEARTBEAT_CHECK_INTERVAL_S:
hb_check_start = time.time()
last_hb_raw = rv.get(f"wotf:req:{request_id}:last_heartbeat")
if last_hb_raw is None:
logger.info(f"[worker] Heartbeat key missing, aborting {request_id}")
rv.set(f"wotf:req:{request_id}:status", "aborted")
rv.rpush(f"wotf:stream:{request_id}:chunks", "[ERROR]")
rv.delete("wotf:processing")
rv.delete(hb_key)
aborted = True
break
last_hb = float(last_hb_raw) # type: ignore[arg-type]
if last_hb > 0 and time.time() - last_hb > HEARTBEAT_TIMEOUT_S:
logger.info(f"[worker] Client disconnected (heartbeat stale), aborting {request_id}")
rv.set(f"wotf:req:{request_id}:status", "aborted")
rv.rpush(f"wotf:stream:{request_id}:chunks", "[ERROR]")
rv.delete("wotf:processing")
rv.delete(hb_key)
aborted = True
break
html_buffer.append(chunk)
encoded = base64.b64encode(chunk.encode("utf-8")).decode("ascii")
rv.rpush(f"wotf:stream:{request_id}:chunks", encoded)
except BaseException as e: # noqa: BLE001
elapsed = time.time() - start
logger.error(f"[worker] stream_llm failed for {request_id} in {elapsed:.1f}s: {e}")
retries = rv.incr(f"wotf:req:{request_id}:retries")
if retries <= 2:
logger.info(f"[worker] Requeuing {request_id} (attempt {retries}/2)")
rv.set(f"wotf:req:{request_id}:status", "pending")
rv.delete(f"wotf:stream:{request_id}:chunks")
rv.rpush("wotf:queue", request_id)
else:
logger.error(f"[worker] {request_id} failed after {retries} attempts")
rv.set(f"wotf:req:{request_id}:status", "error")
rv.set(f"wotf:req:{request_id}:error", str(e))
rv.delete("wotf:processing")
rv.delete(hb_key)
time.sleep(1)
continue
if aborted:
time.sleep(1)
continue
html = "".join(html_buffer)
rv.rpush(f"wotf:stream:{request_id}:chunks", "[STATUS:Validating HTML structure...]")
errors = validate_html(html)
if errors:
short = summarize_errors(errors)
rv.rpush(f"wotf:stream:{request_id}:chunks", f"[STATUS: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)
rv.rpush(f"wotf:stream:{request_id}:chunks", "[STATUS:Still invalid. Fixing (attempt 2/2)...]")
html = fix_html(html, errors)
errors = validate_html(html)
if errors:
short = summarize_errors(errors)
rv.rpush(f"wotf:stream:{request_id}:chunks", f"[STATUS:Still has {len(errors)} issue(s): {short}. Serving as-is.]")
else:
rv.rpush(f"wotf:stream:{request_id}:chunks", "[STATUS:Fixed successfully.]")
rv.rpush(f"wotf:stream:{request_id}:chunks", "[CLEAR]")
for i in range(0, len(html), CHUNK_SIZE):
chunk = html[i : i + CHUNK_SIZE]
encoded = base64.b64encode(chunk.encode("utf-8")).decode("ascii")
rv.rpush(f"wotf:stream:{request_id}:chunks", encoded)
else:
rv.rpush(f"wotf:stream:{request_id}:chunks", "[STATUS:Validation passed.]")
rv.rpush(f"wotf:stream:{request_id}:chunks", "[RENDER]")
elapsed = time.time() - start
rv.set(f"wotf:req:{request_id}:status", "done")
n_raw = rv.get("wotf:stats:gen_count")
n_str: str = str(n_raw) if n_raw is not None else "0"
n = int(n_str)
avg_raw = rv.get("wotf:stats:avg_time")
avg_str: str = str(avg_raw) if avg_raw is not None else "15"
avg = float(avg_str)
new_avg = (avg * n + elapsed) / (n + 1) if n > 0 else elapsed
rv.set("wotf:stats:avg_time", str(new_avg))
rv.set("wotf:stats:gen_count", str(n + 1))
rv.delete("wotf:processing")
rv.delete(hb_key)
logger.info(f"[worker] Generated {request_id} in {elapsed:.1f}s (avg={new_avg:.1f}s)")
if time.time() - lock_hb_start >= WORKER_LOCK_HEARTBEAT_INTERVAL_S:
lock_hb_start = time.time()
rv.set("wotf:worker_lock", "1", xx=True, ex=WORKER_LOCK_TTL_S)
except BaseException as e: # noqa: BLE001
logger.error(f"[worker] Error: {e}")
if request_id:
rv.set(f"wotf:req:{request_id}:status", "error")
rv.set(f"wotf:req:{request_id}:error", str(e))
rv.delete("wotf:processing")
rv.delete(hb_key)
time.sleep(1)
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 make_httpx_client() 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 Exception as e: # noqa: BLE001
elapsed = time.time() - start
logger.error(f"[{req_id}] Stream failed in {elapsed:.1f}s: {e}")
yield f"\n\n"
@app.route("/health")
def health() -> Response:
try:
valkey_ok = rv.get("wotf:health_check") is not None or rv.set("wotf:health_check", "1", ex=60)
except BaseException: # noqa: BLE001
valkey_ok = False
queue_depth = rv.llen("wotf:queue")
worker_active = rv.exists("wotf:worker_lock")
status = "ok" if (valkey_ok and worker_active) else "degraded"
return Response(
json.dumps({
"status": status,
"valkey_connected": valkey_ok,
"worker_active": worker_active,
"queue_depth": queue_depth,
}),
mimetype="application/json",
)
@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}")
request_id, position = enqueue_request(context)
logger.info(f"[{request_id}] Enqueued at position {position}")
def generate() -> Generator[str]:
yield "data: [CLEAR]\n\n"
last_idx = 0
while True:
status_raw = rv.get(f"wotf:req:{request_id}:status")
status = status_raw if isinstance(status_raw, str) else "pending"
if status == "error":
error_raw = rv.get(f"wotf:req:{request_id}:error")
error_msg = error_raw if isinstance(error_raw, str) else "Unknown error"
yield f"data: [STATUS:Error: {error_msg}]\n\n"
yield "data: [ERROR]\n\n"
return
if status == "waiting_confirmation":
chunks_raw = rv.lrange(f"wotf:stream:{request_id}:chunks", 0, -1)
chunks: list[str] = list(chunks_raw) if isinstance(chunks_raw, list) else [] # type: ignore[arg-type]
if chunks:
for c in chunks:
yield f"data: {c}\n\n"
else:
yield "data: [STATUS:Waiting for your connection...]\n\n"
time.sleep(0.2)
continue
if status == "processing":
chunks_raw = rv.lrange(f"wotf:stream:{request_id}:chunks", last_idx, -1)
chunks: list[str] = list(chunks_raw) if isinstance(chunks_raw, list) else [] # type: ignore[arg-type]
if chunks:
for c in chunks:
yield f"data: {c}\n\n"
last_idx += len(chunks)
else:
yield "data: [STATUS:Generating your page...]\n\n"
time.sleep(0.1)
continue
if status == "done":
chunks_raw = rv.lrange(f"wotf:stream:{request_id}:chunks", last_idx, -1)
chunks: list[str] = list(chunks_raw) if isinstance(chunks_raw, list) else [] # type: ignore[arg-type]
for c in chunks:
yield f"data: {c}\n\n"
break
info = get_queue_info(request_id)
pos = info["position"]
eta = info["eta"]
if pos == 0:
yield "data: [STATUS:Your turn is next...]\n\n"
else:
yield f"data: [STATUS:Position {pos} in line | Est. wait: ~{eta:.0f}s]\n\n"
time.sleep(1)
while True:
chunks_raw = rv.lrange(f"wotf:stream:{request_id}:chunks", last_idx, -1)
chunks: list[str] = list(chunks_raw) if isinstance(chunks_raw, list) else [] # type: ignore[arg-type]
if chunks:
for chunk in chunks:
if chunk == "[RENDER]":
yield f"data: {chunk}\n\n"
return
yield f"data: {chunk}\n\n"
last_idx += len(chunks)
time.sleep(0.05)
else:
status_raw = rv.get(f"wotf:req:{request_id}:status")
status = status_raw if isinstance(status_raw, str) else "pending"
if status == "done":
chunks_raw = rv.lrange(f"wotf:stream:{request_id}:chunks", last_idx, -1)
chunks: list[str] = list(chunks_raw) if isinstance(chunks_raw, list) else [] # type: ignore[arg-type]
if chunks:
for chunk in chunks:
yield f"data: {chunk}\n\n"
break
if status == "error":
error_raw = rv.get(f"wotf:req:{request_id}:error")
error_msg = error_raw if isinstance(error_raw, str) else "Unknown error"
yield f"data: [STATUS:Error: {error_msg}]\n\n"
yield "data: [ERROR]\n\n"
return
if status == "aborted":
yield "data: [STATUS:Generation aborted (connection lost)]\n\n"
yield "data: [ERROR]\n\n"
return
time.sleep(0.2)
return Response(
stream_with_context(generate()),
mimetype="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
"Connection": "keep-alive",
},
)
@app.route("/confirm/", methods=["POST"])
def confirm(request_id: str) -> Response:
ctx = rv.exists(f"wotf:req:{request_id}:ctx")
if not ctx:
return Response("not found", status=404)
rv.set(f"wotf:req:{request_id}:confirmed", "1")
return Response("ok", status=200)
@app.route("/heartbeat/", methods=["POST"])
def heartbeat(request_id: str) -> Response:
status = rv.get(f"wotf:req:{request_id}:status")
if status is None:
return Response("not found", status=404)
rv.set(f"wotf:req:{request_id}:last_heartbeat", str(time.time()))
return Response("ok", status=200)
def start_worker_if_needed() -> None:
global worker_running, worker_thread
if worker_running:
return
claimed = rv.set("wotf:worker_lock", "1", nx=True, ex=WORKER_LOCK_TTL_S)
if claimed:
worker_running = True
if os.getenv("GUNICORN_WORKER_ID"):
from gevent import spawn
worker_thread = spawn(worker_loop) # type: ignore[assignment]
logger.info("[startup] Worker greenlet started in this process")
else:
worker_thread = threading.Thread(target=worker_loop, daemon=True)
worker_thread.start()
logger.info("[startup] Worker thread started in this process")
else:
logger.info("[startup] Worker thread already running in another process")
def gunicorn_post_fork(server: Any, worker: Any) -> None:
start_worker_if_needed()
def graceful_shutdown(signum: int, frame: Any) -> None:
global worker_running
sig_name = "SIGINT" if signum == signal.SIGINT else "SIGTERM"
logger.info(f"[shutdown] Received {sig_name}, stopping worker...")
worker_running = False
rv.delete("wotf:worker_lock")
rv.delete("wotf:processing")
rv.delete("wotf:processing:hb")
logger.info("[shutdown] Locks released, exiting")
sys.exit(0)
if __name__ == "__main__":
signal.signal(signal.SIGINT, graceful_shutdown)
signal.signal(signal.SIGTERM, graceful_shutdown)
worker_running = True
worker_thread = threading.Thread(target=worker_loop, daemon=True)
worker_thread.start()
app.run(host="0.0.0.0", port=5000, debug=True)