majuna/app/lists/mirror_mapping.py

#  pylint: disable=too-few-public-methods

import builtins
from typing import Dict, List, Union

from flask import current_app
from pydantic import BaseModel, Field
from tldextract import extract

from app.models.base import Group
from app.models.mirrors import Proxy


class MMMirror(BaseModel):
    origin_domain: str = Field(description="The full origin domain name")
    origin_domain_normalized: str = Field(description="The origin_domain with \"www.\" removed, if present")
    origin_domain_root: str = Field(description="The registered domain name of the origin, excluding subdomains")


class MirrorMapping(BaseModel):
    version: str = Field(
        description="Version number of the mirror mapping schema in use"
    )
    mappings: Dict[str, MMMirror] = Field(
        description="The domain name for the mirror"
    )
    s3_buckets: List[str] = Field(
        description="The names of all S3 buckets used for CloudFront logs"
    )

    class Config:
        title = "Mirror Mapping Version 1.1"


def mirror_mapping() -> Dict[str, Union[str, Dict[str, str]]]:
    return MirrorMapping(
        version="1.1",
        mappings={
            d.url.lstrip("https://"): MMMirror(
                origin_domain=d.origin.description[len("proxy:"):] if d.origin.description.startswith(
                    "proxy:") else d.origin.domain_name,
                origin_domain_normalized=d.origin.description[
                                         len("proxy:"):].replace("www.", "") if d.origin.description.startswith(
                    "proxy:") else d.origin.domain_name.replace("www.", ""),
                origin_domain_root=extract(d.origin.description[len("proxy:"):] if d.origin.description.startswith(
                    "proxy:") else d.origin.domain_name).registered_domain
            ) for d in Proxy.query.all() if d.url is not None
        },
        s3_buckets=[
            f"{current_app.config['GLOBAL_NAMESPACE']}-{g.group_name.lower()}-logs-cloudfront"
            for g in Group.query.filter(Group.destroyed.is_(None)).all()
        ]
    ).dict()


if getattr(builtins, "__sphinx_build__", False):
    schema = MirrorMapping.schema_json()