github-jwinterm/modoboa
1
0
Fork 0
mirror of https://github.com/jwinterm/modoboa.git synced 2026-10-17 05:41:25 +02:00
Code Issues Packages Projects Releases Wiki Activity

PEP 8 compliance.

Browse Source
This commit is contained in:
Antoine Nguyen
2013-02-08 17:19:00 +01:00
parent 6b0be7f4da
commit 724b2f3a08
6 changed files with 113 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modoboa/extensions/amavis/forms.py
+13 -9
View File
@@ -8,6 +8,7 @@ from django.utils.html import conditional_escape
from django.utils.encoding import force_unicode
from django.utils.safestring import mark_safe
class CustomRadioInput(RadioInput):
def __unicode__(self):
if 'id' in self.attrs:
@@ -17,6 +18,7 @@ class CustomRadioInput(RadioInput):
choice_label = conditional_escape(force_unicode(self.choice_label))
return mark_safe(u'<label class="radio inline" %s>%s %s</label>' % (label_for, self.tag(), choice_label))
class InlineRadioRenderer(RadioSelect.renderer):
def __iter__(self):
for i, choice in enumerate(self.choices):
@@ -25,28 +27,30 @@ class InlineRadioRenderer(RadioSelect.renderer):
def render(self):
return mark_safe(u'\n'.join([u'%s\n' % force_unicode(w) for w in self]))
class InlineRadioSelect(RadioSelect):
renderer = InlineRadioRenderer
class DomainPolicyForm(forms.ModelForm):
spam_subject_tag2_act = forms.BooleanField()
class Meta:
model = Policy
fields = ('bypass_virus_checks', 'bypass_spam_checks',
fields = ('bypass_virus_checks', 'bypass_spam_checks',
'spam_tag2_level', 'spam_subject_tag2',
'spam_kill_level', 'bypass_banned_checks')
widgets = {
'bypass_virus_checks' : InlineRadioSelect(),
'bypass_spam_checks' : InlineRadioSelect(),
'spam_tag2_level' : forms.TextInput(attrs={'class' : 'span1'}),
'spam_kill_level' : forms.TextInput(attrs={'class' : 'span1'}),
'spam_subject_tag2' : forms.TextInput(attrs={'class' : 'span2'}),
'bypass_banned_checks' : InlineRadioSelect(),
'bypass_virus_checks': InlineRadioSelect(),
'bypass_spam_checks': InlineRadioSelect(),
'spam_tag2_level': forms.TextInput(attrs={'class': 'span1'}),
'spam_kill_level': forms.TextInput(attrs={'class': 'span1'}),
'spam_subject_tag2': forms.TextInput(attrs={'class': 'span2'}),
'bypass_banned_checks': InlineRadioSelect(),
}
def __init__(self, *args, **kwargs):
if kwargs.has_key("instance"):
if "instance" in kwargs:
self.domain = kwargs["instance"]
try:
policy = Users.objects.get(email="@%s" % self.domain.name).policy
@@ -56,7 +60,7 @@ class DomainPolicyForm(forms.ModelForm):
super(DomainPolicyForm, self).__init__(*args, **kwargs)
for f in self.fields.keys():
self.fields[f].required = False
def save(self, user, commit=True):
p = super(DomainPolicyForm, self).save(commit=False)
for f in ['bypass_spam_checks', 'bypass_virus_checks', 'bypass_banned_checks']:
modoboa/extensions/amavis/lib.py
+2
View File
@@ -11,6 +11,7 @@ from modoboa.lib import parameters
from modoboa.lib.exceptions import ModoboaException
from models import Msgrcpt
def selfservice(ssfunc=None):
"""Decorator used to expose views to the 'self-service' feature
@@ -34,6 +35,7 @@ def selfservice(ssfunc=None):
return wrapped_f
return decorator
class AMrelease(object):
def __init__(self):
mode = parameters.get_admin("AM_PDP_MODE")
modoboa/extensions/amavis/models.py
+21 -6
View File
@@ -12,23 +12,28 @@
from django.db import models
from django.utils.translation import ugettext as _, ugettext_lazy
class Maddr(models.Model):
partition_tag = models.IntegerField(unique=True, null=True, blank=True)
id = models.BigIntegerField(primary_key=True)
email = models.CharField(unique=True, max_length=255)
domain = models.CharField(max_length=765)
class Meta:
db_table = u'maddr'
managed = False
class Mailaddr(models.Model):
id = models.IntegerField(primary_key=True)
priority = models.IntegerField()
email = models.CharField(unique=True, max_length=255)
class Meta:
db_table = u'mailaddr'
managed = False
class Msgs(models.Model):
partition_tag = models.IntegerField(null=True, blank=True)
mail_id = models.CharField(max_length=12, primary_key=True)
@@ -50,11 +55,13 @@ class Msgs(models.Model):
from_addr = models.CharField(max_length=765, blank=True)
subject = models.CharField(max_length=765, blank=True)
host = models.CharField(max_length=765)
class Meta:
db_table = u'msgs'
managed = False
unique_together = ('partition_tag', 'mail_id')
class Msgrcpt(models.Model):
partition_tag = models.IntegerField(null=True, blank=True)
mail = models.ForeignKey(Msgs, primary_key=True)
@@ -68,11 +75,13 @@ class Msgrcpt(models.Model):
wl = models.CharField(max_length=3, blank=True)
bspam_level = models.FloatField(null=True, blank=True)
smtp_resp = models.CharField(max_length=765, blank=True)
class Meta:
db_table = u'msgrcpt'
managed = False
unique_together = ("partition_tag", "mail", "rseqnum")
class Policy(models.Model):
policy_name = models.CharField(max_length=96, blank=True)
virus_lover = models.CharField(max_length=3, blank=True, null=True)
@@ -82,24 +91,24 @@ class Policy(models.Model):
bad_header_lover = models.CharField(max_length=3, blank=True, null=True)
bypass_virus_checks = models.CharField(
ugettext_lazy("Virus filter"), default='', null=True,
choices=(('N', ugettext_lazy('yes')),
('Y', ugettext_lazy('no')),
choices=(('N', ugettext_lazy('yes')),
('Y', ugettext_lazy('no')),
('', ugettext_lazy('default'))),
max_length=3,
help_text=ugettext_lazy("Bypass virus checks or not. Choose 'default' to use global settings.")
)
bypass_spam_checks = models.CharField(
ugettext_lazy("Spam filter"), default='', null=True,
choices=(('N', ugettext_lazy('yes')),
('Y', ugettext_lazy('no')),
choices=(('N', ugettext_lazy('yes')),
('Y', ugettext_lazy('no')),
('', ugettext_lazy('default'))),
max_length=3,
help_text=ugettext_lazy("Bypass spam checks or not. Choose 'default' to use global settings.")
)
bypass_banned_checks = models.CharField(
ugettext_lazy("Banned filter"), default='', null=True,
choices=(('N', ugettext_lazy('yes')),
('Y', ugettext_lazy('no')),
choices=(('N', ugettext_lazy('yes')),
('Y', ugettext_lazy('no')),
('', ugettext_lazy('default'))),
max_length=3,
help_text=ugettext_lazy("Bypass banned checks or not. Choose 'default' to use global settings.")
@@ -148,31 +157,37 @@ class Policy(models.Model):
db_table = u'policy'
managed = False
class Quarantine(models.Model):
partition_tag = models.IntegerField(null=True, blank=True)
mail = models.ForeignKey(Msgs, primary_key=True)
chunk_ind = models.IntegerField()
mail_text = models.TextField()
class Meta:
db_table = u'quarantine'
managed = False
ordering = ["-mail__time_num"]
unique_together = ("partition_tag", "mail", "chunk_ind")
class Users(models.Model):
id = models.AutoField(primary_key=True)
priority = models.IntegerField()
policy = models.ForeignKey(Policy)
email = models.CharField(unique=True, max_length=255)
fullname = models.CharField(max_length=765, blank=True)
class Meta:
db_table = u'users'
managed = False
class Wblist(models.Model):
rid = models.IntegerField(primary_key=True)
sid = models.IntegerField(primary_key=True)
wb = models.CharField(max_length=30)
class Meta:
db_table = u'wblist'
managed = False
modoboa/extensions/amavis/sql_listing.py
+42 -41
View File
@@ -10,6 +10,7 @@ from modoboa.lib.emailutils import *
from modoboa.lib.dbutils import db_type
from models import *
class Qtable(tables.Table):
tableid = "emails"
styles = "table-condensed"
@@ -31,13 +32,14 @@ class Qtable(tables.Table):
def parse_date(self, value):
return datetime.fromtimestamp(value)
class SQLconnector(MBconnector):
orders = {
"from" : "mail__from_addr",
"subject" : "mail__subject",
"date" : "mail__time_num"
"from": "mail__from_addr",
"subject": "mail__subject",
"date": "mail__time_num"
}
def __init__(self, mail_ids=None, filter=None):
self.count = None
self.mail_ids = mail_ids
@@ -51,7 +53,7 @@ class SQLconnector(MBconnector):
if self.filter:
filter &= self.filter
self.messages = Quarantine.objects.filter(filter)
if kwargs.has_key("order"):
if "order" in kwargs:
totranslate = kwargs["order"][1:]
sign = kwargs["order"][:1]
if sign == " ":
@@ -66,12 +68,12 @@ class SQLconnector(MBconnector):
emails = []
for qm in messages:
for rcpt in qm.mail.msgrcpt_set.all():
m = {"from" : qm.mail.from_addr,
"to" : rcpt.rid.email,
"subject" : qm.mail.subject,
"mailid" : qm.mail_id,
"date" : qm.mail.time_num,
"type" : rcpt.content}
m = {"from": qm.mail.from_addr,
"to": rcpt.rid.email,
"subject": qm.mail.subject,
"mailid": qm.mail_id,
"date": qm.mail.time_num,
"type": rcpt.content}
if rcpt.rs == '':
m["class"] = "unseen"
elif rcpt.rs == 'R':
@@ -100,7 +102,7 @@ class SQLWrapper(object):
q = Q(rs='p')
else:
q = ~Q(rs='D')
if request.user.group == 'SimpleUsers':
q &= Q(rid__email=request.user.email)
else:
@@ -126,7 +128,7 @@ class SQLWrapper(object):
def get_pending_requests(self, user):
"""Return the number of current pending requests
:param user: a ``User`` instance
"""
rq = Q(rs='p')
@@ -139,16 +141,17 @@ class SQLWrapper(object):
rq &= doms_q
return Msgrcpt.objects.filter(rq).count()
def get_mail_content(self, mailid):
def get_mail_content(self, mailid):
return Quarantine.objects.filter(mail=mailid)
class PgWrapper(SQLWrapper):
"""The postgres wrapper
Make use of ``QuerySet.extra`` and postgres ``convert_from``
function to let the quarantine manager work as expected !
"""
def get_mails(self, request, rcptfilter=None):
if request.GET.get("viewrequests", None) == "1":
q = Q(rs='p')
@@ -171,43 +174,43 @@ class PgWrapper(SQLWrapper):
def get_recipient_message(self, address, mailid):
qset = Msgrcpt.objects.filter(mail=mailid).extra(
where=["msgrcpt.rid=maddr.id", "convert_from(maddr.email, 'UTF8') = '%s'" % address],
where=["msgrcpt.rid=maddr.id", "convert_from(maddr.email, 'UTF8') = '%s'" % address],
tables=['maddr']
)
return qset.all()[0]
def get_recipient_messages(self, address, mailids):
return Msgrcpt.objects.filter(mail__in=mailids).extra(
where=["U0.rid=maddr.id", "convert_from(maddr.email, 'UTF8') = '%s'" % address],
where=["U0.rid=maddr.id", "convert_from(maddr.email, 'UTF8') = '%s'" % address],
tables=['maddr']
)
def get_domains_pending_requests(self, domains):
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, domains))
return Msgrcpt.objects.filter(rs='p').extra(
where=["msgrcpt.rid=maddr.id", "convert_from(maddr.email, 'UTF8') ~ '%s'" % regexp],
where=["msgrcpt.rid=maddr.id", "convert_from(maddr.email, 'UTF8') ~ '%s'" % regexp],
tables=['maddr']
)
def get_pending_requests(self, user):
rq = Q(rs='p')
if not user.is_superuser:
doms = user.get_domains()
if not doms.count():
return 0
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, doms))
return Msgrcpt.objects.filter(rq).extra(
where=["msgrcpt.rid=maddr.id", "convert_from(maddr.email, 'UTF8') ~ '%s'" % (regexp,)],
tables=['maddr']
).count()
return len(Msgrcpt.objects.filter(rq))
rq = Q(rs='p')
if not user.is_superuser:
doms = user.get_domains()
if not doms.count():
return 0
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, doms))
return Msgrcpt.objects.filter(rq).extra(
where=["msgrcpt.rid=maddr.id", "convert_from(maddr.email, 'UTF8') ~ '%s'" % (regexp,)],
tables=['maddr']
).count()
return len(Msgrcpt.objects.filter(rq))
def get_mail_content(self, mailid):
return Quarantine.objects.filter(mail=mailid).extra(
select={'mail_text' : "convert_from(mail_text, 'UTF8')"}
select={'mail_text': "convert_from(mail_text, 'UTF8')"}
)
def get_wrapper():
"""Return the appropriate *Wrapper class
@@ -217,6 +220,7 @@ def get_wrapper():
return PgWrapper()
return SQLWrapper()
class SQLlisting(EmailListing):
tpl = "amavis/index.html"
tbltype = Qtable
@@ -230,16 +234,16 @@ class SQLlisting(EmailListing):
else:
Qtable.cols_order = ['type', 'rstatus', 'to', 'from_', 'subject', 'time']
self.mbc = SQLconnector(msgs, filter)
super(SQLlisting, self).__init__(**kwargs)
class SQLemail(Email):
def __init__(self, msg, *args, **kwargs):
super(SQLemail, self).__init__(msg, *args, **kwargs)
fields = ["From", "To", "Cc", "Date", "Subject"]
for f in fields:
label = f
self.headers += [{"name" : label, "value" : self.get_header(msg, f)}]
self.headers += [{"name": label, "value": self.get_header(msg, f)}]
try:
label = re.sub("-", "_", label)
setattr(self, label, msg[f])
@@ -255,18 +259,15 @@ class SQLemail(Email):
# Workaround for amavis <= 2.8.0 :p
self.qtype = "BAD HEADER SECTION"
self.qreason = qreason[19:]
def get_header(self, msg, name):
if msg.has_key(name):
return msg[name]
name = name.upper()
if msg.has_key(name):
return msg[name]
for k in [name, name.upper()]:
if k in msg:
return msg[k]
return ""
def render_headers(self, **kwargs):
return render_to_string("amavis/mailheaders.html", {
"qtype" : self.qtype, "qreason" : self.qreason,
"headers" : self.headers,
"qtype": self.qtype, "qreason": self.qreason,
"headers": self.headers,
})
modoboa/extensions/amavis/templatetags/amextras.py
+1 -1
View File
@@ -15,7 +15,7 @@ def viewm_menu(user, mail_id, rcpt):
{"name" : "release",
"img" : "icon-white icon-ok",
"class" : "btn-success",
"url" : reverse(amavis.views.release, args=[mail_id]) \
"url" : reverse(amavis.views.release, args=[mail_id])
+ ("?rcpt=%s" % rcpt if rcpt else ""),
"label" : _("Release")},
{"name" : "delete",
modoboa/extensions/amavis/views.py
+34 -18
View File
@@ -17,23 +17,26 @@ from templatetags.amextras import *
from modoboa.lib.email_listing import parse_search_parameters
from sql_listing import *
def __get_current_url(request):
if request.session.has_key("page"):
if "page" in request.session:
res = "listing?page=%s" % request.session["page"]
else:
res = ""
params = "&".join(map(lambda p: "%s=%s" % (p, request.session[p]),
params = "&".join(map(lambda p: "%s=%s" % (p, request.session[p]),
filter(request.session.has_key, ["criteria", "pattern"])))
if params != "":
res += "?%s" % (params)
return res
def empty_quarantine(request):
content = "<div class='alert alert-info'>%s</div>" % _("Empty quarantine")
ctx = getctx("ok", level=2, listing=content, navbar="",
menu=quar_menu(request.user))
return HttpResponse(simplejson.dumps(ctx), mimetype="application/json")
@login_required
def _listing(request):
flt = None
@@ -45,7 +48,7 @@ def _listing(request):
return empty_quarantine(request)
order = request.GET.get("order", "-date")
if not request.session.has_key("navparams"):
if not "navparams" in request.session:
request.session["navparams"] = {}
request.session["navparams"]["order"] = order
@@ -75,10 +78,10 @@ def _listing(request):
if request.session.has_key("page"):
del request.session["page"]
pageid = 1
lst = SQLlisting(request.user, msgs, flt,
navparams=request.session["navparams"],
elems_per_page=int(parameters.get_user(request.user,
elems_per_page=int(parameters.get_user(request.user,
"MESSAGES_PER_PAGE")))
page = lst.paginator.getpage(pageid)
if not page:
@@ -90,12 +93,14 @@ def _listing(request):
menu=quar_menu(request.user))
return HttpResponse(simplejson.dumps(ctx), mimetype="application/json")
@login_required
def index(request):
return _render(request, "amavis/index.html", dict(
deflocation="listing/", defcallback="listing_cb", selection="quarantine"
))
def getmailcontent_selfservice(request, mail_id):
from sql_listing import SQLemail
@@ -106,10 +111,11 @@ def getmailcontent_selfservice(request, mail_id):
msg = email.message_from_string(content)
mail = SQLemail(msg, mformat="plain", links="0")
return _render(request, "common/viewmail.html", {
"headers" : mail.render_headers(),
"mailbody" : mail.body
"headers": mail.render_headers(),
"mailbody": mail.body
})
@selfservice(getmailcontent_selfservice)
def getmailcontent(request, mail_id):
from sql_listing import SQLemail
@@ -121,11 +127,12 @@ def getmailcontent(request, mail_id):
msg = email.message_from_string(content)
mail = SQLemail(msg, mformat="plain", links="0")
return _render(request, "common/viewmail.html", {
"headers" : mail.render_headers(),
"mailbody" : mail.body
"headers": mail.render_headers(),
"mailbody": mail.body
})
def viewmail_selfservice(request, mail_id,
def viewmail_selfservice(request, mail_id,
tplname="amavis/viewmail_selfservice.html"):
rcpt = request.GET.get("rcpt", None)
secret_id = request.GET.get("secret_id", "")
@@ -134,11 +141,12 @@ def viewmail_selfservice(request, mail_id,
content = Template("""
<iframe src="{% url modoboa.extensions.amavis.views.getmailcontent mail_id %}" id="mailcontent"></iframe>
""").render(Context(dict(mail_id=mail_id)))
return _render(request, tplname, dict(
mail_id=mail_id, rcpt=rcpt, secret_id=secret_id, content=content
))
@selfservice(viewmail_selfservice)
def viewmail(request, mail_id):
if request.user.group != 'SimpleUsers':
@@ -152,14 +160,15 @@ def viewmail(request, mail_id):
msgrcpt = get_wrapper().get_recipient_message(mb.full_address, mail_id)
msgrcpt.rs = 'V'
msgrcpt.save()
content = Template("""
<iframe src="{{ url }}" id="mailcontent"></iframe>
""").render(Context({"url" : reverse(getmailcontent, args=[mail_id])}))
""").render(Context({"url": reverse(getmailcontent, args=[mail_id])}))
menu = viewm_menu(request.user, mail_id, rcpt)
ctx = getctx("ok", menu=menu, listing=content)
return HttpResponse(simplejson.dumps(ctx), mimetype="application/json")
@login_required
def viewheaders(request, mail_id):
content = ""
@@ -167,9 +176,10 @@ def viewheaders(request, mail_id):
content += qm.mail_text
msg = email.message_from_string(content)
return _render(request, 'amavis/viewheader.html', {
"headers" : msg.items()
"headers": msg.items()
})
def check_mail_id(request, mail_id):
if type(mail_id) in [str, unicode]:
if request.GET.has_key("rcpt"):
@@ -178,6 +188,7 @@ def check_mail_id(request, mail_id):
mail_id = [mail_id]
return mail_id
def delete_selfservice(request, mail_id):
rcpt = request.GET.get("rcpt", None)
if rcpt is None:
@@ -190,6 +201,7 @@ def delete_selfservice(request, mail_id):
raise ModoboaException(_("Invalid request"))
return ajax_simple_response(dict(status="ok", respmsg=_("Message deleted")))
@selfservice(delete_selfservice)
def delete(request, mail_id):
mail_id = check_mail_id(request, mail_id)
@@ -210,10 +222,11 @@ def delete(request, mail_id):
message = ungettext("%(count)d message deleted successfully",
"%(count)d messages deleted successfully",
len(mail_id)) % {"count" : len(mail_id)}
len(mail_id)) % {"count": len(mail_id)}
return ajax_response(request, respmsg=message,
url=__get_current_url(request))
def release_selfservice(request, mail_id):
rcpt = request.GET.get("rcpt", None)
secret_id = request.GET.get("secret_id", None)
@@ -239,6 +252,7 @@ def release_selfservice(request, mail_id):
msgrcpt.save()
return ajax_simple_response(dict(status="ok", respmsg=msg))
@selfservice(release_selfservice)
def release(request, mail_id):
mail_id = check_mail_id(request, mail_id)
@@ -256,7 +270,7 @@ def release(request, mail_id):
msgrcpt.save()
message = ungettext("%(count)d request sent",
"%(count)d requests sent",
len(mail_id)) % {"count" : len(mail_id)}
len(mail_id)) % {"count": len(mail_id)}
return ajax_response(request, "ok", respmsg=message,
url=__get_current_url(request))
else:
@@ -280,12 +294,13 @@ def release(request, mail_id):
if not error:
message = ungettext("%(count)d message released successfully",
"%(count)d messages released successfully",
len(mail_id)) % {"count" : len(mail_id)}
len(mail_id)) % {"count": len(mail_id)}
else:
message = error
return ajax_response(request, "ko" if error else "ok", respmsg=message,
url=__get_current_url(request))
@login_required
def process(request):
ids = request.POST.get("selection", "")
@@ -295,10 +310,11 @@ def process(request):
if request.POST["action"] == "release":
return release(request, ids)
if request.POST["action"] == "delete":
return delete(request, ids)
@login_required
@user_passes_test(lambda u: u.group != 'SimpleUsers')
def nbrequests(request):