192 lines
6.6 KiB
Python
192 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Compare committed **deployment / mapping registries** against
|
||
`DEPLOYED_CONTRACTS_UNIFIED_EXTENDED.md` to find (chain, address) pairs
|
||
that should appear in the table but do not.
|
||
|
||
Writes: reports/inventory/INVENTORY_COVERAGE_GAPS.md
|
||
|
||
This does *not* scan uncommitted Foundry `broadcast/` artifacts.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from inventory_onchain import ROOT, UNIFIED_EXTENDED_PATH, parse_table
|
||
|
||
OUT_MD = ROOT / "reports/inventory/INVENTORY_COVERAGE_GAPS.md"
|
||
|
||
# Deployer EOA: not a "contract" line for the extended table
|
||
SKIP_ADDRESSES = {
|
||
"0x4a666f96fc8764181194447a7dfdb7d471b301c8",
|
||
}
|
||
|
||
|
||
def addresses_from_table() -> set[tuple[str, str]]:
|
||
s: set[tuple[str, str]] = set()
|
||
for r in parse_table(UNIFIED_EXTENDED_PATH):
|
||
s.add((r["chain"].strip().lower() if r["chain"] else "", r["address"].lower()))
|
||
return s
|
||
|
||
|
||
def from_address_inventory() -> set[tuple[str, str]]:
|
||
p = ROOT / "smom-dbis-138/config/address-inventory.chain138.json"
|
||
out: set[tuple[str, str]] = set()
|
||
if not p.is_file():
|
||
return out
|
||
j = json.loads(p.read_text())
|
||
for _k, v in j.get("chain138Inventory", {}).items():
|
||
if isinstance(v, str) and v.startswith("0x") and len(v) == 42:
|
||
if v.lower() in SKIP_ADDRESSES:
|
||
continue
|
||
if int(v, 16) == 0:
|
||
continue
|
||
out.add(("138", v.lower()))
|
||
return out
|
||
|
||
|
||
def from_token_mapping() -> set[tuple[str, str]]:
|
||
p = ROOT / "config/token-mapping-multichain.json"
|
||
out: set[tuple[str, str]] = set()
|
||
if not p.is_file():
|
||
return out
|
||
j = json.loads(p.read_text())
|
||
for pair in j.get("pairs", []):
|
||
fc, tc = pair.get("fromChainId"), pair.get("toChainId")
|
||
for t in pair.get("tokens", []):
|
||
a_from = t.get("addressFrom")
|
||
a_to = t.get("addressTo")
|
||
if (
|
||
isinstance(a_from, str)
|
||
and a_from.startswith("0x")
|
||
and len(a_from) == 42
|
||
and int(a_from, 16) != 0
|
||
and fc is not None
|
||
):
|
||
out.add((str(int(fc)), a_from.lower()))
|
||
if (
|
||
isinstance(a_to, str)
|
||
and a_to.startswith("0x")
|
||
and len(a_to) == 42
|
||
and int(a_to, 16) != 0
|
||
and tc is not None
|
||
):
|
||
out.add((str(int(tc)), a_to.lower()))
|
||
return out
|
||
|
||
|
||
def from_csv() -> set[tuple[str, str]]:
|
||
p = ROOT / "reports/inventory/deployed-contracts-by-network.csv"
|
||
out: set[tuple[str, str]] = set()
|
||
if not p.is_file():
|
||
return out
|
||
with p.open(newline="", encoding="utf-8") as f:
|
||
r = csv.DictReader(f)
|
||
for row in r:
|
||
ch = str(row.get("chain_id", "")).strip()
|
||
addr = (row.get("address") or "").strip()
|
||
if addr.startswith("0x") and len(addr) == 42:
|
||
out.add((ch, addr.lower()))
|
||
return out
|
||
|
||
|
||
def from_destination_executors() -> set[tuple[str, str]]:
|
||
out: set[tuple[str, str]] = set()
|
||
|
||
def walk(o: object, chain: str | None) -> None:
|
||
if isinstance(o, dict):
|
||
if o.get("chainId") is not None:
|
||
chain = str(int(o["chainId"]))
|
||
da = o.get("deployedAddress")
|
||
if (
|
||
isinstance(da, str)
|
||
and len(da) == 42
|
||
and int(da, 16) != 0
|
||
and chain is not None
|
||
):
|
||
out.add((chain, da.lower()))
|
||
for v in o.values():
|
||
walk(v, chain)
|
||
elif isinstance(o, list):
|
||
for x in o:
|
||
walk(x, chain)
|
||
|
||
for p in ROOT.glob("atomic-swap-dapp/**/destination-executor*.json"):
|
||
try:
|
||
j = json.loads(p.read_text())
|
||
except (json.JSONDecodeError, OSError):
|
||
continue
|
||
walk(j, None)
|
||
return out
|
||
|
||
|
||
def main() -> int:
|
||
if not UNIFIED_EXTENDED_PATH.is_file():
|
||
print("Missing unified extended", file=sys.stderr)
|
||
return 1
|
||
have = addresses_from_table()
|
||
sources: list[tuple[str, set[tuple[str, str]]]] = [
|
||
("address-inventory.chain138.json", from_address_inventory()),
|
||
("config/token-mapping-multichain.json", from_token_mapping()),
|
||
("deployed-contracts-by-network.csv", from_csv()),
|
||
(
|
||
"atomic-swap-dapp destination-executor*.json (deployedAddress + chainId)",
|
||
from_destination_executors(),
|
||
),
|
||
]
|
||
union: set[tuple[str, str]] = set()
|
||
for _name, s in sources:
|
||
union |= s
|
||
|
||
missing: list[tuple[str, str, str]] = []
|
||
for ch, addr in sorted(union, key=lambda x: (int(x[0]) if x[0].isdigit() else 0, x[1])):
|
||
if (ch, addr) in have or (ch.strip().lower(), addr) in have:
|
||
continue
|
||
if addr in SKIP_ADDRESSES or int(addr, 16) == 0:
|
||
continue
|
||
src = [n for n, st in sources if (ch, addr) in st]
|
||
missing.append((ch, addr, ", ".join(src) if src else "unknown source"))
|
||
|
||
lines = [
|
||
"# Inventory coverage — gaps vs `DEPLOYED_CONTRACTS_UNIFIED_EXTENDED.md`",
|
||
"",
|
||
f"**Compared to:** {len(have)} unique (chain × address) keys parsed from the extended table.",
|
||
f"**Union of registry sources below:** {len(union)} pairs (may overlap across sources).",
|
||
f"**In union but *not* in extended table:** {len(missing)} (excluding zero address / known EOAs in `SKIP_ADDRESSES`).",
|
||
"",
|
||
"## Registry sources",
|
||
"",
|
||
]
|
||
for n, s in sources:
|
||
lines.append(f"- **{n}:** {len(s)} pairs")
|
||
lines += [
|
||
"",
|
||
"## Pairs in registries, missing from extended (add a row for each, then regen live/deduped reports)",
|
||
"",
|
||
]
|
||
if not missing:
|
||
lines.append("*(No gaps with current sources — registries you asked to compare are subsumed by the table.)*")
|
||
else:
|
||
lines.append("| Chain | Address | Found in |")
|
||
lines.append("|---:|:---|:---|")
|
||
for ch, addr, src in missing:
|
||
lines.append(f"| {ch} | `{addr}` | {src} |")
|
||
lines.append("")
|
||
lines.append("## Limitations")
|
||
lines.append("")
|
||
lines.append("- `smom-dbis-138/docs/deployment/DEPLOYED_ADDRESSES.md` is not compared by address+chain (addresses appear without a machine-readable chain in one block).")
|
||
lines.append("- Local **Foundry `broadcast/run-latest.json`** files (if not committed) are not scanned.")
|
||
lines.append("")
|
||
|
||
OUT_MD.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
print("Wrote", OUT_MD)
|
||
print("Gaps:", len(missing))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|