scripts/sync_broker_sdk_refs.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Sync repo-local broker SDK reference mirrors.
The runtime installation is still controlled by ``uv.lock`` and
``packages/integrations/gateway/pyproject.toml``. This script gives agents and
maintainers a repeatable way to keep the corresponding SDK source/artifact
references inside the checkout at ``.local/sdk-audit/`` for parity checks,
without committing vendored SDK code.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import subprocess
import sys
import tomllib
import urllib.error
import urllib.request
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
REPO = Path(__file__).resolve().parents[1]
DEFAULT_LOCK = REPO / "brokers.lock"
DEFAULT_AUDIT_ROOT = REPO / ".local" / "sdk-audit"
@dataclass(frozen=True)
class BrokerSdkSource:
"""Known upstream locations for an installable broker SDK pin."""
package: str
repo_name: str | None = None
git_url: str | None = None
pypi_name: str | None = None
notes: str = ""
SDK_SOURCES: dict[str, BrokerSdkSource] = {
"dhanhq": BrokerSdkSource(
package="dhanhq",
repo_name="dhanhq-py",
git_url="https://github.com/dhan-oss/dhanhq-py.git",
pypi_name="dhanhq",
),
"upstox-python-sdk": BrokerSdkSource(
package="upstox-python-sdk",
repo_name="upstox-python",
git_url="https://github.com/upstox/upstox-python.git",
pypi_name="upstox-python-sdk",
),
"kotakneoapi": BrokerSdkSource(
package="kotakneoapi",
repo_name="kotak-neo-python",
git_url="https://github.com/Kotak-Neo/kotak-neo-python.git",
pypi_name="kotakneoapi",
notes="Kotak Neo v3 freshness follows the pinned release tag and commit, not default-branch HEAD.",
),
"growwapi": BrokerSdkSource(
package="growwapi",
pypi_name="growwapi",
notes="Groww's official Python SDK is currently distributed through PyPI docs/package metadata.",
),
}
SDKLESS_BROKERS: tuple[dict[str, str], ...] = (
{
"broker": "indmoney",
"advertised_package": "indstocks-sdk",
"pypi_json": "https://pypi.org/pypi/indstocks-sdk/json",
"npm_json": "https://registry.npmjs.org/indstocks-sdk",
"reason": "INDstocks advertises SDK install commands, but no package exists on PyPI or npm yet.",
},
)
UrlOpener = Callable[[str], tuple[int, bytes]]
def _open_url(url: str) -> tuple[int, bytes]:
request = urllib.request.Request(url, headers={"User-Agent": "flinttrade-sdk-sync/1.0"})
try:
with urllib.request.urlopen(request, timeout=60) as response:
status = getattr(response, "status", 200)
return int(status), response.read()
except urllib.error.HTTPError as exc:
return int(exc.code), exc.read()
def _run(args: list[str], cwd: Path | None = None) -> str:
result = subprocess.run(args, cwd=cwd, check=True, text=True, capture_output=True)
return result.stdout.strip()
def _display_path(path: Path) -> str:
"""Return a repo-relative path when possible, otherwise an absolute path."""
resolved = path.resolve()
try:
return str(resolved.relative_to(REPO))
except ValueError:
return str(resolved)
def load_broker_lock(path: Path = DEFAULT_LOCK) -> list[dict[str, Any]]:
"""Return broker entries from ``brokers.lock``."""
data = tomllib.loads(path.read_text(encoding="utf-8"))
brokers = data.get("broker", [])
return [entry for entry in brokers if isinstance(entry, dict)]
def active_sdk_entries(entries: list[dict[str, Any]]) -> list[dict[str, str]]:
"""Return non-placeholder SDK entries that have known upstream locations."""
out: list[dict[str, str]] = []
for entry in entries:
name = str(entry.get("name", "")).strip()
version = str(entry.get("version", "")).strip()
if not name or not version or "PLACEHOLDER" in version.upper():
continue
if name not in SDK_SOURCES:
continue
source_commit = str(entry.get("source_commit", "")).strip()
row = {
"name": name,
"version": version,
"source_commit": source_commit,
}
sha256 = str(entry.get("sha256", "")).strip()
if sha256:
# Carried so the downloaded audit artifact can be cross-checked
# against the brokers.lock pin (not only PyPI's self-advertised
# digest) — PyPI may add files to an existing release.
row["sha256"] = sha256
if name == "kotakneoapi":
release_version = entry.get("release_version")
if not isinstance(release_version, str) or re.fullmatch(r"\d+\.\d+\.\d+", release_version) is None:
raise ValueError("Kotak Neo release_version must be an explicit stable version")
row["release_version"] = release_version
for field in (
"source_tree",
"release_tag",
"release_commit",
"release_tree",
"release_wheel_sha256",
"release_sdist_sha256",
):
value = str(entry.get(field, "")).strip()
if value:
row[field] = value
out.append(row)
return out
def sync_git_mirror(source: BrokerSdkSource, audit_root: Path) -> dict[str, str]:
"""Clone or fetch an SDK source repository under ``audit_root``."""
if not source.git_url or not source.repo_name:
return {}
repo = audit_root / source.repo_name
if repo.exists():
if not (repo / ".git").exists():
raise RuntimeError(f"{repo} exists but is not a git checkout")
_run(["git", "-C", str(repo), "fetch", "--tags", "--force", "--prune", "origin"])
else:
repo.parent.mkdir(parents=True, exist_ok=True)
_run(["git", "clone", source.git_url, str(repo)])
# Drift must be measured against the REMOTE head, not the local checkout's
# HEAD: `git fetch` only updates remote-tracking refs, so rev-parse HEAD is
# frozen at clone time and would report `git-current` forever even after
# upstream moves — silently defeating the drift gate for the one SDK whose
# only distribution channel is git (Renovate cannot watch git-pinned uv
# sources either).
remote_head = _run(["git", "-C", str(repo), "ls-remote", "origin", "HEAD"]).split()[0]
result = {
"path": str(repo.relative_to(REPO)),
"remote": _run(["git", "-C", str(repo), "remote", "get-url", "origin"]),
"head": remote_head,
"describe": _run(["git", "-C", str(repo), "describe", "--tags", "--always", "--dirty"]),
}
if source.package == "kotakneoapi":
result["head_tree"] = _run(["git", "-C", str(repo), "rev-parse", f"{remote_head}^{{tree}}"])
tags = _run(["git", "-C", str(repo), "tag", "--list", "v*.*.*"]).splitlines()
stable = [tag for tag in tags if re.fullmatch(r"v\d+\.\d+\.\d+", tag)]
if not stable:
raise RuntimeError("Kotak Neo upstream has no stable release tag")
latest = max(stable, key=lambda tag: tuple(map(int, tag[1:].split("."))))
result["latest_stable_tag"] = latest
result["release_commit"] = _run(["git", "-C", str(repo), "rev-parse", f"refs/tags/{latest}^{{commit}}"])
result["release_tree"] = _run(["git", "-C", str(repo), "rev-parse", f"refs/tags/{latest}^{{tree}}"])
return result
def pypi_release_json(package: str, opener: UrlOpener = _open_url) -> dict[str, Any]:
"""Fetch PyPI metadata for ``package``."""
status, body = opener(f"https://pypi.org/pypi/{package}/json")
if status != 200:
raise RuntimeError(f"PyPI returned HTTP {status} for {package}")
return json.loads(body.decode("utf-8"))
def latest_pypi_version(package: str, opener: UrlOpener = _open_url) -> str:
"""Return the latest version advertised by PyPI for ``package``."""
data = pypi_release_json(package, opener)
version = str((data.get("info") or {}).get("version") or "").strip()
if not version:
raise RuntimeError(f"PyPI metadata for {package} did not include info.version")
return version
def select_pypi_artifact(files: list[Mapping[str, Any]]) -> Mapping[str, Any]:
"""Pick the best audit artifact from a PyPI release file list."""
if not files:
raise RuntimeError("PyPI release has no files")
def score(file: Mapping[str, Any]) -> tuple[int, str]:
packagetype = str(file.get("packagetype", ""))
filename = str(file.get("filename", ""))
python_version = str(file.get("python_version", ""))
if packagetype == "bdist_wheel" and python_version == "py3" and filename.endswith("none-any.whl"):
return (0, filename)
if packagetype == "bdist_wheel":
return (1, filename)
if packagetype == "sdist":
return (2, filename)
return (3, filename)
return min(files, key=score)
def download_pypi_artifact(
package: str,
version: str,
out_dir: Path,
opener: UrlOpener = _open_url,
pinned_sha: str = "",
) -> dict[str, str]:
"""Download the pinned PyPI artifact and verify its SHA256 digest.
The download is verified against BOTH PyPI's self-advertised digest AND the
``pinned_sha`` recorded in brokers.lock (when supplied). PyPI permits adding
files to an existing release, and this tool prefers wheels over sdists, so a
match against PyPI's own digest alone could silently cache an artifact
different from the one the repo pinned. A mismatch against the brokers.lock
pin is a supply-chain red flag and raises.
"""
data = pypi_release_json(package, opener)
releases = data.get("releases", {})
files = releases.get(version)
if not isinstance(files, list):
raise RuntimeError(f"PyPI has no release files for {package}=={version}")
artifact = select_pypi_artifact(files)
url = str(artifact["url"])
filename = str(artifact["filename"])
expected_sha = str((artifact.get("digests") or {}).get("sha256") or "")
status, body = opener(url)
if status != 200:
raise RuntimeError(f"artifact download returned HTTP {status} for {filename}")
actual_sha = hashlib.sha256(body).hexdigest()
if expected_sha and actual_sha != expected_sha:
raise RuntimeError(f"sha256 mismatch for {filename}: expected {expected_sha}, got {actual_sha}")
pin = pinned_sha.strip().lower()
matched_brokers_lock = None
if pin and "placeholder" not in pin:
# brokers.lock records the artifact sha256 for PyPI brokers (the same
# value requirements.lock pins). A mismatch means PyPI is serving a
# different artifact for this pinned release than the repo attested.
if actual_sha != pin:
raise RuntimeError(
f"brokers.lock sha256 mismatch for {package}=={version} ({filename}): "
f"pinned {pin}, downloaded {actual_sha}"
)
matched_brokers_lock = True
out_dir.mkdir(parents=True, exist_ok=True)
target = out_dir / filename
tmp = target.with_suffix(target.suffix + ".tmp")
tmp.write_bytes(body)
tmp.replace(target)
result = {
"path": _display_path(target),
"filename": filename,
"sha256": actual_sha,
"url": url,
}
if matched_brokers_lock is not None:
result["brokers_lock_sha_match"] = "true"
return result
def registry_status(url: str, opener: UrlOpener = _open_url) -> str:
"""Return a compact HTTP status string for a package-registry URL."""
status, _body = opener(url)
return str(status)
def build_readme(manifest: dict[str, Any]) -> str:
"""Render a short human-readable summary for ``.local/sdk-audit``."""
lines = [
"# Broker SDK Audit Cache",
"",
"Generated by `uv run python scripts/sync_broker_sdk_refs.py`.",
"This directory is gitignored and contains repo-local SDK source mirrors and package artifacts.",
"",
f"Generated at: {manifest['generated_at']}",
"",
"## SDKs",
"",
]
for entry in manifest["sdks"]:
lines.append(f"- {entry['package']} {entry['version']}")
if entry.get("release_version"):
lines.append(f" - release compatibility baseline: {entry['release_version']}")
if entry.get("git"):
lines.append(f" - git: {entry['git']['path']} @ {entry['git']['describe']}")
if "git_current" in entry:
if entry["package"] == "kotakneoapi":
status = "current" if entry["git_current"] else "locked behind the pinned release"
else:
status = "current" if entry["git_current"] else "locked behind upstream HEAD"
lines.append(f" - git status: {status}")
if entry.get("pypi"):
lines.append(f" - pypi: {entry['pypi']['path']}")
latest = entry.get("pypi_latest_version")
if latest:
status = "current" if entry.get("pypi_current") else "locked behind latest"
lines.append(f" - PyPI latest: {latest} ({status})")
if entry.get("notes"):
lines.append(f" - notes: {entry['notes']}")
if manifest.get("drift"):
lines.extend(["", "## Drift", ""])
for entry in manifest["drift"]:
lines.append(
f"- {entry['package']}: locked {entry['locked']} vs upstream {entry['upstream']} ({entry['source']})"
)
if manifest.get("sdkless_brokers"):
lines.extend(["", "## SDK-Less Brokers", ""])
for entry in manifest["sdkless_brokers"]:
lines.append(
f"- {entry['broker']}: {entry['advertised_package']} "
f"(PyPI {entry.get('pypi_status', 'unchecked')}, npm {entry.get('npm_status', 'unchecked')})"
)
lines.append(f" - reason: {entry['reason']}")
lines.append("")
return "\n".join(lines)
def sync(
*,
lock_path: Path = DEFAULT_LOCK,
audit_root: Path = DEFAULT_AUDIT_ROOT,
check_missing: bool = True,
opener: UrlOpener = _open_url,
) -> dict[str, Any]:
"""Sync all known broker SDK references and return the manifest."""
audit_root.mkdir(parents=True, exist_ok=True)
pypi_dir = audit_root / "pypi"
entries = active_sdk_entries(load_broker_lock(lock_path))
manifest: dict[str, Any] = {
"generated_at": datetime.now(UTC).replace(microsecond=0).isoformat(),
"lock_path": _display_path(lock_path),
"audit_root": _display_path(audit_root),
"sdks": [],
"drift": [],
"sdkless_brokers": [],
}
for entry in entries:
source = SDK_SOURCES[entry["name"]]
sdk_record: dict[str, Any] = {
"package": entry["name"],
"version": entry["version"],
"source_commit": entry["source_commit"],
"notes": source.notes,
}
if entry.get("release_version"):
sdk_record["release_version"] = entry["release_version"]
if source.git_url:
git = sync_git_mirror(source, audit_root)
sdk_record["git"] = git
if entry["name"] == "kotakneoapi":
# Default-branch HEAD is recorded on the audit mirror, but it is
# not the freshness gate. A move on upstream main must not fail
# the pin while the locked release tag still peels to the same
# commit and tree.
release_commit = entry.get("release_commit", "")
if release_commit:
sdk_record["git_current"] = str(git.get("release_commit", "")) == release_commit
compared = (
("release_tag", "latest_stable_tag", "release-tag"),
("release_commit", "release_commit", "release-commit"),
("release_tree", "release_tree", "release-tree"),
)
for locked_field, observed_field, drift_source in compared:
locked = entry.get(locked_field, "")
observed = git.get(observed_field, "")
if locked and observed != locked:
manifest["drift"].append(
{"package": entry["name"], "source": drift_source, "locked": locked, "upstream": observed}
)
elif entry["source_commit"]:
current = str(git.get("head", "")) == entry["source_commit"]
sdk_record["git_current"] = current
if not current:
manifest["drift"].append(
{
"package": entry["name"],
"source": "git",
"locked": entry["source_commit"],
"upstream": str(git.get("head", "")),
}
)
if source.pypi_name:
release_version = entry.get("release_version", entry["version"])
latest = latest_pypi_version(source.pypi_name, opener)
sdk_record["pypi_latest_version"] = latest
sdk_record["pypi_current"] = latest == release_version
if latest != release_version:
manifest["drift"].append(
{
"package": entry["name"],
"source": "pypi",
"locked": release_version,
"upstream": latest,
}
)
sdk_record["pypi"] = download_pypi_artifact(
source.pypi_name,
release_version,
pypi_dir,
opener,
pinned_sha=entry.get("sha256", ""),
)
if entry["name"] == "kotakneoapi":
release = pypi_release_json(source.pypi_name, opener)
files = release.get("releases", {}).get(release_version, [])
for packagetype, field, drift_source in (
("bdist_wheel", "release_wheel_sha256", "release-wheel"),
("sdist", "release_sdist_sha256", "release-sdist"),
):
matches = [file for file in files if file.get("packagetype") == packagetype]
observed = str((matches[0].get("digests") or {}).get("sha256", "")) if matches else ""
locked = entry.get(field, "")
if locked and observed != locked:
manifest["drift"].append(
{"package": entry["name"], "source": drift_source, "locked": locked, "upstream": observed}
)
manifest["sdks"].append(sdk_record)
for sdkless in SDKLESS_BROKERS:
record = dict(sdkless)
if check_missing:
record["pypi_status"] = registry_status(record["pypi_json"], opener)
record["npm_status"] = registry_status(record["npm_json"], opener)
else:
record["pypi_status"] = "unchecked"
record["npm_status"] = "unchecked"
manifest["sdkless_brokers"].append(record)
(audit_root / "manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
(audit_root / "README.md").write_text(build_readme(manifest), encoding="utf-8")
return manifest
def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--lock", type=Path, default=DEFAULT_LOCK, help="brokers.lock path")
parser.add_argument("--audit-root", type=Path, default=DEFAULT_AUDIT_ROOT, help="output directory")
parser.add_argument(
"--skip-missing-registry-check",
action="store_true",
help="do not query PyPI/npm for SDK-less broker evidence",
)
parser.add_argument(
"--fail-on-drift",
action="store_true",
help="return a non-zero status if any locked SDK is behind upstream metadata",
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv or sys.argv[1:])
manifest = sync(
lock_path=args.lock,
audit_root=args.audit_root,
check_missing=not args.skip_missing_registry_check,
)
print(f"synced {len(manifest['sdks'])} broker SDK refs into {manifest['audit_root']}")
for entry in manifest["sdks"]:
bits = [f"{entry['package']}=={entry['version']}"]
if entry.get("git"):
bits.append(f"git={entry['git']['describe']}")
if "git_current" in entry:
bits.append("git-current" if entry["git_current"] else "git-drift")
if entry.get("pypi"):
bits.append(f"artifact={entry['pypi']['filename']}")
bits.append("pypi-current" if entry.get("pypi_current") else "pypi-drift")
print(" - " + ", ".join(bits))
if manifest.get("drift"):
print("drift detected:")
for entry in manifest["drift"]:
print(
f" - {entry['package']}: locked {entry['locked']} vs upstream {entry['upstream']} ({entry['source']})"
)
return 1 if args.fail_on_drift and manifest.get("drift") else 0
if __name__ == "__main__":
raise SystemExit(main())