#!/usr/bin/env python3
"""
IC Generate FastAPI — Single-file app combining:
- Featherless API proxy with key rotation
- Concurrent UI question & solution generator
- SQLite database with crash recovery
- REST API for management (Admin-protected)
- API Key management routes
- [ENGINE] Question generation engine with randomized prompt templates
- [ENGINE] Runtime prompt management endpoints
Persistent storage: /data
- /data/ic_data.db — SQLite database
- /data/keys.txt — API keys (one per line)
- /data/exported_code/ — File exports
- /data/prompt_engine_state.json — [ENGINE] Prompt engine state
"""
# === Imports ===
import os
import re
import json
import time
import random
import sqlite3
import logging
import signal
import threading
import itertools
from typing import Optional, List, Dict, Any
from concurrent.futures import ThreadPoolExecutor, as_completed
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI, Request, HTTPException, Query, Depends, Security, status
from fastapi.security import APIKeyHeader
from fastapi.responses import StreamingResponse, JSONResponse, HTMLResponse
from pydantic import BaseModel
from dotenv import load_dotenv
from openai import OpenAI, APIError, APIConnectionError, RateLimitError
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
# NOW your imports will work:
from prompt_engine import QuestionGenerationEngine, PromptConfig
load_dotenv()
# === Configuration ===
DATA_DIR = os.environ.get("DATA_DIR", "/data")
os.makedirs(DATA_DIR, exist_ok=True)
DB_PATH = os.path.join(DATA_DIR, "ic_data.db")
KEYS_FILE = os.path.join(DATA_DIR, "keys.txt")
EXPORT_DIR = os.path.join(DATA_DIR, "exported_code")
# [ENGINE] State file for prompt engine persistence
PROMPT_STATE_FILE = os.path.join(DATA_DIR, "prompt_engine_state.json")
FEATHERLESS_API_BASE = os.environ.get(
"FEATHERLESS_API_BASE", "https://api.featherless.ai/v1"
)
PORT = int(os.environ.get("PORT", "7860"))
STACK = "HTML/CSS/JS"
# [ENGINE] Initialize the global prompt engine
prompt_engine = QuestionGenerationEngine(
state_file=PROMPT_STATE_FILE,
min_temperature=float(os.environ.get("PROMPT_MIN_TEMP", "0.85")),
max_temperature=float(os.environ.get("PROMPT_MAX_TEMP", "1.15")),
max_tokens=4096,
)
# === Admin Authentication ===
ADMIN_API_KEY = os.environ.get("ADMIN_API_KEY", "change-me-default-admin-key")
api_key_header_auth = APIKeyHeader(name="X-Admin-Key", auto_error=False)
async def verify_admin_key(api_key: str = Security(api_key_header_auth)):
if api_key == ADMIN_API_KEY:
return api_key
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or missing Admin API Key. Provide it in the 'X-Admin-Key' header.",
)
# === Logging ===
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s [%(threadName)s]: %(message)s",
datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)
# === Shutdown Event ===
_shutdown = threading.Event()
def _handle_signal(sig, frame):
logger.warning("⚠️ Shutdown requested. Finishing in-flight requests...")
_shutdown.set()
signal.signal(signal.SIGINT, _handle_signal)
signal.signal(signal.SIGTERM, _handle_signal)
# === Database Layer ===
DB_LOCK = threading.Lock()
def init_db(db_path: str = DB_PATH):
"""Create all tables including pending_jobs for crash recovery."""
os.makedirs(os.path.dirname(db_path) or ".", exist_ok=True)
conn = sqlite3.connect(db_path, timeout=60)
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA journal_mode = DELETE")
conn.execute("PRAGMA busy_timeout = 30000")
c = conn.cursor()
c.execute(
"""CREATE TABLE IF NOT EXISTS questions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
question_text TEXT NOT NULL,
thinking_trace TEXT,
full_response TEXT,
model TEXT,
finish_reason TEXT,
usage_json TEXT,
raw_chunks_json TEXT,
chunk_count INTEGER,
generation_time_s REAL,
prompt_metadata TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)"""
)
c.execute(
"""CREATE TABLE IF NOT EXISTS solutions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
question_id INTEGER NOT NULL,
stack TEXT NOT NULL DEFAULT 'HTML/CSS/JS',
solution_code TEXT NOT NULL,
thinking_trace TEXT,
full_response TEXT,
model TEXT,
finish_reason TEXT,
usage_json TEXT,
raw_chunks_json TEXT,
chunk_count INTEGER,
generation_time_s REAL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY(question_id) REFERENCES questions(id) ON DELETE CASCADE
)"""
)
c.execute(
"""CREATE TABLE IF NOT EXISTS pending_jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
question_id INTEGER NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
started_at TEXT,
completed_at TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY(question_id) REFERENCES questions(id) ON DELETE CASCADE
)"""
)
c.execute(
"""CREATE TABLE IF NOT EXISTS run_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
started_at TEXT NOT NULL DEFAULT (datetime('now')),
finished_at TEXT,
q_model TEXT, s_model TEXT,
requested INTEGER, completed INTEGER, failed INTEGER, pending INTEGER,
config_json TEXT
)"""
)
# Migrations
_migrate_add_columns(
c,
"questions",
{
"thinking_trace": "TEXT",
"full_response": "TEXT",
"generation_time_s": "REAL",
"prompt_metadata": "TEXT", # [ENGINE] New column
},
)
_migrate_add_columns(
c,
"solutions",
{"thinking_trace": "TEXT", "full_response": "TEXT", "generation_time_s": "REAL"},
)
_migrate_add_columns(
c,
"pending_jobs",
{
"attempts": "INTEGER DEFAULT 0",
"last_error": "TEXT",
"started_at": "TEXT",
"completed_at": "TEXT",
},
)
conn.commit()
conn.close()
logger.info(f"Database ready: {db_path}")
def _migrate_add_columns(cursor, table, columns):
for col, col_type in columns.items():
try:
cursor.execute(f"ALTER TABLE {table} ADD COLUMN {col} {col_type}")
except sqlite3.OperationalError:
pass
def _safe_db_execute(db_path, operations, max_retries=5):
for attempt in range(max_retries):
try:
with DB_LOCK:
conn = sqlite3.connect(db_path, timeout=60)
conn.execute("PRAGMA journal_mode = DELETE")
conn.execute("PRAGMA busy_timeout = 30000")
cur = conn.cursor()
try:
result = operations(cur)
conn.commit()
return result
except Exception:
conn.rollback()
raise
finally:
conn.close()
except sqlite3.OperationalError as e:
if "locked" in str(e).lower() and attempt < max_retries - 1:
wait = 0.5 * (2**attempt)
logger.warning(
f"DB locked (attempt {attempt+1}/{max_retries}), retrying in {wait:.1f}s..."
)
time.sleep(wait)
else:
raise
def save_question(
db_path, q_text, q_thinking, q_full, q_model, q_finish, q_usage, q_chunks, gen_time,
prompt_metadata=None, # [ENGINE] New parameter
):
def ops(cur):
cur.execute(
"""INSERT INTO questions
(question_text, thinking_trace, full_response, model, finish_reason,
usage_json, raw_chunks_json, chunk_count, generation_time_s, prompt_metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
q_text,
q_thinking,
q_full,
q_model,
q_finish,
json.dumps(q_usage, ensure_ascii=False) if q_usage else None,
json.dumps(q_chunks, ensure_ascii=False, default=str),
len(q_chunks),
gen_time,
json.dumps(prompt_metadata, ensure_ascii=False) if prompt_metadata else None, # [ENGINE]
),
)
question_id = cur.lastrowid
cur.execute(
"INSERT INTO pending_jobs (question_id, status) VALUES (?, 'pending')",
(question_id,),
)
job_id = cur.lastrowid
return job_id, question_id
return _safe_db_execute(db_path, ops)
def save_solution(
db_path,
job_id,
question_id,
s_code,
s_thinking,
s_full,
s_model,
s_finish,
s_usage,
s_chunks,
gen_time,
):
def ops(cur):
cur.execute(
"""INSERT INTO solutions
(question_id, stack, solution_code, thinking_trace, full_response,
model, finish_reason, usage_json, raw_chunks_json, chunk_count, generation_time_s)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
question_id,
STACK,
s_code,
s_thinking,
s_full,
s_model,
s_finish,
json.dumps(s_usage, ensure_ascii=False) if s_usage else None,
json.dumps(s_chunks, ensure_ascii=False, default=str),
len(s_chunks),
gen_time,
),
)
cur.execute(
"UPDATE pending_jobs SET status='done', completed_at=datetime('now') WHERE id=?",
(job_id,),
)
_safe_db_execute(db_path, ops)
def mark_job_started(db_path, job_id):
def ops(cur):
cur.execute(
"UPDATE pending_jobs SET started_at=datetime('now'), attempts=attempts+1 WHERE id=?",
(job_id,),
)
_safe_db_execute(db_path, ops)
def mark_job_failed(db_path, job_id, error_msg):
def ops(cur):
cur.execute(
"UPDATE pending_jobs SET last_error=?, status='pending' WHERE id=?",
(str(error_msg)[:500], job_id),
)
_safe_db_execute(db_path, ops)
def get_pending_jobs(db_path):
conn = sqlite3.connect(db_path, timeout=30)
cur = conn.cursor()
cur.execute(
"""SELECT pj.id, pj.question_id, q.question_text, pj.attempts
FROM pending_jobs pj
JOIN questions q ON pj.question_id = q.id
WHERE pj.status = 'pending'
ORDER BY pj.id"""
)
rows = cur.fetchall()
conn.close()
return [(r[0], r[1], r[2], r[3]) for r in rows]
def log_run(
db_path, q_model, s_model, requested, completed, failed, pending, config
):
def ops(cur):
cur.execute(
"""INSERT INTO run_log
(q_model, s_model, requested, completed, failed, pending,
finished_at, config_json)
VALUES (?, ?, ?, ?, ?, ?, datetime('now'), ?)""",
(
q_model,
s_model,
requested,
completed,
failed,
pending,
json.dumps(config, ensure_ascii=False),
),
)
try:
_safe_db_execute(db_path, ops)
except Exception as e:
logger.warning(f"Failed to log run: {e}")
def delete_question_db(db_path, question_id):
def ops(cur):
cur.execute("DELETE FROM questions WHERE id=?", (question_id,))
return cur.rowcount > 0
return _safe_db_execute(db_path, ops)
def get_stats_dict(db_path):
if not os.path.exists(db_path):
return {"error": f"Database {db_path} does not exist."}
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM questions")
q_count = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM solutions")
s_count = cursor.fetchone()[0]
cursor.execute("SELECT COALESCE(SUM(chunk_count), 0) FROM questions")
q_chunks = cursor.fetchone()[0]
cursor.execute("SELECT COALESCE(SUM(chunk_count), 0) FROM solutions")
s_chunks = cursor.fetchone()[0]
cursor.execute(
"SELECT COUNT(*) FROM questions WHERE thinking_trace IS NOT NULL"
)
q_with_thinking = cursor.fetchone()[0]
cursor.execute(
"SELECT COUNT(*) FROM solutions WHERE thinking_trace IS NOT NULL"
)
s_with_thinking = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(*) FROM pending_jobs WHERE status='pending'")
pending = cursor.fetchone()[0]
cursor.execute(
"SELECT COALESCE(AVG(generation_time_s), 0) FROM questions WHERE generation_time_s IS NOT NULL"
)
avg_q_time = cursor.fetchone()[0]
cursor.execute(
"SELECT COALESCE(AVG(generation_time_s), 0) FROM solutions WHERE generation_time_s IS NOT NULL"
)
avg_s_time = cursor.fetchone()[0]
cursor.execute(
"SELECT id, question_text, model, created_at FROM questions ORDER BY id DESC LIMIT 5"
)
recent = cursor.fetchall()
runs = []
try:
cursor.execute(
"SELECT started_at, completed, failed, pending, q_model, s_model FROM run_log ORDER BY id DESC LIMIT 3"
)
runs = cursor.fetchall()
except sqlite3.OperationalError:
pass
return {
"questions": q_count,
"questions_with_thinking": q_with_thinking,
"solutions": s_count,
"solutions_with_thinking": s_with_thinking,
"pending_jobs": pending,
"total_chunks": q_chunks + s_chunks,
"question_chunks": q_chunks,
"solution_chunks": s_chunks,
"avg_question_time_s": round(avg_q_time, 1),
"avg_solution_time_s": round(avg_s_time, 1),
"recent_questions": [
{"id": qid, "model": qm, "created_at": qc, "preview": qt[:80]}
for qid, qt, qm, qc in reversed(recent)
],
"recent_runs": [
{
"started_at": r[0],
"completed": r[1],
"failed": r[2],
"pending": r[3],
"q_model": r[4],
"s_model": r[5],
}
for r in reversed(runs)
],
# [ENGINE] Include prompt engine stats
"prompt_engine_stats": prompt_engine.get_stats(),
}
def export_data_json(db_path):
if not os.path.exists(db_path):
return []
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"""
SELECT q.id, q.question_text, q.thinking_trace, q.full_response,
q.model AS q_model, q.created_at, q.generation_time_s,
q.prompt_metadata,
s.id AS s_id, s.stack, s.solution_code, s.thinking_trace AS s_thinking,
s.full_response AS s_full,
s.model AS s_model, s.finish_reason, s.chunk_count,
s.created_at AS s_created, s.generation_time_s AS s_gen_time
FROM questions q
LEFT JOIN solutions s ON q.id = s.question_id
ORDER BY q.id ASC
"""
)
rows = cursor.fetchall()
questions_map = {}
for (
q_id,
q_text,
q_thinking,
q_full,
q_model,
q_created,
q_gen_time,
q_prompt_meta, # [ENGINE]
s_id,
stack,
code,
s_thinking,
s_full,
s_model,
s_finish,
s_chunks,
s_created,
s_gen_time,
) in rows:
if q_id not in questions_map:
questions_map[q_id] = {
"id": q_id,
"question": q_text,
"thinking_trace": q_thinking,
"full_response": q_full,
"model": q_model,
"created_at": q_created,
"generation_time_s": q_gen_time,
"prompt_metadata": json.loads(q_prompt_meta) if q_prompt_meta else None, # [ENGINE]
"solutions": [],
}
if s_id and code:
questions_map[q_id]["solutions"].append(
{
"id": s_id,
"stack": stack,
"code": code,
"thinking_trace": s_thinking,
"full_response": s_full,
"model": s_model,
"finish_reason": s_finish,
"chunk_count": s_chunks,
"created_at": s_created,
"generation_time_s": s_gen_time,
}
)
return list(questions_map.values())
def export_files(db_path, output_dir):
if not os.path.exists(db_path):
return {"error": "Database does not exist."}
os.makedirs(output_dir, exist_ok=True)
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"""
SELECT q.id, q.question_text, q.thinking_trace, q.full_response,
q.model AS q_model, q.created_at, q.generation_time_s,
q.prompt_metadata,
s.id AS s_id, s.stack, s.solution_code, s.thinking_trace AS s_thinking,
s.full_response AS s_full,
s.model AS s_model, s.finish_reason, s.chunk_count,
s.created_at AS s_created, s.generation_time_s AS s_gen_time
FROM questions q
LEFT JOIN solutions s ON q.id = s.question_id
ORDER BY q.id ASC
"""
)
rows = cursor.fetchall()
exported = 0
for row in rows:
(
q_id,
q_text,
q_thinking,
q_full,
q_model,
q_created,
q_gen_time,
q_prompt_meta, # [ENGINE]
s_id,
stack,
code,
s_thinking,
s_full,
s_model,
s_finish,
s_chunks,
s_created,
s_gen_time,
) = row
if not s_id or not code:
continue
safe_q = re.sub(r"[^\w\-_\. ]", "_", q_text)[:50].strip()
q_folder = os.path.join(output_dir, f"Q{q_id}_{safe_q}")
os.makedirs(q_folder, exist_ok=True)
with open(os.path.join(q_folder, f"solution_{s_id}.html"), "w") as f:
f.write(code)
if q_thinking:
with open(os.path.join(q_folder, f"q_thinking_{q_id}.txt"), "w") as f:
f.write(q_thinking)
if s_thinking:
with open(os.path.join(q_folder, f"s_thinking_{s_id}.txt"), "w") as f:
f.write(s_thinking)
if q_full:
with open(os.path.join(q_folder, f"q_full_response_{q_id}.txt"), "w") as f:
f.write(q_full)
if s_full:
with open(os.path.join(q_folder, f"s_full_response_{s_id}.txt"), "w") as f:
f.write(s_full)
meta = {
"question_id": q_id,
"question": q_text,
"q_model": q_model,
"has_q_thinking": q_thinking is not None,
"q_generation_time_s": q_gen_time,
"prompt_metadata": json.loads(q_prompt_meta) if q_prompt_meta else None, # [ENGINE]
"solution_id": s_id,
"stack": stack,
"s_model": s_model,
"has_s_thinking": s_thinking is not None,
"finish_reason": s_finish,
"chunk_count": s_chunks,
"s_generation_time_s": s_gen_time,
"q_created": q_created,
"s_created": s_created,
}
with open(os.path.join(q_folder, f"meta_{s_id}.json"), "w") as f:
json.dump(meta, f, indent=2, ensure_ascii=False)
exported += 1
return {"exported": exported, "output_dir": output_dir}
# === Key Rotation ===
def load_keys():
"""Load API keys from /data/keys.txt or environment variable."""
keys = []
if os.path.exists(KEYS_FILE):
with open(KEYS_FILE, "r") as f:
keys = [line.strip() for line in f if line.strip()]
env_keys = os.environ.get("FEATHERLESS_API_KEYS", "")
if env_keys:
keys.extend([k.strip() for k in env_keys.split(",") if k.strip()])
seen = set()
unique_keys = []
for k in keys:
if k not in seen:
seen.add(k)
unique_keys.append(k)
if not unique_keys:
logger.warning(f"No keys found in {KEYS_FILE} or env. Using dummy key.")
unique_keys = ["dummy_key"]
return unique_keys
_keys = load_keys()
_key_cycle = itertools.cycle(_keys)
_key_lock = threading.Lock()
def get_next_key():
with _key_lock:
return next(_key_cycle)
def reload_keys():
global _keys, _key_cycle
_keys = load_keys()
_key_cycle = itertools.cycle(_keys)
return len(_keys)
# === Retry Logic ===
def retry_api_call(func, max_retries=5, delay=2):
last_err = None
for attempt in range(max_retries):
if _shutdown.is_set():
raise InterruptedError("Shutdown requested")
try:
return func()
except RateLimitError as e:
last_err = e
backoff = 120.0 + random.uniform(0, 10)
logger.warning(
f"Rate limit (429). Waiting {backoff:.0f}s... (attempt {attempt+1}/{max_retries})"
)
time.sleep(backoff)
except APIConnectionError as e:
last_err = e
backoff = min(60, delay * (2**attempt)) + random.uniform(0, 2)
logger.warning(
f"Connection error: {e}. Retry {attempt+1}/{max_retries} in {backoff:.1f}s..."
)
time.sleep(backoff)
except APIError as e:
last_err = e
if hasattr(e, "status_code") and e.status_code == 429:
backoff = 120.0 + random.uniform(0, 10)
logger.warning(f"Rate limit (via APIError 429). Waiting {backoff:.0f}s...")
elif hasattr(e, "status_code") and e.status_code in (500, 502, 503, 504):
backoff = min(90, delay * (2**attempt)) + random.uniform(0, 5)
logger.warning(
f"Server error {e.status_code}: {e}. Retry in {backoff:.1f}s..."
)
else:
raise
time.sleep(backoff)
except Exception:
raise
raise last_err
# === Streaming Completion ===
def stream_completion(client, model, messages, temperature=0.2, max_tokens=16384):
raw_chunks = []
content_parts = []
thinking_parts = []
tool_call_parts = []
function_call_parts = []
finish_reason = None
usage = None
actual_model = model
t0 = time.monotonic()
def make_call():
return client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=True,
stream_options={"include_usage": True},
)
stream = retry_api_call(make_call)
for chunk in stream:
if _shutdown.is_set():
break
try:
chunk_dict = chunk.model_dump()
except AttributeError:
try:
chunk_dict = json.loads(chunk.json())
except Exception:
chunk_dict = {"_raw": str(chunk)}
raw_chunks.append(chunk_dict)
if chunk.choices:
choice = chunk.choices[0]
delta = choice.delta
if delta:
if delta.content:
content_parts.append(delta.content)
for attr in (
"reasoning_content",
"reasoning",
"thinking",
"thought",
"internal_thoughts",
):
val = getattr(delta, attr, None)
if val:
thinking_parts.append(val)
break
if hasattr(delta, "tool_calls") and delta.tool_calls:
for tc in delta.tool_calls:
try:
tool_call_parts.append(tc.model_dump())
except Exception:
tool_call_parts.append(str(tc))
if hasattr(delta, "function_call") and delta.function_call:
try:
function_call_parts.append(delta.function_call.model_dump())
except Exception:
function_call_parts.append(str(delta.function_call))
if choice.finish_reason:
finish_reason = choice.finish_reason
if hasattr(chunk, "model") and chunk.model:
actual_model = chunk.model
if hasattr(chunk, "usage") and chunk.usage:
try:
usage = chunk.usage.model_dump()
except AttributeError:
usage = {
"prompt_tokens": getattr(chunk.usage, "prompt_tokens", None),
"completion_tokens": getattr(chunk.usage, "completion_tokens", None),
"total_tokens": getattr(chunk.usage, "total_tokens", None),
}
gen_time = time.monotonic() - t0
content = "".join(content_parts)
thinking = "".join(thinking_parts) or None
full_parts = []
if thinking:
full_parts.append(f"
Combined Featherless proxy + UI question/solution generator with full thinking capture.
✨ Now with randomized prompt engine (high temperature, 200+ topics, 8 templates)!
All management routes require the X-Admin-Key header.
Set the ADMIN_API_KEY environment variable in Hugging Face Spaces Secrets.