from flask import Flask, render_template_string, request, jsonify,redirect, url_for,abort
|
import os
|
from auto_media_publisher.scheduler.task_scheduler_thread import auto_media_publisherTaskSchedulerThread
|
from pathlib import Path
|
import json
|
from auto_media_publisher.config.conf_base import TASK_CONFIG_DIR
|
|
|
try:
|
globle_schduler = auto_media_publisherTaskSchedulerThread()
|
globle_schduler.daemon = True
|
globle_schduler.start()
|
except Exception as e:
|
print("[ERROR] 启动调度器失败：", e)
|
|
app = Flask(__name__)
|
|
@app.route("/")
|
def home():
|
return redirect("/tasks")
|
|
@app.route("/tasks")
|
def tasks():
|
infos = globle_schduler.get_task_info()
|
return render_template_string("""
|
<h2>任务状态</h2>
|
<ul>
|
{% for tid, info in infos.items() %}
|
<li><b>{{ tid }}</b> | 上次运行：{{ info.last_run }}<br/>
|
<pre>{{ info.last_log }}</pre></li>
|
{% endfor %}
|
</ul>
|
""", infos=infos)
|
|
@app.route("/users")
|
def list_users():
|
users = [p.name for p in Path(TASK_CONFIG_DIR).rglob("*.json")]
|
return render_template_string("""
|
<h2>用户列表</h2>
|
<form method="POST" action="/user/new" style="margin-top:20px;">
|
<label>用户ID（英文标识）:</label>
|
<input name="user_id" required>
|
<label>用户名（中文）:</label>
|
<input name="username" required>
|
<button type="submit">➕ 新建用户配置</button>
|
</form>
|
<ul>
|
{% for u in users %}
|
<li><a href="/user/{{ u }}">{{ u }}</a></li>
|
{% endfor %}
|
</ul>
|
""", users=users)
|
|
@app.route("/user/<uid>", methods=["GET", "POST"])
|
def edit_user(uid):
|
path = Path(TASK_CONFIG_DIR) / uid
|
if not path.exists():
|
abort(404)
|
if request.method == "POST":
|
raw_text = request.form["content"]
|
cleaned_lines = [line.rstrip() for line in raw_text.splitlines() if line.strip()]
|
cleaned_text = "\n".join(cleaned_lines) + "\n"
|
path.write_text(cleaned_text, encoding="utf-8")
|
return redirect(url_for("list_users"))
|
else:
|
content = ""
|
if path.exists():
|
content = path.read_text(encoding="utf-8")
|
return render_template_string("""
|
<h2>编辑用户配置：{{ uid }}</h2>
|
<form method="POST">
|
<textarea name="content" style="width:90%;height:300px">{{ content }}</textarea><br/>
|
<button type="submit">保存</button>
|
</form>
|
""", uid=uid, content=content)
|
|
|
@app.route('/upload_auth', methods=['POST'])
|
def upload_cookie():
|
username = request.form.get('username')
|
if not username:
|
return jsonify({'success': False, 'error': 'Missing username'})
|
|
file = request.files.get('file')
|
if file.filename == '':
|
return jsonify({'success': False, 'error': 'No selected file'})
|
|
if file and file.filename.endswith('.json'):
|
try:
|
save_path = os.path.join('./data/cookies', username,file.filename)
|
os.makedirs(os.path.dirname(save_path), exist_ok=True)
|
file.save(save_path)
|
print(f"✅ 保存成功: {save_path}")
|
return jsonify({'success': True})
|
except Exception as e:
|
print(f"❌ 保存失败: {e}")
|
return jsonify({'success': False, 'error': str(e)})
|
|
return jsonify({'success': False, 'error': 'Invalid file type'})
|
|
@app.route("/user/new", methods=["POST"])
|
def new_user():
|
user_id = request.form.get("user_id", "").strip()
|
username = request.form.get("username", "").strip()
|
|
if not user_id or not username:
|
return "用户ID 和 用户名 都不能为空", 400
|
|
json_path = Path(TASK_CONFIG_DIR) / f"{user_id}.json"
|
if json_path.exists():
|
return f"配置文件 {user_id}.json 已存在", 400
|
|
default_config = {
|
"id": 1,
|
"user": "13381602515",
|
"username": "施教练",
|
"city": "上海",
|
"proxy": "http://127.0.0.1:8080",
|
"enabled": 1,
|
"tags": [
|
"太极鸟",
|
"机构用户"
|
],
|
"datetime_created": "2025-07-14T09:00:00",
|
"datetime_update": "2025-07-14T09:00:00",
|
"platforms": [
|
{
|
"name": "douyin",
|
"enabled": True,
|
"task_type": "video_upload",
|
"upload_times": [
|
"08:00",
|
"16:00"
|
]
|
},
|
{
|
"name": "xiaohongshu",
|
"enabled": False,
|
"task_type": "video_upload",
|
"upload_times": []
|
}
|
]
|
}
|
|
json_path.write_text(json.dumps(default_config, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
return redirect(url_for("edit_user", uid=f"{user_id}.json"))
|
|
|
# 注册退出处理
|
#import atexit
|
#atexit.register(lambda: globle_schduler.shutdown(wait=False))
|
|
if __name__ == "__main__":
|
app.run(port=5000, debug=True) # 本地访问
|
#app.run(host='0.0.0.0', port=5000,debug=True) # 允许外网访问
|
