97 lines
No EOL
3.5 KiB
Python
97 lines
No EOL
3.5 KiB
Python
# Herramienta CLI y Lib simple para facilitar las consultas y descargar
|
|
# el swagger.json para la api de FORGEJO (usada en MQL5 Algo Forge)
|
|
|
|
#+------------------------------------------------------------------+
|
|
#| Imports |
|
|
#+------------------------------------------------------------------+
|
|
import json
|
|
import os
|
|
import time
|
|
import urllib.request
|
|
import sys
|
|
|
|
#+------------------------------------------------------------------+
|
|
#| Defines |
|
|
#+------------------------------------------------------------------+
|
|
# JSON de la api de forgejo
|
|
SWAGGER_URL : str = "https://forge.mql5.io/swagger.v1.json"
|
|
# Dir donde se ubicara el .json de cache
|
|
CACHE_DIR : str = os.path.join(
|
|
os.environ.get("TEMP") or os.environ.get("TMPDIR") or "/tmp",
|
|
"forgejo-algo-forge",
|
|
)
|
|
# Archivo de cache de swagger.json
|
|
CACHE_FILE : str = os.path.join(CACHE_DIR, "swagger.json")
|
|
|
|
#+------------------------------------------------------------------+
|
|
#| Funciones |
|
|
#+------------------------------------------------------------------+
|
|
def load_spec(max_age_seconds: int = 3600) -> dict:
|
|
"""
|
|
Return the parsed OpenAPI spec as a dict.
|
|
|
|
Downloads it once and caches it in a temp folder. Re-downloads
|
|
automatically if the cache is missing or older than max_age_seconds
|
|
(default: 1 hour). Safe to call repeatedly within a session — cached
|
|
reads are cheap.
|
|
"""
|
|
# Creamos nu nuevo folder siempre
|
|
os.makedirs(CACHE_DIR, exist_ok=True)
|
|
|
|
# Verficamos si requier descargar
|
|
nd : bool = (
|
|
not os.path.exists(CACHE_FILE)
|
|
or (time.time() - os.path.getmtime(CACHE_FILE)) > max_age_seconds
|
|
)
|
|
|
|
# En caso se requiera instalcio lo volvemos a bajar
|
|
# Y actulizamos el archivoa ctual
|
|
if nd:
|
|
urllib.request.urlretrieve(SWAGGER_URL, CACHE_FILE)
|
|
|
|
# Cargamos el json
|
|
with open(CACHE_FILE, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def search_paths(spec: dict, keyword: str) -> list[str]:
|
|
"""
|
|
Return every path in the spec whose URL contains `keyword`
|
|
(case-insensitive). Use this instead of guessing or memorizing
|
|
endpoint URLs.
|
|
"""
|
|
keyword = keyword.lower()
|
|
return [p for p in spec["paths"] if keyword in p.lower()]
|
|
|
|
|
|
def get_path_schema(spec: dict, path: str) -> dict:
|
|
"""
|
|
Return the full schema (methods, params, responses) for one exact
|
|
path, e.g. '/repos/{owner}/{repo}/actions/runs'.
|
|
|
|
Raises KeyError if the path doesn't exist verbatim — use
|
|
search_paths() first to find the exact string.
|
|
"""
|
|
return spec["paths"][path]
|
|
|
|
|
|
def force_refresh() -> dict:
|
|
"""Ignore the cache and re-download the spec unconditionally."""
|
|
return load_spec(max_age_seconds=0)
|
|
|
|
#+------------------------------------------------------------------+
|
|
#| Main |
|
|
#+------------------------------------------------------------------+
|
|
if __name__ == "__main__":
|
|
# Bajamos siempore la spec
|
|
spec = load_spec()
|
|
print(f"Loaded spec with {len(spec['paths'])} paths (cached at {CACHE_FILE})")
|
|
|
|
# En caso se haya proporcionado un argumento
|
|
# (seria el de path) lo buscamos..
|
|
if len(sys.argv) > 1:
|
|
keyword = sys.argv[1]
|
|
matches = search_paths(spec, keyword)
|
|
print(f"\nPaths matching '{keyword}':")
|
|
for m in matches:
|
|
print(f" {m}") |