fix: get url auth

Author: YangSen-qn

src/mcp_server/core/media_processing/__init__.py

@@ -5,7 +5,7 @@
 
 def load(cfg: config.Config):
     cli = processing.MediaProcessingService(cfg)
-    register_tools(cli)
+    register_tools(cfg, cli)
 
 
 __all__ = [

src/mcp_server/core/media_processing/tools.py

@@ -1,8 +1,11 @@
 import logging
+
+import qiniu
 from mcp import types
 
 from . import utils
 from .processing import MediaProcessingService
+from ...config import config
 from ...consts import consts
 from ...tools import tools
 
@@ -12,7 +15,8 @@
 
 
 class _ToolImpl:
-    def __init__(self, cli: MediaProcessingService):
+    def __init__(self, cfg: config.Config, cli: MediaProcessingService):
+        self.auth = qiniu.Auth(cfg.access_key, cfg.secret_key)
         self.client = cli
 
     @tools.tool_meta(
@@ -55,8 +59,8 @@ def image_scale_by_percent(
                 types.TextContent(type="text", text="percent must be between 1 and 999")
             ]
 
-        fop = f"imageMogr2/thumbnail/!{percent}p"
-        object_url = utils.url_add_processing_func(object_url, fop)
+        func = f"imageMogr2/thumbnail/!{percent}p"
+        object_url = utils.url_add_processing_func(auth=self.auth, url=object_url, func=func)
         return [
             types.TextContent(
                 type="text",
@@ -111,16 +115,16 @@ def image_scale_by_size(
         if object_url is None or len(object_url) == 0:
             return [types.TextContent(type="text", text="object_url is required")]
 
-        fop = f"{width}x{height}"
-        if len(fop) == 1:
+        func = f"{width}x{height}"
+        if len(func) == 1:
             return [
                 types.TextContent(
                     type="text", text="At least one width or height must be set"
                 )
             ]
 
-        fop = f"imageMogr2/thumbnail/{fop}"
-        object_url = utils.url_add_processing_func(object_url, fop)
+        func = f"imageMogr2/thumbnail/{func}"
+        object_url = utils.url_add_processing_func(auth=self.auth, url=object_url, func=func)
         return [
             types.TextContent(
                 type="text",
@@ -191,7 +195,7 @@ def image_round_corner(self, **kwargs) -> list[types.TextContent]:
             radius_y = radius_x
 
         func = f"roundPic/radiusx/{radius_x}/radiusy/{radius_y}"
-        object_url = utils.url_add_processing_func(object_url, func)
+        object_url = utils.url_add_processing_func(auth=self.auth, url=object_url, func=func)
         return [
             types.TextContent(
                 type="text",
@@ -228,7 +232,7 @@ def image_info(self, **kwargs) -> list[types.TextContent]:
             ]
 
         func = "imageInfo"
-        object_url = utils.url_add_processing_func(object_url, func)
+        object_url = utils.url_add_processing_func(auth=self.auth, url=object_url, func=func)
         return [
             types.TextContent(
                 type="text",
@@ -259,8 +263,8 @@ def get_fop_status(self, **kwargs) -> list[types.TextContent]:
         return [types.TextContent(type="text", text=str(status))]
 
 
-def register_tools(cli: MediaProcessingService):
-    tool_impl = _ToolImpl(cli)
+def register_tools(cfg: config.Config, cli: MediaProcessingService):
+    tool_impl = _ToolImpl(cfg, cli)
     tools.auto_register_tools(
         [
             tool_impl.image_scale_by_percent,

src/mcp_server/core/media_processing/utils.py

@@ -1,11 +1,16 @@
+import time
 from urllib import parse
 
+import qiniu
+
+from mcp_server.core.storage.tools import logger
+
 FUNC_POSITION_NONE = "none"
 FUNC_POSITION_PREFIX = "prefix"
 FUNC_POSITION_SUFFIX = "suffix"
 
 
-def url_add_processing_func(url: str, func: str) -> str:
+def url_add_processing_func(auth: qiniu.auth, url: str, func: str) -> str:
     func_items = func.split("/")
     func_prefix = func_items[0]
 
@@ -14,7 +19,8 @@ def url_add_processing_func(url: str, func: str) -> str:
     new_query = parse.quote(new_query, safe='&=')
     url_info = url_info._replace(query=new_query)
     new_url = parse.urlunparse(url_info)
-    return str(new_url)
+    new_url = _sign_url(str(new_url), auth)
+    return new_url
 
 
 def _query_add_processing_func(query: str, func: str, func_prefix: str) -> str:
@@ -70,3 +76,33 @@ def _query_add_processing_func(query: str, func: str, func_prefix: str) -> str:
     func = first_query + func.removeprefix(func_prefix)
     queries.insert(0, func)
     return "&".join(queries)
+
+
+def _sign_url(url: str, auth: qiniu.auth) -> str:
+    url_info = parse.urlparse(url)
+    query = url_info.query
+    if (not 'e=' in query) or ('token' not in query):
+        return url
+
+    queries = query.split("&")
+    if '' in queries:
+        queries.remove('')
+
+    # 移除之前的签名信息，但顺序不可变
+    expires = 3600
+    new_queries = []
+    for queryItem in queries:
+        if queryItem.startswith('e='):
+            try:
+                deadline = int(queryItem.removeprefix('e='))
+                expires = deadline - int(time.time())
+            except Exception as e:
+                logger.warning(f"expires parse fail for url:{url}")
+                expires = 3600
+        elif not queryItem.startswith('token='):
+            new_queries.append(queryItem)
+
+    new_query = "&".join(new_queries)
+    url_info = url_info._replace(query=new_query)
+    new_url = parse.urlunparse(url_info)
+    return auth.private_download_url(new_url, expires=expires)