Merge pull request #24 from HisAtri/devc

更新子模块功能、使用国内CDN

app.py

@@ -15,6 +15,7 @@
 
 from mod import search, lrc
 from mod import tag
+from mod import run_process
 from mod.auth import webui, cookie
 from mod.auth.authentication import require_auth
 from mod.args import GlobalArgs
@@ -373,6 +374,7 @@ def main():
     logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
     logger = logging.getLogger('')
     logger.info("正在启动服务器")
+    run_process.run()
     # 注册 Blueprint 到 Flask 应用
     app.register_blueprint(v1_bp)
     # 启动

devtools/__init__.py

@@ -40,4 +40,4 @@ def run(self, func, *args, **kwargs):
         end = time.time()
         all_time = end - start
         avg_time = all_time / (self.threads*self.rounds)
-        return all_time, avg_time
+        return all_time, avg_time

mod/run_process.py

@@ -0,0 +1,27 @@
+import logging
+import pkgutil
+import multiprocessing
+
+
+logger = logging.getLogger(__name__)
+
+def load(package_name):
+    package = __import__(package_name, fromlist=[''])
+    sub_modules = []
+
+    for loader, module_name, is_pkg in pkgutil.walk_packages(package.__path__):
+        if not is_pkg:
+            sub_module = __import__(f"{package_name}.{module_name}", fromlist=[''])
+            sub_modules.append(sub_module)
+
+    return sub_modules
+
+
+def run():
+    modules = load('process')
+    for module in modules:
+        if hasattr(module, 'main'):
+            p = multiprocessing.Process(target=module.main)
+            p.start()
+        else:
+            logger.warning(f"The module {module.__name__} does not have a main method.")

mod/search.py

@@ -1,5 +1,6 @@
 import base64
 import json
+
 import aiohttp
 import asyncio
 
@@ -49,9 +50,10 @@ async def kugou(title, artist, album):
                                         headers=headers,
                                         timeout=10) as response3:
                                     lyrics_data = await response3.json()
-                                    lyrics_encode = lyrics_data["content"]                      # 这里是Base64编码的数据
-                                    lrc_text = base64.b64decode(lyrics_encode).decode('utf-8')  # 这里解码
-                                    return lrc_text
+                                    if lyrics_data:
+                                        lyrics_encode = lyrics_data["content"]                      # 这里是Base64编码的数据
+                                        lrc_text = base64.b64decode(lyrics_encode).decode('utf-8')  # 这里解码
+                                        return lrc_text
     await asyncio.sleep(10)
     return None
 

mod/tools.py

@@ -0,0 +1,82 @@
+import hashlib
+import re
+
+
+def calculate_md5(string: str, base="hex"):
+    """
+    计算字符串的 MD5 哈希值
+
+    参数：
+    - string: 要计算哈希值的字符串
+    - base: 返回结果的表示形式，可选值为 "hex"（十六进制，默认）、"dec"（十进制）、"bin"（二进制）
+    返回：
+    - 根据指定 base 返回相应表示形式的 MD5 哈希值
+    """
+    md5_hash = hashlib.md5()
+    # 将字符串转换为字节流并进行 MD5 计算
+    md5_hash.update(string.encode('utf-8'))
+    # 根据 base 参数返回相应的结果
+    if base == "hex":
+        # 十六进制->str
+        md5_hex = md5_hash.hexdigest()
+        md5_hex = md5_hex.lstrip("0x")
+        return md5_hex
+    elif base == "dec":
+        # 十进制表示->int
+        md5_dec = int(md5_hash.hexdigest(), 16)  # 将十六进制转换为十进制
+        return md5_dec
+    elif base == "bin":
+        # 二进制表示->bin
+        md5_bin = format(int(md5_hash.hexdigest(), 16), '0128b')  # 将十六进制转换为二进制，补齐到128位
+        return md5_bin
+    else:
+        raise ValueError("Invalid base. Supported values are 'hex', 'dec', and 'bin'.")
+
+
+def merge_dictionaries(dict_a: dict, dict_b: dict) -> dict:
+    """
+    合并两字典中的有效数据，前者优先
+    :param dict_a:
+    :param dict_b:
+    :return:
+    """
+    merged_dict = {}
+    if type(dict_a) is not dict:
+        return dict_b
+    # 遍历A和B的所有键
+    for key in set(dict_a.keys()) | set(dict_b.keys()):
+        # 判断A和B中对应键的值
+        value_a = dict_a.get(key)
+        value_b = dict_b.get(key)
+        # 如果A和B中都有有效数据，则优先取A的值
+        if value_a and value_b:
+            merged_dict[key] = value_a
+        # 如果A的值无效，则取B的值
+        elif not value_a:
+            merged_dict[key] = value_b
+        # 如果B的值无效，则取A的值
+        elif not value_b:
+            merged_dict[key] = value_a
+        else:
+            merged_dict[key] = value_a
+    return merged_dict
+
+
+def standard_lrc(lrc_text: str):
+    if not lrc_text or type(lrc_text) is not str:
+        return lrc_text
+    elif '[' in lrc_text and ']' in lrc_text:
+        lrc_text = lrc_text.replace("\r\n", "\n")
+        pattern = re.compile(r'\[([^]]+)]')
+        # 使用findall方法找到所有匹配的字符串
+        matches = pattern.findall(lrc_text)
+        for match_s in matches:
+            replacement = '[' + ']['.join(match_s.split(',')) + ']'
+            lrc_text = lrc_text.replace(f'[{match_s}]', replacement)
+
+        # 匹配时间戳
+        pattern = r"\[(\d{2}:\d{2}\.\d{2})\]"
+        # 进行匹配和替换
+        return re.sub(pattern, lambda match: "[" + match.group(1) + "0]", lrc_text)
+    else:
+        return lrc_text.replace("\r\n", "\n")

process/README.MD

@@ -0,0 +1,3 @@
+# 进程库
+
+此Package下的所有模块将被作为子进程运行，须定义入口函数`main()`。

process/example.py

@@ -0,0 +1,2 @@
+def main():
+    print("Hello, World!")

src/index.html

@@ -7,8 +7,8 @@
     <meta charset="UTF-8">
     <title>LyricsAPI</title>
     <link rel="stylesheet" href="/src/css/mod.css">
-    <link rel="stylesheet" href="//cdn.jsdelivr.net/npm/bootstrap@4.6.2/dist/css/bootstrap.css">
-    <link rel="stylesheet" href="//cdn.jsdelivr.net/npm/bootstrap-icons@1.10.0/font/bootstrap-icons.css">
+    <link rel="stylesheet" href="//cdn.bootcdn.net/ajax/libs/twitter-bootstrap/4.6.2/css/bootstrap.css">
+    <link rel="stylesheet" href="//cdn.bootcdn.net/ajax/libs/bootstrap-icons/1.10.0/font/bootstrap-icons.css">
 </head>
 
 <body>