332 lines
15 KiB
Python
332 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""DRACO — The Book-Forged Code Oracle.
|
|
Flask app: home (hero + brag), library, pricing, API, MCP, SEO kit, RAG chat (SSE).
|
|
"""
|
|
import json, os, re
|
|
import requests as http
|
|
from flask import Flask, Response, jsonify, request, render_template_string, send_from_directory
|
|
|
|
import draco_core as core
|
|
from draco_core import CFG
|
|
from pages import PAGE_HOME, BASE_CSS, NAV
|
|
from pages2 import PAGE_LIBRARY, PAGE_PRICING, PAGE_API
|
|
|
|
app = Flask(__name__)
|
|
|
|
# ------------------------------------------------------------------ pages
|
|
def render(page, active):
|
|
stats = core.get_stats()
|
|
nav = NAV.replace("__ACTIVE__-" + active, "on")
|
|
nav = re.sub(r"__ACTIVE__-\w+", "", nav)
|
|
head = f"<style>{BASE_CSS}</style><nav>{nav}</nav>"
|
|
page = page.replace("__STATS__", f"{stats['books']} books · {stats['chunks']:,} indexed passages · {stats['chars']/1e6:.0f}M chars")
|
|
page = page.replace("__BASE__", CFG["base_url"])
|
|
ctx = dict(books=f"{stats['books']:,}", chunks=f"{stats['chunks']:,}",
|
|
mb=stats['chars'] // 1_000_000, free_day=CFG["anon_daily"],
|
|
free_monthly=CFG["free_monthly"], plans=CFG["plans"], base=CFG["base_url"])
|
|
return render_template_string(head + page, **ctx)
|
|
|
|
@app.route("/")
|
|
def home():
|
|
return render(PAGE_HOME, "home")
|
|
|
|
@app.route("/library")
|
|
def library():
|
|
return render(PAGE_LIBRARY, "library")
|
|
|
|
@app.route("/api/library")
|
|
def api_library():
|
|
try:
|
|
manifest = json.load(open(os.path.join(core.BOOKS_DIR, "manifest.json")))
|
|
except Exception:
|
|
manifest = []
|
|
books = [{"id": m["id"], "title": m["title"][:90], "category": m["category"],
|
|
"format": m["format"], "mb": round(m["bytes"] / 1e6, 1),
|
|
"download": f"/download/{m['id']}"}
|
|
for m in manifest if m.get("txt")]
|
|
books.sort(key=lambda b: b["title"].lower())
|
|
return jsonify(count=len(books), books=books)
|
|
|
|
@app.route("/download/<book_id>")
|
|
def download(book_id):
|
|
try:
|
|
manifest = json.load(open(os.path.join(core.BOOKS_DIR, "manifest.json")))
|
|
except Exception:
|
|
return jsonify(error="library offline"), 503
|
|
m = next((x for x in manifest if x["id"] == book_id), None)
|
|
if not m:
|
|
return jsonify(error="unknown book"), 404
|
|
return send_from_directory(core.BOOKS_DIR, m["source"], as_attachment=True)
|
|
|
|
@app.route("/pricing")
|
|
def pricing():
|
|
return render(PAGE_PRICING, "pricing")
|
|
|
|
@app.route("/api")
|
|
def api_page():
|
|
return render(PAGE_API, "api")
|
|
|
|
@app.route("/health")
|
|
def health():
|
|
try:
|
|
r = http.post(f"{CFG['ollama_url']}/api/generate", stream=True, timeout=4,
|
|
json={"model": CFG["model"], "prompt": "ping", "stream": True, "options": {"num_predict": 1}})
|
|
llm = r.status_code == 200
|
|
except Exception:
|
|
llm = False
|
|
return jsonify(status="ok", llm=llm, model=CFG["model"], **core.get_stats())
|
|
|
|
# ------------------------------------------------------------------ chat (web, SSE)
|
|
@app.route("/api/chat", methods=["POST"])
|
|
def chat():
|
|
data = request.json or {}
|
|
q = (data.get("q") or "").strip()
|
|
if not q:
|
|
return jsonify(error="Empty question."), 400
|
|
key = data.get("api_key")
|
|
user = None
|
|
if key:
|
|
user, err = core.auth_user(request)
|
|
if err:
|
|
return jsonify(error=err), 401
|
|
ok, msg = core.consume(user, "chat", q[:80], core.client_ip(request))
|
|
if not ok:
|
|
return jsonify(error=msg), 402
|
|
else:
|
|
ip = core.client_ip(request)
|
|
if not core.anon_allowed(ip):
|
|
return jsonify(error=f"Free web limit reached ({CFG['anon_daily']}/day). "
|
|
f"Sign up free at {CFG['base_url']}/api for {CFG['free_monthly']}/mo API credits."), 429
|
|
|
|
prompt, hits = core.build_prompt(q, k=CFG.get("rag_k", 6))
|
|
sources = [{"n": i + 1, "title": h["title"], "category": h["category"]}
|
|
for i, h in enumerate(hits)]
|
|
|
|
def generate():
|
|
yield f"data: {json.dumps({'type': 'sources', 'sources': sources})}\n\n"
|
|
acc = ""
|
|
try:
|
|
for channel, piece in core.stream_ollama(prompt):
|
|
if channel == "answer":
|
|
acc += piece
|
|
if core.degenerate(acc):
|
|
core.unload_model() # auto-recovery: next request reloads clean
|
|
yield f"data: {json.dumps({'type': 'error', 'text': 'model hiccup — auto-recovered, ask again'})}\n\n"
|
|
yield f"data: {json.dumps({'type': 'done'})}\n\n"
|
|
return
|
|
yield f"data: {json.dumps({'type': 'token', 'channel': channel, 'text': piece})}\n\n"
|
|
except Exception as e:
|
|
yield f"data: {json.dumps({'type': 'error', 'text': f'Model offline: {e}'})}\n\n"
|
|
yield f"data: {json.dumps({'type': 'done'})}\n\n"
|
|
|
|
return Response(generate(), mimetype="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
|
|
|
|
@app.route("/api/remaining")
|
|
def remaining():
|
|
ip = core.client_ip(request)
|
|
return jsonify(remaining=core.anon_remaining(ip), daily=CFG["anon_daily"])
|
|
|
|
# ------------------------------------------------------------------ REST API (metered)
|
|
@app.route("/api/ask", methods=["POST"])
|
|
def api_ask():
|
|
user, err = core.auth_user(request)
|
|
if err:
|
|
return jsonify(error=err), 401
|
|
q = ((request.json or {}).get("q") or "").strip()
|
|
if not q:
|
|
return jsonify(error="q required"), 400
|
|
ok, msg = core.consume(user, "ask", q[:80], core.client_ip(request))
|
|
if not ok:
|
|
return jsonify(error=msg), 402
|
|
prompt, hits = core.build_prompt(q, k=CFG.get("rag_k", 6))
|
|
try:
|
|
answer = core.ask_ollama(prompt)
|
|
except Exception as e:
|
|
return jsonify(error=f"Model offline: {e}"), 503
|
|
return jsonify(question=q, answer=answer,
|
|
sources=[{"title": h["title"], "category": h["category"],
|
|
"book_id": h["book_id"], "chunk": h["chunk"]} for h in hits])
|
|
|
|
@app.route("/api/search", methods=["GET", "POST"])
|
|
def api_search():
|
|
user, err = core.auth_user(request)
|
|
if err:
|
|
return jsonify(error=err), 401
|
|
q = (request.json or {}).get("q") if request.is_json else request.args.get("q")
|
|
q = (q or "").strip()
|
|
if not q:
|
|
return jsonify(error="q required"), 400
|
|
ok, msg = core.consume(user, "search", q[:80], core.client_ip(request))
|
|
if not ok:
|
|
return jsonify(error=msg), 402
|
|
hits = core.search_library(q, k=int((request.json or {}).get("k", 8) if request.is_json else request.args.get("k", 8)))
|
|
for h in hits:
|
|
h.pop("text", None)
|
|
return jsonify(query=q, results=hits)
|
|
|
|
@app.route("/api/my-usage", methods=["GET", "POST"])
|
|
def my_usage():
|
|
user, err = core.auth_user(request)
|
|
if err:
|
|
return jsonify(error=err), 401
|
|
c = core.db()
|
|
u = c.execute("SELECT credits, free_used, total_calls FROM users WHERE id=?", (user["id"],)).fetchone()
|
|
if u["total_calls"] is None:
|
|
c.execute("UPDATE users SET total_calls=(SELECT COUNT(*) FROM usage_log WHERE user_id=?) WHERE id=?", (user["id"], user["id"]))
|
|
c.commit()
|
|
u = c.execute("SELECT credits, free_used, total_calls FROM users WHERE id=?", (user["id"],)).fetchone()
|
|
recent = [dict(r) for r in c.execute(
|
|
"SELECT kind, detail, created_at FROM usage_log WHERE user_id=? ORDER BY id DESC LIMIT 10", (user["id"],))]
|
|
c.close()
|
|
return jsonify(email=user["email"], credits=u["credits"], free_used_this_month=u["free_used"],
|
|
free_monthly=CFG["free_monthly"], total_calls=u["total_calls"] or 0, recent=recent)
|
|
|
|
@app.route("/api/signup", methods=["POST"])
|
|
def api_signup():
|
|
email = (request.json or {}).get("email", "")
|
|
key, status = core.signup(email)
|
|
if not key:
|
|
return jsonify(error=status), 400
|
|
return jsonify(email=email.strip().lower(), api_key=key, status=status,
|
|
free_monthly=CFG["free_monthly"],
|
|
note="Free tier active. Keep this key safe — it is shown once per signup.")
|
|
|
|
@app.route("/api/create-invoice", methods=["POST"])
|
|
def api_invoice():
|
|
user, err = core.auth_user(request)
|
|
if err:
|
|
return jsonify(error=err), 401
|
|
plan = (request.json or {}).get("plan", "")
|
|
inv, err = core.create_invoice(user, plan)
|
|
if err:
|
|
return jsonify(error=err), 400
|
|
return jsonify(invoice_id=inv["id"], amount=inv.get("amount"),
|
|
checkout_url=inv.get("checkoutLink") or (inv.get("checkout") or {}).get("link"))
|
|
|
|
# ------------------------------------------------------------------ BTCPay webhook
|
|
@app.route("/webhook/btcpay", methods=["POST"])
|
|
def webhook():
|
|
body = request.get_json(silent=True) or {}
|
|
# shared-secret check via webhook URL suffix
|
|
ok = request.args.get("wh") == CFG["btcpay"].get("webhook_secret")
|
|
core.handle_webhook(body, ok)
|
|
return jsonify(status="ok")
|
|
|
|
# ------------------------------------------------------------------ SEO / agent kit
|
|
@app.route("/robots.txt")
|
|
def robots():
|
|
ais = ("GPTBot OAI-SearchBot ChatGPT-User ClaudeBot Claude-Web anthropic-ai PerplexityBot "
|
|
"Perplexity-User Google-Extended GoogleOther Amazonbot Applebot-Extended Bytespider "
|
|
"cohere-ai Meta-ExternalAgent Diffbot CCBot").split()
|
|
lines = [f"User-agent: {a}\nAllow: /" for a in ais]
|
|
lines.append("User-agent: *\nAllow: /\nDisallow: /webhook/")
|
|
lines.append(f"Sitemap: {CFG['base_url']}/sitemap.xml")
|
|
return Response("\n\n".join(lines), mimetype="text/plain")
|
|
|
|
@app.route("/llms.txt")
|
|
def llms_txt():
|
|
return Response(render_template_string(core.load_llms_txt()), mimetype="text/plain")
|
|
|
|
@app.route("/llms-full.txt")
|
|
def llms_full():
|
|
return Response(render_template_string(core.load_llms_txt() + "\n\n" + core.load_llms_full()), mimetype="text/plain")
|
|
|
|
@app.route("/.well-known/ai-plugin.json")
|
|
def ai_plugin():
|
|
return Response(json.dumps({
|
|
"schema_version": "v1",
|
|
"name_for_human": "DRACO — Book-Forged Code Oracle",
|
|
"name_for_model": "draco",
|
|
"description_for_human": "Coding & security answers grounded in a library of hundreds of real programming and hacking books, with citations.",
|
|
"description_for_model": "Ask coding, systems, and security questions. Answers are RAG-grounded in DRACO's book library and include citations [n] plus a sources array with titles. Use /api/ask for full answers, /api/search to find book passages. Auth: X-API-Key header, free tier available.",
|
|
"auth": {"type": "none"},
|
|
"api": {"type": "openapi", "url": f"{CFG['base_url']}/openapi.json"},
|
|
"logo_url": f"{CFG['base_url']}/static/logo.svg",
|
|
"contact_email": "drjones@thetempleofdoom.com",
|
|
"legal_info_url": f"{CFG['base_url']}/api"
|
|
}, indent=1), mimetype="application/json")
|
|
|
|
@app.route("/.well-known/mcp-server.json")
|
|
def mcp_manifest():
|
|
return Response(json.dumps({
|
|
"$schema": "https://cdn.jsdelivr.net/npm/@modelcontextprotocol/sdk@latest/schema.json",
|
|
"name": "com.thetempleofdoom.draco/ask",
|
|
"description": "Coding & security oracle grounded in hundreds of real books. Ask, search, cite.",
|
|
"homepage": CFG["base_url"],
|
|
"remotes": [{"type": "streamable-http", "url": f"{CFG['base_url']}/mcp"}]
|
|
}, indent=1), mimetype="application/json")
|
|
|
|
@app.route("/openapi.json")
|
|
def openapi():
|
|
return Response(json.dumps({
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "DRACO API", "version": "1.0",
|
|
"description": "RAG coding/security oracle over hundreds of real books. Free tier: signup for API key."},
|
|
"servers": [{"url": CFG["base_url"]}],
|
|
"paths": {
|
|
"/api/ask": {"post": {"summary": "Ask a coding/security question (RAG + citations)",
|
|
"requestBody": {"content": {"application/json": {"schema": {"type": "object",
|
|
"properties": {"q": {"type": "string"}, "api_key": {"type": "string"}},
|
|
"required": ["q"]}}}},
|
|
"responses": {"200": {"description": "answer + sources"}}}},
|
|
"/api/search": {"get": {"summary": "BM25 passage search across the library",
|
|
"parameters": [{"name": "q", "in": "query", "required": True, "schema": {"type": "string"}},
|
|
{"name": "k", "in": "query", "schema": {"type": "integer", "default": 8}}],
|
|
"responses": {"200": {"description": "passage results"}}}},
|
|
"/api/signup": {"post": {"summary": "Get a free API key",
|
|
"requestBody": {"content": {"application/json": {"schema": {"type": "object",
|
|
"properties": {"email": {"type": "string"}}, "required": ["email"]}}}},
|
|
"responses": {"200": {"description": "api_key"}}}},
|
|
"/api/my-usage": {"get": {"summary": "Credits + recent calls", "responses": {"200": {"description": "usage"}}}},
|
|
"/api/create-invoice": {"post": {"summary": "Buy credits (Bitcoin via BTCPay)",
|
|
"requestBody": {"content": {"application/json": {"schema": {"type": "object",
|
|
"properties": {"plan": {"type": "string", "enum": list(CFG["plans"].keys())}}}}}},
|
|
"responses": {"200": {"description": "checkout_url"}}}}
|
|
}
|
|
}, indent=1), mimetype="application/json")
|
|
|
|
@app.route("/sitemap.xml")
|
|
def sitemap():
|
|
urls = [CFG["base_url"], f"{CFG['base_url']}/library", f"{CFG['base_url']}/pricing",
|
|
f"{CFG['base_url']}/api", f"{CFG['base_url']}/llms.txt"]
|
|
body = '<?xml version="1.0" encoding="UTF-8"?>\n<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">'
|
|
for u in urls:
|
|
body += f"<url><loc>{u}</loc></url>"
|
|
body += "</urlset>"
|
|
return Response(body, mimetype="application/xml")
|
|
|
|
# ------------------------------------------------------------------ MCP (streamable-http)
|
|
# MCP lives in its own ASGI process (mcp_server.py, uvicorn :8013); nginx routes /mcp there.
|
|
# Status reflects the MCP service health.
|
|
@app.route("/mcp-info")
|
|
def mcp_info():
|
|
mcp_ok = False
|
|
try:
|
|
r = http.post("http://127.0.0.1:8013/mcp", timeout=6,
|
|
headers={"Content-Type": "application/json",
|
|
"Accept": "application/json, text/event-stream"},
|
|
json={"jsonrpc": "2.0", "id": 1, "method": "initialize",
|
|
"params": {"protocolVersion": "2025-03-26", "capabilities": {},
|
|
"clientInfo": {"name": "mcp-info-probe", "version": "1"}}})
|
|
mcp_ok = r.status_code == 200
|
|
err = None if mcp_ok else f"mcp service http {r.status_code}"
|
|
except Exception as e:
|
|
err = str(e)
|
|
return jsonify(mcp=mcp_ok, error=err, url=f"{CFG['base_url']}/mcp")
|
|
|
|
# ------------------------------------------------------------------ static
|
|
@app.route("/static/logo.svg")
|
|
def logo():
|
|
svg = ('<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64">'
|
|
'<rect width="64" height="64" rx="12" fill="%230b0e14"/>'
|
|
'<path d="M32 8 L50 16 V34 C50 46 42 54 32 58 C22 54 14 46 14 34 V16 Z" '
|
|
'fill="none" stroke="%%23e8b64c" stroke-width="3"/>'
|
|
'<text x="32" y="40" font-size="26" text-anchor="middle" fill="%%23e8b64c" '
|
|
'font-family="monospace" font-weight="bold">D</text></svg>')
|
|
return Response(svg.replace("%23", "#"), mimetype="image/svg+xml")
|
|
|
|
# ------------------------------------------------------------------ main
|
|
if __name__ == "__main__":
|
|
app.run(host="127.0.0.1", port=CFG["port"], threaded=True)
|