将数据库合并并添加了配置文件

bin/__init__.py

@@ -7,7 +7,136 @@
 
 
 import os
+import hashlib
+import sys
+import requests
+import threading
+from bin.utils.logger import logger
+from bin.utils.settings import Settings
 
-if not os.path.exists('./bin/log'):
-    os.mkdir('./bin/log')
 
+class MulThreadDownload(threading.Thread):
+    def __init__(self, url, startpos, endpos, f, name):
+        super(MulThreadDownload, self).__init__()
+        self.session = requests.Session()
+        self.session.trust_env = False
+        self.url = url  # 资源Url
+        self.startpos = startpos
+        self.endpos = endpos
+        self.fd = f  # 文件操作
+        self.name = name  # 线程名称
+
+    def download(self):
+        """
+        多线程下载
+        :return:
+        """
+        logger.info(f'线程: Thread-{self.name} 开始下载')
+        headers = {"Range": "bytes=%s-%s" % (self.startpos, self.endpos)}
+        res = self.session.get(self.url, headers=headers)
+        self.fd.seek(self.startpos)
+        self.fd.write(res.content)
+        logger.info(f'线程: Thread-{self.name} 结束下载')
+
+    def run(self):
+        """
+        此处启动
+        :return:
+        """
+        self.download()
+
+
+class Check:
+    """检查md5是否相同和下载数据库"""
+
+    def __init__(self):
+        self.db_name = conf.name
+        self.assets_url = 'https://themedatabases.vercel.app/assets'
+        self.remote_md5 = 'https://themedatabases.vercel.app/md5'
+        self.session = requests.Session()
+        self.session.trust_env = False
+
+    def check_md5(self):
+        """
+        检验本地文件md5是否和远程md5相同
+        :return:
+        """
+        with open(f'./bin/db/{self.db_name}', 'rb') as fp:
+            data = fp.read()
+        local_md5 = hashlib.md5(data).hexdigest()
+        remote_md5 = self.session.get(self.remote_md5).json()['data'][0]
+        logger.info(f'本地数据库md5: {local_md5}')
+        logger.info(f'远程数据库md5: {remote_md5}')
+        if local_md5 != remote_md5:
+            logger.error('下载错误: 本地数据库md5和远程数据库md5检验不通过, 即将开始重新下载\nI: 本次下载将使用单线程下载')
+            self.single_download()
+        else:
+            logger.info('md5检验已通过')
+
+    def download(self):
+        """
+        开始下载
+        :return:
+        """
+        filesize = int(self.session.get(self.assets_url).headers['Content-Length'])
+        threaded_count = 3
+        logger.info(f'数据库大小: {round(filesize / 1024 / 1024, 2)}Mb. 下载线程: {threaded_count}')
+        threading.BoundedSemaphore(threaded_count)
+        step = filesize // threaded_count
+        mtd_list = []
+        start = 0
+        end = -1
+        with open(f'./bin/db/{self.db_name}', 'w') as initial_file:
+            initial_file.close()
+        with open(f'./bin/db/{self.db_name}', 'rb+') as f:
+            name = 1
+            fileno = f.fileno()
+            while end < filesize - 1:
+                start = end + 1
+                end = start + step - 1
+                if end > filesize:
+                    end = filesize
+                dup = os.dup(fileno)
+                fd = os.fdopen(dup, 'rb+', -1)
+                t = MulThreadDownload(self.assets_url, start, end, fd, name)
+                name += 1
+                t.start()
+                mtd_list.append(t)
+
+            for i in mtd_list:
+                i.join()
+
+        self.check_md5()
+
+    def single_download(self):
+        """
+        单线程进行下载
+        :return:
+        """
+        session = requests.Session()
+        session.trust_env = False
+        logger.info(f'正在使用单线程下载中')
+        resp = session.get(self.assets_url)
+        with open(f'./bin/db/{self.db_name}', 'wb') as fp:
+            fp.write(resp.content)
+        logger.info('下载完成 正在检验文件md5')
+        with open(f'./bin/db/{self.db_name}', 'rb') as fp:
+            data = fp.read()
+        local_md5 = hashlib.md5(data).hexdigest()
+        remote_md5 = session.get('https://themedatabases.vercel.app/md5').json()['data'][0]
+        logger.info(f'本地数据库md5: {local_md5}')
+        logger.info(f'远程数据库md5: {remote_md5}')
+        if local_md5 != remote_md5:
+            logger.error('md5检验未通过请手动前往 https://themedatabases.vercel.app/assets 下载文件并放入./bin/db文件夹内')
+            sys.exit(-1)
+        else:
+            logger.info('md5检验已通过')
+
+
+if __name__ != '__main__':
+    conf = Settings()
+    if not os.path.exists('./bin/log'):
+        os.mkdir('./bin/log')
+    if not os.path.exists(f'./bin/db/{conf.name}'):
+        logger.error('没有检测到本地主题数据库即将开始下载')
+        Check().download()

bin/conf/config.yml

@@ -0,0 +1,22 @@
+servers:
+  # 服务器建立的地址
+  host: "0.0.0.0"
+  # 服务器监听的端口
+  port: 5000
+
+
+database:
+  # 数据库类型
+  type: sqlite
+  sqlite:
+    path: "./bin/db/data.db"
+
+  # 暂未开发
+#  mysql:
+#    address: "localhost"
+#    port: 3306
+#    user: ""
+#    password: ""
+
+
+

bin/db/__init__.py

@@ -6,16 +6,4 @@
 # @File Name: __init__.py.py
 
 
-import os
-from bin.utils.logger import logger
-
-if not os.path.exists('./bin/db/count.db'):
-    logger.error('未检测到用户计数数据库')
-    import sqlite3
-    conn = sqlite3.connect('./bin/db/count.db', check_same_thread=False)
-    cursor = conn.cursor()
-    cursor.execute('create table ReqCount (name text primary key, times int)')
-    conn.commit()
-    cursor.close()
-    conn.close()
-    logger.info('已在./db 目录下创建了count.db数据库')
+__all__ = ['sqlite']

bin/db/sqlite.py

@@ -7,110 +7,108 @@
 
 import sqlite3
 from typing import Any
+from bin.utils.settings import Settings
 
 
-def insert_data(name: str) -> None:
-    """
-    插入数据
-    :param name:
-    :return:
-    """
-    conn = sqlite3.connect('./bin/db/count.db', check_same_thread=False)
-    cursor = conn.cursor()
-    try:
-        cursor.execute('insert into ReqCount values(?, ?)', (name, 1))
-        conn.commit()
-    finally:
-        cursor.close()
-        conn.close()
+class Database:
+    def __init__(self):
+        self.sqlite_path = Settings().path
 
+    def insert_data(self, name: str) -> None:
+        """
+        插入数据
+        :param name:
+        :return:
+        """
+        conn = sqlite3.connect(self.sqlite_path, check_same_thread=False)
+        cursor = conn.cursor()
+        try:
+            cursor.execute('insert into ReqCount values(?, ?)', (name, 1))
+            conn.commit()
+        finally:
+            cursor.close()
+            conn.close()
 
-def fetch_data(name: str) -> int:
-    """
-    获取数据
-    :param name:
-    :return:
-    """
-    conn = sqlite3.connect('./bin/db/count.db', check_same_thread=False)
-    cursor = conn.cursor()
-    try:
-        cursor.execute('select * from ReqCount')
-        conn.commit()
-        data = cursor.fetchall()
+    def fetch_data(self, name: str) -> int:
+        """
+        获取数据
+        :param name:
+        :return:
+        """
+        conn = sqlite3.connect(self.sqlite_path, check_same_thread=False)
+        cursor = conn.cursor()
+        try:
+            cursor.execute('select * from ReqCount')
+            conn.commit()
+            data = cursor.fetchall()
 
-        temp_dict = {}
-        for k, v in data:  # 遍历数据将元组数据转换为字典类型
-            temp_dict.setdefault(k, []).append(v)
-        for i, c in zip(temp_dict.keys(), temp_dict.values()):
-            temp_dict[i] = c[0]
+            temp_dict = {}
+            for k, v in data:  # 遍历数据将元组数据转换为字典类型
+                temp_dict.setdefault(k, []).append(v)
+            for i, c in zip(temp_dict.keys(), temp_dict.values()):
+                temp_dict[i] = c[0]
 
-        if name in temp_dict.keys():
-            count = temp_dict[name]  # 获取原数字 为 整型
-            update_data(name, count)
-            return count
-        else:
-            # 新建用户数据
-            insert_data(name)
-            return 0
-    finally:
-        cursor.close()
-        conn.close()
+            if name in temp_dict.keys():
+                count = temp_dict[name]  # 获取原数字 为 整型
+                self.update_data(name, count)
+                return count
+            else:
+                # 新建用户数据
+                self.insert_data(name)
+                return 0
+        finally:
+            cursor.close()
+            conn.close()
 
+    def update_data(self, name: str, times: int) -> None:
+        """
+        更新数据
+        :param name:
+        :param times:
+        :return:
+        """
+        conn = sqlite3.connect(self.sqlite_path, check_same_thread=False)
+        cursor = conn.cursor()
+        try:
+            times += 1
+            cursor.execute('update ReqCount set times=? where name=?', (times, name))
+            conn.commit()
+        finally:
+            cursor.close()
+            conn.close()
 
-def update_data(name: str, times: int) -> None:
-    """
-    更新数据
-    :param name:
-    :param times:
-    :return:
-    """
-    conn = sqlite3.connect('./bin/db/count.db', check_same_thread=False)
-    cursor = conn.cursor()
-    try:
-        times += 1
-        cursor.execute('update ReqCount set times=? where name=?', (times, name))
-        conn.commit()
-    finally:
-        cursor.close()
-        conn.close()
+    def fetch_table(self) -> list:
+        """
+        返回已有主题列表
+        :return:
+        """
+        conn_temp = sqlite3.connect(self.sqlite_path, check_same_thread=False)
+        cursor_temp = conn_temp.cursor()
+        try:
+            lst = []
+            cursor_temp.execute("select * from sqlite_master where type='table'")
+            for i in cursor_temp.fetchall():
+                lst.append(i[1])
+            return lst
+        finally:
+            cursor_temp.close()
+            conn_temp.close()
 
-
-def fetch_table() -> list:
-    """
-    返回已有主题列表
-    :return:
-    """
-    conn_temp = sqlite3.connect('./bin/db/style.db', check_same_thread=False)
-    cursor_temp = conn_temp.cursor()
-    try:
-        lst = []
-        cursor_temp.execute("select * from sqlite_master where type='table'")
-        for i in cursor_temp.fetchall():
-            lst.append(i[1])
-        return lst
-    finally:
-        cursor_temp.close()
-        conn_temp.close()
-
-
-def fetch_style_data(style: str) -> list[dict[str, Any]]:
-    """
-    获取主题数据库内的数据
-    :param style:
-    :return:
-    """
-    conn = sqlite3.connect('./bin/db/style.db', check_same_thread=False)
-    cursor = conn.cursor()
-    try:
-        cursor.execute('select * from %(style_name)s' % {'style_name': style})
-        data = cursor.fetchall()
-        data_set = []
-        for i in data:
-            data_set.append({'index': i[0], 'base64': i[1], 'width': i[2], 'height': i[3]})
-        return data_set
-    finally:
-        cursor.close()
-        conn.close()
-
-
-__all__ = ['fetch_table', 'fetch_data', 'fetch_style_data', 'update_data', 'insert_data']
+    def fetch_style_data(self, style: str) -> list[dict[str, Any]]:
+        """
+        获取主题数据库内的数据
+        :param style:
+        :return:
+        """
+        conn = sqlite3.connect(self.sqlite_path, check_same_thread=False)
+        cursor = conn.cursor()
+        try:
+            cursor.execute('select * from %(style_name)s' % {'style_name': style})
+            data = cursor.fetchall()
+            data_set = []
+            for i in data:
+                data_set.append({'index': i[0], 'base64': i[1], 'width': i[2], 'height': i[3]})
+            return data_set
+        finally:
+            cursor.close()
+            conn.close()

bin/utils/__init__.py

@@ -6,133 +6,4 @@
 # @File Name: __init__.py
 
 
-import os
-import hashlib
-import sys
-import requests
-import threading
-from bin.utils.logger import logger
-
-
-class MulThreadDownload(threading.Thread):
-    def __init__(self, url, startpos, endpos, f, name):
-        super(MulThreadDownload, self).__init__()
-        self.session = requests.Session()
-        self.session.trust_env = False
-        self.url = url  # 资源Url
-        self.startpos = startpos
-        self.endpos = endpos
-        self.fd = f  # 文件操作
-        self.name = name  # 线程名称
-
-    def download(self):
-        """
-        多线程下载
-        :return:
-        """
-        logger.info(f'线程: Thread-{self.name} 开始下载')
-        headers = {"Range": "bytes=%s-%s" % (self.startpos, self.endpos)}
-        res = self.session.get(self.url, headers=headers)
-        self.fd.seek(self.startpos)
-        self.fd.write(res.content)
-        logger.info(f'线程: Thread-{self.name} 结束下载')
-
-    def run(self):
-        """
-        此处启动
-        :return:
-        """
-        self.download()
-
-
-class Check:
-    """检查md5是否相同和下载数据库"""
-
-    def __init__(self):
-        self.assets_url = 'https://themedatabases.vercel.app/assets'
-        self.remote_md5 = 'https://themedatabases.vercel.app/md5'
-        self.session = requests.Session()
-        self.session.trust_env = False
-
-    def check_md5(self):
-        """
-        检验本地文件md5是否和远程md5相同
-        :return:
-        """
-        with open('./bin/db/style.db', 'rb') as fp:
-            data = fp.read()
-        local_md5 = hashlib.md5(data).hexdigest()
-        remote_md5 = self.session.get(self.remote_md5).json()['data'][0]
-        logger.info(f'本地数据库md5: {local_md5}')
-        logger.info(f'远程数据库md5: {remote_md5}')
-        if local_md5 != remote_md5:
-            logger.error('下载错误: 本地数据库md5和远程数据库md5检验不通过, 即将开始重新下载\nI: 本次下载将使用单线程下载')
-            self.single_download()
-        else:
-            logger.info('md5检验已通过')
-
-    def download(self):
-        """
-        开始下载
-        :return:
-        """
-        filesize = int(self.session.get(self.assets_url).headers['Content-Length'])
-        threaded_count = 3
-        logger.info(f'数据库大小: {round(filesize / 1024 / 1024, 2)}Mb. 下载线程: {threaded_count}')
-        threading.BoundedSemaphore(threaded_count)
-        step = filesize // threaded_count
-        mtd_list = []
-        start = 0
-        end = -1
-        with open('./bin/db/style.db', 'w') as initial_file:
-            initial_file.close()
-        with open('./bin/db/style.db', 'rb+') as f:
-            name = 1
-            fileno = f.fileno()
-            while end < filesize - 1:
-                start = end + 1
-                end = start + step - 1
-                if end > filesize:
-                    end = filesize
-                dup = os.dup(fileno)
-                fd = os.fdopen(dup, 'rb+', -1)
-                t = MulThreadDownload(self.assets_url, start, end, fd, name)
-                name += 1
-                t.start()
-                mtd_list.append(t)
-
-            for i in mtd_list:
-                i.join()
-
-        self.check_md5()
-
-    def single_download(self):
-        """
-        单线程进行下载
-        :return:
-        """
-        session = requests.Session()
-        session.trust_env = False
-        logger.info(f'正在使用单线程下载中')
-        resp = session.get(self.assets_url)
-        with open('./bin/db/style.db', 'wb') as fp:
-            fp.write(resp.content)
-        logger.info('下载完成 正在检验文件md5')
-        with open('./bin/db/style.db', 'rb') as fp:
-            data = fp.read()
-        local_md5 = hashlib.md5(data).hexdigest()
-        remote_md5 = session.get('https://themedatabases.vercel.app/md5').json()['data'][0]
-        logger.info(f'本地数据库md5: {local_md5}')
-        logger.info(f'远程数据库md5: {remote_md5}')
-        if local_md5 != remote_md5:
-            logger.error('md5检验未通过请手动前往 https://themedatabases.vercel.app/assets 下载文件并放入./bin/assets文件夹内')
-            sys.exit(-1)
-        else:
-            logger.info('md5检验已通过')
-
-
-if __name__ != '__main__':
-    __all__ = ['error', 'logger', 'view']
-    if not os.path.exists('./bin/db/style.db'):
-        logger.error('没有检测到本地主题数据库即将开始下载')
-        Check().download()
+__all__ = ['error', 'logger', 'packing_logs', 'view']

bin/utils/error.py

@@ -9,7 +9,7 @@
 from typing import Optional
 from flask import jsonify
 from flask import Response
-from bin.db.sqlite import (update_data, fetch_table)
+from bin.db.sqlite import Database
 
 
 class ErrorProcess:
@@ -24,7 +24,7 @@ class ErrorProcess:
         直接获取可选主题
         :return:
         """
-        table_list = fetch_table()
+        table_list = Database().fetch_table()
         self.msg_template['code'] = 200
         self.msg_template['msg'] = '当前已保存到数据库的主题如下'
         self.msg_template['data'] = table_list
@@ -36,7 +36,7 @@ class ErrorProcess:
         :param theme:
         :return:
         """
-        table_list = fetch_table()
+        table_list = Database().fetch_table()
         self.msg_template['code'] = -2
         self.msg_template['msg'] = f'错误的主题: {theme}. 以下是已保存的主题'
         self.msg_template['data'] = table_list
@@ -59,7 +59,7 @@ class ErrorProcess:
         :param name:
         :return:
         """
-        update_data(name, 0)
+        Database().update_data(name, 0)
         self.msg_template['code'] = 200
         self.msg_template['msg'] = '当前长度已超过最大可计数范围. 已将此名称的计数器重置为零'
         self.msg_template['data'] = None

bin/utils/logger.py

@@ -13,7 +13,7 @@ from colorlog import ColoredFormatter
 
 
 class MakeLogger:
-    def __init__(self):
+    def __init__(self) -> None:
         self.date_format = '%H:%M:%S'
         self.format_console = '%(log_color)s[%(asctime)s] |%(filename)s[%(lineno)-3s] |%(levelname)-8s |%(message)s'
         self.format_file = '[%(asctime)s] |%(filename)s[%(funcName)sline:%(lineno)d] |%(levelname)-8s |%(message)s'

bin/utils/packing_logs.py

@@ -3,7 +3,7 @@
 # @Author: markushammered@gmail.com
 # @Development Tool: PyCharm
 # @Create Time: 2022/2/18
-# @File Name: packinglog.py
+# @File Name: packing_logs.py
 
 
 import os

bin/utils/settings.py

@@ -4,3 +4,55 @@
 # @Development Tool: PyCharm
 # @Create Time: 2022/2/23
 # @File Name: settings.py
+
+
+import yaml
+
+
+class Settings:
+    def __init__(self):
+        self.__setting_file = './bin/conf/config.yml'
+        with open(self.__setting_file, encoding='utf-8') as fp:
+            self.data = yaml.load(fp.read(), Loader=yaml.Loader)
+        self.__servers = self.data['servers']
+        self.__database = self.data['database']
+
+    @property
+    def host(self) -> str:
+        """
+        应用地址
+        :return:
+        """
+        return self.__servers['host']
+
+    @property
+    def port(self) -> str:
+        """
+        应用端口
+        :return:
+        """
+        return self.__servers['port']
+
+    @property
+    def type(self) -> str:
+        """
+        数据库类型
+        :return:
+        """
+        return self.__database['type']
+
+    @property
+    def path(self) -> str:
+        """
+        数据库路径
+        :return:
+        """
+        return self.__database['sqlite']['path']
+
+    @property
+    def name(self) -> str:
+        """
+        数据库名称
+        :return:
+        """
+        return self.__database['sqlite']['path'].split('/')[-1]

bin/utils/view.py

@@ -7,8 +7,7 @@
 
 
 from flask import render_template
-from bin.db.sqlite import fetch_style_data
-from bin.db.sqlite import fetch_table
+from bin.db.sqlite import Database
 
 
 def view_template(style: str, length: int, name: str, count: str) -> str or tuple[bool, str]:
@@ -20,9 +19,9 @@ def view_template(style: str, length: int, name: str, count: str) -> str or tupl
     :param style:
     :return:
     """
-    if style not in fetch_table():
+    if style not in Database().fetch_table():
         return [False, 'BadLength']
-    origin_data = fetch_style_data(style)  # 获取数据库内的文件
+    origin_data = Database().fetch_style_data(style)  # 获取数据库内的文件
     context = []
     if count != '0000000000':
         for i in count:  # 通过elif语句依次判断数字