diff --git a/Dockerfile b/Dockerfile index ab9cc09..5c7e4b8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,7 +2,7 @@ FROM python:3.11-slim WORKDIR /app -RUN pip install --no-cache-dir fastapi uvicorn httpx python-dotenv +RUN pip install --no-cache-dir fastapi uvicorn httpx python-dotenv psycopg2-binary COPY backend.py . diff --git a/backend.py b/backend.py index 10b51a7..c12bb98 100644 --- a/backend.py +++ b/backend.py @@ -1,547 +1,517 @@ #!/usr/bin/env python3 """ -AI Research Engine — Backend API -Runs on CT 145, handles all heavy lifting. -Exposes REST API consumed by the MCP proxy (MacBook) and dashboard. +AI Research Engine — Commercial Backend +Postgres auth, BTCPay payments, API key system, premium UI. """ -import os -import json +import os, json, hashlib, re, html as html_mod, random, secrets, time +from datetime import datetime, timezone +from urllib.parse import urlparse + import httpx -import hashlib -from fastapi import FastAPI, Query, HTTPException +import psycopg2 +import psycopg2.extras +from fastapi import FastAPI, Query, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import HTMLResponse +from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse import uvicorn -app = FastAPI(title="AI Research Engine Backend") +app = FastAPI(title="AI Research Engine") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) # ── Config ────────────────────────────────────────────────── -YACY_URL = os.getenv("YACY_URL", "http://localhost:8090") -OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "http://localhost:9200") +YACY_URL = os.getenv("YACY_URL", "http://host.docker.internal:8090") +OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "http://host.docker.internal:9200") QDRANT_URL = os.getenv("QDRANT_URL", "http://10.30.20.68:6333") OLLAMA_URL = os.getenv("OLLAMA_URL", "http://10.30.20.186:11434") OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "ornith:latest") INDEX_NAME = os.getenv("INDEX_NAME", "research_docs") +BTCPAY_URL = "https://10.30.20.140" +BTCPAY_KEY = "6026288e2e315984661c748baafd509e81a75f22" +BTCPAY_STORE = "8ERS6v1UyQ46bQqaWr4LotvMCJLsmUhH2sT8zbzfjkT6" +WEBHOOK_SECRET = "XjfhDd9DzXsUkA91B4SwRz" +PRICE_USD = 5.00 # $5 +CALLS_PER_TIER = 5 # 5 API calls + +# Proxies +PROXIES = [ + "http://10.30.20.154:3128", # Tokyo + "http://10.30.20.71:3128", # London + "http://10.30.20.189:3128", # Sydney +] + +# DB +def get_db(): + return psycopg2.connect( + host="host.docker.internal", port=5432, + user="research", password="ResearchDB2026!", + database="research_engine" + ) client = httpx.Client(timeout=30.0) ollama = httpx.Client(timeout=120.0, base_url=OLLAMA_URL) -# ── Helpers ────────────────────────────────────────────────── +# ── Auth Helpers ───────────────────────────────────────────── + +def _auth(request: Request) -> dict: + """Authenticate by API key, return user row or raise 401.""" + api_key = request.headers.get("X-API-Key") or request.query_params.get("api_key") + if not api_key: + raise HTTPException(401, "API key required. Get one at /signup or /pricing.") + db = get_db() + cur = db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) + cur.execute("SELECT * FROM users WHERE api_key = %s", (api_key,)) + user = cur.fetchone() + db.close() + if not user: + raise HTTPException(401, "Invalid API key.") + if not user["is_admin"] and user["calls_remaining"] <= 0: + raise HTTPException(402, "No calls remaining. Buy more at /pricing.") + return dict(user) + +def _track_usage(user_id: int, tool: str, request: Request): + """Deduct a call and log usage.""" + db = get_db() + cur = db.cursor() + cur.execute("UPDATE users SET calls_remaining = calls_remaining - 1, total_calls = total_calls + 1, last_used_at = NOW() WHERE id = %s AND NOT is_admin", (user_id,)) + cur.execute("INSERT INTO usage_log (user_id, tool_name, endpoint, ip_address) VALUES (%s,%s,%s,%s)", + (user_id, tool, str(request.url), request.client.host if request.client else "")) + db.commit() + db.close() + +# ── Search Helpers (unchanged from v1) ──────────────────────── def _ensure_index(): - try: - client.get(f"{OPENSEARCH_URL}/{INDEX_NAME}") - except Exception: - try: - client.put(f"{OPENSEARCH_URL}/{INDEX_NAME}", json={ - "settings": {"number_of_shards": 1, "number_of_replicas": 0}, - "mappings": {"properties": { - "url": {"type": "keyword"}, "title": {"type": "text"}, - "content": {"type": "text"}, "excerpt": {"type": "text"}, - "category": {"type": "keyword"}, "source_domain": {"type": "keyword"}, - "crawled_at": {"type": "date"}, "indexed_at": {"type": "date"}, - "metadata": {"type": "object"}, - }} - }) - except Exception: - pass - -def _ensure_qdrant(): - try: - client.get(f"{QDRANT_URL}/collections/{INDEX_NAME}") - except Exception: - try: - client.put(f"{QDRANT_URL}/collections/{INDEX_NAME}", json={ - "vectors": {"size": 768, "distance": "Cosine"} - }) - except Exception: - pass + try: client.get(f"{OPENSEARCH_URL}/{INDEX_NAME}") + except: client.put(f"{OPENSEARCH_URL}/{INDEX_NAME}", json={"settings":{"number_of_shards":1,"number_of_replicas":0},"mappings":{"properties":{"url":{"type":"keyword"},"title":{"type":"text"},"content":{"type":"text"},"excerpt":{"type":"text"},"category":{"type":"keyword"},"source_domain":{"type":"keyword"},"crawled_at":{"type":"date"},"indexed_at":{"type":"date"},"metadata":{"type":"object"}}}}) def _ai_chat(prompt: str, system: str = "You are a research assistant. Be concise and factual.") -> str: - r = ollama.post("/api/chat", json={ - "model": OLLAMA_MODEL, "messages": [ - {"role": "system", "content": system}, - {"role": "user", "content": prompt}, - ], "stream": False, - "options": {"temperature": 0.3, "num_predict": 2048}, - }) + r = ollama.post("/api/chat", json={"model":OLLAMA_MODEL,"messages":[{"role":"system","content":system},{"role":"user","content":prompt}],"stream":False,"options":{"temperature":0.3,"num_predict":2048}}) body = r.json() - return body.get("message", {}).get("content", "") or body.get("thinking", "") or "" + return body.get("message",{}).get("content","") or body.get("thinking","") or "" def _get_embedding(text: str) -> list: try: - r = ollama.post("/api/embeddings", json={"model": "nomic-embed-text-v2-moe:latest", "prompt": text[:2048]}) - return r.json().get("embedding", []) - except Exception: - return [] + r = ollama.post("/api/embeddings", json={"model":"nomic-embed-text-v2-moe:latest","prompt":text[:2048]}) + return r.json().get("embedding",[]) + except: return [] + +def _fetch_and_index(url: str, category: str = ""): + proxy_url = random.choice(PROXIES) + try: + pc = httpx.Client(proxy=proxy_url, timeout=15.0) + r = pc.get(url, headers={"User-Agent":"Mozilla/5.0 (compatible; ResearchBot/1.0)"}) + if r.status_code != 200: return None + html_text = r.text + text = re.sub(r']*>.*?','',html_text,flags=re.DOTALL|re.IGNORECASE) + text = re.sub(r']*>.*?','',text,flags=re.DOTALL|re.IGNORECASE) + text = re.sub(r'<[^>]+>',' ',text) + text = re.sub(r'\s+',' ',text).strip() + text = html_mod.unescape(text) + title_m = re.search(r']*>(.*?)',html_text,re.IGNORECASE|re.DOTALL) + title = html_mod.unescape(title_m.group(1).strip()) if title_m else url + desc_m = re.search(r']+name=["\']description["\'][^>]+content=["\']([^"\']+)',html_text,re.IGNORECASE) + excerpt = desc_m.group(1)[:500] if desc_m else text[:500] + domain = urlparse(url).netloc + if not category: + for k,v in {"wikipedia":"reference","github":"software","arxiv":"science","docs.":"documentation","blog.":"blog","news.":"news"}.items(): + if k in domain: category = v; break + if not category: category = "web" + doc = {"url":url,"title":title,"content":text[:50000],"excerpt":excerpt[:1000],"category":category,"source_domain":domain,"crawled_at":datetime.now(timezone.utc).isoformat(),"indexed_at":datetime.now(timezone.utc).isoformat()} + _ensure_index() + client.put(f"{OPENSEARCH_URL}/{INDEX_NAME}/_doc/{hashlib.md5(url.encode()).hexdigest()}",json=doc,params={"refresh":"true"}) + emb = _get_embedding(excerpt[:1000]) + if emb: + try: + client.put(f"{QDRANT_URL}/collections/{INDEX_NAME}/points",json={"points":[{"id":hashlib.md5(url.encode()).hexdigest(),"vector":emb,"payload":{"url":url,"title":title,"excerpt":excerpt[:500]}}]}) + except: pass + return {"title":title,"domain":domain,"category":category,"size":len(text)} + except: return None -# ── Status ──────────────────────────────────────────────────── +# ═══════════════════════════════════════════════════════════════ +# PUBLIC PAGES +# ═══════════════════════════════════════════════════════════════ + +@app.get("/", response_class=HTMLResponse) +def landing(): + return HTMLResponse(LANDING_HTML) + +@app.get("/pricing", response_class=HTMLResponse) +def pricing_page(): + return HTMLResponse(PRICING_HTML) + +@app.get("/signup", response_class=HTMLResponse) +def signup_page(): + return HTMLResponse(SIGNUP_HTML) + +@app.get("/dashboard", response_class=HTMLResponse) +def dashboard_page(): + return HTMLResponse(DASHBOARD_HTML) + + +# ═══════════════════════════════════════════════════════════════ +# AUTH API +# ═══════════════════════════════════════════════════════════════ + +@app.post("/api/signup") +async def api_signup(request: Request): + data = await request.json() + email = data.get("email","").strip().lower() + if not email or "@" not in email: + return JSONResponse({"error":"Valid email required"}, 400) + api_key = "sk-" + secrets.token_hex(24) + db = get_db() + cur = db.cursor() + try: + cur.execute("INSERT INTO users (email, password_hash, api_key, calls_remaining) VALUES (%s,%s,%s,0)", (email, "bcrypt_placeholder", api_key)) + db.commit() + except psycopg2.errors.UniqueViolation: + db.rollback() + cur.execute("SELECT api_key FROM users WHERE email = %s", (email,)) + api_key = cur.fetchone()[0] + db.close() + return {"email": email, "api_key": api_key, "message": "Signup successful. Purchase API calls at /pricing to start using the engine."} + +@app.get("/api/my-usage") +async def my_usage(request: Request): + # Allow even with 0 calls — users need to see their balance + api_key = request.headers.get("X-API-Key") or request.query_params.get("api_key") + if not api_key: + raise HTTPException(401, "API key required.") + db = get_db() + cur = db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) + cur.execute("SELECT * FROM users WHERE api_key = %s", (api_key,)) + user = cur.fetchone() + if not user: + db.close() + raise HTTPException(401, "Invalid API key.") + user = dict(user) + cur.execute("SELECT tool_name, COUNT(*) as cnt FROM usage_log WHERE user_id = %s GROUP BY tool_name ORDER BY cnt DESC", (user["id"],)) + user["usage_by_tool"] = [dict(r) for r in cur.fetchall()] + db.close() + return user + + +# ═══════════════════════════════════════════════════════════════ +# BTCPAY INTEGRATION +# ═══════════════════════════════════════════════════════════════ + +@app.post("/api/create-invoice") +async def create_invoice(request: Request): + """Create BTCPay invoice for $5 = 5 calls. Works even with 0 calls.""" + api_key = request.headers.get("X-API-Key") or request.query_params.get("api_key") + if not api_key: + raise HTTPException(401, "API key required.") + db = get_db() + cur = db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) + cur.execute("SELECT * FROM users WHERE api_key = %s", (api_key,)) + user = cur.fetchone() + if not user: + db.close() + raise HTTPException(401, "Invalid API key.") + user = dict(user) + try: + r = httpx.post(f"{BTCPAY_URL}/api/v1/stores/{BTCPAY_STORE}/invoices", + headers={"Authorization": f"token {BTCPAY_KEY}", "Content-Type": "application/json"}, + json={ + "amount": str(PRICE_USD), + "currency": "USD", + "metadata": { + "user_id": user["id"], + "user_email": user["email"], + "calls": CALLS_PER_TIER, + "orderId": f"research-{user['id']}-{int(time.time())}" + }, + "checkout": {"redirectURL": f"http://10.30.20.249:8000/dashboard"} + }, + verify=False, timeout=15.0 + ) + inv = r.json() + # Log pending payment + db = get_db() + cur = db.cursor() + cur.execute("INSERT INTO payments (user_id, btcpay_invoice_id, amount_usd, calls_purchased, status) VALUES (%s,%s,%s,%s,'pending')", + (user["id"], inv["id"], PRICE_USD, CALLS_PER_TIER)) + db.commit() + db.close() + return {"invoice_id": inv["id"], "checkout_url": inv["checkoutLink"], "amount": f"${PRICE_USD}", "calls": CALLS_PER_TIER} + except Exception as e: + return JSONResponse({"error": str(e)}, 500) + +@app.post("/webhook/btcpay") +async def btcpay_webhook(request: Request): + """BTCPay calls this when invoice is settled.""" + body = await request.json() + event_type = body.get("type", "") + invoice_id = body.get("invoiceId", "") + metadata = body.get("metadata", {}) + + # Handle nested metadata (BTCPay wraps it) + if isinstance(metadata, dict): + user_id = metadata.get("user_id") + calls = metadata.get("calls", CALLS_PER_TIER) + else: + user_id = None + calls = CALLS_PER_TIER + + if event_type in ("InvoiceSettled", "InvoiceProcessing") and user_id: + db = get_db() + cur = db.cursor() + cur.execute("UPDATE payments SET status = 'settled', settled_at = NOW() WHERE btcpay_invoice_id = %s", (invoice_id,)) + cur.execute("UPDATE users SET calls_remaining = calls_remaining + %s WHERE id = %s", (calls, user_id)) + db.commit() + db.close() -@app.get("/health") -def health(): return {"status": "ok"} + +# ═══════════════════════════════════════════════════════════════ +# API TOOLS (auth-gated) +# ═══════════════════════════════════════════════════════════════ + @app.get("/api/status") def status(): svc = {} - try: + try: r = client.get(f"{OPENSEARCH_URL}/_cluster/health") - cnt = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_count").json() if r.status_code == 200 else {"count": 0} - svc["opensearch"] = {"status": r.json().get("status", "down"), "documents": cnt.get("count", 0)} - except Exception: - svc["opensearch"] = {"status": "down"} + cnt = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_count").json() + svc["opensearch"] = {"status": r.json().get("status"), "documents": cnt.get("count",0)} + except: svc["opensearch"] = {"status":"down"} try: - r = client.get(f"{QDRANT_URL}/healthz") + client.get(f"{QDRANT_URL}/healthz") col = client.get(f"{QDRANT_URL}/collections/{INDEX_NAME}").json() - svc["qdrant"] = {"status": "ok", "points": col.get("result", {}).get("points_count", 0)} - except Exception: - svc["qdrant"] = {"status": "down"} + svc["qdrant"] = {"status":"ok","points":col.get("result",{}).get("points_count",0)} + except: svc["qdrant"] = {"status":"down"} try: - r = client.get(f"{YACY_URL}/yacysearch.json", params={"query": "test", "maximumRecords": 1}) - svc["yacy"] = {"status": "ok" if r.status_code == 200 else "down"} - except Exception: - svc["yacy"] = {"status": "down"} + client.get(f"{YACY_URL}/yacysearch.json",params={"query":"test","maximumRecords":1}) + svc["yacy"] = {"status":"ok"} + except: svc["yacy"] = {"status":"down"} try: r = ollama.get("/api/tags") - svc["ollama"] = {"status": "ok", "models": len(r.json().get("models", []))} - except Exception: - svc["ollama"] = {"status": "down"} - return {"services": svc} - - -# ── Search ──────────────────────────────────────────────────── + svc["ollama"] = {"status":"ok","models":len(r.json().get("models",[]))} + except: svc["ollama"] = {"status":"down"} + # DB stats + try: + db = get_db() + cur = db.cursor() + cur.execute("SELECT COUNT(*) FROM users") + users = cur.fetchone()[0] + cur.execute("SELECT COALESCE(SUM(total_calls),0) FROM users") + total = cur.fetchone()[0] + cur.execute("SELECT COUNT(*) FROM payments WHERE status = 'settled'") + payments = cur.fetchone()[0] + db.close() + svc["business"] = {"status":"ok","users":users,"total_api_calls":total,"payments_settled":payments} + except: svc["business"] = {"status":"down"} + return {"services":svc} @app.get("/api/search") -def search_web(q: str = Query(...), category: str = "", limit: int = 10): +def search_web(request: Request, q: str = Query(...), category: str = "", limit: int = 10): + user = _auth(request) + _track_usage(user["id"], "search_web", request) _ensure_index() - body = { - "size": limit, - "query": {"bool": {"must": [{"multi_match": {"query": q, "fields": ["title^3", "content", "excerpt"]}}]}}, - "highlight": {"fields": {"content": {"fragment_size": 200, "number_of_fragments": 2}}}, - } - if category: - body["query"]["bool"]["filter"] = [{"term": {"category": category}}] + body = {"size":limit,"query":{"bool":{"must":[{"multi_match":{"query":q,"fields":["title^3","content","excerpt"]}}]}},"highlight":{"fields":{"content":{"fragment_size":200,"number_of_fragments":2}}}} + if category: body["query"]["bool"]["filter"] = [{"term":{"category":category}}] try: r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json=body) result = r.json() hits = [] - for h in result.get("hits", {}).get("hits", []): + for h in result.get("hits",{}).get("hits",[]): src = h["_source"] - hits.append({"url": src.get("url"), "title": src.get("title"), - "excerpt": src.get("excerpt") or (h.get("highlight", {}).get("content", [""])[0]), - "category": src.get("category"), "crawled_at": src.get("crawled_at"), "score": h["_score"]}) - return {"query": q, "total": result.get("hits", {}).get("total", {}).get("value", 0), "hits": hits} + hits.append({"url":src.get("url"),"title":src.get("title"),"excerpt":src.get("excerpt") or (h.get("highlight",{}).get("content",[""])[0]),"category":src.get("category"),"crawled_at":src.get("crawled_at"),"score":h["_score"]}) + return {"query":q,"total":result.get("hits",{}).get("total",{}).get("value",0),"hits":hits} except Exception as e: - return {"query": q, "total": 0, "hits": [], "note": f"Index may be empty. {e}"} - + return {"query":q,"total":0,"hits":[],"note":str(e)} @app.get("/api/semantic-search") -def semantic_search(q: str = Query(...), limit: int = 10): - _ensure_qdrant() +def semantic_search(request: Request, q: str = Query(...), limit: int = 10): + user = _auth(request) + _track_usage(user["id"], "semantic_search", request) emb = _get_embedding(q) - if not emb: - return {"hits": [], "error": "Embedding model not available"} + if not emb: return {"hits":[],"error":"Embedding model unavailable"} try: - r = client.post(f"{QDRANT_URL}/collections/{INDEX_NAME}/points/search", json={ - "vector": emb, "limit": limit, "with_payload": True, "with_vector": False}) - hits = [{"url": p.get("payload", {}).get("url"), "title": p.get("payload", {}).get("title"), - "excerpt": str(p.get("payload", {}).get("excerpt", ""))[:300], "score": p.get("score")} - for p in r.json().get("result", [])] - return {"query": q, "total": len(hits), "hits": hits} + r = client.post(f"{QDRANT_URL}/collections/{INDEX_NAME}/points/search",json={"vector":emb,"limit":limit,"with_payload":True,"with_vector":False}) + hits = [{"url":p.get("payload",{}).get("url"),"title":p.get("payload",{}).get("title"),"excerpt":str(p.get("payload",{}).get("excerpt",""))[:300],"score":p.get("score")} for p in r.json().get("result",[])] + return {"query":q,"total":len(hits),"hits":hits} except Exception as e: - return {"hits": [], "error": str(e)} - - -# ── Crawl ───────────────────────────────────────────────────── - -import re -import html as html_mod -import random - -# NordVPN HTTP proxies for geo-distributed fetching -PROXIES = [ - "http://10.30.20.154:3128", # Tokyo, Japan - "http://10.30.20.71:3128", # London, UK - "http://10.30.20.189:3128", # Sydney, Australia -] - -def _fetch_and_index(url: str, category: str = ""): - """Fetch a URL, extract text, and index into OpenSearch immediately.""" - proxy_url = random.choice(PROXIES) - try: - # Create a per-request client with proxy - proxy_client = httpx.Client(proxy=proxy_url, timeout=15.0) - r = proxy_client.get(url, - headers={"User-Agent": "Mozilla/5.0 (compatible; ResearchBot/1.0)"}, - ) - if r.status_code != 200: - return None - html_text = r.text - # Basic HTML-to-text - text = re.sub(r']*>.*?', '', html_text, flags=re.DOTALL|re.IGNORECASE) - text = re.sub(r']*>.*?', '', text, flags=re.DOTALL|re.IGNORECASE) - text = re.sub(r'<[^>]+>', ' ', text) - text = re.sub(r'\s+', ' ', text).strip() - text = html_mod.unescape(text) - # Extract title - title_match = re.search(r']*>(.*?)', html_text, re.IGNORECASE|re.DOTALL) - title = html_mod.unescape(title_match.group(1).strip()) if title_match else url - # Extract meta description - desc_match = re.search(r']+name=["\']description["\'][^>]+content=["\']([^"\']+)', html_text, re.IGNORECASE) - excerpt = desc_match.group(1)[:500] if desc_match else text[:500] - # Derive domain + category - from urllib.parse import urlparse - domain = urlparse(url).netloc - if not category: - cat_map = {"wikipedia": "reference", "github": "software", "arxiv": "science", - "docs.": "documentation", "blog.": "blog", "news.": "news"} - for k, v in cat_map.items(): - if k in domain: - category = v - break - if not category: - category = "web" - # Index into OpenSearch - import datetime - doc = { - "url": url, "title": title, "content": text[:50000], - "excerpt": excerpt[:1000], "category": category, - "source_domain": domain, - "crawled_at": datetime.datetime.utcnow().isoformat(), - "indexed_at": datetime.datetime.utcnow().isoformat(), - } - _ensure_index() - client.put(f"{OPENSEARCH_URL}/{INDEX_NAME}/_doc/{hashlib.md5(url.encode()).hexdigest()}", - json=doc, params={"refresh": "true"}) - # Also index into Qdrant - emb = _get_embedding(excerpt[:1000]) - if emb: - try: - _ensure_qdrant() - client.put(f"{QDRANT_URL}/collections/{INDEX_NAME}/points", json={ - "points": [{"id": hashlib.md5(url.encode()).hexdigest(), - "vector": emb, "payload": {"url": url, "title": title, "excerpt": excerpt[:500]}}] - }) - except Exception: - pass - return {"title": title, "domain": domain, "category": category, "size": len(text)} - except Exception as e: - return None - + return {"hits":[],"error":str(e)} @app.get("/api/crawl") -def crawl_url(url: str = Query(...), depth: int = 1): - # 1. Fetch and index immediately into OpenSearch + Qdrant +def crawl_url(request: Request, url: str = Query(...), depth: int = 1): + user = _auth(request) + _track_usage(user["id"], "crawl_url", request) indexed = _fetch_and_index(url) - # 2. Also submit to YaCy for deeper crawling try: - r = client.get(f"{YACY_URL}/Crawler_p.json", params={ - "crawlingDomMaxPages": 50, "crawlingDepth": depth, - "crawlingStart": url, "crawlingQ": "on", - "bookmarkTitle": "research", "bookmarkFolder": "/research", - "indexText": "on", "indexMedia": "on", - "crawlingMode": "url", "cachePolicy": "iffresh", - }) - return {"status": "crawl_started", "url": url, "depth": depth, "indexed": indexed} - except Exception as e: - return {"status": "indexed_only", "url": url, "indexed": indexed, "yacy_error": str(e)} + client.get(f"{YACY_URL}/Crawler_p.json", params={"crawlingDomMaxPages":50,"crawlingDepth":depth,"crawlingStart":url,"crawlingQ":"on","bookmarkTitle":"research","bookmarkFolder":"/research","indexText":"on","indexMedia":"on","crawlingMode":"url","cachePolicy":"iffresh"}) + return {"status":"crawl_started","url":url,"depth":depth,"indexed":indexed} + except: + return {"status":"indexed_only","url":url,"indexed":indexed} - -@app.get("/api/crawl-topic") -def crawl_topic(topic: str = Query(...), max_urls: int = 20): - discovered = [] +@app.get("/api/research") +def research_topic(request: Request, topic: str = Query(...)): + user = _auth(request) + _track_usage(user["id"], "research_topic", request) + steps = []; kw_result = {"hits":[]}; sem_result = {"hits":[]} try: - r = client.get(f"{YACY_URL}/yacysearch.json", params={"query": topic, "maximumRecords": max_urls, "resource": "global"}) - for ch in r.json().get("channels", []): - for item in ch.get("items", []): - if item.get("link"): - discovered.append({"url": item["link"], "title": item.get("title", ""), "snippet": item.get("description", "")}) - except Exception as e: - return {"topic": topic, "error": f"Discovery failed: {e}", "urls_discovered": 0, "urls_crawled": 0} - - crawled = 0 - for u in discovered[:max_urls]: - try: - client.get(f"{YACY_URL}/Crawler_p.json", params={ - "crawlingDomMaxPages": 10, "crawlingDepth": 0, - "crawlingStart": u["url"], "crawlingQ": "on", - "indexText": "on", "indexMedia": "on", - "crawlingMode": "url", "cachePolicy": "iffresh", - }, timeout=5.0) - crawled += 1 - except Exception: - pass - - return {"topic": topic, "urls_discovered": len(discovered), "urls_crawled": crawled} - - -# ── Document Retrieval ──────────────────────────────────────── - -@app.get("/api/document") -def retrieve_document(url: str = Query(...)): + r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size":5,"query":{"multi_match":{"query":topic,"fields":["title^3","content","excerpt"]}},"highlight":{"fields":{"content":{"fragment_size":200,"number_of_fragments":1}}}}) + hits = [{"title":h["_source"].get("title"),"excerpt":h.get("highlight",{}).get("content",[h["_source"].get("excerpt","")])[0]} for h in r.json().get("hits",{}).get("hits",[])] + kw_result = {"total":len(hits),"hits":hits}; steps.append("keyword_search_done") + except: steps.append("keyword_search_skipped") try: - r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={ - "size": 1, "query": {"term": {"url": url}}}) - hits = r.json().get("hits", {}).get("hits", []) - if not hits: - return {"error": "Not found", "url": url} - src = hits[0]["_source"] - return {"url": src.get("url"), "title": src.get("title"), - "content": src.get("content", "")[:10000], "excerpt": src.get("excerpt"), - "category": src.get("category"), "crawled_at": src.get("crawled_at")} - except Exception as e: - return {"error": str(e), "url": url} - - -# ── AI Synthesis ────────────────────────────────────────────── - -@app.get("/api/summarize") -def summarize_sources(urls: str = Query(...), instruction: str = "Summarize key points"): - url_list = [u.strip() for u in urls.split(",") if u.strip()] - combined = "" - for url in url_list[:5]: - try: - r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size": 1, "query": {"term": {"url": url}}}) - hits = r.json().get("hits", {}).get("hits", []) - if hits: - src = hits[0]["_source"] - combined += f"\n\n--- {url} ---\n{src.get('title','')}\n{src.get('content', src.get('excerpt',''))[:2000]}" - except Exception: - pass - if not combined.strip(): - return {"error": "No content retrieved"} - summary = _ai_chat(f"Instruction: {instruction}\n\nSources:{combined}\n\nProvide a structured summary.") - return {"instruction": instruction, "sources": len(url_list), "summary": summary} - - -@app.get("/api/extract") -def extract_information(url: str = Query(...), schema: str = Query("company names, products, prices")): + emb = _get_embedding(topic) + if emb: + r = client.post(f"{QDRANT_URL}/collections/{INDEX_NAME}/points/search",json={"vector":emb,"limit":5,"with_payload":True}) + hits2 = [{"title":p.get("payload",{}).get("title",""),"excerpt":str(p.get("payload",{}).get("excerpt",""))[:200],"score":p.get("score")} for p in r.json().get("result",[])] + sem_result = {"total":len(hits2),"hits":hits2}; steps.append("semantic_search_done") + except: steps.append("semantic_search_skipped") try: - r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size": 1, "query": {"term": {"url": url}}}) - hits = r.json().get("hits", {}).get("hits", []) - if not hits: - return {"error": "Not found", "url": url} - content = hits[0]["_source"].get("content", "")[:8000] - prompt = f"Extract: {schema}\n\nDocument:\n{content}\n\nReturn ONLY valid JSON." - result = _ai_chat(prompt, system="Extract structured data. Return ONLY valid JSON. No explanation.") - return {"url": url, "schema": schema, "extracted": result} - except Exception as e: - return {"error": str(e)} - + r = client.get(f"{YACY_URL}/yacysearch.json",params={"query":topic,"maximumRecords":10,"resource":"global"}) + discovered = [item.get("link") for ch in r.json().get("channels",[]) for item in ch.get("items",[]) if item.get("link")] + for url in discovered[:10]: + try: client.get(f"{YACY_URL}/Crawler_p.json",params={"crawlingDomMaxPages":10,"crawlingDepth":0,"crawlingStart":url,"crawlingQ":"on","indexText":"on","indexMedia":"on","crawlingMode":"url","cachePolicy":"iffresh"},timeout=5.0) + except: pass + steps.append(f"crawl_dispatched_{len(discovered[:10])}") + except: steps.append("crawl_skipped") + all_src = "" + for h in kw_result.get("hits",[])[:3] + sem_result.get("hits",[])[:3]: + all_src += f"- {h.get('title','Unknown')}: {h.get('excerpt','')[:200]}\n" + summary = _ai_chat(f"Research topic: {topic}\n\nSources:\n{all_src}\n\nConcise research summary (3-5 paragraphs): key findings, important sources, knowledge gaps, next steps.") if all_src else "" + return {"topic":topic,"steps":steps,"keyword_results":kw_result,"semantic_results":sem_result,"ai_summary":summary} @app.get("/api/report") -def create_report(topic: str = Query(...), sources: str = ""): +def create_report(request: Request, topic: str = Query(...), sources: str = ""): + user = _auth(request) + _track_usage(user["id"], "create_report", request) if sources: urls = [u.strip() for u in sources.split(",") if u.strip()] else: try: - r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={ - "size": 8, "query": {"multi_match": {"query": topic, "fields": ["title^3", "content"]}}}) - urls = [h["_source"]["url"] for h in r.json().get("hits", {}).get("hits", [])] - except Exception: - urls = [] - + r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size":8,"query":{"multi_match":{"query":topic,"fields":["title^3","content"]}}}) + urls = [h["_source"]["url"] for h in r.json().get("hits",{}).get("hits",[])] + except: urls = [] gathered = "" for url in urls[:8]: try: - r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size": 1, "query": {"term": {"url": url}}}) - hits = r.json().get("hits", {}).get("hits", []) + r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size":1,"query":{"term":{"url":url}}}) + hits = r.json().get("hits",{}).get("hits",[]) if hits: src = hits[0]["_source"] gathered += f"\n\n### {src.get('title','Source')}\nURL: {url}\n{src.get('content',src.get('excerpt',''))[:1500]}" - except Exception: - pass - - prompt = f"""Research topic: {topic}\nSources:{gathered if gathered else ' No sources found.'} - -Generate a comprehensive report: -1. Executive Summary -2. Key Findings (numbered) -3. Source Analysis -4. Knowledge Gaps -5. Recommendations - -Be thorough, use markdown, cite sources.""" - report = _ai_chat(prompt, system="You are a senior research analyst. Produce thorough, structured reports.") - return {"topic": topic, "sources_used": len(urls), "report": report} - - -# ── Research Pipeline ───────────────────────────────────────── - -@app.get("/api/research") -def research_topic(topic: str = Query(...)): - steps = [] - kw_result = {"hits": []} - sem_result = {"hits": []} + except: pass + report = _ai_chat(f"Research topic: {topic}\nSources:{gathered if gathered else ' No sources found.'}\n\nGenerate a comprehensive report:\n1. Executive Summary\n2. Key Findings (numbered)\n3. Source Analysis\n4. Knowledge Gaps\n5. Recommendations\n\nBe thorough, use markdown, cite sources.", system="You are a senior research analyst. Produce thorough, structured reports.") + return {"topic":topic,"sources_used":len(urls),"report":report} +@app.get("/api/document") +def retrieve_document(request: Request, url: str = Query(...)): + user = _auth(request) + _track_usage(user["id"], "retrieve_document", request) try: - r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={ - "size": 5, "query": {"multi_match": {"query": topic, "fields": ["title^3", "content", "excerpt"]}}, - "highlight": {"fields": {"content": {"fragment_size": 200, "number_of_fragments": 1}}}, - }) - hits = [{"title": h["_source"].get("title"), "excerpt": h.get("highlight", {}).get("content", [h["_source"].get("excerpt", "")])[0]} - for h in r.json().get("hits", {}).get("hits", [])] - kw_result = {"total": len(hits), "hits": hits} - steps.append("keyword_search_done") - except Exception: - steps.append("keyword_search_skipped") + r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size":1,"query":{"term":{"url":url}}}) + hits = r.json().get("hits",{}).get("hits",[]) + if not hits: return {"error":"Not found","url":url} + src = hits[0]["_source"] + return {"url":src.get("url"),"title":src.get("title"),"content":src.get("content","")[:10000],"excerpt":src.get("excerpt"),"category":src.get("category"),"crawled_at":src.get("crawled_at")} + except Exception as e: + return {"error":str(e),"url":url} +@app.get("/api/summarize") +def summarize_sources(request: Request, urls: str = Query(...), instruction: str = "Summarize key points"): + user = _auth(request) + _track_usage(user["id"], "summarize_sources", request) + url_list = [u.strip() for u in urls.split(",") if u.strip()] + combined = "" + for url in url_list[:5]: + try: + r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size":1,"query":{"term":{"url":url}}}) + hits = r.json().get("hits",{}).get("hits",[]) + if hits: + src = hits[0]["_source"] + combined += f"\n\n--- {url} ---\n{src.get('title','')}\n{src.get('content',src.get('excerpt',''))[:2000]}" + except: pass + if not combined.strip(): return {"error":"No content retrieved"} + summary = _ai_chat(f"Instruction: {instruction}\n\nSources:{combined}\n\nProvide a structured summary.") + return {"instruction":instruction,"sources":len(url_list),"summary":summary} + +@app.get("/api/extract") +def extract_information(request: Request, url: str = Query(...), schema: str = Query("company names, products, prices")): + user = _auth(request) + _track_usage(user["id"], "extract_information", request) try: - emb = _get_embedding(topic) - if emb: - r = client.post(f"{QDRANT_URL}/collections/{INDEX_NAME}/points/search", json={ - "vector": emb, "limit": 5, "with_payload": True}) - hits2 = [{"title": p.get("payload", {}).get("title", ""), "excerpt": str(p.get("payload", {}).get("excerpt", ""))[:200], - "score": p.get("score")} for p in r.json().get("result", [])] - sem_result = {"total": len(hits2), "hits": hits2} - steps.append("semantic_search_done") - except Exception: - steps.append("semantic_search_skipped") - - try: - r = client.get(f"{YACY_URL}/yacysearch.json", params={"query": topic, "maximumRecords": 10, "resource": "global"}) - discovered = [item.get("link") for ch in r.json().get("channels", []) for item in ch.get("items", []) if item.get("link")] - for url in discovered[:10]: - try: - client.get(f"{YACY_URL}/Crawler_p.json", params={ - "crawlingDomMaxPages": 10, "crawlingDepth": 0, "crawlingStart": url, - "crawlingQ": "on", "indexText": "on", "indexMedia": "on", - "crawlingMode": "url", "cachePolicy": "iffresh"}, timeout=5.0) - except Exception: - pass - steps.append(f"crawl_dispatched_{len(discovered[:10])}") - except Exception: - steps.append("crawl_skipped") - - all_sources = "" - for h in kw_result.get("hits", [])[:3] + sem_result.get("hits", [])[:3]: - all_sources += f"- {h.get('title', 'Unknown')}: {h.get('excerpt', '')[:200]}\n" - - summary = "" - if all_sources: - summary = _ai_chat( - f"Research topic: {topic}\n\nSources:\n{all_sources}\n\nConcise research summary (3-5 paragraphs): " - "key findings, important sources, knowledge gaps, next steps.") - - return {"topic": topic, "steps": steps, - "keyword_results": kw_result, "semantic_results": sem_result, - "ai_summary": summary} + r = client.post(f"{OPENSEARCH_URL}/{INDEX_NAME}/_search", json={"size":1,"query":{"term":{"url":url}}}) + hits = r.json().get("hits",{}).get("hits",[]) + if not hits: return {"error":"Not found","url":url} + content = hits[0]["_source"].get("content","")[:8000] + result = _ai_chat(f"Extract: {schema}\n\nDocument:\n{content}\n\nReturn ONLY valid JSON.", system="Extract structured data. Return ONLY valid JSON.") + return {"url":url,"schema":schema,"extracted":result} + except Exception as e: + return {"error":str(e)} -# ── Dashboard ───────────────────────────────────────────────── +# ═══════════════════════════════════════════════════════════════ +# UI TEMPLATES +# ═══════════════════════════════════════════════════════════════ -@app.get("/", response_class=HTMLResponse) -def dashboard(): - return HTMLResponse(""" - -AI Research Engine - -

🧠 AI Research Engine

Self-hosted · Private
-
-

Your Private Research Cloud

Discover, index, and synthesize knowledge — all on your own infrastructure.

-
- - - -
-
- - - -
- -
-
-
-""") +.btn-primary:hover{background:var(--accent2);transform:translateY(-1px);box-shadow:0 8px 25px rgba(124,58,237,0.3)} +.btn-gold{background:linear-gradient(135deg,var(--gold),#fbbf24);color:#1a1a1a} +.btn-gold:hover{transform:translateY(-1px);box-shadow:0 8px 25px rgba(245,158,11,0.3)} +.btn-outline{background:transparent;border:1px solid var(--border);color:var(--text)} +.container{max-width:1200px;margin:0 auto;padding:40px 24px} +.hero{text-align:center;padding:80px 0 60px} +.hero h2{font-size:3rem;font-weight:900;letter-spacing:-0.04em;line-height:1.1;margin-bottom:20px;background:linear-gradient(135deg,var(--text),var(--accent2));-webkit-background-clip:text;-webkit-text-fill-color:transparent} +.hero p{color:var(--muted);font-size:1.2rem;max-width:650px;margin:0 auto 32px;line-height:1.6} +.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(280px,1fr));gap:20px;margin:40px 0} +.card{background:var(--surface);border:1px solid var(--border);border-radius:16px;padding:28px;transition:all .2s} +.card:hover{border-color:var(--accent);transform:translateY(-2px)} +.card h3{font-size:1.2rem;margin-bottom:10px} +.card p{color:var(--muted);font-size:0.9rem;line-height:1.5} +.price{font-size:3rem;font-weight:900;text-align:center;margin:20px 0} +.price span{font-size:1rem;color:var(--muted);font-weight:400} +.feature-list{list-style:none;margin:20px 0} +.feature-list li{padding:8px 0;color:var(--muted);font-size:0.9rem} +.feature-list li::before{content:'✓ ';color:var(--green);font-weight:700} +pre{background:var(--surface);border:1px solid var(--border);border-radius:12px;padding:16px;overflow-x:auto;font-size:0.85rem;margin:10px 0} +.footer{text-align:center;padding:40px;color:var(--muted);font-size:0.85rem;border-top:1px solid var(--border);margin-top:60px} +.footer a{color:var(--accent2);text-decoration:none} +input,textarea{width:100%;padding:14px 18px;border-radius:12px;border:1px solid var(--border);background:var(--bg);color:var(--text);font-size:1rem;outline:none;transition:border-color .2s;margin-bottom:12px} +input:focus,textarea:focus{border-color:var(--accent)} +""" + +LANDING_HTML = f"""AI Research Engine — Private Knowledge Cloud

🧠 AI Research Engine

The Ultimate Source of
Uncensored Internet Knowledge

A self-hosted AI workforce that continuously gathers, organizes, and acts on information. Your private research cloud — no filters, no tracking, no limits.

🔍 Deep Search

Full-text search across millions of indexed documents. Find anything — no Google bubble, no censorship.

🧬 Semantic Understanding

Search by meaning, not keywords. Our vector engine finds conceptually related content even when the words differ.

🕷️ Autonomous Crawling

Point it at a topic and it discovers, crawls, and indexes every relevant source. Never miss a thing.

🤖 AI Synthesis

Local LLMs summarize, extract, and report. Your research, analyzed by AI running on our hardware.

🌍 Geo-Distributed

Crawls route through proxies in Tokyo, London, and Sydney. Bypass regional blocks automatically.

🔐 Private & Self-Hosted

Everything runs on our metal. No third-party APIs, no data harvesting, no surveillance.

Trusted by AI Agents Worldwide

10 MCP tools. One API key. Infinite knowledge. Agents connect and research autonomously — no browser needed.

Created by drjones · ☕ Buy Me a Coffee

AI Research Engine v2.0 · Self-hosted · No KYC · Bitcoin Only

""" + +PRICING_HTML = f"""Pricing — AI Research Engine

🧠 AI Research Engine

Simple Bitcoin Pricing

No subscriptions. No KYC. Pay with Bitcoin, get API calls.

🚀 Researcher Tier

$5 USD

in Bitcoin · Lightning ⚡

  • 5 API calls
  • Full access to all 10 tools
  • Search, crawl, research, report
  • No expiration
  • No KYC required

👇 Sign up first, then buy from your dashboard

Sign Up & Get Your API Key

⚡ Lightning Fast

Payments settle in seconds. Your API calls are credited instantly.

🔑 Bring Your Own Key

Use your API key in any MCP client, script, or AI agent.

📊 Track Usage

Real-time dashboard shows your remaining calls and history.

Created by drjones · ☕ Buy Me a Coffee

""" + +SIGNUP_HTML = f"""Sign Up — AI Research Engine

🧠 AI Research Engine

Get Your API Key

No KYC. Just an email. Your key is generated instantly.

Created by drjones

""" + +DASHBOARD_HTML = f"""Dashboard — AI Research Engine

🧠 AI Research Engine

Dashboard

Your account, usage, and API access.

🔑 Enter Your API Key

Created by drjones

""" + + +# ── Startup ────────────────────────────────────────────────── if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/server.py b/server.py index 519e873..06c5b21 100644 --- a/server.py +++ b/server.py @@ -1,110 +1,81 @@ #!/usr/bin/env python3 -""" -AI Research Engine — Thin MCP Proxy -Runs on MacBook. Forwards all tool calls to the CT 145 backend. -Minimal resource usage — all heavy lifting on Proxmox. -""" +"""AI Research Engine — Thin MCP Proxy. Admin key for backend auth.""" import json import httpx from mcp.server import FastMCP BACKEND_URL = "http://10.30.20.249:8000" +ADMIN_KEY = "sk-admin-unlimited-2026" client = httpx.Client(timeout=120.0) -mcp = FastMCP( - "ai-research-engine", - instructions=""" -AI Research Engine — private knowledge acquisition system. - -search_web(query) — Full-text search across indexed documents -semantic_search(query) — Find documents by meaning (vector search) -crawl_url(url) — Crawl a URL into the index -crawl_topic(topic) — Discover and crawl sources for a topic -research_topic(topic) — Full pipeline: discover → crawl → summarize -retrieve_document(url) — Get full content of an indexed document -summarize_sources(urls, instruction) — AI summary of multiple sources -extract_information(url, schema) — Structured data extraction -create_report(topic, sources) — Generate comprehensive research report -index_status() — System health and stats -""", -) - def _get(path: str) -> dict: - r = client.get(f"{BACKEND_URL}{path}") + sep = "&" if "?" in path else "?" + r = client.get(f"{BACKEND_URL}{path}{sep}api_key={ADMIN_KEY}") r.raise_for_status() return r.json() +mcp = FastMCP("ai-research-engine", instructions="AI Research Engine — private knowledge acquisition. 10 tools.") @mcp.tool() def search_web(query: str, category: str = "", limit: int = 10) -> str: - """Full-text search across indexed documents. Find by keywords, titles, content.""" + """Full-text search across indexed documents.""" r = _get(f"/api/search?q={query}&category={category}&limit={limit}") return json.dumps(r, indent=2) - @mcp.tool() def semantic_search(query: str, limit: int = 10) -> str: - """Search by meaning using vector embeddings. Finds conceptually related docs.""" + """Search by meaning using vector embeddings.""" r = _get(f"/api/semantic-search?q={query}&limit={limit}") return json.dumps(r, indent=2) - @mcp.tool() def crawl_url(url: str, depth: int = 1) -> str: - """Crawl a URL. depth: 0=just this page, 1=+linked pages.""" + """Crawl a URL and index it.""" r = _get(f"/api/crawl?url={url}&depth={depth}") return json.dumps(r, indent=2) - @mcp.tool() def crawl_topic(topic: str, max_urls: int = 20) -> str: - """Discover and crawl sources for a topic using YaCy.""" + """Discover and crawl sources for a topic.""" r = _get(f"/api/crawl-topic?topic={topic}&max_urls={max_urls}") return json.dumps(r, indent=2) - @mcp.tool() def research_topic(topic: str) -> str: - """Full research pipeline: keyword search → semantic search → crawl new sources → AI summary.""" + """Full pipeline: search → crawl → AI summary.""" r = _get(f"/api/research?topic={topic}") return json.dumps(r, indent=2) - @mcp.tool() def retrieve_document(url: str) -> str: - """Get full indexed content of a document by URL.""" + """Get full indexed content of a document.""" r = _get(f"/api/document?url={url}") return json.dumps(r, indent=2) - @mcp.tool() def summarize_sources(urls: str, instruction: str = "Summarize key points") -> str: - """Summarize multiple URLs using local LLM. urls: comma-separated.""" + """AI summary of multiple URLs. urls: comma-separated.""" r = _get(f"/api/summarize?urls={urls}&instruction={instruction}") return json.dumps(r, indent=2) - @mcp.tool() -def extract_information(url: str, schema: str = "company names, products, prices, specifications") -> str: - """Extract structured information from a document using LLM.""" +def extract_information(url: str, schema: str = "company names, products, prices") -> str: + """Extract structured data from a document using LLM.""" r = _get(f"/api/extract?url={url}&schema={schema}") return json.dumps(r, indent=2) - @mcp.tool() def create_report(topic: str, sources: str = "") -> str: - """Generate a comprehensive research report. sources: optional comma-separated URLs.""" + """Generate comprehensive research report.""" r = _get(f"/api/report?topic={topic}&sources={sources}") return json.dumps(r, indent=2) - @mcp.tool() def index_status() -> str: - """Check health of all backend services: OpenSearch, Qdrant, YaCy, Ollama.""" + """System health + business stats.""" r = _get("/api/status") return json.dumps(r, indent=2) - if __name__ == "__main__": mcp.run(transport="stdio")