(info: dict, dst_dir: str, used_packages: list[str])
| 59 | |
| 60 | |
| 61 | def write_index_cache(info: dict, dst_dir: str, used_packages: list[str]): |
| 62 | cache_dir = join(dst_dir, "cache") |
| 63 | |
| 64 | if not isdir(cache_dir): |
| 65 | os.makedirs(cache_dir) |
| 66 | |
| 67 | _platforms = info["_platform"], "noarch" |
| 68 | _remap_configs = list(info.get("channels_remap", [])) |
| 69 | _env_channels = [] |
| 70 | for env_info in info.get("extra_envs", {}).values(): |
| 71 | _remap_configs += env_info.get("channels_remap", []) |
| 72 | _env_channels += env_info.get("channels", []) |
| 73 | |
| 74 | _remaps = {url["src"].rstrip("/"): url["dest"].rstrip("/") for url in _remap_configs} |
| 75 | _channels = [ |
| 76 | url.rstrip("/") |
| 77 | for url in list(_remaps) |
| 78 | + info.get("channels", []) |
| 79 | + info.get("conda_default_channels", []) |
| 80 | + _env_channels |
| 81 | ] |
| 82 | _urls = all_channel_urls(_channels, subdirs=_platforms) |
| 83 | repodatas = {url: get_repodata(url) for url in _urls if url is not None} |
| 84 | |
| 85 | all_urls = info["_urls"].copy() |
| 86 | for env_info in info.get("_extra_envs_info", {}).values(): |
| 87 | all_urls += env_info["_urls"] |
| 88 | |
| 89 | for url, _ in all_urls: |
| 90 | src, subdir, fn = url.rsplit("/", 2) |
| 91 | dst = _remaps.get(src) |
| 92 | if dst is not None: |
| 93 | src = "%s/%s" % (src, subdir) |
| 94 | dst = "%s/%s" % (dst, subdir) |
| 95 | if dst not in repodatas: |
| 96 | repodatas[dst] = { |
| 97 | "_url": dst, |
| 98 | "info": {"subdir": subdir}, |
| 99 | "packages": {}, |
| 100 | "packages.conda": {}, |
| 101 | "removed": [], |
| 102 | } |
| 103 | loc = "packages.conda" if fn.endswith(".conda") else "packages" |
| 104 | repodatas[dst][loc][fn] = repodatas[src][loc][fn] |
| 105 | for src in _remaps: |
| 106 | for subdir in _platforms: |
| 107 | del repodatas["%s/%s" % (src, subdir)] |
| 108 | |
| 109 | for url, repodata in repodatas.items(): |
| 110 | if repodata is not None: |
| 111 | write_repodata(cache_dir, url, repodata, used_packages, info) |
| 112 | |
| 113 | for cache_file in os.listdir(cache_dir): |
| 114 | if not cache_file.endswith(".json"): |
| 115 | os.unlink(join(cache_dir, cache_file)) |
| 116 | |
| 117 | |
| 118 | def system_info(): |
no test coverage detected