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( """ """ ) 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)