import os from datetime import datetime from pathlib import Path from django.contrib.auth import authenticate, login, logout from django.contrib.auth.decorators import login_required from django.views.decorators.csrf import csrf_exempt from mainProject.models import UserPlus, corretingWord, Wrong from django.http import JsonResponse, Http404, StreamingHttpResponse, FileResponse from django.shortcuts import render, HttpResponse, redirect from mainProject.models import simdFiles from zhipuai import ZhipuAI from docx import Document from docx.shared import Pt def index(request): username = request.session.get("username") word_list = simdFiles.objects.filter(user_id=username).values() profession = request.session.get("profession") return render(request, "index.html", {'word_list': word_list, "username": username, "profession": profession}) def login_page(request): return render(request, "login.html") def upload(request): username = request.session.get("username") word_list = list(simdFiles.objects.filter(user_id=username).values('id', 'title')) returnData = {} for item in word_list: temp = corretingWord.objects.filter(word_id=item['id']).values() for item_x in temp: returnData['id'] = item['id'] returnData['score'] = item_x['score'] returnData['title'] = item['title'] profession = request.session.get("profession") return render(request, "upload.html", {'word_list': returnData, "username": username, "profession": profession}) @csrf_exempt def change(request): if request.method == 'POST': data = request.POST.get('password') # 获取 Ajax 请求中的数据 username = request.session.get("username") UserPlus.objects.filter(username=username).update(password=data) return JsonResponse({"result": 1}) # 返回处理结果 def logout_x(request): logout(request) request.session.clear() return redirect("/index/") @login_required def user_page(request): username = request.session.get("username") profession = request.session.get("profession") number = corretingWord.objects.filter(student_id=username).count() data = {"username": username, "profession": profession, "number": number} return render(request, "user.html", data) def static_page(request): profession = request.session.get("profession") username = request.session.get("username") scores = [ corretingWord.objects.filter(score__range=(0, 10)).count(), corretingWord.objects.filter(score__range=(11, 20)).count(), corretingWord.objects.filter(score__range=(21, 30)).count(), corretingWord.objects.filter(score__range=(31, 40)).count(), corretingWord.objects.filter(score__range=(41, 50)).count(), corretingWord.objects.filter(score__range=(51, 60)).count(), corretingWord.objects.filter(score__range=(61, 70)).count(), corretingWord.objects.filter(score__range=(71, 80)).count(), corretingWord.objects.filter(score__range=(81, 90)).count(), corretingWord.objects.filter(score__range=(91, 100)).count(), ] all_type = Wrong.objects.values('title').distinct() type_a = [] number = [] for i in all_type: count = Wrong.objects.filter(title=i['title']).count() number.append(count) type_a.append(i['title']) return render(request, "statical.html", {"username": username, "profession": profession, "scores": scores, "all_type": type_a, "number": number}) @csrf_exempt def login_x(request): if request.method == 'GET': return render( request, 'login.html', ) elif request.method == 'POST': # 获取参数 user_name = request.POST.get('username', '') pwd = request.POST.get('password', '') # 用户已存在 if UserPlus.objects.filter(username=user_name): # 使用内置方法验证 user = authenticate(username=user_name, password=pwd) usern = UserPlus.objects.get(username=user_name) profession = usern.profession # 验证通过 if user: # 用户已激活 if user.is_active: login(request, usern) request.session["username"] = user_name request.session["profession"] = profession return redirect("/index/") # 未激活 else: return JsonResponse({ 'code': 200, 'msg': '用户未激活' }) # 验证失败 else: return JsonResponse({ 'code': 403, 'msg': '用户认证失败' }) # 用户不存在 else: return redirect('/login_page') def register_page(request): return render(request, "register.html") def register(request): if request.method == 'GET': return render( request, 'register.html' ) elif request.method == 'POST': # 获取参数 user_name = request.POST.get('username', '') pwd = request.POST.get('password', '') pparameter = request.POST.getlist('select') # 将数组转化为字符串 pparameter = ','.join(pparameter) # 用户已存在 if UserPlus.objects.filter(username=user_name): return JsonResponse({ 'code': 200, 'msg': '用户已存在' }) # 用户不存在 else: # 使用User内置方法创建用户 user = UserPlus.objects.create_user( username=user_name, password=pwd, profession=pparameter, email='123@qq.com', is_staff=1, is_active=1, is_superuser=0 ) usern = UserPlus.objects.get(username=user_name) login(request, usern) request.session["username"] = user_name request.session["profession"] = pparameter print(pparameter) return redirect("/index/") else: return JsonResponse({ 'code': 403, 'msg': '被禁止的请求' }) # 这是一个迭代器，减少内存占用 def read_file(url, chunk_size=512): with open(url, "rb") as f: while True: c = f.read(chunk_size) if c: yield c else: break @login_required def download(request): path = corretingWord.objects.filter(word_id=request.GET.get('id')).first() file = Path(str(path.file_correting.name)).resolve() if file.is_file(): response = FileResponse(open(file, 'rb'), as_attachment=True) return response else: return HttpResponse('文件不存在') @login_required def word_upload(request): if request.method == 'POST': username = request.session.get('username') profession = UserPlus.objects.filter(username=username).values('profession') profession_x = profession[0].get('profession') files = request.FILES.getlist('words') word_list = simdFiles.objects.filter(user_id=username).values() if profession_x == 0 and len(files) > 1: return render(request, 'upload.html', {'word_list': word_list}) time = "D:\\upload" + datetime.now().strftime( '%Y-%m-%d-%H-%M-%S') + username # 遍历写入到数据库中 for word in files: if not os.path.exists('D:\\upload'): os.mkdir('D:\\upload') # 写入到数据库中 simdFiles.objects.create(title=word.name, file_upload=time + word.name, user_id=request.session.get('username')) doc = Document(word) s = '' for para in doc.paragraphs: s += para.text client = ZhipuAI(api_key="a8d31831195a45a1570692903062a660.zxJIB74Ibupbpdqh") response = client.chat.completions.create( model="glm-4", messages=[ {"role": "system", "content": "你现在是一名教授python课程的大学老师，现在你要参考标准答案，结合自身知识储备批阅学生的作业，你很清楚Python" "编程类填空题具有一定的主观性，与标准答案稍有差别的学生答案也有可能正确，需要你谨慎批改。"}, {"role": "system", "content": "请你批改作业中的填空题，给出该学生的最终得分,并按照文本格式返回给我"}, { "role": "system", "content": "请在最后以" "学生成绩:" "所有错误题号:" "的形式严格发给我,在最后不需要你的提示或者注意！！！学生成绩后面只需要分数,例如学生成绩:3" }, { "role": "user", "content": "以下是学生的作业:" }, { "role": "user", "content": s } ], stream=True ) file = Document() file.encoding = 'utf-8' var = '' for chunk in response: var += chunk.choices[0].delta.content pg = file.add_paragraph(var) response = client.chat.completions.create( model="glm-4", messages=[ {"role": "user", "content": var}, {"role": "user", "content": "请提取上文中学生的成绩 以及错误题号 只需要返回给我 成绩: 错误题号:,以下为返回样例 成绩:3 错误题号:19,20"} ], stream=True ) x = '' for chunk in response: x += chunk.choices[0].delta.content temp = x.split("\n") score = temp[0].split(":")[1] ti = temp[1].split(":")[1].split(",") for i in ti: Wrong.objects.create(student_id=username, title=i) run = pg.runs[0] run.font.name = '宋体' run.font.size = Pt(12) word_ixd = simdFiles.objects.filter( file_upload=time + word.name).values( 'id') word_realid = word_ixd[0].get('id') corretingWord.objects.create(word_id=word_realid, file_correting="D:\\download\\" + datetime.now().strftime( '%Y-%m-%d-%H-%M-%S') + username + word.name, student_id=username, cor_time=datetime.now(), score=score) if not os.path.exists('D:\\download'): os.mkdir('D:\\download') file.save("D:\\download\\" + datetime.now().strftime('%Y-%m-%d-%H-%M-%S') + username + word.name) return redirect('/upload/')