subconverter/scripts/update_rules.py

import argparse
import configparser
import glob
import logging
import os
import shutil
import stat
from git import InvalidGitRepositoryError, Repo


def del_rw(action, name: str, exc):
    os.chmod(name, stat.S_IWRITE)
    os.remove(name)


def open_repo(path: str):
    if not os.path.exists(path):
        return None
    try:
        return Repo(path)
    except InvalidGitRepositoryError:
        return None


def update_rules(repo_path: str, save_path: str, matches: list[str], keep_tree: bool):
    os.makedirs(save_path, exist_ok=True)
    for pattern in matches:
        files = glob.glob(os.path.join(repo_path, pattern), recursive=True)
        if len(files) == 0:
            logging.warn(f"no files found for pattern {pattern}")
            continue
        for file in files:
            if os.path.isdir(file):
                continue
            file_rel_path, file_name = os.path.split(os.path.relpath(file, repo_path))
            if keep_tree:
                file_dest_dir = os.path.join(save_path, file_rel_path)
                os.makedirs(file_dest_dir, exist_ok=True)
                file_dest_path = os.path.join(file_dest_dir, file_name)
            else:
                file_dest_path = os.path.join(save_path, file_name)
            shutil.copy2(file, file_dest_path)
            logging.info(f"copied {file} to {file_dest_path}")

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-c", "--config", default="rules_config.conf")
    args = parser.parse_args()

    config = configparser.ConfigParser()
    config.read(args.config)
    logging.basicConfig(format="%(asctime)s %(message)s", level=logging.DEBUG)

    for section in config.sections():
        repo = config.get(section, "name", fallback=section)
        url = config.get(section, "url")
        commit = config.get(section, "commit", fallback=None)
        branch = config.get(section, "branch", fallback=None)
        matches = config.get(section, "match").split("|")
        save_path = config.get(section, "dest", fallback=f"base/rules/{repo}")
        keep_tree = config.getboolean(section, "keep_tree", fallback=True)

        logging.info(f"reading files from url {url}, matches {matches}, save to {save_path} keep_tree {keep_tree}")

        repo_path = os.path.join("./tmp/repo/", repo)

        r = open_repo(repo_path)
        if r is None:
            logging.info(f"cloning repo {url} to {repo_path}")
            r = Repo.clone_from(url, repo_path)
        else:
            logging.info(f"repo {repo_path} exists")
            
        try:
            if commit is not None:
                logging.info(f"checking out to commit {commit}")
                r.git.checkout(commit)
            elif branch is not None:
                logging.info(f"checking out to branch {branch}")
                r.git.checkout(branch)
            else:
                logging.info(f"checking out to default branch")
                r.active_branch.checkout()
        except Exception as e:
            logging.error(f"checkout failed {e}")
            continue

        update_rules(repo_path, save_path, matches, keep_tree)

    shutil.rmtree("./tmp", ignore_errors=True)


if __name__ == "__main__":
    main()