app.py

"""
Minimal HF Space for testing ONLY (no bot logic).
Deploy this as app.py in a new, throwaway HF Space (SDK: Gradio).

requirements.txt for this Space:
    gradio
    httpx
"""

import asyncio
import json
import os
import threading
from datetime import datetime, timezone

import gradio as gr
import httpx

# --- CONFIG: edit these two lines --------------------------------------
NORTHFLANK_BASE_URL = os.environ.get("NORTHFLANK_TEST_URL", "https://your-service.northflank.app")
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")  # optional
INTERVAL_SECONDS = 300
# -------------------------------------------------------------------------

LOG_FILE = "connectivity_log.jsonl"
results_lock = threading.Lock()
latest_results = []  # kept in memory for the UI


async def check(name: str, url: str, timeout: float = 10.0) -> dict:
    entry = {"check": name, "url": url, "time_utc": datetime.now(timezone.utc).isoformat()}
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            resp = await client.get(url)
        entry["status"] = "ok" if resp.status_code == 200 else "http_error"
        entry["http_status"] = resp.status_code
    except Exception as e:
        entry["status"] = "connect_error"
        entry["error"] = f"{type(e).__name__}: {e}"
    return entry


async def run_once() -> list[dict]:
    checks = [
        check("hf_to_northflank_ping", f"{NORTHFLANK_BASE_URL}/ping"),
        check("hf_to_northflank_telegram_relay", f"{NORTHFLANK_BASE_URL}/check-telegram"),
    ]
    if TELEGRAM_BOT_TOKEN:
        checks.append(
            check("hf_to_telegram_direct", f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getMe")
        )
    return await asyncio.gather(*checks)


def append_log(entries: list[dict]) -> None:
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        for entry in entries:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")


def background_loop():
    async def loop():
        while True:
            results = await run_once()
            append_log(results)
            with results_lock:
                latest_results.clear()
                latest_results.extend(results)
            await asyncio.sleep(INTERVAL_SECONDS)
    asyncio.run(loop())


def get_status_text() -> str:
    with results_lock:
        if not latest_results:
            return "لسه ما بدأ أول فحص... انتظر بضع ثواني وحدّث الصفحة."
        lines = []
        for r in latest_results:
            line = f"**{r['check']}**: `{r['status']}`"
            if "http_status" in r:
                line += f" (HTTP {r['http_status']})"
            if "error" in r:
                line += f" — {r['error']}"
            line += f"  \n_{r['time_utc']}_"
            lines.append(line)
        return "\n\n".join(lines)


def get_log_tail() -> str:
    if not os.path.exists(LOG_FILE):
        return "لا يوجد سجل بعد."
    with open(LOG_FILE, "r", encoding="utf-8") as f:
        lines = f.readlines()
    return "".join(lines[-30:]) or "لا يوجد سجل بعد."


threading.Thread(target=background_loop, daemon=True).start()

with gr.Blocks(title="Connectivity Test: HF <-> Northflank <-> Telegram") as demo:
    gr.Markdown("## اختبار الاتصال الدوري: HF ↔ Northflank ↔ تيليجرام")
    gr.Markdown(f"يفحص كل {INTERVAL_SECONDS} ثانية. اضغط تحديث لرؤية آخر نتيجة.")
    status_box = gr.Markdown()
    refresh_btn = gr.Button("تحديث النتائج")
    log_box = gr.Textbox(label="آخر 30 سطر من السجل", lines=15)

    refresh_btn.click(fn=get_status_text, outputs=status_box)
    refresh_btn.click(fn=get_log_tail, outputs=log_box)
    demo.load(fn=get_status_text, outputs=status_box)
    demo.load(fn=get_log_tail, outputs=log_box)

demo.launch()