from flask import Flask, render_template, abort, request
import os
from config import LOG_ROOT, ALLOWED_EXTENSIONS, MAX_LINES_PER_PAGE
import vvdutils as vv
import pathlibapp = Flask(__name__)
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 10 # 静态文件缓存10s
def is_safe_path(path):
"""增强路径安全检查"""
target = os.path.abspath(os.path.join(LOG_ROOT, path))
return os.path.commonpath([LOG_ROOT, target]) == LOG_ROOT
@app.route('/')
def index():
return browse('')
@app.route('/browse/<path:subpath>')
def browse(subpath):
if not is_safe_path(subpath):
abort(403)
abs_path = os.path.join(LOG_ROOT, subpath)
if not os.path.exists(abs_path):
abort(404)
entries = []
try:
for entry in sorted(os.listdir(abs_path), key=lambda x: (not os.path.isdir(os.path.join(abs_path, x)), x.lower())):
entry_path = os.path.join(subpath, entry)
full_path = os.path.join(abs_path, entry)
# 最后修改时间(所有系统通用)
if os.path.isdir(full_path):
time_info = vv.get_file_time(full_path, datetime_res=True)
entries.append({
'name': entry,
'path': entry_path,
'is_dir': True,
'size': '-',
'modify_time': vv.time_string(datetime_obj=time_info['modify_time']),
'created_time': vv.time_string(datetime_obj=time_info['create_time'])
})
else:
ext = pathlib.Path(entry).suffix[1:].lower()
if ext not in ALLOWED_EXTENSIONS:
continue
time_info = vv.get_file_time(full_path, datetime_res=True)
entries.append({
'name': entry,
'path': entry_path,
'is_dir': False,
'size': os.path.getsize(full_path),
'modify_time': vv.time_string(datetime_obj=time_info['modify_time']),
'created_time': vv.time_string(datetime_obj=time_info['create_time'])
})
except PermissionError:
abort(403)
parent = os.path.dirname(subpath)
return render_template('browse.html',
entries=entries,
current_path=subpath,
parent=parent)
# @lru_cache(maxsize=100) # 由于时刻在变化,不能缓存
def get_cached_line_count(filepath):
return vv.get_file_line_number(filepath)
@app.route('/view/<path:filepath>')
def view_file(filepath):
if not is_safe_path(filepath):
abort(403)
page = request.args.get('page', 1, type=int)
search_term = request.args.get('search', '').lower()
abs_path = os.path.join(LOG_ROOT, filepath)
if not os.path.isfile(abs_path):
abort(404)
try:
total_lines = get_cached_line_count(abs_path)
except Exception as e:
app.logger.error(f"读取文件行数失败: {str(e)}")
abort(500)
total_pages = max(1, (total_lines + MAX_LINES_PER_PAGE - 1) // MAX_LINES_PER_PAGE)
page = max(1, min(page, total_pages))
try:
with open(abs_path, 'r', encoding='utf-8', errors='ignore') as f:
start_line = (page - 1) * MAX_LINES_PER_PAGE
content = []
for _ in range(start_line):
if not f.readline():
break
for _ in range(MAX_LINES_PER_PAGE):
line = f.readline()
if not line:
break
if search_term in line.lower():
content.append(line)
except IOError as e:
app.logger.error(f"文件读取失败: {str(e)}")
abort(500)
return render_template('view.html',
filename=os.path.basename(filepath),
content=''.join(content),
filepath=filepath,
page=page,
total_pages=total_pages)
@app.errorhandler(404)
def page_not_found(e):
return render_template('404.html'), 404
@app.errorhandler(403)
def forbidden(e):
return render_template('403.html'), 403
if __name__ == '__main__':
app.run(debug=True)