fix: debug and improve backup

fix: fix an authentication issue
fix: fix path combination issue

Author: tokenicrat

src/pelit/plib/route_tool.py

@@ -22,13 +22,13 @@ def authenticate(cfg: dict[str, Any]) -> bool:
         token = token[7:]
 
     # 验证密钥，可用 PELIT_AUTH 环境变量或配置文件
-    if 'hashed' in cfg['auth'] and not 'from_env' in cfg['auth']:
+    if 'PELIT_AUTH' in os.environ and 'from_env' in cfg['auth'] and cfg['auth']['from_env']:
+        token_set: str = os.environ['PELIT_AUTH']
+        return token_set.upper() == token.upper()
+    elif 'hashed' in cfg['auth']:
         token_set: str = cfg['auth']['hashed']
         return token_set.upper() == \
             hashlib.sha256(token.encode('utf-8')).hexdigest().upper()
-    elif 'PELIT_AUTH' in os.environ and 'from_env' in cfg['auth']:
-        token_set: str = os.environ['PELIT_AUTH']
-        return token_set.upper() == token.upper()
     else:
         return False
 
@@ -109,8 +109,9 @@ def backup_to_file(path: Path, file: Path) -> None:
         path: 需要备份的目录
         file: 备份保存的文件
     """
+    print(path, file)
 
     try:
-        shutil.make_archive(str(path), 'gztar', str(file))
+        shutil.make_archive(str(file), 'gztar', str(path))
     except Exception:
         pass

src/pelit/route.py

@@ -174,7 +174,7 @@ def _delete(directory: str, file: str) -> tuple[Response, int]:
 
     @main_route.route('/list', methods=['GET'])
     @main_route.route('/list/<directory>', methods=['GET'])
-    def _list(directory: str = "/") -> tuple[Response, int]:
+    def _list(directory: str = "") -> tuple[Response, int]:
         """
         列出指定目录下所有目录
 
@@ -282,7 +282,7 @@ def _retrieve(directory: str, file: str) -> tuple[Response, int]:
 
     @main_route.route('/backup', methods=['GET'])
     @main_route.route('/backup/<directory>', methods=['GET'])
-    def _backup(directory: str = '/') -> tuple[Response, int]:
+    def _backup(directory: str = '') -> tuple[Response, int]:
         """
         创建一个备份任务，具体方式在配置文件中指定
 
@@ -291,18 +291,29 @@ def _backup(directory: str = '/') -> tuple[Response, int]:
         """
         info_head = f"{request.remote_addr} {request.method} {request.path}"
 
-        bak_name = generate_file_name(cfg['storage']['path'], '.tar.gz')
-        bak_path = Path(cfg['storage']['path']) / (bak_name + '.tar.gz')
-        bak_url = join_url(cfg['network']['base_url'], bak_name + '.tar.gz')
+        if not authenticate(cfg):
+            lg.warn(f"{info_head} 401: 认证失败")
+            return jsonify({
+                "success": False,
+                "message": "认证失败"
+            }), 401
+
+        bak_name = generate_file_name(Path(cfg['storage']['path']), '.tar.gz')
+        bak_path = Path(cfg['storage']['path']) / bak_name
+        bak_url = join_url(
+            cfg['network']['base_url'] if 'network' in cfg['network'] else '',
+            bak_name + '.tar.gz')
 
         path = Path(cfg['storage']['path']) / directory
 
         # 创建新进程压缩备份
         # FIXME: 由于是多进程的，这个任务失败了也不会有表示
+        # FIXME: 备份会包括所有之前所有的 .tar.gz
         p = Process(target=backup_to_file, args=(path, bak_path))
         p.start()
 
         lg.info(f"{info_head} 200 备份任务创建成功")
+        lg.info(f"位置：{bak_path}.tar.gz")
         return jsonify({
             "success": True,
             "url": bak_url,