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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
| 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 pathlib
app = Flask(__name__)
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 10
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)
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)
|
