""" dev.py — AI Answers Plugin Development Server Provides a full-featured local dev environment for ollama_answers.py. Features: - Mock SearXNG environment so the plugin loads without a real SearXNG install - Optional live SearXNG backend (set SEARXNG_URL to proxy real searches) - Debug console with request/response logging - Plugin API explorer (all routes: /ai-stream, /ai-status, /ai-models, /ai-conversation, /ai-auxiliary-search) - Dark UI matching SearXNG's default theme with real CSS vars - Config panel: LLM endpoint, model, token limits, temperature - Live query testing with streaming output in the browser Usage: cd dev/ python dev.py Or with a live SearXNG backend: SEARXNG_URL=http://localhost:8080 python dev.py Environment variables (all optional): LLM_URL Ollama endpoint (default: http://localhost:11434/v1/chat/completions) LLM_MODEL Model name (default: qwen3.5:9b) LLM_MAX_TOKENS Max tokens (default: 200) LLM_TEMPERATURE Temperature (default: 0.2) SEARXNG_URL Live SearXNG URL (default: None, uses mock results) DEV_PORT Port to listen on (default: 5000) DEV_HOST Host to bind to (default: 127.0.0.1) """ import sys import os import json import time import logging import hashlib import secrets import urllib.request import urllib.parse from types import ModuleType # ── path setup ────────────────────────────────────────────────────────────── sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # ── .env loading ───────────────────────────────────────────────────────────── try: from dotenv import load_dotenv # look for .env in the project root (one level up from dev/) _env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.env') if os.path.exists(_env_path): load_dotenv(_env_path) print(f" Loaded .env from {_env_path}") except ImportError: pass # python-dotenv not installed, ignore # ── logging ────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.DEBUG, format='%(asctime)s %(levelname)-8s %(name)s: %(message)s', datefmt='%H:%M:%S' ) logger = logging.getLogger('dev') # ── env defaults ───────────────────────────────────────────────────────────── os.environ.setdefault('LLM_URL', 'http://localhost:11434/v1/chat/completions') os.environ.setdefault('LLM_MODEL', 'qwen3.5:9b') SEARXNG_URL = os.getenv('SEARXNG_URL', '').rstrip('/') DEV_PORT = int(os.getenv('DEV_PORT', 5000)) DEV_HOST = os.getenv('DEV_HOST', '127.0.0.1') # ── SearXNG module mocks ───────────────────────────────────────────────────── searx_mod = ModuleType("searx") searx_plugins_mod = ModuleType("searx.plugins") searx_results_mod = ModuleType("searx.result_types") searx_network_mod = ModuleType("searx.network") class _MockPlugin: def __init__(self, cfg): self.active = getattr(cfg, 'active', True) class _MockPluginInfo: def __init__(self, **kwargs): self.meta = kwargs class _MockEngineResults: def __init__(self): self.types = ModuleType("types") self.types.Answer = lambda *a, **kw: kw.get('answer', a[0] if a else "") self._results = [] def add(self, res): self._results.append(res) searx_plugins_mod.Plugin = _MockPlugin searx_plugins_mod.PluginInfo = _MockPluginInfo searx_results_mod.EngineResults = _MockEngineResults searx_mod.settings = {'server': {'secret_key': 'dev-secret-key'}} # network mock — forwards real HTTP so Ollama calls work import http.client, ssl as _ssl def _real_http(method, url, **kwargs): from urllib.parse import urlparse p = urlparse(url) port = p.port or (443 if p.scheme == 'https' else 80) path = p.path + (f'?{p.query}' if p.query else '') if p.scheme == 'https': conn = http.client.HTTPSConnection(p.hostname, port, timeout=30, context=_ssl.create_default_context()) else: conn = http.client.HTTPConnection(p.hostname, port, timeout=30) headers = kwargs.get('headers', {}) body = json.dumps(kwargs['json']).encode() if kwargs.get('json') else kwargs.get('data') if kwargs.get('params'): qs = urllib.parse.urlencode(kwargs['params']) path += ('&' if '?' in path else '?') + qs conn.request(method, path, body=body, headers=headers) return conn.getresponse() class _MockHTTPResponse: def __init__(self, r): self.status_code = r.status self._content = r.read() self.text = self._content.decode('utf-8', errors='replace') def json(self): return json.loads(self.text) def _mock_get(url, **kw): return _MockHTTPResponse(_real_http('GET', url, **kw)) def _mock_stream(method, url, **kw): r = _real_http(method, url, **kw) class _Resp: status_code = r.status text = "streaming" def _gen(): while True: chunk = r.read(256) if not chunk: break yield chunk return _Resp(), _gen() searx_network_mod.stream = _mock_stream searx_network_mod.get = _mock_get sys.modules['searx'] = searx_mod sys.modules['searx.plugins'] = searx_plugins_mod sys.modules['searx.result_types'] = searx_results_mod sys.modules['searx.network'] = searx_network_mod # ── log capture handler for intent / score extraction ──────────────────────── class _ScoreCaptureHandler(logging.Handler): """Captures plugin log messages so dev routes can extract intent/scores.""" def __init__(self): super().__init__() self.records = [] def emit(self, record): self.records.append(self.format(record)) def clear(self): self.records.clear() _score_handler = _ScoreCaptureHandler() _score_handler.setFormatter(logging.Formatter('%(message)s')) logging.getLogger('ollama_answers').addHandler(_score_handler) # ── load plugin ─────────────────────────────────────────────────────────────── from flask import Flask, request, redirect, jsonify, Response from flask_babel import Babel from ollama_answers import SXNGPlugin app = Flask(__name__) babel = Babel(app) class _MockCfg: active = True plugin = SXNGPlugin(_MockCfg()) plugin.init(app) # Override api_key from env if set env_key = os.getenv('LLM_API_KEY', '').strip() if env_key: plugin.api_key = env_key # Verify Valkey is reachable try: import valkey as _vk _vk_host = os.getenv('VALKEY_HOST', 'localhost') _vk_port = int(os.getenv('VALKEY_PORT', 6379)) _test_vk = _vk.Valkey(host=_vk_host, port=_vk_port, socket_connect_timeout=2) _test_vk.ping() print(f" Valkey: {_vk_host}:{_vk_port} ✓") except Exception as e: print(f""" ⚠ Valkey not reachable at {_vk_host}:{_vk_port} Streaming will not work without Valkey. Start one with Docker: docker run -d --name dev-valkey -p 6379:6379 valkey/valkey:9-alpine Or set VALKEY_HOST and VALKEY_PORT in your .env file. Error: {e} """) # ── request/response debug log storage ─────────────────────────────────────── _DEV_SESSION = secrets.token_hex(12) _dev_log = [] # list of {ts, method, path, status, ms, req_body, resp_body} MAX_LOG = 100 @app.before_request def _log_req(): request._dev_start = time.time() request._dev_body = request.get_data(as_text=True) @app.after_request def _log_resp(resp): if request.path.startswith('/static'): return resp ms = round((time.time() - getattr(request, '_dev_start', time.time())) * 1000, 1) entry = { 'ts': time.strftime('%H:%M:%S'), 'method': request.method, 'path': request.full_path.rstrip('?'), 'status': resp.status_code, 'ms': ms, 'req': getattr(request, '_dev_body', ''), 'resp': resp.get_data(as_text=True)[:2000], } _dev_log.insert(0, entry) if len(_dev_log) > MAX_LOG: _dev_log.pop() return resp # ── mock search results ─────────────────────────────────────────────────────── MOCK_RESULTS = { 'default': [ {"title": "Wikipedia — General Knowledge", "content": "A comprehensive reference covering the query topic in depth with cited sources.", "url": "https://en.wikipedia.org/wiki/Main_Page", "publishedDate": "2026-01-15"}, {"title": "NASA Science — Research Article", "content": "Peer-reviewed research providing scientific context and empirical data on the topic.", "url": "https://science.nasa.gov/", "publishedDate": "2026-01-10"}, {"title": "Scientific American — Analysis", "content": "In-depth analysis covering recent discoveries and expert commentary.", "url": "https://scientificamerican.com/", "publishedDate": "2026-01-01"}, {"title": "BBC Science — Overview", "content": "Accessible explainer covering the fundamentals and current understanding.", "url": "https://bbc.com/science", "publishedDate": "2025-12-20"}, {"title": "MIT OpenCourseWare — Educational Resource", "content": "Course materials and lecture notes providing structured educational content.", "url": "https://ocw.mit.edu/", "publishedDate": "2025-12-15"}, ] + [ {"title": f"Source {i} — Additional Context", "content": f"Supplementary information and supporting details from source {i}.", "url": f"https://example.com/source-{i}", "publishedDate": "2025-12-01"} for i in range(6, 16) ] } def _get_results(query: str): """Return real results from SearXNG if configured, else mock results.""" if SEARXNG_URL: try: qs = urllib.parse.urlencode({'q': query, 'format': 'json', 'categories': 'general'}) url = f"{SEARXNG_URL}/search?{qs}" with urllib.request.urlopen(url, timeout=8) as r: data = json.loads(r.read()) results = data.get('results', []) logger.info(f"Live SearXNG: {len(results)} results for '{query}'") return results, data.get('infoboxes', []), data.get('answers', []) except Exception as e: logger.warning(f"Live SearXNG failed ({e}), falling back to mock results") return MOCK_RESULTS['default'], [], [] # ── search route used by the demo UI ───────────────────────────────────────── _last_render = {'html': '', 'query': '', 'scores': [], 'intent': '—', 'summary': ''} @app.route('/dev/search') def dev_search(): query = request.args.get('q', 'why is the sky blue') global _last_render _last_render['intent'] = '…' _last_render['query'] = query class _MockSearch: class _MockSearchQuery: pageno = 1 lang = 'en-US' categories = ['general'] search_query = _MockSearchQuery() search_query.query = query class _MockResultContainer: def __init__(self, q): self.answers = set() self._q = q self.infoboxes = [] def get_ordered_results(self): results, iboxes, _ = _get_results(self._q) self.infoboxes = iboxes return results result_container = _MockResultContainer(query) session_id = secrets.token_hex(12) # Clear prior conversation so each search starts fresh try: from ollama_answers import _get_valkey vk = _get_valkey() vk.delete(f"ai:conv:{session_id}") except Exception: pass class _MockRequest: cookies = {'sxng_ai_session': session_id} headers = {} args = {} script_root = '' class _MockForm(dict): def get(self, key, default=None): return default # always return html format form = _MockForm() _score_handler.clear() search = _MockSearch() try: plugin.post_search(_MockRequest(), search) except Exception as e: logger.error(f"post_search failed: {e}", exc_info=True) return f"

post_search error: {e}

" html_payload = "" if search.result_container.answers: raw = list(search.result_container.answers)[0] # Markup objects need str() to get the HTML html_payload = str(raw) if raw else "" logger.info(f"Got AI answer, length: {len(html_payload)}") if html_payload: import re as _re text_preview = _re.sub(r'<[^>]+>', '', html_payload) text_preview = ' '.join(text_preview.split())[:300] logger.info(f"AI output preview: {text_preview}") else: logger.warning("post_search produced no answers") # Parse intent, per-URL TF-IDF scores, and rerank summary from captured log records # Parse intent, per-URL TF-IDF scores, and rerank summary from captured log records import re as _re detected_intent = '—' scores = [] fallback_domains = set() summary = '' for msg in _score_handler.records: m = _re.match(r'AI Answers: \[(.+?)\] score=([\d.]+) chunks=(\d+)', msg) if m: scores.append({ 'url': m.group(1), 'score': float(m.group(2)), 'chunks': int(m.group(3)), 'fallback': False }) m = _re.match(r'AI Answers: falling back to snippet for \[\d+\] (.+)', msg) if m: fallback_domains.add(m.group(1)) m = _re.match(r'AI Answers: reranked (\d+) results, top score=([\d.]+)', msg) if m: summary = f"{m.group(1)} reranked · top {float(m.group(2)):.4f}" for s in scores: domain = s['url'].replace('https://', '').replace('http://', '').split('/')[0] if domain in fallback_domains: s['fallback'] = True scores.sort(key=lambda s: s['score'], reverse=True) # Refresh latest intent from Valkey try: from ollama_answers import _get_valkey vk = _get_valkey() latest_intent = vk.get("ai:last_intent") if latest_intent: detected_intent = latest_intent except Exception: pass _last_render['html'] = html_payload _last_render['query'] = query _last_render['intent'] = detected_intent _last_render['scores'] = scores _last_render['summary'] = summary global _last_job _last_job['intent'] = detected_intent _last_job['query'] = query logger.warning(f"RETURNING INTENT TO UI: {repr(_last_job.get('intent'))}") return jsonify({'ok': True, 'query': query}) @app.route('/dev/render') def dev_render(): html = _last_render.get('html', '') return f"""
{html}
""" # ── dev API log endpoint ────────────────────────────────────────────────────── _last_job = {'job_id': None, 'intent': None, 'model': None, 'query': None, 'start_ts': None} def _get_last_job_from_log(): for entry in _dev_log: if '/ai-stream' in entry.get('path', '') and entry.get('method') == 'POST': try: resp = json.loads(entry.get('resp', '{}')) if resp.get('job_id'): return resp['job_id'] except Exception: pass return None @app.route('/dev/log') def dev_log(): return jsonify(_dev_log) @app.route('/dev/log/clear', methods=['POST']) def dev_log_clear(): _dev_log.clear() return jsonify({'ok': True}) @app.route('/dev/last-job') def dev_last_job(): return jsonify(_last_job) @app.route('/dev/last-scores') def dev_last_scores(): return jsonify({ 'scores': _last_render.get('scores', []), 'intent': _last_render.get('intent', '—'), 'query': _last_render.get('query', ''), 'summary': _last_render.get('summary', ''), }) @app.route('/dev/stream-watch') def dev_stream_watch(): job_id = request.args.get('job_id', '') if not job_id: return jsonify({'error': 'no job_id'}), 400 def generate(): import valkey as _vk host = os.getenv('VALKEY_HOST', 'localhost') port = int(os.getenv('VALKEY_PORT', 6379)) v = _vk.Valkey(host=host, port=port, socket_connect_timeout=2, decode_responses=True) offset = 0 max_polls = 600 polls = 0 while polls < max_polls: polls += 1 time.sleep(0.1) try: chunks = v.lrange(f"ai:job:{job_id}:chunks", offset, -1) for chunk in chunks: if chunk == '__DONE__': yield "data: __DONE__\n\n" return if chunk.startswith('__ERROR__'): yield f"data: __ERROR__{chunk[9:]}\n\n" return yield f"data: {json.dumps(chunk)}\n\n" offset += len(chunks) except Exception as e: yield f"data: __ERROR__{e}\n\n" return yield "data: __DONE__\n\n" return Response(generate(), mimetype='text/event-stream', headers={ 'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no' }) # ── dev config update ───────────────────────────────────────────────────────── @app.route('/dev/config', methods=['POST']) def dev_config(): global SEARXNG_URL data = request.json or {} if 'llm_url' in data: url = data.get('llm_url', '').strip() if url: plugin.endpoint_url = url os.environ['LLM_URL'] = url if 'model' in data and data['model']: plugin.model = data['model'].strip() os.environ['LLM_MODEL'] = plugin.model if 'max_tokens' in data: try: plugin.max_tokens = max(1, int(data['max_tokens'])) except (ValueError, TypeError): pass if 'temperature' in data: try: plugin.temperature = float(data['temperature']) except (ValueError, TypeError): pass if 'api_key' in data: plugin.api_key = data['api_key'].strip() or 'ollama' if 'searxng_url' in data: SEARXNG_URL = data['searxng_url'].strip().rstrip('/') if 'valkey_host' in data or 'valkey_port' in data: host = data.get('valkey_host', os.getenv('VALKEY_HOST', 'localhost')).strip() port = int(data.get('valkey_port', os.getenv('VALKEY_PORT', 6379))) os.environ['VALKEY_HOST'] = host os.environ['VALKEY_PORT'] = str(port) # Reset the valkey connection pool so it picks up new settings import ollama_answers as _oa _oa._VALKEY_POOL = None return jsonify({ 'llm_url': plugin.endpoint_url, 'model': plugin.model, 'max_tokens': plugin.max_tokens, 'temperature': plugin.temperature, 'api_key': plugin.api_key, 'searxng_url': SEARXNG_URL or None, }) @app.route('/dev/config', methods=['GET']) def dev_config_get(): return jsonify({ 'llm_url': plugin.endpoint_url, 'model': plugin.model, 'max_tokens': plugin.max_tokens, 'temperature': plugin.temperature, 'api_key': plugin.api_key, 'searxng_url': SEARXNG_URL or None, 'interactive': plugin.interactive, 'allowed_tabs': list(plugin.allowed_tabs), 'valkey_host': os.getenv('VALKEY_HOST', 'localhost'), 'valkey_port': int(os.getenv('VALKEY_PORT', 6379)), }) # ── main dev UI ─────────────────────────────────────────────────────────────── @app.route('/') def index(): return DEV_HTML DEV_HTML = """ AI Answers — Dev Server

✦ AI Answers Dev

Mock Results Ollama: checking… Valkey: checking…

Enter a query above and click Search to test the plugin.

Config
Debug Log
Stream
API Explorer
Plugin Config
loading…
Model—
Max Tokens—
Temperature—
Interactive—
Allowed Tabs—
SearXNG Backendmock
Valkey—
Edit Config
No requests yet
Idle
Shows token chunks arriving from Valkey in real time during streaming.
Waiting for next search…
First token: — | Total: —

Test plugin routes directly. Token is auto-injected.

GET /ai-models List Ollama models 

        

      


      

        

          POST
          /ai-stream
          Start AI stream
        

        

          
          
          

        

      


      

        

          GET
          /ai-status/{job_id}
          Poll stream status
        

        

          

          

          
          

        

      


      

        

          GET
          /ai-conversation
          Get conversation
        

        

          

          
          

        

      


      

        

          DELETE
          /ai-conversation
          Clear conversation
        

        

          

          
          

        

      


    

  







"""

# ── token endpoint (dev only) ─────────────────────────────────────────────────
@app.route('/dev/token')
def dev_token():
    ts = str(time.time())
    sig = hashlib.sha256(f"{ts}{plugin.secret}".encode()).hexdigest()
    return jsonify({'token': f"{ts}.{sig}"})

# ── ollama health check endpoint ──────────────────────────────────────────────
@app.route('/dev/valkey-check')
def dev_valkey_check():
    try:
        import valkey as _vk
        host = os.getenv('VALKEY_HOST', 'localhost')
        port = int(os.getenv('VALKEY_PORT', 6379))
        v = _vk.Valkey(host=host, port=port, socket_connect_timeout=2)
        v.ping()
        return jsonify({'ok': True, 'host': host, 'port': port})
    except Exception as e:
        return jsonify({'ok': False, 'error': str(e)})

@app.route('/dev/ollama-check')
def dev_ollama_check():
    try:
        p = urllib.parse.urlparse(plugin.endpoint_url)
        base = f"{p.scheme}://{p.netloc}"
        url = f"{base}/api/tags"
        req = urllib.request.Request(url, headers={'Authorization': f'Bearer {plugin.api_key}'})
        with urllib.request.urlopen(req, timeout=3) as r:
            data = json.loads(r.read())
        models = len(data.get('models', []))
        return jsonify({'ok': True, 'models': models})
    except Exception as e:
        return jsonify({'ok': False, 'error': str(e)})

# ── entry point ───────────────────────────────────────────────────────────────
if __name__ == '__main__':
    print(f"""
  ✦ AI Answers Dev Server
  ─────────────────────────────────────────
  UI:          http://{DEV_HOST}:{DEV_PORT}
  LLM:         {plugin.endpoint_url}
  Model:       {plugin.model}
  Max tokens:  {plugin.max_tokens}
  Temperature: {plugin.temperature}
  SearXNG:     {SEARXNG_URL or 'mock results'}
  ─────────────────────────────────────────
""")
    app.run(host=DEV_HOST, port=DEV_PORT, debug=False)