Upload 2 files

github_zip_service.py

@@ -1,155 +1,163 @@
-"""授权后的GitHub ZIP交付服务；认证/实时所有权仍由路由负责。"""
-import asyncio
-import os
-from github_archive import cached_archive
-from github_source import github_item_source
-from skill_archive_source import source_header, installation_revision, source_key
-from proxy_resources import ProxyResources
-from workflow_stream_response import OwnedProxyResponse
-from zip_stream import zip_response_length, stream_zip_bytes
-from owned_thread import run_owned_thread
-
-
-async def prepare_github_zip(item, item_id, *, store, token_selector, resolve_commit,
-                             send, client_factory, transport_factory, error_response,
-                             cache_max_bytes, direct_max_bytes, signing_key=None):
-    # 按服务端资源解析仓库，普通作者只使用本人绑定凭据或匿名GitHub。
-    try:
-        source = github_item_source(item)
-        repo_url = item.get("link", "")
-        owner, repo = source["owner"], source["repo"]
-        active_token = await run_owned_thread(token_selector, item)
-        skill_revision = installation_revision(item, source_key(source), active_token, signing_key=signing_key) if source['skill'] else None
-    except ValueError:
-        return error_response(content={"error": "资源仓库或凭据暂时无法核实"}, status_code=503)
-
-    # ===== ZIP 缓存优先机制（鉴权后、GitHub请求前） =====
-    cache_resources = None
-    cache_handed_off = False
-    try:
-        versions_db = await run_owned_thread(store.load_data, "versions.json", default_data={})
-        item_version = versions_db.get(item_id)
-        archive = cached_archive(item, active_token, item_version, repo_id=store.DATASET_REPO_ID)
-        hf_cache_token = os.environ.get("HF_TOKEN")
-        if archive and hf_cache_token:
-            hf_headers = {"Authorization": f"Bearer {hf_cache_token}"}
-            cache_filename = archive["path"]
-            cache_url = f"https://huggingface.co/datasets/{archive['repo_id']}/resolve/{archive['hf_commit']}/{cache_filename}"
-            cache_client = client_factory(
-                follow_redirects=False, trust_env=False, transport=transport_factory(),
-            )
-            cache_resources = ProxyResources(cache_client)
-            head_resp = cache_resources.track(await send(
-                cache_client, "HEAD", cache_url, headers=hf_headers, timeout=10))
-            if head_resp.status_code == 200:
-                await cache_resources.close_response(head_resp)
-                stream_resp = cache_resources.track(await send(
-                    cache_client, "GET", cache_url, headers=hf_headers, timeout=120))
-                if stream_resp.status_code == 200:
-                    # identity响应应与已上传归档大小一致；编码响应仍保持原始透传。
-                    encoding = (stream_resp.headers.get('content-encoding') or 'identity').strip().lower()
-                    expected_size = archive['size'] if encoding == 'identity' else None
-                    cache_length = zip_response_length(stream_resp.headers, max_bytes=cache_max_bytes,
-                                                       expected_size=expected_size)
-                    cache_headers = {}
-                    if source["skill"]:
-                        cache_headers["X-Ranking-Skill-Source"] = source_header(source, ())
-                        cache_headers['X-Ranking-Skill-Revision'] = skill_revision
-                    content_length = stream_resp.headers.get('content-length')
-                    if content_length:
-                        cache_headers["Content-Length"] = content_length
-                    content_encoding = stream_resp.headers.get('content-encoding')
-                    if content_encoding:
-                        cache_headers["Content-Encoding"] = content_encoding
-                    if item_version.get('hash'):
-                        cache_headers["X-Installed-Version"] = archive["commit"]
-
-                    async def cache_stream_generator():
-                        try:
-                            async for chunk in stream_zip_bytes(stream_resp, max_bytes=cache_max_bytes,
-                                                                length=cache_length, expected_size=expected_size):
-                                yield chunk
-                        except Exception as error:
-                            print(f"[ZIP缓存] 响应流中断: error={type(error).__name__}")
-                            raise
-                        finally:
-                            await cache_resources.close()
-
-                    prepared = OwnedProxyResponse(cache_stream_generator(), close_callback=cache_resources.close,
-                                                  media_type="application/zip", headers=cache_headers)
-                    cache_handed_off = True
-                    return prepared
-                print(f"[ZIP缓存] GET失败，回退GitHub: status={stream_resp.status_code}")
-    except asyncio.CancelledError:
-        raise
-    except Exception as error:
-        print(f"[ZIP缓存] 查询失败，回退GitHub: error={type(error).__name__}")
-    finally:
-        if cache_resources is not None and not cache_handed_off:
-            await cache_resources.close()
-
-    # 先取得完整commit，再按它下载；不在下载后另取“最新版本”冒充实际内容。
-    live_commit = await resolve_commit(repo_url, active_token)
-    if not live_commit:
-        return error_response(content={"error": "资源版本暂时无法核实，请稍后重试"}, status_code=502)
-    github_zip_api = f"https://api.github.com/repos/{owner}/{repo}/zipball/{live_commit}"
-
-    headers = {
-        "Accept": "application/vnd.github.v3+json",
-        "User-Agent": "ComfyUI-Ranking-SaaS"
-    }
-    if active_token:
-        headers["Authorization"] = f"Bearer {active_token}"
-
-    # 异步请求固定GitHub commit，原始块保持长度与Content-Encoding契约。
-    client = client_factory(
-        follow_redirects=False, trust_env=False, transport=transport_factory(),
-    )
-    resources = ProxyResources(client)
-    handed_off = False
-    try:
-        try:
-            response = resources.track(await send(
-                client, "GET", github_zip_api, headers=headers, timeout=120))
-        except Exception as error:
-            print(f"[proxy_github_zip] 请求失败: error={type(error).__name__}")
-            return error_response(content={"error": "代理下载时发生网络异常，请稍后重试"}, status_code=502)
-        if response.status_code != 200:
-            status_map = {404: "仓库不存在或无权访问", 401: "Token 无效或已过期", 403: "访问被拒绝（可能触发速率限制）"}
-            err_msg = status_map.get(response.status_code, f"GitHub API 返回 HTTP {response.status_code}")
-            return error_response(content={"error": f"GITHUB_DOWNLOAD_FAILED: {err_msg}"}, status_code=502)
-        declared_length = zip_response_length(response.headers, max_bytes=direct_max_bytes)
-        resp_headers = {}
-        if source["skill"]:
-            resp_headers["X-Ranking-Skill-Source"] = source_header(source, ())
-            resp_headers['X-Ranking-Skill-Revision'] = skill_revision
-        content_length = response.headers.get('content-length', '')
-        if content_length:
-            resp_headers["Content-Length"] = content_length
-        content_encoding = response.headers.get('content-encoding')
-        if content_encoding:
-            resp_headers["Content-Encoding"] = content_encoding
-        resp_headers["X-Installed-Version"] = live_commit
-
-        async def stream_generator():
-            try:
-                async for chunk in stream_zip_bytes(response, max_bytes=direct_max_bytes, length=declared_length):
-                    yield chunk
-            except Exception as error:
-                print(f"[proxy_github_zip] 响应流中断: error={type(error).__name__}")
-                raise
-            finally:
-                await resources.close()
-
-        prepared = OwnedProxyResponse(stream_generator(), close_callback=resources.close,
-                                      media_type="application/zip", headers=resp_headers)
-        handed_off = True
-        return prepared
-    except asyncio.CancelledError:
-        raise
-    except Exception as error:
-        print(f"[proxy_github_zip] 响应准备失败: error={type(error).__name__}")
-        return error_response(content={"error": "代理下载时发生网络异常，请稍后重试"}, status_code=502)
-    finally:
-        if not handed_off:
-            await resources.close()
+"""授权后的GitHub ZIP交付服务；认证/实时所有权仍由路由负责。"""
+import asyncio
+import os
+from github_archive import cached_archive
+from github_source import github_item_source, repository_parts
+from skill_archive_source import source_header, installation_revision, source_key
+from proxy_resources import ProxyResources
+from workflow_stream_response import OwnedProxyResponse
+from zip_stream import zip_response_length, stream_zip_bytes
+from owned_thread import run_owned_thread
+
+
+async def prepare_github_zip(item, item_id, *, expected_repository, store, token_selector, resolve_commit,
+                             send, client_factory, transport_factory, error_response,
+                             cache_max_bytes, direct_max_bytes, signing_key=None):
+    # 按服务端资源解析仓库，普通作者只使用本人绑定凭据或匿名GitHub。
+    try:
+        source = github_item_source(item)
+        repo_url = item.get("link", "")
+        owner, repo = source["owner"], source["repo"]
+        try:
+            expected_owner, expected_repo = repository_parts(expected_repository)
+        except ValueError:
+            expected_owner = expected_repo = None
+        if (expected_owner is None or expected_owner.casefold() != owner.casefold()
+                or expected_repo.casefold() != repo.casefold()):
+            return error_response(content={"error": "资源来源已变更，请刷新资源后重新安装。",
+                "error_code": "source_changed"}, status_code=409)
+        active_token = await run_owned_thread(token_selector, item)
+        skill_revision = installation_revision(item, source_key(source), active_token, signing_key=signing_key) if source['skill'] else None
+    except ValueError:
+        return error_response(content={"error": "资源仓库或凭据暂时无法核实"}, status_code=503)
+
+    # ===== ZIP 缓存优先机制（鉴权后、GitHub请求前） =====
+    cache_resources = None
+    cache_handed_off = False
+    try:
+        versions_db = await run_owned_thread(store.load_data, "versions.json", default_data={})
+        item_version = versions_db.get(item_id)
+        archive = cached_archive(item, active_token, item_version, repo_id=store.DATASET_REPO_ID)
+        hf_cache_token = os.environ.get("HF_TOKEN")
+        if archive and hf_cache_token:
+            hf_headers = {"Authorization": f"Bearer {hf_cache_token}"}
+            cache_filename = archive["path"]
+            cache_url = f"https://huggingface.co/datasets/{archive['repo_id']}/resolve/{archive['hf_commit']}/{cache_filename}"
+            cache_client = client_factory(
+                follow_redirects=False, trust_env=False, transport=transport_factory(),
+            )
+            cache_resources = ProxyResources(cache_client)
+            head_resp = cache_resources.track(await send(
+                cache_client, "HEAD", cache_url, headers=hf_headers, timeout=10))
+            if head_resp.status_code == 200:
+                await cache_resources.close_response(head_resp)
+                stream_resp = cache_resources.track(await send(
+                    cache_client, "GET", cache_url, headers=hf_headers, timeout=120))
+                if stream_resp.status_code == 200:
+                    # identity响应应与已上传归档大小一致；编码响应仍保持原始透传。
+                    encoding = (stream_resp.headers.get('content-encoding') or 'identity').strip().lower()
+                    expected_size = archive['size'] if encoding == 'identity' else None
+                    cache_length = zip_response_length(stream_resp.headers, max_bytes=cache_max_bytes,
+                                                       expected_size=expected_size)
+                    cache_headers = {}
+                    if source["skill"]:
+                        cache_headers["X-Ranking-Skill-Source"] = source_header(source, ())
+                        cache_headers['X-Ranking-Skill-Revision'] = skill_revision
+                    content_length = stream_resp.headers.get('content-length')
+                    if content_length:
+                        cache_headers["Content-Length"] = content_length
+                    content_encoding = stream_resp.headers.get('content-encoding')
+                    if content_encoding:
+                        cache_headers["Content-Encoding"] = content_encoding
+                    if item_version.get('hash'):
+                        cache_headers["X-Installed-Version"] = archive["commit"]
+
+                    async def cache_stream_generator():
+                        try:
+                            async for chunk in stream_zip_bytes(stream_resp, max_bytes=cache_max_bytes,
+                                                                length=cache_length, expected_size=expected_size):
+                                yield chunk
+                        except Exception as error:
+                            print(f"[ZIP缓存] 响应流中断: error={type(error).__name__}")
+                            raise
+                        finally:
+                            await cache_resources.close()
+
+                    prepared = OwnedProxyResponse(cache_stream_generator(), close_callback=cache_resources.close,
+                                                  media_type="application/zip", headers=cache_headers)
+                    cache_handed_off = True
+                    return prepared
+                print(f"[ZIP缓存] GET失败，回退GitHub: status={stream_resp.status_code}")
+    except asyncio.CancelledError:
+        raise
+    except Exception as error:
+        print(f"[ZIP缓存] 查询失败，回退GitHub: error={type(error).__name__}")
+    finally:
+        if cache_resources is not None and not cache_handed_off:
+            await cache_resources.close()
+
+    # 先取得完整commit，再按它下载；不在下载后另取“最新版本”冒充实际内容。
+    live_commit = await resolve_commit(repo_url, active_token)
+    if not live_commit:
+        return error_response(content={"error": "资源版本暂时无法核实，请稍后重试"}, status_code=502)
+    github_zip_api = f"https://api.github.com/repos/{owner}/{repo}/zipball/{live_commit}"
+
+    headers = {
+        "Accept": "application/vnd.github.v3+json",
+        "User-Agent": "ComfyUI-Ranking-SaaS"
+    }
+    if active_token:
+        headers["Authorization"] = f"Bearer {active_token}"
+
+    # 异步请求固定GitHub commit，原始块保持长度与Content-Encoding契约。
+    client = client_factory(
+        follow_redirects=False, trust_env=False, transport=transport_factory(),
+    )
+    resources = ProxyResources(client)
+    handed_off = False
+    try:
+        try:
+            response = resources.track(await send(
+                client, "GET", github_zip_api, headers=headers, timeout=120))
+        except Exception as error:
+            print(f"[proxy_github_zip] 请求失败: error={type(error).__name__}")
+            return error_response(content={"error": "代理下载时发生网络异常，请稍后重试"}, status_code=502)
+        if response.status_code != 200:
+            status_map = {404: "仓库不存在或无权访问", 401: "Token 无效或已过期", 403: "访问被拒绝（可能触发速率限制）"}
+            err_msg = status_map.get(response.status_code, f"GitHub API 返回 HTTP {response.status_code}")
+            return error_response(content={"error": f"GITHUB_DOWNLOAD_FAILED: {err_msg}"}, status_code=502)
+        declared_length = zip_response_length(response.headers, max_bytes=direct_max_bytes)
+        resp_headers = {}
+        if source["skill"]:
+            resp_headers["X-Ranking-Skill-Source"] = source_header(source, ())
+            resp_headers['X-Ranking-Skill-Revision'] = skill_revision
+        content_length = response.headers.get('content-length', '')
+        if content_length:
+            resp_headers["Content-Length"] = content_length
+        content_encoding = response.headers.get('content-encoding')
+        if content_encoding:
+            resp_headers["Content-Encoding"] = content_encoding
+        resp_headers["X-Installed-Version"] = live_commit
+
+        async def stream_generator():
+            try:
+                async for chunk in stream_zip_bytes(response, max_bytes=direct_max_bytes, length=declared_length):
+                    yield chunk
+            except Exception as error:
+                print(f"[proxy_github_zip] 响应流中断: error={type(error).__name__}")
+                raise
+            finally:
+                await resources.close()
+
+        prepared = OwnedProxyResponse(stream_generator(), close_callback=resources.close,
+                                      media_type="application/zip", headers=resp_headers)
+        handed_off = True
+        return prepared
+    except asyncio.CancelledError:
+        raise
+    except Exception as error:
+        print(f"[proxy_github_zip] 响应准备失败: error={type(error).__name__}")
+        return error_response(content={"error": "代理下载时发生网络异常，请稍后重试"}, status_code=502)
+    finally:
+        if not handed_off:
+            await resources.close()

router_proxy.py

@@ -146,7 +146,7 @@ async def proxy_github_zip(request: Request, req_data: ProxyGithubZipRequest, ac
         from 云端_定时版本检测引擎 import fetch_github_commit
         return await fetch_github_commit(repo_url, token)
 
-    return await prepare_github_zip(item, req_data.item_id, store=json_db,
+    return await prepare_github_zip(item, req_data.item_id, expected_repository=req_data.url, store=json_db,
         signing_key=JWT_SECRET,
         token_selector=select_github_token, resolve_commit=resolve_commit,
         send=_send_with_validated_redirects, client_factory=httpx.AsyncClient,