import base64 import io import urllib.parse from io import BytesIO from urllib.parse import parse_qsl, urlencode, urlparse, urlsplit, urlunsplit from django.conf import settings from django.contrib.auth.mixins import LoginRequiredMixin from django.core.mail import EmailMultiAlternatives from django.http import ( FileResponse, HttpResponse, HttpResponseBadRequest, JsonResponse, ) from django.utils import timezone from django.utils.decorators import method_decorator from django.utils.translation import gettext_lazy as _ from django.views import View from django.views.decorators.csrf import csrf_exempt from pdf2image import convert_from_bytes charts_map = { "UserLoginReport": { "title": _('User login report'), "path": "/ui/#/reports/users/user-activity" }, "UserChangePasswordReport": { "title": _('User change password report'), "path": "/ui/#/reports/users/change-password" }, "AssetStatistics": { "title": _('Asset statistics report'), "path": "/ui/#/reports/assets/asset-statistics" }, "AssetReport": { "title": _('Asset activity report'), "path": "/ui/#/reports/assets/asset-activity" }, "AccountStatistics": { "title": _('Account statistics report'), "path": "/ui/#/reports/accounts/account-statistics?days=30" }, "AccountAutomationReport": { "title": _('Account automation report'), "path": "/ui/#/reports/accounts/account-automation" }, "ConsoleDashboard": { "title": _('ConsoleDashboard'), "path": "/ui/#/reports/dashboard/console" }, "AuditsDashboard": { "title": _('AuditsDashboard'), "path": "/ui/#/reports/dashboard/audits" }, "PamDashboard": { "title": _('PamDashboard'), "path": "/ui/#/reports/dashboard/pam" }, "ChangeSecretDashboard": { "title": _('ChangeSecretDashboard'), "path": "/ui/#/reports/dashboard/change-secret" } } def get_trusted_site_url(): # Playwright navigation and cookie scope must never depend on request Host. site_url = getattr(settings, 'SITE_URL', '') if not isinstance(site_url, str): raise ValueError('Invalid SITE_URL') site_url = site_url.rstrip('/') parsed_site = urlparse(site_url) if ( not site_url or parsed_site.scheme not in ('http', 'https') or not parsed_site.hostname or parsed_site.username or parsed_site.password or parsed_site.query or parsed_site.fragment ): raise ValueError('Invalid SITE_URL') return site_url, parsed_site def build_report_url(chart_path): path = urllib.parse.unquote(chart_path) if not path.startswith('/ui/#/reports/'): raise ValueError('Invalid report path') site_url, _ = get_trusted_site_url() return site_url + path def export_chart_to_pdf(chart_name, sessionid, request): chart_info = charts_map.get(chart_name) if not chart_info: return None, None url = build_report_url(chart_info['path']) _, parsed_site = get_trusted_site_url() if settings.DEBUG_DEV: url = url.replace(":8080", ":9528") oid = request.COOKIES.get("X-JMS-ORG") request_data = request.POST if request.method == 'POST' else request.GET query = dict(parse_qsl(urlsplit(url).query, keep_blank_values=True)) for key, value in request_data.items(): if key in {'chart', 'csrfmiddlewaretoken'}: continue query[key] = value query.setdefault('days', 7) query['oid'] = oid parts = list(urlsplit(url)) parts[3] = urlencode(query, doseq=True) url = urlunsplit(parts) with sync_playwright() as p: lang = request.COOKIES.get(settings.LANGUAGE_COOKIE_NAME) browser = p.chromium.launch(headless=True) context = browser.new_context( viewport={"width": 1040, "height": 800}, locale=lang, ignore_https_errors=True ) # 设置 sessionid cookie context.add_cookies([ { 'name': settings.SESSION_COOKIE_NAME, 'value': sessionid, # hostname is derived from SITE_URL and never includes a port. 'domain': parsed_site.hostname, 'path': '/', 'httpOnly': True, 'secure': False, # 如有 https 可改 True } ]) page = context.new_page() try: page.goto(url, wait_until='networkidle') page.wait_for_function('window.echartsFinished === true', timeout=10000) page_title = page.title() print(f"Page title: {page_title}") pdf_bytes = page.pdf(format="A4", landscape=True, margin={"top": "35px", "bottom": "30px", "left": "20px", "right": "20px"}) except Exception as e: print(f'Playwright error: {e}') pdf_bytes = None page_title = chart_info['title'] finally: browser.close() return pdf_bytes, page_title @method_decorator(csrf_exempt, name='dispatch') class ExportPdfView(LoginRequiredMixin, View): def get(self, request): chart_name = request.GET.get('chart') return self._handle_export(request, chart_name) def post(self, request): chart_name = request.POST.get('chart') return self._handle_export(request, chart_name) def _handle_export(self, request, chart_name): if not request.user.is_authenticated: return HttpResponse(status=401) if not chart_name: return HttpResponseBadRequest('Missing chart parameter') sessionid = request.COOKIES.get(settings.SESSION_COOKIE_NAME) if not sessionid: return HttpResponseBadRequest('No sessionid found in cookies') try: pdf_bytes, title = export_chart_to_pdf( chart_name, sessionid, request=request ) except ValueError: return HttpResponseBadRequest('Invalid report configuration') if not pdf_bytes: return HttpResponseBadRequest('Failed to generate PDF') filename = f"{title}-{timezone.now().strftime('%Y%m%d%H%M%S')}.pdf" response = FileResponse(io.BytesIO(pdf_bytes), as_attachment=True, filename=filename, content_type='application/pdf') return response class SendMailView(LoginRequiredMixin, View): def post(self, request): if not request.user.is_authenticated: return HttpResponse(status=401) chart_name = request.GET.get('chart') or request.POST.get('chart') if not chart_name: return HttpResponseBadRequest('Missing chart parameter') email = request.user.email if not email: return HttpResponseBadRequest('Missing email parameter') sessionid = request.COOKIES.get(settings.SESSION_COOKIE_NAME) if not sessionid: return HttpResponseBadRequest('No sessionid found in cookies') # 1. 生成 PDF try: pdf_bytes, title = export_chart_to_pdf( chart_name, sessionid, request=request ) except ValueError: return HttpResponseBadRequest('Invalid report configuration') if not pdf_bytes: return HttpResponseBadRequest('Failed to generate PDF') # 2. PDF 转图片 images = convert_from_bytes(pdf_bytes, dpi=200) # 3. 图片转 base64 img_tags = [] for img in images: buffer = BytesIO() img.save(buffer, format="PNG") encoded = base64.b64encode(buffer.getvalue()).decode("utf-8") img_tags.append(f'') html_content = "
".join(img_tags) # 4. 发送邮件 subject = f"{title} 报表" from_email = settings.EMAIL_FROM or settings.EMAIL_HOST_USER to = [email] msg = EmailMultiAlternatives(subject, '', from_email, to) msg.attach_alternative(html_content, "text/html") filename = f"{title}-{timezone.now().strftime('%Y%m%d%H%M%S')}.pdf" msg.attach(filename, pdf_bytes, "application/pdf") try: msg.send() except Exception as e: return JsonResponse({"error": _('Failed to send email: ') + str(e)}) return JsonResponse({"message": _('Email sent successfully to ') + email})