jumpserver/apps/audits/utils.py

125 lines
4.1 KiB
Python

import codecs
import copy
import csv
from itertools import chain
from django.db import models
from django.http import HttpResponse
from common.utils import validate_ip, get_ip_city, get_logger
from settings.serializers import SettingsSerializer
from .const import DEFAULT_CITY
logger = get_logger(__name__)
def get_excel_response(filename):
excel_response = HttpResponse(content_type='text/csv')
excel_response[
'Content-Disposition'] = 'attachment; filename="%s"' % filename
excel_response.write(codecs.BOM_UTF8)
return excel_response
def write_content_to_excel(response, header=None, login_logs=None, fields=None):
writer = csv.writer(response, dialect='excel', quoting=csv.QUOTE_MINIMAL)
if header:
writer.writerow(header)
if login_logs:
for log in login_logs:
data = [getattr(log, field.name) for field in fields]
writer.writerow(data)
return response
def write_login_log(*args, **kwargs):
from audits.models import UserLoginLog
ip = kwargs.get('ip') or ''
if not (ip and validate_ip(ip)):
ip = ip[:15]
city = DEFAULT_CITY
else:
city = get_ip_city(ip) or DEFAULT_CITY
kwargs.update({'ip': ip, 'city': city})
UserLoginLog.objects.create(**kwargs)
def get_resource_display(resource):
resource_display = str(resource)
setting_serializer = SettingsSerializer()
label = setting_serializer.get_field_label(resource_display)
if label is not None:
resource_display = label
return resource_display
def _get_instance_field_value(
instance, include_model_fields,
model_need_continue_fields, exclude_fields=None
):
data = {}
opts = getattr(instance, '_meta', None)
if opts is not None:
for f in chain(opts.concrete_fields, opts.private_fields):
if not include_model_fields and not getattr(f, 'primary_key', False):
continue
if isinstance(f, (models.FileField, models.ImageField)):
continue
if getattr(f, 'attname', None) in model_need_continue_fields:
continue
value = getattr(instance, f.name) or getattr(instance, f.attname)
if not isinstance(value, bool) and not value:
continue
if getattr(f, 'primary_key', False):
f.verbose_name = 'id'
elif isinstance(value, (list, dict)):
value = copy.deepcopy(value)
elif isinstance(f, models.OneToOneField) and isinstance(value, models.Model):
nested_data = _get_instance_field_value(
value, include_model_fields, model_need_continue_fields, ('id',)
)
for k, v in nested_data.items():
if exclude_fields and k in exclude_fields:
continue
data.setdefault(k, v)
continue
data.setdefault(str(f.verbose_name), value)
return data
def model_to_dict_for_operate_log(
instance, include_model_fields=True, include_related_fields=False
):
model_need_continue_fields = ['date_updated']
m2m_need_continue_fields = ['history_passwords']
data = _get_instance_field_value(
instance, include_model_fields, model_need_continue_fields
)
if include_related_fields:
opts = instance._meta
for f in opts.many_to_many:
value = []
if instance.pk is not None:
related_name = getattr(f, 'attname', '') or getattr(f, 'related_name', '')
if not related_name or related_name in m2m_need_continue_fields:
continue
try:
value = [str(i) for i in getattr(instance, related_name).all()]
except:
pass
if not value:
continue
try:
field_key = getattr(f, 'verbose_name', None) or f.related_model._meta.verbose_name
data.setdefault(str(field_key), value)
except:
pass
return data