Centaur_Quant_Architecture/Python/publish_forge.py

#!/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()