|
import os |
|
import subprocess |
|
import tarfile |
|
import asyncio |
|
from fastapi import FastAPI, HTTPException, Form, UploadFile, File, Depends, Query, Response |
|
from fastapi.responses import HTMLResponse, FileResponse, StreamingResponse |
|
from pydantic import BaseModel |
|
from tempfile import NamedTemporaryFile |
|
from typing import List |
|
from pydantic import BaseModel |
|
from fastapi.staticfiles import StaticFiles |
|
|
|
app = FastAPI() |
|
|
|
REQUIREMENTS_FILE = "requirements1.txt" |
|
|
|
class DockerImageParams(BaseModel): |
|
image_name: str |
|
tag: str = 'latest' |
|
|
|
async def stream_log(file_path): |
|
with open(file_path, 'r') as file: |
|
while True: |
|
line = file.readline() |
|
if line: |
|
yield line |
|
else: |
|
await asyncio.sleep(0.1) |
|
|
|
@app.post("/download-dependencies") |
|
async def download_dependencies(requirements_file: UploadFile = File(...)): |
|
try: |
|
with NamedTemporaryFile(delete=False) as tmp: |
|
tmp.write(requirements_file.file.read()) |
|
tmp_path = tmp.name |
|
|
|
|
|
os.makedirs("/tmp/dependencies", exist_ok=True) |
|
|
|
log_file_path = "/tmp/dependencies/download.log" |
|
with open(log_file_path, "w") as log_file: |
|
|
|
result = subprocess.run( |
|
["pip", "download", "-r", tmp_path, "-d", "/tmp/dependencies"], |
|
|
|
|
|
) |
|
|
|
if result.returncode != 0: |
|
raise HTTPException(status_code=500, detail="Error downloading dependencies. See log file for details.") |
|
|
|
|
|
tar_path = "/tmp/dependencies.tar.gz" |
|
with tarfile.open(tar_path, "w:gz") as tar: |
|
for root, _, files in os.walk("/tmp/dependencies"): |
|
for file in files: |
|
file_path = os.path.join(root, file) |
|
tar.add(file_path, arcname=file) |
|
tar.add(log_file_path, arcname="download.log") |
|
|
|
|
|
file_size = os.path.getsize(tar_path) |
|
|
|
|
|
return StreamingResponse(open(tar_path, "rb"), media_type="application/gzip", headers={ |
|
"Content-Disposition": f"attachment; filename=dependencies.tar.gz", |
|
"Content-Length": str(file_size) |
|
}) |
|
|
|
except subprocess.CalledProcessError as e: |
|
raise HTTPException(status_code=500, detail=f"Error downloading dependencies: {str(e)}") |
|
except Exception as e: |
|
raise HTTPException(status_code=500, detail=str(e)) |
|
finally: |
|
if os.path.exists(tmp_path): |
|
os.remove(tmp_path) |
|
if os.path.exists(log_file_path): |
|
os.remove(log_file_path) |
|
if os.path.exists(tar_path): |
|
os.remove(tar_path) |
|
def download_docker_image(image_name, tag='latest', destination='/tmp/docker-images'): |
|
try: |
|
os.makedirs(destination, exist_ok=True) |
|
|
|
tar_path = os.path.join(destination, f'{image_name.replace("/", "_")}.tar') |
|
|
|
|
|
if os.path.exists(tar_path): |
|
os.remove(tar_path) |
|
|
|
|
|
command = [ |
|
'skopeo', 'copy', |
|
f'docker://{image_name}:{tag}', |
|
f'docker-archive:{tar_path}' |
|
] |
|
subprocess.run(command, check=True) |
|
print(f"Image '{image_name}:{tag}' downloaded successfully to {destination}.") |
|
return tar_path |
|
except subprocess.CalledProcessError as e: |
|
print(f"Error downloading image: {str(e)}") |
|
raise HTTPException(status_code=500, detail=f"Error downloading Docker image: {str(e)}") |
|
|
|
|
|
app.mount("/static", StaticFiles(directory="static"), name="static") |
|
|
|
@app.get("/", response_class=HTMLResponse) |
|
async def read_root(): |
|
html_content = """ |
|
<html> |
|
<head> |
|
<title>Azeez's Help Desk</title> |
|
<link href="/static/css/index.css" rel="stylesheet"> |
|
<script src="/static/js/index.js"></script> |
|
</head> |
|
<body> |
|
<h1>Welcome to Azeez's Help Desk :)</h1> |
|
|
|
<div class="tab"> |
|
<button class="tablink" onclick="openTab(event, 'Docker')" id="defaultOpen">Docker Image Download</button> |
|
<button class="tablink" onclick="openTab(event, 'Pip')">Pip Dependencies Download</button> |
|
<button class="tablink" onclick="openTab(event, 'Deb')">Debian Packages Download</button> |
|
</div> |
|
|
|
<div id="Docker" class="tabcontent"> |
|
<h2>Docker Image Download</h2> |
|
<form onsubmit="handleDockerFormSubmit(event)"> |
|
<label for="image_name">Docker Image Name:</label> |
|
<input type="text" id="image_name" name="image_name" required><br><br> |
|
|
|
<label for="tag">Docker Image Tag:</label> |
|
<input type="text" id="tag" name="tag" value="latest" required><br><br> |
|
|
|
<input type="submit" id="docker-submit-button" value="Download Docker Image"> |
|
</form> |
|
<div id="docker-message" style="display: none; margin-top: 10px;"></div> |
|
</div> |
|
|
|
<div id="Pip" class="tabcontent"> |
|
<h2>Pip Dependencies Download</h2> |
|
<form onsubmit="handlePipFormSubmit(event)"> |
|
<label for="requirements_file">Requirements File:</label> |
|
<input type="file" id="requirements_file" name="requirements_file" accept=".txt" required><br><br> |
|
|
|
<input type="submit" id="pip-submit-button" value="Download Dependencies"> |
|
</form> |
|
<div id="pip-message" style="display: none; margin-top: 10px;"></div> |
|
</div> |
|
|
|
<div id="Deb" class="tabcontent"> |
|
<h2>Debian Packages Download</h2> |
|
<form onsubmit="handleDebFormSubmit(event)"> |
|
<label for="deb_packages">Debian Package Names (comma-separated):</label> |
|
<input type="text" id="deb_packages" name="deb_packages" required><br><br> |
|
|
|
<input type="submit" id="deb-submit-button" value="Download Debian Packages"> |
|
</form> |
|
<div id="deb-message" style="display: none; margin-top: 10px;"></div> |
|
</div> |
|
<div id="progress-container"> |
|
<div id="progress-bar">0%</div> |
|
</div> |
|
</body> |
|
</html> |
|
""" |
|
return HTMLResponse(content=html_content) |
|
|
|
@app.get("/download-docker-image") |
|
async def download_docker_image_endpoint(image_name: str = Query(...), tag: str = Query('latest')): |
|
tar_path = download_docker_image(image_name, tag) |
|
file_size = os.path.getsize(tar_path) |
|
|
|
def iterfile(): |
|
with open(tar_path, 'rb') as file: |
|
while chunk := file.read(1024 * 1024): |
|
yield chunk |
|
|
|
headers = { |
|
"Content-Disposition": f'attachment; filename="{image_name.replace("/", "_")}.tar"', |
|
"Content-Length": str(file_size) |
|
} |
|
|
|
return StreamingResponse(iterfile(), media_type='application/x-tar', headers=headers) |
|
|
|
def create_tar_file(files_to_package: List[str], tar_filename: str, destination_dir: str): |
|
""" |
|
Create a tar file containing specified files. |
|
Args: |
|
- files_to_package (list): List of paths to files to include in the tar file. |
|
- tar_filename (str): Name of the tar file to create. |
|
- destination_dir (str): Directory to save the tar file. |
|
""" |
|
try: |
|
tar_path = os.path.join(destination_dir, tar_filename) |
|
with tarfile.open(tar_path, "w:gz") as tar: |
|
for file_path in files_to_package: |
|
tar.add(file_path, arcname=os.path.basename(file_path)) |
|
print(f"Created tar file '{tar_filename}' successfully in '{destination_dir}'.") |
|
return tar_path |
|
except Exception as e: |
|
print(f"Error creating tar file: {e}") |
|
raise |
|
|
|
def download_deb_packages(package_names: List[str], destination_dir: str) -> List[str]: |
|
""" |
|
Download Debian packages (`.deb`) and their dependencies using `apt-get download`. |
|
Args: |
|
- package_names (list): List of package names to download. |
|
- destination_dir (str): Directory to save downloaded packages. |
|
Returns: |
|
- List of paths to downloaded `.deb` packages. |
|
""" |
|
try: |
|
|
|
os.makedirs(destination_dir, exist_ok=True) |
|
|
|
downloaded_packages = [] |
|
|
|
|
|
for package_name in package_names: |
|
|
|
|
|
|
|
subprocess.run(['apt-get', 'download', package_name, '-d', destination_dir], check=True) |
|
|
|
deb_filename = f"{package_name}.deb" |
|
downloaded_packages.append(os.path.join(destination_dir, deb_filename)) |
|
|
|
return downloaded_packages |
|
|
|
except subprocess.CalledProcessError as e: |
|
print(f"Error downloading packages: {e}") |
|
raise |
|
|
|
import subprocess |
|
|
|
def download_make_package(destination_dir): |
|
try: |
|
|
|
subprocess.run(['mkdir', '-p', destination_dir]) |
|
|
|
|
|
result = subprocess.run(['apt-get', 'download', 'make', '-d', destination_dir], check=True) |
|
|
|
if result.returncode == 0: |
|
print(f"Downloaded make package and dependencies to {destination_dir} successfully.") |
|
else: |
|
print("Failed to download make package.") |
|
|
|
except subprocess.CalledProcessError as e: |
|
print(f"Error downloading packages: {e}") |
|
|
|
@app.post("/download-deb-packages") |
|
async def download_deb_packages_handler(deb_packages: str = Form(...)): |
|
try: |
|
destination_dir = '/tmp/downloaded_packages' |
|
subprocess.run(['mkdir', '-p', destination_dir]) |
|
|
|
package_name='make' |
|
|
|
|
|
result = subprocess.run(['apt-get', 'install', package_name], check=True) |
|
|
|
|
|
if result.returncode == 0: |
|
tar_filename = f'{package_name}.tar.gz' |
|
tar_path = f'{destination_dir}/{tar_filename}' |
|
return FileResponse(tar_path, filename=tar_filename) |
|
|
|
|
|
else: |
|
raise HTTPException(status_code=500, detail=f"Failed to download {package_name} package.") |
|
|
|
except subprocess.CalledProcessError as e: |
|
error_message = str(e.stderr) |
|
print(f"Error downloading packages: {error_message}") |
|
raise HTTPException(status_code=500, detail=f"Error downloading {package_name} package: {error_message}") |
|
|
|
except Exception as e: |
|
print(f"Error: {str(e)}") |
|
raise HTTPException(status_code=500, detail=f"Failed to download {package_name} package: {str(e)}") |
|
|
|
|
|
if __name__ == "__main__": |
|
import uvicorn |
|
uvicorn.run(app, host="0.0.0.0", port=5000) |
|
|
