#!/usr/bin/env python3 """ 极简文件浏览和下载服务器使用Flask实现 """ from flask import Flask, send_file, render_template_string, request import os import time import sqlite3 app = Flask(__name__) # 数据库连接 def get_db_connection(): try: conn = sqlite3.connect(r'\\ll\D\xtrssvjj\db\dpd.db', timeout=30.0) conn.row_factory = sqlite3.Row # 调整同步模式，提高性能 conn.execute('PRAGMA synchronous=NORMAL') # 调整缓存大小 conn.execute('PRAGMA cache_size=-64000') return conn except Exception as e: import traceback import inspect func_name = inspect.currentframe().f_code.co_name line_no = inspect.currentframe().f_lineno print(f"错误 [函数: {func_name}, 行号: {line_no}]: 连接数据库失败: {e}") traceback.print_exc() raise # 主页 - 显示文件列表 @app.route('/') def index(): # 获取排序参数，默认按时间倒序 sort_by = request.args.get('sort', 'time') files = [] for filename in os.listdir('.'): filepath = os.path.join('.', filename) if os.path.isfile(filepath): files.append({ 'name': filename, 'size': os.path.getsize(filepath) / 1024 / 1024, # MB 'path': filename, 'mtime': os.path.getmtime(filepath) }) elif os.path.isdir(filepath): files.append({ 'name': f"📁 {filename}/", 'size': 'DIR', 'path': filename, 'mtime': os.path.getmtime(filepath) }) # 排序逻辑 if sort_by == 'name': files.sort(key=lambda x: x['name'].lower()) else: # time files.sort(key=lambda x: x['mtime'], reverse=True) # 简单的HTML模板 html = ''' 文件服务器

文件列表

排序: 按时间按字母 pub

{% for file in files %} {% endfor %}

文件名	大小	修改时间
{% if file.size != 'DIR' %} {{ file.name }} {% else %} {{ file.name }} {% endif %}	{% if file.size == 'DIR' %}目录{% else %}{{ "%.2f"\|format(file.size) }} MB{% endif %}	{{ file.mtime\|strftime }}

''' # 添加时间格式化过滤器 def strftime_filter(timestamp): return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(timestamp)) app.jinja_env.filters['strftime'] = strftime_filter return render_template_string(html, files=files, sort_by=sort_by) # 目录浏览 @app.route('/') def browse_directory(dirname): if not os.path.isdir(dirname): return f"

404

目录不存在

", 404 # 获取排序参数，默认按时间倒序 sort_by = request.args.get('sort', 'time') files = [] for filename in os.listdir(dirname): filepath = os.path.join(dirname, filename) if os.path.isfile(filepath): files.append({ 'name': filename, 'size': os.path.getsize(filepath) / 1024 / 1024, 'path': os.path.join(dirname, filename), 'mtime': os.path.getmtime(filepath) }) elif os.path.isdir(filepath): files.append({ 'name': f"📁 {filename}/", 'size': 'DIR', 'path': os.path.join(dirname, filename), 'mtime': os.path.getmtime(filepath) }) # 排序逻辑 if sort_by == 'name': files.sort(key=lambda x: x['name'].lower()) else: # time files.sort(key=lambda x: x['mtime'], reverse=True) html = ''' 文件服务器 - {{ dirname }}

← 返回上级

目录: {{ dirname }}

排序: 按时间按字母

{% for file in files %} {% endfor %}

文件名	大小	修改时间
{% if file.size != 'DIR' %} {{ file.name }} {% else %} {{ file.name }} {% endif %}	{% if file.size == 'DIR' %}目录{% else %}{{ "%.2f"\|format(file.size) }} MB{% endif %}	{{ file.mtime\|strftime }}

''' # 添加时间格式化过滤器 def strftime_filter(timestamp): return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(timestamp)) app.jinja_env.filters['strftime'] = strftime_filter return render_template_string(html, files=files, dirname=dirname, sort_by=sort_by) # 文件下载 @app.route('/download/') def download_file(filepath): if not os.path.isfile(filepath): return f"

404

文件不存在

", 404 return send_file(filepath, as_attachment=True) # 文件播放 @app.route('/play/') def play_file(filepath): if not os.path.isfile(filepath): return f"

404

文件不存在

", 404 # 获取文件扩展名 ext = os.path.splitext(filepath)[1].lower() # 对于HTML文件，直接读取并返回内容 if ext == '.html': try: with open(filepath, 'r', encoding='utf-8') as f: content = f.read() return content, 200, {'Content-Type': 'text/html'} except Exception as e: return f"

500

读取文件失败: {e}

", 500 # 设置MIME类型 mime_types = { '.mp3': 'audio/mpeg', '.mp4': 'video/mp4', '.wav': 'audio/wav', '.m4a': 'audio/mp4', '.ogg': 'audio/ogg', '.webm': 'video/webm', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.png': 'image/png', '.gif': 'image/gif' } mimetype = mime_types.get(ext, 'application/octet-stream') return send_file(filepath, mimetype=mimetype) # 显示 ads 表记录 @app.route('/listpub') def listpub(): import inspect import time func_name = inspect.currentframe().f_code.co_name line_no = inspect.currentframe().f_lineno conn = None max_retries = 3 retry_delay = 1 for attempt in range(max_retries): try: # 获取排序参数，默认按 pub 状态排序 sort_by = request.args.get('sort', 'status') # 连接数据库 conn = get_db_connection() # 根据排序方式构建查询 if sort_by == 'status': # 按 pub 状态排序：2（处理中）-> 1（待处理）-> 其他 -> NULL query = ''' SELECT rowid, tts, ttscnt, totsrts, name, pub FROM ads ORDER BY CASE WHEN pub = 2 THEN 0 WHEN pub = 1 THEN 1 WHEN pub IS NULL THEN 3 ELSE 2 END, rowid DESC LIMIT 100 ''' elif sort_by == 'rowid': query = ''' SELECT rowid, tts, ttscnt, totsrts, name, pub FROM ads ORDER BY rowid DESC LIMIT 100 ''' elif sort_by == 'createtime': # 按 createtime 倒序排序 query = ''' SELECT rowid, tts, ttscnt, totsrts, name, pub FROM ads ORDER BY createtime DESC LIMIT 100 ''' else: query = ''' SELECT rowid, tts, ttscnt, totsrts, name, pub FROM ads ORDER BY rowid DESC LIMIT 100 ''' cursor = conn.execute(query) ads = cursor.fetchall() # 立即关闭数据库连接，释放锁 conn.close() conn = None # 构建响应 html = ''' pub 列表

← 返回主页

排序: 按状态按 rowid 按 createtime

{% for ad in ads %} {% endfor %}

rowid	tts	ttscnt	totsrts	pub	name
{{ ad[0] }}	{{ ad[1] }}	{{ ad[2] }}	{{ ad[3] }}	{% if ad[1] == 0 %} {{ ad[5] }} pub {% elif ad[5] == 1 %} {{ ad[5] }} pub {% else %} {{ ad[5] }} pub {% endif %}	{{ ad[4] }}

''' return render_template_string(html, ads=ads, sort_by=sort_by) except Exception as e: import traceback print(f"错误 [函数: {func_name}, 行号: {line_no}]: 操作失败 (尝试 {attempt+1}/{max_retries}): {e}") traceback.print_exc() if conn: try: conn.close() except: pass conn = None if attempt < max_retries - 1: print(f"等待 {retry_delay} 秒后重试...") time.sleep(retry_delay) retry_delay *= 2 else: return f"

500

数据库查询失败: {e}

", 500 # pub 操作路由 @app.route('/pub//') def pub(adid, puber): import inspect func_name = inspect.currentframe().f_code.co_name line_no = inspect.currentframe().f_lineno conn = None try: # 连接数据库 conn = get_db_connection() # 更新 ads 表中的 pub 字段 conn.execute('UPDATE ads SET pub = ? WHERE rowid = ?', (puber, adid)) conn.commit() # 立即关闭数据库连接，释放锁 conn.close() conn = None html = ''' pub 操作

操作成功

adid: {{ adid }}

pub: {{ puber }}

reset pub

← 返回 pub 列表

''' return render_template_string(html, adid=adid, puber=puber) except Exception as e: import traceback print(f"错误 [函数: {func_name}, 行号: {line_no}]: 操作失败: {e}") traceback.print_exc() return f"

500

操作失败: {e}

", 500 finally: if conn: conn.close() # reset pub 操作路由 @app.route('/resetpub/') def resetpub(adid): import inspect func_name = inspect.currentframe().f_code.co_name line_no = inspect.currentframe().f_lineno conn = None try: # 连接数据库 conn = get_db_connection() # 将 pub 值复位为 0 conn.execute('UPDATE ads SET pub = 0 WHERE rowid = ?', (adid,)) conn.commit() # 立即关闭数据库连接，释放锁 conn.close() conn = None html = ''' reset pub 操作

操作成功

adid: {{ adid }}

pub 已复位为 0

← 返回 pub 列表

''' return render_template_string(html, adid=adid) except Exception as e: import traceback print(f"错误 [函数: {func_name}, 行号: {line_no}]: 操作失败: {e}") traceback.print_exc() return f"

500

操作失败: {e}

", 500 finally: if conn: conn.close() if __name__ == '__main__': print("文件服务器启动中...") print("访问地址: http://localhost:5000") print("按 Ctrl+C 停止服务器") app.run(host='0.0.0.0', port=5000, debug=False)