dingxin_toolbox_drf/update/utils/git_util.py

from git import Repo
from update.models import Release
import os
import shutil
import hashlib
import pymysql
from dingxin_toolbox_drf import settings

"""
功能：
1. clone代码（仅内部使用）
2. 拉去指定版本代码
3. 获取所有版本并写入数据库
4. 将指定版本的cine.sql拉取到本地
5. 对本执行版本的cine.sql和本地同版本的文件是否一致
6. 将本地的cine.sql写入数据库
"""


class GitUtil:
    def __init__(self, short_release):
        self.short_release = short_release
        self.local_code_path = os.path.join(settings.BASE_DIR, r'dx\code')
        self.cine_sql_path = os.path.join(settings.BASE_DIR, r'dx\sql')
        self.db_name = self.get_db_name(self.short_release)
        self.db_config = {
            'host': '172.16.3.112',
            'user': 'test',
            'password': 'cine123456',
            'port': '3306',
            'connect_timeout': 5,
        }

    # 执行一次 clone代码到本地
    def clone(self):
        git_url = 'http://dingxin_readonly:cine123456@172.16.3.3:8081/root/dingxin.git'
        Repo.clone_from(git_url, to_path=self.local_code_path, branch='master')

    # 获取指定版本的最新代码
    def checkout_release(self):
        local_repo = Repo(self.local_code_path)
        r = local_repo.git.checkout(self.short_release)
        print(r)
        r = local_repo.git.pull()
        print(r)

    # 复制cine.sql到本地路径cine_xxx_org.sql,然后生成新数据库名称的cine_xxx.sql
    # 如果cine.sql和cine_xxx_org.sql文件md5一致则跳过
    # 如果cine.sql和cine_xxx_org.sql文件md5不一致，则重新拷贝cine_xxx_org.sql并生成cine_xxx.sql
    # 如果生成新的cine_xxx.sql则返回True，否则返回False
    def create_cine_sql(self):
        self.checkout_release()
        org_path = os.path.join(self.local_code_path, 'install', 'cine.sql')
        target_org_path = os.path.join(self.cine_sql_path, f'{self.db_name}_org.sql')
        target_path = os.path.join(self.cine_sql_path, f'{self.db_name}.sql')

        if os.path.exists(target_org_path):
            if self.get_md5(org_path) == self.get_md5(target_org_path):
                print('md5相同')
                if os.path.exists(target_path):
                    return False
            else:
                print('md5不同')
                os.remove(target_org_path)
                os.remove(target_path)
                shutil.copyfile(org_path, target_org_path)
        else:
            shutil.copyfile(org_path, target_org_path)

        with open(target_path, 'w', encoding='utf-8') as target_file:
            with open(org_path, 'r', encoding='utf') as f:
                lines = f.readlines()
                target_lines = []
                for line in lines:
                    if line.startswith('CREATE DATABASE'):
                        print('CREATE DATABASE')
                        target_lines.append(f'DROP DATABASE IF EXISTS `{self.db_name}`;\r\n')
                        target_lines.append(
                            f'CREATE DATABASE /*!32312 IF NOT EXISTS*/`{self.db_name}` /*!40100 DEFAULT CHARACTER SET utf8 */;\r\n')
                    elif line.startswith('USE `cine`'):
                        print('USE `cine`')
                        target_lines.append(f'USE `{self.db_name}`;\r\n')
                    else:
                        target_lines.append(line)
                f.close()
            target_file.writelines(target_lines)
            target_file.close()
            return True

    # 将生成的cine.sql写入数据库
    def write_cine_sql_by_mysql(self):
        if os.path.exists(sql_path := os.path.join(self.cine_sql_path, f'{self.db_name}.sql')):
            cmd = f'mysql -h{self.db_config["host"]} -P{self.db_config["port"]} -u{self.db_config["user"]} -p{self.db_config["password"]} < {sql_path}'
            r = os.system(cmd)
            print(r)

    # 获取数据库名称
    def get_db_name(self, _release=''):
        if _release == '':
            release = self.short_release
        else:
            release = _release
        print(release)
        rel = Release.objects.filter(short_release=release).first()
        print(rel)
        return 'cine_' + str(rel.ver_id)

    # 写入cine.sql
    def handle_create_cine(self):
        # 如果生成了新的文件则写库，如果sql没有变化则跳过
        if self.create_cine_sql():
            self.write_cine_sql_by_mysql()

    # # 获取release信息
    # def get_release(self, _release=''):
    #     if _release == '':
    #         release = self.short_release
    #     else:
    #         release = _release
    #     rel = Release.objects.filter(short_release=release).first()
    #     return rel.release

    @staticmethod
    def get_short_version():
        return Release.objects.exclude(status='0').order_by('-ver_id').all()

    @staticmethod
    def get_md5(path):
        with open(path, "rb") as f:
            m = hashlib.md5()  # 创建md5对象
            with open(path, 'rb') as fobj:
                while True:
                    data = fobj.read(4096)
                    if not data:
                        break
                    m.update(data)  # 更新md5对象
            return m.hexdigest()  # 返回md5对象


class GitDbUtil:
    def __init__(self):
        self.local_code_path = os.path.join(settings.BASE_DIR, r'dx\code')

    # 获取可用的版本，可用状态需要手动维护数据库
    @staticmethod
    def get_short_version():
        return Release.objects.exclude(status='0').order_by('-ver_id').all()

    # 从git中获取版本列表后和数据库对比，添加新的版本到数据库
    def set_release_to_db(self):
        local_repo = Repo(self.local_code_path)
        release = []
        for remote in local_repo.remotes:
            for ref in remote.refs:
                release.append(ref.name)
        saved_release = Release.objects.all()
        saved_release_version_list = [rel.release for rel in saved_release]
        for rel in release:
            if rel in saved_release_version_list:
                print('pass')
            else:
                data = {
                    'release': rel,
                    'short_release': rel.replace('origin/', ''),
                    'main_ver': rel[7:18],
                    'ver_id': int(rel[14:18]),
                    'status': '1'
                }
                Release.objects.create(**data)

# if __name__ == '__main__':
#     git_util = GitUtil()
#     # git_util.clone()