"""
๐จ Agent UX (Interface & API Gateway)
-------------------------------------------------
This agent acts as the central controller and router. It manages the Gradio
Lab Admin dashboard, provides headless API endpoints for the React frontend games,
and seamlessly orchestrates data flow between the users, the Brain, and the Trust agents.
"""
import os
import time
import uuid
import glob
import hmac
import inspect
import json
import re
import shutil
import threading
import traceback
import zipfile
from datetime import datetime
import gradio as gr
import pandas as pd
from config import AppConfig
from huggingface_hub import HfApi, hf_hub_download
try:
import spaces
except Exception:
spaces = None
from src import prepare_training_data # ๐ข Import your new script
from src.dialect_rules import (
hausa_variety_instruction,
nigerian_variety_instruction,
nigerian_variety_retry_prompt,
nigerian_variety_retry_reason,
)
from src.asr_language import resolve_asr_language_code
def zero_gpu_optional(duration=60):
def decorator(fn):
if spaces is not None and hasattr(spaces, "GPU"):
return spaces.GPU(duration=duration)(fn)
return fn
return decorator
# ==========================================
# โ๏ธ CLOUD SYNC CHECK
# ==========================================
def check_cloud_sync_status():
"""Checks Hugging Face for the last commit time and returns a HUD status string."""
try:
# ๐ข FIX: Use AppConfig to guarantee the token is found, instead of relying on os.environ
hf_token = AppConfig.HF_TOKEN
if not hf_token:
print("Sync check warning: HF_TOKEN is None")
return "๐ด SYNC OFFLINE (NO TOKEN)"
api = HfApi(token=hf_token)
repo_id = "toecm/IEDID"
commits = api.list_repo_commits(repo_id=repo_id, repo_type="dataset")
if commits:
last_sync = commits[0].created_at.strftime("%H:%M")
return f"๐ข CLOUD SYNCED (Last: {last_sync})"
return "๐ก REPO EMPTY"
except Exception as e:
print(f"Sync check error: {e}")
return "๐ด SYNC OFFLINE"
class AgentUX:
def __init__(self, input_agent, brain_agent, trust_agent, acoustic_agent=None):
print("\n" + "="*40)
print("๐ต๏ธโโ๏ธ UX AGENT: SECRETS AUDIT")
pk = os.environ.get("PRIVATE_KEY")
rpc = os.environ.get("PURECHAIN_RPC_URL")
print(f"๐ PRIVATE_KEY : {'โ
LOADED' if pk else 'โ MISSING'}")
print(f"๐ RPC_URL : {rpc if rpc else 'โ MISSING'}")
print(f"๐ง GEMINI_KEY : {'โ
LOADED' if os.environ.get('GOOGLE_API_KEY') else 'โ MISSING'}")
print(f"โ๏ธ HF_TOKEN : {'โ
LOADED' if os.environ.get('HF_TOKEN') else 'โ MISSING'}")
print(f"๐ก๏ธ REVIEW_KEY : {'โ
LOADED' if (os.environ.get('PURE_REVIEW_KEY') or os.environ.get('PURE_ADMIN_KEY') or os.environ.get('ADMIN_PASSCODE')) else 'โ MISSING - review/admin mutations disabled'}")
print("="*40 + "\n")
self.input = input_agent
self.brain = brain_agent
self.trust = trust_agent
self.acoustic = acoustic_agent
if self.acoustic and hasattr(self.acoustic, "attach_input_agent"):
self.acoustic.attach_input_agent(self.input)
self.TONES = ["Neutral / Conversational", "Casual / Slang", "Formal / Professional", "Proverb / Idiom"]
self.PENDING_FILE = "/app/pending_approvals.csv"
self.ADMIN_ACTION_LOG = os.environ.get("ADMIN_ACTION_LOG_FILE", "/app/admin_action_log.jsonl")
self.last_audio_path = None
self.last_pending_count = 0
# ๐ข NEW: Matchmaking & Live Room Memory
self.waiting_pool = [] # List of operators looking for a match
self.active_matches = {} # Maps Operator ID -> Room Code
self.live_rooms = {} # Stores the actual chat logs per room
self.alert_sound = None
print("๐จ Agent UX Online: PhD Research Hub Ready.")
self.sync_pending_queue(direction="down")
self.sync_admin_action_log(direction="down")
def get_quota_status(self):
if hasattr(self.brain, 'check_quota'): return self.brain.check_quota()
if hasattr(self.brain.gemini_manager, 'get_status_string'): return self.brain.gemini_manager.get_status_string()
return "Active"
def _resolve_ai_model_label(self):
manager = getattr(getattr(self, "brain", None), "gemini_manager", None)
model = getattr(manager, "model_flash", None) or getattr(manager, "last_used_model", None)
if model:
return str(model)
if os.environ.get("QWEN_MODEL_NAME"):
return os.environ.get("QWEN_MODEL_NAME")
if os.environ.get("GROQ_API_KEY"):
return "llama-3.3-70b-versatile"
return "Dataset/Persona/Fallback"
def _resolve_asr_model_label(self):
model_name = getattr(getattr(self, "input", None), "model_name", None)
return f"Whisper {model_name}" if model_name else "Whisper tiny"
def get_blockchain_health(self):
try:
if hasattr(self.trust, 'w3') and self.trust.w3 and self.trust.w3.is_connected():
return "๐ข PureChain Network: ONLINE & SYNCED
"
except Exception:
pass
return "๐ด PureChain Network: OFFLINE / DISCONNECTED
"
def check_background_status(self):
tasks = self.trust.active_tasks
current_time = datetime.now().strftime('%H:%M:%S')
if tasks > 0:
return f"๐ Processing {tasks} background task(s)... | {current_time}"
return f"โ
System Active (Ready) | {current_time}"
def _review_key(self):
return (os.environ.get("PURE_REVIEW_KEY") or os.environ.get("PURE_ADMIN_KEY") or os.environ.get("ADMIN_PASSCODE") or "").strip()
def _review_scope(self, review_key):
provided = str(review_key or "").strip()
if not provided:
return None
for env_name in ["PURE_REVIEW_KEY", "PURE_ADMIN_KEY", "ADMIN_PASSCODE"]:
expected = os.environ.get(env_name, "").strip()
if expected and hmac.compare_digest(provided, expected):
return {"kind": "full", "label": "Full admin"}
scoped = [
("PURE_ADMIN_KEY_CH", "Chinese pending reviewer", ["chinese"], ["mandarin", "cantonese", "hakka", "gan", "jin", "wu", "yue", "min", "xiang", "gan chinese", "jin chinese", "mandarin chinese", "yue chinese", "hakka chinese", "wu chinese", "xiang chinese", "min nan chinese", "min dong chinese", "min bei chinese", "standard mandarin", "beijing dialect", "taiwanese mandarin", "singaporean mandarin", "guangzhou cantonese", "hong kong cantonese", "macau cantonese", "nanchang", "ji'an", "jian", "jian'ou", "yichun", "zhangjiakou-hohhot", "lรผliang", "luliang", "bingzhou", "fuzhou", "hainanese", "hokkien", "teochew", "hailu", "meixian", "sixian", "shanghainese", "suzhounese", "wenzhounese", "loushao", "chenxรผ", "chenxu", "changyi"]),
("PURE_ADMIN_KEY_KO", "Korean pending reviewer", ["korean"], ["satoori", "jeju", "gyeongsang", "jeolla", "chungcheong", "gangwon", "hamgyong", "pyongan", "hwanghae"]),
]
for env_name, label, language_terms, dialect_terms in scoped:
expected = os.environ.get(env_name, "").strip()
if expected and hmac.compare_digest(provided, expected):
return {
"kind": "scoped",
"label": label,
"language_terms": language_terms,
"dialect_terms": dialect_terms,
}
return None
def _is_review_authorized(self, review_key):
return self._review_scope(review_key) is not None
def _is_full_admin_authorized(self, review_key):
scope = self._review_scope(review_key)
return bool(scope and scope.get("kind") == "full")
def _scope_allows_language_dialect(self, scope, language="", dialect="", *extra_texts):
if not scope:
return False
if scope.get("kind") == "full":
return True
language_text = str(language or "").lower()
dialect_text = str(dialect or "").lower()
extra_text = " ".join(str(value or "").lower() for value in extra_texts)
combined = f"{language_text} {dialect_text} {extra_text}"
if any(self._scope_term_matches(combined, term) for term in scope.get("language_terms", [])):
return True
if any(self._scope_term_matches(combined, term) for term in scope.get("dialect_terms", [])):
return True
return False
def _scope_term_matches(self, text, term):
text = str(text or "").lower()
term = str(term or "").lower()
if not term:
return False
if len(term) <= 3:
import re
return re.search(r"(? {final_list}")
print("="*40 + "\n")
return json.dumps(final_list)
except Exception as e:
print(f"๐จ CRITICAL API ERROR: {e}")
return json.dumps(["American English", "British English", "+ Add New Dialect"])
def api_get_language_hierarchy(self):
print("\n" + "="*40)
print("๐ก REACT API WAKEUP: Requesting Language Hierarchy...")
hierarchy = {}
ignore_list = ["minted_history", "system_feedback", "pending_approvals", "train", "dataset_export", "rejected_graveyard", "eeqs_events"]
try:
dataset_dir = getattr(self.brain.config, 'DATASET_DIR', "/app/IEDID")
print(f"๐ Scanning directory for hierarchy: {dataset_dir}")
if os.path.exists(dataset_dir):
files = glob.glob(os.path.join(dataset_dir, "**", "*.csv"), recursive=True)
for f in files:
name = os.path.basename(f).replace(".csv", "")
if name in ignore_list:
continue
parent_dir = os.path.dirname(f)
grandparent_dir = os.path.dirname(parent_dir)
if os.path.basename(parent_dir) == "data":
language = os.path.basename(grandparent_dir)
else:
language = os.path.basename(parent_dir)
if language == os.path.basename(dataset_dir):
language = "Other"
if language not in hierarchy:
hierarchy[language] = set()
hierarchy[language].add(name)
else:
print(f"โ ๏ธ Directory {dataset_dir} does not exist!")
final_hierarchy = {}
for lang, dialects in hierarchy.items():
final_hierarchy[lang] = sorted(list(dialects)) + ["+ Add New Dialect"]
if not final_hierarchy:
final_hierarchy = {"English": ["American English", "British English", "+ Add New Dialect"]}
print(f"โ
SUCCESS: Sending Hierarchy to React")
print("="*40 + "\n")
return json.dumps(final_hierarchy)
except Exception as e:
print(f"๐จ CRITICAL API ERROR: {e}")
return json.dumps({"English": ["American English", "British English", "+ Add New Dialect"]})
def api_generate_mission(self, topic):
topic_str = str(topic).strip("['\"]")
if hasattr(self.brain, 'generate_conversation_starter'):
return self.brain.generate_conversation_starter(topic_str)
return json.dumps({"text": f"Let's talk about {topic_str}."})
def api_transcribe(self, audio_path, dialect):
return self.api_transcribe_with_model(audio_path, dialect, "auto", "on")
def api_acoustic_models(self):
if self.acoustic and hasattr(self.acoustic, "models"):
return json.dumps(self.acoustic.models())
return json.dumps({"ok": False, "error": "Acoustic agent unavailable."})
def api_acoustic_tts(self, text, language="", dialect="", voice="browser-native"):
if self.acoustic and hasattr(self.acoustic, "tts"):
return json.dumps(self.acoustic.tts(text, language=language, dialect=dialect, voice=voice))
return json.dumps({"ok": False, "error": "Acoustic agent unavailable.", "text": text or ""})
@zero_gpu_optional(duration=60)
def api_acoustic_transcribe(self, audio_path, language="", dialect="", speech_model="auto", audio_sanitation="on"):
if not self.acoustic or not hasattr(self.acoustic, "transcribe"):
return json.dumps({"ok": False, "text": "", "error": "Acoustic agent unavailable."})
try:
result = self.acoustic.transcribe(
audio_path,
language=language,
dialect=dialect,
speech_model=speech_model,
audio_sanitation=audio_sanitation,
)
return json.dumps(result)
except Exception as e:
return json.dumps({"ok": False, "text": "", "error": str(e)})
@zero_gpu_optional(duration=60)
def api_transcribe_with_model(self, audio_path, dialect, speech_model="auto", audio_sanitation="on"):
if not audio_path: return ""
try:
parsed_language = ""
parsed_dialect = str(dialect or "")
if "|||" in parsed_dialect:
parsed_language, parsed_dialect = (parsed_dialect.split("|||", 1) + [""])[:2]
if self.acoustic and hasattr(self.acoustic, "transcribe"):
result = self.acoustic.transcribe(
audio_path,
language=parsed_language,
dialect=parsed_dialect,
speech_model=speech_model,
audio_sanitation=audio_sanitation,
)
return result.get("text", "") if isinstance(result, dict) else str(result)
if hasattr(self.input, 'transcribe'):
asr_code, source_language, source_dialect = resolve_asr_language_code(dialect)
speech_model_choice = str(speech_model or "auto").strip() or "auto"
sanitation_choice = str(audio_sanitation or "on").strip().lower() or "on"
print(f"๐ง ASR hint resolved: language={source_language or 'auto'} dialect={source_dialect or 'auto'} code={asr_code or 'auto'} speech_model={speech_model_choice} clean_audio={sanitation_choice}")
res = self.input.transcribe(
audio_path,
language=asr_code,
model_choice=speech_model_choice,
dialect_hint=f"{source_language or ''} {source_dialect or ''} {dialect or ''}",
sanitize_audio=sanitation_choice
)
return res[0]['text'] if isinstance(res, list) else str(res)
except Exception as e:
return f"Transcription error: {str(e)}"
return ""
def api_clarify(self, text, dialect):
try:
if hasattr(self.brain, 'analyze_dialect_single'):
res = self.brain.analyze_dialect_single(text, dialect)
clarification = res.get("clarification", text)
return json.dumps({"clarification": clarification})
except: pass
return json.dumps({"clarification": text})
def _ai_model_hint(self, ai_model):
choice = str(ai_model or "auto").strip().lower()
hints = {
"auto": "Use the system's smart default model for this language and task.",
"llama": "Prefer the Llama/Groq route when available.",
"qwen": "Prefer the Qwen route when available, especially for multilingual and non-Latin-script handling.",
"nemotron": "Prefer the free OpenRouter Nemotron route when available, especially for reasoning and ambiguity handling.",
"gpt-oss": "Prefer the free OpenRouter GPT-OSS route for general multilingual text work.",
"lfm": "Prefer the free OpenRouter LFM tiny route when speed is more important than depth.",
"openrouter-free": "Prefer OpenRouter's free auto route as a cost-free fallback.",
"deepseek": "Prefer the paid DeepSeek route when configured, especially for careful reasoning and ambiguity handling.",
"gemini": "Prefer the Gemini route when available.",
}
return hints.get(choice, hints["auto"])
def api_clarify_with_model(self, text, dialect, ai_model="auto"):
if not text:
return json.dumps({"clarification": ""})
if str(ai_model or "auto").strip().lower() == "auto":
return self.api_clarify(text, dialect)
try:
if hasattr(self.brain, "_safe_generate"):
prompt = f"""
{self._ai_model_hint(ai_model)}
Explain the meaning of this utterance for PureHumCom data review.
Dialect/language: {dialect}
Utterance: "{text}"
Return only concise JSON with a clarification field.
"""
raw = self.brain._safe_generate(prompt)
cleaned = self._clean_peer_translation(raw)
try:
parsed = json.loads(cleaned)
clarification = parsed.get("clarification") or parsed.get("text") or parsed.get("Meaning") or cleaned
except Exception:
clarification = cleaned
return json.dumps({"clarification": clarification or text})
except Exception as e:
print(f"Selector-aware clarify Error: {e}")
return self.api_clarify(text, dialect)
def _target_script_rule(self, dialect):
d = (dialect or "").lower()
rules = [
(("korean", "seoul", "satoori", "jeju", "gyeongsang", "chungcheong", "jeolla", "busan"), "Hangul", r"[\uac00-\ud7af\u1100-\u11ff\u3130-\u318f]", "Write natural Korean Hangul syllables only. Do not use romanization, Chinese characters, Japanese kana, or punctuation-only placeholders."),
(("arabic", "urdu", "uyghur", "persian", "farsi", "dari", "pashto", "shahmukhi"), "Arabic", r"[\u0600-\u06ff]", "Use Arabic-script characters appropriate for the target language or dialect."),
(("greek", "cretan"), "Greek", r"[\u0370-\u03ff]", "Use Greek script characters."),
(("chinese", "mandarin", "cantonese", "yue", "wu ", "xiang"), "Han", r"[\u4e00-\u9fff]", "Use Chinese Han characters where natural for the target variety."),
(("japanese", "kansai"), "Japanese", r"[\u3040-\u30ff\u4e00-\u9fff]", "Use Japanese kana/kanji, not romanization."),
(("thai",), "Thai", r"[\u0e00-\u0e7f]", "Use Thai script."),
(("russian", "ukrainian", "bulgarian", "serbian", "cyrillic"), "Cyrillic", r"[\u0400-\u04ff]", "Use Cyrillic script."),
(("hindi", "marathi", "nepali", "sanskrit"), "Devanagari", r"[\u0900-\u097f]", "Use Devanagari script."),
(("bengali", "bangla"), "Bengali", r"[\u0980-\u09ff]", "Use Bengali script."),
(("tamil",), "Tamil", r"[\u0b80-\u0bff]", "Use Tamil script."),
(("telugu",), "Telugu", r"[\u0c00-\u0c7f]", "Use Telugu script."),
]
for tokens, name, pattern, instruction in rules:
if any(token in d for token in tokens):
return {"name": name, "pattern": pattern, "instruction": instruction}
return None
def _clean_peer_translation(self, response):
raw_text = getattr(response, "text", response)
cleaned = str(raw_text or "").replace("```json", "").replace("```", "").strip()
try:
parsed = json.loads(cleaned)
if isinstance(parsed, dict):
cleaned = parsed.get("translation") or parsed.get("text") or parsed.get("translated_text") or cleaned
except Exception:
pass
for prefix in ("Translation:", "Translated text:", "Output:"):
if cleaned.lower().startswith(prefix.lower()):
cleaned = cleaned[len(prefix):].strip()
return cleaned.strip().strip('"').strip("'").strip()
def _translation_has_required_script(self, translated, target_dialect):
if not translated or re.fullmatch(r"[\s\W_]+", translated, flags=re.UNICODE):
return False
rule = self._target_script_rule(target_dialect)
if not rule:
return True
return bool(re.search(rule["pattern"], translated))
def _fallback_script_translation(self, text, target_dialect):
rule = self._target_script_rule(target_dialect)
if not rule:
return ""
d = (target_dialect or "").lower()
lower_text = str(text or "").lower()
if rule["name"] == "Hangul":
if "echo" in lower_text or "ai mock peer" in lower_text:
if "gyeongsang" in d or "busan" in d:
return "์๋
ํ์ด์! ๋๋ ์์ฝ๋ผ์. ๋ AI ๋ํ ์๋๋ผ์. ๊ฒฝ์๋ ์ฌํฌ๋ฆฌ๋ก ์๊ธฐํ ๋ผ๋ผ์. ๊ฐ์ด ์๊ธฐํด๋ณด์
์๋!"
if "jeju" in d:
return "์๋
ํ์ฐ๊ฝ! ๋๋ ์์ฝ์ฐ๋ค. ๋ค AI ๋ํ ์๋์ฐ๋ค. ์ ์ฃผ ์ฌํฌ๋ฆฌ๋ก ๋งํด๋ณด์ฟ ๋ค. ๊ฐ์ด ์ด์ผ๊ธฐํด๋ด
์!"
if "chungcheong" in d:
return "์๋
ํ์ธ์ ! ๋๋ ์์ฝ์ฌ์ . ๋ค AI ๋ํ ์๋์ฌ์ . ์ถฉ์ฒญ๋ ์ฌํฌ๋ฆฌ๋ก ์๊ธฐํ ๊ฒ์ . ๊ฐ์ด ์๊ธฐํด๋ด์ !"
if "jeolla" in d:
return "์๋
ํ์! ๋๋ ์์ฝ์ฌ๋ผ. ๋ค AI ๋ํ ์๋์ฌ๋ผ. ์ ๋ผ๋ ์ฌํฌ๋ฆฌ๋ก ๋งํ ๋๊ป ๊ฐ์ด ์๊ธฐํด๋ณด์์!"
if "seoul" in d:
return "์๋
! ๋๋ ์์ฝ์ผ. ๋ค AI ๋ชจ์ ๋ํ ์๋์ผ. ์์ธ๋ง๋ก ์ด์ผ๊ธฐํ ๊ฒ. ๊ฐ์ด ์๊ธฐํ์!"
return "์๋
ํ์ธ์! ์ ๋ ์์ฝ์์. ๋น์ ์ AI ๋ชจ์ ๋ํ ์๋์
๋๋ค. ํ๊ตญ์ด๋ก ์ด์ผ๊ธฐํ ๊ฒ์. ํจ๊ป ๋ํํด์!"
common_phrases = [
(("no wahala", "no problem", "it's okay", "it is okay"), "๊ด์ฐฎ์์."),
(("i dey alright", "i am alright", "i'm alright", "i am fine"), "์ ๋ ๊ด์ฐฎ์์."),
(("you still dey", "are you still there", "you still there"), "์์ง ๊ฑฐ๊ธฐ ์์ด์?"),
(("sleep", "slept"), "๋ฒ์จ ์ค์ด์?"),
(("what happened", "wetin dey happen"), "๋ฌด์จ ์ผ์ด์์?"),
]
for needles, phrase in common_phrases:
if any(needle in lower_text for needle in needles):
return phrase
return "๋ค์ ํ ๋ฒ ๋งํด ์ฃผ์ธ์."
fallback_by_script = {
"Arabic": "ู
ู ูุถูู ุฃุนุฏ ุฅุฑุณุงู ุงูุฑุณุงูุฉ.",
"Greek": "ฮ ฮฑฯฮฑฮบฮฑฮปฯ ฯฯฮตฮฏฮปฯฮต ฮพฮฑฮฝฮฌ ฯฮฟ ฮผฮฎฮฝฯ
ฮผฮฑ.",
"Han": "่ฏทๅ่ฏดไธ้ใ",
"Japanese": "ใใไธๅบฆ่จใฃใฆใใ ใใใ",
"Thai": "เธเธฃเธธเธเธฒเธเธนเธเธญเธตเธเธเธฃเธฑเนเธ",
"Cyrillic": "ะะพะฒัะพัะธัะต, ะฟะพะถะฐะปัะนััะฐ.",
"Devanagari": "เคเฅเคชเคฏเคพ เคซเคฟเคฐ เคธเฅ เคเคนเฅเคเฅค",
"Bengali": "เฆ
เฆจเงเฆเงเฆฐเฆน เฆเฆฐเง เฆเฆฌเฆพเฆฐ เฆฌเฆฒเงเฆจเฅค",
"Tamil": "เฎฎเฏเฎฃเฏเฎเฏเฎฎเฏ เฎเฏเฎฒเฏเฎฒเฏเฎเฏเฎเฎณเฏ.",
"Telugu": "เฐฆเฐฏเฐเฑเฐธเฐฟ เฐฎเฐณเฑเฐฒเฑ เฐเฑเฐชเฑเฐชเฐเฐกเฐฟ.",
}
return fallback_by_script.get(rule["name"], "")
def _request_peer_translation(self, prompt):
if hasattr(self.brain, '_safe_generate'):
try:
return self.brain._safe_generate(prompt)
except Exception as e:
print(f"Peer Translation safe-generate Error: {e}")
manager = getattr(self.brain, "gemini_manager", None)
client = getattr(manager, "client", None)
if client and hasattr(client, "models"):
return client.models.generate_content(model='gemini-2.0-flash', contents=prompt)
if manager and hasattr(manager, "generate_fast") and not inspect.iscoroutinefunction(manager.generate_fast):
try:
return manager.generate_fast(prompt)
except Exception as e:
print(f"Peer Translation manager Error: {e}")
return None
def api_translate_peer(self, text, source_dialect, target_dialect, ai_model="auto"):
"""Translates an utterance directly from one dialect to another."""
if not text: return ""
script_rule = self._target_script_rule(target_dialect)
script_instruction = script_rule["instruction"] if script_rule else "Use the target language or dialect's normal writing system. For Igbo, keep proper Igbo letters and tone/dot marks such as แป, แปฅ, แป, แน
, แบน, รก, and ร where natural."
variety_instruction = "\n".join(filter(None, [
nigerian_variety_instruction(source_dialect, target_dialect),
hausa_variety_instruction(source_dialect, target_dialect),
]))
prompt = f"""
MODEL ROUTING PREFERENCE: {self._ai_model_hint(ai_model)}
You are a careful translation engine for PureHumCom.
Translate the following utterance from {source_dialect} to {target_dialect}.
Utterance: "{text}"
CRITICAL INSTRUCTIONS:
1. Output ONLY the translated text. No conversational filler.
2. Preserve the cultural pragmatics and emotional tone.
3. Do not explain the translation, just provide the direct equivalent in {target_dialect}.
4. SCRIPT FIDELITY: {script_instruction}
5. LANGUAGE CONSISTENCY: Once a language or dialect is established, strictly maintain translations in that dialect. Do not randomly switch to another language mid-conversation.
6. Never answer with punctuation-only text or romanization when a native script is required.
7. {variety_instruction}
"""
try:
translated = self._clean_peer_translation(self._request_peer_translation(prompt))
boundary_reason = nigerian_variety_retry_reason(translated, target_dialect)
if self._translation_has_required_script(translated, target_dialect) and not boundary_reason:
return translated
if boundary_reason:
retry_prompt = nigerian_variety_retry_prompt(
text, source_dialect, target_dialect, translated, boundary_reason
)
retry = self._clean_peer_translation(self._request_peer_translation(retry_prompt))
if self._translation_has_required_script(retry, target_dialect) and not nigerian_variety_retry_reason(retry, target_dialect):
return retry
if script_rule:
retry_prompt = f"""
Your previous output failed the script requirement for {target_dialect}.
Previous invalid output: "{translated}"
Rewrite this utterance from {source_dialect} to {target_dialect}.
Utterance: "{text}"
Output ONLY the corrected translation.
Required script: {script_rule["instruction"]}
"""
retry = self._clean_peer_translation(self._request_peer_translation(retry_prompt))
if self._translation_has_required_script(retry, target_dialect):
return retry
fallback = self._fallback_script_translation(text, target_dialect)
if fallback:
return fallback
return translated or text
except Exception as e:
print(f"Peer Translation Error: {e}")
return self._fallback_script_translation(text, target_dialect) or text
# ==========================================
# DIALECT RELAY (MATCHMAKING & CHAT)
# ==========================================
def api_join_queue(self, operator_id, dialect, target_partner_id=""):
import uuid, json, time
# Clean up any old ghost matches
if operator_id in self.active_matches:
del self.active_matches[operator_id]
# 1. Check if a specific target partner is requested and available
partner = None
if target_partner_id:
for p in self.waiting_pool:
if p['operator_id'] == target_partner_id:
partner = p
break
# If targeting someone and they exist, or just picking someone (if we wanted auto)
if partner:
# 2. Match found! Create a room.
self.waiting_pool.remove(partner)
room_id = f"FREQ-{uuid.uuid4().hex[:6].upper()}"
self.active_matches[operator_id] = {"room_id": room_id, "partner_dialect": partner['dialect']}
self.active_matches[partner['operator_id']] = {"room_id": room_id, "partner_dialect": dialect}
self.live_rooms[room_id] = []
return json.dumps({"status": "matched", "room_id": room_id, "partner_dialect": partner['dialect']})
else:
# 3. Add self to pool to wait.
self.waiting_pool = [p for p in self.waiting_pool if p['operator_id'] != operator_id] # Prevent duplicates
self.waiting_pool.append({"operator_id": operator_id, "dialect": dialect, "time": time.time()})
return json.dumps({"status": "waiting"})
def api_get_lobby(self):
import json
# Return only dialect and masked operator ID for privacy/display
lobby_data = [{"id": p["operator_id"], "dialect": p["dialect"]} for p in self.waiting_pool]
return json.dumps(lobby_data)
def api_check_match(self, operator_id):
import json
if operator_id in self.active_matches:
match = self.active_matches[operator_id]
# Consume the match so it cannot be re-used on a second poll (prevents ghost room reconnection)
del self.active_matches[operator_id]
return json.dumps({"status": "matched", "room_id": match["room_id"], "partner_dialect": match["partner_dialect"]})
return json.dumps({"status": "waiting"})
def api_leave_queue(self, operator_id):
import json
self.waiting_pool = [p for p in self.waiting_pool if p['operator_id'] != operator_id]
if operator_id in self.active_matches:
del self.active_matches[operator_id]
return json.dumps({"status": "left"})
def api_remote_eval_and_send(self, room_code, sender_id, text, source_dialect, target_dialect, meaning_to_send=""):
import json, os, re, uuid, threading
import pandas as pd
if not text: return json.dumps({"status": "error", "msg": "Empty text"})
standard_meaning = meaning_to_send.strip() if meaning_to_send and meaning_to_send.strip() else text
# 1. Forward Lookup (Understand Source - used for dataset clarity if meaning wasn't provided)
if not meaning_to_send or not meaning_to_send.strip():
eval_result = self.brain.search_local_dataset(text)
if not eval_result: eval_result = self.brain.search_personas(text)
if eval_result:
standard_meaning = eval_result.get("clarification", text)
# ๐ข THE DATA COLLECTION HOOK (Silent Background Saving)
clarification_to_save = standard_meaning
try:
threading.Thread(
target=self.check_and_submit_logic,
args=(text, source_dialect, "", clarification_to_save, "Conversational", "Relay Peer-to-Peer Chat", "Automated Relay Extraction", "Game: Relay Pair", "AI Relay Bouncer", sender_id, None, False),
daemon=True
).start()
except Exception as save_err:
print(f"Data Hook Error: {save_err}")
target_translation = standard_meaning
target_label = target_dialect or "Standard English"
if target_dialect and target_dialect != "Standard English":
try:
target_translation = self.api_translate_peer(standard_meaning, "Standard English", target_dialect)
except Exception as translate_err:
print(f"Relay target translation failed: {translate_err}")
# 4. Route to Room
if room_code not in self.live_rooms: self.live_rooms[room_code] = []
msg = {"sender": sender_id, "original": text, "translation": target_translation, "dialect": source_dialect, "target_dialect": target_label, "standard_meaning": standard_meaning, "id": str(uuid.uuid4())[:8]}
self.live_rooms[room_code].append(msg)
return json.dumps({"status": "success", "message": msg})
def api_remote_poll(self, room_code, last_index):
import json
try:
idx = int(float(last_index or 0))
except Exception:
idx = 0
return json.dumps(self.live_rooms.get(room_code, [])[idx:])
def api_get_user_xp(self, operator_id, dialect):
import json, os
try:
w3 = self.trust.w3
contract_addr = os.environ.get("PURECHAIN_CONTRACT_ADDRESS", getattr(self.trust.config, 'PURECHAIN_CONTRACT_ADDRESS', None))
if not contract_addr or not w3: return json.dumps({"xp": 0})
raw_operator = str(operator_id).strip().lower()
try:
final_operator_id = w3.to_checksum_address(raw_operator)
except:
return json.dumps({"xp": 0})
contract = w3.eth.contract(address=contract_addr, abi=self.trust.config.CONTRACT_ABI)
xp = contract.functions.getXP(final_operator_id, dialect).call()
return json.dumps({"xp": xp})
except Exception as e:
print(f"XP Fetch Error: {e}")
return json.dumps({"xp": 0})
def api_get_oracle_task(self, operator_id, dialect, review_key=""):
import json, pandas as pd, os
scope = self._review_scope(review_key)
if not scope:
return self._review_unauthorized_json()
if not self._scope_allows_language_dialect(scope, "", dialect):
return self._review_unauthorized_json()
if not os.path.exists(self.PENDING_FILE): return json.dumps({"status": "empty"})
try:
df = pd.read_csv(self.PENDING_FILE)
df = self._infer_pending_languages(df)
if "Dialect" not in df.columns or "Chain_ID" not in df.columns: return json.dumps({"status": "empty"})
target_df = df[(df["Dialect"] == dialect) & (df["Chain_ID"].notna()) & (df["Chain_ID"] != "")]
target_df = self._filter_dataframe_by_review_scope(target_df, scope)
for idx, row in target_df.iterrows():
approvers = str(row.get("Approvers", "")).split(",")
if operator_id not in approvers and row.get("User", "") != operator_id:
return json.dumps({
"status": "success",
"row_id": int(idx),
"chain_id": str(row["Chain_ID"]),
"phrase": row.get("Utterance", ""),
"clarification": row.get("Clarification", ""),
"audio": row.get("Audio", "")
})
return json.dumps({"status": "empty"})
except Exception as e:
print(f"Get Oracle Task Error: {e}")
return json.dumps({"status": "empty"})
def api_submit_oracle_review(self, row_id, chain_id, operator_id, dialect, is_approve, reason="Oracle Peer Review Failed", review_key=""):
import json, pandas as pd, os, threading
if not self._is_review_authorized(review_key) and self._is_review_authorized(reason):
review_key, reason = reason, "Oracle Peer Review Failed"
scope = self._review_scope(review_key)
if not scope:
return self._review_unauthorized_json()
try:
df = pd.read_csv(self.PENDING_FILE)
df = self._infer_pending_languages(df)
if int(row_id) not in df.index: return json.dumps({"status": "error"})
if not self._pending_row_allowed_by_scope(df.loc[int(row_id)], scope):
return self._review_unauthorized_json()
approvers = str(df.at[int(row_id), "Approvers"]).split(",")
approvers.append(operator_id)
df.at[int(row_id), "Approvers"] = ",".join(filter(None, approvers))
df.to_csv(self.PENDING_FILE, index=False)
self.sync_pending_queue(direction="up")
def run_chain_verify():
try:
w3 = self.trust.w3
contract_addr = os.environ.get("PURECHAIN_CONTRACT_ADDRESS", getattr(self.trust.config, 'PURECHAIN_CONTRACT_ADDRESS', None))
contract = w3.eth.contract(address=contract_addr, abi=self.trust.config.CONTRACT_ABI)
final_operator = w3.to_checksum_address(operator_id)
nonce = w3.eth.get_transaction_count(self.trust.account.address)
approve_bool = str(is_approve).lower() == "true"
if approve_bool:
tx = contract.functions.verifyEntry(int(chain_id), final_operator).build_transaction({
'chainId': getattr(self.trust.config, 'PURECHAIN_ID', 900520900520),
'gas': 2000000, 'gasPrice': 0, 'nonce': nonce
})
else:
tx = contract.functions.rejectEntry(int(chain_id), final_operator, reason).build_transaction({
'chainId': getattr(self.trust.config, 'PURECHAIN_ID', 900520900520),
'gas': 2000000, 'gasPrice': 0, 'nonce': nonce
})
signed_tx = w3.eth.account.sign_transaction(tx, self.trust.config.PRIVATE_KEY)
tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction)
receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
if approve_bool:
events = contract.events.EntryVerified().process_receipt(receipt)
if events:
print("๐ Oracle consensus reached (2 approvals). Moving to IEDID!")
row = df.loc[int(row_id)]
self.trust.update_dataset_csv(
dialect=row["Dialect"], utterance=row["Utterance"], clarification=row["Clarification"],
tone=row["Tone"], context=row.get("Context", ""), syntax="", audio_path=row.get("Audio", ""),
pragmatics=row.get("Pragmatic_Analysis", ""), sourceTag="Game (Verified)", clar_source="User",
userKey=row.get("User", "")
)
else:
row = df.loc[int(row_id)]
self.trust.add_to_graveyard(
dialect=row.get("Dialect", ""), utterance=row.get("Utterance", ""), clarification=row.get("Clarification", ""),
tone=row.get("Tone", ""), context=row.get("Context", ""), syntax="", audio_path=row.get("Audio", ""),
pragmatics=row.get("Pragmatic_Analysis", ""), sourceTag="Game (Rejected)", clar_source="User",
userKey=row.get("User", ""), reason=reason, rejector=operator_id, chain_id=chain_id
)
print("๐ฅ Oracle rejected entry. Added to Burn Ward.")
except Exception as e:
print(f"Blockchain verify error: {e}")
threading.Thread(target=run_chain_verify, daemon=True).start()
return json.dumps({"status": "success"})
except Exception as e:
print(f"Submit Oracle Error: {e}")
return json.dumps({"status": "error"})
def api_get_graveyard_task(self, operator_id, dialect, review_key=""):
import json, pandas as pd, os
scope = self._review_scope(review_key)
if not scope:
return self._review_unauthorized_json()
if not self._scope_allows_language_dialect(scope, "", dialect):
return self._review_unauthorized_json()
graveyard_path = os.path.join(self.trust.config.DATASET_DIR, "..", "rejected_graveyard.csv")
if not os.path.exists(graveyard_path): return json.dumps({"status": "empty"})
try:
df = pd.read_csv(graveyard_path, encoding='utf-8-sig', on_bad_lines='skip')
if "Dialect" not in df.columns or "Chain_ID" not in df.columns: return json.dumps({"status": "empty"})
target_df = df[(df["Dialect"] == dialect) & (df["Chain_ID"].notna()) & (df["Chain_ID"] != "")]
for idx, row in target_df.iterrows():
if str(row.get("User", "")) != operator_id:
return json.dumps({
"status": "success",
"row_id": int(idx),
"chain_id": str(row["Chain_ID"]),
"phrase": str(row.get("Utterance", "")),
"clarification": str(row.get("Clarification", "")),
"audio": str(row.get("audio_file_name", "")),
"reason": str(row.get("Rejection_Reason", "Unknown")),
"rejector": str(row.get("Rejector", "Unknown"))
})
return json.dumps({"status": "empty"})
except Exception as e:
print(f"Get Graveyard Task Error: {e}")
return json.dumps({"status": "empty"})
def api_submit_appeal(self, chain_id, operator_id, dialect, review_key=""):
import json, os, threading, pandas as pd
scope = self._review_scope(review_key)
if not scope:
return self._review_unauthorized_json()
if not self._scope_allows_language_dialect(scope, "", dialect):
return self._review_unauthorized_json()
try:
graveyard_path = os.path.join(self.trust.config.DATASET_DIR, "..", "rejected_graveyard.csv")
if os.path.exists(graveyard_path):
try:
df = pd.read_csv(graveyard_path, encoding='utf-8-sig', on_bad_lines='skip')
df['Chain_ID'] = df['Chain_ID'].astype(str)
df = df[df["Chain_ID"] != str(chain_id)]
df.to_csv(graveyard_path, index=False, encoding='utf-8-sig')
except Exception as e:
print(f"Graveyard update error: {e}")
def run_chain_appeal():
try:
w3 = self.trust.w3
contract_addr = os.environ.get("PURECHAIN_CONTRACT_ADDRESS", getattr(self.trust.config, 'PURECHAIN_CONTRACT_ADDRESS', None))
contract = w3.eth.contract(address=contract_addr, abi=self.trust.config.CONTRACT_ABI)
final_operator = w3.to_checksum_address(operator_id)
nonce = w3.eth.get_transaction_count(self.trust.account.address)
tx = contract.functions.appealEntry(int(chain_id), final_operator).build_transaction({
'chainId': getattr(self.trust.config, 'PURECHAIN_ID', 900520900520),
'gas': 2000000, 'gasPrice': 0, 'nonce': nonce
})
signed_tx = w3.eth.account.sign_transaction(tx, self.trust.config.PRIVATE_KEY)
tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction)
w3.eth.wait_for_transaction_receipt(tx_hash)
print(f"โ๏ธ Appeal filed for Chain ID: {chain_id} by {final_operator}")
except Exception as e:
print(f"Blockchain appeal error: {e}")
threading.Thread(target=run_chain_appeal, daemon=True).start()
return json.dumps({"status": "success"})
except Exception as e:
print(f"Submit Appeal Error: {e}")
return json.dumps({"status": "error"})
def api_get_room_messages(self, room_code, last_index):
idx = int(last_index)
return json.dumps(self.live_rooms[room_code][idx:])
# ==========================================
# SOCIOLINGUISTIC PIPELINE
# ==========================================
def automated_pipeline(self, audio_path, language_code, request: gr.Request = None):
client_ip = request.headers.get("x-forwarded-for") or request.client.host if request else "Unknown_IP"
source_tag = f"Lab_Admin_{client_ip}"
print(f"๐ PIPELINE TRIGGERED by {source_tag}")
headers = ["Source", "Speaker", "Utterance", "Dialect", "Clarification", "Tone", "Context", "Pragmatic Analysis"]
empty_df = pd.DataFrame(columns=headers)
if not audio_path:
yield empty_df, "Waiting for input...", self.get_quota_status(), "", None, "Neutral / Conversational", "", ""
return
print("\n" + "="*40)
print(f"๐ PIPELINE TRIGGERED for {language_code}!")
print(f"Audio Path: {audio_path}")
self.last_audio_path = audio_path
status_log = "๐ง Transcribing Audio...\n"
try:
print("โณ STEP 1: Calling Whisper Transcription...")
if hasattr(self.input, 'transcribe'):
asr_code, _, _ = resolve_asr_language_code(language_code)
transcription_result = self.input.transcribe(audio_path, language=asr_code)
print(f"โ
Whisper Result: {transcription_result}")
if isinstance(transcription_result, list) and len(transcription_result) > 0:
transcribed_text = transcription_result[0].get('text', str(transcription_result))
else:
transcribed_text = str(transcription_result)
else:
transcribed_text = "Audio Received."
status_log += f"๐ฃ๏ธ Heard: '{transcribed_text}'\n\n"
print(f"โณ STEP 2: Checking Local Dataset...")
status_log += "๐๏ธ Checking Local Dataset (Fast Match)...\n"
final_result = self.brain.search_local_dataset(transcribed_text)
if not final_result:
print("โณ STEP 3: Checking Persona...")
status_log += "โ No local match. ๐ญ Checking Persona Context...\n"
final_result = self.brain.search_personas(transcribed_text)
if not final_result:
print("โณ STEP 4: Sending to Gemini API...")
status_log += "โ No persona hit. ๐ง Generating AI interpretation...\n"
if hasattr(self.brain, 'analyze_dialect_single'):
final_result = self.brain.analyze_dialect_single(transcribed_text, language_code)
print(f"โ
Gemini Result: {final_result}")
else:
raise Exception("analyze_dialect_single missing from brain_agent.")
status_log += f"\nโ
Analysis Complete via {final_result.get('Source', 'Unknown')}."
def safe_str(val, default=""):
return default if pd.isna(val) or val is None else str(val)
df_data = [[
safe_str(final_result.get("Source", "Unknown")),
source_tag,
safe_str(transcribed_text),
safe_str(final_result.get("dialect", "")),
safe_str(final_result.get("clarification", "")),
safe_str(final_result.get("tone", "")),
safe_str(final_result.get("context", "")),
safe_str(final_result.get("pragmatics", ""))
]]
df = pd.DataFrame(df_data, columns=headers)
print("โ
PIPELINE COMPLETED SUCCESSFULLY.")
print("="*40 + "\n")
yield (
df, safe_str(status_log), self.get_quota_status(), safe_str(transcribed_text),
safe_str(final_result.get("dialect")), safe_str(final_result.get("clarification")),
safe_str(final_result.get("tone", "Neutral / Conversational")),
safe_str(final_result.get("context")), safe_str(final_result.get("pragmatics"))
)
return
except Exception as e:
print("\n๐จ CRITICAL PIPELINE ERROR ๐จ")
traceback.print_exc()
print("="*40 + "\n")
status_log += f"\nโ System Error: {e}\n"
yield empty_df, status_log, self.get_quota_status(), "", None, "Neutral / Conversational", "", ""
return
# ==========================================
# RESEARCH DATA SUBMISSION
# ==========================================
def check_and_submit_logic(
self, transcribed, dialect, customD, clarification, tone, context, pragmatics,
sourceTag="Web", clar_source="User", userKey="", blob=None, confirm=False,
language="", request: gr.Request = None
):
print("\n" + "="*40)
print(f"๐ฅ API HIT: /check_and_submit_logic")
print(f" Source: {sourceTag}")
print(f" Text: '{transcribed}'")
print(f" Dialect: {dialect}")
print("="*40)
clean_key = str(userKey).strip()
if not clean_key.startswith("0x"):
clean_key = "0x" + uuid.uuid4().hex + uuid.uuid4().hex[:8]
final_user = clean_key
if sourceTag == "Web" or sourceTag == "":
final_origin = "Gradio Admin UI"
else:
final_origin = sourceTag
if not transcribed or not clarification:
return "โ ๏ธ Cannot submit empty analysis.", gr.update(visible=False)
final_d = customD if (dialect == "+ Add New Dialect" and customD) else dialect
if not final_d or final_d == "+ Add New Dialect":
return "โ ๏ธ Select a dialect first.", gr.update(visible=False)
voice_privacy = self._voice_privacy_metadata(final_origin, pragmatics, context, clar_source)
voice_acoustic_profile = json.dumps({"voice_privacy": voice_privacy}, ensure_ascii=False)
permanent_audio_path = ""
if voice_privacy.get("raw_audio_uploaded") and blob is not None:
actual_path = None
if isinstance(blob, str): actual_path = blob
elif isinstance(blob, dict) and 'path' in blob: actual_path = blob['path']
elif hasattr(blob, 'name'): actual_path = blob.name
if actual_path and os.path.exists(actual_path):
save_dir = os.path.join(self.brain.config.DATASET_DIR, "audio")
os.makedirs(save_dir, exist_ok=True)
unique_name = f"rec_{int(time.time())}_{uuid.uuid4().hex[:6]}.wav"
permanent_audio_path = os.path.join(save_dir, unique_name)
shutil.copy(actual_path, permanent_audio_path)
print(f"โ
Audio securely extracted to: {permanent_audio_path}")
else:
print(f"โ ๏ธ Audio skipped! Blob invalid: {blob}")
elif voice_privacy.get("raw_audio_uploaded") and self.last_audio_path:
permanent_audio_path = self.last_audio_path
elif blob is not None or self.last_audio_path:
print("๐ Voice privacy active: raw audio was not persisted; transcript-only metadata will be stored.")
is_game_submission = ("Game" in final_origin)
if not is_game_submission:
success = self.trust.update_dataset_csv(
final_d, transcribed, clarification, tone, context, "",
permanent_audio_path, pragmatics, final_origin, clar_source, final_user,
acoustic_profile=voice_acoustic_profile, language=language
)
if success:
payload = { "original": transcribed, "dialect": final_d, "clarification": clarification, "tone": tone, "user": final_user }
threading.Thread(target=self.trust.stamp_on_chain, args=(payload,), daemon=True).start()
return f"๐ Approved and Minted to {final_d}", gr.update(visible=False)
else:
new_entry = {
"User": final_user,
"Data_Origin": final_origin,
"Utterance": transcribed,
"Dialect": final_d,
"Clarification": clarification,
"Clarification_Source": clar_source,
"Tone": tone,
"Context": context,
"Pragmatic_Analysis": pragmatics,
"Acoustic_Profile": voice_acoustic_profile,
"Audio": permanent_audio_path,
"Timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"Chain_ID": "",
"Approvers": "",
"Language": language,
"Voice_Privacy_Mode": voice_privacy.get("voice_privacy_mode", "transcript_only"),
"Voice_Audio_Consent": str(bool(voice_privacy.get("voice_audio_consent", False))),
"Raw_Audio_Uploaded": str(bool(voice_privacy.get("raw_audio_uploaded", False))),
}
# Start a thread to stamp on chain and then save to CSV
def process_game_submission():
payload = { "original": transcribed, "dialect": final_d, "clarification": clarification, "tone": tone, "user": final_user }
entry_id = self.trust.stamp_on_chain(payload)
if str(entry_id).isdigit():
new_entry["Chain_ID"] = str(entry_id)
try:
if os.path.exists(self.PENDING_FILE):
df = pd.read_csv(self.PENDING_FILE)
else:
df = pd.DataFrame(columns=new_entry.keys())
df = pd.concat([df, pd.DataFrame([new_entry])], ignore_index=True)
df.to_csv(self.PENDING_FILE, index=False)
self.sync_pending_queue(direction="up")
if permanent_audio_path and os.path.exists(permanent_audio_path):
try:
api = HfApi(token=os.environ.get("HF_TOKEN"))
api.upload_file(
path_or_fileobj=permanent_audio_path,
path_in_repo=f"pending_audio/{os.path.basename(permanent_audio_path)}",
repo_id="toecm/PureChain_Dataset",
repo_type="dataset",
commit_message=f"๐๏ธ Staging pending audio from {final_origin}"
)
except Exception as e:
print(f"โ ๏ธ Failed to stage audio: {e}")
except Exception as e:
print(f"Failed to save game submission: {e}")
threading.Thread(target=process_game_submission, daemon=True).start()
return "๐ฅ Submitted for Peer Review (XP pending approval)", gr.update(visible=False)
def force_overwrite_logic(self, *args):
return self.check_and_submit_logic(*args, confirm=True)
def admin_check_and_submit_logic(self, *args):
review_key = args[-1] if args else ""
if not self._is_full_admin_authorized(review_key):
return self._full_admin_required_message(), gr.update(visible=False)
return self.check_and_submit_logic(*args[:-1])
def admin_force_overwrite_logic(self, *args):
review_key = args[-1] if args else ""
if not self._is_full_admin_authorized(review_key):
return self._full_admin_required_message(), gr.update(visible=False)
return self.check_and_submit_logic(*args[:-1], confirm=True)
# ==========================================
# FEEDBACK & AUDIT HELPERS
# ==========================================
def handle_feedback_submission(self, op_id, text, img_blob):
"""Catches secure feedback from React games and logs it with images."""
feedback_file = "/app/system_feedback.csv"
image_path = ""
print("\n" + "="*40)
print("๐ก๏ธ SECURE FEEDBACK RECEIVED")
print(f"Operator: {op_id}")
# 1. Process Image if attached
if img_blob is not None:
actual_path = None
if isinstance(img_blob, str): actual_path = img_blob
elif hasattr(img_blob, 'name'): actual_path = img_blob.name
if actual_path and os.path.exists(actual_path):
unique_name = f"bug_img_{int(time.time())}.png"
image_path = os.path.join("/app", unique_name)
shutil.copy(actual_path, image_path)
try:
api = HfApi(token=os.environ.get("HF_TOKEN"))
api.upload_file(
path_or_fileobj=image_path,
path_in_repo=f"feedback_images/{unique_name}",
repo_id="toecm/PureChain_Dataset",
repo_type="dataset",
commit_message="๐ธ New Bug Report Image"
)
except Exception as e:
print(f"Image upload failed: {e}")
# 2. Log to CSV
new_row = {
"Timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"Operator_ID": op_id,
"Feedback_Text": text,
"Image_Reference": image_path
}
try:
if os.path.exists(feedback_file):
df = pd.read_csv(feedback_file)
else:
df = pd.DataFrame(columns=new_row.keys())
df = pd.concat([df, pd.DataFrame([new_row])], ignore_index=True)
df.to_csv(feedback_file, index=False)
try:
api = HfApi(token=os.environ.get("HF_TOKEN"))
api.upload_file(
path_or_fileobj=feedback_file,
path_in_repo="system_feedback.csv",
repo_id="toecm/PureChain_Dataset",
repo_type="dataset",
commit_message="๐ Updated System Feedback Log"
)
except: pass
except Exception as e:
print(f"Feedback save error: {e}")
return "Success"
def get_feedback_dataframe(self, review_key=""):
if not self._is_full_admin_authorized(review_key):
return self._review_empty_feedback_dataframe()
feedback_file = "/app/system_feedback.csv"
hf_token = os.environ.get("HF_TOKEN")
repo_id = "toecm/PureChain_Dataset"
if hf_token:
try:
downloaded = hf_hub_download(repo_id=repo_id, filename="system_feedback.csv", repo_type="dataset", token=hf_token)
shutil.copy(downloaded, feedback_file)
except Exception as e:
pass
if os.path.exists(feedback_file):
try:
df = pd.read_csv(feedback_file)
for col in ["Timestamp", "Operator_ID", "Feedback_Text", "Image_Reference"]:
if col not in df.columns: df[col] = ""
return df.sort_values(by="Timestamp", ascending=False)
except:
pass
return self._review_empty_feedback_dataframe()
def _eeqs_local_file(self):
base_dir = "/app" if os.path.isdir("/app") else os.getcwd()
return os.path.join(base_dir, "eeqs_events.csv")
def _manual_eeqs_local_file(self):
base_dir = "/app" if os.path.isdir("/app") else os.getcwd()
return os.path.join(base_dir, "eeqs_manual_events.csv")
def _empty_eeqs_dashboard(self):
return pd.DataFrame(columns=[
"Game", "AI_Model", "ASR_Model", "Source_Language", "Source_Dialect",
"Target_Language", "Target_Dialect", "Interaction_Type", "Event_Count",
"Mean_EEQ", "Last_EEQ", "Last_Seen"
])
def _load_eeqs_events(self):
eeqs_file = self._eeqs_local_file()
hf_token = os.environ.get("HF_TOKEN")
repo_id = "toecm/PureChain_Dataset"
if hf_token:
try:
downloaded = hf_hub_download(
repo_id=repo_id,
filename="eeqs_events.csv",
repo_type="dataset",
token=hf_token
)
shutil.copy(downloaded, eeqs_file)
except Exception:
pass
if os.path.exists(eeqs_file):
try:
return pd.read_csv(eeqs_file)
except Exception:
pass
return pd.DataFrame()
def _load_manual_eeqs_events(self):
manual_file = self._manual_eeqs_local_file()
hf_token = os.environ.get("HF_TOKEN")
repo_id = "toecm/PureChain_Dataset"
if hf_token:
try:
downloaded = hf_hub_download(
repo_id=repo_id,
filename="eeqs_manual_events.csv",
repo_type="dataset",
token=hf_token
)
shutil.copy(downloaded, manual_file)
except Exception:
pass
if os.path.exists(manual_file):
try:
return pd.read_csv(manual_file)
except Exception:
pass
return pd.DataFrame()
def handle_eeqs_submission(self, op_id, payload_json):
"""Stores user-synced passive and manual EEQ-s events for audit dashboards."""
eeqs_file = self._eeqs_local_file()
manual_file = self._manual_eeqs_local_file()
hf_token = os.environ.get("HF_TOKEN")
repo_id = "toecm/PureChain_Dataset"
try:
payload = json.loads(payload_json or "{}")
events = payload.get("events", [])
manual_events = payload.get("manual_events", [])
if not isinstance(events, list):
events = []
if not isinstance(manual_events, list):
manual_events = []
if not events and not manual_events:
return "No EEQ-s events received."
rows = []
manual_rows = []
sync_ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
for event in events:
if not isinstance(event, dict):
continue
items = event.get("items") if isinstance(event.get("items"), dict) else {}
row = {
"Sync_Timestamp": sync_ts,
"Timestamp": event.get("timestamp", ""),
"Operator_ID": op_id or "Anonymous",
"App": payload.get("app") or event.get("app", ""),
"Host": payload.get("host") or event.get("host", ""),
"AI_Model": event.get("aiModel") or payload.get("ai_model") or self._resolve_ai_model_label(),
"ASR_Model": event.get("asrModel") or payload.get("asr_model") or self._resolve_asr_model_label(),
"Event_ID": event.get("id", ""),
"Game": event.get("game", ""),
"Interaction_Type": event.get("interaction", ""),
"Source_Language": event.get("sourceLanguage", ""),
"Source_Dialect": event.get("sourceDialect", ""),
"Target_Language": event.get("targetLanguage", ""),
"Target_Dialect": event.get("targetDialect", ""),
"Index": event.get("index", ""),
"Accepted": event.get("accepted", ""),
"Error": event.get("error", ""),
"Regenerated": event.get("regenerated", ""),
"Explained": event.get("explained", ""),
"Played_Audio": event.get("playedAudio", ""),
"Sent": event.get("sent", ""),
"Transcript_Edit_Ratio": event.get("transcriptEditRatio", ""),
"Meaning_Edit_Ratio": event.get("meaningEditRatio", ""),
"Translation_Edit_Ratio": event.get("translationEditRatio", ""),
}
for key, value in items.items():
row[key] = value
rows.append(row)
for event in manual_events:
if not isinstance(event, dict):
continue
scores = event.get("scores") if isinstance(event.get("scores"), dict) else {}
manual_rows.append({
"Sync_Timestamp": sync_ts,
"Timestamp": event.get("timestamp", ""),
"Operator_ID": op_id or "Anonymous",
"App": payload.get("app") or event.get("app", ""),
"Host": payload.get("host") or event.get("host", ""),
"Event_ID": event.get("id", ""),
"Source_Event_ID": event.get("source_event_id", ""),
"Game": event.get("game", ""),
"Interaction_Type": event.get("interaction", ""),
"Source_Language": event.get("sourceLanguage", ""),
"Source_Dialect": event.get("sourceDialect", ""),
"Target_Language": event.get("targetLanguage", ""),
"Target_Dialect": event.get("targetDialect", ""),
"Manual_Index": event.get("manual_index", ""),
"Passive_Index": event.get("passive_index", ""),
"Score_Understood": scores.get("understood", ""),
"Score_Tone": scores.get("tone", ""),
"Score_Dialect": scores.get("dialect", ""),
"Score_Repair": scores.get("repair", ""),
"Note": event.get("note", ""),
})
if not rows and not manual_rows:
return "No valid EEQ-s events received."
if rows:
existing = self._load_eeqs_events()
new_df = pd.DataFrame(rows)
df = pd.concat([existing, new_df], ignore_index=True)
if "Event_ID" in df.columns:
df = df.drop_duplicates(subset=["Event_ID"], keep="last")
df.to_csv(eeqs_file, index=False)
if manual_rows:
existing_manual = self._load_manual_eeqs_events()
new_manual_df = pd.DataFrame(manual_rows)
manual_df = pd.concat([existing_manual, new_manual_df], ignore_index=True)
if "Event_ID" in manual_df.columns:
manual_df = manual_df.drop_duplicates(subset=["Event_ID"], keep="last")
manual_df.to_csv(manual_file, index=False)
if hf_token:
api = HfApi(token=hf_token)
if rows:
api.upload_file(
path_or_fileobj=eeqs_file,
path_in_repo="eeqs_events.csv",
repo_id=repo_id,
repo_type="dataset",
commit_message="Update Passive EEQ-s dashboard events"
)
if manual_rows:
api.upload_file(
path_or_fileobj=manual_file,
path_in_repo="eeqs_manual_events.csv",
repo_id=repo_id,
repo_type="dataset",
commit_message="Update Manual EEQ-s audit events"
)
return f"EEQ-s synced: {len(rows)} passive and {len(manual_rows)} manual events saved to global dashboard."
return f"EEQ-s saved locally on backend: {len(rows)} passive and {len(manual_rows)} manual events. HF_TOKEN is not configured."
except Exception as e:
print(f"EEQ-s save error: {e}")
return f"EEQ-s sync error: {e}"
def get_eeqs_dashboard_dataframe(self, review_key=""):
if not self._is_full_admin_authorized(review_key):
return self._empty_eeqs_dashboard()
df = self._load_eeqs_events()
if df.empty:
return self._empty_eeqs_dashboard()
required = [
"Game", "AI_Model", "ASR_Model", "Source_Language", "Source_Dialect",
"Target_Language", "Target_Dialect", "Interaction_Type", "Index", "Timestamp"
]
for col in required:
if col not in df.columns:
df[col] = ""
df["Index"] = pd.to_numeric(df["Index"], errors="coerce")
df = df.dropna(subset=["Index"])
if df.empty:
return self._empty_eeqs_dashboard()
group_cols = [
"Game", "AI_Model", "ASR_Model", "Source_Language", "Source_Dialect",
"Target_Language", "Target_Dialect", "Interaction_Type"
]
grouped = df.groupby(group_cols, dropna=False).agg(
Event_Count=("Index", "count"),
Mean_EEQ=("Index", "mean"),
Last_EEQ=("Index", "last"),
Last_Seen=("Timestamp", "last")
).reset_index()
grouped["Mean_EEQ"] = grouped["Mean_EEQ"].round(1)
grouped["Last_EEQ"] = grouped["Last_EEQ"].round(1)
return grouped.sort_values(by=["Event_Count", "Mean_EEQ"], ascending=[False, False])
def get_eeqs_download_file(self, review_key=""):
if not self._is_full_admin_authorized(review_key):
return None
eeqs_file = self._eeqs_local_file()
manual_file = self._manual_eeqs_local_file()
base_dir = "/app" if os.path.isdir("/app") else os.getcwd()
export_zip = os.path.join(base_dir, "eeqs_audit_export.zip")
df = self._load_eeqs_events()
if df.empty:
df = pd.DataFrame(columns=[
"Sync_Timestamp", "Timestamp", "Operator_ID", "App", "Host",
"AI_Model", "ASR_Model", "Event_ID", "Game", "Interaction_Type",
"Source_Language", "Source_Dialect", "Target_Language",
"Target_Dialect", "Index", "Accepted", "Error", "Regenerated",
"Explained", "Played_Audio", "Sent", "Transcript_Edit_Ratio",
"Meaning_Edit_Ratio", "Translation_Edit_Ratio"
])
df.to_csv(eeqs_file, index=False)
manual_df = self._load_manual_eeqs_events()
if manual_df.empty:
manual_df = pd.DataFrame(columns=[
"Sync_Timestamp", "Timestamp", "Operator_ID", "App", "Host",
"Event_ID", "Source_Event_ID", "Game", "Interaction_Type",
"Source_Language", "Source_Dialect", "Target_Language",
"Target_Dialect", "Manual_Index", "Passive_Index",
"Score_Understood", "Score_Tone", "Score_Dialect",
"Score_Repair", "Note"
])
manual_df.to_csv(manual_file, index=False)
with zipfile.ZipFile(export_zip, "w", zipfile.ZIP_DEFLATED) as zf:
zf.write(eeqs_file, arcname="eeqs_passive_events.csv")
zf.write(manual_file, arcname="eeqs_manual_events.csv")
return export_zip
def get_pending_dataframe(self, language_filter="All", dialect_filter="All", review_key=""):
cols = [
"User", "Data_Origin", "Utterance", "Language", "Dialect",
"Clarification", "Clarification_Source", "Tone", "Audio", "Timestamp",
"App_Source", "Source_Language", "Source_Dialect", "Source_Input_Mode",
"Machine_Transcript_Initial", "User_Transcript_Final",
"Transcript_Edit_Distance", "Machine_Translation_Initial",
"User_Translation_Final", "Translation_Edit_Distance", "ASR_Model",
"AI_Model", "Audio_Sanitation", "Consent_Confirmed", "Queue_ID"
]
scope = self._review_scope(review_key)
if not scope:
return pd.DataFrame(columns=cols)
if os.path.exists(self.PENDING_FILE):
df = pd.read_csv(self.PENDING_FILE)
for c in cols:
if c not in df.columns: df[c] = ""
df = self._infer_pending_languages(df)
df = self._filter_dataframe_by_review_scope(df, scope)
if language_filter and language_filter != "All":
df = df[df["Language"] == language_filter]
if dialect_filter and dialect_filter != "All":
df = df[df["Dialect"] == dialect_filter]
return df[cols]
return pd.DataFrame(columns=cols)
def sync_pending_queue(self, direction="up"):
hf_token = os.environ.get("HF_TOKEN")
repo_id = "toecm/PureChain_Dataset"
if not hf_token:
print("โ ๏ธ Skipping Pending Sync: No HF_TOKEN found.")
return
api = HfApi(token=hf_token)
if direction == "up":
if os.path.exists(self.PENDING_FILE):
try:
api.upload_file(
path_or_fileobj=self.PENDING_FILE,
path_in_repo="pending_approvals.csv",
repo_id=repo_id,
repo_type="dataset",
commit_message="๐ Auto-sync: Updated pending approvals queue"
)
print("โ๏ธ Pending queue backed up to PureChain_Dataset.")
except Exception as e:
print(f"โ ๏ธ Failed to upload pending queue: {e}")
elif direction == "down":
try:
downloaded_path = hf_hub_download(
repo_id=repo_id,
filename="pending_approvals.csv",
repo_type="dataset",
token=hf_token
)
shutil.copy(downloaded_path, self.PENDING_FILE)
print("โ๏ธ Pending queue CSV restored from PureChain_Dataset.")
df = pd.read_csv(self.PENDING_FILE)
for audio_path in df['Audio'].dropna():
if audio_path and not os.path.exists(audio_path):
try:
audio_filename = os.path.basename(audio_path)
print(f"โ๏ธ Recovering missing audio: {audio_filename}...")
audio_dl = hf_hub_download(
repo_id=repo_id,
filename=f"pending_audio/{audio_filename}",
repo_type="dataset",
token=hf_token
)
os.makedirs(os.path.dirname(audio_path), exist_ok=True)
shutil.copy(audio_dl, audio_path)
except Exception as dl_err:
print(f"โ ๏ธ Could not recover {audio_filename}: {dl_err}")
except Exception as e:
print("โน๏ธ No remote pending queue found. Starting fresh.")
def _admin_action_log_path(self):
try:
os.makedirs(os.path.dirname(self.ADMIN_ACTION_LOG), exist_ok=True)
return self.ADMIN_ACTION_LOG
except Exception:
fallback = os.path.join(os.getcwd(), "admin_action_log.jsonl")
os.makedirs(os.path.dirname(fallback), exist_ok=True)
self.ADMIN_ACTION_LOG = fallback
return fallback
def _clean_cell(self, value):
if value is None:
return ""
try:
if pd.isna(value):
return ""
except Exception:
pass
return str(value)
def _row_to_action_dict(self, row):
return {str(k): self._clean_cell(v) for k, v in row.to_dict().items()}
def sync_admin_action_log(self, direction="up"):
hf_token = os.environ.get("HF_TOKEN")
repo_id = "toecm/PureChain_Dataset"
if not hf_token:
return
api = HfApi(token=hf_token)
log_path = self._admin_action_log_path()
if direction == "up":
if os.path.exists(log_path):
try:
api.upload_file(
path_or_fileobj=log_path,
path_in_repo="admin_action_log.jsonl",
repo_id=repo_id,
repo_type="dataset",
commit_message="Sync admin approval/rejection action log"
)
except Exception as e:
print(f"Failed to upload admin action log: {e}")
elif direction == "down":
try:
downloaded_path = hf_hub_download(
repo_id=repo_id,
filename="admin_action_log.jsonl",
repo_type="dataset",
token=hf_token
)
shutil.copy(downloaded_path, log_path)
print("Admin action log restored from PureChain_Dataset.")
except Exception:
pass
def _load_admin_actions(self):
log_path = self._admin_action_log_path()
if not os.path.exists(log_path):
return []
actions = []
with open(log_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
actions.append(json.loads(line))
except Exception:
continue
return actions
def _write_admin_actions(self, actions):
log_path = self._admin_action_log_path()
with open(log_path, "w", encoding="utf-8") as f:
for action in actions:
f.write(json.dumps(action, ensure_ascii=False) + "\n")
self.sync_admin_action_log(direction="up")
def _append_admin_action(self, action):
action.setdefault("action_id", uuid.uuid4().hex)
action.setdefault("created_at", datetime.now().isoformat())
action.setdefault("undone", False)
log_path = self._admin_action_log_path()
with open(log_path, "a", encoding="utf-8") as f:
f.write(json.dumps(action, ensure_ascii=False) + "\n")
self.sync_admin_action_log(direction="up")
return action["action_id"]
def _mark_admin_action_undone(self, action_id):
actions = self._load_admin_actions()
for action in reversed(actions):
if action.get("action_id") == action_id:
action["undone"] = True
action["undone_at"] = datetime.now().isoformat()
break
self._write_admin_actions(actions)
def _restore_pending_row(self, row_data):
row_data = {str(k): self._clean_cell(v) for k, v in dict(row_data or {}).items()}
if not row_data:
return False, "No saved pending row was found."
if os.path.exists(self.PENDING_FILE):
df = pd.read_csv(self.PENDING_FILE, dtype=str).fillna("")
else:
df = pd.DataFrame()
for col in row_data.keys():
if col not in df.columns:
df[col] = ""
for col in df.columns:
row_data.setdefault(col, "")
timestamp = row_data.get("Timestamp", "")
utterance = row_data.get("Utterance", "")
if not df.empty and "Timestamp" in df.columns and "Utterance" in df.columns:
duplicate = df[
(df["Timestamp"].astype(str) == str(timestamp)) &
(df["Utterance"].astype(str) == str(utterance))
]
if not duplicate.empty:
return True, "Entry was already back in pending."
df = pd.concat([df, pd.DataFrame([{c: row_data.get(c, "") for c in df.columns}])], ignore_index=True)
df.to_csv(self.PENDING_FILE, index=False)
self.sync_pending_queue(direction="up")
return True, "Entry restored to pending."
def _approval_csv_path(self, language, dialect):
dialect_str = "" if pd.isna(dialect) else str(dialect)
clean_dialect = dialect_str.strip().title()
if "Pidgin" in clean_dialect and not clean_dialect.endswith("English") and not clean_dialect.endswith("Dialect"):
clean_dialect += " English"
def infer_language(name):
name_lower = name.lower()
if "patois" in name_lower or "pidgin" in name_lower or "creole" in name_lower:
return "Creole"
if "english" in name_lower or "american" in name_lower or "british" in name_lower or "south african" in name_lower:
return "English"
if "korean" in name_lower:
return "Korean"
if "igbo" in name_lower:
return "Asusu Igbo"
if "tagalog" in name_lower or "filipino" in name_lower:
return "Filipino"
if "indonesian" in name_lower:
return "Indonesian"
return "Other"
if language is not None and not pd.isna(language) and str(language).strip():
lang = str(language).strip().title()
else:
lang = infer_language(clean_dialect)
return os.path.join(self.trust.config.DATASET_DIR, lang, "data", f"{clean_dialect}.csv"), clean_dialect
def _upload_file_to_hf(self, local_path, path_in_repo, repo_id, message):
hf_token = os.environ.get("HF_TOKEN")
if not hf_token or not os.path.exists(local_path):
return
try:
HfApi(token=hf_token).upload_file(
path_or_fileobj=local_path,
path_in_repo=path_in_repo,
repo_id=repo_id,
repo_type="dataset",
commit_message=message
)
except Exception as e:
print(f"HF rollback sync warning: {e}")
def _remove_approved_dataset_row(self, action):
import csv
approved = action.get("approved_row", {})
csv_path, clean_dialect = self._approval_csv_path(approved.get("Language", ""), approved.get("Dialect", ""))
if not os.path.exists(csv_path):
return False, "Approved CSV was not found locally."
df = pd.read_csv(csv_path, dtype=str, encoding="utf-8-sig", on_bad_lines="skip").fillna("")
if df.empty:
return False, "Approved CSV was empty."
def series(col):
if col not in df.columns:
return pd.Series([""] * len(df), index=df.index)
return df[col].astype(str).str.strip()
mask = (
(series("Utterance") == str(approved.get("Utterance", "")).strip()) &
(series("Dialect") == str(clean_dialect).strip()) &
(series("Clarification") == str(approved.get("Clarification", "")).strip()) &
(series("Data_Origin") == str(approved.get("Data_Origin", "")).strip()) &
(series("User") == str(approved.get("User", "")).strip())
)
matches = df[mask]
if matches.empty:
mask = (
(series("Utterance") == str(approved.get("Utterance", "")).strip()) &
(series("Dialect") == str(clean_dialect).strip()) &
(series("Clarification") == str(approved.get("Clarification", "")).strip())
)
matches = df[mask]
if matches.empty:
return False, "Could not find the matching approved CSV row."
df = df.drop(matches.index[-1])
df.to_csv(csv_path, index=False, encoding="utf-8-sig", quoting=csv.QUOTE_ALL)
relative_path = os.path.relpath(csv_path, self.trust.config.DATASET_DIR).replace(os.sep, "/")
self._upload_file_to_hf(
csv_path,
relative_path,
getattr(self.trust.config, "HF_REPO_ID", "toecm/IEDID"),
"Undo admin approval CSV row"
)
return True, "Approved CSV row removed."
def _remove_rejected_graveyard_row(self, action):
import csv
row = action.get("pending_row", {})
graveyard_path = os.path.join(self.trust.config.DATASET_DIR, "..", "rejected_graveyard.csv")
if not os.path.exists(graveyard_path):
return False, "Rejected graveyard CSV was not found locally."
df = pd.read_csv(graveyard_path, dtype=str, encoding="utf-8-sig", on_bad_lines="skip").fillna("")
if df.empty:
return False, "Rejected graveyard CSV was empty."
def series(col):
if col not in df.columns:
return pd.Series([""] * len(df), index=df.index)
return df[col].astype(str).str.strip()
mask = (
(series("Utterance") == str(row.get("Utterance", "")).strip()) &
(series("Dialect") == str(row.get("Dialect", "")).strip()) &
(series("Clarification") == str(row.get("Clarification", "")).strip()) &
(series("Data_Origin") == str(row.get("Data_Origin", "")).strip()) &
(series("User") == str(row.get("User", "")).strip()) &
(series("Rejection_Reason") == str(action.get("reject_reason", "")).strip())
)
matches = df[mask]
if matches.empty:
mask = (
(series("Utterance") == str(row.get("Utterance", "")).strip()) &
(series("Dialect") == str(row.get("Dialect", "")).strip()) &
(series("Rejection_Reason") == str(action.get("reject_reason", "")).strip())
)
matches = df[mask]
if matches.empty:
return False, "Could not find the matching rejected graveyard row."
df = df.drop(matches.index[-1])
df.to_csv(graveyard_path, index=False, encoding="utf-8-sig", quoting=csv.QUOTE_ALL)
self._upload_file_to_hf(
graveyard_path,
"rejected_graveyard.csv",
getattr(self.trust.config, "HF_REPO_ID", "toecm/IEDID"),
"Undo admin rejection graveyard row"
)
return True, "Rejected graveyard row removed."
def _restore_pending_audio(self, action):
row = action.get("pending_row", {})
audio_path = str(row.get("Audio", "") or "")
if not audio_path or audio_path == "nan":
return "No pending audio path to restore."
audio_name = os.path.basename(audio_path)
if not audio_name:
return "No pending audio filename to restore."
if not os.path.exists(audio_path):
hf_token = os.environ.get("HF_TOKEN")
if hf_token:
candidates = []
if action.get("action_type") == "reject":
candidates.append(("toecm/PureChain_Dataset", f"rejected_audio/{audio_name}"))
candidates.append((getattr(self.trust.config, "HF_REPO_ID", "toecm/IEDID"), f"rejected_audio/{audio_name}"))
candidates.append((getattr(self.trust.config, "HF_REPO_ID", "toecm/IEDID"), f"audio/{audio_name}"))
for repo_id, filename in candidates:
try:
downloaded = hf_hub_download(repo_id=repo_id, filename=filename, repo_type="dataset", token=hf_token)
os.makedirs(os.path.dirname(audio_path), exist_ok=True)
shutil.copy(downloaded, audio_path)
break
except Exception:
continue
if os.path.exists(audio_path):
self._upload_file_to_hf(
audio_path,
f"pending_audio/{audio_name}",
"toecm/PureChain_Dataset",
"Restore pending audio after admin undo"
)
return "Pending audio restored."
return "Pending audio could not be restored automatically."
def admin_undo_last_action(self, review_key=""):
if not self._is_full_admin_authorized(review_key):
return self._full_admin_required_message()
try:
actions = self._load_admin_actions()
action = next((a for a in reversed(actions) if not a.get("undone") and a.get("action_type") in ["approve", "reject"]), None)
if not action:
return "No approval/rejection action is available to undo."
status_parts = []
restored, msg = self._restore_pending_row(action.get("pending_row", {}))
status_parts.append(msg)
if action.get("action_type") == "approve":
ok, msg = self._remove_approved_dataset_row(action)
status_parts.append(msg)
elif action.get("action_type") == "reject":
ok, msg = self._remove_rejected_graveyard_row(action)
status_parts.append(msg)
status_parts.append(self._restore_pending_audio(action))
self._mark_admin_action_undone(action.get("action_id"))
undone_label = "approval" if action.get("action_type") == "approve" else "rejection"
return f"Undo complete for last {undone_label}: " + " ".join(status_parts)
except Exception as e:
return f"Undo failed: {e}"
def get_pending_label(self):
if os.path.exists(self.PENDING_FILE):
count = len(pd.read_csv(self.PENDING_FILE))
if count > 0:
return f"๐ฎ Pending ({count})", count
return "๐ฎ Pending", 0
def monitor_pending_state(self):
label, count = self.get_pending_label()
if count > self.last_pending_count and self.alert_sound:
sound = gr.update(value=self.alert_sound, autoplay=True)
else:
sound = gr.skip()
self.last_pending_count = count
return f"### {label} - Review Submissions from React Games", sound
def admin_approve_pending(self, timestamp, orig_utt, edited_utt, edited_lang, edited_dialect, edited_clar, edited_tone, trimmed_audio_path, review_key=""):
scope = self._review_scope(review_key)
if not scope:
return self._review_unauthorized_message()
try:
df = pd.read_csv(self.PENDING_FILE)
df = self._infer_pending_languages(df)
match = df[(df["Timestamp"] == timestamp) & (df["Utterance"] == orig_utt)]
if len(match) == 0:
return "โ Approval failed: Entry not found in pending database."
index_in_csv = match.index[0]
row = df.loc[index_in_csv]
if not self._pending_row_allowed_by_scope(row, scope):
return "๐ Scoped reviewer passcode cannot approve this language/dialect."
# ๐ข Use the edited text instead of the original row data
final_utt = edited_utt if edited_utt else row["Utterance"]
final_lang = edited_lang if edited_lang else row.get("Language", "")
final_dialect = edited_dialect if edited_dialect else row["Dialect"]
final_clar = edited_clar if edited_clar else row["Clarification"]
final_tone = edited_tone if edited_tone else row["Tone"]
if not self._scope_allows_language_dialect(scope, final_lang, final_dialect):
return "๐ Scoped reviewer passcode cannot approve edits outside its language/dialect scope."
final_audio = trimmed_audio_path if isinstance(trimmed_audio_path, str) and os.path.exists(trimmed_audio_path) else row.get("Audio", "")
review_metadata = {
column: self._clean_cell(row.get(column, ""))
for column in [
"Queue_ID", "Interaction_ID", "Supersedes_Interaction_ID",
"App_Source", "Source_Input_Mode", "Machine_Transcript_Initial",
"User_Transcript_Final", "Transcript_Edit_Distance",
"ASR_Model", "Audio_Sanitation", "Audio_Retained",
"Machine_Translation_Initial",
"User_Translation_Final", "Translation_Edit_Distance",
"AI_Model", "Translation_Route", "Consent_Confirmed",
"Consent_Version", "Source_Language", "Source_Dialect",
"Target_Language", "Target_Dialect", "Review_Submitted_At"
]
}
review_metadata["Review_Status"] = "Approved"
self.trust.update_dataset_csv(
final_dialect, final_utt, final_clar, final_tone,
row.get("Context", ""), "", final_audio, row.get("Pragmatic_Analysis", ""),
row.get("Data_Origin", ""), "Admin Edit", row.get("User", ""),
language=final_lang, review_metadata=review_metadata
)
self._append_admin_action({
"action_type": "approve",
"pending_row": self._row_to_action_dict(row),
"approved_row": {
"Utterance": self._clean_cell(final_utt),
"Dialect": self._clean_cell(final_dialect),
"Language": self._clean_cell(final_lang),
"Clarification": self._clean_cell(final_clar),
"Tone_Category": self._clean_cell(final_tone),
"Linguistic_Context": self._clean_cell(row.get("Context", "")),
"Pragmatic_Analysis": self._clean_cell(row.get("Pragmatic_Analysis", "")),
"Data_Origin": self._clean_cell(row.get("Data_Origin", "")),
"Clarification_Source": "Admin Edit",
"User": self._clean_cell(row.get("User", "")),
"Audio": self._clean_cell(final_audio)
}
})
payload = {
"original": final_utt,
"dialect": final_dialect,
"clarification": final_clar,
"tone": final_tone,
"user": str(row.get("User", "")),
"Data_Origin": str(row.get("Data_Origin", ""))
}
threading.Thread(target=self.trust.stamp_on_chain, args=(payload,), daemon=True).start()
df.drop(index_in_csv).to_csv(self.PENDING_FILE, index=False)
self.sync_pending_queue(direction="up")
audio_to_delete = row.get("Audio")
if audio_to_delete and str(audio_to_delete) != "nan":
try:
api = HfApi(token=os.environ.get("HF_TOKEN"))
api.delete_file(
path_in_repo=f"pending_audio/{os.path.basename(audio_to_delete)}",
repo_id="toecm/PureChain_Dataset",
repo_type="dataset",
commit_message="๐๏ธ Cleaned up processed pending audio"
)
except Exception as e:
pass
return f"โ
Approved & Minted: {final_utt[:20]}..."
except Exception as e:
return f"โ Approval failed: {e}"
def admin_reject_pending(self, timestamp, orig_utt, reject_reason, custom_reason, review_key=""):
scope = self._review_scope(review_key)
if not scope:
return self._review_unauthorized_message()
try:
df = pd.read_csv(self.PENDING_FILE)
df = self._infer_pending_languages(df)
match = df[(df["Timestamp"] == timestamp) & (df["Utterance"] == orig_utt)]
if len(match) == 0:
return "โ Rejection failed: Entry not found."
index_in_csv = match.index[0]
row = df.loc[index_in_csv]
if not self._pending_row_allowed_by_scope(row, scope):
return "๐ Scoped reviewer passcode cannot reject this language/dialect."
final_reason = custom_reason if reject_reason == "Other" else reject_reason
if not final_reason:
final_reason = "Admin Override"
# If it has a Chain_ID, it was proposed. Reject it on-chain!
chain_id = str(row.get("Chain_ID", ""))
# Log to Graveyard
self.trust.add_to_graveyard(
dialect=row.get("Dialect", ""),
utterance=row.get("Utterance", ""),
clarification=row.get("Clarification", ""),
tone=row.get("Tone", ""),
context=row.get("Context", ""),
syntax="",
audio_path=row.get("Audio", ""),
pragmatics=row.get("Pragmatic_Analysis", ""),
sourceTag=row.get("Data_Origin", "Pending"),
clar_source="Lab Admin",
userKey=row.get("User", ""),
reason=final_reason,
rejector="Admin",
chain_id=chain_id
)
self._append_admin_action({
"action_type": "reject",
"pending_row": self._row_to_action_dict(row),
"reject_reason": self._clean_cell(final_reason),
"rejector": "Admin"
})
if chain_id.isdigit():
try:
w3 = self.trust.w3
contract_addr = os.environ.get("PURECHAIN_CONTRACT_ADDRESS", getattr(self.trust.config, 'PURECHAIN_CONTRACT_ADDRESS', None))
contract = w3.eth.contract(address=contract_addr, abi=self.trust.config.CONTRACT_ABI)
nonce = w3.eth.get_transaction_count(self.trust.account.address)
tx = contract.functions.rejectEntry(int(chain_id), self.trust.account.address, final_reason).build_transaction({
'chainId': getattr(self.trust.config, 'PURECHAIN_ID', 900520900520),
'gas': 2000000, 'gasPrice': 0, 'nonce': nonce
})
signed_tx = w3.eth.account.sign_transaction(tx, self.trust.config.PRIVATE_KEY)
tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction)
w3.eth.wait_for_transaction_receipt(tx_hash)
print(f"๐ฅ On-chain burn applied for {chain_id} with reason: {final_reason}")
except Exception as e:
print(f"Failed to burn on-chain: {e}")
df.drop(index_in_csv).to_csv(self.PENDING_FILE, index=False)
self.sync_pending_queue(direction="up")
audio_to_delete = row.get("Audio")
if audio_to_delete and str(audio_to_delete) != "nan":
try:
api = HfApi(token=os.environ.get("HF_TOKEN"))
if os.path.exists(audio_to_delete):
api.upload_file(
path_or_fileobj=audio_to_delete, # Local file
path_in_repo=f"rejected_audio/{os.path.basename(audio_to_delete)}",
repo_id="toecm/PureChain_Dataset",
repo_type="dataset",
commit_message=f"๐ซ Moved rejected audio: {final_reason}"
)
api.delete_file(
path_in_repo=f"pending_audio/{os.path.basename(audio_to_delete)}",
repo_id="toecm/PureChain_Dataset",
repo_type="dataset",
commit_message="๐๏ธ Cleaned up rejected pending audio"
)
except Exception as e:
print(f"HF Audio Move Error: {e}")
return f"๐๏ธ Entry Rejected ({final_reason}) & Moved to Burn Ward."
except Exception as e:
return f"โ Rejection Error: {e}"
def admin_clear_all_pending(self, review_key=""):
if not self._is_full_admin_authorized(review_key):
return self._full_admin_required_message()
try:
if os.path.exists(self.PENDING_FILE):
os.remove(self.PENDING_FILE)
self.sync_pending_queue(direction="up")
return "๐งน All pending entries swept!"
except Exception as e: return f"โ Clear failed: {e}"
def export_analysis_to_csv(self, df, review_key=""):
if not self._is_full_admin_authorized(review_key):
return None
if df is None or not hasattr(df, 'columns') or df.empty:
return None
path = "/app/sociolinguistic_export.csv"
df.to_csv(path, index=False, encoding='utf-8-sig')
return path
def api_generate_training_data(self, review_key=""):
if not self._is_full_admin_authorized(review_key):
return self._full_admin_required_message()
try:
prepare_training_data.main()
return f"โ
Success! 'train.csv' created."
except Exception as e:
return f"โ Error generating data: {e}"
def api_get_full_dataset_zip(self, review_key=""):
if not self._is_full_admin_authorized(review_key):
return None
try:
shutil.make_archive("/app/dataset_export", 'zip', self.brain.config.DATASET_DIR)
return "/app/dataset_export.zip"
except Exception as e:
return f"Error zipping: {e}"
def auto_regenerate_analysis(self, text, clar, tone, ctx, prag, new_dialect):
show_new = (new_dialect == "+ Add New Dialect")
# Guard: don't fire heavy AI calls if there is no utterance text (e.g. tab-switching with empty field)
if not text or not text.strip() or show_new or not new_dialect:
return clar, tone, ctx, prag, gr.update(visible=show_new)
print(f"๐ UI Trigger: Re-analyzing '{text}' for dialect: {new_dialect}")
try:
if hasattr(self.brain, 'analyze_dialect_single'):
res = self.brain.analyze_dialect_single(text, new_dialect)
return (
res.get("clarification", clar),
res.get("tone", tone),
res.get("context", ctx),
res.get("pragmatics", prag),
gr.update(visible=show_new)
)
except Exception as e:
print(f"Auto-regenerate error: {e}")
return clar, tone, ctx, prag, gr.update(visible=show_new)
def lab_analyze_and_mint(self, text, dialect, force_ai, userKey, review_key="", request: gr.Request = None):
if not self._is_full_admin_authorized(review_key):
return {"error": "unauthorized"}, self._full_admin_required_message()
status_log = f"๐ LAB PIPELINE TRIGGERED for '{text}'\n"
final_result = None
if not text or not dialect:
return {"error": "Missing input"}, "โ ๏ธ Please provide text and select a dialect."
if not force_ai:
status_log += "๐๏ธ Checking Local Dataset...\n"
final_result = self.brain.search_local_dataset(text)
if final_result:
status_log += "โ
Found in Local Dataset.\n"
else:
status_log += "๐ญ Checking Persona Context...\n"
final_result = self.brain.search_personas(text)
else:
status_log += "๐ Force AI Enabled: Bypassing local lookups...\n"
if not final_result:
status_log += "๐ง Generating AI interpretation...\n"
if hasattr(self.brain, 'analyze_dialect_single'):
final_result = self.brain.analyze_dialect_single(text, dialect)
status_log += "โ
AI Engine Analysis Complete.\n"
else:
return {"error": "Missing AI function"}, status_log + "โ Error."
clarification = final_result.get("clarification", "")
tone = final_result.get("tone", "Neutral")
context = final_result.get("context", "")
pragmatics = final_result.get("pragmatics", "")
client_ip = request.headers.get("x-forwarded-for") or request.client.host if request else "Unknown_IP"
source_tag = f"Lab_Admin_{client_ip}"
success = self.trust.update_dataset_csv(
dialect=dialect, utterance=text, clarification=clarification,
tone=tone, context=context, syntax="", audio_path=None,
pragmatics=pragmatics, sourceTag=source_tag,
clar_source=final_result.get("Source", "AI"), userKey=userKey
)
if success:
status_log += "\n๐ SUCCESS: Entry saved to CSV and synced to HF Cloud!"
payload = {
"original": text,
"dialect": dialect,
"clarification": clarification,
"tone": tone,
"user": userKey
}
threading.Thread(target=self.trust.stamp_on_chain, args=(payload,), daemon=True).start()
status_log += "\nโ๏ธ PureChain minting triggered in background."
else:
status_log += "\nโ ๏ธ ERROR: Database save failed."
return final_result, status_log
# ==========================================
# THE RESEARCH UI
# ==========================================
def create_ui(self):
def generate_admin_op_id():
return "0x" + uuid.uuid4().hex + uuid.uuid4().hex[:8]
custom_css = """
.gradio-container { max-width: 95% !important; }
table { width: 100% !important; table-layout: auto !important; }
td { white-space: normal !important; word-wrap: break-word !important; }
"""
import json
hierarchy_str = self.api_get_language_hierarchy()
language_hierarchy_dict = json.loads(hierarchy_str)
language_choices = ["All"] + sorted(list(language_hierarchy_dict.keys()))
all_dialects = []
for d_list in language_hierarchy_dict.values():
all_dialects.extend([d for d in d_list if d != "+ Add New Dialect"])
dropdown_choices = sorted(list(set(all_dialects)))
available_profiles = self.brain.get_available_profiles() if hasattr(self.brain, 'get_available_profiles') else []
backup_files = []
if hasattr(self.brain, 'config'):
if os.path.exists(self.brain.config.DATASET_DIR):
backup_files.extend([os.path.basename(f) for f in glob.glob(os.path.join(self.brain.config.DATASET_DIR, "**", "*.csv"), recursive=True)])
if os.path.exists(self.brain.config.PROFILES_DIR):
backup_files.extend([os.path.basename(f) for f in glob.glob(os.path.join(self.brain.config.PROFILES_DIR, "*.json"))])
backup_files = sorted(list(set(backup_files))) if backup_files else ["No files found"]
with gr.Blocks(theme=gr.themes.Default(primary_hue=gr.themes.colors.red, secondary_hue=gr.themes.colors.orange)) as ui:
gr.Markdown("## ๐ PurePolyglot: Decentralized Multi-Dialect Mediator (Lab View)")
ui_source_tag = gr.Textbox(visible=False, value="Web")
ui_clar_source = gr.Textbox(visible=False, value="Lab Admin")
ui_operator_id = gr.Textbox(visible=False, value=generate_admin_op_id)
api_audio_blob = gr.Audio(visible=False, type="filepath")
api_confirm = gr.State(False)
admin_review_key = gr.Textbox(
label="Reviewer/Admin passcode",
type="password",
placeholder="Required for backend dashboard tabs and privileged actions"
)
with gr.Tabs():
with gr.Tab("๐๏ธ Live Field Analysis"):
health_display = gr.HTML(self.get_blockchain_health())
with gr.Row():
with gr.Column(scale=1):
audio_in = gr.Audio(label="Step 1: Speak/Upload", sources=["microphone", "upload"], type="filepath")
lang_sel = gr.Dropdown(["en", "yo", "ig", "ko", "ha"], value="en", label="Language Context")
btn_run = gr.Button("Analyze Audio ๐", variant="primary")
quota_box = gr.Textbox(label="๐ API Quota", value=self.get_quota_status(), interactive=False)
with gr.Row(variant="compact"):
background_status_display = gr.Textbox(label="Status", value="Checking...", interactive=False, show_label=False)
with gr.Column(scale=5):
log_box = gr.Textbox(label="Linguistic Analysis Log", interactive=False)
gr.Markdown("### ๐ฅ AI Interpretation Baseline")
results_table = gr.Dataframe(
headers=["Source", "Speaker", "Utterance", "Dialect", "Clarification", "Tone", "Context", "Pragmatic Analysis"],
interactive=True,
wrap=False,
row_count=(1, "dynamic")
)
with gr.Row():
export_btn = gr.Button("๐ฅ Download Analysis CSV", variant="secondary")
export_file = gr.File(label="Export Result", visible=False)
gr.Markdown("### โ๏ธ Active Sociolinguistic Feedback Loop (Edit & Approve)")
with gr.Row():
with gr.Column(scale=1):
orig_text = gr.Textbox(visible=True, label="Utterance (Transcribed)")
dialect_sel = gr.Dropdown(choices=dropdown_choices, label="Assigned Dialect", interactive=True, allow_custom_value=True)
new_dialect = gr.Textbox(label="Enter New Dialect Name", visible=False, interactive=True)
with gr.Column():
clar_text = gr.Textbox(label="Final Clarification", interactive=True, lines=2)
tone_sel = gr.Dropdown(choices=self.TONES, value="Neutral / Conversational", label="Pragmatic Tone", interactive=True, allow_custom_value=True)
ctx_area = gr.TextArea(label="Linguistic Context", interactive=True, lines=1)
prag_area = gr.TextArea(label="Pragmatic Analysis ([Force], [Deixis], [Register])", interactive=True, lines=1)
with gr.Row():
btn_save = gr.Button("๐พ Validate & Save", variant="primary")
btn_over = gr.Button("โ ๏ธ Confirm Overwrite", variant="stop", visible=False)
feedback_msg = gr.Markdown()
gr.Markdown("### ๐ฅ PhD Data Export & Training")
with gr.Row():
export_data_btn = gr.Button("๐ฆ Generate Full Dataset ZIP", variant="secondary")
train_btn = gr.Button("๐ง Generate AutoTrain CSV", variant="primary")
export_zip_file = gr.File(label="Download")
train_status = gr.Textbox(label="Training Data Status", lines=1)
with gr.Tab("๐งช THE LAB (Force AI)"):
gr.Markdown("### ๐ฌ Test text inputs directly and force AI generation")
with gr.Row():
with gr.Column():
lab_input = gr.Textbox(label="Test Phrase (Text Only)")
lab_dialect = gr.Dropdown(choices=dropdown_choices, label="Target Dialect")
force_ai_toggle = gr.Checkbox(label="Force Live AI (Skip Local Cache)", value=False)
lab_user_key = gr.Textbox(label="Admin User Key", value="Admin_001")
lab_btn = gr.Button("RUN ANALYSIS & MINT", variant="primary")
with gr.Column():
lab_output = gr.JSON(label="Analysis Result")
lab_log = gr.Textbox(label="System Logs", lines=10)
lab_btn.click(
fn=self.lab_analyze_and_mint,
inputs=[lab_input, lab_dialect, force_ai_toggle, lab_user_key, admin_review_key],
outputs=[lab_output, lab_log]
)
with gr.Tab("โ๏ธ Persona Management"):
with gr.Row():
load_all_btn = gr.Button("๐ Load ALL Profiles", variant="secondary")
profile_selector = gr.Dropdown(choices=available_profiles, label="Select Profile", allow_custom_value=True)
profile_filename = gr.Textbox(label="Filename")
load_profile_btn = gr.Button("๐ฅ Load Selected Profile", variant="primary") # ๐ข NEW
profile_editor = gr.Textbox(label="Profile Content (JSON)", lines=20) # ๐ข CHANGED TO TEXTBOX
save_profile_btn = gr.Button("๐พ Save Profile modifications", variant="primary")
profile_status = gr.Textbox(label="System Response", interactive=False)
def change_profile(val, review_key):
if not self._is_full_admin_authorized(review_key):
return "", ""
if not val: return "", ""
return json.dumps(self.brain.load_profile_by_name(val), indent=2), val
def save_and_refresh_profile(filename, content, review_key):
if not self._is_full_admin_authorized(review_key):
return self._full_admin_required_message(), gr.update()
msg = self.brain.save_specific_profile(filename, content)
return msg, gr.update(choices=self.brain.get_available_profiles(), value=filename)
def load_all_profiles_with_key(review_key):
if not self._is_full_admin_authorized(review_key):
return self._full_admin_required_message(), ""
self.brain.load_all_profiles_simultaneously()
return "Profiles loaded.", self.brain.get_current_profile_text()
# ๐ข Changed to click instead of auto-change
load_profile_btn.click(change_profile, inputs=[profile_selector, admin_review_key], outputs=[profile_editor, profile_filename])
save_profile_btn.click(save_and_refresh_profile, inputs=[profile_filename, profile_editor, admin_review_key], outputs=[profile_status, profile_selector])
if hasattr(self.brain, 'load_all_profiles_simultaneously'):
load_all_btn.click(load_all_profiles_with_key, inputs=[admin_review_key], outputs=[profile_status, profile_editor])
with gr.Tab("๐ฎ Pending Audit/Approvals"):
pending_header = gr.Markdown("### ๐ฎ Pending (0) - Review Submissions from React Games")
gr.Markdown("Use the global reviewer/admin passcode field above before refreshing or changing audit data.")
with gr.Row():
with gr.Column(scale=3):
with gr.Row():
filter_language = gr.Dropdown(choices=language_choices, value="All", label="Filter by Language")
filter_dialect = gr.Dropdown(choices=["All"] + dropdown_choices, value="All", label="Filter by Dialect")
pending_df = gr.Dataframe(headers=[
"User", "Data_Origin", "Utterance", "Language", "Dialect",
"Clarification", "Clarification_Source", "Tone", "Audio", "Timestamp",
"App_Source", "Source_Language", "Source_Dialect", "Source_Input_Mode",
"Machine_Transcript_Initial", "User_Transcript_Final",
"Transcript_Edit_Distance", "Machine_Translation_Initial",
"User_Translation_Final", "Translation_Edit_Distance", "ASR_Model",
"AI_Model", "Audio_Sanitation", "Consent_Confirmed", "Queue_ID"
], interactive=False, wrap=False, row_count=(1, "dynamic"))
with gr.Column(scale=1):
gr.Markdown("#### ๐ง Audio Auditor")
btn_refresh = gr.Button("๐ Refresh List")
pending_audio_player = gr.Audio(label="Trim or Preview Audio", type="filepath", interactive=True)
audit_log = gr.Textbox(label="Audit Status", interactive=False)
# ๐ข NEW: Editable Textboxes for Admin Corrections
gr.Markdown("#### โ๏ธ Edit Selected Entry Before Minting")
with gr.Row():
pending_timestamp = gr.Textbox(label="Timestamp ID", interactive=False)
pending_orig_utt = gr.Textbox(visible=False)
edit_utt = gr.Textbox(label="Utterance", interactive=True)
edit_lang = gr.Textbox(label="Language", interactive=True)
edit_dialect = gr.Textbox(label="Dialect", interactive=True)
edit_clar = gr.Textbox(label="Clarification / Meaning", interactive=True)
edit_tone = gr.Textbox(label="Tone", interactive=True)
with gr.Row():
reject_reason_dropdown = gr.Dropdown(["Spam", "Audio Quality", "Dialect Mismatch", "Other"], label="Rejection Reason", value="Audio Quality")
reject_reason_text = gr.Textbox(label="Custom Reason (if Other)", visible=False)
def toggle_reason(val):
return gr.update(visible=(val == "Other"))
reject_reason_dropdown.change(toggle_reason, inputs=[reject_reason_dropdown], outputs=[reject_reason_text])
with gr.Row():
btn_appr_p = gr.Button("โ
Approve & Mint (With Edits)", variant="primary")
btn_rejt_p = gr.Button("๐๏ธ Reject entry", variant="stop")
btn_undo_admin = gr.Button("Undo Last Approval/Rejection", variant="secondary")
btn_clear_pending = gr.Button("Sweep All Pending", variant="secondary")
with gr.Tab("โ๏ธ PureChain History"):
gr.Markdown("### ๐ Immutable Transaction Log & Audit Reports")
with gr.Row():
legacy_checkboxes = gr.CheckboxGroup(
choices=["PureIUUY", "PureConvo", "PureVersation", "PureBi (Jun 11)", "PureBi (Jun 12)", "PureBi (Jun 13)", "PureBi"],
label="Fetch Legacy Blockchain Data",
value=["PureBi"]
)
with gr.Row():
start_date = gr.DateTime(label="Start Date", type="string")
end_date = gr.DateTime(label="End Date", type="string")
btn_filter = gr.Button("๐ Filter & Refresh", variant="primary")
gr.HTML("๐ Open PureChain Explorer")
history_df = gr.Dataframe(
headers=["Timestamp", "Utterance", "Dialect", "Data_Origin", "Data_Approved", "Block", "TX Hash"],
interactive=False,
wrap=False,
row_count=(5, "dynamic")
)
with gr.Row():
export_report_btn = gr.Button("๐ฅ Generate CSV Report", variant="secondary")
report_file = gr.File(label="Download Audit Report")
explorer_link = gr.Markdown("Select a row to generate Explorer Link")
def run_filter(s, e, selected_contracts, review_key):
if not self._is_full_admin_authorized(review_key):
return self._review_empty_history_dataframe()
if selected_contracts:
try:
import recover_chain
recover_chain.main(selected_contracts)
except Exception as exc:
print(f"Recovery failed: {exc}")
df = self.trust.get_filtered_history(s, e)
display_cols = ["Timestamp", "Utterance", "Dialect", "Data_Origin", "Data_Approved", "Block", "TX Hash"]
available = [c for c in display_cols if c in df.columns]
# Truncate TX Hash for UI only
if "TX Hash" in available:
df["TX Hash"] = df["TX Hash"].apply(lambda x: str(x)[:15] + "..." if len(str(x)) > 15 else x)
return df[available]
def generate_report(s, e, review_key):
if not self._is_full_admin_authorized(review_key):
return None
df = self.trust.get_filtered_history(s, e)
report_path = "/app/purechain_audit_report.csv"
export_cols = ["Timestamp", "Utterance", "Dialect", "Clarification", "Data_Origin", "Data_Approved", "Block", "TX Hash"]
available_cols = [c for c in export_cols if c in df.columns]
df[available_cols].to_csv(report_path, index=False, encoding='utf-8-sig')
return report_path
def make_explorer_link(evt: gr.SelectData, df):
try:
tx_hash = df.iloc[evt.index[0]]["TX Hash"]
return f"๐ **[View Transaction on Explorer](http://3.34.161.207:3000/tx/{tx_hash})**"
except: return "Select a valid row"
btn_filter.click(run_filter, [start_date, end_date, legacy_checkboxes, admin_review_key], [history_df])
export_report_btn.click(generate_report, inputs=[start_date, end_date, admin_review_key], outputs=[report_file]).then(run_filter, [start_date, end_date, legacy_checkboxes, admin_review_key], [history_df])
history_df.select(make_explorer_link, [history_df], [explorer_link])
with gr.Tab("๐พ System Backups"):
with gr.Row():
backup_target = gr.Dropdown(choices=backup_files, label="Select File")
backup_desc = gr.Textbox(label="Backup Note", value="Routine check")
backup_btn = gr.Button("๐ Create Immutable Backup", variant="primary")
recover_btn = gr.Button("๐ Recover Data from Blockchain", variant="secondary")
backup_log = gr.Textbox(label="Backup Status", interactive=False)
gr.Markdown("---")
with gr.Row():
bytecode_input = gr.Textbox(label="Paste Contract Bytecode", lines=3)
deploy_btn = gr.Button("๐ Force Deploy (Zero Gas)", variant="stop")
deployment_output = gr.Textbox(label="Deployment Result", interactive=False)
def force_deploy_with_key(bytecode, review_key):
if not self._is_full_admin_authorized(review_key):
return self._full_admin_required_message()
return self.trust.force_deploy_contract(bytecode)
deploy_btn.click(
fn=force_deploy_with_key,
inputs=[bytecode_input, admin_review_key],
outputs=[deployment_output]
)
# --- NEW TAB: REJECTED ENTRIES AUDIT ---
with gr.Tab("๐ซ Rejected Entries Audit"):
gr.Markdown("### ๐ชฆ The Burn Ward (Rejected & Burned Data)")
with gr.Row():
btn_refresh_graveyard = gr.Button("๐ Refresh Graveyard", variant="primary")
graveyard_df = gr.Dataframe(
headers=["Timestamp", "Utterance", "Dialect", "Rejection_Reason", "Rejector"],
interactive=False,
wrap=False,
row_count=(5, "dynamic")
)
def refresh_graveyard(review_key):
import os
import pandas as pd
if not self._is_full_admin_authorized(review_key):
return pd.DataFrame(columns=["Timestamp", "Utterance", "Dialect", "Rejection_Reason", "Rejector"])
path = os.path.join(self.trust.config.DATASET_DIR, "..", "rejected_graveyard.csv")
if os.path.exists(path):
df = pd.read_csv(path)
cols = ["Timestamp", "Utterance", "Dialect", "Rejection_Reason", "Rejector"]
return df[[c for c in cols if c in df.columns]]
return pd.DataFrame()
btn_refresh_graveyard.click(refresh_graveyard, inputs=[admin_review_key], outputs=[graveyard_df])
# --- TAB 6: BUG REPORTS & FEEDBACK ---
with gr.Tab("๐ Bug Reports & Feedback"):
gr.Markdown("### ๐ก๏ธ Secure System Feedback Log")
with gr.Row():
with gr.Column(scale=3):
btn_refresh_fb = gr.Button("๐ Refresh Feedback List", variant="secondary")
feedback_df = gr.Dataframe(
headers=["Timestamp", "Operator_ID", "Feedback_Text", "Image_Reference"],
interactive=False,
wrap=False,
row_count=(5, "dynamic")
)
with gr.Column(scale=1):
gr.Markdown("#### ๐ธ Attached Screenshot")
feedback_image = gr.Image(label="Click a row to view screenshot", interactive=False)
# --- TAB 7: PASSIVE EEQ-S DASHBOARD ---
with gr.Tab("๐ EEQ-s Dashboard"):
gr.Markdown("### Passive EEQ-s Interaction Index")
gr.Markdown("Aggregated from frontend passive/manual EEQ-s auto-sync and manual sync events.")
with gr.Row():
btn_refresh_eeqs = gr.Button("๐ Refresh EEQ-s Dashboard", variant="secondary")
btn_download_eeqs = gr.Button("โฌ๏ธ Prepare EEQ-s Audit ZIP", variant="secondary")
eeqs_df = gr.Dataframe(
headers=[
"Game", "AI_Model", "ASR_Model", "Source_Language", "Source_Dialect",
"Target_Language", "Target_Dialect", "Interaction_Type", "Event_Count",
"Mean_EEQ", "Last_EEQ", "Last_Seen"
],
interactive=False,
wrap=False,
row_count=(8, "dynamic")
)
eeqs_download_file = gr.File(label="Download passive/manual EEQ-s audit ZIP")
# ==========================================
# EVENT BINDINGS
# ==========================================
export_data_btn.click(self.api_get_full_dataset_zip, inputs=[admin_review_key], outputs=[export_zip_file])
train_btn.click(self.api_generate_training_data, inputs=[admin_review_key], outputs=[train_status])
def admin_automated_pipeline(audio_path, lang, review_key):
if not self._is_full_admin_authorized(review_key):
return (
pd.DataFrame(),
self._review_unauthorized_message(),
self.get_quota_status(),
"",
None,
"",
"Neutral / Conversational",
"",
""
)
return self.automated_pipeline(audio_path, lang, None)
btn_run.click(
admin_automated_pipeline,
[audio_in, lang_sel, admin_review_key],
[results_table, log_box, quota_box, orig_text, dialect_sel, clar_text, tone_sel, ctx_area, prag_area]
)
audio_in.stop_recording(
admin_automated_pipeline,
[audio_in, lang_sel, admin_review_key],
[results_table, log_box, quota_box, orig_text, dialect_sel, clar_text, tone_sel, ctx_area, prag_area]
)
audio_in.upload(
admin_automated_pipeline,
[audio_in, lang_sel, admin_review_key],
[results_table, log_box, quota_box, orig_text, dialect_sel, clar_text, tone_sel, ctx_area, prag_area]
)
def handle_selection(evt: gr.SelectData, df):
if df is None or not hasattr(df, 'columns') or len(df) == 0:
return "", "", "", "Neutral / Conversational", "", ""
try:
row = df.iloc[evt.index[0]]
d = row["Dialect"] if row["Dialect"] in dropdown_choices else None
return row["Utterance"], d, row["Clarification"], row["Tone"], row.get("Context", ""), row.get("Pragmatic Analysis", "")
except: return "", "", "", "Neutral / Conversational", "", ""
results_table.select(handle_selection, [results_table], [orig_text, dialect_sel, clar_text, tone_sel, ctx_area, prag_area])
export_btn.click(self.export_analysis_to_csv, [results_table, admin_review_key], [export_file]).then(lambda: gr.update(visible=True), None, [export_file])
btn_save.click(
fn=self.admin_check_and_submit_logic,
inputs=[
orig_text, dialect_sel, new_dialect, clar_text, tone_sel, ctx_area, prag_area,
ui_source_tag, # 8. sourceTag ("Web")
ui_clar_source, # 9. clar_source ("Lab Admin")
ui_operator_id, # 10. userKey (Generated ID)
audio_in, # 11. blob
admin_review_key
],
outputs=[feedback_msg, btn_over]
)
btn_over.click(
fn=self.admin_force_overwrite_logic,
inputs=[
orig_text, dialect_sel, new_dialect, clar_text, tone_sel, ctx_area, prag_area,
ui_source_tag, # 8. sourceTag
ui_clar_source, # 9. clar_source
ui_operator_id, # 10. userKey
audio_in, # 11. blob
admin_review_key
],
outputs=[feedback_msg, btn_over]
)
# Audit / Pending
def select_pending_row(evt: gr.SelectData, df):
try:
idx = evt.index[0]
row = df.iloc[idx]
audio_path = row.get("Audio")
timestamp = str(row.get("Timestamp", ""))
utt = str(row.get("Utterance", ""))
clar = str(row.get("Clarification", ""))
tone = str(row.get("Tone", ""))
dialect = str(row.get("Dialect", ""))
lang = str(row.get("Language", ""))
return audio_path, timestamp, utt, utt, lang, dialect, clar, tone
except:
return None, "", "", "", "", "", "", ""
def update_dialects(lang):
if lang == "All":
return gr.update(choices=["All"] + dropdown_choices, value="All")
elif lang in language_hierarchy_dict:
new_choices = [d for d in language_hierarchy_dict[lang] if d != "+ Add New Dialect"]
return gr.update(choices=["All"] + new_choices, value="All")
return gr.update(choices=["All"], value="All")
filter_language.change(update_dialects, inputs=[filter_language], outputs=[filter_dialect]).then(
self.get_pending_dataframe, inputs=[filter_language, filter_dialect, admin_review_key], outputs=[pending_df]
)
filter_dialect.change(self.get_pending_dataframe, inputs=[filter_language, filter_dialect, admin_review_key], outputs=[pending_df])
btn_refresh.click(self.get_pending_dataframe, inputs=[filter_language, filter_dialect, admin_review_key], outputs=[pending_df])
pending_df.select(select_pending_row, [pending_df], [pending_audio_player, pending_timestamp, pending_orig_utt, edit_utt, edit_lang, edit_dialect, edit_clar, edit_tone])
btn_appr_p.click(self.admin_approve_pending, inputs=[pending_timestamp, pending_orig_utt, edit_utt, edit_lang, edit_dialect, edit_clar, edit_tone, pending_audio_player, admin_review_key], outputs=[audit_log]).then(self.get_pending_dataframe, inputs=[filter_language, filter_dialect, admin_review_key], outputs=[pending_df])
btn_rejt_p.click(self.admin_reject_pending, inputs=[pending_timestamp, pending_orig_utt, reject_reason_dropdown, reject_reason_text, admin_review_key], outputs=[audit_log]).then(self.get_pending_dataframe, inputs=[filter_language, filter_dialect, admin_review_key], outputs=[pending_df])
btn_undo_admin.click(self.admin_undo_last_action, inputs=[admin_review_key], outputs=[audit_log]).then(self.get_pending_dataframe, inputs=[filter_language, filter_dialect, admin_review_key], outputs=[pending_df])
btn_clear_pending.click(self.admin_clear_all_pending, inputs=[admin_review_key], outputs=[audit_log]).then(self.get_pending_dataframe, inputs=[filter_language, filter_dialect, admin_review_key], outputs=[pending_df])
def trigger_recovery(review_key):
if not self._is_full_admin_authorized(review_key):
return self._full_admin_required_message()
try:
import recover_chain # Left local specifically to prevent circular dependency at startup
result_message = recover_chain.main()
return result_message
except Exception as e:
return f"โ Recovery Error: {e}"
recover_btn.click(trigger_recovery, inputs=[admin_review_key], outputs=[backup_log])
# --- Feedback Tab Events ---
def show_feedback_image(evt: gr.SelectData, df):
try:
img_path = str(df.iloc[evt.index[0]].get("Image_Reference", ""))
if img_path == "nan" or not img_path:
return None
if not os.path.exists(img_path) and os.environ.get("HF_TOKEN"):
img_name = os.path.basename(img_path)
try:
dl_img = hf_hub_download(
repo_id="toecm/PureChain_Dataset",
filename=f"feedback_images/{img_name}",
repo_type="dataset",
token=os.environ.get("HF_TOKEN")
)
os.makedirs(os.path.dirname(img_path), exist_ok=True)
shutil.copy(dl_img, img_path)
except:
pass
return img_path if os.path.exists(img_path) else None
except:
return None
btn_refresh_fb.click(self.get_feedback_dataframe, inputs=[admin_review_key], outputs=[feedback_df])
feedback_df.select(show_feedback_image, [feedback_df], [feedback_image])
btn_refresh_eeqs.click(self.get_eeqs_dashboard_dataframe, inputs=[admin_review_key], outputs=[eeqs_df])
btn_download_eeqs.click(self.get_eeqs_download_file, inputs=[admin_review_key], outputs=[eeqs_download_file])
# โฌ๏ธ COMPREHENSIVE API BRIDGE FOR REACT FRONTEND โฌ๏ธ
gr.Markdown("---")
gr.Markdown("### ๐ก API Gateway (Headless endpoints for React)")
with gr.Row(visible=False):
api_sync_out = gr.Textbox()
api_btn_sync = gr.Button()
api_btn_sync.click(fn=check_cloud_sync_status, inputs=[], outputs=[api_sync_out], api_name="check_cloud_sync")
api_btn_admin_undo = gr.Button()
api_admin_undo_out = gr.Textbox()
api_admin_undo_key = gr.Textbox()
api_btn_admin_undo.click(fn=self.admin_undo_last_action, inputs=[api_admin_undo_key], outputs=[api_admin_undo_out], api_name="admin_undo_last_action")
# ๐ข FIX: Added an invisible input box to satisfy the JS client routing
api_btn_dialects = gr.Button()
api_dialects_out = gr.Textbox()
api_btn_dialects.click(fn=self.api_get_dialects, inputs=[], outputs=[api_dialects_out], api_name="api_get_dialects")
api_btn_hierarchy = gr.Button()
api_hierarchy_out = gr.Textbox()
api_btn_hierarchy.click(fn=self.api_get_language_hierarchy, inputs=[], outputs=[api_hierarchy_out], api_name="api_get_language_hierarchy")
api_admin_key = gr.Textbox()
api_history_start = gr.Textbox()
api_history_end = gr.Textbox()
api_history_contracts = gr.CheckboxGroup(
choices=["PureIUUY", "PureConvo", "PureVersation", "PureBi (Jun 11)", "PureBi (Jun 12)", "PureBi (Jun 13)", "PureBi"],
value=["PureBi"]
)
api_admin_df_out = gr.Dataframe()
api_btn_history = gr.Button()
api_btn_history.click(
fn=run_filter,
inputs=[api_history_start, api_history_end, api_history_contracts, api_admin_key],
outputs=[api_admin_df_out],
api_name="run_filter"
)
api_btn_eeqs_dashboard = gr.Button()
api_btn_eeqs_dashboard.click(
fn=self.get_eeqs_dashboard_dataframe,
inputs=[api_admin_key],
outputs=[api_admin_df_out],
api_name="get_eeqs_dashboard_dataframe"
)
api_btn_eeqs_dashboard_alias = gr.Button()
api_btn_eeqs_dashboard_alias.click(
fn=self.get_eeqs_dashboard_dataframe,
inputs=[api_admin_key],
outputs=[api_admin_df_out],
api_name="refresh_eeqs_dashboard"
)
api_btn_mission = gr.Button()
api_topic_in = gr.Textbox()
api_mission_out = gr.Textbox()
api_btn_mission.click(fn=self.api_generate_mission, inputs=[api_topic_in], outputs=[api_mission_out], api_name="generate_mission")
api_btn_transcribe = gr.Button()
api_audio_in = gr.File()
api_dialect_in = gr.Textbox()
api_transcribe_out = gr.Textbox()
api_btn_transcribe.click(fn=self.api_transcribe, inputs=[api_audio_in, api_dialect_in], outputs=[api_transcribe_out], api_name="transcribe_check")
api_btn_transcribe_v2 = gr.Button()
api_speech_model_in = gr.Textbox(value="auto")
api_audio_sanitation_in = gr.Textbox(value="on")
api_btn_transcribe_v2.click(
fn=self.api_transcribe_with_model,
inputs=[api_audio_in, api_dialect_in, api_speech_model_in, api_audio_sanitation_in],
outputs=[api_transcribe_out],
api_name="transcribe_check_v2"
)
api_text_in = gr.Textbox()
api_acoustic_language_in = gr.Textbox()
api_acoustic_voice_in = gr.Textbox(value="browser-native")
api_acoustic_models_out = gr.Textbox()
api_acoustic_json_out = gr.Textbox()
api_btn_acoustic_models = gr.Button()
api_btn_acoustic_models.click(
fn=self.api_acoustic_models,
inputs=[],
outputs=[api_acoustic_models_out],
api_name="api_acoustic_models"
)
api_btn_acoustic_transcribe = gr.Button()
api_btn_acoustic_transcribe.click(
fn=self.api_acoustic_transcribe,
inputs=[api_audio_in, api_acoustic_language_in, api_dialect_in, api_speech_model_in, api_audio_sanitation_in],
outputs=[api_acoustic_json_out],
api_name="api_acoustic_transcribe"
)
api_btn_acoustic_tts = gr.Button()
api_btn_acoustic_tts.click(
fn=self.api_acoustic_tts,
inputs=[api_text_in, api_acoustic_language_in, api_dialect_in, api_acoustic_voice_in],
outputs=[api_acoustic_json_out],
api_name="api_acoustic_tts"
)
api_btn_clarify = gr.Button()
api_clarify_out = gr.Textbox()
api_btn_clarify.click(fn=self.api_clarify, inputs=[api_text_in, api_dialect_in], outputs=[api_clarify_out], api_name="generate_clarifications")
api_ai_model_in = gr.Textbox(value="auto")
api_btn_clarify_v2 = gr.Button()
api_btn_clarify_v2.click(
fn=self.api_clarify_with_model,
inputs=[api_text_in, api_dialect_in, api_ai_model_in],
outputs=[api_clarify_out],
api_name="generate_clarifications_v2"
)
api_btn_submit = gr.Button()
api_custom_d = gr.Textbox()
api_tone = gr.Textbox()
api_context = gr.Textbox()
api_pragmatics = gr.Textbox()
api_source_tag = gr.Textbox(visible=False, value="Web")
api_clar_source = gr.Textbox(visible=False, value="AI")
api_user_key = gr.Textbox(visible=False, value="")
api_confirm = gr.State(False)
# ๐ข NEW: Peer-to-Peer Translation Endpoint
api_btn_translate = gr.Button()
api_translate_text_in = gr.Textbox()
api_translate_source_in = gr.Textbox()
api_translate_target_in = gr.Textbox()
api_translate_out = gr.Textbox()
api_btn_translate.click(
fn=self.api_translate_peer,
inputs=[api_translate_text_in, api_translate_source_in, api_translate_target_in],
outputs=[api_translate_out],
api_name="translate_peer"
)
api_btn_translate_v2 = gr.Button()
api_btn_translate_v2.click(
fn=self.api_translate_peer,
inputs=[api_translate_text_in, api_translate_source_in, api_translate_target_in, api_ai_model_in],
outputs=[api_translate_out],
api_name="translate_peer_v2"
)
# ๐ข NEW: Dialect Relay Endpoints
api_btn_join = gr.Button()
api_q_op = gr.Textbox()
api_q_dialect = gr.Textbox()
api_q_target = gr.Textbox()
api_q_out = gr.Textbox()
api_btn_join.click(fn=self.api_join_queue, inputs=[api_q_op, api_q_dialect, api_q_target], outputs=[api_q_out], api_name="join_queue")
api_btn_lobby = gr.Button()
api_btn_lobby.click(fn=self.api_get_lobby, inputs=[], outputs=[api_q_out], api_name="get_lobby")
api_btn_check = gr.Button()
api_c_out = gr.Textbox()
api_btn_check.click(fn=self.api_check_match, inputs=[api_q_op], outputs=[api_c_out], api_name="check_match")
api_btn_leave = gr.Button()
api_btn_leave.click(fn=self.api_leave_queue, inputs=[api_q_op], outputs=[api_c_out], api_name="leave_queue")
api_btn_relay_send = gr.Button()
api_relay_room = gr.Textbox()
api_relay_text = gr.Textbox()
api_relay_target = gr.Textbox()
api_relay_meaning = gr.Textbox()
api_relay_out = gr.Textbox()
api_btn_relay_send.click(fn=self.api_remote_eval_and_send, inputs=[api_relay_room, api_q_op, api_relay_text, api_q_dialect, api_relay_target, api_relay_meaning], outputs=[api_relay_out], api_name="relay_send")
api_btn_relay_poll = gr.Button()
api_poll_idx = gr.Number()
api_btn_relay_poll.click(fn=self.api_remote_poll, inputs=[api_relay_room, api_poll_idx], outputs=[api_c_out], api_name="relay_poll")
# 7. Secure Feedback Endpoint
# 8. Oracle XP and Task Endpoints
api_xp_op = gr.Textbox()
api_xp_dia = gr.Textbox()
api_review_key = gr.Textbox()
api_xp_out = gr.Textbox()
api_btn_xp = gr.Button()
api_btn_xp.click(fn=self.api_get_user_xp, inputs=[api_xp_op, api_xp_dia], outputs=[api_xp_out], api_name="get_xp")
api_btn_task = gr.Button()
api_btn_task.click(fn=self.api_get_oracle_task, inputs=[api_xp_op, api_xp_dia, api_review_key], outputs=[api_xp_out], api_name="get_oracle_task")
api_btn_review = gr.Button()
api_rev_row = gr.Textbox()
api_rev_chain = gr.Textbox()
api_rev_appr = gr.Textbox()
api_rev_reason = gr.Textbox()
api_btn_review.click(fn=self.api_submit_oracle_review, inputs=[api_rev_row, api_rev_chain, api_xp_op, api_xp_dia, api_rev_appr, api_rev_reason, api_review_key], outputs=[api_xp_out], api_name="submit_oracle_review")
api_btn_appeal = gr.Button()
api_btn_appeal.click(fn=self.api_submit_appeal, inputs=[api_rev_chain, api_xp_op, api_xp_dia, api_review_key], outputs=[api_xp_out], api_name="submit_appeal")
api_fb_btn = gr.Button()
api_fb_op = gr.Textbox()
api_fb_text = gr.Textbox()
api_fb_img = gr.File()
api_fb_out = gr.Textbox()
api_fb_btn.click(
fn=self.handle_feedback_submission,
inputs=[api_fb_op, api_fb_text, api_fb_img],
outputs=[api_fb_out],
api_name="submit_feedback"
)
api_eeqs_btn = gr.Button()
api_eeqs_op = gr.Textbox()
api_eeqs_payload = gr.Textbox()
api_eeqs_out = gr.Textbox()
api_eeqs_btn.click(
fn=self.handle_eeqs_submission,
inputs=[api_eeqs_op, api_eeqs_payload],
outputs=[api_eeqs_out],
api_name="submit_eeqs"
)
# 8. Oracle XP and Task Endpoints
api_btn_get_xp = gr.Button()
api_xp_op = gr.Textbox()
api_xp_dialect = gr.Textbox()
api_review_key_v2 = gr.Textbox()
api_xp_out = gr.Textbox()
api_btn_get_xp.click(fn=self.api_get_user_xp, inputs=[api_xp_op, api_xp_dialect], outputs=[api_xp_out], api_name="get_user_xp")
api_btn_get_oracle = gr.Button()
api_btn_get_oracle.click(fn=self.api_get_oracle_task, inputs=[api_xp_op, api_xp_dialect, api_review_key_v2], outputs=[api_xp_out], api_name="get_oracle_task")
api_btn_get_graveyard = gr.Button()
api_btn_get_graveyard.click(fn=self.api_get_graveyard_task, inputs=[api_xp_op, api_xp_dialect, api_review_key_v2], outputs=[api_xp_out], api_name="get_graveyard_task")
api_btn_sub_oracle = gr.Button()
api_sub_row = gr.Textbox()
api_sub_chain = gr.Textbox()
api_sub_approve = gr.Textbox()
api_btn_sub_oracle.click(fn=self.api_submit_oracle_review, inputs=[api_sub_row, api_sub_chain, api_xp_op, api_xp_dialect, api_sub_approve, api_review_key_v2], outputs=[api_xp_out], api_name="submit_oracle_review")
api_btn_appeal = gr.Button()
api_btn_appeal.click(fn=self.api_submit_appeal, inputs=[api_sub_chain, api_xp_op, api_xp_dialect, api_review_key_v2], outputs=[api_xp_out], api_name="submit_appeal")
api_language_in = gr.Textbox()
api_btn_submit.click(
fn=self.check_and_submit_logic,
inputs=[
api_text_in, # 1. transcribed
api_dialect_in, # 2. dialect
api_custom_d, # 3. customD
api_clarify_out, # 4. clarification
api_tone, # 5. tone
api_context, # 6. context
api_pragmatics, # 7. pragmatics
api_source_tag, # 8. sourceTag
api_clar_source, # 9. clar_source
api_user_key, # 10. userKey
api_audio_in, # 11. blob (audio)
api_confirm, # 12. confirm
api_language_in # 13. language
],
outputs=[feedback_msg, btn_over],
api_name="check_and_submit_logic"
)
# โฌ๏ธ END OF API BRIDGE โฌ๏ธ
return ui