-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapp.py
More file actions
92 lines (80 loc) · 3.01 KB
/
app.py
File metadata and controls
92 lines (80 loc) · 3.01 KB
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
# /// script
# requires-python = ">=3.7"
# dependencies = [
# "Flask==2.3.3",
# "Flask-CORS==4.0.0",
# "requests==2.31.0",
# "gunicorn==21.2.0; sys_platform != 'win32'",
# ]
# ///
from flask import Flask, request, jsonify, send_from_directory
from flask_cors import CORS
import requests
import os
from lib.apt_parser import AptParser
import io
import gzip
app = Flask(__name__, static_folder='static', template_folder='templates')
CORS(app)
@app.route('/')
def index():
"""Serve the main HTML page"""
return send_from_directory('templates', 'index.html')
@app.route('/config')
def config():
"""Return configuration values from environment variables"""
return jsonify({
'APTREPO': os.environ.get('APTREPO', '')
})
@app.route('/proxy')
def proxy():
"""
Proxy requests to APT repositories to avoid CORS issues
"""
target_url = request.args.get('url')
if not target_url:
return jsonify({'error': 'Missing url parameter'}), 400
try:
print(f"Proxying request to: {target_url}")
response = requests.get(
target_url,
headers={'User-Agent': 'APT-Repository-Previewer/1.0'},
timeout=(3.05, 10) # (connect_timeout, read_timeout) in seconds
)
# Pass through the original status code from the upstream server
status_code = response.status_code
print(f"Received status code {status_code} from {target_url}")
# Check if the response is a gzipped file
if target_url.endswith('.gz') and status_code == 200:
try:
print(f"Decompressing gzipped content from {target_url}")
# Decompress the gzipped content
gzip_content = io.BytesIO(response.content)
with gzip.GzipFile(fileobj=gzip_content, mode='rb') as f:
decompressed_content = f.read().decode('utf-8')
print(f"Successfully decompressed content from {target_url}")
return decompressed_content, status_code, {
'Content-Type': 'text/plain',
'Access-Control-Allow-Origin': '*'
}
except Exception as gz_error:
print(f"Error decompressing content from {target_url}: {str(gz_error)}")
return jsonify({
'error': f'Error decompressing gzipped content: {str(gz_error)}'
}), 500
return response.text, status_code, {
'Content-Type': 'text/plain',
'Access-Control-Allow-Origin': '*'
}
except Exception as e:
print(f"Error proxying request to {target_url}: {str(e)}")
return jsonify({
'error': f'Error fetching from repository: {str(e)}'
}), 500
@app.route('/static/<path:path>')
def serve_static(path):
"""Serve static files"""
return send_from_directory('static', path)
if __name__ == '__main__':
port = int(os.environ.get('PORT', 5000))
app.run(host='0.0.0.0', port=port, debug=True)