taolib.testing.email_service.services

目录

taolib.testing.email_service.services#

业务逻辑层。

Submodules#

Classes#

BounceHandler

退信处理器。

EmailService

核心邮件发送服务。

QueueProcessor

邮件队列处理器。

SubscriptionService

订阅管理服务。

TemplateService

模板管理服务。

EmailAnalytics

邮件分析数据。

TrackingService

追踪和分析服务。

Package Contents#

class taolib.testing.email_service.services.BounceHandler(tracking_service: taolib.testing.email_service.services.tracking_service.TrackingService, subscription_service: taolib.testing.email_service.services.subscription_service.SubscriptionService, email_repo: taolib.testing.email_service.repository.email_repo.EmailRepository, hard_bounce_threshold: int = 1)#

退信处理器。

_tracking#
_subscription#
_email_repo#
_hard_bounce_threshold = 1#
async handle_bounce(email_id: str, bounce_type: taolib.testing.email_service.models.enums.BounceType, reason: str, recipient: str, provider: str | None = None, raw_payload: dict | None = None) → None#

处理退信事件。

参数:

  • email_id -- 邮件 ID

  • bounce_type -- 退信类型

  • reason -- 退信原因

  • recipient -- 收件人

  • provider -- 提供商名称

  • raw_payload -- 原始 Webhook 数据

async handle_complaint(email_id: str, recipient: str, provider: str | None = None, raw_payload: dict | None = None) → None#

处理投诉事件。

投诉视为退订请求。

参数:

  • email_id -- 邮件 ID

  • recipient -- 收件人

  • provider -- 提供商名称

  • raw_payload -- 原始 Webhook 数据

class taolib.testing.email_service.services.EmailService(email_repo: taolib.testing.email_service.repository.email_repo.EmailRepository, template_service: taolib.testing.email_service.services.template_service.TemplateService, subscription_service: taolib.testing.email_service.services.subscription_service.SubscriptionService, provider_manager: taolib.testing.email_service.providers.failover.ProviderFailoverManager, queue: taolib.testing.email_service.queue.protocol.EmailQueueProtocol, tracking_service: taolib.testing.email_service.services.tracking_service.TrackingService)#

核心邮件发送服务。

编排完整的邮件发送流程： 1. 模板渲染（如有模板） 2. 订阅状态检查（营销邮件） 3. 创建邮件文档 4. 入队或直接发送

_email_repo#
_template_service#
_subscription_service#
_provider_manager#
_queue#
_tracking_service#
async send_email(data: taolib.testing.email_service.models.email.EmailCreate, enqueue: bool = True) → taolib.testing.email_service.models.email.EmailResponse#

发送单封邮件。

参数:

  • data -- 邮件创建数据

  • enqueue -- 是否入队异步发送，False 则直接发送

返回:

邮件响应

async send_bulk(emails: list[taolib.testing.email_service.models.email.EmailCreate]) → list[taolib.testing.email_service.models.email.EmailResponse]#

批量发送邮件。

async _send_now(email_doc: taolib.testing.email_service.models.email.EmailDocument) → taolib.testing.email_service.models.email.EmailDocument#

立即发送邮件。

参数:

email_doc -- 邮件文档

返回:

更新后的邮件文档

async get_email(email_id: str) → taolib.testing.email_service.models.email.EmailResponse#

获取邮件详情。

async list_emails(status: taolib.testing.email_service.models.enums.EmailStatus | None = None, email_type: taolib.testing.email_service.models.enums.EmailType | None = None, skip: int = 0, limit: int = 100) → list[taolib.testing.email_service.models.email.EmailResponse]#

查询邮件列表。

class taolib.testing.email_service.services.QueueProcessor(queue: taolib.testing.email_service.queue.protocol.EmailQueueProtocol, email_repo: taolib.testing.email_service.repository.email_repo.EmailRepository, send_callback, poll_interval: float = 1.0, batch_size: int = 10)#

邮件队列处理器。

作为后台 asyncio.Task 运行，持续从队列取出邮件并发送。 参考 data_sync.services.scheduler.AsyncScheduler 模式。

_queue#
_email_repo#
_send_callback#
_poll_interval = 1.0#
_batch_size = 10#
_running = False#
_task: asyncio.Task[None] | None = None#
async start() → None#

启动处理器。

async stop() → None#

停止处理器。

async _run_loop() → None#

主处理循环。

async _process_batch() → None#

处理一批邮件。

async _process_single(email_id: str) → None#

处理单封邮件。

async process_scheduled() → None#

处理到期的计划邮件。

class taolib.testing.email_service.services.SubscriptionService(subscription_repo: taolib.testing.email_service.repository.subscription_repo.SubscriptionRepository)#

订阅管理服务。

_repo#
async get_or_create_subscription(email: str) → taolib.testing.email_service.models.subscription.SubscriptionDocument#

获取或创建订阅记录。

如果邮箱没有订阅记录，则创建一个新的激活状态记录。

参数:

email -- 邮箱地址

返回:

订阅文档

async unsubscribe(token: str, reason: str | None = None) → taolib.testing.email_service.models.subscription.SubscriptionResponse#

处理退订请求。

参数:

  • token -- 退订令牌

  • reason -- 退订原因

返回:

更新后的订阅响应

抛出:

SubscriptionError -- 令牌无效或已退订

async unsubscribe_by_email(email: str, reason: str | None = None) → taolib.testing.email_service.models.subscription.SubscriptionResponse | None#

按邮箱退订（用于硬退信自动退订）。

async resubscribe(email: str) → taolib.testing.email_service.models.subscription.SubscriptionResponse#

重新订阅。

参数:

email -- 邮箱地址

返回:

更新后的订阅响应

抛出:

SubscriptionError -- 订阅记录不存在

async is_subscribed(email: str) → bool#

检查是否订阅。

async get_unsubscribe_token(email: str) → str#

获取退订令牌（自动创建订阅记录）。

class taolib.testing.email_service.services.TemplateService(template_repo: taolib.testing.email_service.repository.template_repo.TemplateRepository, engine: taolib.testing.email_service.template.engine.TemplateEngine)#

模板管理服务。

_repo#
_engine#
async create_template(data: taolib.testing.email_service.models.template.TemplateCreate) → taolib.testing.email_service.models.template.TemplateResponse#

创建模板。

async update_template(template_id: str, data: taolib.testing.email_service.models.template.TemplateUpdate) → taolib.testing.email_service.models.template.TemplateResponse | None#

更新模板。

async get_template(template_id: str) → taolib.testing.email_service.models.template.TemplateResponse | None#

获取模板。

async get_template_by_name(name: str) → taolib.testing.email_service.models.template.TemplateResponse | None#

按名称获取模板。

async list_templates(email_type: taolib.testing.email_service.models.enums.EmailType | None = None, is_active: bool | None = None, skip: int = 0, limit: int = 100) → list[taolib.testing.email_service.models.template.TemplateResponse]#

查询模板列表。

async delete_template(template_id: str) → bool#

删除模板。

async render_template(template_id: str, variables: dict, email_type: taolib.testing.email_service.models.enums.EmailType = EmailType.TRANSACTIONAL, recipient_email: str | None = None, unsubscribe_token: str | None = None) → taolib.testing.email_service.template.engine.RenderedEmail#

渲染模板。

参数:

  • template_id -- 模板 ID

  • variables -- 模板变量

  • email_type -- 邮件类型

  • recipient_email -- 收件人邮箱

  • unsubscribe_token -- 退订令牌

返回:

渲染后的邮件内容

抛出:

TemplateNotFoundError -- 模板不存在

class taolib.testing.email_service.services.EmailAnalytics#

Bases: pydantic.BaseModel

邮件分析数据。

total_sent: int#
total_delivered: int#
total_opened: int#
total_clicked: int#
total_bounced: int#
total_failed: int#
delivery_rate: float#
open_rate: float#
click_rate: float#
bounce_rate: float#
class taolib.testing.email_service.services.TrackingService(tracking_repo: taolib.testing.email_service.repository.tracking_repo.TrackingRepository, email_repo: taolib.testing.email_service.repository.email_repo.EmailRepository)#

追踪和分析服务。

_tracking_repo#
_email_repo#
async record_event(email_id: str, event_type: taolib.testing.email_service.models.enums.TrackingEventType, recipient: str, provider: str | None = None, ip_address: str | None = None, user_agent: str | None = None, click_url: str | None = None, bounce_type: str | None = None, bounce_reason: str | None = None, raw_payload: dict | None = None) → taolib.testing.email_service.models.tracking.TrackingEventResponse#

记录追踪事件。

同时更新关联的 EmailDocument 状态。

async get_events_for_email(email_id: str) → list[taolib.testing.email_service.models.tracking.TrackingEventResponse]#

获取邮件的所有追踪事件。

async get_analytics(start: datetime.datetime, end: datetime.datetime) → EmailAnalytics#

获取时间范围内的分析数据。

async get_daily_stats(start: datetime.datetime, end: datetime.datetime) → list[dict]#

获取按日统计数据。

async _update_email_status(email_id: str, event_type: taolib.testing.email_service.models.enums.TrackingEventType, timestamp: datetime.datetime) → None#

根据追踪事件更新邮件状态。

目录