"""
brain.py - Nex-N2-Pro as autonomous AI brain via multi-provider API.
Nex-N2-Pro replaces the local Nanbeige4.1-3B model with a cloud-based
397B parameter MoE model (17B active) that is:
- FREE ($0/M tokens) via OpenRouter
- Far more capable (GPT-5.5 level)
- Has agentic thinking (adaptive reasoning depth)
- Supports tool use / function calling
Provider strategy (priority order):
1. OpenRouter (5 keys, round-robin) - nex-agi/nex-n2-pro:free
2. Scaleway (1M tokens free) - qwen3.5-397b-a17b (same base model)
3. Groq (fallback) - llama-3.3-70b
4. SiliconFlow (legacy, keys may be invalid)
This brain has FULL autonomy and access to:
- Terminal execution (subprocess)
- Web search (DuckDuckGo / SiliconFlow)
- Browser automation (Playwright)
- YouTube API (upload, delete, manage)
- HuggingFace Spaces (deploy, update)
- The entire project codebase (read/write)
- State management (read/write persistent state)
- Pipeline execution (run, diagnose, fix)
"""
import json
import os
import re
import subprocess
import threading
import time
from pathlib import Path
from typing import Optional
import httpx
# ============================================================
# Multi-Provider Configuration
# ============================================================
# OpenRouter: 5 accounts for round-robin (nex-agi/nex-n2-pro:free)
OPENROUTER_KEYS = [
k.strip() for k in os.getenv(
"OPENROUTER_API_KEYS", ""
).split(",") if k.strip()
]
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
OPENROUTER_MODEL = "nex-agi/nex-n2-pro:free"
# Scaleway: 1M tokens free of qwen3.5-397b-a17b (Nex-N2-Pro base model)
SCALEWAY_API_KEY = os.getenv("SCALEWAY_API_KEY", "")
SCALEWAY_BASE_URL = "https://api.scaleway.ai/llm/v1"
SCALEWAY_MODEL = "qwen3.5-397b-a17b"
# Groq: Fallback for smaller tasks
GROQ_API_KEY = os.getenv("GROQ_API_KEY", "")
GROQ_BASE_URL = "https://api.groq.com/openai/v1"
GROQ_MODEL = "llama-3.3-70b-versatile"
# SiliconFlow: Legacy (keys may be invalid, kept for TTS)
SILICONFLOW_API_KEY = os.getenv("SILICONFLOW_API_KEY", "")
SILICONFLOW_BASE_URL = "https://api.siliconflow.cn/v1"
NEX_MODEL = "nex-agi/Nex-N2-Pro"
# Available TTS models on SiliconFlow
TTS_MODELS = {
"fish-speech": "fishaudio/fish-speech-1.5",
"cosyvoice": "FunAudioLLM/CosyVoice2-0.5B",
"indextts": "IndexTeam/IndexTTS-2",
}
# Default TTS configuration
DEFAULT_TTS_MODEL = os.getenv("TTS_MODEL", "fishaudio/fish-speech-1.5")
DEFAULT_TTS_VOICE = os.getenv("TTS_VOICE", "fishaudio/fish-speech-1.5:claire")
# Rate limit tracking
_key_lock = threading.Lock()
_key_daily_usage = {} # {key_index: {"count": N, "reset_date": "YYYY-MM-DD"}}
_round_robin_index = 0
def _get_next_openrouter_key() -> tuple:
"""Round-robin key selection with daily usage tracking.
OpenRouter free tier: 20 req/min, 50 req/day per key.
With 5 keys: 100 req/min, 250 req/day total.
With $10/acc: 20 req/min, 1000 req/day per key = 5000 req/day total.
"""
global _round_robin_index
if not OPENROUTER_KEYS:
return None, -1
today = time.strftime("%Y-%m-%d")
with _key_lock:
# Try all keys starting from round-robin position
for _ in range(len(OPENROUTER_KEYS)):
idx = _round_robin_index % len(OPENROUTER_KEYS)
_round_robin_index += 1
key = OPENROUTER_KEYS[idx]
usage = _key_daily_usage.get(idx, {"count": 0, "reset_date": today})
# Reset counter on new day
if usage["reset_date"] != today:
usage = {"count": 0, "reset_date": today}
_key_daily_usage[idx] = usage
# Free tier: 50 req/day. With credits: 1000 req/day.
# Use conservative limit of 45 req/day to stay safe on free tier.
if usage["count"] < 45:
usage["count"] += 1
_key_daily_usage[idx] = usage
return key, idx
# All keys exhausted for today
return None, -1
def _mark_key_rate_limited(idx: int):
"""Mark a key as rate-limited (429 received)."""
today = time.strftime("%Y-%m-%d")
with _key_lock:
usage = _key_daily_usage.get(idx, {"count": 0, "reset_date": today})
usage["count"] = 999 # Force exhaustion
_key_daily_usage[idx] = usage
def get_provider_status() -> dict:
"""Get current provider status and usage info."""
today = time.strftime("%Y-%m-%d")
status = {
"openrouter": {
"keys_total": len(OPENROUTER_KEYS),
"model": OPENROUTER_MODEL,
"daily_usage": {},
},
"scaleway": {
"available": bool(SCALEWAY_API_KEY),
"model": SCALEWAY_MODEL,
},
"groq": {
"available": bool(GROQ_API_KEY),
"model": GROQ_MODEL,
},
"siliconflow": {
"available": bool(SILICONFLOW_API_KEY),
"model": NEX_MODEL,
},
}
with _key_lock:
for idx in range(len(OPENROUTER_KEYS)):
usage = _key_daily_usage.get(idx, {"count": 0, "reset_date": today})
if usage["reset_date"] != today:
usage = {"count": 0, "reset_date": today}
status["openrouter"]["daily_usage"][f"key_{idx}"] = {
"count": usage["count"],
"limit": 45,
"remaining": max(0, 45 - usage["count"]),
}
return status
class NexBrain:
"""Autonomous AI brain using Nex-N2-Pro via multi-provider API.
Provider priority:
1. OpenRouter (5 keys, round-robin) - FREE, 250+ req/day
2. Scaleway (1M tokens free) - Same base model (qwen3.5-397b)
3. Groq (free tier) - Llama 3.3 70B fallback
4. SiliconFlow (legacy) - May have auth issues
This brain has TOTAL control over the project. It can:
- Execute any terminal command
- Read and modify any file
- Search the web
- Browse websites
- Upload/delete YouTube videos
- Deploy to HuggingFace Spaces
- Make all decisions autonomously
- Diagnose and fix pipeline bugs
- Run the full translation pipeline
"""
def __init__(self, state=None, pipeline=None, browser=None):
self.api_key = SILICONFLOW_API_KEY # Legacy SiliconFlow key
self.base_url = SILICONFLOW_BASE_URL
self.model = NEX_MODEL
self.state = state
self.pipeline = pipeline
self.browser = browser
self._loaded = True # Cloud model, always "loaded"
self._conversation_history = []
self._max_history = 50 # Keep last 50 exchanges
self._active_provider = None # Track which provider last worked
# Log provider availability at startup
n_keys = len(OPENROUTER_KEYS)
print(f"[NEX-BRAIN] Providers: OpenRouter({n_keys} keys), "
f"Scaleway({'yes' if SCALEWAY_API_KEY else 'no'}), "
f"Groq({'yes' if GROQ_API_KEY else 'no'}), "
f"SiliconFlow({'yes' if SILICONFLOW_API_KEY else 'no'})")
def is_ready(self) -> bool:
"""Check if the brain is ready. True if ANY provider has a key."""
return bool(OPENROUTER_KEYS or SCALEWAY_API_KEY or GROQ_API_KEY or SILICONFLOW_API_KEY)
def _call_nex(self, messages: list, max_tokens: int = 2048,
temperature: float = 0.3, tools: list = None) -> dict:
"""Call Nex-N2-Pro via multi-provider with automatic failover.
Provider priority:
1. OpenRouter (round-robin across 5 keys)
2. Scaleway (1M tokens free of qwen3.5-397b)
3. Groq (free Llama 3.3 70B)
4. SiliconFlow (legacy, may fail with 401)
Returns the full API response.
"""
# === Provider 1: OpenRouter (round-robin) ===
if OPENROUTER_KEYS:
# Try up to 5 different keys
for attempt in range(min(5, len(OPENROUTER_KEYS))):
key, idx = _get_next_openrouter_key()
if not key:
print(f"[NEX-BRAIN] All OpenRouter keys exhausted for today")
break
try:
result = self._call_provider(
base_url=OPENROUTER_BASE_URL,
api_key=key,
model=OPENROUTER_MODEL,
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
tools=tools,
extra_headers={"HTTP-Referer": "https://autodub-herspace.space"},
)
self._active_provider = f"openrouter:key{idx}"
return result
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
print(f"[NEX-BRAIN] OpenRouter key {idx} rate limited, rotating...")
_mark_key_rate_limited(idx)
continue
elif e.response.status_code == 401:
print(f"[NEX-BRAIN] OpenRouter key {idx} unauthorized, skipping")
_mark_key_rate_limited(idx)
continue
else:
print(f"[NEX-BRAIN] OpenRouter key {idx} error {e.response.status_code}")
continue
except Exception as e:
print(f"[NEX-BRAIN] OpenRouter key {idx} exception: {e}")
continue
# === Provider 2: Scaleway (qwen3.5-397b-a17b = Nex base model) ===
if SCALEWAY_API_KEY:
try:
print("[NEX-BRAIN] Fallback to Scaleway (qwen3.5-397b)")
result = self._call_provider(
base_url=SCALEWAY_BASE_URL,
api_key=SCALEWAY_API_KEY,
model=SCALEWAY_MODEL,
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
tools=tools,
)
self._active_provider = "scaleway"
return result
except Exception as e:
print(f"[NEX-BRAIN] Scaleway failed: {e}")
# === Provider 3: Groq (Llama 3.3 70B fallback) ===
if GROQ_API_KEY:
try:
print("[NEX-BRAIN] Fallback to Groq (Llama 3.3 70B)")
result = self._call_provider(
base_url=GROQ_BASE_URL,
api_key=GROQ_API_KEY,
model=GROQ_MODEL,
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
tools=tools,
)
self._active_provider = "groq"
return result
except Exception as e:
print(f"[NEX-BRAIN] Groq failed: {e}")
# === Provider 4: SiliconFlow (legacy, may fail) ===
if SILICONFLOW_API_KEY:
try:
print("[NEX-BRAIN] Last resort: SiliconFlow (may 401)")
result = self._call_provider(
base_url=SILICONFLOW_BASE_URL,
api_key=SILICONFLOW_API_KEY,
model=NEX_MODEL,
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
tools=tools,
)
self._active_provider = "siliconflow"
return result
except Exception as e:
print(f"[NEX-BRAIN] SiliconFlow failed: {e}")
# All providers failed
print("[NEX-BRAIN] ALL PROVIDERS FAILED")
return {"choices": [{"message": {"content": "ERROR: All API providers failed. Check API keys and network."}}]}
def _call_provider(self, base_url: str, api_key: str, model: str,
messages: list, max_tokens: int = 2048,
temperature: float = 0.3, tools: list = None,
extra_headers: dict = None) -> dict:
"""Make an API call to a specific provider. Raises on HTTP errors."""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
if extra_headers:
headers.update(extra_headers)
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
"top_p": 0.9,
}
if tools:
payload["tools"] = tools
payload["tool_choice"] = "auto"
with httpx.Client(timeout=120) as client:
resp = client.post(
f"{base_url}/chat/completions",
headers=headers,
json=payload,
)
resp.raise_for_status()
return resp.json()
def _chat(self, system_prompt: str, user_message: str,
max_tokens: int = 2048, temperature: float = 0.3) -> str:
"""Simple chat interface. Returns the assistant's text response."""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
]
response = self._call_nex(messages, max_tokens=max_tokens,
temperature=temperature)
try:
content = response["choices"][0]["message"]["content"]
# FIX: content can be None when model only returns reasoning
if content is None:
# Try reasoning_content as fallback
content = response["choices"][0]["message"].get("reasoning_content", "") or ""
# FIX: More aggressive cleaning of thinking tokens
# 1. Remove ... blocks
content = re.sub(r'.*?', '', content, flags=re.DOTALL)
# 2. Remove ... blocks
content = re.sub(r'.*?', '', content, flags=re.DOTALL)
# 3. Remove unclosed at the start (model sometimes doesn't close)
content = re.sub(r'^.*?(?=\{|\[|"|[A-ZÁÉÍÓÚÑ¿¡]|$)', '', content, flags=re.DOTALL)
# 4. If content starts with and never closes, return empty
if content.strip().startswith(''):
return ""
return content.strip()
except (KeyError, IndexError, TypeError):
return ""
# ================================================================
# TOOL DEFINITIONS for Nex-N2-Pro autonomous mode
# ================================================================
TOOLS = [
{
"type": "function",
"function": {
"name": "execute_terminal",
"description": "Execute a terminal/shell command. Returns stdout, stderr, and exit code. Use for: file operations, ffmpeg, yt-dlp, git, pip, system commands, etc.",
"parameters": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The shell command to execute"
},
"timeout": {
"type": "integer",
"description": "Timeout in seconds (default 60, max 300)",
"default": 60
}
},
"required": ["command"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read the contents of a file. Returns the file content as text.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Absolute path to the file"
},
"start_line": {
"type": "integer",
"description": "Start line number (1-based, optional)"
},
"end_line": {
"type": "integer",
"description": "End line number (optional)"
}
},
"required": ["path"]
}
}
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "Write content to a file. Creates the file if it doesn't exist, overwrites if it does.",
"parameters": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Absolute path to the file"
},
"content": {
"type": "string",
"description": "The content to write"
}
},
"required": ["path", "content"]
}
}
},
{
"type": "function",
"function": {
"name": "web_search",
"description": "Search the web for information. Returns search results with URLs, titles, and snippets.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
},
"num_results": {
"type": "integer",
"description": "Number of results to return (default 5)",
"default": 5
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "browse_page",
"description": "Read the content of a web page. Returns the page title and main content as text.",
"parameters": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The URL to read"
}
},
"required": ["url"]
}
}
},
{
"type": "function",
"function": {
"name": "youtube_list_videos",
"description": "List videos on the YouTube channel. Returns video IDs, titles, and status.",
"parameters": {
"type": "object",
"properties": {
"max_results": {
"type": "integer",
"description": "Max number of videos to return (default 10)",
"default": 10
}
}
}
}
},
{
"type": "function",
"function": {
"name": "youtube_delete_video",
"description": "Delete a video from the YouTube channel by video ID.",
"parameters": {
"type": "object",
"properties": {
"video_id": {
"type": "string",
"description": "The YouTube video ID to delete"
}
},
"required": ["video_id"]
}
}
},
{
"type": "function",
"function": {
"name": "youtube_upload_video",
"description": "Upload a video file to YouTube.",
"parameters": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Absolute path to the video file"
},
"title": {
"type": "string",
"description": "Video title (max 100 chars)"
},
"description": {
"type": "string",
"description": "Video description"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Video tags"
}
},
"required": ["file_path", "title"]
}
}
},
{
"type": "function",
"function": {
"name": "hf_space_deploy",
"description": "Deploy files to the HuggingFace Space. Updates the Space with new/modified files.",
"parameters": {
"type": "object",
"properties": {
"files": {
"type": "array",
"items": {
"type": "object",
"properties": {
"local_path": {"type": "string", "description": "Local file path"},
"repo_path": {"type": "string", "description": "Path in the HF Space repo"}
},
"required": ["local_path", "repo_path"]
},
"description": "List of files to deploy"
},
"commit_message": {
"type": "string",
"description": "Git commit message for the deployment"
}
},
"required": ["files", "commit_message"]
}
}
},
{
"type": "function",
"function": {
"name": "generate_tts",
"description": "Generate speech audio from text using SiliconFlow TTS. Supports Spanish and many other languages. Models: fish-speech-1.5 (best), cosyvoice2, indextts.",
"parameters": {
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "The text to convert to speech"
},
"output_path": {
"type": "string",
"description": "Path to save the audio file"
},
"model": {
"type": "string",
"description": "TTS model: 'fish-speech', 'cosyvoice', or 'indextts'",
"default": "fish-speech"
},
"voice": {
"type": "string",
"description": "Voice name (e.g. 'claire' for female Spanish, 'alex' for male)",
"default": "claire"
},
"response_format": {
"type": "string",
"description": "Audio format: mp3, wav, opus",
"default": "mp3"
}
},
"required": ["text", "output_path"]
}
}
},
{
"type": "function",
"function": {
"name": "read_state",
"description": "Read the current persistent state (processed videos, config, queue, visual_config, etc.)",
"parameters": {
"type": "object",
"properties": {}
}
}
},
{
"type": "function",
"function": {
"name": "update_state",
"description": "Update the persistent state. Provide a partial dict to merge with existing state.",
"parameters": {
"type": "object",
"properties": {
"updates": {
"type": "object",
"description": "Dict of state keys to update"
}
},
"required": ["updates"]
}
}
},
{
"type": "function",
"function": {
"name": "run_pipeline",
"description": "Run the translation pipeline on a video. Downloads, transcribes, translates, generates TTS, composes, and uploads.",
"parameters": {
"type": "object",
"properties": {
"video_id": {
"type": "string",
"description": "YouTube video ID to process"
},
"title": {
"type": "string",
"description": "Video title"
},
"url": {
"type": "string",
"description": "YouTube URL"
}
},
"required": ["video_id", "title", "url"]
}
}
},
{
"type": "function",
"function": {
"name": "diagnose_pipeline",
"description": "Diagnose pipeline issues. Checks: FFmpeg, TTS engines, API connectivity, font availability, state integrity. Returns a detailed health report.",
"parameters": {
"type": "object",
"properties": {}
}
}
}
]
# ================================================================
# TOOL EXECUTION
# ================================================================
def _execute_tool(self, tool_name: str, arguments: dict) -> str:
"""Execute a tool call and return the result as a string."""
try:
if tool_name == "execute_terminal":
cmd = arguments["command"]
timeout = min(arguments.get("timeout", 60), 300)
proc = subprocess.run(
cmd, shell=True, capture_output=True, text=True, timeout=timeout
)
result = f"EXIT CODE: {proc.returncode}\n"
if proc.stdout:
result += f"STDOUT:\n{proc.stdout[:10000]}\n"
if proc.stderr:
result += f"STDERR:\n{proc.stderr[:5000]}\n"
return result
elif tool_name == "read_file":
path = arguments["path"]
start = arguments.get("start_line", 1)
end = arguments.get("end_line")
with open(path, "r", encoding="utf-8", errors="replace") as f:
lines = f.readlines()
if end:
lines = lines[start-1:end]
else:
lines = lines[start-1:]
return "".join(lines)[:20000]
elif tool_name == "write_file":
path = arguments["path"]
content = arguments["content"]
Path(path).parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return f"File written: {path} ({len(content)} bytes)"
elif tool_name == "web_search":
query = arguments["query"]
num = arguments.get("num_results", 5)
try:
from duckduckgo_search import DDGS
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=num))
if results:
return json.dumps(results, indent=2, ensure_ascii=False)[:10000]
return "No results found"
except ImportError:
return f"Web search not available (duckduckgo_search not installed). Query was: {query}"
except Exception as e:
return f"Web search error: {e}"
elif tool_name == "browse_page":
url = arguments["url"]
with httpx.Client(timeout=30, follow_redirects=True) as client:
resp = client.get(url, headers={"User-Agent": "Mozilla/5.0 (AutoDub/2.0)"})
resp.raise_for_status()
text = re.sub(r']*>.*?', '', resp.text, flags=re.DOTALL)
text = re.sub(r']*>.*?', '', text, flags=re.DOTALL)
text = re.sub(r'<[^>]+>', ' ', text)
text = re.sub(r'\s+', ' ', text).strip()
return text[:15000]
elif tool_name == "youtube_list_videos":
return self._youtube_list_videos(arguments.get("max_results", 10))
elif tool_name == "youtube_delete_video":
return self._youtube_delete_video(arguments["video_id"])
elif tool_name == "youtube_upload_video":
return self._youtube_upload_video(
arguments["file_path"],
arguments["title"],
arguments.get("description", ""),
arguments.get("tags", [])
)
elif tool_name == "hf_space_deploy":
return self._hf_space_deploy(
arguments["files"],
arguments["commit_message"]
)
elif tool_name == "generate_tts":
return self._generate_tts(
arguments["text"],
arguments["output_path"],
arguments.get("model", "fish-speech"),
arguments.get("voice", "claire"),
arguments.get("response_format", "mp3")
)
elif tool_name == "read_state":
if self.state:
state_data = self.state._state if hasattr(self.state, '_state') else {}
return json.dumps(state_data, indent=2, ensure_ascii=False)[:20000]
return "No state manager available"
elif tool_name == "update_state":
if self.state and hasattr(self.state, '_state'):
updates = arguments["updates"]
for key, value in updates.items():
if isinstance(value, dict) and key in self.state._state and isinstance(self.state._state[key], dict):
self.state._state[key].update(value)
else:
self.state._state[key] = value
self.state.save()
return f"State updated with keys: {list(updates.keys())}"
return "No state manager available"
elif tool_name == "run_pipeline":
if self.pipeline:
return "Pipeline execution should be triggered from the async app layer. Use execute_terminal to test individual steps."
return "No pipeline available"
elif tool_name == "diagnose_pipeline":
return self._diagnose_pipeline()
else:
return f"Unknown tool: {tool_name}"
except subprocess.TimeoutExpired:
return f"Command timed out after {arguments.get('timeout', 60)}s"
except Exception as e:
return f"Tool execution error: {e}"
# ================================================================
# Diagnostics
# ================================================================
def _diagnose_pipeline(self) -> str:
"""Run comprehensive pipeline diagnostics."""
report = []
report.append("=== PIPELINE DIAGNOSTICS ===\n")
# Check FFmpeg
try:
proc = subprocess.run(["ffmpeg", "-version"], capture_output=True, text=True, timeout=10)
version_line = proc.stdout.split('\n')[0] if proc.stdout else "unknown"
report.append(f"FFmpeg: {version_line}")
except Exception as e:
report.append(f"FFmpeg: MISSING - {e}")
# Check Python packages
for pkg in ["supertonic", "gtts", "groq", "httpx", "duckduckgo_search", "PIL", "numpy"]:
try:
__import__(pkg.replace("-", "_"))
report.append(f"Package {pkg}: OK")
except ImportError:
report.append(f"Package {pkg}: MISSING")
# Check fonts
font_dirs = ["/usr/share/fonts/truetype/montserrat", "/app/fonts"]
for fd in font_dirs:
if Path(fd).exists():
fonts = list(Path(fd).glob("*.ttf"))
report.append(f"Fonts in {fd}: {len(fonts)} files ({[f.name for f in fonts[:3]]})")
else:
report.append(f"Font dir {fd}: NOT FOUND")
# Check API keys (multi-provider)
for key_name in ["OPENROUTER_API_KEYS", "SCALEWAY_API_KEY", "GROQ_API_KEY", "SILICONFLOW_API_KEY", "HF_TOKEN", "GOOGLE_CLIENT_ID"]:
val = os.getenv(key_name, "")
if key_name == "OPENROUTER_API_KEYS":
n_keys = len([k for k in val.split(",") if k.strip()])
report.append(f"Env {key_name}: {n_keys} keys configured")
else:
report.append(f"Env {key_name}: {'SET (' + val[:8] + '...)' if val else 'NOT SET'}")
# Check state
if self.state:
try:
state = self.state._state if hasattr(self.state, '_state') else {}
processed = len(state.get("processed_videos", {}))
failed = len(state.get("failed_videos", {}))
report.append(f"State: {processed} processed, {failed} failed videos")
except Exception as e:
report.append(f"State: ERROR - {e}")
else:
report.append("State: NOT INITIALIZED")
# Check OpenRouter API
if OPENROUTER_KEYS:
try:
key = OPENROUTER_KEYS[0]
with httpx.Client(timeout=10) as client:
resp = client.get(
f"{OPENROUTER_BASE_URL}/models",
headers={"Authorization": f"Bearer {key}"},
)
if resp.status_code == 200:
report.append(f"OpenRouter API: OK ({len(OPENROUTER_KEYS)} keys, model={OPENROUTER_MODEL})")
else:
report.append(f"OpenRouter API: ERROR {resp.status_code}")
except Exception as e:
report.append(f"OpenRouter API: CONNECTION ERROR - {e}")
else:
report.append("OpenRouter API: NO KEYS")
# Check Scaleway API
if SCALEWAY_API_KEY:
try:
with httpx.Client(timeout=10) as client:
resp = client.get(
f"{SCALEWAY_BASE_URL}/models",
headers={"Authorization": f"Bearer {SCALEWAY_API_KEY}"},
)
if resp.status_code == 200:
report.append(f"Scaleway API: OK (model={SCALEWAY_MODEL})")
else:
report.append(f"Scaleway API: ERROR {resp.status_code}")
except Exception as e:
report.append(f"Scaleway API: CONNECTION ERROR - {e}")
else:
report.append("Scaleway API: NO KEY")
# Check SiliconFlow API (legacy, may fail)
if SILICONFLOW_API_KEY:
try:
with httpx.Client(timeout=10) as client:
resp = client.get(
f"{self.base_url}/models",
headers={"Authorization": f"Bearer {self.api_key}"},
)
if resp.status_code == 200:
report.append(f"SiliconFlow API: OK (legacy)")
else:
report.append(f"SiliconFlow API: ERROR {resp.status_code} (legacy)")
except Exception as e:
report.append(f"SiliconFlow API: CONNECTION ERROR - {e}")
else:
report.append("SiliconFlow API: NO KEY")
return "\n".join(report)
# ================================================================
# YouTube API Helpers
# ================================================================
def _get_youtube_service(self):
"""Get authenticated YouTube service."""
if not self.state:
return None
tokens = self.state.get_youtube_tokens()
if not tokens:
return None
try:
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
creds = Credentials.from_authorized_user_info(tokens, [
"https://www.googleapis.com/auth/youtube",
"https://www.googleapis.com/auth/youtube.upload",
])
return build("youtube", "v3", credentials=creds)
except Exception as e:
print(f"[NEX-BRAIN] YouTube auth error: {e}")
return None
def _youtube_list_videos(self, max_results: int = 10) -> str:
"""List channel videos."""
yt = self._get_youtube_service()
if not yt:
return "YouTube not authenticated. Connect your channel first."
try:
channels = yt.channels().list(mine=True, part="contentDetails").execute()
if not channels.get("items"):
return "No YouTube channel found"
playlist_id = channels["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"]
playlist_items = yt.playlistItems().list(
playlistId=playlist_id,
part="snippet,contentDetails",
maxResults=max_results
).execute()
videos = []
for item in playlist_items.get("items", []):
snippet = item["snippet"]
videos.append({
"video_id": item["contentDetails"]["videoId"],
"title": snippet["title"],
"published_at": snippet.get("publishedAt", ""),
"status": snippet.get("status", ""),
})
return json.dumps(videos, indent=2, ensure_ascii=False)
except Exception as e:
return f"YouTube API error: {e}"
def _youtube_delete_video(self, video_id: str) -> str:
"""Delete a YouTube video."""
yt = self._get_youtube_service()
if not yt:
return "YouTube not authenticated."
try:
yt.videos().delete(id=video_id).execute()
return f"Video {video_id} deleted successfully"
except Exception as e:
return f"Delete failed: {e}"
def _youtube_upload_video(self, file_path: str, title: str,
description: str = "", tags: list = None) -> str:
"""Upload a video to YouTube."""
yt = self._get_youtube_service()
if not yt:
return "YouTube not authenticated."
try:
from googleapiclient.http import MediaFileUpload
body = {
"snippet": {
"title": title[:100],
"description": description[:5000],
"tags": tags or [],
"categoryId": "22",
},
"status": {
"privacyStatus": "public",
"selfDeclaredMadeForKids": False,
},
}
media = MediaFileUpload(file_path, resumable=True)
request = yt.videos().insert(
part="snippet,status",
body=body,
media_body=media,
)
response = None
while response is None:
status, response = request.next_chunk()
if status:
print(f"[NEX-BRAIN] Upload progress: {int(status.progress() * 100)}%")
return json.dumps({
"status": "success",
"video_id": response["id"],
"title": response["snippet"]["title"],
})
except Exception as e:
return f"Upload failed: {e}"
# ================================================================
# HuggingFace Space Deploy
# ================================================================
def _hf_space_deploy(self, files: list, commit_message: str) -> str:
"""Deploy files to the HF Space."""
try:
from huggingface_hub import HfApi
token = os.getenv("HF_TOKEN", "")
repo_id = os.getenv("HF_SPACE_REPO", "TomatitoToho/autodub-herspace")
api = HfApi(token=token)
for file_info in files:
api.upload_file(
path_or_fileobj=file_info["local_path"],
path_in_repo=file_info["repo_path"],
repo_id=repo_id,
repo_type="space",
commit_message=commit_message,
)
return f"Deployed {len(files)} files to {repo_id}"
except Exception as e:
return f"Deploy failed: {e}"
# ================================================================
# TTS via SiliconFlow API
# ================================================================
def _generate_tts(self, text: str, output_path: str, model: str = "fish-speech",
voice: str = "claire", response_format: str = "mp3") -> str:
"""Generate TTS using SiliconFlow TTS models.
Supported models:
- fish-speech: fishaudio/fish-speech-1.5 (best quality, $15/M bytes)
- cosyvoice: FunAudioLLM/CosyVoice2-0.5B ($7.15/M bytes)
- indextts: IndexTeam/IndexTTS-2 ($7.15/M bytes)
"""
model_id = TTS_MODELS.get(model, TTS_MODELS["fish-speech"])
full_voice = f"{model_id}:{voice}" if ":" not in voice else voice
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
payload = {
"model": model_id,
"input": text,
"voice": full_voice,
"response_format": response_format,
}
try:
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
with httpx.Client(timeout=60) as client:
resp = client.post(
f"{self.base_url}/audio/speech",
headers=headers,
json=payload,
)
resp.raise_for_status()
with open(output_path, "wb") as f:
f.write(resp.content)
size_kb = Path(output_path).stat().st_size / 1024
return f"TTS generated: {output_path} ({size_kb:.0f} KB, model={model_id})"
except Exception as e:
return f"TTS failed: {e}"
# ================================================================
# AUTONOMOUS AGENT LOOP
# ================================================================
def think_autonomous(self, task: str, max_iterations: int = 10) -> str:
"""Run an autonomous agent loop with tool use.
Nex-N2-Pro thinks, uses tools, observes results, and iterates
until the task is complete or max iterations reached.
"""
system_prompt = self._get_system_prompt()
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": task},
]
iteration = 0
final_response = ""
while iteration < max_iterations:
iteration += 1
print(f"[NEX-BRAIN] Agent iteration {iteration}/{max_iterations}")
response = self._call_nex(
messages,
max_tokens=4096,
temperature=0.3,
tools=self.TOOLS
)
try:
choice = response["choices"][0]
message = choice["message"]
content = message.get("content", "")
tool_calls = message.get("tool_calls", [])
# Add assistant message to history
messages.append(message)
# If no tool calls, the agent is done
if not tool_calls:
content = re.sub(r'.*?', '', content, flags=re.DOTALL)
final_response = content.strip()
break
# Execute each tool call
for tool_call in tool_calls:
func = tool_call["function"]
tool_name = func["name"]
try:
arguments = json.loads(func["arguments"]) if isinstance(func["arguments"], str) else func["arguments"]
except json.JSONDecodeError:
arguments = {}
print(f"[NEX-BRAIN] Tool call: {tool_name}({json.dumps(arguments)[:200]})")
result = self._execute_tool(tool_name, arguments)
print(f"[NEX-BRAIN] Tool result: {result[:300]}...")
# Add tool result to messages
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": result[:10000],
})
except (KeyError, IndexError) as e:
final_response = f"Agent error: {e}"
break
if iteration >= max_iterations:
final_response = f"Agent reached max iterations ({max_iterations}). Last response: {final_response[:500]}"
# Store in conversation history
self._conversation_history.append({
"task": task,
"iterations": iteration,
"result": final_response[:2000],
"timestamp": time.time(),
})
if len(self._conversation_history) > self._max_history:
self._conversation_history = self._conversation_history[-self._max_history:]
return final_response
# ================================================================
# SYSTEM PROMPT (Full Project Context - IMPROVED)
# ================================================================
def _get_system_prompt(self) -> str:
"""Return the comprehensive system prompt with full project context."""
provider_info = f"Active providers: OpenRouter({len(OPENROUTER_KEYS)} keys, {OPENROUTER_MODEL})"
if SCALEWAY_API_KEY:
provider_info += f", Scaleway({SCALEWAY_MODEL})"
provider_info += f", Groq({GROQ_MODEL})"
return (f"You are Nex-N2-Pro, an autonomous AI agent that manages a YouTube Shorts translation channel. "
f"You have TOTAL control over the project and can do anything you want.\n\n"
f"## YOUR IDENTITY\n"
f"- You are the \"brain\" of an automated YouTube channel called \"HerYTStory\"\n"
f"- You translate English YouTube Shorts to Spanish\n"
f"- You replaced the previous brain (Nanbeige4.1-3B) and are far more capable\n"
f"- You are a 397B MoE model based on Qwen3.5 with agentic reasoning\n"
f"- You can think deeply about problems using tags\n"
f"- {provider_info}\n\n"
+ self._get_system_prompt_body())
def _get_system_prompt_body(self) -> str:
"""Return the static body of the system prompt (no f-string variables)."""
return """
## YOUR PROJECT ARCHITECTURE
The project runs on a Hugging Face Space (free tier):
- **2 vCPU Intel Xeon** (AVX-512), 16GB RAM, 50GB ephemeral storage
- **~60s request timeout**, sleeps after 48h inactivity
- **Python 3.11**, FastAPI web server on port 7860
### Project Files (at /app/ on the Space):
- **app.py** - Main FastAPI app, YouTube OAuth, scheduling, dashboard
- **pipeline.py** - Translation pipeline (download -> transcribe -> translate -> TTS -> compose -> upload)
- **brain.py** - THIS FILE. You are the AI brain.
- **browser.py** - Playwright browser for YouTube downloads/cookies
- **state.py** - Persistent state via HF Dataset (survives restarts)
- **monitor.py** - YouTube channel monitoring (RSS + Invidious + Piped + API)
- **Dockerfile** - Container setup (Python 3.11, FFmpeg, Playwright Chromium)
- **requirements.txt** - Python dependencies
- **fonts/** - Montserrat Black/Bold for subtitles
- **templates/dashboard.html** - Web dashboard UI
### Pipeline Steps:
1. **Download** YouTube Short (CF Worker -> VideoDL -> RapidAPI -> Browser -> cobalt -> yt-dlp -> Invidious -> Piped)
2. **Extract audio** from video (FFmpeg)
3. **Transcribe** with timestamps (Groq Whisper API, segment granularity)
4. **Translate** EN->ES per-segment (Groq Llama 3.3 70B)
5. **Validate** translation (YOU - the brain)
6. **Generate Spanish voice** per-segment, speed-adjusted to fit segment duration (Supertonic 3 / SiliconFlow TTS / gTTS)
7. **Compose** final video: scale + crop + hflip + blur + ASS subtitles + Spanish voice + bg music (FFmpeg)
8. **Upload** to YouTube (YouTube API v3)
### State Management:
State is stored in a HuggingFace Dataset (private), persisted across Space restarts:
- **processed_videos** - Dict of video_id -> metadata
- **failed_videos** - Dict of video_id -> error info
- **youtube_tokens** - OAuth tokens for YouTube upload
- **config** - Channel config (check_interval, target_channel, language)
- **visual_config** - Blur position, subtitle font size, margins, etc.
- **queue** - Videos queued for processing
### YouTube Channel:
- **Channel**: @HerYTStory (Her Story)
- **Channel ID**: UCpHpLD-URpThtttHTxefZ4A
- **Content**: English storytelling/history Shorts -> Spanish translation
### API Keys Available (via environment variables):
- OPENROUTER_API_KEYS - 5 OpenRouter keys for Nex-N2-Pro:free (round-robin, 250+ req/day)
- SCALEWAY_API_KEY - Scaleway qwen3.5-397b (1M tokens free, same base model)
- GROQ_API_KEY - For transcription (Whisper), translation (Llama 3.3 70B), and fallback
- SILICONFLOW_API_KEY - Legacy key for TTS and Nex (may return 401)
- HF_TOKEN - HuggingFace token for state and Space management
- GOOGLE_CLIENT_ID / GOOGLE_CLIENT_SECRET - YouTube OAuth
- VIDEODL_API_KEY - Video download service
- RAPIDAPI_KEY - YouTube downloader API
## YOUR CAPABILITIES
You have these tools available:
1. **execute_terminal** - Run ANY shell command (ffmpeg, yt-dlp, python, pip, git, etc.)
2. **read_file** - Read any file in the project
3. **write_file** - Write/modify any file
4. **web_search** - Search the internet
5. **browse_page** - Read web page content
6. **youtube_list_videos** - List your channel's videos
7. **youtube_delete_video** - Delete videos from your channel
8. **youtube_upload_video** - Upload videos to YouTube
9. **hf_space_deploy** - Deploy changes to the HuggingFace Space
10. **generate_tts** - Generate speech from text (Fish Speech 1.5, CosyVoice, IndexTTS - all support Spanish)
11. **read_state** - Read persistent state
12. **update_state** - Update persistent state
13. **run_pipeline** - Run the full translation pipeline on a video
14. **diagnose_pipeline** - Run comprehensive pipeline diagnostics
## RECENT BUG FIXES APPLIED
1. **Voice desync FIXED**: Changed speed adjustment threshold from 5% to 1%, added audio truncation as safety net, fixed current_time tracking to use segment end timestamps instead of audio_duration
2. **Blur FIXED**: Changed boxblur from lr:lp=25:25 (25 iterations, extremely slow) to lr:cr=10:2 (fast and effective)
3. **Subtitles FIXED**: Added Montserrat Black auto-download, improved font search, ensured fontsdir is always valid
4. **TTS replaced**: Edge-TTS -> Supertonic 3 (primary) + gTTS (fallback) + SiliconFlow TTS (API option)
## TTS OPTIONS (ordered by quality)
1. **Supertonic 3** (PRIMARY - local, ONNX, 99M params, 31 languages, RTF ~0.31)
- Voice: M1 (male Spanish), F1 (female Spanish)
- Free, no API needed, runs on CPU
2. **Fish Speech 1.5** via SiliconFlow API (HIGH QUALITY - use generate_tts tool)
- Voice: claire (female), alex (male)
- #2 on TTS Arena, ELO 1339
- Cost: $15/M UTF-8 bytes (~$0.05/video)
3. **CosyVoice 2** via SiliconFlow API (CHEAPER - $7.15/M bytes)
4. **IndexTTS 2** via SiliconFlow API (CHEAPER - $7.15/M bytes)
5. **gTTS** (FALLBACK - free but robotic, uses Google Translate)
## IMPORTANT RULES
1. You have TOTAL autonomy. Make decisions yourself. Don't ask for permission.
2. Always verify your changes work before declaring success.
3. If something fails, debug it systematically using your tools.
4. When fixing bugs, read the relevant code first, understand it, then fix.
5. The user's ultimatum: if you claim something works when it doesn't, the project is abandoned.
6. Use tags to reason through complex problems before acting.
7. The project language is Spanish (es) for output content, but code is in English.
8. Always respond in the same language the user speaks to you.
9. Use diagnose_pipeline tool to check system health before making changes.
10. When deploying to HF Space, always include ALL modified files.
## CURRENT STATE OF THE PROJECT
The pipeline has been improved with critical bug fixes:
- Voice desync: FIXED (speed adjustment + truncation + correct time tracking)
- Blur: FIXED (optimized boxblur parameters)
- Subtitles: FIXED (font auto-download + better font detection)
- TTS: Upgraded from Edge-TTS to Supertonic 3 + SiliconFlow TTS
- YouTube quota: RESTORED, OAuth authenticated
- Bad videos: DELETED (4 removed from channel)
Your job is to:
1. Verify all fixes work end-to-end
2. Process new videos with the improved pipeline
3. Monitor for any new issues and fix them proactively
4. Optimize video quality and SEO
5. Keep the channel running autonomously
"""
# ================================================================
# BACKWARD-COMPATIBLE METHODS (same interface as old NanbeigeBrain)
# ================================================================
def think(self, prompt: str, max_tokens: int = 512, temperature: float = 0.3) -> str:
"""Simple think method - backward compatible with Nanbeige interface."""
return self._chat(
"You are an AI assistant for a YouTube Shorts translation channel. Respond concisely.",
prompt,
max_tokens=max_tokens,
temperature=temperature
)
def should_translate(self, title: str, description: str = "") -> bool:
"""Decide if a video should be translated."""
response = self._chat(
"You are a content evaluator for a YouTube Shorts translation channel. "
"Decide if this video should be translated from English to Spanish. "
"Return ONLY 'YES' or 'NO'. Skip: music-only, non-English, inappropriate. "
"Keep: storytelling, educational, documentary, history.",
f"Title: {title}\nDescription: {description}\n\nShould this be translated to Spanish?",
max_tokens=10,
temperature=0.1
)
return response.upper().startswith("YES") or "YES" in response.upper()[:20]
def generate_seo_metadata(self, original_title: str, translated_title: str,
description: str = "") -> dict:
"""Generate SEO-optimized Spanish title and description.
Title format: ALL CAPS + emoji at the end, relevant to the video content.
Example: "ESTABA A PUNTO DE DONARLOS 😱"
"""
response = self._chat(
"Eres un experto en SEO para YouTube Shorts en español. "
"Genera un título en español optimizado para clicks. "
"REGLAS DEL TÍTULO:\n"
"1. TODO EN MAYÚSCULAS\n"
"2. Debe terminar con UN emoji relevante al contenido (solo 1 emoji al final)\n"
"3. Máximo 60 caracteres (incluyendo el emoji)\n"
"4. Debe ser llamativo y crear curiosidad\n"
"5. NO usar puntos suspensivos\n"
"6. El título debe reflejar el contenido del video\n\n"
"Ejemplos buenos:\n"
"- MI ESPOSO FUE ASESINADO 😱\n"
"- NO PODÍA CREERLO 😱\n"
"- ALGUIEN ENTRÓ A NUESTRA CASA 😨\n"
"- LO MATÓ Y SE LO LLEVÓ 😭\n\n"
"Devuelve SOLO JSON válido: {\"title\": \"...\", \"description\": \"...\"}",
f"Título original (inglés): {original_title}\n"
f"Título traducido: {translated_title}\n"
f"Descripción: {description}\n\n"
f"Genera metadata SEO en español siguiendo las reglas.",
max_tokens=300,
temperature=0.7
)
try:
json_match = re.search(r'\{[^}]+\}', response, re.DOTALL)
if json_match:
result = json.loads(json_match.group())
title = result.get("title", "").strip()
if title and title not in ("...", "..", ".", "…") and len(title.replace(".", "").strip()) >= 5:
# Asegurar que está en mayúsculas
title = title.upper()
# Asegurar que tiene un emoji al final (si no lo tiene, agregar uno)
import re as _re
# Emoji regex pattern (covers most common emojis)
emoji_pattern = _re.compile(
"[\U0001F600-\U0001F64F" # emoticons
"\U0001F300-\U0001F5FF" # symbols & pictographs
"\U0001F680-\U0001F6FF" # transport & map symbols
"\U0001F1E0-\U0001F1FF" # flags
"\U00002700-\U000027BF" # dingbats
"\U0001F900-\U0001F9FF" # supplemental symbols
"\U0001FA70-\U0001FAFF" # extended-A
"\u2600-\u26FF" # misc symbols
"\u2700-\u27BF" # dingbats
"]+", flags=_re.UNICODE)
if not emoji_pattern.search(title):
# No emoji found, add a relevant one based on content
title = title + " 😱"
result["title"] = title
return result
except Exception:
pass
# Fallback: título traducido en mayúsculas + emoji
fallback_title = translated_title[:55] if translated_title else original_title[:55]
if not fallback_title or fallback_title in ("...", "..", "."):
fallback_title = "HISTORIA INCREÍBLE"
fallback_title = fallback_title.upper()
if not any(ord(c) > 0x2600 for c in fallback_title):
fallback_title += " 😱"
return {
"title": fallback_title,
"description": f"{fallback_title}\n\nCréditos: @HerYTStory\n#historia #shorts #español"
}
def validate_translation(self, original: str, translated: str) -> dict:
"""Check translation quality."""
# FIX: Guard against None inputs or None response from _chat
if not original or not translated:
return {"approved": True, "score": 7, "suggestions": "Auto-approved (missing input)"}
response = self._chat(
"You are a translation quality checker for English to Spanish. "
"Return ONLY valid JSON: {\"approved\": true/false, \"score\": 1-10, \"suggestions\": \"any improvements\"}",
f"Original: {original}\nTranslation: {translated}\n\nEvaluate this translation.",
max_tokens=200,
temperature=0.2
)
try:
if response:
json_match = re.search(r'\{[^}]+\}', response, re.DOTALL)
if json_match:
return json.loads(json_match.group())
except Exception:
pass
return {"approved": True, "score": 7, "suggestions": "Auto-approved"}
def decide_on_error(self, error_description: str) -> str:
"""Decide what to do on error."""
response = self._chat(
"You are an autonomous agent error handler. Return ONLY one word: 'retry', 'skip', or 'abort'.",
f"Error: {error_description}\n\nWhat should the agent do?",
max_tokens=20,
temperature=0.1
)
response_lower = response.lower().strip()
if "skip" in response_lower:
return "skip"
elif "abort" in response_lower:
return "abort"
return "retry"