romfast/echo-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: auto-commit from dashboard

Browse Source
This commit is contained in:
Marius Mutu
2026-07-10 15:12:22 +00:00
parent 7b44749e5a
commit 738dc0181a
13 changed files with 383 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
tools/roa2web_client.py Normal file
View File

@@ -0,0 +1,214 @@
#!/usr/bin/env python3
"""
Client generic pentru API-ul roa2web — autentificare + query-uri rapoarte.
Autentificare: user/parolă Oracle (ca la login-ul normal ROA) + 2FA email la
prima logare. După primul login cu trust_device=True, sesiunile următoare
folosesc refresh_token din keyring — fără OTP.
Credențiale (keyring, service "echo-core", vezi src/credential_store.py):
roa2web_user, roa2web_pass, roa2web_server,
roa2web_access_token, roa2web_refresh_token, roa2web_trusted_device_token
Firma implicită și base_url sunt în config.json -> "roa2web".
Usage:
python3 tools/roa2web_client.py summary [--company COD_SAU_NUME] [--luna N] [--an N]
python3 tools/roa2web_client.py companies
python3 tools/roa2web_client.py get <path> [--company COD_SAU_NUME] [--param k=v ...]
python3 tools/roa2web_client.py verify-2fa <cod> <email>
"""
import argparse
import json
import sys
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
import requests
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from src.credential_store import get_secret, set_secret
from src.config import Config
class ROA2WebError(Exception):
pass
class ROA2WebClient:
def __init__(self):
self.config = Config()
self.base_url = self.config.get("roa2web.base_url", "https://roa2web.romfast.ro/roa2web/api")
self.session = requests.Session()
self._token = get_secret("roa2web_access_token")
parsed = urlsplit(self.base_url)
self._origin = f"{parsed.scheme}://{parsed.netloc}"
# IIS/ARR reverse-proxy-ul din fata backend-ului serveste totul sub /roa2web,
# dar FastAPI nu stie de acest prefix -> redirect-urile lui (307, ex. trailing
# slash pe rutele definite la "/") vin fara el si pica in 404. Il reinjectam.
self._prefix = parsed.path[: -len("/api")] if parsed.path.endswith("/api") else parsed.path
@property
def default_company(self) -> str:
return self.config.get("roa2web.default_company", "114")
def _store_tokens(self, data: dict) -> None:
set_secret("roa2web_access_token", data["access_token"])
if data.get("refresh_token"):
set_secret("roa2web_refresh_token", data["refresh_token"])
if data.get("trusted_device_token"):
set_secret("roa2web_trusted_device_token", data["trusted_device_token"])
self._token = data["access_token"]
def _login_password(self) -> None:
user = get_secret("roa2web_user")
pw = get_secret("roa2web_pass")
server = get_secret("roa2web_server") or "romfast"
if not user or not pw:
raise ROA2WebError("Lipsesc roa2web_user/roa2web_pass din keyring (eco secrets set ...)")
payload = {"username": user, "password": pw, "server_id": server}
trusted = get_secret("roa2web_trusted_device_token")
if trusted:
payload["trusted_device_token"] = trusted
r = self.session.post(f"{self.base_url}/auth/login", json=payload, timeout=15)
r.raise_for_status()
data = r.json()
if data.get("requires_2fa"):
raise ROA2WebError(
f"Necesar cod OTP nou, trimis pe {data.get('masked_email')}. "
"Cere-i lui Marius codul, apoi ruleaza verify_2fa(code, email)."
)
self._store_tokens(data)
def _refresh(self) -> bool:
refresh_token = get_secret("roa2web_refresh_token")
if not refresh_token:
return False
r = self.session.post(
f"{self.base_url}/auth/refresh",
json={"refresh_token": refresh_token},
timeout=15,
)
if r.status_code != 200:
return False
self._store_tokens(r.json())
return True
def verify_2fa(self, code: str, email: str, server_id: str | None = None) -> dict:
server = server_id or get_secret("roa2web_server") or "romfast"
r = self.session.post(
f"{self.base_url}/auth/verify-2fa-code",
json={"code": code, "email": email, "server_id": server, "trust_device": True},
timeout=15,
)
r.raise_for_status()
data = r.json()
self._store_tokens(data)
return data
def request(self, method: str, path: str, params: dict | None = None, _retried: bool = False) -> dict:
if not self._token and not self._refresh():
self._login_password()
url = f"{self.base_url}{path}" if path.startswith("/") else f"{self.base_url}/{path}"
headers = {"Authorization": f"Bearer {self._token}"}
r = self.session.request(method, url, params=params, headers=headers, timeout=30, allow_redirects=False)
if r.status_code in (307, 308) and "Location" in r.headers:
loc = urlsplit(r.headers["Location"])
loc_path = loc.path
if self._prefix and not loc_path.startswith(self._prefix):
loc_path = self._prefix + loc_path
fixed_url = urlunsplit((urlsplit(self.base_url).scheme, urlsplit(self.base_url).netloc, loc_path, loc.query, ""))
r = self.session.request(method, fixed_url, headers=headers, timeout=30)
if r.status_code == 401 and not _retried:
if self._refresh() or (self._login_password() or True):
return self.request(method, path, params=params, _retried=True)
r.raise_for_status()
return r.json()
def get(self, path: str, params: dict | None = None) -> dict:
return self.request("GET", path, params=params)
def companies(self) -> list:
return self.get("/companies").get("companies", [])
def dashboard_summary(self, company: str | None = None, luna: int | None = None, an: int | None = None) -> dict:
params = {"company": company or self.default_company}
if luna:
params["luna"] = luna
if an:
params["an"] = an
return self.get("/reports/dashboard/summary", params)
def resolve_company(client: ROA2WebClient, value: str | None) -> str:
"""Acceptă cod numeric sau (parte din) numele firmei; implicit firma din config."""
if not value:
return client.default_company
if value.isdigit():
return value
needle = value.lower()
matches = [c for c in client.companies() if needle in c["name"].lower()]
if not matches:
raise ROA2WebError(f"Firma '{value}' nu a fost gasita in lista de firme.")
if len(matches) > 1:
names = ", ".join(f"{c['id_firma']}={c['name']}" for c in matches)
raise ROA2WebError(f"'{value}' e ambiguu, se potriveste cu mai multe firme: {names}")
return str(matches[0]["id_firma"])
def main():
parser = argparse.ArgumentParser(description="Client CLI pentru roa2web API")
sub = parser.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("summary", help="Dashboard financiar (trezorerie, clienti, furnizori)")
p.add_argument("--company", help="Cod sau nume firma (implicit: cea din config.json)")
p.add_argument("--luna", type=int)
p.add_argument("--an", type=int)
sub.add_parser("companies", help="Lista firmelor disponibile")
p = sub.add_parser("get", help="Query generic pe orice endpoint /api/...")
p.add_argument("path", help='ex: "/reports/invoices" sau "/reports/treasury/bank-cash-register"')
p.add_argument("--company", help="Daca e dat, se adauga ca param 'company'")
p.add_argument("--param", action="append", default=[], metavar="k=v", help="Parametru query, repetabil")
v = sub.add_parser("verify-2fa", help="Finalizeaza login-ul cu un cod OTP primit pe email")
v.add_argument("code")
v.add_argument("email")
args = parser.parse_args()
client = ROA2WebClient()
try:
if args.cmd == "summary":
company = resolve_company(client, args.company)
data = client.dashboard_summary(company=company, luna=args.luna, an=args.an)
elif args.cmd == "companies":
for c in client.companies():
print(f"{c['id_firma']}\t{c['name']}")
return
elif args.cmd == "get":
params = dict(kv.split("=", 1) for kv in args.param)
if args.company:
params["company"] = resolve_company(client, args.company)
data = client.get(args.path, params or None)
elif args.cmd == "verify-2fa":
raw = client.verify_2fa(args.code, args.email)
data = {k: v for k, v in raw.items() if k not in ("access_token", "refresh_token", "trusted_device_token")}
except ROA2WebError as e:
print(f"Eroare: {e}", file=sys.stderr)
sys.exit(1)
print(json.dumps(data, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()