152 lines
4.9 KiB
Python
152 lines
4.9 KiB
Python
from google.oauth2 import service_account
|
|
from googleapiclient.discovery import build
|
|
import os
|
|
import sys
|
|
import json
|
|
import re
|
|
|
|
global Request, InstalledAppFlow, Credentials, service_account, build, os, sys, json, LISTA_BUSCAR_REEMPLAZAR, base_dir, libs_dir, SCOPES, CREDENTIALS_PATH, TOKEN_PATH, _load_json, get_google_services, replace_vars_doc, drive_service, docs_service, gdoc_id, lista_oficial_reemplazar, success
|
|
|
|
# Lista oficial de variables a reemplazar
|
|
LISTA_BUSCAR_REEMPLAZAR = {list_buscar}
|
|
|
|
|
|
# ===== Rocketbot-compatible py_libs path =====
|
|
base_dir = os.path.dirname(sys.executable) if getattr(
|
|
sys, 'frozen', False) else os.getcwd()
|
|
libs_dir = os.path.join(base_dir, 'librerias', 'py_libs', 'py310')
|
|
if os.path.isdir(libs_dir) and libs_dir not in sys.path:
|
|
sys.path.insert(0, libs_dir)
|
|
|
|
|
|
# SCOPES actualizados (solo Drive y Docs)
|
|
SCOPES = {scopes_api_google}
|
|
|
|
|
|
CREDENTIALS_PATH = '{gdoc_sa_json2}'
|
|
IMPERSONATED_USER = '{gdoc_impersonated_user}'
|
|
|
|
|
|
def rb_get_var(name, required=True, default=''):
|
|
try:
|
|
value = GetVar(name)
|
|
except NameError:
|
|
value = os.environ.get(name, default)
|
|
|
|
if value is None:
|
|
value = ''
|
|
|
|
value = str(value).strip()
|
|
if value == f'{{{name}}}':
|
|
value = ''
|
|
|
|
if required and not value:
|
|
raise RuntimeError(f'La variable Rocketbot "{name}" está vacía o no fue leída correctamente.')
|
|
|
|
return value
|
|
|
|
|
|
def extract_doc_id_from_url(url):
|
|
match = re.search(r"/document/d/([a-zA-Z0-9_-]+)", url or "")
|
|
return match.group(1) if match else ""
|
|
|
|
|
|
def _load_json(path):
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
|
|
def get_google_services(credentials_json_path=CREDENTIALS_PATH, impersonated_user=IMPERSONATED_USER):
|
|
info = _load_json(credentials_json_path)
|
|
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.')
|
|
|
|
impersonated_user = (impersonated_user or '').strip()
|
|
if not impersonated_user:
|
|
raise RuntimeError('Falta la variable gdoc_impersonated_user.')
|
|
|
|
creds = service_account.Credentials.from_service_account_file(
|
|
credentials_json_path, scopes=SCOPES, subject=impersonated_user)
|
|
drive_service = build(
|
|
'drive', 'v3', credentials=creds, cache_discovery=False)
|
|
docs_service = build(
|
|
'docs', 'v1', credentials=creds, cache_discovery=False)
|
|
|
|
return drive_service, docs_service
|
|
|
|
|
|
def replace_vars_doc(docs_service, gdoc_id, replacements_values):
|
|
"""
|
|
Reemplaza las variables en el documento de Google Docs.
|
|
Recibe el gdoc_id y un array de valores en el mismo orden que LISTA_BUSCAR_REEMPLAZAR.
|
|
"""
|
|
if not replacements_values:
|
|
print("No se proporcionaron valores para reemplazar.")
|
|
return False
|
|
|
|
requests = []
|
|
|
|
# Iterar sobre las variables oficiales y construir la lista de reemplazo
|
|
for i in range(min(len(LISTA_BUSCAR_REEMPLAZAR), len(replacements_values))):
|
|
key = LISTA_BUSCAR_REEMPLAZAR[i]
|
|
value = str(replacements_values[i]) if replacements_values[i] is not None else ""
|
|
print('Buscando variable a remplazar:', key)
|
|
print('Remplazando por:', value)
|
|
|
|
requests.append({
|
|
'replaceAllText': {
|
|
'containsText': {
|
|
'text': key,
|
|
'matchCase': True
|
|
},
|
|
'replaceText': value
|
|
}
|
|
})
|
|
|
|
if requests:
|
|
print(f"Enviando {len(requests)} solicitudes de reemplazo al documento {gdoc_id}")
|
|
docs_service.documents().batchUpdate(
|
|
documentId=gdoc_id, body={'requests': requests}).execute()
|
|
print("Reemplazo completado exitosamente.")
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
# Iniciar servicios (Solo Drive y Docs)
|
|
drive_service, docs_service = get_google_services(CREDENTIALS_PATH, IMPERSONATED_USER)
|
|
|
|
# Variables de entrada (esto usualmente viene de Rocketbot via GetVar)
|
|
current_url = rb_get_var('current_url')
|
|
gdoc_id = extract_doc_id_from_url(current_url)
|
|
if not gdoc_id:
|
|
raise RuntimeError('No pude extraer documentId desde current_url: ' + current_url)
|
|
lista_oficial_reemplazar = {lista_oficial_reemplazar}
|
|
|
|
|
|
try:
|
|
try:
|
|
SetVar('gdoc_id', gdoc_id)
|
|
SetVar('gdoc_url_used', current_url)
|
|
except NameError:
|
|
pass
|
|
|
|
# Ejecutar reemplazo si se cuenta con la información
|
|
if gdoc_id and lista_oficial_reemplazar:
|
|
success = replace_vars_doc(docs_service, gdoc_id, lista_oficial_reemplazar)
|
|
try:
|
|
SetVar('res_reemplazo_variables', success)
|
|
except NameError:
|
|
pass
|
|
else:
|
|
print("Falta gdoc_id o lista_oficial_reemplazar para ejecutar el reemplazo.")
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
print(f"Error en el reemplazo de variables: {str(e)}")
|
|
print(traceback.format_exc())
|
|
try:
|
|
SetVar('error', {'error': 'Error al remplazar las variables en Google Docs', 'descripcion': str(e)})
|
|
except NameError:
|
|
pass
|