Spaces:

zaheer140
/

liveChatBotBackend

Runtime error

liveChatBotBackend / SuperAgent /main.py

Update SuperAgent/main.py

875af83 verified 5 months ago

21.9 kB

import os
import sys
import re
import httpx
from pathlib import Path

# Add project root and sub-agent directories to sys.path to resolve dynamic package namespaces (Trigger Reload)
root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if root_dir not in sys.path:
    sys.path.insert(0, root_dir)

sys.path.append(os.path.join(root_dir, "agents", "live"))
sys.path.append(os.path.join(root_dir, "agents", "analysis", "backend"))
sys.path.append(os.path.join(root_dir, "agents", "anomaly", "backend"))
sys.path.append(os.path.join(root_dir, "agents", "telemetry"))
sys.path.append(os.path.join(root_dir, "agents", "report"))

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv

env_path = Path(__file__).parent.parent / ".env"
load_dotenv(dotenv_path=env_path)

from app.models.schemas import ChatRequest, ChatResponse
from app.services.router import router_service

from fastapi.responses import RedirectResponse, FileResponse
from fastapi.staticfiles import StaticFiles

app = FastAPI(title="Visibility AI Multi-Agent Gateway", version="0.1.0")

# Import sub-agents applications
from agents.live.app.main import app as live_app
from agents.analysis.backend.app.main import app as analysis_app
from agents.anomaly.backend.api_chat import app as anomaly_app
from agents.telemetry.main import app as telemetry_app
from agents.report.main import app as report_app

# Mount sub-agents natively on their respective sub-paths
app.mount("/live", live_app)
app.mount("/analysis", analysis_app)
app.mount("/anomaly", anomaly_app)
app.mount("/telemetry", telemetry_app)
app.mount("/report", report_app)

# Custom route to force download report files or preview in browser
@app.get("/reports/{filename}")
async def download_report(filename: str):
    
    filepath = os.path.join("scratch/reports", filename)
    if not os.path.exists(filepath):
        raise HTTPException(status_code=404, detail="File not found")
    
    if filename.endswith(".pdf"):
        # Serve PDF inline for browser previewing in a new tab
        return FileResponse(
            path=filepath,
            media_type="application/pdf",
            headers={"Content-Disposition": f"inline; filename={filename}"}
        )
    else:
        # Serve spreadsheet as a forced direct download
        media_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" if filename.endswith(".xlsx") else "application/octet-stream"
        return FileResponse(
            path=filepath,
            filename=filename,
            media_type=media_type
        )

# Serve report files statically (fallback)
os.makedirs("scratch/reports", exist_ok=True)
app.mount("/reports", StaticFiles(directory="scratch/reports"), name="reports")


DEFAULT_USER_ID = os.getenv("DEFAULT_USER_ID", "ad7103e4-9595-4000-a56a-8f650e3eddef")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/", tags=["Root"])
async def read_index():
    return {
        "status": "online",
        "service": "Multi-Agent Telemetry SuperAgent API",
        "integrations": {
            "live_agent": "Connected [OK]",
            "analysis_agent": "Connected [OK]",
            "anomaly_agent": "Connected [OK]",
            "postgresql_db": "Connected [OK]"
        },
        "api_endpoints_guide": {
            "/health": "GET - Check server status.",
            "/api/v1/chat/message": "POST - Unified AI chat endpoint (Live/Analysis/Anomaly).",
            "/api/v1/assets": "GET - List authorized assets for a user.",
            "/api/v1/schedules": "GET - List available automated report types.",
            "/api/v1/schedules/run/{id}": "POST - Manually trigger a specific report.",
            "/api/v1/anomalies": "GET - Real-time summary of all active incidents.",
            "/docs": "GET - Interactive Swagger UI documentation."
        },
        "message": "Backend is running independently. Connect your frontend to this URL."
    }

@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}

@app.get("/api/v1/assets")
async def list_assets(user_id: str = None):
    from app.services.asset_registry import asset_registry
    assets = asset_registry.list_all_assets(user_id=user_id or DEFAULT_USER_ID)
    return assets

@app.get("/api/v1/assets/{asset_id}")
async def get_asset(asset_id: str, user_id: str = None):
    from app.services.asset_registry import asset_registry
    asset = asset_registry.get_asset_by_id(asset_id, user_id=user_id or DEFAULT_USER_ID)
    if not asset:
        raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found or not authorized.")
    return asset

@app.get("/debug/logs")
async def get_logs():
    logs = {"cwd": os.getcwd()}
    for agent in ["startup", "live_agent", "analysis_agent", "anomaly_agent"]:
        log_path = f"/app/{agent}.log"
        if os.path.exists(log_path):
            with open(log_path, "r") as f:
                logs[agent] = f.read()[-2000:] # Last 2000 chars
        else:
            # Fallback to relative path if /app/ doesn't exist
            rel_path = f"../{agent}.log"
            if os.path.exists(rel_path):
                 with open(rel_path, "r") as f:
                    logs[agent] = f.read()[-2000:]
            else:
                logs[agent] = f"Log file not found at {log_path} or {rel_path}"
    return logs

@app.post("/api/v1/chat/message", response_model=ChatResponse)
async def chat_message(payload: ChatRequest) -> ChatResponse:
    print(f"DEBUG: Received request: {payload}")
    try:
        return await router_service.route_request(payload)
    except Exception as exc:
        import traceback
        error_msg = f"{type(exc).__name__}: {str(exc)}"
        tb = traceback.format_exc()
        print(f"ERROR: {error_msg}\n{tb}")
        raise HTTPException(status_code=500, detail=f"Internal error: {error_msg}\n{tb[:200]}") from exc

@app.get("/api/v1/schedules")
async def list_schedules():
    # Hugging Face injects SPACE_ID. If present, force port to 7860. Otherwise default to 8000.
    port = "7860" if os.getenv("SPACE_ID") else os.getenv("PORT", "8000")
    analysis_agent_url = os.getenv("ANALYSIS_AGENT_URL", f"http://127.0.0.1:{port}/analysis")
    if "localhost" in analysis_agent_url or "127.0.0.1" in analysis_agent_url:
        analysis_agent_url = analysis_agent_url.replace("localhost", "127.0.0.1")
        analysis_agent_url = re.sub(r":\d+", f":{port}", analysis_agent_url)
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(f"{analysis_agent_url}/api/v1/schedules")
        return response.json()

@app.post("/api/v1/schedules/run/{schedule_id}")
async def run_schedule(schedule_id: str, offset: int = 0, span_days: int = 1):
    # Hugging Face injects SPACE_ID. If present, force port to 7860. Otherwise default to 8000.
    port = "7860" if os.getenv("SPACE_ID") else os.getenv("PORT", "8000")
    analysis_agent_url = os.getenv("ANALYSIS_AGENT_URL", f"http://127.0.0.1:{port}/analysis")
    if "localhost" in analysis_agent_url or "127.0.0.1" in analysis_agent_url:
        analysis_agent_url = analysis_agent_url.replace("localhost", "127.0.0.1")
        analysis_agent_url = re.sub(r":\d+", f":{port}", analysis_agent_url)
    async with httpx.AsyncClient(timeout=60.0) as client:
        response = await client.post(
            f"{analysis_agent_url}/api/v1/schedules/run/{schedule_id}",
            params={"offset": offset, "span_days": span_days}
        )
        return response.json()

@app.get("/api/v1/anomalies")
async def get_active_anomalies(user_id: str = None, schema_name: str = None):
    # Hugging Face injects SPACE_ID. If present, force port to 7860. Otherwise default to 8000.
    port = "7860" if os.getenv("SPACE_ID") else os.getenv("PORT", "8000")
    anomaly_agent_url = os.getenv("ANOMALY_AGENT_URL", f"http://127.0.0.1:{port}/anomaly")
    if "localhost" in anomaly_agent_url or "127.0.0.1" in anomaly_agent_url:
        anomaly_agent_url = anomaly_agent_url.replace("localhost", "127.0.0.1")
        anomaly_agent_url = re.sub(r":\d+", f":{port}", anomaly_agent_url)
    async with httpx.AsyncClient(timeout=300.0) as client:
        response = await client.get(
            f"{anomaly_agent_url}/api/v1/anomalies",
            params={
                "user_id": user_id or DEFAULT_USER_ID,
                "schema_name": schema_name
            }
        )
        return response.json()

@app.get("/api/v1/reports/daily")
async def get_daily_reports():
    import csv
    import ast
    import sys
    csv.field_size_limit(2147483647)
    from pathlib import Path
    
    reports = {"coldrooms": [], "tanks": [], "smoke_alarms": []}
    base_dir = Path(__file__).parent.parent
    cr_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "coldroom_reports.csv"
    tank_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "tank_reports.csv"
    smoke_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "smoke_reports.csv"
    
    def parse_csv(file_path, target_type):
        if not file_path.exists(): return []
        with open(file_path, mode='r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            all_rows = list(reader)
        if not all_rows: return []
        filtered_rows = [row for row in all_rows if row.get('report_type') == target_type]
        if not filtered_rows: return []
        latest_time = max(row['report_timestamp'] for row in filtered_rows)
        latest_rows = [row for row in filtered_rows if row['report_timestamp'] == latest_time]
        for row in latest_rows:
            if 'anomalies' in row:
                try: row['anomalies'] = ast.literal_eval(row['anomalies'])
                except: pass
        return latest_rows

    reports["coldrooms"] = parse_csv(cr_path, "daily")
    reports["tanks"] = parse_csv(tank_path, "daily")
    reports["smoke_alarms"] = parse_csv(smoke_path, "daily")
    return reports

@app.get("/api/v1/reports/latest")
async def get_latest_rolling_reports():
    import csv
    import ast
    import sys
    csv.field_size_limit(2147483647)
    from pathlib import Path
    
    reports = {"coldrooms": [], "tanks": [], "smoke_alarms": []}
    base_dir = Path(__file__).parent.parent
    cr_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "coldroom_reports.csv"
    tank_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "tank_reports.csv"
    smoke_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "smoke_reports.csv"
    
    def parse_csv(file_path, target_type):
        if not file_path.exists(): return []
        with open(file_path, mode='r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            all_rows = list(reader)
        if not all_rows: return []
        filtered_rows = [row for row in all_rows if row.get('report_type') == target_type]
        if not filtered_rows: return []
        latest_time = max(row['report_timestamp'] for row in filtered_rows)
        latest_rows = [row for row in filtered_rows if row['report_timestamp'] == latest_time]
        for row in latest_rows:
            if 'anomalies' in row:
                try: row['anomalies'] = ast.literal_eval(row['anomalies'])
                except: pass
        return latest_rows

    reports["coldrooms"] = parse_csv(cr_path, "rolling")
    reports["tanks"] = parse_csv(tank_path, "rolling")
    reports["smoke_alarms"] = parse_csv(smoke_path, "rolling")
    return reports

import uuid
import json
import datetime
import time
import threading
from pydantic import BaseModel
from typing import Optional

class ManualReportRequest(BaseModel):
    user_id: Optional[str] = None
    asset_id: str
    timeframe: str
    email: Optional[str] = None

class ScheduleSaveRequest(BaseModel):
    user_id: Optional[str] = None
    asset_id: str
    frequency: str
    schedule_time: str
    weekly_day: Optional[str] = None
    recipient_email: str
    is_active: bool = True

SCHEDULES_FILE = Path("scratch/report_schedules.json")

def load_schedules():
    if not SCHEDULES_FILE.exists():
        return {"schedules": []}
    try:
        with open(SCHEDULES_FILE, "r") as f:
            return json.load(f)
    except Exception:
        return {"schedules": []}

def save_schedules(data):
    SCHEDULES_FILE.parent.mkdir(parents=True, exist_ok=True)
    try:
        with open(SCHEDULES_FILE, "w") as f:
            json.dump(data, f, indent=2)
    except Exception as e:
        print(f"Error saving schedules: {e}")

def trigger_automated_report(user_id: str, asset_id: str, recipient_email: str, frequency: str):
    try:
        from app.services.asset_registry import asset_registry
        from agents.report.app.services.report_generator import report_generator_service
        from app.services.email_service import email_service
        
        # 1. Resolve asset
        asset = asset_registry.get_asset_by_id(asset_id, user_id=user_id)
        asset_name = asset["name"] if asset else asset_id
        
        # Determine timeframe
        timeframe = "yesterday" if frequency == "daily" else "last week"
        
        print(f"SCHEDULER: Generating {frequency} report for {asset_name}...")
        res = report_generator_service.generate_report(
            user_id=user_id,
            asset_query=f"{asset_name} {timeframe}",
            format_type="pdf"
        )
        
        if res.get("success") and recipient_email:
            filepath = res.get("filepath")
            subject = f"Automated {frequency.capitalize()} Report: {asset_name}"
            body_html = f"""
            <html>
                <body>
                    <h2 style="color: #1A365D;">Automated Operational Report</h2>
                    <p>Dear Operator,</p>
                    <p>Please find attached the automated <b>{frequency}</b> PDF report for <b>{asset_name}</b>.</p>
                    <p><b>Reporting Timeframe:</b> {timeframe.capitalize()}</p>
                    <hr/>
                    <p style="font-size: 11px; color: #718096;">This is an automated notification. Please do not reply directly to this email.</p>
                </body>
            </html>
            """
            email_service.send_report(
                recipient=recipient_email,
                subject=subject,
                body_html=body_html,
                attachment_path=filepath
            )
            print(f"SCHEDULER: Automated report sent to {recipient_email}")
        else:
            print(f"SCHEDULER: Report generation failed or no email configured. Success: {res.get('success')}")
    except Exception as e:
        print(f"SCHEDULER ERROR: Failed to run automated report: {e}")

# def run_background_scheduler():
#     print("Background report scheduler loop running...")
#     while True:
#         try:
#             time.sleep(30)
#             data = load_schedules()
#             schedules = data.get("schedules", [])
#             updated = False
#             
#             now = datetime.datetime.now()
#             today_str = now.strftime("%Y-%m-%d")
#             current_time_str = now.strftime("%H:%M")
#             current_day_name = now.strftime("%A")
#             
#             for s in schedules:
#                 if not s.get("is_active", True):
#                     continue
#                 
#                 frequency = s.get("frequency", "daily").lower()
#                 schedule_time = s.get("schedule_time", "09:00")
#                 weekly_day = s.get("weekly_day")
#                 last_run = s.get("last_run")
#                 
#                 should_run = False
#                 if current_time_str == schedule_time:
#                     if frequency == "daily":
#                         if last_run != today_str:
#                             should_run = True
#                     elif frequency == "weekly":
#                         if weekly_day and weekly_day.lower() == current_day_name.lower():
#                             if last_run != today_str:
#                                 should_run = True
#                 
#                 if should_run:
#                     asset_id = s.get("asset_id")
#                     user_id = s.get("user_id") or "ad7103e4-9595-4000-a56a-8f650e3eddef"
#                     recipient_email = s.get("recipient_email")
#                     
#                     threading.Thread(
#                         target=trigger_automated_report,
#                         args=(user_id, asset_id, recipient_email, frequency)
#                     ).start()
#                     
#                     s["last_run"] = today_str
#                     updated = True
#             
#             if updated:
#                 save_schedules(data)
#         except Exception as e:
#             print(f"Error in background scheduler loop: {e}")
# 
# @app.on_event("startup")
# async def start_scheduler_on_startup():
#     threading.Thread(target=run_background_scheduler, daemon=True).start()

@app.post("/api/v1/analysis/schedule/manual")
async def manual_report_endpoint(payload: ManualReportRequest):
    try:
        from app.services.asset_registry import asset_registry
        from agents.report.app.services.report_generator import report_generator_service
        from app.services.email_service import email_service
        
        user_id = payload.user_id or DEFAULT_USER_ID
        asset = asset_registry.get_asset_by_id(payload.asset_id, user_id=user_id)
        if not asset:
            raise HTTPException(status_code=404, detail="Asset not found")
            
        asset_name = asset["name"]
        asset_query = f"{asset_name} {payload.timeframe}"
        
        res = report_generator_service.generate_report(
            user_id=user_id,
            asset_query=asset_query,
            format_type="pdf"
        )
        
        if not res.get("success"):
            raise HTTPException(status_code=400, detail=res.get("answer", "Failed to generate report"))
            
        if payload.email:
            filepath = res.get("filepath")
            subject = f"Manual Telemetry Report: {asset_name}"
            body_html = f"""
            <html>
                <body>
                    <h2 style="color: #1A365D;">Manual Telemetry Report</h2>
                    <p>Hello,</p>
                    <p>You requested a manual report for <b>{asset_name}</b>.</p>
                    <p><b>Timeframe:</b> {payload.timeframe.capitalize()}</p>
                    <hr/>
                    <p style="font-size: 11px; color: #718096;">Generated by Visbot Multi-Agent Gateway.</p>
                </body>
            </html>
            """
            email_service.send_report(
                recipient=payload.email,
                subject=subject,
                body_html=body_html,
                attachment_path=filepath
            )
            
        return {
            "success": True,
            "download_url": res.get("download_url"),
            "filename": res.get("filename")
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/v1/analysis/schedule/save")
async def save_schedule_endpoint(payload: ScheduleSaveRequest):
    try:
        user_id = payload.user_id or DEFAULT_USER_ID
        data = load_schedules()
        schedules = data.setdefault("schedules", [])
        
        existing = None
        for s in schedules:
            if s.get("asset_id") == payload.asset_id:
                existing = s
                break
                
        if existing:
            existing["frequency"] = payload.frequency
            existing["schedule_time"] = payload.schedule_time
            existing["weekly_day"] = payload.weekly_day
            existing["recipient_email"] = payload.recipient_email
            existing["is_active"] = payload.is_active
            existing["last_run"] = None
        else:
            new_schedule = {
                "id": str(uuid.uuid4()),
                "user_id": user_id,
                "asset_id": payload.asset_id,
                "frequency": payload.frequency,
                "schedule_time": payload.schedule_time,
                "weekly_day": payload.weekly_day,
                "recipient_email": payload.recipient_email,
                "is_active": payload.is_active,
                "last_run": None
            }
            schedules.append(new_schedule)
            
        save_schedules(data)
        return {"success": True, "message": "Schedule saved successfully."}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/v1/analysis/schedule/{asset_id}")
async def get_schedule_endpoint(asset_id: str):
    try:
        data = load_schedules()
        schedules = data.get("schedules", [])
        for s in schedules:
            if s.get("asset_id") == asset_id:
                return s
        return {"is_active": False}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

	import os
	import sys
	import re
	import httpx
	from pathlib import Path

	# Add project root and sub-agent directories to sys.path to resolve dynamic package namespaces (Trigger Reload)
	root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
	if root_dir not in sys.path:
	sys.path.insert(0, root_dir)

	sys.path.append(os.path.join(root_dir, "agents", "live"))
	sys.path.append(os.path.join(root_dir, "agents", "analysis", "backend"))
	sys.path.append(os.path.join(root_dir, "agents", "anomaly", "backend"))
	sys.path.append(os.path.join(root_dir, "agents", "telemetry"))
	sys.path.append(os.path.join(root_dir, "agents", "report"))

	from fastapi import FastAPI, HTTPException
	from fastapi.middleware.cors import CORSMiddleware
	from dotenv import load_dotenv

	env_path = Path(__file__).parent.parent / ".env"
	load_dotenv(dotenv_path=env_path)

	from app.models.schemas import ChatRequest, ChatResponse
	from app.services.router import router_service

	from fastapi.responses import RedirectResponse, FileResponse
	from fastapi.staticfiles import StaticFiles

	app = FastAPI(title="Visibility AI Multi-Agent Gateway", version="0.1.0")

	# Import sub-agents applications
	from agents.live.app.main import app as live_app
	from agents.analysis.backend.app.main import app as analysis_app
	from agents.anomaly.backend.api_chat import app as anomaly_app
	from agents.telemetry.main import app as telemetry_app
	from agents.report.main import app as report_app

	# Mount sub-agents natively on their respective sub-paths
	app.mount("/live", live_app)
	app.mount("/analysis", analysis_app)
	app.mount("/anomaly", anomaly_app)
	app.mount("/telemetry", telemetry_app)
	app.mount("/report", report_app)

	# Custom route to force download report files or preview in browser
	@app.get("/reports/{filename}")
	async def download_report(filename: str):

	filepath = os.path.join("scratch/reports", filename)
	if not os.path.exists(filepath):
	raise HTTPException(status_code=404, detail="File not found")

	if filename.endswith(".pdf"):
	# Serve PDF inline for browser previewing in a new tab
	return FileResponse(
	path=filepath,
	media_type="application/pdf",
	headers={"Content-Disposition": f"inline; filename={filename}"}
	)
	else:
	# Serve spreadsheet as a forced direct download
	media_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" if filename.endswith(".xlsx") else "application/octet-stream"
	return FileResponse(
	path=filepath,
	filename=filename,
	media_type=media_type
	)

	# Serve report files statically (fallback)
	os.makedirs("scratch/reports", exist_ok=True)
	app.mount("/reports", StaticFiles(directory="scratch/reports"), name="reports")


	DEFAULT_USER_ID = os.getenv("DEFAULT_USER_ID", "ad7103e4-9595-4000-a56a-8f650e3eddef")

	app.add_middleware(
	CORSMiddleware,
	allow_origins=["*"],
	allow_credentials=True,
	allow_methods=["*"],
	allow_headers=["*"],
	)

	@app.get("/", tags=["Root"])
	async def read_index():
	return {
	"status": "online",
	"service": "Multi-Agent Telemetry SuperAgent API",
	"integrations": {
	"live_agent": "Connected [OK]",
	"analysis_agent": "Connected [OK]",
	"anomaly_agent": "Connected [OK]",
	"postgresql_db": "Connected [OK]"
	},
	"api_endpoints_guide": {
	"/health": "GET - Check server status.",
	"/api/v1/chat/message": "POST - Unified AI chat endpoint (Live/Analysis/Anomaly).",
	"/api/v1/assets": "GET - List authorized assets for a user.",
	"/api/v1/schedules": "GET - List available automated report types.",
	"/api/v1/schedules/run/{id}": "POST - Manually trigger a specific report.",
	"/api/v1/anomalies": "GET - Real-time summary of all active incidents.",
	"/docs": "GET - Interactive Swagger UI documentation."
	},
	"message": "Backend is running independently. Connect your frontend to this URL."
	}

	@app.get("/health")
	def health() -> dict[str, str]:
	return {"status": "ok"}

	@app.get("/api/v1/assets")
	async def list_assets(user_id: str = None):
	from app.services.asset_registry import asset_registry
	assets = asset_registry.list_all_assets(user_id=user_id or DEFAULT_USER_ID)
	return assets

	@app.get("/api/v1/assets/{asset_id}")
	async def get_asset(asset_id: str, user_id: str = None):
	from app.services.asset_registry import asset_registry
	asset = asset_registry.get_asset_by_id(asset_id, user_id=user_id or DEFAULT_USER_ID)
	if not asset:
	raise HTTPException(status_code=404, detail=f"Asset {asset_id} not found or not authorized.")
	return asset

	@app.get("/debug/logs")
	async def get_logs():
	logs = {"cwd": os.getcwd()}
	for agent in ["startup", "live_agent", "analysis_agent", "anomaly_agent"]:
	log_path = f"/app/{agent}.log"
	if os.path.exists(log_path):
	with open(log_path, "r") as f:
	logs[agent] = f.read()[-2000:] # Last 2000 chars
	else:
	# Fallback to relative path if /app/ doesn't exist
	rel_path = f"../{agent}.log"
	if os.path.exists(rel_path):
	with open(rel_path, "r") as f:
	logs[agent] = f.read()[-2000:]
	else:
	logs[agent] = f"Log file not found at {log_path} or {rel_path}"
	return logs

	@app.post("/api/v1/chat/message", response_model=ChatResponse)
	async def chat_message(payload: ChatRequest) -> ChatResponse:
	print(f"DEBUG: Received request: {payload}")
	try:
	return await router_service.route_request(payload)
	except Exception as exc:
	import traceback
	error_msg = f"{type(exc).__name__}: {str(exc)}"
	tb = traceback.format_exc()
	print(f"ERROR: {error_msg}\n{tb}")
	raise HTTPException(status_code=500, detail=f"Internal error: {error_msg}\n{tb[:200]}") from exc

	@app.get("/api/v1/schedules")
	async def list_schedules():
	# Hugging Face injects SPACE_ID. If present, force port to 7860. Otherwise default to 8000.
	port = "7860" if os.getenv("SPACE_ID") else os.getenv("PORT", "8000")
	analysis_agent_url = os.getenv("ANALYSIS_AGENT_URL", f"http://127.0.0.1:{port}/analysis")
	if "localhost" in analysis_agent_url or "127.0.0.1" in analysis_agent_url:
	analysis_agent_url = analysis_agent_url.replace("localhost", "127.0.0.1")
	analysis_agent_url = re.sub(r":\d+", f":{port}", analysis_agent_url)
	async with httpx.AsyncClient(timeout=10.0) as client:
	response = await client.get(f"{analysis_agent_url}/api/v1/schedules")
	return response.json()

	@app.post("/api/v1/schedules/run/{schedule_id}")
	async def run_schedule(schedule_id: str, offset: int = 0, span_days: int = 1):
	# Hugging Face injects SPACE_ID. If present, force port to 7860. Otherwise default to 8000.
	port = "7860" if os.getenv("SPACE_ID") else os.getenv("PORT", "8000")
	analysis_agent_url = os.getenv("ANALYSIS_AGENT_URL", f"http://127.0.0.1:{port}/analysis")
	if "localhost" in analysis_agent_url or "127.0.0.1" in analysis_agent_url:
	analysis_agent_url = analysis_agent_url.replace("localhost", "127.0.0.1")
	analysis_agent_url = re.sub(r":\d+", f":{port}", analysis_agent_url)
	async with httpx.AsyncClient(timeout=60.0) as client:
	response = await client.post(
	f"{analysis_agent_url}/api/v1/schedules/run/{schedule_id}",
	params={"offset": offset, "span_days": span_days}
	)
	return response.json()

	@app.get("/api/v1/anomalies")
	async def get_active_anomalies(user_id: str = None, schema_name: str = None):
	# Hugging Face injects SPACE_ID. If present, force port to 7860. Otherwise default to 8000.
	port = "7860" if os.getenv("SPACE_ID") else os.getenv("PORT", "8000")
	anomaly_agent_url = os.getenv("ANOMALY_AGENT_URL", f"http://127.0.0.1:{port}/anomaly")
	if "localhost" in anomaly_agent_url or "127.0.0.1" in anomaly_agent_url:
	anomaly_agent_url = anomaly_agent_url.replace("localhost", "127.0.0.1")
	anomaly_agent_url = re.sub(r":\d+", f":{port}", anomaly_agent_url)
	async with httpx.AsyncClient(timeout=300.0) as client:
	response = await client.get(
	f"{anomaly_agent_url}/api/v1/anomalies",
	params={
	"user_id": user_id or DEFAULT_USER_ID,
	"schema_name": schema_name
	}
	)
	return response.json()

	@app.get("/api/v1/reports/daily")
	async def get_daily_reports():
	import csv
	import ast
	import sys
	csv.field_size_limit(2147483647)
	from pathlib import Path

	reports = {"coldrooms": [], "tanks": [], "smoke_alarms": []}
	base_dir = Path(__file__).parent.parent
	cr_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "coldroom_reports.csv"
	tank_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "tank_reports.csv"
	smoke_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "smoke_reports.csv"

	def parse_csv(file_path, target_type):
	if not file_path.exists(): return []
	with open(file_path, mode='r', encoding='utf-8') as f:
	reader = csv.DictReader(f)
	all_rows = list(reader)
	if not all_rows: return []
	filtered_rows = [row for row in all_rows if row.get('report_type') == target_type]
	if not filtered_rows: return []
	latest_time = max(row['report_timestamp'] for row in filtered_rows)
	latest_rows = [row for row in filtered_rows if row['report_timestamp'] == latest_time]
	for row in latest_rows:
	if 'anomalies' in row:
	try: row['anomalies'] = ast.literal_eval(row['anomalies'])
	except: pass
	return latest_rows

	reports["coldrooms"] = parse_csv(cr_path, "daily")
	reports["tanks"] = parse_csv(tank_path, "daily")
	reports["smoke_alarms"] = parse_csv(smoke_path, "daily")
	return reports

	@app.get("/api/v1/reports/latest")
	async def get_latest_rolling_reports():
	import csv
	import ast
	import sys
	csv.field_size_limit(2147483647)
	from pathlib import Path

	reports = {"coldrooms": [], "tanks": [], "smoke_alarms": []}
	base_dir = Path(__file__).parent.parent
	cr_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "coldroom_reports.csv"
	tank_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "tank_reports.csv"
	smoke_path = base_dir / "agents" / "anomaly" / "backend" / "data" / "reports" / "smoke_reports.csv"

	def parse_csv(file_path, target_type):
	if not file_path.exists(): return []
	with open(file_path, mode='r', encoding='utf-8') as f:
	reader = csv.DictReader(f)
	all_rows = list(reader)
	if not all_rows: return []
	filtered_rows = [row for row in all_rows if row.get('report_type') == target_type]
	if not filtered_rows: return []
	latest_time = max(row['report_timestamp'] for row in filtered_rows)
	latest_rows = [row for row in filtered_rows if row['report_timestamp'] == latest_time]
	for row in latest_rows:
	if 'anomalies' in row:
	try: row['anomalies'] = ast.literal_eval(row['anomalies'])
	except: pass
	return latest_rows

	reports["coldrooms"] = parse_csv(cr_path, "rolling")
	reports["tanks"] = parse_csv(tank_path, "rolling")
	reports["smoke_alarms"] = parse_csv(smoke_path, "rolling")
	return reports

	import uuid
	import json
	import datetime
	import time
	import threading
	from pydantic import BaseModel
	from typing import Optional

	class ManualReportRequest(BaseModel):
	user_id: Optional[str] = None
	asset_id: str
	timeframe: str
	email: Optional[str] = None

	class ScheduleSaveRequest(BaseModel):
	user_id: Optional[str] = None
	asset_id: str
	frequency: str
	schedule_time: str
	weekly_day: Optional[str] = None
	recipient_email: str
	is_active: bool = True

	SCHEDULES_FILE = Path("scratch/report_schedules.json")

	def load_schedules():
	if not SCHEDULES_FILE.exists():
	return {"schedules": []}
	try:
	with open(SCHEDULES_FILE, "r") as f:
	return json.load(f)
	except Exception:
	return {"schedules": []}

	def save_schedules(data):
	SCHEDULES_FILE.parent.mkdir(parents=True, exist_ok=True)
	try:
	with open(SCHEDULES_FILE, "w") as f:
	json.dump(data, f, indent=2)
	except Exception as e:
	print(f"Error saving schedules: {e}")

	def trigger_automated_report(user_id: str, asset_id: str, recipient_email: str, frequency: str):
	try:
	from app.services.asset_registry import asset_registry
	from agents.report.app.services.report_generator import report_generator_service
	from app.services.email_service import email_service

	# 1. Resolve asset
	asset = asset_registry.get_asset_by_id(asset_id, user_id=user_id)
	asset_name = asset["name"] if asset else asset_id

	# Determine timeframe
	timeframe = "yesterday" if frequency == "daily" else "last week"

	print(f"SCHEDULER: Generating {frequency} report for {asset_name}...")
	res = report_generator_service.generate_report(
	user_id=user_id,
	asset_query=f"{asset_name} {timeframe}",
	format_type="pdf"
	)

	if res.get("success") and recipient_email:
	filepath = res.get("filepath")
	subject = f"Automated {frequency.capitalize()} Report: {asset_name}"
	body_html = f"""
	<html>
	<body>
	<h2 style="color: #1A365D;">Automated Operational Report</h2>
	<p>Dear Operator,</p>
	<p>Please find attached the automated <b>{frequency}</b> PDF report for <b>{asset_name}</b>.</p>
	<p><b>Reporting Timeframe:</b> {timeframe.capitalize()}</p>
	<hr/>
	<p style="font-size: 11px; color: #718096;">This is an automated notification. Please do not reply directly to this email.</p>
	</body>
	</html>
	"""
	email_service.send_report(
	recipient=recipient_email,
	subject=subject,
	body_html=body_html,
	attachment_path=filepath
	)
	print(f"SCHEDULER: Automated report sent to {recipient_email}")
	else:
	print(f"SCHEDULER: Report generation failed or no email configured. Success: {res.get('success')}")
	except Exception as e:
	print(f"SCHEDULER ERROR: Failed to run automated report: {e}")

	# def run_background_scheduler():
	# print("Background report scheduler loop running...")
	# while True:
	# try:
	# time.sleep(30)
	# data = load_schedules()
	# schedules = data.get("schedules", [])
	# updated = False
	#
	# now = datetime.datetime.now()
	# today_str = now.strftime("%Y-%m-%d")
	# current_time_str = now.strftime("%H:%M")
	# current_day_name = now.strftime("%A")
	#
	# for s in schedules:
	# if not s.get("is_active", True):
	# continue
	#
	# frequency = s.get("frequency", "daily").lower()
	# schedule_time = s.get("schedule_time", "09:00")
	# weekly_day = s.get("weekly_day")
	# last_run = s.get("last_run")
	#
	# should_run = False
	# if current_time_str == schedule_time:
	# if frequency == "daily":
	# if last_run != today_str:
	# should_run = True
	# elif frequency == "weekly":
	# if weekly_day and weekly_day.lower() == current_day_name.lower():
	# if last_run != today_str:
	# should_run = True
	#
	# if should_run:
	# asset_id = s.get("asset_id")
	# user_id = s.get("user_id") or "ad7103e4-9595-4000-a56a-8f650e3eddef"
	# recipient_email = s.get("recipient_email")
	#
	# threading.Thread(
	# target=trigger_automated_report,
	# args=(user_id, asset_id, recipient_email, frequency)
	# ).start()
	#
	# s["last_run"] = today_str
	# updated = True
	#
	# if updated:
	# save_schedules(data)
	# except Exception as e:
	# print(f"Error in background scheduler loop: {e}")
	#
	# @app.on_event("startup")
	# async def start_scheduler_on_startup():
	# threading.Thread(target=run_background_scheduler, daemon=True).start()

	@app.post("/api/v1/analysis/schedule/manual")
	async def manual_report_endpoint(payload: ManualReportRequest):
	try:
	from app.services.asset_registry import asset_registry
	from agents.report.app.services.report_generator import report_generator_service
	from app.services.email_service import email_service

	user_id = payload.user_id or DEFAULT_USER_ID
	asset = asset_registry.get_asset_by_id(payload.asset_id, user_id=user_id)
	if not asset:
	raise HTTPException(status_code=404, detail="Asset not found")

	asset_name = asset["name"]
	asset_query = f"{asset_name} {payload.timeframe}"

	res = report_generator_service.generate_report(
	user_id=user_id,
	asset_query=asset_query,
	format_type="pdf"
	)

	if not res.get("success"):
	raise HTTPException(status_code=400, detail=res.get("answer", "Failed to generate report"))

	if payload.email:
	filepath = res.get("filepath")
	subject = f"Manual Telemetry Report: {asset_name}"
	body_html = f"""
	<html>
	<body>
	<h2 style="color: #1A365D;">Manual Telemetry Report</h2>
	<p>Hello,</p>
	<p>You requested a manual report for <b>{asset_name}</b>.</p>
	<p><b>Timeframe:</b> {payload.timeframe.capitalize()}</p>
	<hr/>
	<p style="font-size: 11px; color: #718096;">Generated by Visbot Multi-Agent Gateway.</p>
	</body>
	</html>
	"""
	email_service.send_report(
	recipient=payload.email,
	subject=subject,
	body_html=body_html,
	attachment_path=filepath
	)

	return {
	"success": True,
	"download_url": res.get("download_url"),
	"filename": res.get("filename")
	}
	except Exception as e:
	raise HTTPException(status_code=500, detail=str(e))

	@app.post("/api/v1/analysis/schedule/save")
	async def save_schedule_endpoint(payload: ScheduleSaveRequest):
	try:
	user_id = payload.user_id or DEFAULT_USER_ID
	data = load_schedules()
	schedules = data.setdefault("schedules", [])

	existing = None
	for s in schedules:
	if s.get("asset_id") == payload.asset_id:
	existing = s
	break

	if existing:
	existing["frequency"] = payload.frequency
	existing["schedule_time"] = payload.schedule_time
	existing["weekly_day"] = payload.weekly_day
	existing["recipient_email"] = payload.recipient_email
	existing["is_active"] = payload.is_active
	existing["last_run"] = None
	else:
	new_schedule = {
	"id": str(uuid.uuid4()),
	"user_id": user_id,
	"asset_id": payload.asset_id,
	"frequency": payload.frequency,
	"schedule_time": payload.schedule_time,
	"weekly_day": payload.weekly_day,
	"recipient_email": payload.recipient_email,
	"is_active": payload.is_active,
	"last_run": None
	}
	schedules.append(new_schedule)

	save_schedules(data)
	return {"success": True, "message": "Schedule saved successfully."}
	except Exception as e:
	raise HTTPException(status_code=500, detail=str(e))

	@app.get("/api/v1/analysis/schedule/{asset_id}")
	async def get_schedule_endpoint(asset_id: str):
	try:
	data = load_schedules()
	schedules = data.get("schedules", [])
	for s in schedules:
	if s.get("asset_id") == asset_id:
	return s
	return {"is_active": False}
	except Exception as e:
	raise HTTPException(status_code=500, detail=str(e))

	if __name__ == "__main__":
	import uvicorn
	uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)