Create main.py

main.py

@@ -0,0 +1,226 @@
+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:///./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:8080/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:8080/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:8080/notifications/{username}"
+    response = requests.get(url)
+    if response.status_code != 200:
+        return f"Failed to send notification. Status code: {response.status_code}", ""
+    title = f"# {response.json()['notifications'][0]['title']}"
+    notification = response.json()['notifications'][0]['message']
+    return title, notification
+
+
+"""
+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")
+    title = gr.Markdown(label="title")
+    notification = gr.Markdown(label="Notification")
+    submit.click(
+        get_notifications,
+        inputs=[username],
+        outputs=[title, notification],
+    )
+
+with gr.Blocks() as demo:
+    NavBar()
+    gr.Image(
+                value="./background.png", 
+                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)
+