script de plan de pago contado con ajustes para el uso de la cuenta de servicio
This commit is contained in:
@@ -25,9 +25,6 @@ for k in list(sys.modules.keys()):
|
|||||||
from googleapiclient.discovery import build
|
from googleapiclient.discovery import build
|
||||||
from googleapiclient.errors import HttpError
|
from googleapiclient.errors import HttpError
|
||||||
from google.oauth2 import service_account
|
from google.oauth2 import service_account
|
||||||
from google.oauth2.credentials import Credentials
|
|
||||||
from google_auth_oauthlib.flow import InstalledAppFlow
|
|
||||||
from google.auth.transport.requests import Request
|
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
# Rocketbot vars helpers
|
# Rocketbot vars helpers
|
||||||
@@ -338,36 +335,23 @@ def _load_json(path): #Función auxiliar para cargar un archivo JSON
|
|||||||
return json.load(f)
|
return json.load(f)
|
||||||
|
|
||||||
|
|
||||||
def get_services(credentials_json_path, token_json_path): #Función auxiliar para obtener los servicios de Google
|
def get_services(credentials_json_path, impersonated_user): #Funci?n auxiliar para obtener los servicios de Google
|
||||||
info = _load_json(credentials_json_path) # carga el archivo JSON
|
info = _load_json(credentials_json_path) # carga el archivo JSON
|
||||||
|
if not isinstance(info, dict) or info.get("type") != "service_account":
|
||||||
|
raise RuntimeError("gdoc_sa_json2 debe apuntar a un JSON de cuenta de servicio.")
|
||||||
|
|
||||||
if isinstance(info, dict) and info.get("type") == "service_account": # si el archivo JSON es un diccionario y tiene el tipo "service_account"
|
impersonated_user = (impersonated_user or "").strip()
|
||||||
creds = service_account.Credentials.from_service_account_file(credentials_json_path, scopes=SCOPES) # crea las credenciales
|
if not impersonated_user:
|
||||||
docs = build("docs", "v1", credentials=creds, cache_discovery=False) # crea el cliente de Google Docs
|
raise RuntimeError("Falta la variable gdoc_impersonated_user.")
|
||||||
drive = build("drive", "v3", credentials=creds, cache_discovery=False) # crea el cliente de Google Drive
|
|
||||||
return docs, drive, "service_account" # devuelve los clientes y el tipo de cuenta
|
|
||||||
|
|
||||||
creds = None # siempre devuelve None si el valor es None
|
creds = service_account.Credentials.from_service_account_file(
|
||||||
if os.path.exists(token_json_path):
|
credentials_json_path,
|
||||||
creds = Credentials.from_authorized_user_file(token_json_path, SCOPES)
|
scopes=SCOPES,
|
||||||
|
subject=impersonated_user,
|
||||||
if (not creds) or (not creds.valid):
|
) # crea las credenciales delegadas
|
||||||
if creds and creds.expired and creds.refresh_token:
|
docs = build("docs", "v1", credentials=creds, cache_discovery=False) # crea el cliente de Google Docs
|
||||||
creds.refresh(Request())
|
drive = build("drive", "v3", credentials=creds, cache_discovery=False) # crea el cliente de Google Drive
|
||||||
else:
|
return docs, drive, "service_account_impersonated" # devuelve los clientes y el tipo de cuenta
|
||||||
flow = InstalledAppFlow.from_client_secrets_file(credentials_json_path, SCOPES)
|
|
||||||
try:
|
|
||||||
creds = flow.run_local_server(port=0)
|
|
||||||
except Exception:
|
|
||||||
creds = flow.run_console()
|
|
||||||
|
|
||||||
os.makedirs(os.path.dirname(token_json_path), exist_ok=True)
|
|
||||||
with open(token_json_path, "w", encoding="utf-8") as f:
|
|
||||||
f.write(creds.to_json())
|
|
||||||
|
|
||||||
docs = build("docs", "v1", credentials=creds, cache_discovery=False)
|
|
||||||
drive = build("drive", "v3", credentials=creds, cache_discovery=False)
|
|
||||||
return docs, drive, "oauth"
|
|
||||||
|
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
@@ -1239,24 +1223,20 @@ try:
|
|||||||
marker = _gvs("gdoc_marker", "TABLA_PROFORMA")
|
marker = _gvs("gdoc_marker", "TABLA_PROFORMA")
|
||||||
_sv("dbg_marker_var", marker)
|
_sv("dbg_marker_var", marker)
|
||||||
|
|
||||||
cred_path = _gvs("gdoc_sa_json", "")
|
cred_path = _gvs("gdoc_sa_json2", "")
|
||||||
if cred_path == "":
|
if cred_path == "":
|
||||||
raise RuntimeError("Falta gdoc_sa_json (ruta a credentials.json)")
|
raise RuntimeError("Falta gdoc_sa_json2 (ruta al JSON de cuenta de servicio)")
|
||||||
if not os.path.isabs(cred_path):
|
if not os.path.isabs(cred_path):
|
||||||
cred_path = os.path.join(base_dir, cred_path)
|
cred_path = os.path.join(base_dir, cred_path)
|
||||||
if not os.path.exists(cred_path):
|
if not os.path.exists(cred_path):
|
||||||
raise RuntimeError("No existe credentials.json: " + cred_path)
|
raise RuntimeError("No existe JSON de cuenta de servicio: " + cred_path)
|
||||||
|
|
||||||
token_path = _gvs("gdoc_token_json", "")
|
impersonated_user = _gvs("gdoc_impersonated_user", "")
|
||||||
if token_path == "":
|
|
||||||
token_path = os.path.join(base_dir, "credentials", "token_gdocs_drive.json")
|
|
||||||
if not os.path.isabs(token_path):
|
|
||||||
token_path = os.path.join(base_dir, token_path)
|
|
||||||
|
|
||||||
_sv("gdoc_credentials_used", cred_path)
|
_sv("gdoc_credentials_used", cred_path)
|
||||||
_sv("gdoc_token_used", token_path)
|
_sv("gdoc_impersonated_user_used", impersonated_user)
|
||||||
|
|
||||||
docs_service, drive_service, auth_mode = get_services(cred_path, token_path)
|
docs_service, drive_service, auth_mode = get_services(cred_path, impersonated_user)
|
||||||
_sv("gdoc_auth_mode", auth_mode)
|
_sv("gdoc_auth_mode", auth_mode)
|
||||||
|
|
||||||
url = _gvs("gdoc_url", "")
|
url = _gvs("gdoc_url", "")
|
||||||
|
|||||||
Reference in New Issue
Block a user