leonxie/v1-backend
Archived
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
LeonspaceX
2025-10-18 17:32:46 +08:00
parent 59e391d84a
commit 245ef74698
4 changed files with 209 additions and 886 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
api_server.py
View File

@@ -90,30 +90,29 @@ def set_config(key, value):
# === 变量 ===
BANNED_KEYWORDS = [
"傻逼", "煞笔", "傻叉", "脑残", "狗东西",
"操你妈", "你妈的", "滚", "神经病", "贱人", "杂种", "王八蛋",
"臭婊子", "蠢货", "白痴", "妈的",
"约吗", "开房", "一夜情", "裸聊", "床照",
"小电影", "嫖娼", "成人网", "🈷吗",
"毒品", "枪支", "赌博", "六合彩", "博彩", "赌球",
"诈骗", "洗钱", "开票", "假证", "网监",
"习近平", "毛泽东", "共产党", "台湾独立", "台独", "法轮功",
"反动", "民主运动", "六四", "政变",
"割腕", "跳楼"
DEFAULT_BANNED_KEYWORDS = [
"傻逼"
]
ADMIN_TOKEN = "I_Love_SFLS_128936^"
# === 延迟初始化配置 ===
DEFAULT_ADMIN_TOKEN = "Sycamore_whisper"
DEFAULT_UPLOAD_FOLDER = "img"
DEFAULT_ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}
DEFAULT_MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB
UPLOAD_FOLDER = "img"
ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}
MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB
CONFIG = {}
INIT = False
NEED_AUDIT = False
if not os.path.exists(UPLOAD_FOLDER):
os.makedirs(UPLOAD_FOLDER)
# 运行时使用的变量，初始为默认值
ADMIN_TOKEN = DEFAULT_ADMIN_TOKEN
UPLOAD_FOLDER = DEFAULT_UPLOAD_FOLDER
ALLOWED_EXTENSIONS = set(DEFAULT_ALLOWED_EXTENSIONS)
MAX_FILE_SIZE = DEFAULT_MAX_FILE_SIZE
BANNED_KEYWORDS = list(DEFAULT_BANNED_KEYWORDS)
DB_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'instance', 'database.db')
IMG_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'img')
IMG_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), UPLOAD_FOLDER)
BACKUP_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'backups')
os.makedirs(BACKUP_FOLDER, exist_ok=True)
@@ -122,6 +121,140 @@ ALLOWED_BACKUP_EXTENSIONS = {'zip'}
def allowed_backup_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_BACKUP_EXTENSIONS
def apply_config_to_globals():
global ADMIN_TOKEN, UPLOAD_FOLDER, ALLOWED_EXTENSIONS, MAX_FILE_SIZE, IMG_FOLDER, BANNED_KEYWORDS
ADMIN_TOKEN = CONFIG.get('ADMIN_TOKEN', DEFAULT_ADMIN_TOKEN)
UPLOAD_FOLDER = CONFIG.get('UPLOAD_FOLDER', DEFAULT_UPLOAD_FOLDER)
ALLOWED_EXTENSIONS = set(CONFIG.get('ALLOWED_EXTENSIONS', DEFAULT_ALLOWED_EXTENSIONS))
MAX_FILE_SIZE = int(CONFIG.get('MAX_FILE_SIZE', DEFAULT_MAX_FILE_SIZE))
BANNED_KEYWORDS = list(CONFIG.get('BANNED_KEYWORDS', DEFAULT_BANNED_KEYWORDS))
IMG_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), UPLOAD_FOLDER)
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
def load_config():
global CONFIG, INIT
try:
import importlib, sys
importlib.invalidate_caches()
if 'config' in sys.modules:
cfg = importlib.reload(sys.modules['config'])
else:
cfg = importlib.import_module('config')
CONFIG = {
'ADMIN_TOKEN': getattr(cfg, 'ADMIN_TOKEN'),
'UPLOAD_FOLDER': getattr(cfg, 'UPLOAD_FOLDER'),
'ALLOWED_EXTENSIONS': set(getattr(cfg, 'ALLOWED_EXTENSIONS')),
'MAX_FILE_SIZE': int(getattr(cfg, 'MAX_FILE_SIZE')),
'BANNED_KEYWORDS': list(getattr(cfg, 'BANNED_KEYWORDS', DEFAULT_BANNED_KEYWORDS)),
}
INIT = True
apply_config_to_globals()
except Exception:
INIT = False
CONFIG = {}
# 启动时尝试加载配置
load_config()
# 全部接口在初始化完成前返回 503（仅 /init 允许）
@app.before_request
def gate_uninitialized():
if request.path == '/init':
return None
global INIT
# 若未初始化，尝试动态加载配置（兼容多进程/热重载场景）
if not INIT:
try:
load_config()
except Exception:
pass
if not INIT:
return jsonify({"status": "Fail", "reason": "Uninitialized"}), 503
def write_config_py(token, upload_folder, allowed_exts, max_file_size, banned_keywords=None):
# 归一化扩展名为小写且唯一
exts = sorted(set(str(e).strip().lower() for e in allowed_exts if str(e).strip()))
# 归一化敏感词为去空格的字符串列表
banned = banned_keywords if banned_keywords is not None else DEFAULT_BANNED_KEYWORDS
banned = [str(w).strip() for w in banned if str(w).strip()]
content = (
"# Auto-generated by /init\n"
f"ADMIN_TOKEN = {repr(token)}\n"
f"UPLOAD_FOLDER = {repr(upload_folder)}\n"
f"ALLOWED_EXTENSIONS = {repr(exts)}\n"
f"MAX_FILE_SIZE = {int(max_file_size)}\n"
f"BANNED_KEYWORDS = {repr(banned)}\n"
)
config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'config.py')
with open(config_path, 'w', encoding='utf-8') as f:
f.write(content)
@app.route('/init', methods=['POST'])
def init_service():
global INIT
if INIT:
return jsonify({"status": "Fail", "reason": "Already initialized"}), 403
data = request.get_json() or {}
required = ["ADMIN_TOKEN", "UPLOAD_FOLDER", "ALLOWED_EXTENSIONS", "MAX_FILE_SIZE"]
missing = [k for k in required if k not in data]
if missing:
return jsonify({"status": "Fail", "reason": f"Missing fields: {', '.join(missing)}"}), 400
token = str(data["ADMIN_TOKEN"])
upload_folder = str(data["UPLOAD_FOLDER"]).strip()
# 接受 list 或 逗号字符串
exts = data["ALLOWED_EXTENSIONS"]
if isinstance(exts, str):
allowed_exts = [x.strip() for x in exts.split(',')]
elif isinstance(exts, list):
allowed_exts = exts
else:
return jsonify({"status": "Fail", "reason": "ALLOWED_EXTENSIONS must be list or comma string"}), 400
try:
max_file_size = int(data["MAX_FILE_SIZE"])
except Exception:
return jsonify({"status": "Fail", "reason": "MAX_FILE_SIZE must be int"}), 400
# 可选的 BANNED_KEYWORDS
bk = data.get("BANNED_KEYWORDS", DEFAULT_BANNED_KEYWORDS)
if isinstance(bk, str):
banned_keywords = [x.strip() for x in bk.split(',') if x.strip()]
elif isinstance(bk, list):
banned_keywords = [str(x).strip() for x in bk if str(x).strip()]
else:
return jsonify({"status": "Fail", "reason": "BANNED_KEYWORDS must be list or comma string"}), 400
try:
write_config_py(token, upload_folder, allowed_exts, max_file_size, banned_keywords)
load_config()
initialize_database()
try:
global NEED_AUDIT
NEED_AUDIT = get_config("need_audit", "false").lower() == "true"
except Exception:
NEED_AUDIT = False
return jsonify({"status": "OK"}), 200
except Exception as e:
return jsonify({"status": "Fail", "reason": str(e)}), 500
# 在服务收到请求且已配置后，确保数据库表创建并加载审核状态
@app.before_request
def ensure_db_and_audit():
global NEED_AUDIT
if not getattr(ensure_db_and_audit, "_has_run", False) and INIT:
try:
initialize_database()
try:
NEED_AUDIT = get_config("need_audit", "false").lower() == "true"
except Exception:
NEED_AUDIT = False
except Exception:
pass
finally:
setattr(ensure_db_and_audit, "_has_run", True)
# === 管理端文章状态修改工具函数 ===
def admin_change_status(submission_id, from_status, to_status):
@@ -291,7 +424,7 @@ def serve_image(filename):
if not allowed_file(filename):
return 'Request not allowed', 403 # 后缀不允许
# 返回图片
return send_from_directory('img', filename)
return send_from_directory(UPLOAD_FOLDER, filename)
@app.route('/report', methods=['POST'])
@@ -473,10 +606,6 @@ def return_418():
def return_200():
return 'API OK!!!', 200
@app.route('/get/api_info', methods=['GET'])
def get_api_info():
return '<a>Sycamore_whisper API v1.0.0</a> Made with ❤️ By <a href="https://leonxie.cn">Leonxie</a>', 200
@app.route('/admin/need_audit', methods=['POST'])
@require_admin
def toggle_audit():
@@ -499,6 +628,36 @@ def get_need_audit():
global NEED_AUDIT
return jsonify({"status": NEED_AUDIT}), 200
# 动态敏感词配置
@app.route('/admin/get/banned_keywords', methods=['GET'])
@require_admin
def get_banned_keywords():
return jsonify({"keywords": BANNED_KEYWORDS}), 200
@app.route('/admin/banned_keywords', methods=['POST'])
@require_admin
def set_banned_keywords():
data = request.get_json() or {}
keywords = data.get("BANNED_KEYWORDS", data.get("banned_keywords"))
if keywords is None:
return jsonify({"status": "Fail", "reason": "BANNED_KEYWORDS not found"}), 400
if isinstance(keywords, str):
new_keywords = [x.strip() for x in keywords.split(',') if x.strip()]
elif isinstance(keywords, list):
new_keywords = [str(x).strip() for x in keywords if str(x).strip()]
else:
return jsonify({"status": "Fail", "reason": "BANNED_KEYWORDS must be list or comma string"}), 400
global BANNED_KEYWORDS
BANNED_KEYWORDS = new_keywords
try:
# 重写配置文件，确保重启后仍生效
write_config_py(ADMIN_TOKEN, UPLOAD_FOLDER, list(ALLOWED_EXTENSIONS), MAX_FILE_SIZE, BANNED_KEYWORDS)
load_config()
return jsonify({"status": "OK"}), 200
except Exception as e:
return jsonify({"status": "Fail", "reason": str(e)}), 500
@app.route('/admin/approve', methods=['POST'])
@require_admin
def admin_approve():
@@ -634,7 +793,7 @@ def admin_del_pic():
return jsonify({"status": "Fail", "reason": "filename not found"}), 400
filename = data["filename"]
file_path = os.path.join(os.getcwd(), "img", filename)
file_path = os.path.join(os.getcwd(), UPLOAD_FOLDER, filename)
if not os.path.isfile(file_path):
return jsonify({"status": "Fail", "reason": "file not found"}), 404
@@ -832,12 +991,12 @@ def admin_get_pic_links():
page = 1
per_page = 20
if not os.path.exists('img'):
if not os.path.exists(UPLOAD_FOLDER):
return jsonify([]), 200
all_files = sorted(
[f for f in os.listdir('img') if os.path.isfile(os.path.join('img', f))],
key=lambda x: os.path.getmtime(os.path.join('img', x)),
[f for f in os.listdir(UPLOAD_FOLDER) if os.path.isfile(os.path.join(UPLOAD_FOLDER, f))],
key=lambda x: os.path.getmtime(os.path.join(UPLOAD_FOLDER, x)),
reverse=True
)
@@ -881,7 +1040,4 @@ def initialize_database():
# === 启动 ===
if __name__ == '__main__':
with app.app_context():
initialize_database()
NEED_AUDIT = get_config("need_audit", "false").lower() == "true"
app.run(host='0.0.0.0', port=5000)
app.run(host='127.0.0.1', port=5000) # 监听IP&端口，建议监听127.0.0.1并配置反向代理