#!/usr/bin/env python3
"""
Sync a curated subset of the Keiyoushi extension repo into this self-hosted
Mihon repo.
Emits all three index formats so every client works:
index.pb - Mihon 0.20.1+ (gzipped protobuf, the current format)
index.json - new-schema JSON (human readable / debugging)
index.min.json - legacy Tachiyomi schema (old forks: Aniyomi, TachiyomiSY, ...)
APKs / JARs / icons are mirrored locally and the URLs in the indexes are
rewritten to point at BASE_URL, so the repo is fully self-contained and keeps
working even if upstream release assets disappear.
The signing key is left as Keiyoushi's because the mirrored APKs are their
original, unmodified, signed artifacts.
Usage:
./sync.py # sync packages listed in packages.txt
./sync.py --dry-run # report what would change, download nothing
./sync.py --prune # also delete mirrored files no longer referenced
./sync.py --add en.weebcentral --add all.mangaplus
./sync.py --list-available mangadex # search upstream catalogue
"""
import argparse
import hashlib
import json
import os
import re
import shutil
import struct
import sys
sys.dont_write_bytecode = True # keep __pycache__ out of the public web root
import time
import zipfile
import urllib.error
import urllib.request
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import mihon_pb # noqa: E402
ROOT = os.path.dirname(os.path.abspath(__file__))
BASE_URL = "https://changed.dami3nz.au"
UPSTREAM = "https://raw.githubusercontent.com/keiyoushi/extensions/repo/index.json"
APK_DIR = os.path.join(ROOT, "apk")
JAR_DIR = os.path.join(ROOT, "jar")
ICON_DIR = os.path.join(ROOT, "icon")
PKG_FILE = os.path.join(ROOT, "packages.txt")
STORES_FILE = os.path.join(ROOT, "stores.txt")
PKG_PREFIX = "eu.kanade.tachiyomi.extension."
META = {
"name": "dami3nz",
"badgeLabel": "D3N",
# Keiyoushi's APK signing cert fingerprint - the APKs are theirs, unmodified.
"signingKey": "9add655a78e96c4ec7a53ef89dccb557cb5d767489fac5e785d671a5a75d4da2",
"contact": {"website": BASE_URL, "discord": ""},
}
def log(msg):
print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
def icon_fallbacks(url):
"""Upstream icons are jsdelivr URLs. jsdelivr 403s on some paths (seen on
all.onisaga) even though the file exists in extensions-source, so try the
equivalent raw.githubusercontent URL before giving up."""
urls = [url]
pre = "https://cdn.jsdelivr.net/gh/"
if url.startswith(pre):
rest = url[len(pre):] # owner/repo@ref/path
if "@" in rest and "/" in rest:
repo, _, tail = rest.partition("@")
ref, _, path = tail.partition("/")
urls.append(f"https://raw.githubusercontent.com/{repo}/{ref}/{path}")
return urls
def _png_size(data):
if data[:8] != b"\x89PNG\r\n\x1a\n":
return None
return struct.unpack(">II", data[16:24])
def extract_icon_from_apk(apk_path, dest):
"""Last-resort icon source: pull the launcher icon out of the APK itself.
Some small stores publish no icon/ directory at all (copymanga ships one
icon for four extensions). Resource names inside an APK are obfuscated, so
pick the largest square PNG in a plausible launcher-icon size - that is the
mipmap launcher in practice.
"""
try:
with zipfile.ZipFile(apk_path) as z:
best = None
for name in z.namelist():
if not (name.startswith("res/") and name.lower().endswith(".png")):
continue
data = z.read(name)
size = _png_size(data)
if not size:
continue
w, h = size
if w != h or not (32 <= w <= 512):
continue
if best is None or w > best[0]:
best = (w, data)
if best:
os.makedirs(os.path.dirname(dest), exist_ok=True)
with open(dest, "wb") as fh:
fh.write(best[1])
return True
except Exception:
pass
return False
def fetch(url, timeout=60):
req = urllib.request.Request(url, headers={"User-Agent": "mihon-repo-sync/1.0"})
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.read()
def download(url, dest, timeout=180):
"""Download to dest atomically. Returns True if a download happened."""
if os.path.exists(dest) and os.path.getsize(dest) > 0:
return False
os.makedirs(os.path.dirname(dest), exist_ok=True)
tmp = dest + ".part"
req = urllib.request.Request(url, headers={"User-Agent": "mihon-repo-sync/1.0"})
with urllib.request.urlopen(req, timeout=timeout) as r, open(tmp, "wb") as fh:
shutil.copyfileobj(r, fh)
if os.path.getsize(tmp) == 0:
os.remove(tmp)
raise IOError(f"empty download: {url}")
os.replace(tmp, dest)
return True
def read_excluded():
"""excluded.txt: full package names that must never be mirrored."""
path = os.path.join(ROOT, "excluded.txt")
if not os.path.exists(path):
return set()
return {ln.split("#", 1)[0].strip() for ln in open(path)} - {""}
def read_packages():
"""packages.txt holds short names (en.toonily) or full package names."""
if not os.path.exists(PKG_FILE):
return []
out = []
for line in open(PKG_FILE):
line = line.split("#", 1)[0].strip()
if not line:
continue
out.append(line if line.startswith(PKG_PREFIX) else PKG_PREFIX + line)
excluded = read_excluded()
return [p for p in out if p not in excluded]
def write_packages(pkgs):
header = (
"# One extension per line. Short form (en.toonily) or full package name.\n"
"# Blank lines and # comments ignored. Run ./sync.py after editing.\n\n"
)
body = "\n".join(sorted(p[len(PKG_PREFIX):] if p.startswith(PKG_PREFIX) else p
for p in set(pkgs)))
with open(PKG_FILE, "w") as fh:
fh.write(header + body + "\n")
def version_code(version):
"""Upstream encodes 1.6.4 as 106004 = major*100000 + minor*1000 + patch.
Legacy stores only carry a small 'code' int, so derive it from the string."""
parts = [int(x) for x in re.findall(r"\d+", version or "")][:3]
while len(parts) < 3:
parts.append(0)
return parts[0] * 100000 + parts[1] * 1000 + parts[2]
def from_legacy(e, base):
"""Convert a legacy index.min.json entry to the new schema.
Legacy stores resolve assets relative to the index URL:
/apk/ and /icon/.png
"""
ver = e.get("version", "0.0.0")
parts = [int(x) for x in re.findall(r"\d+", ver)][:2]
while len(parts) < 2:
parts.append(0)
return {
"name": re.sub(r"^(Tachiyomi|Aniyomi):\s*", "", e.get("name", "")),
"packageName": e["pkg"],
"resources": {
"apkUrl": f"{base}/apk/{e['apk']}",
"iconUrl": f"{base}/icon/{e['pkg']}.png",
},
"extensionLib": f"{parts[0]}.{parts[1]}",
"versionCode": str(version_code(ver)),
"versionName": ver,
"contentWarning": ("CONTENT_WARNING_NSFW" if e.get("nsfw")
else "CONTENT_WARNING_SAFE"),
# Empty values must be omitted, not emitted as "": the pb encoder skips
# empty strings, so a "homeUrl": "" here would fail the round-trip
# assert. Server connectors (Kavita, Suwayomi) have no baseUrl.
"sources": [
{k: v for k, v in (
("id", str(s.get("id", ""))),
("name", s.get("name", "")),
("language", s.get("lang", "")),
("homeUrl", s.get("baseUrl", "")),
) if v}
for s in e.get("sources", [])
],
}
def read_stores():
"""stores.txt: 'label | index-url', one per line. Extra Mihon-scheme stores
merged on top of Keiyoushi."""
if not os.path.exists(STORES_FILE):
return []
out = []
for line in open(STORES_FILE):
line = line.split("#", 1)[0].strip()
if not line:
continue
label, _, url = line.partition("|")
if url.strip():
out.append((label.strip(), url.strip()))
return out
def load_store(label, url):
"""-> list of new-schema extension dicts from any store format."""
body = fetch(url)
if body[:2] == b"\x1f\x8b":
import gzip
body = gzip.decompress(body)
data = json.loads(body)
if isinstance(data, dict) and "extensionList" in data:
return data["extensionList"]["extensions"]
if isinstance(data, list):
base = url.rsplit("/", 1)[0]
return [from_legacy(e, base) for e in data]
raise ValueError(f"{label}: unrecognised index format")
def merge_catalogue(primary, stores):
"""Merge extra stores over the primary catalogue, deduping by packageName.
Highest versionCode wins; ties go to whoever was there first (Keiyoushi).
Returns (bypkg, duplicate_report).
"""
bypkg = {e["packageName"]: e for e in primary}
dupes = []
for label, exts in stores:
for e in exts:
pkg = e["packageName"]
cur = bypkg.get(pkg)
if cur is None:
bypkg[pkg] = dict(e, _store=label)
continue
new_v, old_v = int(e.get("versionCode") or 0), int(cur.get("versionCode") or 0)
winner = label if new_v > old_v else (cur.get("_store") or "Keiyoushi")
dupes.append((pkg, cur.get("_store") or "Keiyoushi", old_v, label, new_v, winner))
if new_v > old_v:
bypkg[pkg] = dict(e, _store=label)
return bypkg, dupes
def to_legacy(e, apk_name):
"""Convert a new-schema extension dict to the old index.min.json schema."""
langs = {s.get("language") for s in e.get("sources", [])}
lang = langs.pop() if len(langs) == 1 else "all"
nsfw = 1 if e.get("contentWarning") in ("CONTENT_WARNING_NSFW",
"CONTENT_WARNING_MIXED") else 0
return {
"name": "Tachiyomi: " + e["name"],
"pkg": e["packageName"],
"apk": apk_name,
"lang": lang,
"code": int(e["versionCode"]),
"version": e["versionName"],
"nsfw": nsfw,
"sources": [
{
"name": s.get("name", ""),
"lang": s.get("language", ""),
"id": s.get("id", ""),
"baseUrl": s.get("homeUrl", ""),
}
for s in e.get("sources", [])
],
}
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--dry-run", action="store_true")
ap.add_argument("--prune", action="store_true",
help="delete mirrored apk/jar/icon files no longer referenced")
ap.add_argument("--add", action="append", default=[],
help="add a package to packages.txt before syncing")
ap.add_argument("--list-available", metavar="QUERY",
help="search the upstream catalogue and exit")
ap.add_argument("--no-jars", action="store_true",
help="don't mirror .jar files (leave jarUrl pointing upstream)")
args = ap.parse_args()
log(f"fetching upstream index ({UPSTREAM})")
up = json.loads(fetch(UPSTREAM))
catalogue = up["extensionList"]["extensions"]
log(f"Keiyoushi has {len(catalogue)} extensions")
extra = []
for label, url in read_stores():
try:
exts = load_store(label, url)
extra.append((label, exts))
log(f" + store {label}: {len(exts)} extensions")
except Exception as ex:
log(f" ! store {label} unreachable: {ex}")
bypkg, dupes = merge_catalogue(catalogue, extra)
if dupes:
log(f"deduped {len(dupes)} package(s) present in more than one store")
for pkg, s1, v1, s2, v2, win in dupes[:10]:
log(f" {pkg.replace(PKG_PREFIX, '')}: {s1} {v1} vs {s2} {v2} -> {win}")
log(f"merged catalogue: {len(bypkg)} unique packages")
catalogue = list(bypkg.values())
if args.list_available:
q = args.list_available.lower()
hits = [e for e in catalogue
if q in e["name"].lower() or q in e["packageName"].lower()]
for e in sorted(hits, key=lambda x: x["name"]):
cw = (e.get("contentWarning") or "").replace("CONTENT_WARNING_", "")
short = e["packageName"][len(PKG_PREFIX):]
print(f" {e['name']:<34} {e['versionName']:<9} {cw:<6} {short}")
print(f"\n{len(hits)} match(es)")
return 0
pkgs = read_packages()
if args.add:
extra = [p if p.startswith(PKG_PREFIX) else PKG_PREFIX + p for p in args.add]
unknown = [p for p in extra if p not in bypkg]
if unknown:
log("ERROR: not in upstream catalogue: " + ", ".join(unknown))
return 1
pkgs = sorted(set(pkgs) | set(extra))
if not args.dry_run:
write_packages(pkgs)
log(f"added {len(extra)} package(s) to packages.txt")
if not pkgs:
log("packages.txt is empty - nothing to sync")
return 1
missing = [p for p in pkgs if p not in bypkg]
for p in missing:
log(f"WARNING: {p} no longer exists upstream - skipping")
selected = [bypkg[p] for p in pkgs if p in bypkg]
# what we already publish, for a change report
old_versions = {}
old_index = os.path.join(ROOT, "index.json")
if os.path.exists(old_index):
try:
for e in json.load(open(old_index))["extensionList"]["extensions"]:
old_versions[e["packageName"]] = e["versionName"]
except Exception:
pass
out_exts = []
keep = {"apk": set(), "jar": set(), "icon": set()}
changed, added, downloaded = [], [], 0
for e in sorted(selected, key=lambda x: x["name"].lower()):
pkg = e["packageName"]
res = e.get("resources", {})
apk_url = res.get("apkUrl", "")
jar_url = res.get("jarUrl", "")
icon_url = res.get("iconUrl", "")
apk_name = os.path.basename(apk_url)
jar_name = os.path.basename(jar_url) if jar_url else ""
icon_name = pkg + ".png"
prev = old_versions.get(pkg)
if prev is None:
added.append((e["name"], e["versionName"]))
elif prev != e["versionName"]:
changed.append((e["name"], prev, e["versionName"]))
keep["apk"].add(apk_name)
if jar_name:
keep["jar"].add(jar_name)
keep["icon"].add(icon_name)
if not args.dry_run:
try:
if download(apk_url, os.path.join(APK_DIR, apk_name)):
downloaded += 1
log(f" + apk {apk_name}")
except Exception as ex:
log(f" ! FAILED apk {pkg}: {ex}")
continue
if jar_url and not args.no_jars:
try:
if download(jar_url, os.path.join(JAR_DIR, jar_name)):
downloaded += 1
log(f" + jar {jar_name}")
except Exception as ex:
log(f" ! jar {pkg}: {ex} (leaving upstream URL)")
jar_name = ""
if icon_url:
# jsdelivr 403s on some paths; fall back to raw.githubusercontent
for url in icon_fallbacks(icon_url):
try:
if download(url, os.path.join(ICON_DIR, icon_name)):
downloaded += 1
break
except Exception as ex:
last = ex
else:
if extract_icon_from_apk(os.path.join(APK_DIR, apk_name),
os.path.join(ICON_DIR, icon_name)):
downloaded += 1
log(f" ~ icon {pkg}: not published upstream, "
f"extracted from apk")
else:
log(f" ! icon {pkg}: {last} (leaving upstream URL)")
icon_name = ""
new = json.loads(json.dumps(e)) # deep copy
# the pb encoder omits empty strings, so an empty key here would break
# the round-trip assert below - only emit keys that have a value
res_out = {"apkUrl": f"{BASE_URL}/apk/{apk_name}"}
icon_final = f"{BASE_URL}/icon/{icon_name}" if icon_name else icon_url
if icon_final:
res_out["iconUrl"] = icon_final
jar_final = (f"{BASE_URL}/jar/{jar_name}"
if jar_name and not args.no_jars else jar_url)
if jar_final:
res_out["jarUrl"] = jar_final
new["resources"] = res_out
new.pop("_store", None)
out_exts.append(new)
# ---- report -------------------------------------------------------
print()
log(f"selected {len(out_exts)} extension(s), {downloaded} new file(s) downloaded")
if added:
print("\n NEW:")
for n, v in added:
print(f" + {n} {v}")
if changed:
print("\n UPDATED:")
for n, a, b in changed:
print(f" ^ {n}: {a} -> {b}")
if not added and not changed:
print("\n everything already up to date")
if args.dry_run:
print("\n[dry-run] no files written")
return 0
# ---- write indexes ------------------------------------------------
with open(os.path.join(ROOT, "index.json"), "w") as fh:
json.dump({**META, "extensionList": {"extensions": out_exts}}, fh, indent=2)
legacy = [to_legacy(e, os.path.basename(e["resources"]["apkUrl"])) for e in out_exts]
with open(os.path.join(ROOT, "index.min.json"), "w") as fh:
json.dump(legacy, fh, separators=(",", ":"))
size = mihon_pb.write_index_pb(os.path.join(ROOT, "index.pb"), META, out_exts)
# verify what we just wrote actually decodes back to what we meant
_, rt = mihon_pb.decode_repo(open(os.path.join(ROOT, "index.pb"), "rb").read())
assert rt == out_exts, "index.pb round-trip verification FAILED"
log(f"wrote index.pb ({size} bytes uncompressed), index.json, index.min.json")
# repo.json - REQUIRED, do not delete. Two different apps demand it:
#
# * Aniyomi's *manga* side (ExtensionRepoService.fetchRepoDetails) only ever
# GETs "/repo.json". No repo.json = the repo cannot be added at all.
# * Mihon / Aniyomi new-format (ExtensionStoreService.fetch): if the pasted
# URL returns a JSON array it rewrites ".../index.min.json" -> ".../repo.json"
# and reads the store metadata from there.
#
# "index_v2" is the bridge that upgrades those legacy clients onto index.pb;
# Aniyomi's manga path ignores the key (ignoreUnknownKeys = true).
repo_json = {
"index_v2": f"{BASE_URL}/index.pb",
"meta": {
"name": META["name"],
"shortName": META["badgeLabel"],
"website": META["contact"]["website"],
"signingKeyFingerprint": META["signingKey"],
},
}
with open(os.path.join(ROOT, "repo.json"), "w") as fh:
json.dump(repo_json, fh, indent=2)
log("wrote repo.json (required by Aniyomi manga + legacy index.min.json clients)")
# keep the legacy /repo/ path in sync with the root so both URLs work
os.makedirs(os.path.join(ROOT, "repo"), exist_ok=True)
for f in ("index.pb", "index.json", "index.min.json", "repo.json"):
shutil.copy2(os.path.join(ROOT, f), os.path.join(ROOT, "repo", f))
log("mirrored indexes into ./repo/ for backwards compatibility")
# ---- prune --------------------------------------------------------
if args.prune:
for sub, kept in (("apk", keep["apk"]), ("jar", keep["jar"]),
("icon", keep["icon"])):
d = os.path.join(ROOT, sub)
if not os.path.isdir(d):
continue
for f in os.listdir(d):
p = os.path.join(d, f)
if os.path.isdir(p):
log(f" - removing stray dir {sub}/{f}")
shutil.rmtree(p)
elif f not in kept:
log(f" - pruning {sub}/{f}")
os.remove(p)
print()
log("done. Repo URL for Mihon 0.20.1+ : " + BASE_URL + "/index.pb")
log(" legacy forks : " + BASE_URL + "/index.min.json")
return 0
if __name__ == "__main__":
sys.exit(main())