Da

✏️ Editing: migration_list.py

from collections import defaultdict

from parallels.core.migration_list.entities.subscription_source_info import SubscriptionSourceInfo
from parallels.core.migration_list.source_data import MigrationListSourceData
from parallels.core.registry import Registry
from parallels.core.utils.common import if_not_none, group_by, group_by_id, obj

class ExpandMigrationListSourceData(MigrationListSourceData):
    def __init__(self, global_context):
        """Class constructor

:type global_context: parallels.plesk.source.expand.global_context.ExpandGlobalMigrationContext
        """
        self._expand_objects = global_context.expand_data
        self._plesks = global_context.source_servers.keys()

def extract_subscriptions(
        self, subscriptions_mapping=None, subscription_filter=None,
        include_addon_plans=False, target_panel=None
    ):
        processed_domains = set()
        subscriptions = []

plesk_clients_by_id = group_by_id(self._expand_objects.plesk_clients, lambda c: c.id)
        expand_clients_by_id = group_by_id(self._expand_objects.expand_clients, lambda c: c.id)
        assigned_clients_by_client_id = group_by(
            self._expand_objects.assigned_plesk_clients, lambda c: c.client_id
        )
        expand_resellers_by_id = group_by_id(self._expand_objects.expand_resellers, lambda r: r.id)
        plesk_domains_by_plesk_id = group_by(self._expand_objects.plesk_domains, lambda d: d.plesk_id)
        assigned_reseller_by_domain_id = {
            d.domain_id: d.reseller_id
            for d in self._expand_objects.assigned_plesk_domains
        }

for plesk_id in self._plesks:
            for domain in plesk_domains_by_plesk_id[plesk_id]:
                owner_id = (
                    assigned_clients_by_client_id[domain.client_id][0].reseller_id
                    if domain.client_id in assigned_clients_by_client_id
                    else None
                )
                reseller_login = if_not_none(owner_id, lambda oid: expand_resellers_by_id[oid].login)
                plesk_client = plesk_clients_by_id[domain.client_id]
                if plesk_client.expand_client_id is None:
                    # Plesk Client/Plesk Domain are not assigned to Expand Client
                    if plesk_client.login != 'admin':
                        # Domain is assigned to Plesk client
                        client_login = plesk_client.login
                    else:
                        # Domain is assigned to admin in Plesk

if domain.id in assigned_reseller_by_domain_id:
                            # For domains assigned to admin in Plesk (but not assigned to any kind of client)
                            # check domain assignment to Expand Reseller - it can be:
                            # - directly assigned to reseller,
                            # - located on a server assigned to reseller
                            # - located on a server withing a server group assigned to reseller
                            # For domain that are assigned to a client, their assignment to reseller
                            # is taken from client's assignment to reseller.
                            assigned_reseller_id = assigned_reseller_by_domain_id[domain.id]
                            if assigned_reseller_id in expand_resellers_by_id:
                                reseller_login = expand_resellers_by_id[assigned_reseller_id].login

target_panel_obj = Registry.get_instance().get_context().target_panel_obj
                        if reseller_login is None and not target_panel_obj.has_admin_subscriptions_feature():
                            # Domain is assigned to Plesk admin,
                            # and not assigned to an Expand reseller -> move under default admin client
                            client_login = target_panel_obj.get_default_admin_name()
                        elif reseller_login is not None and not target_panel_obj.has_reseller_subscriptions_feature():
                            # Domain is assigned to Plesk admin, and assigned
                            # to an Expand reseller 'reseller' -> move under default reseller client
                            client_login = target_panel_obj.get_default_reseller_name(obj(login=reseller_login))
                        else:
                            # for Plesk target panel assign domain directly to admin/reseller
                            client_login = None
                else:
                    # Plesk Client/Plesk Domain are assigned to Expand Client
                    client_login = expand_clients_by_id[plesk_client.expand_client_id].info['login']

if domain.name not in processed_domains:
                    subscriptions.append(SubscriptionSourceInfo(
                        name=domain.name,
                        name_canonical=domain.name_canonical,
                        plan=domain.tmpl_name,
                        addon_plans=[],
                        reseller=reseller_login,
                        customer=client_login
                    ))
                    processed_domains.add(domain.name)

return subscriptions

def get_plans(self, target_service_templates, addon_plans=False, include_source_plans=True):
        """Return the following structure:
        plans[reseller_login][plan_name] = bool: whether this plan exists in the target panel
        reseller_login is either a reseller login or None if owner is admin
        """
        plans = defaultdict(lambda: defaultdict(bool))

if not addon_plans:
            subscriptions = self.extract_subscriptions()
            for subscription in subscriptions:
                plans[subscription.reseller][subscription.plan] = False

expand_resellers_by_id = group_by_id(self._expand_objects.expand_resellers, lambda r: r.id)
            for p in self._expand_objects.expand_plans:
                reseller_login = (
                    expand_resellers_by_id[p.reseller_id].login
                    if p.reseller_id in expand_resellers_by_id else None
                )
                plans[reseller_login][p.name] = False
        else:
            # we don't have addon plans in Expand
            pass

for reseller_login, template_names in target_service_templates.items():
            for template_name in template_names:
                plans[reseller_login][template_name] = True

return plans

def get_reseller_contacts(self):
        reseller_contacts = {}
        for reseller in self._expand_objects.expand_resellers:
            if reseller.login not in reseller_contacts:
                reseller_contacts[reseller.login] = reseller.name
        return reseller_contacts

def get_reseller_plans(self):
        """Get dictionary with keys - reseller logins, values - plans they are assigned to

For Expand this function returns empty dictionary, as source Expand reseller plans
        do not map directly to target panel plans. This could be improved in the future.

:rtype: dict[str | unicode, str | unicode]
        """
        return {}

def get_customer_contacts(self):
        customer_contacts = {}
        expand_clients_by_id = group_by_id(self._expand_objects.expand_clients, lambda c: c.id)
        for customer in self._expand_objects.plesk_clients:
            if customer.expand_client_id is None:
                customer_login = customer.login
                customer_pname = customer.pname
            else:
                customer_login = expand_clients_by_id[customer.expand_client_id].info['login']
                customer_pname = expand_clients_by_id[customer.expand_client_id].personal_info['pname']

if customer_login not in customer_contacts:
                customer_contacts[customer_login] = customer_pname
        return customer_contacts

def get_source_subscriptions(self):
        return [domain.name for domain in self._expand_objects.plesk_domains]

📁 Sid Gifari File Manager