Script parrafos_dinamicos_forma_pago.py ajustado para cuenta de servicio.
This commit is contained in:
mabejoyb
@@ -9,23 +9,41 @@ import unicodedata
|
||||
import datetime as dt
|
||||
from decimal import Decimal, InvalidOperation
|
||||
|
||||
def _rb_set_early(name, value):
|
||||
try:
|
||||
SetVar(name, value)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# --------------------
|
||||
# Boot libs (Rocketbot portable)
|
||||
# --------------------
|
||||
base_dir = os.path.dirname(sys.executable)
|
||||
libs_dir = os.path.join(base_dir, "py_libs", "py310")
|
||||
sys.path.insert(0, libs_dir)
|
||||
_rb_set_early("gdoc_textos_status", "BOOTING")
|
||||
|
||||
for k in list(sys.modules.keys()):
|
||||
try:
|
||||
base_dir = os.path.dirname(sys.executable)
|
||||
libs_candidates = [
|
||||
os.path.join(base_dir, "py_libs", "py310"),
|
||||
os.path.join(base_dir, "librerias", "py_libs", "py310"),
|
||||
os.path.join(os.getcwd(), "librerias", "py_libs", "py310"),
|
||||
]
|
||||
for libs_dir in libs_candidates:
|
||||
if os.path.isdir(libs_dir) and libs_dir not in sys.path:
|
||||
sys.path.insert(0, libs_dir)
|
||||
|
||||
for k in list(sys.modules.keys()):
|
||||
if k == "pyparsing" or k.startswith("pyparsing."):
|
||||
del sys.modules[k]
|
||||
|
||||
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
|
||||
from googleapiclient.discovery import build
|
||||
from googleapiclient.errors import HttpError
|
||||
from google.oauth2 import service_account
|
||||
except Exception as e:
|
||||
_rb_set_early("gdoc_textos_status", "ERROR")
|
||||
_rb_set_early("gdoc_textos_error", "BOOT_IMPORT: " + repr(e))
|
||||
print("BOOT_IMPORT error:", repr(e))
|
||||
raise
|
||||
|
||||
|
||||
# -----------------------------------------------------------
|
||||
@@ -474,46 +492,70 @@ def _build_refuerzos_post_paragraphs(refuerzos_post, sep_line="-----------------
|
||||
# -----------------------------------------------------------
|
||||
# GOOGLE AUTH
|
||||
# -----------------------------------------------------------
|
||||
SCOPES = {scopes_api_google}
|
||||
def _resolve_google_scopes():
|
||||
default_scopes = [
|
||||
"https://www.googleapis.com/auth/documents",
|
||||
"https://www.googleapis.com/auth/drive",
|
||||
]
|
||||
|
||||
try:
|
||||
raw = scopes_api_google
|
||||
except NameError:
|
||||
raw = _gv("scopes_api_google", None)
|
||||
|
||||
if _missing(raw):
|
||||
return default_scopes
|
||||
|
||||
if isinstance(raw, (list, tuple, set)):
|
||||
scopes = [str(x).strip() for x in raw if str(x).strip()]
|
||||
return scopes or default_scopes
|
||||
|
||||
s = str(raw).strip()
|
||||
if not s:
|
||||
return default_scopes
|
||||
|
||||
try:
|
||||
parsed = json.loads(s)
|
||||
except Exception:
|
||||
try:
|
||||
parsed = ast.literal_eval(s)
|
||||
except Exception:
|
||||
parsed = None
|
||||
|
||||
if isinstance(parsed, (list, tuple, set)):
|
||||
scopes = [str(x).strip() for x in parsed if str(x).strip()]
|
||||
return scopes or default_scopes
|
||||
|
||||
if "," in s:
|
||||
scopes = [x.strip().strip("'\"") for x in s.split(",") if x.strip()]
|
||||
return scopes or default_scopes
|
||||
|
||||
return [s.strip("'\"")]
|
||||
|
||||
|
||||
SCOPES = _resolve_google_scopes()
|
||||
|
||||
def _load_json(path):
|
||||
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):
|
||||
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.")
|
||||
|
||||
if isinstance(info, dict) and info.get("type") == "service_account":
|
||||
creds = service_account.Credentials.from_service_account_file(
|
||||
credentials_json_path,
|
||||
scopes=SCOPES
|
||||
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, "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())
|
||||
|
||||
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"
|
||||
|
||||
# -----------------------------------------------------------
|
||||
# DRIVE / DOC HELPERS
|
||||
@@ -892,6 +934,10 @@ globals().update(locals())
|
||||
# MAIN
|
||||
# -----------------------------------------------------------
|
||||
try:
|
||||
_sv("gdoc_textos_status", "STARTED")
|
||||
_sv("gdoc_textos_error", "")
|
||||
print("parrafos_dinamicos_forma_pago.py STARTED")
|
||||
|
||||
NL = "\n"
|
||||
|
||||
# usar SOLO current_url
|
||||
@@ -909,23 +955,20 @@ 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_impersonated_user_used", impersonated_user)
|
||||
_sv("gdoc_auth_mode", auth_mode)
|
||||
|
||||
doc_id, meta_file, converted = ensure_docs_api_compatible(drive_service, raw_id)
|
||||
@@ -1790,9 +1833,12 @@ try:
|
||||
except HttpError as e:
|
||||
_sv("gdoc_textos_status", "ERROR")
|
||||
_sv("gdoc_textos_error", "HttpError: " + str(e))
|
||||
_sv("gdoc_textos_traceback", "HttpError: " + str(e))
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
import traceback
|
||||
_sv("gdoc_textos_status", "ERROR")
|
||||
_sv("gdoc_textos_error", str(e))
|
||||
_sv("gdoc_textos_traceback", traceback.format_exc())
|
||||
raise
|
||||
Reference in New Issue
Block a user