Create app.py

app.py

@@ -0,0 +1,83 @@
+import os
+import csv
+import requests
+import threading
+from flask import Flask, render_template, request, jsonify, send_file
+
+app = Flask(__name__)
+
+results = []
+lock = threading.Lock()
+
+@app.route('/')
+def index():
+    return render_template('index.html', results=results)
+
+@app.route('/upload', methods=['POST'])
+def upload():
+    file = request.files['file']
+    if file.filename.endswith('.csv'):
+        file.save('/tmp/hosts.csv')
+        return '', 204
+    return '请上传 CSV 文件', 400
+
+@app.route('/start', methods=['POST'])
+def start():
+    threading.Thread(target=run_tests).start()
+    return jsonify({'status': 'started'})
+
+@app.route('/progress')
+def progress():
+    with lock:
+        return jsonify(results)
+
+@app.route('/download')
+def download():
+    return send_file('/tmp/result.csv', as_attachment=True)
+
+def test_login(host):
+    url = f"http://{host}/login"
+    payload = "username=admin&password=admin"
+    headers = {
+        "accept": "application/json",
+        "content-type": "application/x-www-form-urlencoded"
+    }
+    try:
+        response = requests.post(url, headers=headers, data=payload, timeout=10)
+        status = response.status_code
+        success = False
+        try:
+            response_json = response.json()
+            success = response_json.get("success", False)
+        except:
+            pass
+        with lock:
+            results.append(f"{host} - {'成功' if success else '失败'} ({status})")
+            if success:
+                with open('/tmp/result.csv', 'a', newline='', encoding='utf-8') as f:
+                    writer = csv.writer(f)
+                    writer.writerow([host])
+    except Exception as e:
+        with lock:
+            results.append(f"{host} - 异常: {str(e)}")
+
+def run_tests():
+    with lock:
+        results.clear()
+        if os.path.exists('/tmp/result.csv'):
+            os.remove('/tmp/result.csv')
+
+    with open('/tmp/hosts.csv', 'r', encoding='utf-8') as f:
+        reader = csv.reader(f)
+        hosts = [row[0] for row in reader]
+
+    threads = []
+    for host in hosts:
+        t = threading.Thread(target=test_login, args=(host,))
+        threads.append(t)
+        t.start()
+    for t in threads:
+        t.join()
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=7860)