agrego el scriot encargado de crear la copia de la plantilla del contrato mediante el uso de la API de google
This commit is contained in:
170
duplicar_plantilla_contrato.py
Normal file
170
duplicar_plantilla_contrato.py
Normal file
@@ -0,0 +1,170 @@
|
||||
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
|
||||
import sys
|
||||
import json
|
||||
import re
|
||||
|
||||
global Request, InstalledAppFlow, Credentials, service_account, build, os, sys, json, re, base_dir, libs_dir, SCOPES, CREDENTIALS_PATH, TOKEN_PATH, _load_json, get_drive_service, extract_folder_id_from_url, get_doc_name, copy_doc_to_folder, get_doc_url, TEMPLATE_DOC_ID, FOLDER_URL, NEW_DOC_NAME, drive_service, folder_id, copied_doc, doc_url
|
||||
|
||||
# ===== 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 = {scopes_api_google}
|
||||
|
||||
CREDENTIALS_PATH = '{gdoc_sa_json}'
|
||||
TOKEN_PATH = '{gmail_token_json}'
|
||||
|
||||
|
||||
def _load_json(path):
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def get_drive_service(credentials_json_path=CREDENTIALS_PATH, token_json_path=TOKEN_PATH):
|
||||
info = _load_json(credentials_json_path)
|
||||
|
||||
# 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)
|
||||
return drive_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())
|
||||
|
||||
drive_service = build(
|
||||
'drive', 'v3', credentials=creds, cache_discovery=False)
|
||||
return drive_service
|
||||
|
||||
|
||||
def extract_folder_id_from_url(folder_url):
|
||||
"""Extrae el ID de una carpeta desde una URL de Google Drive"""
|
||||
match = re.search(r'/folders/([a-zA-Z0-9_-]+)', folder_url)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return folder_url # Si ya es un ID, lo retorna como está
|
||||
|
||||
|
||||
def get_doc_name(drive_service, doc_id):
|
||||
"""Obtiene el nombre del documento de Google Drive"""
|
||||
file_metadata = drive_service.files().get(
|
||||
fileId=doc_id, fields='name'
|
||||
).execute()
|
||||
return file_metadata.get('name', 'Documento')
|
||||
|
||||
|
||||
def copy_doc_to_folder(drive_service, template_doc_id, folder_id, new_doc_name=None):
|
||||
"""
|
||||
Copia un documento a una carpeta específica
|
||||
|
||||
Args:
|
||||
drive_service: Servicio de Drive autenticado
|
||||
template_doc_id: ID del documento plantilla a copiar
|
||||
folder_id: ID de la carpeta destino
|
||||
new_doc_name: Nombre opcional para el documento copiado
|
||||
|
||||
Returns:
|
||||
dict con 'id' y 'webViewLink' del documento creado
|
||||
"""
|
||||
if new_doc_name is None:
|
||||
new_doc_name = get_doc_name(
|
||||
drive_service, template_doc_id) + ' (copia)'
|
||||
|
||||
# Copiar el documento
|
||||
body = {
|
||||
'name': new_doc_name,
|
||||
'parents': [folder_id]
|
||||
}
|
||||
|
||||
copied_file = drive_service.files().copy(
|
||||
fileId=template_doc_id,
|
||||
body=body,
|
||||
fields='id, webViewLink, name'
|
||||
).execute()
|
||||
|
||||
return copied_file
|
||||
|
||||
|
||||
def get_doc_url(doc_id):
|
||||
"""Genera la URL de edición del documento en Google Drive"""
|
||||
return f"https://docs.google.com/document/d/{doc_id}/edit"
|
||||
|
||||
|
||||
# IDs y URLs
|
||||
TEMPLATE_DOC_ID = '{template_contrato_id}'
|
||||
FOLDER_URL = '{drive_contratos_carpeta}'
|
||||
# Opcional
|
||||
NEW_DOC_NAME = '{proyecto} - {numeracion_dpto} - {nombre_comprador}' # Nombre del contrato
|
||||
|
||||
|
||||
try:
|
||||
if not os.path.exists(CREDENTIALS_PATH):
|
||||
raise FileNotFoundError(
|
||||
f'No existe el archivo de credenciales: {CREDENTIALS_PATH}')
|
||||
|
||||
# Inicializar servicio
|
||||
drive_service = get_drive_service(CREDENTIALS_PATH, TOKEN_PATH)
|
||||
|
||||
# Extraer ID de carpeta desde URL
|
||||
folder_id = extract_folder_id_from_url(FOLDER_URL)
|
||||
|
||||
# Copiar documento
|
||||
copied_doc = copy_doc_to_folder(
|
||||
drive_service, TEMPLATE_DOC_ID, folder_id, NEW_DOC_NAME)
|
||||
|
||||
# Generar URL del documento creado
|
||||
doc_url = get_doc_url(copied_doc['id'])
|
||||
|
||||
print(f'Documento copiado exitosamente')
|
||||
print(f'Nombre: {copied_doc.get("name")}')
|
||||
print(f'ID: {copied_doc.get("id")}')
|
||||
print(f'URL: {doc_url}')
|
||||
|
||||
# Retornar valores (Rocketbot compatible)
|
||||
try:
|
||||
SetVar('current_url', doc_url)
|
||||
SetVar('doc_copy_created', True)
|
||||
except NameError:
|
||||
pass # Si no está en Rocketbot, solo imprime
|
||||
|
||||
except Exception as e:
|
||||
import traceback
|
||||
error = {
|
||||
'error': str(e),
|
||||
'descripcion': traceback.format_exc()
|
||||
}
|
||||
try:
|
||||
SetVar('error', error)
|
||||
SetVar('doc_copy_created', False)
|
||||
except NameError:
|
||||
pass
|
||||
print(f"Error detectado: {str(e)}")
|
||||
Reference in New Issue
Block a user