관리-도구

편집 파일: 087_ignored_by_port_fields.py

import logging
from ipaddress import ip_network

logger = logging.getLogger(__name__)

def migrate(migrator, database, fake=False, **kwargs):
    IgnoredByPort = migrator.orm["ignored_by_port_proto"]
    try:
        from im360.utils.net import pack_ip_network
    except ImportError:
        ips = []  # keep database structure in AV too
    else:
        q = IgnoredByPort.select(IgnoredByPort.ip).distinct().tuples()
        ips = [ip for ip, in q]

for ip in ips:
        try:
            net, mask, version = pack_ip_network(ip_network(ip))
        except ValueError:
            logger.warning("Invalid IP network %s", ip)
            IgnoredByPort.delete().where(IgnoredByPort.ip == ip).execute()
        else:
            IgnoredByPort.update(
                network_address=net, netmask=mask, version=version
            ).where(IgnoredByPort.ip == ip).execute()

if ips:
        try:
            from im360.internals import geo

with geo.reader() as geo_reader:
                for ip in ips:
                    country = geo_reader.get_id(ip)
                    IgnoredByPort.update(
                        country=country,
                    ).where(IgnoredByPort.ip == ip).execute()
        except (ImportError, OSError):
            logger.warning(
                "Failed to update countries data in ignored_by_port"
            )

migrator.add_not_null(IgnoredByPort, "network_address")
    migrator.add_not_null(
        IgnoredByPort,
        "netmask",
    )
    migrator.add_not_null(IgnoredByPort, "version")

def rollback(migrator, database, fake=False, **kwargs):
    IgnoredByPort = migrator.orm["ignored_by_port_proto"]
    migrator.drop_not_null(
        IgnoredByPort, "network_address", "netmask", "version"
    )