-
Notifications
You must be signed in to change notification settings - Fork 1
/
web_server.py
99 lines (77 loc) · 2.34 KB
/
web_server.py
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
import logging
import zproc
from flask import Flask, Response, abort
from flask import request
from flask_autoindex import AutoIndex
import build_server
from app_builder import OUTPUT_DIR
app = Flask(__name__)
ctx = zproc.Context()
ax = AutoIndex(app, browse_root=OUTPUT_DIR)
@app.route("/do_build/", methods=["POST"])
def on_push():
data = request.get_json()
print(f"got build request: {data}")
state = ctx.create_state()
state["next_build_request"] = (
data["project"]["name"],
data["project"]["git_http_url"],
data["ref"][len("refs/heads/") :],
data["checkout_sha"],
)
return "OK"
def fmt_log(levelno: int, msg: str) -> str:
color = "black"
if levelno == logging.DEBUG:
color = "MediumOrchid"
elif levelno == logging.ERROR:
color = "red"
return f"<span style='color: {color};'>{msg}</span><br>"
def stream_build_logs(state: zproc.State, name: str, branch: str, url: str):
yield f"""
<html>
<head>
<meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body>
<h3>Project: {name}</h3>
<h3>Branch: {branch}</h3>
<h3>Url: {url}</h3>
<pre>
"""
footer = "</pre></body></html>"
if "logs" in state:
logs = state["logs"]
else:
logs = next(state.when_available("logs"))
yield from (fmt_log(*it) for it in logs)
last_len = len(logs)
if state["completed"]:
yield footer
return
for snapshot in state.when(
lambda it: len(it["logs"]) > last_len or it["completed"]
):
logs = snapshot["logs"]
yield from (fmt_log(*it) for it in logs[last_len:])
last_len = len(logs)
if snapshot["completed"]:
break
yield footer
@app.route("/build_logs/<string:git_hash>/")
def build_logs(git_hash: str):
state = ctx.create_state()
state.namespace = "request_history"
try:
build_info = state[git_hash]
except KeyError:
return abort(404)
state.namespace = git_hash
return Response(stream_build_logs(state, *build_info))
if __name__ == "__main__":
build_server.run(ctx)
try:
app.run(host="0.0.0.0", port=80)
except PermissionError:
print("Permission denied on port 80! Falling back to 8000...")
app.run(host="0.0.0.0", port=8000)