124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
publish_forge.py — Publish Centaur Quant Architecture to MQL5 Algo Forge.
|
|
|
|
Algo Forge (forge.mql5.io) is a Gitea/Forgejo-based Git platform tied to
|
|
mql5.com accounts. This pipeline:
|
|
1. Authenticates via a personal API token (Settings -> Applications).
|
|
2. Ensures the repository exists (creates it via Gitea API if missing).
|
|
3. Pushes the local git history over HTTPS with the token in an
|
|
Authorization header ONLY (never persisted into .git/config).
|
|
|
|
Usage:
|
|
python publish_forge.py --token <FORGE_TOKEN> [--private]
|
|
# or set FORGE_TOKEN env var
|
|
|
|
Token: forge.mql5.io -> sign in (mql5.com) -> Settings -> Applications
|
|
-> Generate New Token (scope: read:user, write:repository).
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
BASE = "https://forge.mql5.io"
|
|
API = BASE + "/api/v1"
|
|
REPO = "Centaur_Quant_Architecture"
|
|
DESCRIPTION = ("Centaur Quant Architecture — Multi-Asset SMC Executor + "
|
|
"SDP Telemetry Bridge (MQL5 + Python). OB/FVG scanner, "
|
|
"anti-veto LLM scoring, file-based telemetry fallback.")
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DEFAULT_DIR = os.path.dirname(SCRIPT_DIR) # project root (parent of Python/)
|
|
|
|
|
|
def api(method, path, token, body=None):
|
|
req = urllib.request.Request(API + path, method=method)
|
|
req.add_header("Authorization", "token " + token)
|
|
req.add_header("Accept", "application/json")
|
|
data = None
|
|
if body is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
data = json.dumps(body).encode("utf-8")
|
|
try:
|
|
with urllib.request.urlopen(req, data=data, timeout=30) as r:
|
|
raw = r.read()
|
|
return r.status, (json.loads(raw) if raw else None)
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read()
|
|
try:
|
|
err = json.loads(raw)
|
|
except Exception:
|
|
err = raw.decode("utf-8", errors="replace")
|
|
return e.code, err
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="Publish project to MQL5 Algo Forge")
|
|
ap.add_argument("--token", default=os.environ.get("FORGE_TOKEN", ""),
|
|
help="Forge API token (or FORGE_TOKEN env)")
|
|
ap.add_argument("--repo", default=REPO, help="repository name")
|
|
ap.add_argument("--dir", default=DEFAULT_DIR, help="git repo root")
|
|
ap.add_argument("--branch", default="main")
|
|
ap.add_argument("--description", default=DESCRIPTION)
|
|
ap.add_argument("--private", action="store_true", help="create as private repo")
|
|
args = ap.parse_args()
|
|
|
|
if not args.token:
|
|
sys.exit("ERROR: no token. Set FORGE_TOKEN or pass --token "
|
|
"(create at forge.mql5.io -> Settings -> Applications).")
|
|
|
|
code, me = api("GET", "/user", args.token)
|
|
if code != 200:
|
|
sys.exit(f"ERROR: token rejected ({code}): {me}")
|
|
owner = me.get("login") or me.get("username")
|
|
print(f"[1/3] Authenticated as '{owner}'")
|
|
|
|
code, repo = api("GET", f"/repos/{owner}/{args.repo}", args.token)
|
|
if code == 404:
|
|
print(f"[2/3] Creating repository '{owner}/{args.repo}' ...")
|
|
code, repo = api("POST", "/user/repos", args.token, {
|
|
"name": args.repo,
|
|
"description": args.description,
|
|
"private": args.private,
|
|
"default_branch": args.branch,
|
|
"auto_init": False,
|
|
"has_issues": True,
|
|
"has_wiki": True,
|
|
})
|
|
if code != 201:
|
|
sys.exit(f"ERROR: repo create failed ({code}): {repo}")
|
|
print(" Repo created.")
|
|
elif code == 200:
|
|
print(f"[2/3] Repository exists: {repo['html_url']}")
|
|
else:
|
|
sys.exit(f"ERROR: repo check failed ({code}): {repo}")
|
|
|
|
subprocess.run(["git", "-C", args.dir, "remote", "remove", "forge"],
|
|
capture_output=True)
|
|
subprocess.run(["git", "-C", args.dir, "remote", "add", "forge",
|
|
f"{BASE}/{owner}/{args.repo}.git"], check=True)
|
|
|
|
print(f"[3/3] Pushing {args.branch} -> forge ...")
|
|
auth = base64.b64encode(f"git:{args.token}".encode()).decode()
|
|
p = subprocess.run(
|
|
["git", "-C", args.dir,
|
|
"-c", f"http.extraheader=Authorization: Basic {auth}",
|
|
"push", "forge", f"HEAD:{args.branch}"],
|
|
capture_output=True, text=True)
|
|
if p.stdout:
|
|
print(p.stdout)
|
|
if p.returncode != 0:
|
|
print(p.stderr)
|
|
sys.exit(f"ERROR: push failed ({p.returncode})")
|
|
|
|
print(f"\nPUBLISHED: {BASE}/{owner}/{args.repo}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|