99 lines
3 KiB
Python
99 lines
3 KiB
Python
|
import asyncio
|
||
|
|
import os
|
||
|
|
import signal
|
||
|
|
import sys
|
||
|
|
import typing as t
|
||
|
|
from dataclasses import dataclass
|
||
|
|
|
||
|
|
import aiohttp
|
||
|
|
from flask import Flask, Response, render_template
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class Config():
|
||
|
|
base_url: str
|
||
|
|
token: t.Optional[str]
|
||
|
|
user: str
|
||
|
|
repo: str
|
||
|
|
task_name: str
|
||
|
|
page_size: int
|
||
|
|
artifact_names: t.List[str]
|
||
|
|
debug: bool
|
||
|
|
|
||
|
|
config = {
|
||
|
|
"base_url": os.environ.get('FORGEJO_BASE_URL'),
|
||
|
|
"token": os.environ.get('FORGEJO_TOKEN') or None,
|
||
|
|
"user": os.environ.get('FORGEJO_USER'),
|
||
|
|
"repo": os.environ.get('FORGEJO_REPO'),
|
||
|
|
"task_name": os.environ.get('FORGEJO_TASK_NAME'),
|
||
|
|
"page_size": os.environ.get('FORGEJO_PAGE_SIZE') or 10,
|
||
|
|
"artifact_names": os.environ.get('FORGEJO_ARTIFACT_NAMES', '').split(','),
|
||
|
|
"debug": os.environ.get('FORGEJO_DEBUG') or False,
|
||
|
|
}
|
||
|
|
config = Config(**config)
|
||
|
|
|
||
|
|
if not config.base_url:
|
||
|
|
raise ValueError("FORGEJO_BASE_URL is required")
|
||
|
|
if not config.user:
|
||
|
|
raise ValueError("FORGEJO_USER is required")
|
||
|
|
if not config.repo:
|
||
|
|
raise ValueError("FORGEJO_REPO is required")
|
||
|
|
if not config.task_name:
|
||
|
|
raise ValueError("FORGEJO_TASK_NAME is required")
|
||
|
|
if not config.artifact_names:
|
||
|
|
raise ValueError("FORGEJO_ARTIFACT_NAMES is required")
|
||
|
|
|
||
|
|
app = Flask(__name__, template_folder='templates')
|
||
|
|
|
||
|
|
@app.route('/')
|
||
|
|
def index():
|
||
|
|
return render_template('404.html')
|
||
|
|
|
||
|
|
async def fetch_tasks():
|
||
|
|
url = f'{config.base_url}/api/v1/repos/{config.user}/{config.repo}/actions/tasks'
|
||
|
|
headers = {
|
||
|
|
"accept": "application/json",
|
||
|
|
}
|
||
|
|
if config.token:
|
||
|
|
headers["Authorization"] = f"token {config.token}"
|
||
|
|
params = {
|
||
|
|
"page_size": config.page_size,
|
||
|
|
}
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
async with session.get(url, headers=headers, params=params) as response:
|
||
|
|
response.raise_for_status()
|
||
|
|
data = await response.json()
|
||
|
|
for task in data['workflow_runs']:
|
||
|
|
if task['name'] == config.task_name:
|
||
|
|
return task['url']
|
||
|
|
|
||
|
|
async def fetch_artifact(artifact_url):
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
async with session.get(artifact_url) as response:
|
||
|
|
data = await response.read()
|
||
|
|
return data
|
||
|
|
|
||
|
|
def create_handle_artifact(artifact_name: str):
|
||
|
|
def handle_artifact():
|
||
|
|
task = asyncio.run(fetch_tasks())
|
||
|
|
artifact_url = f"{task}/artifacts/{artifact_name}"
|
||
|
|
artifact_data = asyncio.run(fetch_artifact(artifact_url))
|
||
|
|
response = Response(artifact_data)
|
||
|
|
response.headers['Content-Disposition'] = 'attachment; filename="GalacticFactory.zip"'
|
||
|
|
return response
|
||
|
|
return handle_artifact
|
||
|
|
|
||
|
|
for artifact_name in config.artifact_names:
|
||
|
|
route = f'/{artifact_name}'
|
||
|
|
endpoint = f'handle_{artifact_name}'
|
||
|
|
app.add_url_rule(route, view_func=create_handle_artifact(artifact_name), endpoint=endpoint)
|
||
|
|
|
||
|
|
def handle_sigterm(*args):
|
||
|
|
print("Received SIGTERM, exiting gracefully...")
|
||
|
|
sys.exit(0)
|
||
|
|
|
||
|
|
signal.signal(signal.SIGTERM, handle_sigterm)
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
app.run(debug=config.debug, host='0.0.0.0', port=80)
|