671 lines
25 KiB
Python
671 lines
25 KiB
Python
"""
|
||
ES相关的API视图
|
||
"""
|
||
import os
|
||
import re
|
||
import uuid
|
||
import base64
|
||
import json
|
||
from django.conf import settings
|
||
from django.http import JsonResponse
|
||
from django.shortcuts import render
|
||
from django.views.decorators.http import require_http_methods
|
||
from django.views.decorators.csrf import ensure_csrf_cookie
|
||
from django.views.decorators.csrf import csrf_exempt, ensure_csrf_cookie, csrf_protect
|
||
from .es_connect import *
|
||
from .es_connect import update_user_by_id as es_update_user_by_id, delete_user_by_id as es_delete_user_by_id
|
||
from .es_connect import (
|
||
analytics_trend as es_analytics_trend,
|
||
analytics_types as es_analytics_types,
|
||
analytics_types_trend as es_analytics_types_trend,
|
||
analytics_recent as es_analytics_recent,
|
||
)
|
||
from PIL import Image
|
||
|
||
|
||
@require_http_methods(["GET", "POST"])
|
||
@csrf_exempt
|
||
def init_index(request):
|
||
"""初始化ES索引"""
|
||
print("⚠️ init_index 被调用了！")
|
||
try:
|
||
create_index_with_mapping()
|
||
return JsonResponse({"status": "success", "message": "索引初始化成功"})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["POST"])
|
||
@csrf_exempt
|
||
def add_data(request):
|
||
"""添加数据到ES"""
|
||
try:
|
||
data = json.loads(request.body.decode('utf-8'))
|
||
success = insert_data(data)
|
||
if success:
|
||
return JsonResponse({"status": "success", "message": "数据添加成功"})
|
||
else:
|
||
return JsonResponse({"status": "error", "message": "数据添加失败"}, status=500)
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
def search(request):
|
||
"""搜索数据"""
|
||
try:
|
||
query = request.GET.get('q', '')
|
||
if not query:
|
||
return JsonResponse({"status": "error", "message": "搜索关键词不能为空"}, status=400)
|
||
|
||
results = search_data(query)
|
||
return JsonResponse({"status": "success", "data": results})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
def fuzzy_search(request):
|
||
"""模糊搜索"""
|
||
try:
|
||
keyword = request.GET.get('keyword', '')
|
||
if not keyword:
|
||
return JsonResponse({"status": "error", "message": "搜索关键词不能为空"}, status=400)
|
||
|
||
results = search_by_any_field(keyword)
|
||
return JsonResponse({"status": "success", "data": results})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
@require_http_methods(["GET"])
|
||
def get_all_data(request):
|
||
"""获取所有数据"""
|
||
try:
|
||
results = search_all()
|
||
return JsonResponse({"status": "success", "data": results})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
@require_http_methods(["DELETE"])
|
||
@csrf_exempt
|
||
def delete_data(request, doc_id):
|
||
"""删除数据（需登录；管理员或作者本人）"""
|
||
request_user=request.session.get("user_id")
|
||
# request_admin=request.session.get("permisssion")
|
||
if request_user is None:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
|
||
|
||
try:
|
||
existing = get_by_id(doc_id)
|
||
user_existing=get_user_by_id(request_user)
|
||
|
||
if not existing:
|
||
return JsonResponse({"status": "error", "message": "数据不存在"}, status=404)
|
||
|
||
is_admin = int(user_existing.get('permission')) == 0
|
||
is_owner = str(existing.get("writer_id", "")) == str(request.session.get("user_id"))
|
||
|
||
if not (is_admin or is_owner):
|
||
return JsonResponse({"status": "error", "message": "无权限"}, status=403)
|
||
success = delete_by_id(doc_id)
|
||
|
||
if success:
|
||
return JsonResponse({"status": "success", "message": "数据删除成功"})
|
||
else:
|
||
return JsonResponse({"status": "error", "message": "数据删除失败"}, status=500)
|
||
|
||
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["PUT"])
|
||
@csrf_exempt
|
||
def update_data(request, doc_id):
|
||
"""更新数据（需登录；管理员或作者本人）"""
|
||
request_user = request.session.get("user_id")
|
||
if request_user is None:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
|
||
try:
|
||
payload = json.loads(request.body.decode('utf-8'))
|
||
except Exception:
|
||
return JsonResponse({"status": "error", "message": "JSON无效"}, status=400)
|
||
try:
|
||
existing = get_by_id(doc_id)
|
||
user_existing = get_user_by_id(request_user)
|
||
|
||
if not existing:
|
||
return JsonResponse({"status": "error", "message": "数据不存在"}, status=404)
|
||
|
||
is_admin = int(user_existing.get('permission')) == 0
|
||
is_owner = str(existing.get("writer_id", "")) == str(request.session.get("user_id"))
|
||
|
||
if not (is_admin or is_owner):
|
||
return JsonResponse({"status": "error", "message": "无权限"}, status=403)
|
||
|
||
updated = {}
|
||
if "writer_id" in payload:
|
||
updated["writer_id"] = payload["writer_id"]
|
||
if "image" in payload:
|
||
updated["image"] = payload["image"]
|
||
if "data" in payload:
|
||
v = payload["data"]
|
||
if isinstance(v, dict):
|
||
updated["data"] = json.dumps(v, ensure_ascii=False)
|
||
else:
|
||
try:
|
||
obj = json.loads(str(v))
|
||
updated["data"] = json.dumps(obj, ensure_ascii=False)
|
||
except Exception:
|
||
updated["data"] = str(v)
|
||
|
||
success = update_by_id(doc_id, updated)
|
||
if success:
|
||
return JsonResponse({"status": "success", "message": "数据更新成功"})
|
||
else:
|
||
return JsonResponse({"status": "error", "message": "数据更新失败"}, status=500)
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
def get_data(request, doc_id):
|
||
"""获取单个数据"""
|
||
try:
|
||
result = get_by_id(doc_id)
|
||
if result:
|
||
return JsonResponse({"status": "success", "data": result})
|
||
else:
|
||
return JsonResponse({"status": "error", "message": "数据不存在"}, status=404)
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["POST"])
|
||
@csrf_protect
|
||
def add_user(request):
|
||
if request.session.get("user_id") is None:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
if int(request.session.get("permission", 1)) != 0:
|
||
return JsonResponse({"status": "error", "message": "无权限"}, status=403)
|
||
try:
|
||
payload = json.loads(request.body.decode("utf-8"))
|
||
except Exception:
|
||
return JsonResponse({"status": "error", "message": "JSON无效"}, status=400)
|
||
username = (payload.get("username") or "").strip()
|
||
password = (payload.get("password") or "").strip()
|
||
try:
|
||
permission = int(payload.get("permission", 1))
|
||
except Exception:
|
||
permission = 1
|
||
if not username:
|
||
return JsonResponse({"status": "error", "message": "用户名不能为空"}, status=400)
|
||
if password and len(password) < 6:
|
||
return JsonResponse({"status": "error", "message": "密码长度至少为6位"}, status=400)
|
||
existing = get_user_by_username(username)
|
||
if existing:
|
||
return JsonResponse({"status": "error", "message": "用户名已存在"}, status=409)
|
||
users = get_all_users()
|
||
next_id = (max([int(u.get("user_id", 0)) for u in users]) + 1) if users else 1
|
||
ok = write_user_data({
|
||
"user_id": next_id,
|
||
"username": username,
|
||
"password": password,
|
||
"permission": permission,
|
||
})
|
||
if not ok:
|
||
return JsonResponse({"status": "error", "message": "用户添加失败"}, status=500)
|
||
return JsonResponse({"status": "success", "message": "用户添加成功"})
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
def get_users(request):
|
||
if request.session.get("user_id") is None:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
if int(request.session.get("permission", 1)) != 0:
|
||
return JsonResponse({"status": "error", "message": "无权限"}, status=403)
|
||
try:
|
||
q = (request.GET.get("search") or "").strip()
|
||
users = get_all_users()
|
||
if q:
|
||
users = [u for u in users if q in str(u.get("username", ""))]
|
||
return JsonResponse({"status": "success", "data": users})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["POST"])
|
||
@csrf_protect
|
||
def update_user_by_id_view(request, user_id):
|
||
if request.session.get("user_id") is None:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
if int(request.session.get("permission", 1)) != 0:
|
||
return JsonResponse({"status": "error", "message": "无权限"}, status=403)
|
||
try:
|
||
payload = json.loads(request.body.decode("utf-8"))
|
||
except Exception:
|
||
return JsonResponse({"status": "error", "message": "JSON无效"}, status=400)
|
||
new_username = (payload.get("username") or "").strip()
|
||
new_permission = payload.get("permission")
|
||
new_password = (payload.get("password") or "").strip()
|
||
if new_username:
|
||
other = get_user_by_username(new_username)
|
||
if other and int(other.get("user_id", -1)) != int(user_id):
|
||
return JsonResponse({"status": "error", "message": "用户名已存在"}, status=409)
|
||
if new_password and len(new_password) < 6:
|
||
return JsonResponse({"status": "error", "message": "密码长度至少为6位"}, status=400)
|
||
ok = es_update_user_by_id(
|
||
user_id,
|
||
username=new_username if new_username else None,
|
||
permission=int(new_permission) if new_permission is not None else None,
|
||
password=new_password if new_password else None,
|
||
)
|
||
if not ok:
|
||
return JsonResponse({"status": "error", "message": "用户更新失败"}, status=500)
|
||
return JsonResponse({"status": "success", "message": "用户更新成功"})
|
||
|
||
@require_http_methods(["POST"])
|
||
@csrf_protect
|
||
def delete_user_by_id_view(request, user_id):
|
||
if request.session.get("user_id") is None:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
if int(request.session.get("permission", 1)) != 0:
|
||
return JsonResponse({"status": "error", "message": "无权限"}, status=403)
|
||
ok = es_delete_user_by_id(user_id)
|
||
if not ok:
|
||
return JsonResponse({"status": "error", "message": "用户删除失败"}, status=500)
|
||
return JsonResponse({"status": "success", "message": "用户删除成功"})
|
||
|
||
|
||
# 辅助：JSON 转换（兼容 a.py 行为）
|
||
def json_to_string(obj):
|
||
try:
|
||
return json.dumps(obj, ensure_ascii=False)
|
||
except Exception:
|
||
return str(obj)
|
||
|
||
|
||
def string_to_json(s):
|
||
try:
|
||
return json.loads(s)
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
# 移植自 a.py 的核心：调用大模型进行 OCR/信息抽取
|
||
def ocr_and_extract_info(image_path: str):
|
||
from openai import OpenAI
|
||
def encode_image(path: str) -> str:
|
||
with open(path, "rb") as f:
|
||
return base64.b64encode(f.read()).decode("utf-8")
|
||
|
||
base64_image = encode_image(image_path)
|
||
|
||
# api_key = getattr(settings, "AISTUDIO_API_KEY", "188f57db3766e02ed2c7e18373996d84f4112272")
|
||
# base_url = getattr(settings, "OPENAI_BASE_URL", "https://aistudio.baidu.com/llm/lmapi/v3")
|
||
# if not api_key:
|
||
# raise RuntimeError("缺少 AISTUDIO_API_KEY，请在环境变量或 settings 中配置")
|
||
|
||
|
||
api_key = getattr(settings, "AISTUDIO_API_KEY", "")
|
||
base_url = getattr(settings, "OPENAI_BASE_URL", "")
|
||
if not api_key or not base_url:
|
||
raise RuntimeError("缺少模型服务配置，请设置 AISTUDIO_API_KEY 与 OPENAI_BASE_URL")
|
||
client = OpenAI(api_key=api_key, base_url=base_url)
|
||
|
||
types = get_type_list()
|
||
chat_completion = client.chat.completions.create(
|
||
messages=[
|
||
{"role": "system", "content": "你是一个能理解图片和文本的助手，请根据用户提供的信息进行回答。"},
|
||
{
|
||
"role": "user",
|
||
"content": [
|
||
{"type": "text", "text": f"请识别这张图片中的信息，将你认为重要的数据转换为不包含嵌套的json，不要显示其它信息以便于解析，直接输出json结果即可。使用“数据类型”字段表示这个东西的大致类型，除此之外你可以自行决定使用哪些json字段。“数据类型”的内容有严格规定，请查看{json.dumps(types, ensure_ascii=False)}中是否包含你所需要的类型，确定不包含后你才可以填入你觉得合适的大致分类。"},
|
||
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{base64_image}"}},
|
||
],
|
||
},
|
||
],
|
||
model="ernie-4.5-turbo-vl-32k",
|
||
)
|
||
|
||
response_text = chat_completion.choices[0].message.content
|
||
|
||
def parse_response(text: str):
|
||
try:
|
||
result = json.loads(text)
|
||
if result:
|
||
return result
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
m = re.search(r"```json\n(.*?)```", text, re.DOTALL)
|
||
if m:
|
||
try:
|
||
result = json.loads(m.group(1))
|
||
if result:
|
||
return result
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
try:
|
||
fixed = text.replace("'", '"')
|
||
result = json.loads(fixed)
|
||
if result:
|
||
return result
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
return None
|
||
|
||
return parse_response(response_text)
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
@ensure_csrf_cookie
|
||
def upload_page(request):
|
||
session_user_id = request.session.get("user_id")
|
||
if session_user_id is None:
|
||
from django.shortcuts import redirect
|
||
return redirect("/accounts/login/")
|
||
user_id_qs = request.GET.get("user_id")
|
||
context = {"user_id": user_id_qs or session_user_id}
|
||
return render(request, "elastic/upload.html", context)
|
||
|
||
|
||
# 上传并识别（不入库）
|
||
@require_http_methods(["POST"])
|
||
def upload(request):
|
||
if request.session.get("user_id") is None:
|
||
fallback_uid = request.POST.get("user_id") or request.GET.get("user_id")
|
||
if fallback_uid:
|
||
request.session["user_id"] = fallback_uid
|
||
request.session.setdefault("permission", 1)
|
||
else:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
|
||
file = request.FILES.get("file")
|
||
if not file:
|
||
return JsonResponse({"status": "error", "message": "未选择文件"}, status=400)
|
||
|
||
images_dir = os.path.join(settings.MEDIA_ROOT, "images")
|
||
os.makedirs(images_dir, exist_ok=True)
|
||
filename = f"{uuid.uuid4()}_{file.name}"
|
||
abs_path = os.path.join(images_dir, filename)
|
||
|
||
with open(abs_path, "wb") as dst:
|
||
for chunk in file.chunks():
|
||
dst.write(chunk)
|
||
|
||
try:
|
||
data = ocr_and_extract_info(abs_path)
|
||
if not data:
|
||
return JsonResponse({"status": "error", "message": "无法识别图片内容"}, status=400)
|
||
|
||
rel_path = f"images/{filename}"
|
||
image_url = request.build_absolute_uri(settings.MEDIA_URL + rel_path)
|
||
return JsonResponse({
|
||
"status": "success",
|
||
"message": "识别成功，请确认数据后点击录入",
|
||
"data": data,
|
||
"image": rel_path,
|
||
"image_url": image_url,
|
||
})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
# 确认并入库
|
||
@require_http_methods(["POST"])
|
||
def confirm(request):
|
||
if request.session.get("user_id") is None:
|
||
# 允许从payload中带入user_id作为后备（便于前端已知用户时继续操作）
|
||
try:
|
||
payload_for_uid = json.loads(request.body.decode("utf-8"))
|
||
except Exception:
|
||
payload_for_uid = {}
|
||
fb_uid = (payload_for_uid or {}).get("user_id")
|
||
if fb_uid:
|
||
request.session["user_id"] = fb_uid
|
||
request.session.setdefault("permission", 1)
|
||
else:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
|
||
try:
|
||
payload = json.loads(request.body.decode("utf-8"))
|
||
except json.JSONDecodeError:
|
||
return JsonResponse({"status": "error", "message": "JSON无效"}, status=400)
|
||
|
||
edited = payload.get("data") or {}
|
||
image_rel = payload.get("image") or ""
|
||
if not isinstance(edited, dict) or not edited:
|
||
return JsonResponse({"status": "error", "message": "数据不能为空"}, status=400)
|
||
|
||
ensure_type_in_list(edited.get("数据类型"))
|
||
final_image_rel = image_rel
|
||
try:
|
||
if image_rel:
|
||
images_dir = os.path.join(settings.MEDIA_ROOT, "images")
|
||
os.makedirs(images_dir, exist_ok=True)
|
||
src_abs = os.path.join(settings.MEDIA_ROOT, image_rel)
|
||
base = os.path.splitext(os.path.basename(image_rel))[0]
|
||
webp_name = base + ".webp"
|
||
webp_abs = os.path.join(images_dir, webp_name)
|
||
with Image.open(src_abs) as im:
|
||
if im.mode in ("RGBA", "LA", "P"):
|
||
im = im.convert("RGBA")
|
||
else:
|
||
im = im.convert("RGB")
|
||
im.save(webp_abs, format="WEBP", quality=80)
|
||
final_image_rel = f"images/{webp_name}"
|
||
except Exception:
|
||
final_image_rel = image_rel
|
||
|
||
to_store = {
|
||
"writer_id": str(request.session.get("user_id")),
|
||
"data": json_to_string(edited),
|
||
"image": final_image_rel,
|
||
}
|
||
|
||
ok = insert_data(to_store)
|
||
if not ok:
|
||
return JsonResponse({"status": "error", "message": "写入ES失败"}, status=500)
|
||
|
||
try:
|
||
if image_rel and final_image_rel != image_rel:
|
||
orig_abs = os.path.join(settings.MEDIA_ROOT, image_rel)
|
||
if os.path.isfile(orig_abs):
|
||
os.remove(orig_abs)
|
||
except Exception:
|
||
pass
|
||
|
||
return JsonResponse({"status": "success", "message": "数据录入成功", "data": edited})
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
@ensure_csrf_cookie
|
||
def manage_page(request):
|
||
session_user_id = request.session.get("user_id")
|
||
if session_user_id is None:
|
||
from django.shortcuts import redirect
|
||
return redirect("/accounts/login/")
|
||
|
||
is_admin = int(request.session.get("permission", 1)) == 0
|
||
if is_admin:
|
||
raw_results = search_all()
|
||
else:
|
||
uid = str(session_user_id)
|
||
raw_results = [r for r in search_all() if str(r.get("writer_id", "")) == uid]
|
||
|
||
results = []
|
||
for r in raw_results:
|
||
try:
|
||
r_data = string_to_json(r.get("data", "{}"))
|
||
r_data["_id"] = r["id"]
|
||
r_data["_image"] = r.get("image", "")
|
||
results.append(r_data)
|
||
except Exception:
|
||
pass
|
||
|
||
return render(
|
||
request,
|
||
"elastic/manage.html",
|
||
{
|
||
"results": results,
|
||
"is_admin": is_admin,
|
||
"user_id": session_user_id,
|
||
},
|
||
)
|
||
# 规范化键，避免模板点号访问下划线前缀字段
|
||
results = []
|
||
for r in raw_results:
|
||
results.append({
|
||
"id": r.get("_id", ""),
|
||
"writer_id": r.get("writer_id", ""),
|
||
"image": r.get("image", ""),
|
||
"data": r.get("data", ""),
|
||
})
|
||
user_id_qs = request.GET.get("user_id")
|
||
context = {"items": results, "user_id": user_id_qs or session_user_id}
|
||
return render(request, "elastic/manage.html", context)
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
def analytics_trend_view(request):
|
||
try:
|
||
gte = request.GET.get("from")
|
||
lte = request.GET.get("to")
|
||
interval = request.GET.get("interval", "day")
|
||
data = es_analytics_trend(gte=gte, lte=lte, interval=interval)
|
||
return JsonResponse({"status": "success", "data": data})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
def analytics_types_view(request):
|
||
try:
|
||
gte = request.GET.get("from")
|
||
lte = request.GET.get("to")
|
||
size = request.GET.get("size")
|
||
try:
|
||
size_int = int(size) if size is not None else 10
|
||
except Exception:
|
||
size_int = 10
|
||
data = es_analytics_types(gte=gte, lte=lte, size=size_int)
|
||
return JsonResponse({"status": "success", "data": data})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
def analytics_types_trend_view(request):
|
||
try:
|
||
gte = request.GET.get("from")
|
||
lte = request.GET.get("to")
|
||
interval = request.GET.get("interval", "week")
|
||
size = request.GET.get("size")
|
||
try:
|
||
size_int = int(size) if size is not None else 8
|
||
except Exception:
|
||
size_int = 8
|
||
data = es_analytics_types_trend(gte=gte, lte=lte, interval=interval, size=size_int)
|
||
return JsonResponse({"status": "success", "data": data})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
|
||
@require_http_methods(["GET"])
|
||
def analytics_recent_view(request):
|
||
try:
|
||
limit = request.GET.get("limit")
|
||
gte = request.GET.get("from")
|
||
lte = request.GET.get("to")
|
||
try:
|
||
limit_int = int(limit) if limit is not None else 10
|
||
except Exception:
|
||
limit_int = 10
|
||
data = es_analytics_recent(limit=limit_int, gte=gte, lte=lte)
|
||
return JsonResponse({"status": "success", "data": data})
|
||
except Exception as e:
|
||
return JsonResponse({"status": "error", "message": str(e)}, status=500)
|
||
|
||
@require_http_methods(["GET"])
|
||
@ensure_csrf_cookie
|
||
def user_manage(request):
|
||
session_user_id = request.session.get("user_id")
|
||
if session_user_id is None:
|
||
from django.shortcuts import redirect
|
||
return redirect("/accounts/login/")
|
||
if int(request.session.get("permission", 1)) != 0:
|
||
from django.shortcuts import redirect
|
||
return redirect("/main/home/")
|
||
user_id_qs = request.GET.get("user_id")
|
||
context = {"user_id": user_id_qs or session_user_id}
|
||
return render(request, "elastic/users.html", context)
|
||
|
||
@require_http_methods(["GET"])
|
||
@ensure_csrf_cookie
|
||
def registration_code_manage_page(request):
|
||
session_user_id = request.session.get("user_id")
|
||
if session_user_id is None:
|
||
from django.shortcuts import redirect
|
||
return redirect("/accounts/login/")
|
||
if int(request.session.get("permission", 1)) != 0:
|
||
from django.shortcuts import redirect
|
||
return redirect("/main/home/")
|
||
user_id_qs = request.GET.get("user_id")
|
||
context = {"user_id": user_id_qs or session_user_id}
|
||
return render(request, "elastic/registration_codes.html", context)
|
||
|
||
@require_http_methods(["GET"])
|
||
def get_keys_list_view(request):
|
||
if request.session.get("user_id") is None:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
if int(request.session.get("permission", 1)) != 0:
|
||
return JsonResponse({"status": "error", "message": "无权限"}, status=403)
|
||
lst = get_keys_list()
|
||
return JsonResponse({"status": "success", "data": lst})
|
||
|
||
@require_http_methods(["POST"])
|
||
@csrf_protect
|
||
def add_key_view(request):
|
||
if request.session.get("user_id") is None:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
if int(request.session.get("permission", 1)) != 0:
|
||
return JsonResponse({"status": "error", "message": "无权限"}, status=403)
|
||
try:
|
||
payload = json.loads(request.body.decode("utf-8"))
|
||
except Exception:
|
||
return JsonResponse({"status": "error", "message": "JSON无效"}, status=400)
|
||
key_name = (payload.get("key") or "").strip()
|
||
if not key_name:
|
||
return JsonResponse({"status": "error", "message": "key不能为空"}, status=400)
|
||
ok = ensure_key_in_list(key_name)
|
||
if not ok:
|
||
return JsonResponse({"status": "error", "message": "key已存在或写入失败"}, status=409)
|
||
return JsonResponse({"status": "success"})
|
||
|
||
@require_http_methods(["POST"])
|
||
@csrf_protect
|
||
def generate_registration_code_view(request):
|
||
if request.session.get("user_id") is None:
|
||
return JsonResponse({"status": "error", "message": "未登录"}, status=401)
|
||
if int(request.session.get("permission", 1)) != 0:
|
||
return JsonResponse({"status": "error", "message": "无权限"}, status=403)
|
||
try:
|
||
payload = json.loads(request.body.decode("utf-8"))
|
||
except Exception:
|
||
return JsonResponse({"status": "error", "message": "JSON无效"}, status=400)
|
||
keys = list(payload.get("keys") or [])
|
||
manage_keys = list(payload.get("manage_keys") or [])
|
||
try:
|
||
days = int(payload.get("expires_in_days", 30))
|
||
except Exception:
|
||
days = 30
|
||
result = generate_registration_code(keys=keys, manage_keys=manage_keys, expires_in_days=days, created_by=request.session.get("user_id"))
|
||
if not result:
|
||
return JsonResponse({"status": "error", "message": "生成失败"}, status=500)
|
||
return JsonResponse({"status": "success", "data": result})
|