import time import os.path import typing import json import public import threading import logging from power_mta.maillog_stat import query_maillog_with_time_section, last_maillog_time from public.regexplib import match_email class Trigger: # 触发器类型 TYPE_SUBSCRIBER_ADDED = 'subscriber_added' TYPE_SUBSCRIBED = 'subscribed' TYPE_OPENED = 'opened' TYPE_CLICKED = 'clicked' TYPE_UNSUBSCRIBED = 'unsubscribed' __slots__ = ['type', '__attributes'] def __init__(self, trigger_dict: typing.Dict): self.__attributes: typing.Dict = {} for k, v in trigger_dict.items(): if k.startswith('_'): continue if k in self.__slots__: self.__setattr__(k, v) continue self.__attributes[k] = v def get_attribute(self, attribute: typing.Optional[str] = None, default: typing.Any = None): # 返回所有attribute if attribute is None: return self.__attributes return self.__attributes.get(attribute, default) def to_dict(self): d = {} for k in self.__slots__: if not hasattr(self, k) or k.startswith('_'): continue d[k] = getattr(self, k) d.update(self.__attributes) return d class Triggers: __slots__ = ['triggers'] def __init__(self, triggers: typing.List[typing.Dict]): self.triggers: typing.List[Trigger] = [] for trigger_dict in triggers: self.triggers.append(Trigger(trigger_dict)) def match(self, trigger_type: str, group_ids: typing.Union[int, typing.List[int]] = 0) -> bool: trigger_type = trigger_type.strip() for trigger in self.triggers: if trigger.type == trigger_type: if trigger_type in (Trigger.TYPE_SUBSCRIBED, Trigger.TYPE_SUBSCRIBER_ADDED): if group_ids == 0: return True if isinstance(group_ids, (int, str)): group_ids = [int(group_ids)] for group_id in group_ids: if int(group_id) in trigger.get_attribute('group_ids', []): return True continue return True return False def complete_check(self) -> bool: for trigger in self.triggers: if trigger.type in (Trigger.TYPE_SUBSCRIBED, Trigger.TYPE_SUBSCRIBER_ADDED): # 检查订阅触发器 if len(trigger.get_attribute('group_ids', [])) == 0: return False elif trigger.type == Trigger.TYPE_OPENED: # 检查打开邮件触发器 pass elif trigger.type == Trigger.TYPE_CLICKED: # 检查点击邮件触发器 pass elif trigger.type == Trigger.TYPE_UNSUBSCRIBED: # 检查取消订阅触发器 pass return True def to_list(self): return list(map(lambda x: x.to_dict(), self.triggers)) def json_dumps(self): return json.dumps(self.to_list()) class Node: TYPE_DELAY = 'delay' TYPE_EMAIL = 'email' TYPE_CONDITION = 'condition' TYPE_ACTION = 'action' TYPE_WEBHOOK = 'webhook' TYPE_ABTEST = 'abtest' CONDITION_LOGIC_AND = 'and' CONDITION_LOGIC_OR = 'or' CONDITION_BRANCH_YES = 'yes' CONDITION_BRANCH_NO = 'no' DELAY_UNIT_DAYS = 'days' DELAY_UNIT_HOURS = 'hours' DELAY_UNIT_MINUTES = 'minutes' DELAY_UNIT_SECONDS = 'seconds' ACTION_ADD_TO_SUBSCRIBERS = 'add_to_subscribers' ACTION_REMOVE_FROM_SUBSCRIBERS = 'remove_from_subscribers' ACTION_MARK_AS_UNSUBSCRIBE = 'mark_as_unsubscribe' ACTION_MOVE_TO_STEP = 'move_to_step' __slots__ = ['id', 'parent_id', 'broken', 'complete', 'type', 'attributes', 'create_time', 'update_time', '__attributes', '__next', '__yes', '__no', '__branches'] __int_bool_converts = ('complete', 'broken') __ignore_props = ('id',) def __init__(self, node_dict: typing.Dict, parent_id: int = -1): self.id = node_dict.get('id', 0) self.parent_id = 0 if self.id == 0: self.id = public.snow_flake() self.type = '' self.complete = False self.broken = False self.__attributes: typing.Dict = {} self.__next: typing.Optional[Node] = None self.__yes: typing.Optional[Node] = None self.__no: typing.Optional[Node] = None self.__branches: typing.List[Node] = [] for k, v in node_dict.items(): if k.startswith('_') or k in self.__ignore_props: continue if k == 'next': if isinstance(v, dict): self.set_next(Node(v, self.id)) continue if k in (self.CONDITION_BRANCH_YES, self.CONDITION_BRANCH_NO) and isinstance(v, dict): v['condition_branch'] = k self.set_next(Node(v, self.id)) continue if k in self.__slots__: if k in self.__int_bool_converts: v = bool(v) self.__setattr__(k, v) continue self.__attributes[k] = v if parent_id > -1: self.parent_id = parent_id if hasattr(self, 'attributes'): try: m = json.loads(self.attributes) if isinstance(m, dict): m.update(self.__attributes) self.__attributes = m except: pass self.attributes = json.dumps(self.__attributes, ensure_ascii=False) self.complete_check() self.broken_check() cur_time = int(time.time()) self.update_time = cur_time if not hasattr(self, 'create_time'): self.create_time = cur_time def to_tree_dict(self): root = {} for k in self.__slots__: if not hasattr(self, k) or k.startswith('_') or k == 'attributes': continue root[k] = getattr(self, k) for k, v in self.get_attribute().items(): root[k] = v if self.type == self.TYPE_CONDITION: if isinstance(self.__yes, Node): root['yes'] = self.__yes.to_tree_dict() if isinstance(self.__no, Node): root['no'] = self.__no.to_tree_dict() elif self.type == self.TYPE_ABTEST: root['branches'] = [] for branch in self.__branches: root['branches'].append(branch.to_tree_dict()) root['next'] = None if self.next() is None else self.next().to_tree_dict() return root def next(self, attribute_or_index: typing.Optional[typing.Union[str, int]] = None, default: typing.Any = None): if attribute_or_index is not None: if isinstance(attribute_or_index, str): if attribute_or_index == 'yes': return self.__yes elif attribute_or_index == 'no': return self.__no elif attribute_or_index == 'branches': return self.__branches return default elif isinstance(attribute_or_index, int): return self.__branches[attribute_or_index] return self.__next def set_next(self, node): if self.type == self.TYPE_CONDITION: if node.get_attribute('condition_branch') == self.CONDITION_BRANCH_YES: self.__yes = node elif node.get_attribute('condition_branch') == self.CONDITION_BRANCH_NO: self.__no = node return if self.type == self.TYPE_ABTEST: pass self.__next = node def get_attribute(self, attribute: typing.Optional[str] = None, default: typing.Any = None): # 返回所有attribute if attribute is None: return self.__attributes return self.__attributes.get(attribute, default) def set_attribute(self, name: str, value: typing.Any): self.__attributes[name] = value self.attributes = json.dumps(self.__attributes, ensure_ascii=False) return self def to_dict(self): d = {} for k in self.__slots__: if not hasattr(self, k) or k.startswith('_'): continue d[k] = getattr(self, k) return d def complete_check(self): if self.type == self.TYPE_DELAY: self.complete = self.get_attribute('value', 0) > 0 and self.get_attribute('unit', '') in ('days', 'hours', 'minutes', 'seconds') elif self.type == self.TYPE_EMAIL: try: public.to_dict_obj(self.get_attribute()).validate([ public.Param('email_id').Require().Integer('>', 0), public.Param('campaign_id').Require().Integer('>', 0), public.Param('name').Require().string('>', 0), public.Param('subject').Require().string('>', 0), public.Param('from').Require().Email(), public.Param('get_attribute').Require().String('>', 0), ]) except: return self.complete = True elif self.type == self.TYPE_CONDITION: self.complete = len(self.get_attribute('rules', [])) > 0 elif self.type == self.TYPE_ACTION: self.complete = self.get_attribute('action', '') != '' elif self.type == self.TYPE_WEBHOOK: try: public.to_dict_obj(self.get_attribute()).validate([ public.Param('url').Require().Url(), public.Param('secret').Require().string('>', 0), ]) except: return self.complete = True elif self.type == self.TYPE_ABTEST: self.complete = self.get_attribute('name', '') != '' and len(self.get_attribute('branches', [])) > 0 def broken_check(self): return def base_validate(self): # node转换成dict_obj对象，方便校验数据结构 node_obj = public.to_dict_obj(self.to_dict()) # node基本数据结构验证 node_obj.validate([ public.Param('id').Filter(int), public.Param('parent_id').Filter(int), public.Param('type').Require().String('in', [Node.TYPE_DELAY, Node.TYPE_EMAIL, Node.TYPE_CONDITION, Node.TYPE_ACTION, Node.TYPE_ABTEST, Node.TYPE_WEBHOOK]), public.Param('next').Require(), public.Param('broken').Require().Bool(), public.Param('complete').Require().Bool(), ]) # 根据node_type再次验证专属数据结构 if node_obj.type == Node.TYPE_EMAIL: # email节点数据结构验证 node_obj.validate([ public.Param('email_id').Filter(int), public.Param('campaign_id').Filter(int), public.Param('name').Require(), public.Param('subject').Require(), public.Param('from').Require(), public.Param('from_name').Require(), public.Param('track_opens').Require().Bool(), public.Param('track_clicks').Require().Bool(), public.Param('track_unsubscribe').Require().Bool(), ]) elif node_obj.type == Node.TYPE_DELAY: # delay节点数据结构验证 node_obj.validate([ public.Param('value').Require().Filter(int), public.Param('unit').Require().String('in', [Node.DELAY_UNIT_DAYS, Node.DELAY_UNIT_HOURS, Node.DELAY_UNIT_MINUTES, Node.DELAY_UNIT_SECONDS]), public.Param('description').Require(), ]) elif node_obj.type == Node.TYPE_CONDITION: # condition节点数据结构验证 node_obj.validate([ public.Param('rules').Require().List(), public.Param('logic_type').Require().String('in', [Node.CONDITION_LOGIC_AND, Node.CONDITION_LOGIC_OR]), public.Param('yes').Require(), public.Param('no').Require(), public.Param('description').Require(), ]) elif node_obj.type == Node.TYPE_WEBHOOK: # webhook节点数据结构验证 node_obj.validate([ public.Param('url').Require(), public.Param('secret').Require(), ]) elif node_obj.type == Node.TYPE_ACTION: # action节点数据结构验证 node_obj.validate([ public.Param('action').Require(), public.Param('group_ids').Require().List(), public.Param('groups').List(), public.Param('description').Require(), ]) elif node_obj.type == Node.TYPE_ABTEST: # abtest节点数据结构验证 node_obj.validate([ public.Param('name').Require(), public.Param('branches').Require().List(), ]) elif node_obj.get('abtestbranch', False): # abtestfbranch节点数据结构验证 node_obj.validate([ public.Param('label').Require(), public.Param('path').Require().String('in', ['a', 'b', 'c']), public.Param('percentage').Require().Integer('between', [1, 100]), ]) else: raise public.HintException(public.lang('Invalid node type: {}', self.type)) class ConditionRule: TYPE_CAMPAIGN = 'campaign' TYPE_WORKFLOW = 'workflow' TYPE_RECIPIENT = 'recipient' __slots__ = ['type', 'action', '__attributes'] def __init__(self, condition_rule_dict: typing.Dict): self.type = '' self.action = '' self.__attributes: typing.Dict = {} for k, v in condition_rule_dict.items(): if k.startswith('_'): continue if k in self.__slots__: self.__setattr__(k, v) continue self.__attributes[k] = v def get_attribute(self, attribute: typing.Optional[str] = None, default: typing.Any = None): # 返回所有attribute if attribute is None: return self.__attributes return self.__attributes.get(attribute, default) class ConditionRules: __slots__ = ['rules', 'logic'] def __init__(self, rules: typing.List[typing.Dict], logic: str = Node.CONDITION_LOGIC_AND): self.rules: typing.List[ConditionRule] = [] self.logic: str = logic for rule_dict in rules: self.rules.append(ConditionRule(rule_dict)) def match(self) -> bool: for rule in self.rules: if rule.type in (ConditionRule.TYPE_CAMPAIGN, ConditionRule.TYPE_WORKFLOW): continue if rule.type == ConditionRule.TYPE_RECIPIENT: continue return False return True class Workflow: __db_path = '/www/vmail/mail_automation_workflows' def __init__(self, automation_id: int): self.__automation_id = automation_id self.__create_tables() def __create_tables(self): if not os.path.exists(self.__db_path): os.makedirs(self.__db_path, 0o700) with public.S(None, '{}/workflow_{}'.format(self.__db_path, self.__automation_id)) as query: query.execute_script('''-- 开启wal模式 PRAGMA journal_mode = wal; -- 关闭同步 PRAGMA synchronous = 0; -- 开启事务 begin; -- 节点表 create table if not exists `nodes` ( `id` integer primary key, `parent_id` integer not null default 0, `broken` integer not null default 0, `complete` integer not null default 0, `create_time` integer not null default (strftime('%s')), `update_time` integer not null default (strftime('%s')), `type` text not null default '', `attributes` text not null default '{}' ); create index if not exists `node_parentId` on `nodes` (`parent_id`); -- 工作流程进行表 create table if not exists `schedules` ( `id` integer primary key autoincrement, `node_id` integer not null default 0, `status` integer not null default 0, `create_time` integer not null default (strftime('%s')), `update_time` integer not null default (strftime('%s')), `subscriber_email` text not null default '' ); create index if not exists `schedule_subscriberEmail_status` on `schedules` (`subscriber_email`, `status`); -- 提交事务 commit; ''') def query(self, table_name: str): return public.S(table_name, '{}/workflow_{}'.format(self.__db_path, self.__automation_id)) def __get_node_map(self) -> typing.Dict[int, Node]: with self.query('nodes') as query: nodes = [Node(node_dict) for node_dict in query.select()] node_map = {} for node in nodes: node_map[node.id] = node return node_map def load_workflow_tree(self): return self.build_workflow_tree(self.__get_node_map()) def set_nodes(self, root: typing.Optional[typing.Dict]) -> typing.Optional[Node]: if root is None: # 清空节点树 with self.query('nodes') as query: query.delete() return None insert_data = [] node_map = {} # 解析nodes for node in self.walk_node(root): # 处理特殊节点 if node.type == Node.TYPE_EMAIL: # 邮件发送任务节点 pass elif node.type == Node.TYPE_CONDITION: # 条件分支节点 pass elif node.type == Node.TYPE_ABTEST: # AB测试节点 pass elif node.get_attribute('abtestbranch', False): # AB测试子节点根节点 pass insert_data.append(node.to_dict()) node_map[node.id] = node # 将节点数据写入数据库 if len(insert_data) > 0: with self.query('nodes') as query: query.insert_all(insert_data, option='REPLACE') # 删除无效节点 query.where_not_in('id', list(node_map.keys())).delete() # 生成新的节点树并返回 return self.build_workflow_tree(node_map) def build_workflow_tree(self, node_map: typing.Dict[int, Node], root_id: int = 0) -> typing.Optional[Node]: root = None for node_id, node in node_map.items(): # 检查是否root节点 if node.parent_id == 0 or node.id == root_id: root = node continue if node.parent_id not in node_map: continue node_map[node.parent_id].set_next(node) return root def walk_node(self, node_or_dict: typing.Union[typing.Dict, Node], p_node: typing.Optional[Node] = None): # node_dict转换成Node对象，方便校验数据结构 if isinstance(node_or_dict, Node): node = node_or_dict else: node = Node(node_or_dict, p_node.id if p_node is not None else 0) # 返回节点信息 yield node # 处理特殊节点 if node.type == Node.TYPE_CONDITION: if node.next('yes'): yield from self.walk_node(node.next('yes'), node) if node.next('no'): yield from self.walk_node(node.next('no'), node) elif node.type == Node.TYPE_ABTEST: for abtestbranch in node.next('branches', []): yield from self.walk_node(abtestbranch, node) if node.next(): yield from self.walk_node(node.next(), node) def walk_node_for_schedule(self, root_id: int = 0): with self.query('nodes') as query: query.where('parent_id', root_id) if root_id > 0: query.where_or('id', root_id) nodes = [Node(node_dict) for node_dict in query.select()] if len(nodes) == 0: return # 生成新的节点树并返回 workflow_tree = self.build_workflow_tree({node.id: node for node in nodes}, root_id) if workflow_tree is None: return yield from self.walk_node(workflow_tree) def walk_node_with_database(self): yield from self.walk_node(self.load_workflow_tree()) class Automation: __db_path = '/www/vmail' def __init__(self): self.__IS_INIT = False self.__create_tables() def is_init(self) -> bool: self.__create_tables() return self.__IS_INIT def __create_tables(self): if self.__IS_INIT: return if not os.path.exists(self.__db_path): return self.__IS_INIT = True with self.query() as query: query.execute_script('''-- 开启wal模式 PRAGMA journal_mode = wal; -- 关闭同步 PRAGMA synchronous = 0; -- 开启事务 begin; -- 任务表 create table if not exists `automations` ( `id` integer primary key autoincrement, `status` integer not null default 0, `sent` integer not null default 0, `delivered` integer not null default 0, `opened` integer not null default 0, `clicked` integer not null default 0, `create_time` integer not null default (strftime('%s')), `update_time` integer not null default (strftime('%s')), `last_stat_time` integer not null default 0, `name` text not null default '', `triggers` text not null default '[]' ); -- 提交事务 commit; ''') def query(self): self.__create_tables() return public.S('automations', '{}/automations'.format(self.__db_path)) # 获取自动化任务列表 def get_tasks(self, args: public.dict_obj): args.validate([ public.Param('p').Integer('>', 0), public.Param('p_size').Integer('>', 0), public.Param('keyword').Xss(), ]) with self.query() as query: query.where_in('status', [0, 1])\ .field('id', 'status', 'name', 'create_time', 'update_time', 'sent', 'delivered', 'ifnull(round(1.0 * opened / sent * 100, 2), 0) as `opened`', 'ifnull(round(1.0 * clicked / sent * 100, 2), 0) as `clicked`') if 'keyword' in args and args.keyword != '': query.where('`name` like ?', '%{}%'.format(args.keyword)) ret = { 'total': query.fork().count(), 'list': query.order('id', 'desc').skip((args.get('p', 1) - 1) * args.get('p_size', 20)).limit(args.get('p_size', 20)).select(), } return ret # 创建/编辑邮件自动化任务 def set(self, args: public.dict_obj): args.validate([ public.Param('id').Integer('>', 0), public.Param('name').String('>', 0), public.Param('triggers').Require().List(), public.Param('root').Require(), ]) triggers = Triggers(args.triggers) with self.query() as query: if 'id' in args: automation_id = args.id if not query.where('id', automation_id).exists(): raise public.HintException(public.lang('Invalid automation')) if query.where('id', automation_id).where('status', 1).exists(): raise public.HintException(public.lang('Cannot edit automation in running state.')) update_data = { 'triggers': triggers.json_dumps(), } if 'name' in args: update_data['name'] = args.name query.where('id', automation_id).update(update_data) else: insert_data = { 'triggers': triggers.json_dumps(), } if 'name' in args: insert_data['name'] = args.name automation_id = query.insert(insert_data) wf = Workflow(automation_id) root = wf.set_nodes(args.root) return { 'id': automation_id, 'triggers': triggers.to_list(), 'root': root.to_tree_dict() if root is not None else None, } # 更新任务名称 def set_name(self, args: public.dict_obj): args.validate([ public.Param('id').Require().Integer('>', 0), public.Param('name').Require().String('>', 0), ]) with self.query() as query: query.where('id', args.id).update({ 'name': args.name }) return public.lang('Success') # 获取节点树 def get_workflow(self, args: public.dict_obj): args.validate([ public.Param('id').Require().Integer('>', 0), ]) with self.query() as query: triggers_str = query.where('id', args.id).value('triggers') if triggers_str is None: raise public.HintException(public.lang('Invalid automation_id: {}', args.id)) try: triggers = Triggers(json.loads(triggers_str)) except: raise public.HintException(public.lang('Invalid automation: failed to load triggers')) wf = Workflow(args.id) return { 'id': args.id, 'triggers': triggers.to_list(), 'root': wf.load_workflow_tree().to_tree_dict(), } # 启停邮件自动化任务 def set_status(self, args: public.dict_obj): args.validate([ public.Param('id').Require().Integer('>', 0), public.Param('status').Require().Integer('in', [0, 1]), ]) # 启动任务时检查所有节点是否配置 if args.status == 1: # 检查triggers是否配置 with self.query() as query: try: triggers = Triggers(json.loads(query.where('id', args.id).value('triggers'))) except: raise public.HintException(public.lang('Invalid automation: failed to load triggers')) if not triggers.complete_check(): raise public.HintException(public.lang('Invalid automation: incomplete triggers')) task = Task() wf = Workflow(args.id) for node in wf.walk_node_with_database(): if not node.complete: raise public.HintException(public.lang('You must complete workflow')) # 为Email节点创建邮件任务 if node.type == Node.TYPE_EMAIL: # 更新邮件任务 node.set_attribute('campaign_id', task.set_email_campaign(node)) with wf.query('nodes') as query: query.where('id', node.id).update(node.to_dict()) with self.query() as query: query.where('id', args.id).update({ 'status': args.status, }) return public.lang('Success') # 删除邮件自动化任务 def remove(self, args: public.dict_obj): args.validate([ public.Param('id').Require().Integer('>', 0), ]) with self.query() as query: # 标记软删除 query.where('id', args.id).update({ 'status': 2, }) return public.lang('Success') class ScheduleNode: STATUS_WAITING = 0 STATUS_PROCESSING = 1 STATUS_COMPLETED = 2 STATUS_BROKEN = 3 __slots__ = ['id', 'node_id', 'status', 'subscriber_email', 'update_time'] def __init__(self, node_dict: typing.Dict): self.id = 0 self.node_id = 0 self.status = self.STATUS_WAITING self.subscriber_email = '' for k in self.__slots__: if k not in node_dict: continue self.__setattr__(k, node_dict[k]) if not hasattr(self, 'update_time'): self.update_time = int(time.time()) def to_dict(self): d = {} for k in self.__slots__: if not hasattr(self, k) or k.startswith('_'): continue d[k] = getattr(self, k) return d class Scheduler: def __init__(self, automation_id: int): # 参数格式验证 public.to_dict_obj({ 'automation_id': automation_id, }).validate([ public.Param('automation_id').Integer('>', 0), ]) self.automation_id: int = automation_id self.automation: Automation = Automation() self.workflow: Workflow = Workflow(automation_id) def schedule(self, subscriber: typing.Optional[str] = None): if subscriber is None: self.__schedule_all() return # 参数格式验证 public.to_dict_obj({ 'subscriber': subscriber }).validate([ public.Param('subscriber').Email(), ]) with self.workflow.query('schedules') as query: schedule_node = query.where('subscriber_email', subscriber)\ .order('id', 'desc')\ .field('id', 'node_id', 'subscriber_email', 'status', 'update_time')\ .find() cur_time = int(time.time()) # 任务首次调度 if schedule_node is None: schedule_node = ScheduleNode({ 'id': 0, 'node_id': 0, 'subscriber_email': subscriber, 'status': ScheduleNode.STATUS_WAITING, 'update_time': cur_time, }) else: schedule_node = ScheduleNode(schedule_node) self.__schedule_one(schedule_node, cur_time) def schedule_async(self, subscriber: typing.Union[str, typing.List[str]]): if isinstance(subscriber, str): subscriber = [subscriber] # 过滤邮件 subscriber = list(filter(lambda x: match_email.match(x), subscriber)) if len(subscriber) == 0: return cur_time = int(time.time()) with self.workflow.query('schedules') as query: exists_emails = query.where_in('subscriber_email', subscriber).field('distinct `subscriber_email`').column('subscriber_email') insert_data = [] for email in subscriber: if email in exists_emails: continue insert_data.append({ 'node_id': 0, 'status': ScheduleNode.STATUS_WAITING, 'subscriber_email': email, 'create_time': cur_time, 'update_time': cur_time, }) if len(insert_data) > 0: query.insert_all(insert_data) def schedule_email_sending(self): cur_time = int(time.time()) # Add timeout threshold for stuck processing tasks (e.g., 1 hour) processing_timeout = 120 for node in self.workflow.walk_node_with_database(): if node.type not in (Node.TYPE_EMAIL,): continue logging.debug('schedule node: {} {}'.format(node.id, node.get_attribute('campaign_id', 0))) with self.workflow.query('schedules') as query: schedules = query.where('node_id', node.id).where_in('status', [ ScheduleNode.STATUS_WAITING, ScheduleNode.STATUS_PROCESSING ]).field('id', 'subscriber_email', 'status', 'update_time').select() if len(schedules) == 0: continue campaign_id = int(node.get_attribute('campaign_id', 0)) schedule_ids = set() emails = set() for schedule in schedules: # Only process waiting tasks and stuck processing tasks if schedule['status'] == ScheduleNode.STATUS_PROCESSING and ( cur_time - schedule['update_time'] < processing_timeout): continue # Skip tasks that are still being processed within timeout schedule_ids.add(schedule['id']) emails.add(schedule['subscriber_email']) if len(schedule_ids) == 0: continue if campaign_id < 1: query.where_in('id', list(schedule_ids)).update({ 'status': ScheduleNode.STATUS_BROKEN, 'update_time': cur_time, }) continue with public.S('email_task', '/www/vmail/postfixadmin') as query: if not query.where('id', campaign_id).where('type', 1).exists(): with self.workflow.query('scheduels') as query2: query2.where_in('id', list(schedule_ids)).update({ 'status': ScheduleNode.STATUS_BROKEN, 'update_time': cur_time, }) continue campaign = query.where('id', campaign_id).where('type', 1).field( 'task_process').find() # 邮件任务执行中，等待一次调度 if int(campaign['task_process']) == 1: continue recipient_file = '{}/data/mail/in_bulk/recipient/{}_verify_{}'.format(public.get_panel_path(), public.Md5(str(node.id)), campaign_id) recipient_dict = { 'gmail.com': {"count": 0, "info": list(emails)}, 'googlemail.com': {"count": 0, "info": []}, 'hotmail.com': {"count": 0, "info": []}, 'outlook.com': {"count": 0, "info": []}, 'yahoo.com': {"count": 0, "info": []}, 'icloud.com': {"count": 0, "info": []}, 'other': {"count": 0, "info": []}, } with open(recipient_file, 'w') as fp: fp.write(json.dumps(recipient_dict)) query.where('id', campaign_id).update({ 'task_process': 0, 'recipient': recipient_file, }) db_dir = '/www/vmail/bulk' db_path = '{}/task_{}.db'.format(db_dir, campaign_id) if os.path.exists(db_path): insert_data = [] for email in emails: insert_data.append({ 'recipient': email, 'is_sent': 0, 'mail_provider': email.split('@')[1], 'sent_time': 0, 'created': cur_time, }) with public.S('recipient_info', db_path) as query: query.insert_all(insert_data, option='IGNORE') logging.debug('sending emails with automation: {}'.format(list(emails))) with self.workflow.query('schedules') as query: query.where_in('id', list(schedule_ids)).update({ 'status': ScheduleNode.STATUS_PROCESSING, 'update_time': cur_time, }) ret = public.run_plugin('mail_sys', 'check_task_status_new', public.to_dict_obj({ 'task_id': node.get_attribute('campaign_id', 0), })) logging.debug(ret) if not isinstance(ret, dict): logging.debug('check_task_status_new failed: {}'.format(ret)) continue with self.workflow.query('schedules') as query: query.where_in('id', list(schedule_ids)).update({ 'status': ScheduleNode.STATUS_COMPLETED, 'update_time': cur_time, }) def sync_maillog_stat(self): logging.debug('sync automation-{} maillog statistics'.format(self.automation_id)) cur_time = int(time.time()) with self.automation.query() as query: last_stat_time = query.where('id', self.automation_id).value('case when `last_stat_time` > 0 then `last_stat_time` else `create_time` end as `last_stat_time`') start_time = int(last_stat_time) end_time = cur_time msgids = set() msgid_sent_file = '{}/data/mail/in_bulk/recipient/sent_recipient/msgid_{}.sent'.format(public.get_panel_path(), self.automation_id) msgid_sent_set = set() if os.path.exists(msgid_sent_file): with open(msgid_sent_file, 'r') as fp: for msgid in fp: msgid = msgid.strip().strip('<>') if msgid == '': continue msgid_sent_set.add(msgid) msgid_opened_file = '{}/data/mail/in_bulk/recipient/sent_recipient/msgid_{}.opened'.format(public.get_panel_path(), self.automation_id) msgid_opened_set = set() if os.path.exists(msgid_opened_file): with open(msgid_opened_file, 'r') as fp: for msgid in fp: msgid = msgid.strip().strip('<>') if msgid == '': continue msgid_opened_set.add(msgid) msgid_clicked_file = '{}/data/mail/in_bulk/recipient/sent_recipient/msgid_{}.clicked'.format(public.get_panel_path(), self.automation_id) msgid_clicked_set = set() if os.path.exists(msgid_clicked_file): with open(msgid_clicked_file, 'r') as fp: for msgid in fp: msgid = msgid.strip().strip('<>') if msgid == '': continue msgid_clicked_set.add(msgid) for node in self.workflow.walk_node_with_database(): if node.type not in (Node.TYPE_EMAIL,): continue campaign_id = int(node.get_attribute('campaign_id', 0)) msgid_file = '{}/data/mail/in_bulk/recipient/sent_recipient/msgid_{}.log'.format(public.get_panel_path(), campaign_id) if campaign_id < 1 or not os.path.exists(msgid_file): continue with open(msgid_file, 'r') as fp: for msgid in fp: msgid = msgid.strip().strip('<>') if msgid == '': continue msgids.add(msgid) # Check if there's anything new to process not_sent = msgids - msgid_sent_set not_opened = msgids - msgid_opened_set not_clicked = msgids - msgid_clicked_set if len(not_sent) + len(not_opened) + len(not_clicked) == 0: logging.debug('sync automation-{} maillog statistics >>> no data for update'.format(self.automation_id)) with self.automation.query() as query: query.where('id', self.automation_id).update({ 'last_stat_time': int(last_maillog_time()), 'update_time': cur_time, }) return with self.automation.query() as query: automation_info = query.where('id', self.automation_id).field('sent', 'delivered', 'opened', 'clicked').find() stat_data = { 'sent': len(msgids), 'delivered': int(automation_info['delivered']), 'opened': int(automation_info['opened']), 'clicked': int(automation_info['clicked']), 'last_stat_time': int(last_maillog_time()), 'update_time': cur_time, } if len(not_sent) > 0: with public.S('message_ids') as query: query.alias('mi').prefix('') query.inner_join('send_mails sm', 'mi.postfix_message_id=sm.postfix_message_id') query.where('sm.log_time > ?', start_time - 1) query.where('sm.log_time < ?', end_time + 1) query.where('sm.status', 'sent') query.where('sm.dsn like ?', '2.%') query.where_in('mi.message_id', list(not_sent)) query.field('mi.message_id') ret = query_maillog_with_time_section(query, start_time, end_time) msgid_sent_add = set() for item in ret: msgid_sent_add.add(item['message_id']) stat_data['delivered'] += int(item['message_id'] not in msgid_sent_set) if len(msgid_sent_add) > 0: with open(msgid_sent_file, 'a') as fp: fp.write('{}\n'.format('\n'.join(msgid_sent_add))) if len(not_opened) > 0: with public.S('opened') as query: query.alias('o').prefix('') query.where('o.log_time > ?', start_time - 1) query.where('o.log_time < ?', end_time + 1) query.where_in('o.message_id', list(not_opened)) query.group('o.message_id') query.field('o.message_id') ret = query_maillog_with_time_section(query, start_time, end_time) msgid_opened_add = set() for item in ret: msgid_opened_add.add(item['message_id']) stat_data['opened'] += int(item['message_id'] not in msgid_opened_set) if len(msgid_opened_add) > 0: with open(msgid_opened_file, 'a') as fp: fp.write('{}\n'.format('\n'.join(msgid_opened_add))) if len(not_clicked) > 0: with public.S('clicked') as query: query.alias('c').prefix('') query.where('c.log_time > ?', start_time - 1) query.where('c.log_time < ?', end_time + 1) query.where_in('c.message_id', list(not_clicked)) query.group('c.message_id') query.field('c.message_id') ret = query_maillog_with_time_section(query, start_time, end_time) msgid_clicked_add = set() for item in ret: msgid_clicked_add.add(item['message_id']) stat_data['clicked'] += int(item['message_id'] not in msgid_clicked_set) if len(msgid_clicked_add) > 0: with open(msgid_clicked_file, 'a') as fp: fp.write('{}\n'.format('\n'.join(msgid_clicked_add))) with self.automation.query() as query: query.where('id', self.automation_id).update(stat_data) logging.debug('sync automation-{} maillog statistics >>> OK'.format(self.automation_id)) def __schedule_one(self, schedule_node: ScheduleNode, cur_time: int = 0): if int(schedule_node.status) == ScheduleNode.STATUS_BROKEN: return if cur_time < 1: cur_time = int(time.time()) schedule_id = int(schedule_node.id) root_id = int(schedule_node.node_id) for node in self.workflow.walk_node_for_schedule(root_id): # 处理当前节点 if node.id == root_id: # 当前节点正在处理中，等待下一次调度 if schedule_node.status == ScheduleNode.STATUS_PROCESSING: break # 当前节点已处理完成，可以跳转到下一流程 if schedule_node.status == ScheduleNode.STATUS_COMPLETED: continue if node.type == Node.TYPE_DELAY: logging.debug('schedule delay node - {} {} {}'.format(node.id, node.get_attribute('value', 0), node.get_attribute('unit', Node.DELAY_UNIT_SECONDS))) # delay节点 seconds = int(node.get_attribute('value', 0)) if node.get_attribute('unit', Node.DELAY_UNIT_SECONDS) == Node.DELAY_UNIT_MINUTES: seconds *= 60 elif node.get_attribute('unit', Node.DELAY_UNIT_SECONDS) == Node.DELAY_UNIT_HOURS: seconds *= 3600 elif node.get_attribute('unit', Node.DELAY_UNIT_SECONDS) == Node.DELAY_UNIT_DAYS: seconds *= 86400 # 当前任务未到可执行下一步的时间 if schedule_node.update_time + seconds > cur_time: return # 更新Delay节点已完成 with self.workflow.query('schedules') as query: query.where('id', schedule_id).update({ 'status': ScheduleNode.STATUS_COMPLETED, 'update_time': cur_time, }) continue schedule_node.node_id = node.id schedule_node.status = ScheduleNode.STATUS_PROCESSING schedule_node.update_time = cur_time # 更新当前任务正在进行节点 if schedule_id == 0: with self.workflow.query('schedules') as query: schedule_id = query.insert(schedule_node.to_dict()) else: with self.workflow.query('schedules') as query: query.where('id', schedule_id).update(schedule_node.to_dict()) logging.debug('schedule node: {} {} {}'.format(node.type, node.id, schedule_node.status)) # 处理子节点 if node.type == Node.TYPE_CONDITION: # Condition节点 condition_rules = ConditionRules(node.get_attribute('rules', []), node.get_attribute('logic_type', Node.CONDITION_LOGIC_AND)) condition_matched = condition_rules.match() for sub_node in self.workflow.walk_node_for_schedule(node.id): if sub_node.id == node.id: continue if condition_matched and node.get_attribute('condition_branch', '') == Node.CONDITION_BRANCH_YES: # Yes分支 # 更新当前任务正在进行节点 with self.workflow.query('schedules') as query: query.where('id', schedule_id).update({ 'node_id': sub_node.id, 'status': ScheduleNode.STATUS_WAITING, 'update_time': cur_time, }) break if not condition_matched and node.get_attribute('condition_branch', '') == Node.CONDITION_BRANCH_NO: # No分支 # 更新当前任务正在进行节点 with self.workflow.query('schedules') as query: query.where('id', schedule_id).update({ 'node_id': sub_node.id, 'status': ScheduleNode.STATUS_WAITING, 'update_time': cur_time, }) break break elif node.type == Node.TYPE_ABTEST: # TODO Abtest节点（留到下一次迭代） pass elif node.type == Node.TYPE_EMAIL: # Email节点 if schedule_node.status in (ScheduleNode.STATUS_WAITING,): break with self.workflow.query('schedules') as query: query.where('id', schedule_id).update({ 'status': ScheduleNode.STATUS_WAITING, }) logging.debug('update Email node status to waiting') # Task().send_email_with_campaign(node, schedule_node.subscriber_email) break elif node.type == Node.TYPE_WEBHOOK: # Webhook节点 # 发送请求 logging.debug('send request to {}'.format(node.get_attribute('url'))) threading.Thread(target=public.HttpPost, args=(node.get_attribute('url'), { 'secret': node.get_attribute('secret', ''), 'subscriber': schedule_node.subscriber_email, })).start() elif node.type == Node.TYPE_ACTION: # Action节点 if node.get_attribute('action', '') == Node.ACTION_ADD_TO_SUBSCRIBERS: # TODO 将订阅者邮箱添加到指定用户组中 pass elif node.get_attribute('action', '') == Node.ACTION_REMOVE_FROM_SUBSCRIBERS: # TODO 将订阅者邮箱从指定用户组中移除 pass elif node.get_attribute('action', '') == Node.ACTION_MARK_AS_UNSUBSCRIBE: # TODO 将订阅者邮箱标记为 “已退订” pass elif node.get_attribute('action', '') == Node.ACTION_MOVE_TO_STEP: if node.get_attribute('next_node_id', 0) > 0: # 更新当前任务正在进行的节点 with self.workflow.query('schedules') as query: query.where('id', schedule_id).update({ 'node_id': node.get_attribute('next_node_id', 0), 'status': ScheduleNode.STATUS_WAITING, 'update_time': cur_time, }) break elif node.type == Node.TYPE_DELAY: # Delay节点 if schedule_node.status in (ScheduleNode.STATUS_WAITING,): break with self.workflow.query('schedules') as query: query.where('id', schedule_id).update({ 'status': ScheduleNode.STATUS_WAITING, }) logging.debug('update Delay node status to waiting') break # 更新该阶段的处理已完成 with self.workflow.query('schedules') as query: query.where('id', schedule_id).update({ 'status': ScheduleNode.STATUS_COMPLETED, 'update_time': cur_time, }) def __schedule_all(self): with self.workflow.query('schedules') as query: schedule_nodes = query.order('id', 'desc') \ .field('id', 'node_id', 'subscriber_email', 'status', 'update_time') \ .select() if len(schedule_nodes) == 0: return for schedule_node in schedule_nodes: self.__schedule_one(ScheduleNode(schedule_node)) class Task: def __init__(self): self.__interval = 15 self.__automation = Automation() def set_email_campaign(self, node: Node) -> int: if node.type != Node.TYPE_EMAIL: return False cur_time = int(time.time()) with public.S('email_task', '/www/vmail/postfixadmin') as query: if node.get_attribute('campaign_id', 0) < 1 or not query.where('id', node.get_attribute('campaign_id', 0)).exists(): task_id = query.insert({ 'task_name': node.get_attribute('name', node.get_attribute('subject', '')), 'addresser': node.get_attribute('from', ''), 'recipient_count': 0, 'task_process': 0, 'pause': 0, 'temp_id': node.get_attribute('email_id', 0), 'is_record': 0, 'unsubscribe': int(node.get_attribute('track_unsubscribe', False)), 'threads': 0, 'created': cur_time, 'modified': cur_time, 'start_time': 0, 'remark': '', 'etypes': '', 'recipient': '', 'subject': node.get_attribute('subject', ''), 'full_name': node.get_attribute('from_name', node.get_attribute('from', '')), 'type': 1, }) else: task_id = node.get_attribute('campaign_id', 0) query.where('id', node.get_attribute('campaign_id', 0)).update({ 'task_name': node.get_attribute('name', node.get_attribute('subject', '')), 'addresser': node.get_attribute('from', ''), 'recipient_count': 0, 'task_process': 0, 'pause': 0, 'temp_id': node.get_attribute('email_id', 0), 'is_record': 0, 'unsubscribe': int(node.get_attribute('track_unsubscribe', False)), 'threads': 0, 'created': cur_time, 'modified': cur_time, 'start_time': 0, 'remark': '', 'etypes': '', 'recipient': '', 'subject': node.get_attribute('subject', ''), 'full_name': node.get_attribute('from_name', node.get_attribute('from', '')), 'type': 1, }) db_dir = '/www/vmail/bulk' db_path = '{}/task_{}'.format(db_dir, task_id) if not os.path.exists(db_dir): os.makedirs(db_dir) os.system('chown -R vmail:mail /www/vmail/bulk') # 建表 # 全量统计 message_id与收件人联合唯一 sql = '''CREATE TABLE IF NOT EXISTS `task_count` ( `id` INTEGER PRIMARY KEY AUTOINCREMENT, `queue_id` varchar(320) NOT NULL, -- 邮件队列id `message_id` TEXT NOT NULL, -- 邮件 message_id `created` INTEGER NOT NULL, -- 邮件时间 时间戳 `recipient` varchar(320) NOT NULL, -- 收件人 `delay` varchar(320) NOT NULL, -- 延时 `delays` varchar(320) NOT NULL, -- 各阶段延时 `dsn` varchar(320) NOT NULL, -- dsn `relay` text NOT NULL, -- 中继服务器 `domain` varchar(320) NOT NULL, -- 域名 `status` varchar(255) NOT NULL, -- 状态 `code` INTEGER, -- 状态码 250 5xx 101 `err_info` text NOT NULL, -- 详情 UNIQUE(message_id,recipient) ); CREATE INDEX IF NOT EXISTS `message_id_recipient_index` ON `task_count` (`message_id`, `recipient`, `queue_id`); CREATE INDEX IF NOT EXISTS `id_status_index` ON `task_count` (`id`, `status`); CREATE TABLE IF NOT EXISTS `recipient_info` ( `id` INTEGER PRIMARY KEY AUTOINCREMENT, `recipient` varchar(320) NOT NULL, -- 收件人 `is_sent` tinyint(1) NOT NULL DEFAULT 0, -- 是否发送 `mail_provider` varchar(320) NOT NULL, -- 邮件提供商域名 `sent_time` INTEGER NOT NULL, -- 发送时间 `created` INTEGER NOT NULL, -- 添加时间 UNIQUE(recipient) ); create index if not exists `recipient_isSent` on `recipient_info` (`recipient`, `is_sent`);''' with public.S(db_name=db_path) as query: query.execute_script(sql) return task_id def send_email_with_campaign(self, node: Node, to: str) -> bool: campaign_id = node.get_attribute('campaign_id', 0) if campaign_id < 1: return False with public.S('email_task', '/www/vmail/postfixadmin') as query: campaign = query.where('id', campaign_id).where('type', 1).field('task_process').find() if campaign is None or int(campaign['task_process']) == 1: return False recipient_file = '{}/data/mail/in_bulk/recipient/{}_verify_{}'.format(public.get_panel_path(), public.Md5(str(node.id)), campaign_id) recipient_dict = { 'gmail.com': {"count": 0, "info": [ to, ]}, 'googlemail.com': {"count": 0, "info": []}, 'hotmail.com': {"count": 0, "info": []}, 'outlook.com': {"count": 0, "info": []}, 'yahoo.com': {"count": 0, "info": []}, 'icloud.com': {"count": 0, "info": []}, 'other': {"count": 0, "info": []}, } with open(recipient_file, 'w') as fp: fp.write(json.dumps(recipient_dict)) query.where('id', campaign_id).update({ 'task_process': 0, 'recipient': recipient_file, }) ret = public.run_plugin('mail_sys', 'check_task_status_new', public.to_dict_obj({ 'task_id': campaign_id, })) logging.debug(ret) return True def walk_running_tasks(self, include_inactive=False): status_lst = [1] if include_inactive: status_lst.append(0) with self.__automation.query() as query: tasks = query.where_in('status', status_lst).field('id', 'triggers').select() for task in tasks: try: triggers = Triggers(json.loads(task['triggers'])) except: continue yield int(task['id']), triggers def schedule_async(self, automation_id: int, subscriber: typing.Union[str, typing.List[str]]): Scheduler(automation_id).schedule_async(subscriber) def schedule(self, automation_id: int, subscriber: typing.Optional[str] = None): Scheduler(automation_id).schedule(subscriber) def schedule_once(self): if self.__automation.is_init(): try: for automation_id, _ in self.walk_running_tasks(): logging.debug('schedule mail automation -- {}'.format(automation_id)) scheduler = Scheduler(automation_id) scheduler.schedule() scheduler.schedule_email_sending() scheduler.sync_maillog_stat() except: public.print_error() def schedule_forever(self): while True: time.sleep(self.__interval) if self.__automation.is_init(): try: for automation_id, _ in self.walk_running_tasks(): logging.debug('schedule mail automation -- {}'.format(automation_id)) scheduler = Scheduler(automation_id) scheduler.schedule() scheduler.schedule_email_sending() scheduler.sync_maillog_stat() except: public.print_error() # 发生未知错误时，短暂休眠以防循环过于频繁地失败 time.sleep(60) def fire(self, trigger_type: str, subscriber: typing.Union[str, typing.List[str]], group_ids: typing.Union[int, typing.List[int]] = 0): for automation_id, triggers in self.walk_running_tasks(include_inactive=True): if triggers.match(trigger_type, group_ids): try: self.schedule_async(automation_id, subscriber) except: public.print_log(public.get_error_info()) def subscriber_added(self, subscriber: typing.Union[str, typing.List[str]], group_ids: typing.Union[int, typing.List[int]] = 0): self.fire(Trigger.TYPE_SUBSCRIBER_ADDED, subscriber, group_ids) # TODO 当订阅者通过API添加到用户组 def subscribed(self, subscriber: str, group_ids: typing.Union[int, typing.List[int]] = 0): self.fire(Trigger.TYPE_SUBSCRIBED, subscriber, group_ids) # TODO 当订阅者打开邮件 def opened(self, subscriber: str, group_ids: typing.Union[int, typing.List[int]] = 0): self.fire(Trigger.TYPE_OPENED, subscriber, group_ids) # TODO 当订阅者点击邮件链接 def clicked(self, subscriber: str, group_ids: typing.Union[int, typing.List[int]] = 0): self.fire(Trigger.TYPE_CLICKED, subscriber, group_ids) # TODO 当订阅者退订邮件 def unsubscribed(self, subscriber: str, group_ids: typing.Union[int, typing.List[int]] = 0): self.fire(Trigger.TYPE_UNSUBSCRIBED, subscriber, group_ids) def schedule_automations_forever(): Task().schedule_forever()