diff options
Diffstat (limited to 'src/mailman/database/mailinglist.py')
| -rw-r--r-- | src/mailman/database/mailinglist.py | 465 |
1 files changed, 0 insertions, 465 deletions
diff --git a/src/mailman/database/mailinglist.py b/src/mailman/database/mailinglist.py deleted file mode 100644 index 447d4657a..000000000 --- a/src/mailman/database/mailinglist.py +++ /dev/null @@ -1,465 +0,0 @@ -# Copyright (C) 2006-2009 by the Free Software Foundation, Inc. -# -# This file is part of GNU Mailman. -# -# GNU Mailman is free software: you can redistribute it and/or modify it under -# the terms of the GNU General Public License as published by the Free -# Software Foundation, either version 3 of the License, or (at your option) -# any later version. -# -# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT -# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or -# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for -# more details. -# -# You should have received a copy of the GNU General Public License along with -# GNU Mailman. If not, see <http://www.gnu.org/licenses/>. - -"""Model for mailing lists.""" - -from __future__ import absolute_import, unicode_literals - -__metaclass__ = type -__all__ = [ - 'MailingList', - ] - - -import os -import string - -from storm.locals import ( - And, Bool, DateTime, Float, Int, Pickle, Reference, Store, TimeDelta, - Unicode) -from urlparse import urljoin -from zope.interface import implements - -from mailman.config import config -from mailman.database import roster -from mailman.database.digests import OneLastDigest -from mailman.database.mime import ContentFilter -from mailman.database.model import Model -from mailman.database.types import Enum -from mailman.interfaces.domain import IDomainManager -from mailman.interfaces.mailinglist import ( - IAcceptableAlias, IAcceptableAliasSet, IMailingList, Personalization) -from mailman.interfaces.mime import FilterType -from mailman.utilities.filesystem import makedirs -from mailman.utilities.string import expand - - -SPACE = ' ' -UNDERSCORE = '_' - - - -class MailingList(Model): - implements(IMailingList) - - id = Int(primary=True) - - # List identity - list_name = Unicode() - host_name = Unicode() - list_id = Unicode() - include_list_post_header = Bool() - include_rfc2369_headers = Bool() - # Attributes not directly modifiable via the web u/i - created_at = DateTime() - admin_member_chunksize = Int() - # Attributes which are directly modifiable via the web u/i. The more - # complicated attributes are currently stored as pickles, though that - # will change as the schema and implementation is developed. - next_request_id = Int() - next_digest_number = Int() - digest_last_sent_at = DateTime() - volume = Int() - last_post_time = DateTime() - # Implicit destination. - acceptable_aliases_id = Int() - acceptable_alias = Reference(acceptable_aliases_id, 'AcceptableAlias.id') - # Attributes which are directly modifiable via the web u/i. The more - # complicated attributes are currently stored as pickles, though that - # will change as the schema and implementation is developed. - accept_these_nonmembers = Pickle() - admin_immed_notify = Bool() - admin_notify_mchanges = Bool() - administrivia = Bool() - advertised = Bool() - anonymous_list = Bool() - archive = Bool() - archive_private = Bool() - archive_volume_frequency = Int() - # Automatic responses. - autoresponse_grace_period = TimeDelta() - autorespond_owner = Enum() - autoresponse_owner_text = Unicode() - autorespond_postings = Enum() - autoresponse_postings_text = Unicode() - autorespond_requests = Enum() - autoresponse_request_text = Unicode() - # Content filters. - filter_content = Bool() - collapse_alternatives = Bool() - convert_html_to_plaintext = Bool() - # Bounces and bans. - ban_list = Pickle() - bounce_info_stale_after = TimeDelta() - bounce_matching_headers = Unicode() - bounce_notify_owner_on_disable = Bool() - bounce_notify_owner_on_removal = Bool() - bounce_processing = Bool() - bounce_score_threshold = Int() - bounce_unrecognized_goes_to_list_owner = Bool() - bounce_you_are_disabled_warnings = Int() - bounce_you_are_disabled_warnings_interval = TimeDelta() - default_member_moderation = Bool() - description = Unicode() - digest_footer = Unicode() - digest_header = Unicode() - digest_is_default = Bool() - digest_send_periodic = Bool() - digest_size_threshold = Float() - digest_volume_frequency = Enum() - digestable = Bool() - discard_these_nonmembers = Pickle() - emergency = Bool() - encode_ascii_prefixes = Bool() - first_strip_reply_to = Bool() - forward_auto_discards = Bool() - gateway_to_mail = Bool() - gateway_to_news = Bool() - generic_nonmember_action = Int() - goodbye_msg = Unicode() - header_matches = Pickle() - hold_these_nonmembers = Pickle() - info = Unicode() - linked_newsgroup = Unicode() - max_days_to_hold = Int() - max_message_size = Int() - max_num_recipients = Int() - member_moderation_action = Enum() - member_moderation_notice = Unicode() - mime_is_default_digest = Bool() - moderator_password = Unicode() - msg_footer = Unicode() - msg_header = Unicode() - new_member_options = Int() - news_moderation = Enum() - news_prefix_subject_too = Bool() - nntp_host = Unicode() - nondigestable = Bool() - nonmember_rejection_notice = Unicode() - obscure_addresses = Bool() - personalize = Enum() - pipeline = Unicode() - post_id = Int() - _preferred_language = Unicode(name='preferred_language') - private_roster = Bool() - real_name = Unicode() - reject_these_nonmembers = Pickle() - reply_goes_to_list = Enum() - reply_to_address = Unicode() - require_explicit_destination = Bool() - respond_to_post_requests = Bool() - scrub_nondigest = Bool() - send_goodbye_msg = Bool() - send_reminders = Bool() - send_welcome_msg = Bool() - start_chain = Unicode() - subject_prefix = Unicode() - subscribe_auto_approval = Pickle() - subscribe_policy = Int() - topics = Pickle() - topics_bodylines_limit = Int() - topics_enabled = Bool() - unsubscribe_policy = Int() - welcome_msg = Unicode() - - def __init__(self, fqdn_listname): - super(MailingList, self).__init__() - listname, at, hostname = fqdn_listname.partition('@') - assert hostname, 'Bad list name: {0}'.format(fqdn_listname) - self.list_name = listname - self.host_name = hostname - # For the pending database - self.next_request_id = 1 - self._restore() - self.personalization = Personalization.none - self.real_name = string.capwords( - SPACE.join(listname.split(UNDERSCORE))) - makedirs(self.data_path) - - # XXX FIXME - def _restore(self): - self.owners = roster.OwnerRoster(self) - self.moderators = roster.ModeratorRoster(self) - self.administrators = roster.AdministratorRoster(self) - self.members = roster.MemberRoster(self) - self.regular_members = roster.RegularMemberRoster(self) - self.digest_members = roster.DigestMemberRoster(self) - self.subscribers = roster.Subscribers(self) - - def __repr__(self): - return '<mailing list "{0}" at {1:#x}>'.format( - self.fqdn_listname, id(self)) - - @property - def fqdn_listname(self): - """See `IMailingList`.""" - return '{0}@{1}'.format(self.list_name, self.host_name) - - @property - def web_host(self): - """See `IMailingList`.""" - return IDomainManager(config)[self.host_name] - - def script_url(self, target, context=None): - """See `IMailingList`.""" - # Find the domain for this mailing list. - domain = IDomainManager(config)[self.host_name] - # XXX Handle the case for when context is not None; those would be - # relative URLs. - return urljoin(domain.base_url, target + '/' + self.fqdn_listname) - - @property - def data_path(self): - """See `IMailingList`.""" - return os.path.join(config.LIST_DATA_DIR, self.fqdn_listname) - - # IMailingListAddresses - - @property - def posting_address(self): - """See `IMailingList`.""" - return self.fqdn_listname - - @property - def no_reply_address(self): - """See `IMailingList`.""" - return '{0}@{1}'.format(config.mailman.noreply_address, self.host_name) - - @property - def owner_address(self): - """See `IMailingList`.""" - return '{0}-owner@{1}'.format(self.list_name, self.host_name) - - @property - def request_address(self): - """See `IMailingList`.""" - return '{0}-request@{1}'.format(self.list_name, self.host_name) - - @property - def bounces_address(self): - """See `IMailingList`.""" - return '{0}-bounces@{1}'.format(self.list_name, self.host_name) - - @property - def join_address(self): - """See `IMailingList`.""" - return '{0}-join@{1}'.format(self.list_name, self.host_name) - - @property - def leave_address(self): - """See `IMailingList`.""" - return '{0}-leave@{1}'.format(self.list_name, self.host_name) - - @property - def subscribe_address(self): - """See `IMailingList`.""" - return '{0}-subscribe@{1}'.format(self.list_name, self.host_name) - - @property - def unsubscribe_address(self): - """See `IMailingList`.""" - return '{0}-unsubscribe@{1}'.format(self.list_name, self.host_name) - - def confirm_address(self, cookie): - """See `IMailingList`.""" - local_part = expand(config.mta.verp_confirm_format, dict( - address = '{0}-confirm'.format(self.list_name), - cookie = cookie)) - return '{0}@{1}'.format(local_part, self.host_name) - - @property - def preferred_language(self): - """See `IMailingList`.""" - return config.languages[self._preferred_language] - - @preferred_language.setter - def preferred_language(self, language): - """See `IMailingList`.""" - # Accept both a language code and a `Language` instance. - try: - self._preferred_language = language.code - except AttributeError: - self._preferred_language = language - - def send_one_last_digest_to(self, address, delivery_mode): - """See `IMailingList`.""" - digest = OneLastDigest(self, address, delivery_mode) - Store.of(self).add(digest) - - @property - def last_digest_recipients(self): - """See `IMailingList`.""" - results = Store.of(self).find( - OneLastDigest, - OneLastDigest.mailing_list == self) - recipients = [(digest.address, digest.delivery_mode) - for digest in results] - results.remove() - return recipients - - @property - def filter_types(self): - """See `IMailingList`.""" - results = Store.of(self).find( - ContentFilter, - And(ContentFilter.mailing_list == self, - ContentFilter.filter_type == FilterType.filter_mime)) - for content_filter in results: - yield content_filter.filter_pattern - - @filter_types.setter - def filter_types(self, sequence): - """See `IMailingList`.""" - # First, delete all existing MIME type filter patterns. - store = Store.of(self) - results = store.find( - ContentFilter, - And(ContentFilter.mailing_list == self, - ContentFilter.filter_type == FilterType.filter_mime)) - results.remove() - # Now add all the new filter types. - for mime_type in sequence: - content_filter = ContentFilter( - self, mime_type, FilterType.filter_mime) - store.add(content_filter) - - @property - def pass_types(self): - """See `IMailingList`.""" - results = Store.of(self).find( - ContentFilter, - And(ContentFilter.mailing_list == self, - ContentFilter.filter_type == FilterType.pass_mime)) - for content_filter in results: - yield content_filter.filter_pattern - - @pass_types.setter - def pass_types(self, sequence): - """See `IMailingList`.""" - # First, delete all existing MIME type pass patterns. - store = Store.of(self) - results = store.find( - ContentFilter, - And(ContentFilter.mailing_list == self, - ContentFilter.filter_type == FilterType.pass_mime)) - results.remove() - # Now add all the new filter types. - for mime_type in sequence: - content_filter = ContentFilter( - self, mime_type, FilterType.pass_mime) - store.add(content_filter) - - @property - def filter_extensions(self): - """See `IMailingList`.""" - results = Store.of(self).find( - ContentFilter, - And(ContentFilter.mailing_list == self, - ContentFilter.filter_type == FilterType.filter_extension)) - for content_filter in results: - yield content_filter.filter_pattern - - @filter_extensions.setter - def filter_extensions(self, sequence): - """See `IMailingList`.""" - # First, delete all existing file extensions filter patterns. - store = Store.of(self) - results = store.find( - ContentFilter, - And(ContentFilter.mailing_list == self, - ContentFilter.filter_type == FilterType.filter_extension)) - results.remove() - # Now add all the new filter types. - for mime_type in sequence: - content_filter = ContentFilter( - self, mime_type, FilterType.filter_extension) - store.add(content_filter) - - @property - def pass_extensions(self): - """See `IMailingList`.""" - results = Store.of(self).find( - ContentFilter, - And(ContentFilter.mailing_list == self, - ContentFilter.filter_type == FilterType.pass_extension)) - for content_filter in results: - yield content_filter.pass_pattern - - @pass_extensions.setter - def pass_extensions(self, sequence): - """See `IMailingList`.""" - # First, delete all existing file extensions pass patterns. - store = Store.of(self) - results = store.find( - ContentFilter, - And(ContentFilter.mailing_list == self, - ContentFilter.filter_type == FilterType.pass_extension)) - results.remove() - # Now add all the new filter types. - for mime_type in sequence: - content_filter = ContentFilter( - self, mime_type, FilterType.pass_extension) - store.add(content_filter) - - - -class AcceptableAlias(Model): - implements(IAcceptableAlias) - - id = Int(primary=True) - - mailing_list_id = Int() - mailing_list = Reference(mailing_list_id, MailingList.id) - - alias = Unicode() - - def __init__(self, mailing_list, alias): - self.mailing_list = mailing_list - self.alias = alias - - -class AcceptableAliasSet: - implements(IAcceptableAliasSet) - - def __init__(self, mailing_list): - self._mailing_list = mailing_list - - def clear(self): - """See `IAcceptableAliasSet`.""" - Store.of(self._mailing_list).find( - AcceptableAlias, - AcceptableAlias.mailing_list == self._mailing_list).remove() - - def add(self, alias): - if not (alias.startswith('^') or '@' in alias): - raise ValueError(alias) - alias = AcceptableAlias(self._mailing_list, alias.lower()) - Store.of(self._mailing_list).add(alias) - - def remove(self, alias): - Store.of(self._mailing_list).find( - AcceptableAlias, - And(AcceptableAlias.mailing_list == self._mailing_list, - AcceptableAlias.alias == alias.lower())).remove() - - @property - def aliases(self): - aliases = Store.of(self._mailing_list).find( - AcceptableAlias, - AcceptableAlias.mailing_list == self._mailing_list) - for alias in aliases: - yield alias.alias |