| """Same-origin, single-user development server. No audio or transcript logging."""
|
| from functools import partial
|
| from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
| from pathlib import Path
|
| import json
|
| import secrets
|
| import threading
|
| import time
|
| from urllib.parse import urlsplit, parse_qs
|
|
|
| MAX_BYTES = 20 * 1024 * 1024
|
|
|
|
|
| class BusyError(Exception):
|
| pass
|
|
|
|
|
| def make_server(app_dir, transcribe=None, language=None, port=8765, answer=None, progress=None, speak=None):
|
| token = secrets.token_urlsafe(32)
|
| lock = threading.Lock()
|
| jobs = {}
|
| jobs_lock = threading.Lock()
|
| latest_answer = {}
|
|
|
| def answer_and_remember(data):
|
| latest_answer.clear()
|
| result = answer(data)
|
| latest_answer.update(result=result, expires=time.monotonic() + 900)
|
| return result
|
|
|
| class Handler(SimpleHTTPRequestHandler):
|
| def setup(self):
|
| super().setup()
|
| self.connection.settimeout(60)
|
|
|
| def log_message(self, *args):
|
| pass
|
|
|
| def list_directory(self, path):
|
| self.send_error(404)
|
|
|
| def end_headers(self):
|
| self.send_header('Cache-Control', 'no-store')
|
| self.send_header('X-Content-Type-Options', 'nosniff')
|
| self.send_header('Referrer-Policy', 'no-referrer')
|
| self.send_header('Content-Security-Policy', "default-src 'self'; script-src 'self'; style-src 'self'; media-src 'self' blob:; connect-src 'self'; object-src 'none'; base-uri 'none'; form-action 'none'")
|
| super().end_headers()
|
|
|
| def reply(self, status, data):
|
| payload = json.dumps(data, ensure_ascii=False).encode('utf-8')
|
| self.send_response(status)
|
| self.send_header('Content-Type', 'application/json; charset=utf-8')
|
| self.send_header('Content-Length', str(len(payload)))
|
| self.end_headers()
|
| try:
|
| self.wfile.write(payload)
|
| except (BrokenPipeError, ConnectionResetError):
|
| pass
|
|
|
| def do_GET(self):
|
| if urlsplit(self.path).path.startswith('/api/jobs/'):
|
| if not secrets.compare_digest(self.headers.get('X-FieldSabi-Token', ''), token):
|
| return self.reply(403, {'error': 'Refresh the app to reconnect.'})
|
| job_id = urlsplit(self.path).path.rsplit('/', 1)[-1]
|
| with jobs_lock:
|
| job = jobs.get(job_id)
|
| result = dict(job) if job else None
|
| if result is None:
|
| return self.reply(404, {'error': 'Job expired or runtime restarted. Submit the question again.'})
|
| if result['state'] == 'running':
|
| result['stage'] = result.get('stage') or (progress() if progress else 'Searching sources, loading the model or checking the answer')
|
| result['elapsed_seconds'] = round(time.monotonic() - result.pop('started'))
|
| return self.reply(200, result)
|
|
|
| if urlsplit(self.path).path == '/api/session':
|
| return self.reply(200, {'ready': transcribe is not None, 'language': language,
|
| 'csrf': token, 'answers_ready': answer is not None,
|
| 'speech_languages': ['en','yo','ha'] if speak else []})
|
| if urlsplit(self.path).path.startswith('/api/'):
|
| return self.reply(404, {'error': 'Unknown endpoint.'})
|
| return super().do_GET()
|
|
|
| def do_POST(self):
|
| is_answer = urlsplit(self.path).path == '/api/answer'
|
| is_speak = urlsplit(self.path).path == '/api/speak'
|
| if urlsplit(self.path).path not in {'/api/transcribe', '/api/answer', '/api/speak'}:
|
| return self.reply(404, {'error': 'Unknown endpoint.'})
|
| if not secrets.compare_digest(self.headers.get('X-FieldSabi-Token', ''), token):
|
| return self.reply(403, {'error': 'Refresh the app before sending audio.'})
|
| if (speak if is_speak else answer if is_answer else transcribe) is None:
|
| return self.reply(503, {'error': 'Speech model is not connected. Open the app in the Colab launcher.'})
|
| selected = parse_qs(urlsplit(self.path).query).get('language', [''])[0]
|
| if not is_answer and not is_speak and selected != language:
|
| return self.reply(422, {'error': 'Choose the language loaded by this speech service.'})
|
| try:
|
| length = int(self.headers.get('Content-Length', '0'))
|
| except ValueError:
|
| length = 0
|
| if self.headers.get('Transfer-Encoding') or not 0 < length <= (100 if is_speak else 4300000 if is_answer else MAX_BYTES):
|
| return self.reply(413, {'error': 'Send a non-empty audio file up to 20 MB.'})
|
| if not lock.acquire(blocking=False):
|
| return self.reply(429, {'error': 'Model service is busy. Please retry shortly.'})
|
| handed_off = False
|
| try:
|
| data = self.rfile.read(length)
|
| if len(data) != length:
|
| return self.reply(400, {'error': 'Audio upload was incomplete.'})
|
| if is_speak:
|
| from .tts import spoken_text
|
| payload = json.loads(data)
|
| selected_language = payload.get('language') if isinstance(payload, dict) else None
|
| if selected_language not in {'en','yo','ha'}:
|
| return self.reply(422, {'error':'Choose a supported speech language.'})
|
| if latest_answer.get('expires', 0) < time.monotonic():
|
| return self.reply(422, {'error':'The answer expired. Generate it again before listening.'})
|
| text = spoken_text(latest_answer['result'], selected_language)
|
| job_id = secrets.token_urlsafe(24)
|
| with jobs_lock:
|
| jobs.clear()
|
| jobs[job_id] = {'state':'running', 'started':time.monotonic(), 'stage':'Preparing spoken answer'}
|
| def run_speech():
|
| try:
|
| result = speak(text, selected_language)
|
| with jobs_lock:
|
| jobs[job_id].update(state='done', result=result)
|
| except Exception:
|
| with jobs_lock:
|
| jobs[job_id].update(state='failed', error='Speech generation failed. The written answer is still available. Check model access in the runtime.')
|
| finally:
|
| lock.release()
|
| threading.Thread(target=run_speech, daemon=True).start()
|
| handed_off = True
|
| return self.reply(202, {'job_id':job_id})
|
| if is_answer:
|
| try:
|
| payload = json.loads(data)
|
| except (ValueError, UnicodeError):
|
| return self.reply(422, {'error': 'Send a valid question.'})
|
| limits = {'question':1500, 'language':2, 'answer_mode':5, 'previous_question':1500, 'state':80, 'locality':100, 'topic':80, 'farm_context':500}
|
| if not isinstance(payload, dict) or any(not isinstance(payload.get(k, ''), str) or len(payload.get(k, '')) > limit for k, limit in limits.items()):
|
| return self.reply(422, {'error': 'Question or farm details are invalid or too long.'})
|
| if not payload.get('question', '').strip() or payload.get('language') not in {'en','ha','yo','ig'}:
|
| return self.reply(422, {'error': 'Enter a question and choose a supported language.'})
|
| if payload.get('answer_mode', '') not in {'','en','yo','ha','en-yo','en-ha','ig','en-ig'}:
|
| return self.reply(422, {'error':'Choose a supported answer language.'})
|
| clean = {k:payload.get(k, '').strip() for k in limits}
|
| document = payload.get('document')
|
| if document is not None:
|
| if not isinstance(document, dict) or any(not isinstance(document.get(k), str) or len(document[k]) > size for k,size in {'name':150,'scope':100,'data':4200000}.items()):
|
| return self.reply(422, {'error':'Invalid reference document. Use a PDF up to 3 MB.'})
|
| clean['document'] = {k:document[k] for k in ('name','scope','data')}
|
|
|
| if self.headers.get('Prefer') == 'respond-async':
|
| job_id = secrets.token_urlsafe(24)
|
| with jobs_lock:
|
| jobs.clear()
|
| jobs[job_id] = {'state':'running', 'started':time.monotonic()}
|
| def run():
|
| try:
|
| result = answer_and_remember(clean)
|
| with jobs_lock:
|
| jobs[job_id].update(state='done', result=result)
|
| except ValueError as error:
|
| with jobs_lock:
|
| jobs[job_id].update(state='failed', error=str(error))
|
| except Exception as error:
|
|
|
| print('FieldSabi answer failed:', type(error).__name__, flush=True)
|
| with jobs_lock:
|
| jobs[job_id].update(state='failed', error='Answer failed in the runtime. Check GPU memory and model access; restart the runtime if necessary.')
|
| finally:
|
| lock.release()
|
| worker = threading.Thread(target=run, daemon=True)
|
| worker.start()
|
| handed_off = True
|
| return self.reply(202, {'job_id':job_id})
|
| return self.reply(200, answer_and_remember(clean))
|
| text = transcribe(data).strip()
|
| words = text.split()
|
| if not text or len(text) > 1500 or (len(words) >= 12 and len(set(words[-8:])) <= 2):
|
| return self.reply(422, {'error': 'No usable transcript returned. Try a clearer, shorter recording.'})
|
| self.reply(200, {'text': text, 'language': language})
|
| except ValueError as error:
|
| self.reply(422, {'error': str(error)})
|
| except Exception:
|
| self.reply(503, {'error': 'Model request failed. Check the runtime and retry.'})
|
| finally:
|
| if not handed_off:
|
| lock.release()
|
|
|
| class Server(ThreadingHTTPServer):
|
| def server_close(self):
|
| super().server_close()
|
|
|
| with lock:
|
| pass
|
|
|
| return Server(('127.0.0.1', port), partial(Handler, directory=str(Path(app_dir).resolve())))
|
|
|
