diff --git a/clausulas_durante_obra.py b/clausulas_durante_obra.py index b4484b2..5690738 100644 --- a/clausulas_durante_obra.py +++ b/clausulas_durante_obra.py @@ -26,9 +26,6 @@ for k in list(sys.modules.keys()): from googleapiclient.discovery import build from googleapiclient.errors import HttpError 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 as GRequest # ----------------------------------------------------------- @@ -42,39 +39,23 @@ def _load_json(path): return json.load(f) -def get_services(credentials_json_path, token_json_path): +def get_services(credentials_json_path, 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.") - if isinstance(info, dict) and info.get("type") == "service_account": - creds = service_account.Credentials.from_service_account_file( - credentials_json_path, - scopes=SCOPES, - ) - docs = build("docs", "v1", credentials=creds, cache_discovery=False) - drive = build("drive", "v3", credentials=creds, cache_discovery=False) - return docs, drive, "service_account" - - creds = None - if os.path.exists(token_json_path): - creds = Credentials.from_authorized_user_file(token_json_path, SCOPES) - - if (not creds) or (not creds.valid): - if creds and creds.expired and creds.refresh_token: - creds.refresh(GRequest()) - else: - 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()) + 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, + ) docs = build("docs", "v1", credentials=creds, cache_discovery=False) drive = build("drive", "v3", credentials=creds, cache_discovery=False) - return docs, drive, "oauth" + return docs, drive, "service_account_impersonated" # ----------------------------------------------------------- @@ -691,24 +672,21 @@ try: _sv("gdoc_original_id", raw_id) - cred_path = _gvs("gdoc_sa_json", "") + cred_path = _gvs("gdoc_sa_json2", "") if cred_path == "": - raise RuntimeError("Falta la variable gdoc_sa_json.") + raise RuntimeError("Falta la variable gdoc_sa_json2.") if not os.path.isabs(cred_path): cred_path = os.path.join(base_dir, 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", "") - 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) + impersonated_user = _gvs("gdoc_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_impersonated_user_used", impersonated_user) doc_id, meta_file, converted = ensure_docs_api_compatible(drive_service, raw_id) _sv("gdoc_id", doc_id) diff --git a/clausulas_financiado.py b/clausulas_financiado.py index f6550d6..a092b93 100644 --- a/clausulas_financiado.py +++ b/clausulas_financiado.py @@ -26,9 +26,6 @@ for k in list(sys.modules.keys()): from googleapiclient.discovery import build from googleapiclient.errors import HttpError 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 as GRequest # ----------------------------------------------------------- @@ -42,39 +39,23 @@ def _load_json(path): return json.load(f) -def get_services(credentials_json_path, token_json_path): +def get_services(credentials_json_path, 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.") - if isinstance(info, dict) and info.get("type") == "service_account": - creds = service_account.Credentials.from_service_account_file( - credentials_json_path, - scopes=SCOPES, - ) - docs = build("docs", "v1", credentials=creds, cache_discovery=False) - drive = build("drive", "v3", credentials=creds, cache_discovery=False) - return docs, drive, "service_account" - - creds = None - if os.path.exists(token_json_path): - creds = Credentials.from_authorized_user_file(token_json_path, SCOPES) - - if (not creds) or (not creds.valid): - if creds and creds.expired and creds.refresh_token: - creds.refresh(GRequest()) - else: - 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()) + 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, + ) docs = build("docs", "v1", credentials=creds, cache_discovery=False) drive = build("drive", "v3", credentials=creds, cache_discovery=False) - return docs, drive, "oauth" + return docs, drive, "service_account_impersonated" # ----------------------------------------------------------- @@ -691,24 +672,21 @@ try: _sv("gdoc_original_id", raw_id) - cred_path = _gvs("gdoc_sa_json", "") + cred_path = _gvs("gdoc_sa_json2", "") if cred_path == "": - raise RuntimeError("Falta la variable gdoc_sa_json.") + raise RuntimeError("Falta la variable gdoc_sa_json2.") if not os.path.isabs(cred_path): cred_path = os.path.join(base_dir, 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", "") - 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) + impersonated_user = _gvs("gdoc_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_impersonated_user_used", impersonated_user) doc_id, meta_file, converted = ensure_docs_api_compatible(drive_service, raw_id) _sv("gdoc_id", doc_id) diff --git a/plan_pagos_financiado_durante_obra.py b/plan_pagos_financiado_durante_obra.py index fc874fb..321de43 100644 --- a/plan_pagos_financiado_durante_obra.py +++ b/plan_pagos_financiado_durante_obra.py @@ -485,41 +485,26 @@ def run(): with open(path, "r", encoding="utf-8") as f: return json.load(f) - def get_services(credentials_json_path, token_json_path): + def get_services(credentials_json_path, impersonated_user): from googleapiclient.discovery import build 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 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.") - if isinstance(info, dict) and info.get("type") == "service_account": - creds = service_account.Credentials.from_service_account_file(credentials_json_path, scopes=SCOPES) - docs = build("docs", "v1", credentials=creds, cache_discovery=False) - drive = build("drive", "v3", credentials=creds, cache_discovery=False) - return docs, drive, "service_account" - - creds = None - if os.path.exists(token_json_path): - creds = Credentials.from_authorized_user_file(token_json_path, SCOPES) - - if (not creds) or (not creds.valid): - if creds and creds.expired and creds.refresh_token: - creds.refresh(Request()) - else: - 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()) + 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, + ) docs = build("docs", "v1", credentials=creds, cache_discovery=False) drive = build("drive", "v3", credentials=creds, cache_discovery=False) - return docs, drive, "oauth" + return docs, drive, "service_account_impersonated" def extract_doc_id_from_url(url): m = re.search(r"/document/d/([a-zA-Z0-9_-]+)", url or "") @@ -1354,21 +1339,18 @@ def run(): # ============================================================ marker = _gvs("gdoc_marker", "TABLA_PROFORMA") - cred_path = _gvs("gdoc_sa_json", "") + cred_path = _gvs("gdoc_sa_json2", "") if cred_path == "": - raise RuntimeError("Falta gdoc_sa_json") + raise RuntimeError("Falta gdoc_sa_json2") if not os.path.isabs(cred_path): cred_path = os.path.join(base_dir, 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", "") - 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) + impersonated_user = _gvs("gdoc_impersonated_user", "") - docs_service, drive_service, _ = get_services(cred_path, token_path) + docs_service, drive_service, _ = get_services(cred_path, impersonated_user) + _sv("gdoc_impersonated_user_used", impersonated_user) url = _gvs("current_url", "") if url == "": diff --git a/reemplazar_variables_contrato.py b/reemplazar_variables_contrato.py index 9003d90..823d4c2 100644 --- a/reemplazar_variables_contrato.py +++ b/reemplazar_variables_contrato.py @@ -1,6 +1,3 @@ -from google.auth.transport.requests import Request -from google_auth_oauthlib.flow import InstalledAppFlow -from google.oauth2.credentials import Credentials from google.oauth2 import service_account from googleapiclient.discovery import build import os @@ -25,8 +22,8 @@ if os.path.isdir(libs_dir) and libs_dir not in sys.path: SCOPES = {scopes_api_google} -CREDENTIALS_PATH = '{gdoc_sa_json}' -TOKEN_PATH = '{gdoc_token_json}' +CREDENTIALS_PATH = '{gdoc_sa_json2}' +IMPERSONATED_USER = '{gdoc_impersonated_user}' def _load_json(path): @@ -34,39 +31,17 @@ def _load_json(path): return json.load(f) -def get_google_services(credentials_json_path=CREDENTIALS_PATH, token_json_path=TOKEN_PATH): +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.') - # Service account - if isinstance(info, dict) and info.get('type') == 'service_account': - creds = service_account.Credentials.from_service_account_file( - credentials_json_path, scopes=SCOPES) - 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 - - # OAuth flow - creds = None - if os.path.exists(token_json_path): - creds = Credentials.from_authorized_user_file(token_json_path, SCOPES) - - if not creds or not creds.valid: - if creds and creds.expired and creds.refresh_token: - creds.refresh(Request()) - else: - 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()) + 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( @@ -114,7 +89,7 @@ def replace_vars_doc(docs_service, gdoc_id, replacements_values): # Iniciar servicios (Solo Drive y Docs) -drive_service, docs_service = get_google_services(CREDENTIALS_PATH, TOKEN_PATH) +drive_service, docs_service = get_google_services(CREDENTIALS_PATH, IMPERSONATED_USER) # Variables de entrada (esto usualmente viene de Rocketbot via GetVar) gdoc_id = '{gdoc_id}'