import base64 import json import random import string import threading from datetime import timedelta from bleach import Cleaner from bleach.css_sanitizer import CSSSanitizer from django import forms from django.contrib.auth.forms import UserCreationForm from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin, UserPassesTestMixin from django.contrib.auth.models import User, Group from django.contrib.auth.views import LoginView from django.core.exceptions import ValidationError from django.core.files.base import ContentFile from django.core.paginator import Paginator from django.db import IntegrityError, transaction from django.db.models import Value, Q, Sum, Count, Max, F from django.db.models.functions import Replace from django.forms import modelform_factory, inlineformset_factory, \ ClearableFileInput from django.http import HttpResponse, JsonResponse, HttpResponseForbidden, HttpResponseRedirect from django.shortcuts import render, get_object_or_404, redirect from django.urls import reverse_lazy, reverse from django.utils import timezone from django.utils.decorators import method_decorator from django.views.decorators.cache import never_cache from django.views.decorators.csrf import csrf_protect from django.views.decorators.debug import sensitive_post_parameters from django.views.generic import TemplateView, View, FormView from CMSBlueprintStation.settings import BASE_DIR from home.model_util import update_assets, thread_lock, get_item_face, analyze_blueprint, \ refresh_single_item_ico from home.models import Choice, Function, Tag, Post, Blueprint, ActiveCode, Assets, Record, ItemIn, ItemOut, Version, \ PostImage, Comment STATIC_ROOT = BASE_DIR / 'static' # Create your views here. def choice_selector(t, n): return forms.ModelMultipleChoiceField(queryset=Choice.objects.filter(type=t) .annotate(ver=Replace('name', Value('.'), Value('~'))) .order_by("-ver"), label=n, required=False, widget=forms.CheckboxSelectMultiple) def generate_active_code(length=10): characters = string.ascii_letters + string.digits return ''.join(random.choice(characters) for _ in range(length)) def get_all_sub_functions(function): sub_functions = [function] for sub_function in function.parent_function.all(): # 假设 related_name='parent_function' sub_functions.extend(get_all_sub_functions(sub_function)) return sub_functions class IndexView(TemplateView): template_name = "index.html" def get_context_data(self, **kwargs): return { "form": SearchForm(initial={'order': 'down', 'loader_type': 'any', 'mc_type': 'any', 'create_type': 'any'}), "function": Function.objects.all()} class TreeView(TemplateView): template_name = "tree.html" def get_context_data(self, **kwargs): return {"functions": Function.objects.filter(parent=None).prefetch_related(), "form": SearchForm( initial={'order': 'down', 'loader_type': 'any', 'mc_type': 'any', 'create_type': 'any'})} class SearchForm(forms.Form): q = forms.CharField(label='搜索', required=False) product = forms.CharField(label='产物', required=False) consume = forms.CharField(label='消耗', required=False) loader = choice_selector('loader', "加载器") loader_type = forms.ChoiceField(label="匹配", choices=[("any", "所有"), ("all", "任意")], widget=forms.RadioSelect) mc_ver = choice_selector('mc_ver', "我的世界版本") mc_type = forms.ChoiceField(label="匹配", choices=[("any", "所有"), ("all", "任意")], widget=forms.RadioSelect) create_ver = choice_selector("create_ver", "机械动力版本") create_type = forms.ChoiceField(label="匹配", choices=[("any", "所有"), ("all", "任意")], widget=forms.RadioSelect) function = forms.ModelChoiceField(queryset=Function.objects.all(), label='功能类型', required=False) working_behavior = choice_selector("working_behavior", "工作特性") space_behavior = choice_selector("space_behavior", "空间特性") design = choice_selector("design", "设计理念") x_size = forms.IntegerField(label="X轴最大尺寸", required=False) y_size = forms.IntegerField(label="Y轴最大尺寸", required=False) z_size = forms.IntegerField(label="Z轴最大尺寸", required=False) sort = forms.ChoiceField(label="排序", choices=[("time", "发布日期"), ("speed", "产量"), ("download", "下载量")]) order = forms.ChoiceField(label="顺序", choices=[("up", "arrow-up.svg"), ("down", "arrow-down.svg")], widget=forms.RadioSelect) page = forms.IntegerField(label="页", required=False) class SearchView(View): def post(self, request, *args, **kwargs): search = SearchForm(request.POST) context = { 'posts': [], 'count': 0, 'page': 0, 'current_page': 0, } if not search.is_valid(): return render(self.request, "result/search.html", context) # 获取表单数据 mc_versions = search.cleaned_data.get('mc_ver') create_versions = search.cleaned_data.get('create_ver') loaders = search.cleaned_data.get('loader') loader_type = search.cleaned_data.get('loader_type') mc_type = search.cleaned_data.get('mc_type') create_type = search.cleaned_data.get('create_type') working_behavior = search.cleaned_data.get('working_behavior') space_behavior = search.cleaned_data.get('space_behavior') design = search.cleaned_data.get('design') x_size = search.cleaned_data.get('x_size') y_size = search.cleaned_data.get('y_size') z_size = search.cleaned_data.get('z_size') root_function = search.cleaned_data.get('function') consume = search.cleaned_data.get('consume') product = search.cleaned_data.get('product') sort = search.cleaned_data.get('sort') order = search.cleaned_data.get('order') page = search.cleaned_data.get('page') or 1 # 初始查询集，加入 prefetch_related posts = Post.objects.filter(posted=True, removed=False).prefetch_related( 'version__mc_ver', 'version__create_ver', 'version__loader', 'itemin_set__item', 'itemout_set__item', 'blueprint_set', 'working_behavior', 'space_behavior', 'design', 'function', ) # 模糊搜索 q = search.cleaned_data.get('q') if q: posts = posts.filter( Q(title__icontains=q) | Q(author__icontains=q) | Q(description__icontains=q) | Q(content__icontains=q) ) # 构建“必须满足所有条件”的查询 def add_all_conditions(queryset, field, values, field_type): if values and field_type == "all": for value in values: queryset = queryset.filter(**{field: value}) return queryset posts = add_all_conditions(posts, 'version__mc_ver', mc_versions, mc_type) posts = add_all_conditions(posts, 'version__create_ver', create_versions, create_type) posts = add_all_conditions(posts, 'version__loader', loaders, loader_type) # 构建“任意满足条件”的查询 any_conditions = Q() if mc_versions and mc_type == "any": any_conditions |= Q(version__mc_ver__in=mc_versions) if create_versions and create_type == "any": any_conditions |= Q(version__create_ver__in=create_versions) if loaders and loader_type == "any": any_conditions |= Q(version__loader__in=loaders) if any_conditions: posts = posts.filter(any_conditions) # 其他条件过滤 def add_filter(queryset, field, values): return queryset.filter(**{field: values}) if values else queryset posts = add_filter(posts, 'working_behavior__in', working_behavior) posts = add_filter(posts, 'space_behavior__in', space_behavior) posts = add_filter(posts, 'design__in', design) posts = add_filter(posts, 'x_size__lte', x_size) posts = add_filter(posts, 'y_size__lte', y_size) posts = add_filter(posts, 'z_size__lte', z_size) if root_function: posts = posts.filter(function__in=get_all_sub_functions(root_function)) if consume: consume_tags = Tag.objects.filter( Q(name__icontains=consume) | Q(sec_name__icontains=consume), type='item' ) posts = posts.filter(itemin__item__in=consume_tags) if product: product_tags = Tag.objects.filter( Q(name__icontains=product) | Q(sec_name__icontains=product), type='item' ) posts = posts.filter(itemout__item__in=product_tags) # 排序 order_prefix = '' if order == 'up' else '-' if sort == 'time': posts = posts.order_by(f'{order_prefix}create_time') elif sort == 'speed': posts = posts.annotate(speed=Max('itemout__count')).order_by(f'{order_prefix}speed') elif sort == 'download': posts = posts.annotate(download_count=Sum('blueprint__downloads')).order_by(f'{order_prefix}download_count') # 分页 paginator = Paginator(posts.distinct(), 20) posts_page = paginator.get_page(page) # 数据处理 for post in posts_page: post.mc_vers = post.version.values_list('mc_ver__name', flat=True).distinct().order_by('-mc_ver__name') post.mc_ver = post.mc_vers[0] if post.mc_vers else '-.--.-' post.mc_count = post.mc_vers.count() - 1 post.mc_vers = post.mc_vers[1:] post.create_vers = post.version.values_list('create_ver__name', flat=True).distinct().order_by( '-create_ver__name') post.create_ver = post.create_vers[0] if post.create_vers else '-.-.--' post.create_count = post.create_vers.count() - 1 post.create_vers = post.create_vers[1:] post.item_in = post.itemin_set.order_by('-count')[:3] post.item_out = post.itemout_set.order_by('-count')[:3] post.download = post.blueprint_set.aggregate(Sum('downloads')).get('downloads__sum', 0) or 0 context = { 'posts': posts_page, 'page': paginator.page_range, 'last_page': paginator.num_pages, 'current_page': page, 'count': paginator.count, } return render(self.request, "result/search.html", context) class DetailView(TemplateView): template_name = "detail.html" def get_context_data(self, **kwargs): post = get_object_or_404( Post.objects.prefetch_related( 'comment_set__user', 'version', 'verify_set__version', 'link', 'depend', 'itemin_set__item', 'itemout_set__item', 'blueprint_set', ).select_related( 'function', 'working_behavior', 'space_behavior', 'design', ), pk=self.kwargs["pk"], posted=True, removed=False, ) # 处理蓝图的材料统计 def count_inv(blueprints): bps = [] for bp in blueprints: items = [] if bp.inventory: try: inventory_data = json.loads(bp.inventory) for k, v in inventory_data.items(): try: item = Tag.objects.get(sec_name__endswith=k.replace(':', '.')) item.count = v items.append(item) except Tag.DoesNotExist: pass except json.JSONDecodeError: pass bp.inv = items bps.append(bp) return bps post.comments = post.comment_set.filter(reply_to__isnull=True) post.vers = post.version.all() post.vver = post.verify_set.all() post.unver = Version.objects.exclude(pk__in=post.vver.values_list('version', flat=True)) post.cleaned_link = post.link.filter(posted=True, removed=False) post.bps = count_inv(post.blueprint_set.all()) return { 'post': post, 'vff': VerifyForm(), } class CommentForm(forms.Form): comment = forms.CharField(widget=forms.Textarea) reply = forms.ModelChoiceField(queryset=Comment.objects.filter(post__posted=True, post__removed=False), required=False) class CommentView(PermissionRequiredMixin, View): permission_required = 'home.send_post' permission_denied_message = '你已被封禁' def post(self, request, *args, **kwargs): post = get_object_or_404(Post, pk=self.kwargs["pk"], posted=True, removed=False) form = CommentForm(request.POST) if form.is_valid(): comment = Comment(post=post, content=form.cleaned_data['comment'], reply_to=form.cleaned_data['reply'], user=request.user) comment.save() return HttpResponse("") return HttpResponseForbidden("") class VerifyForm(forms.Form): version = forms.ModelChoiceField(queryset=Version.objects.all()) can_use = forms.BooleanField(required=False) class VerifyView(PermissionRequiredMixin, View): permission_required = 'home.send_post' permission_denied_message = '你已被封禁' def post(self, request, *args, **kwargs): post = get_object_or_404(Post, pk=self.kwargs["pk"], posted=True, removed=False) form = VerifyForm(request.POST) if form.is_valid(): verify, c = post.verify_set.get_or_create(version=form.cleaned_data['version']) if form.cleaned_data['can_use']: verify.agree.add(self.request.user) if verify.disagree.filter(pk=request.user.pk).exists(): verify.disagree.remove(self.request.user) else: verify.disagree.add(self.request.user) if verify.agree.filter(pk=request.user.pk).exists(): verify.agree.remove(self.request.user) verify.save() if verify.disagree.count() > verify.agree.count(): if verify.version in post.version.all(): post.version.remove(verify.version) else: post.version.add(verify.version) return HttpResponse("") return HttpResponseForbidden("") class DownloadView(TemplateView): template_name = "download.html" def dispatch(self, request, *args, **kwargs): # 获取当前时间 now = timezone.now().timestamp() # 检查是否被限制下载 block_until = request.session.get('block_download', now) if block_until > now: return HttpResponse("服务器拒绝冲泡咖啡", status=418) # 检查下载频率 last_download = request.session.get('last_download', now) download_counter = request.session.get('download_counter', 0) if last_download + 60 > now: # 如果在一分钟内连续下载 download_counter += 1 else: download_counter = 1 # 重置计数 # 更新会话数据 request.session['last_download'] = now request.session['download_counter'] = download_counter # 如果下载次数超过限制，封锁一天 if download_counter > 19: request.session['block_download'] = now + 86400 # 一天 return super().dispatch(request, *args, **kwargs) def get_context_data(self, **kwargs): # 获取蓝图并更新下载计数 blueprint = get_object_or_404(Blueprint, pk=self.kwargs["pk"]) Blueprint.objects.filter(pk=blueprint.pk).update(downloads=F('downloads') + 1) # 返回下载 URL 和文件名 return { "url": blueprint.file.url, "name": blueprint.name } class TagView(View): def get(self, request, *args, **kwargs): try: tag = Tag.objects.get(pk=self.kwargs["pk"]) return JsonResponse( {'name': tag.name, 'sec_name': tag.sec_name, 'ico': tag.ico.url, 'description': tag.description}) except Tag.DoesNotExist: return JsonResponse({'name': '', 'sec_name': '', 'ico': '', 'description': ''}) class SearchTagView(View): def post(self, request, *args, **kwargs): name = self.request.POST.get('name', '') item = Tag.objects.filter( (Q(name__icontains=name) | Q(sec_name__icontains=name)) & Q(type=self.request.POST.get('type', 'item')))[:30] return render(self.request, 'result/tag_search.html', {'items': item}) class SearchLinkView(View): def post(self, request, *args, **kwargs): name = self.request.POST.get('name', '') q = Q(title__icontains=name) try: pk = int(name) q |= Q(pk=pk) except ValueError: pass q &= Q(posted=True, removed=False) posts = Post.objects.filter(q) return render(self.request, 'result/link_search.html', {'posts': posts}) class ProfileView(LoginRequiredMixin, TemplateView): template_name = "profile.html" def get_context_data(self, **kwargs): return { "codes": ActiveCode.objects.filter(creator=self.request.user), 'drafts': Post.objects.filter(posted=False, sender=self.request.user, removed=False), 'posts': Post.objects.filter(posted=True, sender=self.request.user, removed=False).annotate( downloads=Sum('blueprint__downloads')).order_by('-create_time'), } def is_staff(user): return user.is_staff and user.groups.filter(name='管理员').exists() class PostActionForm(forms.Form): post = forms.ModelChoiceField(queryset=Post.objects.filter(removed=False), label='发布') action = forms.ChoiceField(label='操作', choices=[('unpost', '撤下'), ('copy', '复制'), ('delete', '删除'), ('rebuild', '重建预览')]) class ProfileActionView(LoginRequiredMixin, View): def post(self, request, *args, **kwargs): form = PostActionForm(self.request.POST) if not form.is_valid(): return HttpResponseForbidden("无效的请求。") post = form.cleaned_data['post'] action = form.cleaned_data['action'] # 检查 post 是否属于当前用户 if post.sender != self.request.user and not is_staff(self.request.user): return HttpResponseForbidden("你无权操作此发布内容。") # 检查 post 是否未被移除 if post.removed: return HttpResponseForbidden("该内容已被移除，无法执行操作。") if action == 'rebuild': if is_staff(self.request.user): process_blueprints(post) else: return HttpResponseForbidden("你无权操作此发布内容。") if action == 'unpost': # 将 posted 设置为 False post.posted = False post.save() elif action == 'copy' and post.sender == request.user: # 使用事务处理复制操作 with transaction.atomic(): # 创建新 Post 并复制基本字段 new_post = Post.objects.create( sender=request.user, title=f"{post.title} - 复制", description=post.description, content=post.content, function=post.function, working_behavior=post.working_behavior, space_behavior=post.space_behavior, design=post.design, stress=post.stress, delay=post.delay, x_size=post.x_size, y_size=post.y_size, z_size=post.z_size, posted=False # 复制的内容不直接发布 ) # 复制多对多关系 new_post.version.set(post.version.all()) new_post.depend.set(post.depend.all()) new_post.link.set(post.link.all()) # 复制 ItemIn 和 ItemOut 相关数据 for item_in in post.itemin_set.all(): ItemIn.objects.create( item=item_in.item, post=new_post, count=item_in.count ) for item_out in post.itemout_set.all(): ItemOut.objects.create( item=item_out.item, post=new_post, count=item_out.count ) # 复制封面图像，避免多个对象指向同一图片 if post.cover: new_post.cover.save( post.cover.name, ContentFile(post.cover.read()), save=True ) # 复制 PostImage 实例并避免共享图片文件 for post_image in post.postimage_set.all(): new_post_image = PostImage.objects.create( post=new_post, image=ContentFile(post_image.image.read(), post_image.image.name) ) new_post_image.save() new_post.content.replace(post_image.image.name, new_post_image.image.name) elif action == 'delete': if post.posted: return HttpResponseForbidden("无法删除已发布的内容。") download_count = post.blueprint_set.aggregate(download_count=Sum('downloads')).get('download_count') or 0 if download_count > 10: post.removed = True post.save() else: post.delete() return HttpResponse("") class GenerateActiveCodeView(PermissionRequiredMixin, View): permission_required = 'home.generate_active_code' permission_denied_message = '你没有权限这么做' def post(self, request, *args, **kwargs): while True: try: code = ActiveCode(code=generate_active_code()) code.creator = self.request.user code.save() break except IntegrityError: pass return HttpResponseRedirect(reverse("profile") + "#tab3") class SimplifyFileInput(ClearableFileInput): template_name = "widget/simplify_file_input.html" PostForm = modelform_factory(Post, exclude=('create_time', 'sender', 'removed'), widgets={ "version": forms.CheckboxSelectMultiple, 'depend': forms.CheckboxSelectMultiple, 'content': forms.HiddenInput, 'cover': SimplifyFileInput, 'link': forms.CheckboxSelectMultiple, 'author': forms.TextInput(attrs={'placeholder': "作者（们）"}), 'description': forms.Textarea(attrs={'placeholder': "简单描述一下这个蓝图？"}) }, labels={'version': '版本列表', 'cover': '封面'}) PartPostForm = modelform_factory(Post, exclude=('create_time', 'sender', 'cover', 'content', 'removed', 'posted'), widgets={ "version": forms.CheckboxSelectMultiple, 'depend': forms.CheckboxSelectMultiple, 'link': forms.CheckboxSelectMultiple, 'author': forms.TextInput(attrs={'placeholder': "作者（们）"}), 'description': forms.Textarea(attrs={'placeholder': "简单描述一下这个蓝图？"}) }, labels={'version': '版本列表'}) ItemInFormSet = inlineformset_factory(Post, ItemIn, exclude=(), labels={'item': '物品', 'count': '数量'}, widgets={'item': forms.HiddenInput}, extra=0, max_num=200, absolute_max=200) ItemOutFormSet = inlineformset_factory(Post, ItemOut, exclude=(), labels={'item': '物品', 'count': '数量'}, widgets={'item': forms.HiddenInput}, extra=0, max_num=200, absolute_max=200) BlueprintFormSet = inlineformset_factory(Post, Blueprint, fields=('notes', 'name', 'file'), labels={'notes': '注释', 'name': '名称', 'file': '蓝图文件'}, widgets={'file': SimplifyFileInput(attrs={'accept': ".nbt"})}, extra=0, max_num=100, absolute_max=100) def ver_and_ft(post: Post): """ 可用版本和分类树 """ vers = [] checked_ver = post.version.all() for ver in Version.objects.all(): if ver in checked_ver: ver.checked = True vers.append(ver) ft = '' if post.function: f = post.function.parent ft = str(post.function.pk) while f is not None: ft = str(f.pk) + "," + ft f = f.parent return vers, ft def addup_errors(post_form, inline_forms) -> str: error_text = "" for field, errors in post_form.errors.items(): label = post_form.fields[field].label error_text += f"{label}: " for err in errors: error_text += f"{err}" for inline_form in inline_forms: for err_dic in inline_form.errors: for field, errors in err_dic.items(): label = inline_form.form.base_fields[field].label error_text += f"{label}: " for err in errors: error_text += f"{err}" return error_text # 允许的标签（与示例文本和 TinyMCE 一致） ALLOWED_TAGS = [ 'h1', 'h6', 'p', 'br', 'hr', 'div', 'span', 'blockquote', 'code', 'pre', 'a', 'strong', 'em', 's', 'u', 'sub', 'sup', 'b', 'i', 'ol', 'ul', 'li', 'table', 'thead', 'tbody', 'tr', 'th', 'td', 'colgroup', 'col', 'video', 'source', 'img', 'iframe', 'details', 'summary' ] # 允许的属性 ALLOWED_ATTRIBUTES = { '*': ['style', 'class', 'id'], # 全局属性 'a': ['href', 'title', 'target', 'referrerpolicy', 'rel'], # 链接属性 'img': ['src', 'alt', 'width', 'height'], # 图片属性 'video': ['controls', 'width', 'height'], # 视频属性 'source': ['src'], # 视频源 'col': ['span'], # 表格列 'td': ['colspan', 'rowspan'], # 表格单元格 'th': ['colspan', 'rowspan'], # 表格表头 'iframe': ['src', 'width', 'height', 'frameborder', 'allowfullscreen'], 'details': ['open'] } css_sanitizer = CSSSanitizer() cleaner = Cleaner(tags=ALLOWED_TAGS, attributes=ALLOWED_ATTRIBUTES, css_sanitizer=css_sanitizer, strip=True, strip_comments=False, protocols=['http', 'https']) def bleach_clean(html_content): """ 使用 bleach 清理 HTML 内容。 :param html_content: 用户输入的富文本内容 :return: 经过清理的安全 HTML """ return cleaner.clean(html_content) def process_blueprints(post): counter = 0 size = 0 for blueprint in post.blueprint_set.all(): size += blueprint.file.size if size > 1024 * 2560: blueprint.delete() continue if not post.posted: continue counter += 1 blueprint.model = None blueprint.inventory = None # 校验文件签名 file_signature = blueprint.file.read(10) if file_signature != b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff': blueprint.warn = 'invalid' blueprint.save() continue # 检查文件大小 if blueprint.file.size > 256 * 1024: blueprint.warn = 'tooBig' blueprint.save() continue blueprint.warn = None blueprint.save() # 分析蓝图文件 if counter <= 5 and blueprint.file.size < 50 * 1024: blueprint.model, blueprint.inventory = analyze_blueprint(blueprint.file.path) blueprint.save() def cleanup_unused_images(post): for image in post.postimage_set.all(): if image.image.name not in post.content: image.delete() class PostView(LoginRequiredMixin, UserPassesTestMixin, View): def test_func(self): if not self.request.user.has_perm('home.send_post'): return False # 如果是编辑已有草稿，直接通过 if 'pk' in self.kwargs: return True # 检查是否有未发布且 version 为空的草稿 if Post.objects.filter(sender=self.request.user, posted=False, removed=False, version__isnull=True).exists(): return True # 检查草稿数量限制 max_draft = 9 if self.request.user.has_perm('home.infinite_send_post') else 3 if Post.objects.filter(sender=self.request.user, posted=False, removed=False).count() >= max_draft: return False return True def get_permission_denied_message(self): if not self.request.user.has_perm('home.send_post'): return '你已被封禁' return '你不能产生更多草稿了' def get(self, request, *args, **kwargs): # 如果没有提供 pk，尝试打开未发布的 version 为空的草稿 if 'pk' not in self.kwargs: existing_draft = Post.objects.filter( sender=self.request.user, posted=False, removed=False, version__isnull=True ).first() if existing_draft: return redirect('post', existing_draft.pk) # 创建新草稿 post = Post.objects.create(sender=self.request.user) return redirect('post', post.pk) # 编辑指定草稿 post = get_object_or_404(Post, pk=self.kwargs.get('pk'), posted=False, removed=False, sender=self.request.user) context = { 'post': PostForm(instance=post), 'blueprints': BlueprintFormSet(instance=post), 'itemin': ItemInFormSet(instance=post), 'itemout': ItemOutFormSet(instance=post), 'function': Function.objects.all(), 'ft': ver_and_ft(post)[1], 'vers': ver_and_ft(post)[0], 'links': post.link.all(), 'depends': post.depend.all(), 'pk': post.pk, } return render(request, "post.html", context) def post(self, request, *args, **kwargs): post = get_object_or_404(Post, pk=self.kwargs.get('pk'), posted=False, removed=False, sender=self.request.user) pf = PostForm(request.POST, request.FILES, instance=post) bf = BlueprintFormSet(request.POST, request.FILES, instance=post) iof = ItemOutFormSet(request.POST, instance=post) iif = ItemInFormSet(request.POST, instance=post) error_text = addup_errors(pf, [bf, iof, iif]) if error_text: return HttpResponse(error_text) with transaction.atomic(): # 保存表单 pf.save() bf.save() iof.save() iif.save() # 清理内容 post.content = bleach_clean(post.content) post.save() # 检查是否可以发布 if pf.data.get('posted') and not self.handle_post_limit(request, post): return HttpResponse('你不能发布更多文章了') # 处理蓝图 process_blueprints(post) # 清理未使用的图片 cleanup_unused_images(post) return HttpResponse(error_text) def handle_post_limit(self, request, post): if not self.request.user.has_perm('home.infinite_send_post'): three_days_ago = timezone.now() - timedelta(days=3) post_count = Record.objects.filter(user=request.user, type='post', time__gte=three_days_ago).count() if post_count > 3: post.posted = False post.save() return False Record(user=request.user, type='post', comment=post.get_absolute_url()).save() return True class QuickEditView(PermissionRequiredMixin, View): permission_required = 'home.send_post' permission_denied_message = '你已被封禁' def get(self, request, *args, **kwargs): # 获取文章 post = self.get_post() # 生成表单和其他上下文数据 context = self.get_context_data(post) return render(request, "quickedit.html", context) def post(self, request, *args, **kwargs): # 获取文章 post = self.get_post() # 表单处理 pf = PartPostForm(request.POST, request.FILES, instance=post) iof = ItemOutFormSet(request.POST, instance=post) iif = ItemInFormSet(request.POST, instance=post) # 检查错误 error_text = addup_errors(pf, [iof, iif]) if not error_text: # 保存所有表单 with transaction.atomic(): pf.save() iof.save() iif.save() return HttpResponse(error_text) def get_post(self): """获取指定的 Post 对象""" pk = self.kwargs.get('pk') user = self.request.user if is_staff(user): return get_object_or_404(Post, pk=pk, removed=False) return get_object_or_404(Post, pk=pk, removed=False, sender=user) def get_context_data(self, post): """构建模板上下文数据""" pf = PostForm(instance=post) iof = ItemOutFormSet(instance=post) iif = ItemInFormSet(instance=post) vers, ft = ver_and_ft(post) return { 'post': pf, "itemin": iif, "itemout": iof, "function": Function.objects.all(), 'ft': ft, 'vers': vers, 'links': post.link.all(), 'depends': post.depend.all(), 'pk': post.pk, 'cover': post.cover, } class UploadPostImg(LoginRequiredMixin, UserPassesTestMixin, View): def test_func(self): if not self.request.user.has_perm('home.send_post'): return False post = get_object_or_404(Post, pk=self.kwargs.get('pk')) if post.sender != self.request.user: return False if post.postimage_set.count() > 40: return False return True def get_permission_denied_message(self): if not self.request.user.has_perm('home.send_post'): return '你已被封禁' else: return '你不能这么做' def post(self, request, *args, **kwargs): post = get_object_or_404(Post, pk=self.kwargs.get('pk')) post_img = PostImage(post=post, image=request.FILES['file']) post_img.save() return JsonResponse({'location': post_img.image.url}) class RegisterForm(UserCreationForm): code = forms.CharField(max_length=16, required=True) class Meta: model = User fields = ("username", "email",) def clean_code(self): code = self.cleaned_data.get("code") err = ValidationError( "激活码不可用", code="code_invalid", ) try: active_code = ActiveCode.objects.get(code=code) except ActiveCode.DoesNotExist: raise err if not active_code.valid: raise err if active_code.email is not None and active_code.email != self.cleaned_data["email"]: raise err if active_code.valid_time is not None and active_code.valid_time < timezone.now(): raise err return code def save(self, commit=True): user = super().save(commit=True) active_code = ActiveCode.objects.get(code=self.cleaned_data.get("code")) active_code.user = user active_code.valid = False active_code.save() user_group = Group.objects.get(name="用户") user.groups.add(user_group) return user class RegisterView(FormView): template_name = "registration/register.html" form_class = RegisterForm success_url = reverse_lazy('login') @method_decorator(sensitive_post_parameters()) @method_decorator(csrf_protect) @method_decorator(never_cache) def dispatch(self, request, *args, **kwargs): return super().dispatch(request, *args, **kwargs) def form_valid(self, form): form.save() return super().form_valid(form) class ProtectLoginView(LoginView): def post(self, request, *args, **kwargs): # 获取会话中的登录尝试次数 login_attempts = request.session.get('login_attempts', 0) if login_attempts >= 10: form = self.get_form() form.add_error(None, "用户名或密码错误") return self.form_invalid(form) # 增加尝试计数并保存到会话 request.session['login_attempts'] = login_attempts + 1 return super().post(request, *args, **kwargs) def form_valid(self, form): # 登录成功时重置尝试计数 if 'login_attempts' in self.request.session: del self.request.session['login_attempts'] return super().form_valid(form) class UpdateView(LoginRequiredMixin, UserPassesTestMixin, View): def test_func(self): return self.request.user.is_staff def get(self, request, *args, **kwargs): return render(self.request, "staff/update.html") class UpdateAssetsView(PermissionRequiredMixin, View): permission_required = 'home.upload_assets' permission_denied_message = '你没有权限这么做' def get(self, request, *args, **kwargs): updating = False if thread_lock.locked(): updating = True elif any((BASE_DIR / 'tmp').iterdir()): t = threading.Thread(target=update_assets, args=[self.request.user]) t.start() updating = True last_record = Record.objects.filter(type='update').order_by('-time').first() if 'check' in request.GET: return HttpResponse(updating) context = { "assets_count": len(Assets.objects.all()), 'tag_count': len(Tag.objects.all()), 'last_record': last_record, 'updating': updating } return render(self.request, "staff/update_assets.html", context=context) def post(self, request, *args, **kwargs): if thread_lock.locked(): return HttpResponse("更新正在进行……") files = (BASE_DIR / 'tmp').iterdir() if any(files): t = threading.Thread(target=update_assets, args=[self.request.user]) t.start() return HttpResponse("更新正在进行……") if 'mod_file' not in self.request.FILES: return HttpResponse("上传模组文件") mod_file = self.request.FILES['mod_file'] if not mod_file.name.endswith(".jar"): return HttpResponse("上传模组文件") with (BASE_DIR / "tmp_mod_file").open('wb') as f: for chunk in mod_file.chunks(): f.write(chunk) t = threading.Thread(target=update_assets, args=[self.request.user]) t.start() return HttpResponse("更新正在进行……") class UpdateItemForm(forms.Form): name = forms.CharField(required=False) sec_name = forms.CharField() description = forms.CharField(required=False) img = forms.CharField(required=False) def tag_context(get, t): q = Q(type=t) if 'q' in get: name = get['q'] q &= Q(name__icontains=name) | Q(description__icontains=name) | Q(sec_name__icontains=name) items = Tag.objects.filter(q).order_by('ico', 'sec_name') pages = Paginator(items, 200) page = int(get.get('page', 1)) context = { 'items': pages.page(page) if pages.count > 0 else [], 'count': pages.count, 'page': pages.page_range, 'current_page': page, 'q': get.get('q', ''), } return context def update_tag(tag, form): if form.cleaned_data['img']: image_data = form.cleaned_data['img'] f, i = image_data.split(';base64,') data = ContentFile(base64.b64decode(i)) tag.ico.save(tag.sec_name + '.png', data, save=False) if form.cleaned_data['description']: tag.description = form.cleaned_data['description'] if form.cleaned_data['name']: tag.name = form.cleaned_data['name'] tag.save() class UpdateItemView(PermissionRequiredMixin, View): permission_required = "home.update_tag" permission_denied_message = '你没有权限这么做' def get(self, request, *args, **kwargs): return render(self.request, 'staff/update_item.html', context=tag_context(request.GET, 'item')) def post(self, request, *args, **kwargs): form = UpdateItemForm(self.request.POST) if form.is_valid(): tag = get_object_or_404(Tag, sec_name=form.cleaned_data['sec_name'], type='item') update_tag(tag, form) if tag.ico: return HttpResponse(tag.ico.url) return HttpResponse('') class GetItemModelView(View): def get(self, request, *args, **kwargs): item = Tag.objects.get(sec_name=self.kwargs['item']) if not item.generated: refresh_single_item_ico(item) return JsonResponse({}) return JsonResponse(get_item_face(self.kwargs['item'])) class UpdateDependView(PermissionRequiredMixin, View): permission_required = "home.update_tag" permission_denied_message = '你没有权限这么做' def get(self, request, *args, **kwargs): return render(self.request, 'staff/update_depend.html', context=tag_context(self.request.GET, 'depend')) def post(self, request, *args, **kwargs): form = UpdateItemForm(self.request.POST) if form.is_valid(): tag, c = Tag.objects.get_or_create(sec_name=form.cleaned_data['sec_name'], type='depend') update_tag(tag, form) if tag.ico: return HttpResponse(tag.ico.url) return HttpResponse('') class AddChoiceForm(forms.Form): type = forms.ChoiceField( choices=[("working_behavior", "工作特性"), ("space_behavior", "空间特性"), ("design", "设计理念")]) name = forms.CharField() class AddVersionForm(forms.Form): mc = forms.CharField() create = forms.CharField() loader = forms.CharField() class UpdateChoiceView(PermissionRequiredMixin, View): permission_required = "home.edit_choice" permission_denied_message = '你没有权限这么做' def get(self, request, *args, **kwargs): context = { 'working_behavior': Choice.objects.filter(type="working_behavior").annotate( post_count=Count('working_behavior')), 'space_behavior': Choice.objects.filter(type="space_behavior").annotate(post_count=Count('space_behavior')), 'design': Choice.objects.filter(type="design").annotate(post_count=Count('design')), 'mc_ver': Choice.objects.filter(type="mc_ver"), 'create_ver': Choice.objects.filter(type="create_ver"), 'loader': Choice.objects.filter(type="loader"), 'version': Version.objects.all().annotate(post_count=Count('post_version')), } return render(self.request, 'staff/update_choice.html', context=context) def post(self, request, *args, **kwargs): if 'type' in request.POST: form = AddChoiceForm(self.request.POST) if form.is_valid(): choice, c = Choice.objects.get_or_create(type=form.cleaned_data['type'], name=form.cleaned_data['name']) choice.save() else: form = AddVersionForm(self.request.POST) if form.is_valid(): c_mc, c = Choice.objects.get_or_create(type='mc_ver', name=form.cleaned_data['mc']) c_create, c = Choice.objects.get_or_create(type='create_ver', name=form.cleaned_data['create']) c_loader, c = Choice.objects.get_or_create(type='loader', name=form.cleaned_data['loader']) _, c = Version.objects.get_or_create(mc_ver=c_mc, create_ver=c_create, loader=c_loader) return HttpResponseRedirect(reverse('update') + "#tab4") class AddFunctionForm(forms.Form): parent = forms.ModelChoiceField(queryset=Function.objects.all(), required=False) name = forms.CharField() class UpdateFunctionView(PermissionRequiredMixin, View): permission_required = "home.edit_function" permission_denied_message = '你没有权限这么做' def get(self, request, *args, **kwargs): context = { 'functions': Function.objects.filter(parent=None).prefetch_related() } return render(self.request, 'staff/update_function.html', context=context) def post(self, request, *args, **kwargs): form = AddFunctionForm(self.request.POST) if form.is_valid(): Function.objects.get_or_create(name=form.cleaned_data['name'], parent=form.cleaned_data['parent']) return HttpResponseRedirect(reverse('update') + "#tab5") class UserActionForm(forms.Form): user = forms.ModelChoiceField(queryset=User.objects.all(), label='用户') action = forms.ChoiceField(label='操作', choices=[('ban', '封禁'), ('unban', '解封'), ('cert', '认证'), ('uncert', '取消认证')]) class CommentActionForm(forms.Form): comment = forms.ModelChoiceField(queryset=Comment.objects.all(), label='评论') action = forms.ChoiceField(label='操作', choices=[('this', '这个'), ('full', '整楼')]) class ManagePostView(LoginRequiredMixin, UserPassesTestMixin, View): def test_func(self): return self.request.user.is_staff and self.request.user.groups.filter(name='管理员').exists() def get(self, request, *args, **kwargs): post = get_object_or_404(Post, pk=self.kwargs.get('pk'), removed=False) pf = PostForm(instance=post) iof = ItemOutFormSet(instance=post) iif = ItemInFormSet(instance=post) vers, ft = ver_and_ft(post) post.downloads = 0 post.blueprint_total_size = 0 post.blueprint_count = 0 for blueprint in post.blueprint_set.all(): post.downloads += blueprint.downloads post.blueprint_total_size += blueprint.file.size post.blueprint_count += 1 post.img_count = 0 post.img_total_size = 0 for img in post.postimage_set.all(): post.img_count += 1 post.img_total_size += img.image.size post.blueprint_total_size_k = '{:.2f}'.format(post.blueprint_total_size / 1024) post.comments = post.comment_set.filter(reply_to__post=None) return render(request, 'staff/manage.html', context={'post': pf, 'postData': post, "itemin": iif, "itemout": iof, "function": Function.objects.all(), 'ft': ft, 'vers': vers, 'links': post.link.all(), 'depends': post.depend.all(), 'pk': post.pk}) def post(self, request, *args, **kwargs): form = CommentActionForm(self.request.POST) if not form.is_valid(): return HttpResponseForbidden("无效的请求。") comment = form.cleaned_data['comment'] if comment.comment_set.exclude(content='').exists() and form.cleaned_data['action'] != 'full': comment.user = None comment.content = '' comment.save() else: # 如果 是回复 评论已删 评论下面只有它了 那么删楼 if comment.reply_to and comment.reply_to.content == "" and not comment.reply_to.comment_set.exclude( content='').exclude(pk=comment.pk).exists(): comment.reply_to.delete() else: comment.delete() return HttpResponse('') class SearchUserForm(forms.Form): q = forms.CharField(label='搜索', required=False) class ManageUserView(LoginRequiredMixin, UserPassesTestMixin, View): def test_func(self): return is_staff(self.request.user) def get(self, request, *args, **kwargs): form = SearchUserForm(self.request.GET) if not form.is_valid(): return HttpResponseForbidden("无效的请求。") if form.cleaned_data['q']: q = Q(username__icontains=form.cleaned_data['q']) | Q( email__icontains=form.cleaned_data['q']) | Q(user__creator__username__icontains=form.cleaned_data['q']) user = User.objects.filter(q) else: user = User.objects.all() user = user.annotate(post_count=Count('post')).order_by('-last_login') pages = Paginator(user, 50) page = int(self.request.GET.get('page', 1)) context = { 'users': pages.page(page) if pages.count > 0 else [], 'count': pages.count, 'page': pages.page_range, 'current_page': page, 'q': self.request.GET.get('q', ''), } return render(self.request, 'staff/user_manager.html', context=context) def post(self, request, *args, **kwargs): form = UserActionForm(self.request.POST) if not form.is_valid(): return HttpResponseForbidden("无效的请求。") user: User = form.cleaned_data['user'] action = form.cleaned_data['action'] user_group = Group.objects.get(name="用户") cert_group = Group.objects.get(name="认证作者") if action == 'ban': user.groups.remove(user_group) elif action == 'unban': user.groups.add(user_group) elif action == 'cert': user.groups.add(cert_group) elif action == 'uncert': user.groups.remove(cert_group) return HttpResponse('') class ThreeView(TemplateView): template_name = "three.html" def get_context_data(self, **kwargs): blueprint = get_object_or_404(Blueprint, pk=self.kwargs["pk"]) return {"model": blueprint.model, 'name': blueprint.name}