main.py

import os
import requests
import gradio as gr
import uvicorn
from fastapi import FastAPI, WebSocket, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Text
from sqlalchemy.orm import declarative_base, sessionmaker, relationship, Session

import secrets


app = FastAPI(title="Pushover Clone")

# Database setup
DATABASE_URL = "sqlite:////tmp/pushover_clone.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()

# Models
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    token = Column(String, unique=True, index=True)

    notifications = relationship("Notification", back_populates="recipient")

class Notification(Base):
    __tablename__ = "notifications"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    message = Column(Text)
    user_id = Column(Integer, ForeignKey("users.id"))

    recipient = relationship("User", back_populates="notifications")

Base.metadata.create_all(bind=engine)

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Schemas
class RegisterUser(BaseModel):
    username: str

class PushMessage(BaseModel):
    token: str
    title: str
    message: str


@app.post("/register_user")
def register_user(payload: RegisterUser, db: Session = Depends(get_db)):
    token = secrets.token_hex(16)
    user = User(username=payload.username, token=token)
    db.add(user)
    db.commit()
    db.refresh(user)
    return {"username": user.username, "token": user.token}

@app.post("/notify")
def send_notification(payload: PushMessage, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.token == payload.token).first()
    if not user:
        raise HTTPException(status_code=403, detail="Invalid token")
    notif = Notification(title=payload.title, message=payload.message, recipient=user)
    db.add(notif)
    db.commit()
    return {"status": "queued", "user": user.username}

@app.get("/notifications/{username}")
def get_notifications(username: str, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.username == username).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    notes = [{"title": n.title, "message": n.message} for n in user.notifications]
    return {"notifications": notes}

@app.websocket("/ws/{username}")
async def websocket_notifications(ws: WebSocket, username: str, db: Session = Depends(get_db)):
    await ws.accept()
    user = db.query(User).filter(User.username == username).first()
    if not user:
        await ws.send_text("User not found")
        await ws.close()
        return

    await ws.send_text(f"Connected to {username}'s notification stream")

    # Keep connection open (simplified)
    while True:
        db = SessionLocal()
        user = db.query(User).filter(User.username == username).first()
        if user and user.notifications:
            last = user.notifications[-1]
            await ws.send_json({"title": last.title, "message": last.message})
        await ws.receive_text()  # wait for client ping

"""
APIs for Gradio Interfaces
"""
def register_user(username):
    url = "http://127.0.0.1:7860/register_user"
    response = requests.post(url, json={"username": username})
    if response.status_code != 200:
        return f"Failed to register user. Status code: {response.status_code}", ""
    return response.json().get("username", ""), response.json().get("token", "")

def send_notification(token, title, message):
    url = "http://127.0.0.1:7860/notify"
    response = requests.post(url, json={"token": token, "title": title, "message": message})
    if response.status_code != 200:
        return f"Failed to send notification. Status code: {response.status_code}"
    return f"Notification sent to user with token ending in ...{token[-4:]}"

def get_notifications(username):
    url = f"http://127.0.0.1:7860/notifications/{username}"
    response = requests.get(url)
    if response.status_code != 200:
        return f"Failed to send notification. Status code: {response.status_code}", ""
    msg_list = response.json()['notifications']
    if isinstance(msg_list, list):
        md_text = ""
        for msg in msg_list:
            md_text += f"# {msg['title']}\n{msg['message']}\n\n"
        return md_text

    md_text = f"# {msg['title']}\n{msg['message']}\n\n"
    return md_text


"""
Gradio Interfaces
"""

def NavBar():
    gr.HTML(
        """
        <nav style="
            background: #4f46e5; 
            padding: 12px 24px; 
            display: flex; 
            justify-content: space-between; 
            align-items: center;
            color: white;
            font-family: sans-serif;">
            
            <div style='font-weight: bold; font-size: 20px; style="color:white;"'>
            PushOver Notifier
            </div>
            
            <div style="display: flex; gap: 20px; font-size: 16px;">
                <a href="/register_new_user/" style="color:white; text-decoration:none;">Register New User</a>
                <a href="/notify_users_device/" style="color:white; text-decoration:none;">Notify Users</a>
                <a href="/get_notifications_page/" style="color:white; text-decoration:none;">Notifications</a>
            </div>
        </nav>
        """
    )



with gr.Blocks() as register_new_user:
    NavBar()
    gr.Markdown("## Register User")
    register_username = gr.Textbox(label="Enter Username")
    register_button = gr.Button("Register")
    your_username = gr.Textbox(label="Your Username", interactive=False)
    created_token = gr.Textbox(label="Your Token", interactive=False, type="password", show_copy_button=True)
    register_button.click(
        register_user,
        inputs=[register_username],
        outputs=[your_username, created_token],
    )


with gr.Blocks() as notify_users_device:
    NavBar()
    gr.Markdown("## Send Push Notification")
    token = gr.Textbox(label="Enter your Token", type="password")
    title = gr.Textbox(label="Enter Notification Title")
    message = gr.Textbox(label="Enter Notification Message")
    submit = gr.Button("Send Notification")
    alert = gr.Textbox(label="Alert", interactive=False)
    submit.click(
        send_notification,
        inputs=[token, title, message],
        outputs=[alert],
    )

with gr.Blocks() as get_notifications_page:
    NavBar()
    gr.Markdown("## Notifications")
    username = gr.Textbox(label="Enter your Username")
    submit = gr.Button("Submit")
    gr.Markdown("### Latest Notification")
    notification = gr.Markdown(label="Notification")
    submit.click(
        get_notifications,
        inputs=[username],
        outputs=[notification],
    )


BASE_DIR = os.path.dirname(os.path.abspath(__file__))
background_path = os.path.join(BASE_DIR, "background.png")

with gr.Blocks() as demo:
    NavBar()
    gr.Image(
                value=background_path, 
                show_label=False,
                elem_classes="full-img",
            )

app = gr.mount_gradio_app(app, register_new_user, path="/register_new_user")
app = gr.mount_gradio_app(app, notify_users_device, path="/notify_users_device")
app = gr.mount_gradio_app(app, get_notifications_page, path="/get_notifications_page")

app = gr.mount_gradio_app(app, demo, path="/")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)