File size: 5,953 Bytes
89419ea | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
from __future__ import annotations
import random
from dataclasses import dataclass
from typing import Any, Optional
from sqlalchemy.orm import Session
from ..database import crud
@dataclass
class EmailItemAssignment:
item_index: int
service_id: Optional[int]
service_type: str
failure_reason: Optional[str] = None
config_snapshot: Optional[dict[str, Any]] = None
@dataclass
class UploadTarget:
provider: str
service_id: int
service_name: str
config_snapshot: dict[str, Any]
def _get_enabled_email_services(db: Session, email_type: str):
return crud.get_email_services(db, service_type=email_type, enabled=True)
def _get_email_service_or_raise(db: Session, email_type: str, requested_service_id: int):
service = crud.get_email_service_by_id(db, requested_service_id)
if not service:
raise ValueError(f"email service {requested_service_id} not found")
if not service.enabled:
raise ValueError(f"email service {requested_service_id} is disabled")
if service.service_type != email_type:
raise ValueError(f"email service {requested_service_id} does not belong to type {email_type}")
return service
def build_email_item_assignments(
db: Session,
*,
email_type: str,
count: int,
requested_service_id: Optional[int] = None,
) -> list[EmailItemAssignment]:
if count < 1:
raise ValueError('count must be >= 1')
if requested_service_id is not None:
service = _get_email_service_or_raise(db, email_type, requested_service_id)
if email_type == 'outlook' and count > 1:
raise ValueError('outlook requested_service_id cannot be reused when count > 1')
return [
EmailItemAssignment(
item_index=index,
service_id=service.id,
service_type=email_type,
config_snapshot=(service.config or {}).copy(),
)
for index in range(count)
]
services = _get_enabled_email_services(db, email_type)
if not services:
raise ValueError(f'no enabled email services for type {email_type}')
if email_type == 'outlook':
assignments: list[EmailItemAssignment] = []
for index in range(count):
if index < len(services):
service = services[index]
assignments.append(
EmailItemAssignment(
item_index=index,
service_id=service.id,
service_type=email_type,
config_snapshot=(service.config or {}).copy(),
)
)
else:
assignments.append(
EmailItemAssignment(
item_index=index,
service_id=None,
service_type=email_type,
failure_reason='no_available_email_service',
)
)
return assignments
if email_type == 'temp_mail':
min_priority = min(service.priority for service in services)
pool = [service for service in services if service.priority == min_priority]
return [
EmailItemAssignment(
item_index=index,
service_id=(selected := random.choice(pool)).id,
service_type=email_type,
config_snapshot=(selected.config or {}).copy(),
)
for index in range(count)
]
service = services[0]
return [
EmailItemAssignment(
item_index=index,
service_id=service.id,
service_type=email_type,
config_snapshot=(service.config or {}).copy(),
)
for index in range(count)
]
def resolve_upload_target(db: Session, *, provider: str, requested_service_id: Optional[int] = None) -> UploadTarget:
provider = (provider or '').strip().lower()
provider_to_crud = {
'cpa': (crud.get_cpa_service_by_id, lambda session: crud.get_cpa_services(session, enabled=True)),
'sub2api': (crud.get_sub2api_service_by_id, lambda session: crud.get_sub2api_services(session, enabled=True)),
'tm': (crud.get_tm_service_by_id, lambda session: crud.get_tm_services(session, enabled=True)),
}
if provider not in provider_to_crud:
raise ValueError(f'unsupported upload provider: {provider}')
get_one, list_enabled = provider_to_crud[provider]
if requested_service_id is not None:
service = get_one(db, requested_service_id)
if not service:
for other_provider, (other_get_one, _) in provider_to_crud.items():
if other_provider == provider:
continue
if other_get_one(db, requested_service_id):
raise ValueError(f'upload service {requested_service_id} does not belong to provider {provider}')
raise ValueError(f'upload service {requested_service_id} not found')
if not getattr(service, 'enabled', False):
raise ValueError(f'upload service {requested_service_id} is disabled')
service_pool = {svc.id for svc in list_enabled(db)}
if service.id not in service_pool:
raise ValueError(f'upload service {requested_service_id} does not belong to provider {provider}')
else:
services = list_enabled(db)
if not services:
raise ValueError(f'no enabled upload services for provider {provider}')
service = services[0]
if provider == 'cpa':
config_snapshot = {'api_url': service.api_url, 'api_token': service.api_token}
else:
config_snapshot = {'api_url': service.api_url, 'api_key': service.api_key}
return UploadTarget(
provider=provider,
service_id=service.id,
service_name=service.name,
config_snapshot=config_snapshot,
)
|