File size: 4,214 Bytes
0b0871b 5825f9f 0b0871b 5825f9f 0b0871b 78b9bfc 0b0871b 5825f9f 0b0871b 5825f9f 0b0871b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 |
from fastapi import FastAPI, Request, HTTPException, UploadFile, File, Form
from pydantic import BaseModel
import uuid
from fastapi.responses import RedirectResponse
from fastapi.middleware.cors import CORSMiddleware
import os
import requests
from urllib.parse import urlencode
import aiofiles
app = FastAPI()
deployments = {}
# GitHub OAuth config (replace with your own client_id and client_secret)
GITHUB_CLIENT_ID = os.getenv("GITHUB_CLIENT_ID", "your_github_client_id")
GITHUB_CLIENT_SECRET = os.getenv("GITHUB_CLIENT_SECRET", "your_github_client_secret")
GITHUB_OAUTH_REDIRECT = os.getenv("GITHUB_OAUTH_REDIRECT", "https://dragxd-host.hf.space/github/callback")
# In-memory user token storage (for demo; use DB in production)
user_tokens = {}
# In-memory user selected repo storage (for demo)
user_selected_repo = {}
# In-memory .env storage: {(user_id, repo_full_name): env_content}
env_files = {}
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
class DeployRequest(BaseModel):
github_url: str
user_id: str
class RepoSelectRequest(BaseModel):
user_id: str
repo_full_name: str
@app.post("/deploy")
async def deploy(req: DeployRequest):
# Simulate deployment
deploy_id = str(uuid.uuid4())
deployments[deploy_id] = {
"github_url": req.github_url,
"user_id": req.user_id,
"status": "deploying",
"url": None
}
# Simulate deployment done after a few seconds (in real app, use background task)
deployments[deploy_id]["status"] = "success"
deployments[deploy_id]["url"] = f"https://demo-deployment/{deploy_id}"
return {"deploy_id": deploy_id}
@app.get("/status/{deploy_id}")
async def status(deploy_id: str):
if deploy_id in deployments:
return deployments[deploy_id]
return {"error": "Not found"}
@app.get("/github/login")
def github_login(user_id: str):
params = {
"client_id": GITHUB_CLIENT_ID,
"redirect_uri": GITHUB_OAUTH_REDIRECT,
"scope": "repo",
"state": user_id
}
url = f"https://github.com/login/oauth/authorize?{urlencode(params)}"
return {"auth_url": url}
@app.get("/github/callback")
def github_callback(code: str, state: str):
# state is user_id
token_url = "https://github.com/login/oauth/access_token"
headers = {"Accept": "application/json"}
data = {
"client_id": GITHUB_CLIENT_ID,
"client_secret": GITHUB_CLIENT_SECRET,
"code": code,
"redirect_uri": GITHUB_OAUTH_REDIRECT,
"state": state
}
resp = requests.post(token_url, headers=headers, data=data)
token = resp.json().get("access_token")
if token:
user_tokens[state] = token
return RedirectResponse(url=f"https://t.me/your_bot_username?start=github_connected")
return {"error": "Failed to get token"}
@app.get("/github/repos")
def github_repos(user_id: str):
token = user_tokens.get(user_id)
if not token:
raise HTTPException(status_code=401, detail="User not authenticated with GitHub.")
headers = {"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
resp = requests.get("https://api.github.com/user/repos", headers=headers)
if resp.status_code != 200:
raise HTTPException(status_code=resp.status_code, detail="Failed to fetch repos.")
repos = resp.json()
# Return only repo name and full_name for selection
return [{"name": r["name"], "full_name": r["full_name"]} for r in repos]
@app.post("/repo/select")
def select_repo(req: RepoSelectRequest):
user_selected_repo[req.user_id] = req.repo_full_name
return {"message": "Repo selected", "repo_full_name": req.repo_full_name}
@app.post("/env/upload")
async def upload_env(user_id: str = Form(...), file: UploadFile = File(...)):
content = await file.read()
# Find selected repo for user
repo_full_name = user_selected_repo.get(user_id)
if not repo_full_name:
return {"error": "No repo selected for user."}
env_files[(user_id, repo_full_name)] = content.decode()
return {"message": ".env uploaded", "repo_full_name": repo_full_name} |