OwlCyberSecurity - MANAGER

Edit File: compilemessages.py

import codecs
import concurrent.futures
import glob
import os
from pathlib import Path

from django.core.management.base import BaseCommand, CommandError
from django.core.management.utils import find_command, is_ignored_path, popen_wrapper

def has_bom(fn):
    with fn.open("rb") as f:
        sample = f.read(4)
    return sample.startswith(
        (codecs.BOM_UTF8, codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE)
    )

def is_writable(path):
    # Known side effect: updating file access/modified time to current time if
    # it is writable.
    try:
        with open(path, "a"):
            os.utime(path, None)
    except OSError:
        return False
    return True

class Command(BaseCommand):
    help = "Compiles .po files to .mo files for use with builtin gettext support."

requires_system_checks = []

program = "msgfmt"
    program_options = ["--check-format"]

def add_arguments(self, parser):
        parser.add_argument(
            "--locale",
            "-l",
            action="append",
            default=[],
            help="Locale(s) to process (e.g. de_AT). Default is to process all. "
            "Can be used multiple times.",
        )
        parser.add_argument(
            "--exclude",
            "-x",
            action="append",
            default=[],
            help="Locales to exclude. Default is none. Can be used multiple times.",
        )
        parser.add_argument(
            "--use-fuzzy",
            "-f",
            dest="fuzzy",
            action="store_true",
            help="Use fuzzy translations.",
        )
        parser.add_argument(
            "--ignore",
            "-i",
            action="append",
            dest="ignore_patterns",
            default=[],
            metavar="PATTERN",
            help="Ignore directories matching this glob-style pattern. "
            "Use multiple times to ignore more.",
        )

def handle(self, **options):
        locale = options["locale"]
        exclude = options["exclude"]
        ignore_patterns = set(options["ignore_patterns"])
        self.verbosity = options["verbosity"]
        if options["fuzzy"]:
            self.program_options = self.program_options + ["-f"]

if find_command(self.program) is None:
            raise CommandError(
                "Can't find %s. Make sure you have GNU gettext "
                "tools 0.15 or newer installed." % self.program
            )

basedirs = [os.path.join("conf", "locale"), "locale"]
        if os.environ.get("DJANGO_SETTINGS_MODULE"):
            from django.conf import settings

basedirs.extend(settings.LOCALE_PATHS)

# Walk entire tree, looking for locale directories
        for dirpath, dirnames, filenames in os.walk(".", topdown=True):
            for dirname in dirnames:
                if is_ignored_path(
                    os.path.normpath(os.path.join(dirpath, dirname)), ignore_patterns
                ):
                    dirnames.remove(dirname)
                elif dirname == "locale":
                    basedirs.append(os.path.join(dirpath, dirname))

# Gather existing directories.
        basedirs = set(map(os.path.abspath, filter(os.path.isdir, basedirs)))

if not basedirs:
            raise CommandError(
                "This script should be run from the Django Git "
                "checkout or your project or app tree, or with "
                "the settings module specified."
            )

# Build locale list
        all_locales = []
        for basedir in basedirs:
            locale_dirs = filter(os.path.isdir, glob.glob("%s/*" % basedir))
            all_locales.extend(map(os.path.basename, locale_dirs))

# Account for excluded locales
        locales = locale or all_locales
        locales = set(locales).difference(exclude)

self.has_errors = False
        for basedir in basedirs:
            if locales:
                dirs = [
                    os.path.join(basedir, locale, "LC_MESSAGES") for locale in locales
                ]
            else:
                dirs = [basedir]
            locations = []
            for ldir in dirs:
                for dirpath, dirnames, filenames in os.walk(ldir):
                    locations.extend(
                        (dirpath, f) for f in filenames if f.endswith(".po")
                    )
            if locations:
                self.compile_messages(locations)

if self.has_errors:
            raise CommandError("compilemessages generated one or more errors.")

def compile_messages(self, locations):
        """
        Locations is a list of tuples: [(directory, file), ...]
        """
        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = []
            for i, (dirpath, f) in enumerate(locations):
                po_path = Path(dirpath) / f
                mo_path = po_path.with_suffix(".mo")
                try:
                    if mo_path.stat().st_mtime >= po_path.stat().st_mtime:
                        if self.verbosity > 0:
                            self.stdout.write(
                                "File “%s” is already compiled and up to date."
                                % po_path
                            )
                        continue
                except FileNotFoundError:
                    pass
                if self.verbosity > 0:
                    self.stdout.write("processing file %s in %s" % (f, dirpath))

if has_bom(po_path):
                    self.stderr.write(
                        "The %s file has a BOM (Byte Order Mark). Django only "
                        "supports .po files encoded in UTF-8 and without any BOM."
                        % po_path
                    )
                    self.has_errors = True
                    continue

# Check writability on first location
                if i == 0 and not is_writable(mo_path):
                    self.stderr.write(
                        "The po files under %s are in a seemingly not writable "
                        "location. mo files will not be updated/created." % dirpath
                    )
                    self.has_errors = True
                    return

args = [self.program, *self.program_options, "-o", mo_path, po_path]
                futures.append(executor.submit(popen_wrapper, args))

for future in concurrent.futures.as_completed(futures):
                output, errors, status = future.result()
                if status:
                    if self.verbosity > 0:
                        if errors:
                            self.stderr.write(
                                "Execution of %s failed: %s" % (self.program, errors)
                            )
                        else:
                            self.stderr.write("Execution of %s failed" % self.program)
                    self.has_errors = True