129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""
|
||
主页面蓝图：首页、home、图片工作台、品牌页、静态文件、Logo
|
||
"""
|
||
import os
|
||
import requests
|
||
from flask import Blueprint, Response, request, send_file
|
||
|
||
from app_common import (
|
||
get_db,
|
||
_render_html,
|
||
_is_session_user_valid,
|
||
login_required,
|
||
admin_required,
|
||
STATIC_DIR,
|
||
BASE_DIR,
|
||
ASSETS_DIR
|
||
)
|
||
from flask import redirect, url_for, session
|
||
|
||
from config import base_url,version,java_api_base
|
||
|
||
JAVA_API_BASE = java_api_base
|
||
|
||
main_bp = Blueprint('main', __name__)
|
||
|
||
|
||
@main_bp.route('/')
|
||
def index():
|
||
if session.get('user_id') and _is_session_user_valid():
|
||
return redirect(url_for('main.home'))
|
||
return redirect(url_for('auth.login'))
|
||
|
||
|
||
@main_bp.route('/home')
|
||
@login_required
|
||
def home():
|
||
try:
|
||
conn = get_db()
|
||
with conn.cursor() as cur:
|
||
cur.execute("SELECT username, is_admin FROM users WHERE id = %s", (session['user_id'],))
|
||
row = cur.fetchone()
|
||
conn.close()
|
||
return _render_html('home.html', username=row.get('username', ''), is_admin=bool(row.get('is_admin')), user_id=session.get('user_id'),baseUrl=base_url,version=version)
|
||
except Exception:
|
||
return _render_html('home.html', username=session.get('username', ''), is_admin=False, user_id=session.get('user_id'),baseUrl=base_url,version=version)
|
||
|
||
|
||
@main_bp.route('/image')
|
||
@login_required
|
||
def wb():
|
||
return _render_html('index.html')
|
||
|
||
|
||
@main_bp.route('/brand')
|
||
@login_required
|
||
def brand_page():
|
||
return _render_html('brand.html')
|
||
|
||
|
||
|
||
|
||
@main_bp.route('/static/<path:filename>')
|
||
def serve_static(filename):
|
||
"""提供 static 目录及子目录下的静态文件访问。"""
|
||
filepath = os.path.normpath(os.path.join(STATIC_DIR, filename))
|
||
static_abs = os.path.abspath(STATIC_DIR)
|
||
file_abs = os.path.abspath(filepath)
|
||
if not file_abs.startswith(static_abs) or not os.path.isfile(file_abs):
|
||
return '', 404
|
||
return send_file(file_abs, as_attachment=False)
|
||
|
||
|
||
@main_bp.route('/assets/<path:filename>')
|
||
def serve_assets(filename):
|
||
"""提供 static 目录及子目录下的静态文件访问。"""
|
||
filepath = os.path.normpath(os.path.join(ASSETS_DIR, filename))
|
||
static_abs = os.path.abspath(ASSETS_DIR)
|
||
file_abs = os.path.abspath(filepath)
|
||
if not file_abs.startswith(static_abs) or not os.path.isfile(file_abs):
|
||
return '', 404
|
||
return send_file(file_abs, as_attachment=False)
|
||
|
||
|
||
@main_bp.route('/new_web_source/<path:filename>')
|
||
def serve_new_web_source(filename):
|
||
"""提供 static 目录及子目录下的静态文件访问。"""
|
||
filepath = os.path.normpath(os.path.join("new_web_source", filename))
|
||
static_abs = os.path.abspath("new_web_source")
|
||
file_abs = os.path.abspath(filepath)
|
||
if not file_abs.startswith(static_abs) or not os.path.isfile(file_abs):
|
||
return '', 404
|
||
return send_file(file_abs, as_attachment=False)
|
||
|
||
|
||
|
||
@main_bp.route('/newApi/<path:subpath>', methods=['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS'])
|
||
def proxy_new_api(subpath):
|
||
target_url = f"{JAVA_API_BASE}/{subpath}"
|
||
try:
|
||
headers = {
|
||
key: value
|
||
for key, value in request.headers.items()
|
||
if key.lower() not in {'host', 'content-length'}
|
||
}
|
||
upstream = requests.request(
|
||
method=request.method,
|
||
url=target_url,
|
||
params=request.args,
|
||
data=request.get_data(),
|
||
headers=headers,
|
||
cookies=request.cookies,
|
||
allow_redirects=False,
|
||
timeout=300,
|
||
)
|
||
excluded_headers = {'content-encoding', 'content-length', 'transfer-encoding', 'connection'}
|
||
response_headers = [
|
||
(name, value)
|
||
for name, value in upstream.headers.items()
|
||
if name.lower() not in excluded_headers
|
||
]
|
||
return Response(upstream.content, upstream.status_code, response_headers)
|
||
except requests.RequestException as exc:
|
||
return {'success': False, 'message': str(exc), 'data': None}, 502
|
||
|
||
|
||
@main_bp.route('/logo.jpg', methods=['GET'])
|
||
def get_logo_image():
|
||
return send_file(os.path.join(BASE_DIR, "logo.jpg"), mimetype='image/jpeg')
|