"""授权后的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()