import os import requests import json import time import csv from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from webdriver_manager.chrome import ChromeDriverManager from selenium.common.exceptions import TimeoutException, NoSuchElementException from collections import defaultdict from datetime import datetime, timedelta import webbrowser import re import threading import glob from http.server import HTTPServer, BaseHTTPRequestHandler import urllib.parse # ⬇️ NOVO: Import do stock.py (ficheiro na mesma pasta ou Desktop) import sys sys.path.append(os.path.expanduser("~/Desktop")) import stock import comentarios #modulo de comentarios # Import defensivo do motor de pesquisa visual # (se falhar, o Dimo.py continua a funcionar sem o HTML do motor) try: import motor_pesquisa_visual _MOTOR_DISPONIVEL = True except Exception as e: print(f"⚠️ Aviso: motor_pesquisa_visual.py não pôde ser importado: {e}") print(" O Dimo.py vai continuar, mas o motor de pesquisa NÃO será regenerado.") motor_pesquisa_visual = None _MOTOR_DISPONIVEL = False # ============================================================================ # SERVIDOR PROXY PARA RESOLVER CORS - CORRIGIDO # ============================================================================ # ============================================================================ # SERVIDOR PROXY PARA RESOLVER CORS - COM RENOVAÇÃO AUTOMÁTICA DE TOKEN # ============================================================================ class ProxyHandler(BaseHTTPRequestHandler): """Handler para o servidor proxy que encaminha requisições para a API""" # Variáveis de classe token = None api_instance = None # Referência à instância da API para renovar token username = None password = None def do_GET(self): """Processa requisições GET""" parsed_path = urllib.parse.urlparse(self.path) # Verificar se é uma requisição para a API if parsed_path.path.startswith('/api/workOrderReader/Get'): # Extrair o ID da query string query_params = urllib.parse.parse_qs(parsed_path.query) ot_id = query_params.get('id', [''])[0] if ot_id: self._proxy_request_with_retry(ot_id) else: self._send_error(400, "ID da OT não fornecido") else: self._send_error(404, "Endpoint não encontrado") def _proxy_request_with_retry(self, ot_id, max_retries=2): """ Tenta fazer a requisição, renovando o token se necessário """ for attempt in range(max_retries + 1): try: # Obter o token token = ProxyHandler.token if not token: print("❌ Proxy: Token não disponível. A renovar...") if self._renovar_token(): token = ProxyHandler.token else: self._send_error(401, "Não foi possível obter token.") return url = f"https://prod.mydimomaintmx.cloud/SERRATEC/api/workOrderReader/Get?id={ot_id}" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "tenant": "SERRATEC" } print(f"📤 Proxy: A buscar detalhes da OT {ot_id} (tentativa {attempt + 1})") response = requests.get(url, headers=headers, timeout=30) # Se for 401 e ainda houver tentativas, renovar token e tentar novamente if response.status_code == 401 and attempt < max_retries: print(f"⚠️ Proxy: Token expirado (401). A renovar...") if self._renovar_token(): print("✅ Proxy: Token renovado. A tentar novamente...") continue # Tentar novamente com o novo token else: print("❌ Proxy: Falha ao renovar token.") self._send_error(401, "Token expirado e não foi possível renovar.") return # Sucesso ou erro final self.send_response(response.status_code) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization') self.end_headers() self.wfile.write(response.content) print(f"✅ Proxy: Resposta enviada (status {response.status_code})") return except requests.exceptions.Timeout: if attempt < max_retries: print(f"⚠️ Proxy: Timeout, a tentar novamente...") continue self._send_error(504, "Timeout ao contactar a API") return except requests.exceptions.ConnectionError: if attempt < max_retries: print(f"⚠️ Proxy: Erro de ligação, a tentar novamente...") continue self._send_error(502, "Erro de ligação à API") return except Exception as e: if attempt < max_retries: print(f"⚠️ Proxy: Erro {e}, a tentar novamente...") continue self._send_error(500, f"Erro interno: {str(e)}") return # Se chegou aqui, todas as tentativas falharam self._send_error(500, "Todas as tentativas falharam.") def _renovar_token(self): """ Renova o token usando Selenium """ try: api = ProxyHandler.api_instance username = ProxyHandler.username password = ProxyHandler.password if not api or not username or not password: print("❌ Proxy: Credenciais não disponíveis para renovar token.") return False print("🔑 Proxy: A renovar token com Selenium...") # Forçar novo login if api.login_with_selenium(username, password, headless=True, force=True): token = api.token if token: ProxyHandler.token = token print(f"✅ Proxy: Token renovado com sucesso: {token[:30]}...") return True else: print("❌ Proxy: Token não foi obtido após login.") return False else: print("❌ Proxy: Falha no login para renovar token.") return False except Exception as e: print(f"❌ Proxy: Erro ao renovar token: {e}") return False def do_OPTIONS(self): """Responde a requisições OPTIONS (preflight CORS)""" self.send_response(200) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization') self.end_headers() def _send_error(self, code, message): """Envia uma resposta de erro""" self.send_response(code) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() error_response = json.dumps({"error": message, "code": code}).encode('utf-8') self.wfile.write(error_response) print(f"❌ Proxy: Erro {code} - {message}") def log_message(self, format, *args): """Log simplificado""" print(f"📡 Proxy: {format % args}") def iniciar_servidor_proxy(porta=8888): """Inicia o servidor proxy numa thread separada""" try: server = HTTPServer(('localhost', porta), ProxyHandler) print(f"🚀 Servidor proxy iniciado em http://localhost:{porta}") print(f"📡 Endpoint: http://localhost:{porta}/api/workOrderReader/Get?id=OT_ID") print("=" * 60) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() return server except OSError as e: if "Address already in use" in str(e): print(f"❌ Porta {porta} já está em uso. Tente outra porta.") print(" Exemplo: iniciar_servidor_proxy(porta=8889)") raise def configurar_proxy(api_instance, username, password): """Configura o proxy com a instância da API e credenciais""" ProxyHandler.api_instance = api_instance ProxyHandler.username = username ProxyHandler.password = password if api_instance and api_instance.token: ProxyHandler.token = api_instance.token print(f"✅ Proxy configurado com token: {api_instance.token[:30]}...") else: print("⚠️ Proxy configurado mas sem token disponível.") # ============================================================================ # CLASSE PRINCIPAL DimoMaintAPI # ============================================================================ class DimoMaintAPI: def __init__(self, base_url="https://prod.mydimomaintmx.cloud"): self.base_url = base_url self.session = requests.Session() self.token = None self.token_file = os.path.join(os.path.expanduser("~/Desktop"), "dimo_token.json") self.driver = None self.estados_cache = {} self.estados_workrequest_cache = {} self.token_expiry = None def _load_token(self): """Carrega o token do ficheiro""" try: if os.path.exists(self.token_file): with open(self.token_file, 'r', encoding='utf-8') as f: data = json.load(f) self.token = data.get('token') self.token_expiry = data.get('expiry') if self.token and self.token_expiry: expiry = datetime.fromisoformat(self.token_expiry) if datetime.now() < expiry - timedelta(minutes=5): print(f"✅ Token carregado do ficheiro (válido até {self.token_expiry})") self.session.cookies.set('MX-Access-Token', self.token) return True else: print("⚠️ Token expirado. A renovar...") return False return False except Exception as e: print(f"⚠️ Erro ao carregar token: {e}") return False def _save_token(self): """Guarda o token no ficheiro""" try: expiry = datetime.now() + timedelta(hours=1) data = { 'token': self.token, 'expiry': expiry.isoformat(), 'timestamp': datetime.now().isoformat() } with open(self.token_file, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"✅ Token guardado (válido até {expiry.isoformat()})") return True except Exception as e: print(f"⚠️ Erro ao guardar token: {e}") return False def login_with_selenium(self, username, password, headless=False, force=False): """Faz login usando Selenium, reutilizando token se possível""" if not force and self._load_token(): return True print("🌐 A iniciar navegador para login...") options = Options() if headless: print("🔇 Modo headless (sem janela visível)") options.add_argument("--headless=new") else: print("🪟 Modo visível (janela aberta)") options.add_argument("--disable-blink-features=AutomationControlled") options.add_experimental_option("excludeSwitches", ["enable-automation"]) options.add_experimental_option('useAutomationExtension', False) options.add_argument("--window-size=1280,720") options.add_argument("--no-sandbox") options.add_argument("--disable-dev-shm-usage") options.add_argument("--disable-gpu") options.add_argument("--disable-logging") options.add_argument("--log-level=3") options.add_argument("--ignore-certificate-errors") options.add_argument("--ignore-ssl-errors") try: service = Service(ChromeDriverManager().install()) self.driver = webdriver.Chrome(service=service, options=options) self.driver.implicitly_wait(10) print("📄 A carregar página de login...") self.driver.get(f"{self.base_url}/SERRATEC/account") time.sleep(3) if self._check_already_logged_in(): print("✅ Já estás logado!") return self._extract_token_from_browser() print("🔍 A preencher credenciais...") time.sleep(2) username_field = self._find_element([ "input[name='username']", "input[name='email']", "input[type='text']", "#username", "#email" ]) if username_field: username_field.clear() username_field.send_keys(username) print("✅ Username preenchido") else: print("❌ Campo de username não encontrado") self.driver.quit() return False password_field = self._find_element([ "input[name='password']", "input[type='password']", "#password" ]) if password_field: password_field.clear() password_field.send_keys(password) print("✅ Password preenchida") else: print("❌ Campo de password não encontrado") self.driver.quit() return False print("🔍 A submeter login...") login_button = self._find_element([ "button[type='submit']", "input[type='submit']", "button.btn-primary", "#login-button" ]) if login_button: login_button.click() print("✅ Login submetido (botão)") else: password_field.send_keys("\n") print("✅ Login submetido (Enter)") print("⏳ A aguardar resposta do login...") for i in range(30): time.sleep(1) if self._check_already_logged_in(): print("✅ Login bem-sucedido!") success = self._extract_token_from_browser() if success: self._save_token() return success if self._handle_session_modal(): print("✅ Modal de sessão ativa resolvido!") time.sleep(2) if self._check_already_logged_in(): print("✅ Login bem-sucedido após resolver modal!") success = self._extract_token_from_browser() if success: self._save_token() return success if i % 5 == 0 and i > 0: print(f"⏳ A aguardar... ({i}s)") print(f"❌ Login falhou. URL: {self.driver.current_url}") self.driver.quit() return False except Exception as e: print(f"❌ Erro: {e}") if self.driver: self.driver.quit() return False def _handle_session_modal(self): """Lida especificamente com o modal 'A conta... deseja desligar?'""" try: modal_indicators = [ "A conta a que está a tentar ligar-se está a ser utilizada", "conta a que está a tentar ligar-se", "Gostaria de a desligar", "desligar", "Dimo Maint" ] page_source = self.driver.page_source modal_present = False for indicator in modal_indicators: if indicator.lower() in page_source.lower(): modal_present = True break if not modal_present: return False print("⚠️ Modal de sessão ativa detetado!") try: sim_button = self.driver.find_element(By.XPATH, "//button[contains(text(), 'Sim')]") if sim_button.is_displayed() and sim_button.is_enabled(): print("✅ A clicar em 'Sim' para desligar a outra sessão...") sim_button.click() time.sleep(2) return True except: pass try: result = self.driver.execute_script(""" const buttons = document.querySelectorAll('button'); for (const btn of buttons) { if (btn.textContent && btn.textContent.trim() === 'Sim') { btn.click(); return true; } } return false; """) if result: print("✅ Clicou em 'Sim' via JavaScript") time.sleep(2) return True except: pass return False except Exception as e: print(f"⚠️ Erro ao lidar com modal: {e}") return False def _check_already_logged_in(self): """Verifica se já está logado""" try: current_url = self.driver.current_url if '/SERRATEC' in current_url and 'account' not in current_url: cookies = self.driver.get_cookies() for cookie in cookies: if cookie['name'] == 'MX-Access-Token': return True return False except: return False def _find_element(self, selectors): """Tenta encontrar um elemento por múltiplos selectors""" for selector in selectors: try: element = self.driver.find_element(By.CSS_SELECTOR, selector) if element.is_displayed(): return element except: pass return None def _extract_token_from_browser(self): """Extrai o token do navegador""" try: cookies = self.driver.get_cookies() for cookie in cookies: if cookie['name'] == 'MX-Access-Token': self.token = cookie['value'] print(f"🔑 Token obtido: {self.token[:30]}...") self.session.cookies.set(cookie['name'], cookie['value']) self.driver.quit() return True token = self.driver.execute_script(""" const cookies = document.cookie.split('; '); for (const c of cookies) { const parts = c.trim().split('='); if (parts[0] === 'MX-Access-Token') { return parts[1]; } } return null; """) if token: self.token = token self.session.cookies.set('MX-Access-Token', token) print(f"🔑 Token obtido via JS: {token[:30]}...") self.driver.quit() return True print("⚠️ Token não encontrado") self.driver.quit() return False except Exception as e: print(f"❌ Erro: {e}") if self.driver: self.driver.quit() return False def _carregar_estados(self): """Carrega a lista de estados do sistema através da API""" if not self.token: return {} if self.estados_cache: return self.estados_cache try: print("📊 A carregar lista de estados do sistema...") query = """ query { states { items { id name } totalCount } } """ headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", "controller": "workorder", "action": "list", "tenant": "SERRATEC" } response = self.session.post( f"{self.base_url}/graphql", headers=headers, json={"query": query}, timeout=30 ) if response.status_code == 200: data = response.json() if "data" in data and "states" in data["data"]: items = data["data"]["states"]["items"] for item in items: state_id = item.get('id', '') state_name = item.get('name', '') if state_id and state_name: self.estados_cache[state_id] = state_name print(f"✅ Carregados {len(self.estados_cache)} estados do sistema") return self.estados_cache except Exception as e: print(f"⚠️ Erro ao carregar estados: {e}") self.estados_cache = { '8b959fa9-92da-48af-83a8-8960b86e5d3e' : 'Por fazer', '08059026-de6e-8b90-bde1-4ae92f676643': 'Suspenso', '32fbd621-0458-4e3e-968d-77ffc56c892a': 'Em curso', 'd17964fa-1174-ed36-eb9c-f50d58980859': 'A aguardar intervenção externa', '658ee047-7054-cae7-8e55-b9ab543b67c7': 'A aguardar material', 'df9e4a1c-4670-8242-3843-d0a3e8d175da': 'A aguardar reparação da peça', '7b419f98-ebc2-dcd4-8db2-312cdfbe8072' : 'A aguardar validação qualidade', '8240c123-1dd5-860b-6c08-041805b85dbf' : 'A aguardar validação segurança', 'ecdfc42d-e39a-d2e4-1455-78fc9d35dc05' : 'Em analise com o fabricante', '85c3cbee-2503-4bf5-8271-b13b41b02a08' : 'Concluída', '529a1f42-ebb9-0688-3523-54e7b4dd5783' : 'A aguardar documentação', '1ee4fca1-325d-948d-af5d-bc5fefa53635' : 'Em teste', '0ef45d0d-1901-959e-8727-869a178e0f6b' : 'Reparação corretiva temporaria', 'd67894ef-2d53-4e63-a840-71089d2cc851' : 'Anulada', 'ada41e02-5e6a-4dc7-9f5c-e6b14340f1a6' : 'Encerrada', } return self.estados_cache def _carregar_estados_workrequest(self): """Carrega a lista de estados de workrequest do sistema""" if not self.token: return {} if self.estados_workrequest_cache: return self.estados_workrequest_cache try: print("📊 A carregar estados de workrequest...") self.estados_workrequest_cache = { 'f890de1b-26b3-45a9-b8b8-9887afbf3360': 'Em curso', 'a7386d0d-5244-4811-94d9-1e77b58693f4': 'A processar', '50f51fa2-045d-48c2-92e7-a81637ea33f4': 'Encerrado', '70f32bd4-875b-48e9-a51d-133ebe30f0f8': 'Efetuado', 'a579a008-d90b-48fe-850c-43ff3f6c4de7' : 'Anulado', } return self.estados_workrequest_cache except Exception as e: print(f"⚠️ Erro ao carregar estados de workrequest: {e}") return {} def _obter_nome_estado(self, state_id): """Obtém o nome do estado a partir do ID""" if state_id is None: return 'Desconhecido' state_id_str = str(state_id) if not self.estados_cache: self._carregar_estados() if state_id_str in self.estados_cache: return self.estados_cache[state_id_str] return state_id_str[:8] + '...' def _obter_nome_estado_workrequest(self, state_id): """Obtém o nome do estado de workrequest a partir do ID""" if state_id is None: return 'Desconhecido' state_id_str = str(state_id) if not self.estados_workrequest_cache: self._carregar_estados_workrequest() if state_id_str in self.estados_workrequest_cache: return self.estados_workrequest_cache[state_id_str] mapping = { 'f890de1b-26b3-45a9-b8b8-9887afbf3360': 'Em curso', 'a7386d0d-5244-4811-94d9-1e77b58693f4': 'Por fazer', '50f51fa2-045d-48c2-92e7-a81637ea33f4': 'Encerrado', '70f32bd4-875b-48e9-a51d-133ebe30f0f8': 'Concluído', } if state_id_str in mapping: return mapping[state_id_str] return state_id_str[:8] + '...' def get_workorders(self, skip=0, take=100): """Obtém ordens de trabalho""" if not self.token: print("❌ Erro: Token não definido!") return None query = """ query workorders($skip: Int, $take: Int) { workorder( skip: $skip take: $take order: [{beginDateTime: DESC}] where: {and: [{archived: {eq: false}}]} filter: {} ) { items { id code name assetCode assetName brand brandModel serialNumber criticality beginDateTime closed stateId technician issuerFullName urgency remedy alert archived } totalCount } } """ headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", "controller": "workorder", "action": "list", "tenant": "SERRATEC" } try: response = self.session.post( f"{self.base_url}/graphql", headers=headers, json={ "query": query, "variables": { "skip": skip, "take": take } }, timeout=30 ) if response.status_code == 200: return response.json() elif response.status_code == 401: print("⚠️ Token expirado!") self.token = None if os.path.exists(self.token_file): os.remove(self.token_file) return None else: print(f"❌ Erro HTTP: {response.status_code}") return None except Exception as e: print(f"❌ Erro: {e}") return None def get_all_workorders(self, limit=5000): """ Obtém as últimas N ordens de trabalho (mais recentes) - limit: número máximo de OTs a buscar (default 2000) """ if not self.token: print("❌ Erro: Não estás autenticado!") return [] self._carregar_estados() all_items = [] skip = 0 take = min(100, limit) print(f"📥 A buscar as últimas {limit} OTs (mais recentes)...") while len(all_items) < limit: print(f"📥 A buscar registos de {skip} a {skip + take}...") result = self.get_workorders(skip, take) if result is None: break if "errors" in result: print(f"❌ Erro GraphQL: {result['errors']}") break data = result["data"]["workorder"] items = data["items"] total_count = data["totalCount"] all_items.extend(items) print(f"✅ Obtidos {len(items)} registos (total: {len(all_items)}/{min(limit, total_count)})") if len(all_items) >= limit: break if len(all_items) >= total_count: break skip += take time.sleep(0.5) print(f"📊 Total de OTs extraídas: {len(all_items)} (limite: {limit})") return all_items # ======================================================================== # MÉTODOS PARA WORK REQUESTS (PEDIDOS DE INTERVENÇÃO) # ======================================================================== def get_workrequests(self, skip=0, take=100): """Obtém pedidos de intervenção (work requests)""" if not self.token: print("❌ Erro: Token não definido!") return None query = """ query workrequests($skip: Int, $take: Int) { workrequest( skip: $skip take: $take order: [{creationDateTime: DESC}] filter: {} ) { items { id code name asset brand brandModel serialNumber criticality creationDateTime stateId stateCode requesterFullName urgency alert archived color } totalCount } } """ headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", "controller": "workrequest", "action": "list", "tenant": "SERRATEC" } try: response = self.session.post( f"{self.base_url}/graphql", headers=headers, json={ "query": query, "variables": { "skip": skip, "take": take } }, timeout=30 ) if response.status_code == 200: return response.json() elif response.status_code == 401: print("⚠️ Token expirado!") self.token = None if os.path.exists(self.token_file): os.remove(self.token_file) return None else: print(f"❌ Erro HTTP: {response.status_code}") return None except Exception as e: print(f"❌ Erro: {e}") return None def get_all_workrequests(self, limit=5000): """ Obtém os últimos N pedidos de intervenção (mais recentes) - limit: número máximo de PIs a buscar (default 2000) """ if not self.token: print("❌ Erro: Não estás autenticado!") return [] self._carregar_estados_workrequest() all_items = [] skip = 0 take = min(100, limit) print(f"📥 A buscar os últimos {limit} PIs (mais recentes)...") while len(all_items) < limit: print(f"📥 A buscar pedidos de {skip} a {skip + take}...") result = self.get_workrequests(skip, take) if result is None: break if "errors" in result: print(f"❌ Erro GraphQL: {result['errors']}") break data = result["data"]["workrequest"] items = data["items"] total_count = data["totalCount"] all_items.extend(items) print(f"✅ Obtidos {len(items)} pedidos (total: {len(all_items)}/{min(limit, total_count)})") if len(all_items) >= limit: break if len(all_items) >= total_count: break skip += take time.sleep(0.5) print(f"📊 Total de PIs extraídos: {len(all_items)} (limite: {limit})") return all_items def export_workrequests_to_csv(self, workrequests, filename): """Exporta pedidos de intervenção para CSV""" if not workrequests: print("❌ Sem dados para exportar") return False fieldnames = [ 'code', 'name', 'asset', 'brand', 'brandModel', 'serialNumber', 'criticality', 'creationDateTime', 'stateId', 'stateCode', 'requesterFullName', 'urgency', 'alert', 'archived', 'color', 'id' ] try: os.makedirs(os.path.dirname(filename), exist_ok=True) with open(filename, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=';') writer.writeheader() for wr in workrequests: row = {k: wr.get(k, '') for k in fieldnames} writer.writerow(row) print(f"✅ Exportado {len(workrequests)} pedidos para:") print(f" 📄 {filename}") return True except Exception as e: print(f"❌ Erro ao guardar CSV: {e}") return False def gerar_relatorio_ultimos_pis(self, workrequests): """Gera relatório com os pedidos de intervenção""" if not workrequests: print("❌ Sem dados para gerar relatório") return [] relatorio = [] for wr in workrequests: state_id = wr.get('stateId', '') estado_nome = self._obter_nome_estado_workrequest(state_id) cor = wr.get('color', '#FFFFFF') mapa_cores = { '#FFEB3B': '🟡 Amarelo', '#ff3b3b': '🔴 Vermelho', '#FECD55': '🟠 Laranja', '#FFFFFF': '⚪ Branco' } cor_nome = mapa_cores.get(cor, cor) relatorio.append({ 'Código PI': wr.get('code', ''), 'Nome': wr.get('name', ''), 'Equipamento': wr.get('asset', ''), 'Marca': wr.get('brand', ''), 'Modelo': wr.get('brandModel', ''), 'Nº Série': wr.get('serialNumber', ''), 'Criticidade': wr.get('criticality', ''), 'Data Criação': wr.get('creationDateTime', '')[:16] if wr.get('creationDateTime') else '', 'Estado': estado_nome, 'Estado Código': wr.get('stateCode', ''), 'Solicitante': wr.get('requesterFullName', ''), 'Urgência': wr.get('urgency', ''), 'Alerta': 'Sim' if wr.get('alert') else 'Não', 'Arquivado': 'Sim' if wr.get('archived') else 'Não', 'Cor': cor_nome, 'Alterações': wr.get('_alteracoes') }) print(f"📊 Relatório gerado com {len(relatorio)} pedidos") return relatorio def gerar_html_pis(self, relatorio, filename, refresh_interval=300): """Gera um HTML visual e interativo com os pedidos de intervenção""" print(f"📝 A gerar HTML visual dos PIs: {filename}") if not relatorio: print("❌ Sem dados para gerar o relatório HTML") return False total_pis = len(relatorio) estados = defaultdict(int) urgencias = defaultdict(int) criticidades = defaultdict(int) for r in relatorio: estado = r.get('Estado', 'Desconhecido') urgencia = r.get('Urgência', 'Desconhecida') criticidade = r.get('Criticidade', 'Desconhecida') estados[estado] += 1 urgencias[urgencia] += 1 criticidades[criticidade] += 1 ativos = sum(1 for r in relatorio if r.get('Estado') in ['Em curso', 'Por fazer']) concluidos = sum(1 for r in relatorio if r.get('Estado') in ['Concluído', 'Encerrado']) now = datetime.now() novos_pis = 0 for r in relatorio: data_criacao = r.get('Data Criação', '') if data_criacao: try: data_str = data_criacao.strip() if 'T' in data_str: data_str = data_str.replace('T', ' ') if len(data_str) > 16: data_str = data_str[:16] data_obj = datetime.strptime(data_str, "%Y-%m-%d %H:%M") diff = (now - data_obj).total_seconds() if diff < 86400: novos_pis += 1 except: pass refresh_minutes = refresh_interval // 60 if refresh_interval > 0 else 0 html_content = f''' 📋 Pedidos de Intervenção (PIs) - Últimos {total_pis} ''' if refresh_interval > 0: html_content += f' \n' html_content += f'''

📋 Pedidos de Intervenção (PIs)

Últimos {total_pis} pedidos (mais recentes)
📅 Gerado em {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}
''' if refresh_interval > 0: html_content += f'''
🔄 Atualização automática a cada {refresh_minutes} minutos ({refresh_interval}s)
''' html_content += f'''
🆕 {novos_pis} novos nas últimas 24h
{total_pis}
📋 Total de PIs
{ativos}
🔄 Ativos
{concluidos}
✅ Concluídos
''' for urgencia, count in sorted(urgencias.items(), key=lambda x: -x[1]): emoji = '🔴' if 'Muito' in urgencia else '🟠' if 'Urgente' in urgencia else '🟡' if 'Moderado' in urgencia else '🟢' html_content += f'''
{count}
{emoji} {urgencia}
''' html_content += f'''
Mostrando 0 de 0
''' if not relatorio: html_content += '''
📭

Nenhum pedido de intervenção encontrado

Verifique se há dados disponíveis na API.

''' else: for idx, r in enumerate(relatorio): estado = r.get('Estado', 'Desconhecido') estado_class = estado.replace(' ', '_') if estado_class not in ['Em_curso', 'Por_fazer', 'Concluído', 'Encerrado']: estado_class = 'default' urgencia = r.get('Urgência', 'Não definida') urgencia_class = urgencia.replace(' ', '_') if urgencia_class not in ['Muito_Urgente', 'Urgente', 'Moderado', 'Não_urgente']: urgencia_class = 'default' criticidade = r.get('Criticidade', 'Não definida') criticidade_class = criticidade.replace(' ', '_') if criticidade_class not in ['Muito_Crítico', 'Crítico', 'Não_Crítico']: criticidade_class = 'default' border_color = '#f5576c' if 'Muito Urgente' in urgencia else '#f5a623' if 'Urgente' in urgencia else '#4a90d9' data_criacao = r.get('Data Criação', '') solicitante = r.get('Solicitante', '-') equipamento = r.get('Equipamento', '-') alteracoes = r.get('Alterações') change_badge = '' change_description = '' if alteracoes: change_badge = '
🔔 Alterado
'.format( alteracoes['at'] + stock.ChangeTracker.DURACAO_SEGUNDOS ) change_description = '
🔔 {}
'.format( ' • '.join(alteracoes['fields']) ) eh_novo = False eh_muito_novo = False if data_criacao: try: data_str = data_criacao.strip() if 'T' in data_str: data_str = data_str.replace('T', ' ') if len(data_str) > 16: data_str = data_str[:16] data_obj = datetime.strptime(data_str, "%Y-%m-%d %H:%M") diff = (now - data_obj).total_seconds() if diff < 7200: eh_muito_novo = True eh_novo = True elif diff < 86400: eh_novo = True except: pass card_classes = "pi-card" if alteracoes: card_classes += " alterado" if eh_muito_novo: card_classes += " muito-novo novo" elif eh_novo: card_classes += " novo" html_content += f'''
{r.get('Código PI', 'N/A')}
{equipamento}
{estado} {change_badge} ''' if eh_novo: html_content += f''' 🆕 NOVO ''' html_content += f'''
{change_description}
{r.get('Nome', 'Sem título')}
🕐 Data Criação
{data_criacao}
👤 Solicitante
{solicitante}
🔴 Urgência
{urgencia}
⚠️ Criticidade
{criticidade}
🏷️ Marca
{r.get('Marca', '-')}
📦 Modelo
{r.get('Modelo', '-')}
🔢 Nº Série
{r.get('Nº Série', '-')}
''' html_content += f'''

📋 Relatório de Pedidos de Intervenção (PIs) • Últimos {total_pis} pedidos • {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}

Total de PIs: {total_pis} | 🆕 Novos (24h): {novos_pis}

''' try: os.makedirs(os.path.dirname(filename), exist_ok=True) with open(filename, 'w', encoding='utf-8') as f: f.write(html_content) print(f"✅ Relatório HTML dos PIs gerado com sucesso!") print(f" 📄 {filename}") print(f" 📊 Tamanho: {os.path.getsize(filename)} bytes") return True except Exception as e: print(f"❌ Erro ao gerar HTML: {e}") import traceback traceback.print_exc() return False def export_to_csv(self, workorders, filename): """Exporta para CSV""" if not workorders: print("❌ Sem dados para exportar") return False fieldnames = [ 'id', 'name', 'assetCode', 'assetName', 'brand', 'brandModel', 'serialNumber', 'criticality', 'beginDateTime', 'closed', 'stateId', 'technician', 'issuerFullName', 'urgency', 'remedy', 'alert', 'archived' ] try: os.makedirs(os.path.dirname(filename), exist_ok=True) with open(filename, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=';') writer.writeheader() for wo in workorders: row = {k: wo.get(k, '') for k in fieldnames} writer.writerow(row) print(f"✅ Exportado {len(workorders)} registos para:") print(f" 📄 {filename}") return True except Exception as e: print(f"❌ Erro ao guardar CSV: {e}") return False # ======================================================================== # FUNÇÕES PARA RELATÓRIO VISUAL DAS ÚLTIMAS OTs # ======================================================================== def _formatar_valor(self, valor): """Formata um valor para string, tratando diferentes tipos""" if valor is None: return '' if isinstance(valor, bool): return 'Sim' if valor else 'Não' if isinstance(valor, (int, float)): return str(valor) if isinstance(valor, str): return valor return str(valor) def _formatar_data(self, valor): """Formata uma data para exibição (apenas os primeiros 10 caracteres)""" valor_str = self._formatar_valor(valor) if valor_str and len(valor_str) >= 10: return valor_str[:10] return valor_str def _normalizar_codigo_maquina(self, codigo): """Normaliza o código da máquina removendo sufixos (A, B, C, etc.)""" if not codigo: return codigo codigo_normalizado = re.sub(r'[A-Da-d]$', '', codigo) if codigo_normalizado == codigo: return codigo return codigo_normalizado def _is_ot_planeada(self, wo): """Verifica se uma OT é planeada""" palavras_planeadas = [ 'anual', 'semestral', 'trimestral', 'mensal', 'planeada', 'planejada', 'preventiva', 'revisão', 'revisao', 'calibração', 'calibracao', 'inspeção', 'inspecao', 'manutenção programada', 'manutencao programada' ] nome = wo.get('name', '').lower() for palavra in palavras_planeadas: if palavra in nome: return True return False def gerar_relatorio_ultimas_ots(self, workorders, num_ultimas=1): """Gera um relatório com as últimas N ordens de trabalho por máquina""" if not workorders: print("❌ Sem dados para gerar relatório") return [] workorders_filtradas = [] ots_planeadas = 0 for wo in workorders: if self._is_ot_planeada(wo): ots_planeadas += 1 else: workorders_filtradas.append(wo) print(f"📊 Filtradas {ots_planeadas} OTs planeadas") print(f"📊 Restam {len(workorders_filtradas)} OTs não-planeadas") maquinas = defaultdict(list) for wo in workorders_filtradas: machine_id = wo.get('assetCode') or wo.get('assetName') or 'SEM_MÁQUINA' if machine_id: machine_id_normalizado = self._normalizar_codigo_maquina(machine_id) maquinas[machine_id_normalizado].append(wo) print(f"📊 Encontradas {len(maquinas)} máquinas (após normalização)") for machine_id in maquinas: maquinas[machine_id].sort( key=lambda x: x.get('beginDateTime', ''), reverse=True ) relatorio = [] for machine_id, ots in sorted(maquinas.items()): ultimas = ots[:num_ultimas] for wo in ultimas: nome_maquina = wo.get('assetName', '') state_id = wo.get('stateId', '') estado_nome = self._obter_nome_estado(state_id) relatorio.append({ 'Máquina': machine_id, 'Nome Máquina': self._formatar_valor(nome_maquina), 'ID OT': self._formatar_valor(wo.get('id', '')), 'Código OT': self._formatar_valor(wo.get('code', '')), 'Título': self._formatar_valor(wo.get('name', '')), 'Estado': estado_nome, 'Estado ID': self._formatar_valor(state_id), 'Urgência': self._formatar_valor(wo.get('urgency', '')), 'Data Início': self._formatar_data(wo.get('beginDateTime', '')), 'Data Fecho': self._formatar_data(wo.get('closed', '')), 'Técnico': self._formatar_valor(wo.get('technician', '')), 'Emissor': self._formatar_valor(wo.get('issuerFullName', '')), 'Solução': self._formatar_valor(wo.get('remedy', '')), 'Marca': self._formatar_valor(wo.get('brand', '')), 'Modelo': self._formatar_valor(wo.get('brandModel', '')), 'Série': self._formatar_valor(wo.get('serialNumber', '')), 'Criticidade': self._formatar_valor(wo.get('criticality', '')), 'Arquivada': 'Sim' if wo.get('archived') else 'Não', 'Alerta': 'Sim' if wo.get('alert') else 'Não', 'Total OTs': len(ots), 'Alterações': wo.get('_alteracoes') }) print(f"📊 Relatório gerado com {len(relatorio)} registos") return relatorio def gerar_html_visual(self, relatorio, filename, refresh_interval=300): """ Gera um HTML visual e interativo com as últimas OTs por máquina Ao clicar no ID da OT, faz uma requisição para o proxy local (resolução CORS) """ print(f"📝 A gerar HTML visual: {filename}") if not relatorio: print("❌ Sem dados para gerar o relatório HTML") return False # Agrupar OTs por máquina maquinas = defaultdict(list) for r in relatorio: machine_id = r.get('Máquina', 'SEM_MÁQUINA') maquinas[machine_id].append(r) total_maquinas = len(maquinas) total_ots = len(relatorio) estados = defaultdict(int) urgencias = defaultdict(int) for r in relatorio: estado = r.get('Estado', 'Desconhecido') urgencia = r.get('Urgência', 'Desconhecida') estados[estado] += 1 urgencias[urgencia] += 1 ativas = sum(1 for r in relatorio if r.get('Estado') in ['Em curso', 'Por fazer', 'Suspenso']) encerradas = sum(1 for r in relatorio if r.get('Estado') == 'Encerrada') refresh_minutes = refresh_interval // 60 if refresh_interval > 0 else 0 # Proxy local para resolver CORS proxy_url = "http://localhost:8888" html_content = f''' 📊 Últimas Ordens de Trabalho por Máquina ''' if refresh_interval > 0: html_content += f' \n' html_content += f'''

🔧 Últimas Ordens de Trabalho

Resumo das últimas ordens de trabalho por máquina (exclui OTs planeadas)
📅 Gerado em {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}
''' if refresh_interval > 0: html_content += f'''
🔄 Atualização automática a cada {refresh_minutes} minutos ({refresh_interval}s)
''' html_content += f'''
{total_maquinas}
🖥️ Máquinas
{total_ots}
📝 Total de OTs
{encerradas}
✅ Encerradas
{ativas}
🔄 Ativas
Mostrando 0 de 0
''' if not relatorio: html_content += '''
📭

Nenhuma ordem de trabalho encontrada

Verifique se há dados disponíveis na API.

''' else: for machine_id, ots in sorted(maquinas.items()): # Primeira OT da máquina (a mais recente) para mostrar no resumo primeira_ot = ots[0] if ots else {} alteracoes_ots = [ot.get('Alterações') for ot in ots if ot.get('Alterações')] alteracoes = alteracoes_ots[0] if alteracoes_ots else None change_badge = '' change_description = '' if alteracoes: change_badge = '
🔔 Alterado
'.format( alteracoes['at'] + stock.ChangeTracker.DURACAO_SEGUNDOS ) change_description = '
🔔 OT alterada: {}
'.format( ' • '.join(alteracoes['fields']) ) estado = primeira_ot.get('Estado', 'Desconhecido') estado_class = estado.replace(' ', '_') if estado_class not in ['Encerrada', 'Em_curso', 'A_aguardar_documentação', 'Anulada', 'Por_fazer', 'Suspenso']: estado_class = 'default' urgencia = primeira_ot.get('Urgência', 'Não definida') urgencia_class = urgencia.replace(' ', '_') if urgencia_class not in ['Muito_Urgente', 'Urgente', 'Moderado', 'Não_urgente']: urgencia_class = 'default' criticidade = primeira_ot.get('Criticidade', 'Não definida') criticidade_class = criticidade.replace(' ', '_') if criticidade_class not in ['Muito_Crítico', 'Crítico', 'Não_Crítico']: criticidade_class = 'default' data_ini = primeira_ot.get('Data Início', '') data_fecho = primeira_ot.get('Data Fecho', '') id_ot = primeira_ot.get('ID OT', '') nome_maquina = primeira_ot.get('Nome Máquina', '') html_content += f'''
🖥️ {machine_id}
{nome_maquina}
📝 {len(ots)} OTs {change_badge} ▼
{change_description}
ID da OT (última)
🔍 {id_ot}
Título
{primeira_ot.get('Título', '-')}
Estado
{estado}
Urgência
{urgencia}
📅 Data Início
{data_ini}
📅 Data Fecho
{data_fecho}
👤 Técnico
{primeira_ot.get('Técnico', '-')}
👤 Emissor
{primeira_ot.get('Emissor', '-')}
🏷️ Marca/Modelo
{primeira_ot.get('Marca', '-')} {primeira_ot.get('Modelo', '')}
🔢 Nº Série
{primeira_ot.get('Série', '-')}
⚠️ Criticidade
{criticidade}
📦 Arquivada
{'✅ Sim' if primeira_ot.get('Arquivada') == 'Sim' else '❌ Não'}
💡 Solução: {primeira_ot.get('Solução', 'N/A')}
📋 Todas as OTs desta máquina ({len(ots)})
''' # Listar todas as OTs da máquina for ot in ots: ot_estado = ot.get('Estado', 'Desconhecido') ot_estado_class = ot_estado.replace(' ', '_') ot_id = ot.get('ID OT', '') ot_code = ot.get('Código OT', '') ot_titulo = ot.get('Título', 'Sem título') ot_data = ot.get('Data Início', '') html_content += f'''
{ot_data} 🔍 {ot_code} {ot_titulo} {ot_estado}
''' html_content += '''
''' html_content += f'''

📊 Relatório gerado automaticamente • Última OT por máquina (exclui OTs planeadas) • {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}

Total de máquinas: {total_maquinas} | Total de OTs: {total_ots}

''' # JavaScript no final do body js_code = f''' ''' html_content += js_code html_content += ''' ''' try: os.makedirs(os.path.dirname(filename), exist_ok=True) with open(filename, 'w', encoding='utf-8') as f: f.write(html_content) print(f"✅ Relatório HTML visual gerado com sucesso!") print(f" 📄 {filename}") print(f" 📊 Tamanho: {os.path.getsize(filename)} bytes") return True except Exception as e: print(f"❌ Erro ao gerar HTML: {e}") import traceback traceback.print_exc() return False # ======================================================================== # MÉTODOS PARA PLANEAMENTO DE TÉCNICOS (PLANNING) # ======================================================================== def get_planning_technician(self, start_date=None, end_date=None, days=5): """ Obtém o planeamento de técnicos para um período Args: start_date (str): Data de início no formato 'YYYY-MM-DDTHH:MM:SS' end_date (str): Data de fim no formato 'YYYY-MM-DDTHH:MM:SS' days (int): Número de dias a buscar (se start_date e end_date não forem fornecidos) """ if not self.token: print("❌ Erro: Token não definido!") return None # Se não forem fornecidas datas, usa os últimos N dias if start_date is None or end_date is None: hoje = datetime.now() start = hoje - timedelta(days=days) end = hoje + timedelta(days=7) start_date = start.strftime("%Y-%m-%dT%H:%M:%S") end_date = end.strftime("%Y-%m-%dT%H:%M:%S") print(f"📅 A buscar planeamento dos últimos {days} dias...") url = f"{self.base_url}/SERRATEC/api/planningTechnician/getPlanning" params = { "startDate": start_date, "endDate": end_date } headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", "tenant": "SERRATEC" } try: print(f"📅 A buscar planeamento de {start_date} a {end_date}...") response = self.session.get(url, headers=headers, params=params, timeout=30) if response.status_code == 200: print("✅ Planning obtido com sucesso!") data = response.json() if isinstance(data, dict): data['period'] = f"{start_date[:10]} a {end_date[:10]}" elif isinstance(data, list): data = { 'items': data, 'period': f"{start_date[:10]} a {end_date[:10]}" } return data elif response.status_code == 401: print("⚠️ Token expirado! A renovar...") self.token = None if os.path.exists(self.token_file): os.remove(self.token_file) return None else: print(f"❌ Erro HTTP: {response.status_code}") print(f" Resposta: {response.text[:200]}...") return None except Exception as e: print(f"❌ Erro ao buscar planning: {e}") return None def gerar_html_planeamento(self, planning_data, filename): """ Gera um HTML visual em formato de escala/calendário com as OTs organizadas por dia e técnico Usa os dados da API de planning (lista de OTs) Args: planning_data: Lista de OTs obtidas pela API getPlanning filename: Nome do ficheiro HTML a gerar """ print(f"📝 A gerar HTML do planeamento em escala: {filename}") if not planning_data: print("❌ Sem dados para gerar o planeamento") return False # Debug: mostrar estrutura print(f"🔍 Total de OTs recebidas: {len(planning_data)}") if len(planning_data) > 0: print(f" - Primeiro item: {list(planning_data[0].keys()) if isinstance(planning_data[0], dict) else 'N/A'}") # Extrair técnicos e suas OTs tecnicos_ots = {} ots_sem_tecnico = [] # Processar cada item (que é uma OT) for item in planning_data: if not isinstance(item, dict): continue # Extrair nome do técnico tech_nome = None technician = item.get('technician') if isinstance(technician, dict): tech_nome = technician.get('fullName') or technician.get('name') elif isinstance(technician, str) and technician.strip(): tech_nome = technician.strip() # Se não tem technician, verificar outros campos if not tech_nome: if item.get('technicianName'): tech_nome = item.get('technicianName') elif item.get('assignedTo'): tech_nome = item.get('assignedTo') # Se ainda não tem, verificar no objeto technician por outros nomes if not tech_nome and isinstance(technician, dict): # Tentar firstName + lastName first = technician.get('firstName', '') last = technician.get('lastName', '') if first or last: tech_nome = f"{first} {last}".strip() # Se encontrou técnico, adicionar if tech_nome: if tech_nome not in tecnicos_ots: tecnicos_ots[tech_nome] = [] tecnicos_ots[tech_nome].append(item) else: # Sem técnico - guardar separadamente ots_sem_tecnico.append(item) # Adicionar OTs sem técnico como "Não atribuído" if ots_sem_tecnico: tecnicos_ots['Não atribuído'] = ots_sem_tecnico if not tecnicos_ots: print("⚠️ Nenhum técnico com OTs encontrado no planning.") print(" Usando 'Não atribuído' para todas as OTs...") tecnicos_ots['Não atribuído'] = [item for item in planning_data if isinstance(item, dict)] # Ordenar técnicos tecnicos = sorted(tecnicos_ots.keys()) total_ots = sum(len(ots) for ots in tecnicos_ots.values()) print(f"👤 Técnicos encontrados: {len(tecnicos)}") for tech, ots in tecnicos_ots.items(): print(f" - {tech}: {len(ots)} OTs") print(f"📋 Total de OTs: {total_ots}") # Dias da semana dias_semana = ['Segunda', 'Terça', 'Quarta', 'Quinta', 'Sexta', 'Sábado', 'Domingo'] # Agrupar OTs por dia e técnico - usar ID único agenda = {} ots_sem_data = 0 # Mapeamento de ID da OT para dados mapa_ots = {} for tech, ots in tecnicos_ots.items(): for ot in ots: # Extrair data begin = ot.get('beginDateTime') if not begin: begin = ot.get('begin') or ot.get('startDateTime') or ot.get('start') or ot.get('date') if not begin: ots_sem_data += 1 continue data_str = str(begin)[:10] if data_str not in agenda: agenda[data_str] = {} if tech not in agenda[data_str]: agenda[data_str][tech] = [] # Extrair hora hora = '' if len(str(begin)) >= 16: hora = str(begin)[11:16] # Extrair hora de fim end = ot.get('endDateTime') or ot.get('end') or ot.get('finishDateTime') or '' hora_fim = '' if end and len(str(end)) >= 16: hora_fim = str(end)[11:16] # Extrair estado estado_obj = ot.get('state', {}) if isinstance(estado_obj, dict): estado_nome = estado_obj.get('name', '') estado_code = estado_obj.get('code', '') else: estado_nome = str(estado_obj) if estado_obj else '' estado_code = '' # Extrair asset asset_display = '' asset_code = '' asset_name = '' asset_obj = ot.get('asset', {}) if isinstance(asset_obj, dict): asset_code = asset_obj.get('code', '') asset_name = asset_obj.get('name', '') if asset_code: asset_display = f"{asset_code} - {asset_name}"[:40] else: asset_display = asset_name[:40] else: asset_code = ot.get('assetCode', '') or '' asset_name = ot.get('assetName', '') or '' asset_display = asset_code or asset_name # Extrair código da OT codigo = ot.get('code', '') if not codigo: codigo = ot.get('id', '')[:12] # ID único da OT ot_id = ot.get('id', '') # Criar dados do evento evento = { 'code': codigo, 'name': ot.get('name', ''), 'asset': asset_display, 'asset_code': asset_code, 'asset_name': asset_name, 'asset_id': asset_obj.get('id', '') if isinstance(asset_obj, dict) else '', 'hora': hora, 'hora_fim': hora_fim, 'state': estado_nome, 'state_code': estado_code, 'id': ot_id, 'beginDateTime': begin, 'endDateTime': end, 'preventiveMaintenanceCode': ot.get('preventiveMaintenanceCode', ''), 'workRequestCode': ot.get('workRequestCode', ''), 'labourId': ot.get('labourId', ''), 'anyLabourAchievedLine': ot.get('anyLabourAchievedLine', False), 'isLabourAchieved': ot.get('isLabourAchieved', False), 'isWorkOrderClosed': ot.get('isWorkOrderClosed', False), 'technician': tech, 'bodyId': ot.get('bodyId', ''), 'workOrderGroupId': ot.get('workOrderGroupId', ''), 'timeZoneId': ot.get('timeZoneId', ''), 'documentType': ot.get('documentType', ''), } agenda[data_str][tech].append(evento) # Guardar no mapa por ID if ot_id: mapa_ots[ot_id] = evento # Guardar também por código (fallback) if codigo: mapa_ots[codigo] = evento if ots_sem_data > 0: print(f"⚠️ {ots_sem_data} OTs ignoradas por não terem data válida") # Verificar se há dados na agenda if not agenda: html_content = ''' 📅 Planeamento de Técnicos

📅 Planeamento de Técnicos

📭 Nenhuma OT com data válida encontrada
''' else: datas_ordenadas = sorted(agenda.keys()) if datas_ordenadas: period = f"{datas_ordenadas[0]} a {datas_ordenadas[-1]}" else: period = "Semana atual" import json # Passar tanto a agenda como o mapa de OTs por ID dados_json = json.dumps({ 'agenda': agenda, 'mapa_ots': mapa_ots }, ensure_ascii=False) html_content = f''' 📅 Planeamento de Técnicos

📅 Planeamento de Técnicos

Escala de ordens de trabalho por dia e técnico
📆 {period}
👤 {len(tecnicos)} Técnicos 📋 {total_ots} OTs 📅 {len(datas_ordenadas)} Dias
Mostrando 0 OTs
Legenda: Concluída Encerrada Em curso Por fazer Suspenso Aguardar material Aguardar externa Análise fabricante
📅 Data
''' for tech in tecnicos: html_content += f'
{tech}
\n' html_content += '''
''' # Corpo do calendário for data_str in datas_ordenadas: try: data_obj = datetime.strptime(data_str, "%Y-%m-%d") dia_semana = dias_semana[data_obj.weekday()] dia_num = data_obj.day mes_num = data_obj.month data_formatada = f"{dia_num:02d}/{mes_num:02d}" except: dia_semana = '' data_formatada = data_str[5:10] html_content += f'''
{data_formatada} {dia_semana}
''' for tech in tecnicos: eventos = agenda.get(data_str, {}).get(tech, []) eventos_ordenados = sorted(eventos, key=lambda x: x.get('hora', '99:99')) if eventos_ordenados: html_content += f'
\n' for idx, evento in enumerate(eventos_ordenados): estado = evento.get('state', '') ot_id = evento.get('id', '') # Mapear estado para classe CSS estado_class = '' if 'Concluída' in estado or 'DONE' in estado: estado_class = 'evento-estado-Concluída' elif 'Encerrada' in estado or 'CLOSED' in estado: estado_class = 'evento-estado-Encerrada' elif 'Em curso' in estado or 'INPROGRESS' in estado: if 'aguardar material' in estado.lower(): estado_class = 'evento-estado-A_aguardar_material' elif 'intervenção externa' in estado.lower() or 'externa' in estado.lower(): estado_class = 'evento-estado-A_aguardar_intervencao_externa' elif 'análise com o fabricante' in estado.lower() or 'analise com o fabricante' in estado.lower(): estado_class = 'evento-estado-Em_analise_com_o_fabricante' else: estado_class = 'evento-estado-Em_curso' elif 'Por fazer' in estado or 'TODO' in estado: estado_class = 'evento-estado-Por_fazer' elif 'Suspenso' in estado: estado_class = 'evento-estado-Suspenso' elif 'aguardar material' in estado.lower(): estado_class = 'evento-estado-A_aguardar_material' elif 'analise com o fabricante' in estado.lower(): estado_class = 'evento-estado-Em_analise_com_o_fabricante' titulo = evento.get('name', '')[:50] titulo = titulo.replace('<', '<').replace('>', '>') codigo = evento.get('code', '') hora = evento.get('hora', '--:--') asset = evento.get('asset', '') # USAR O ID DA OT COMO IDENTIFICADOR ÚNICO evento_id = ot_id if ot_id else f"evt_{data_str}_{tech}_{idx}_{codigo}" html_content += f'''
{hora} {codigo} {titulo} 🏷️ {asset}
''' html_content += '
\n' else: html_content += f'
—
\n' html_content += '
\n' html_content += '''

📅 Planeamento gerado automaticamente • ''' + datetime.now().strftime('%d/%m/%Y %H:%M:%S') + '''

Total: ''' + str(len(tecnicos)) + ''' técnicos | ''' + str(total_ots) + ''' OTs

''' try: os.makedirs(os.path.dirname(filename), exist_ok=True) with open(filename, 'w', encoding='utf-8') as f: f.write(html_content) print(f"✅ HTML do planeamento gerado com sucesso!") print(f" 📄 {filename}") print(f" 📊 Tamanho: {os.path.getsize(filename)} bytes") return True except Exception as e: print(f"❌ Erro ao gerar HTML do planeamento: {e}") import traceback traceback.print_exc() return False # ============================================================================ # FUNÇÃO DE ATUALIZAÇÃO AUTOMÁTICA # ============================================================================ def auto_update(api, username, password, desktop, limite=5000, intervalo=300): """ Executa a atualização automática a cada X segundos Sobrescreve sempre os mesmos ficheiros (nomes fixos) Args: api: Instância da API username: Nome de utilizador password: Password desktop: Caminho da Área de Trabalho limite: Número máximo de registos intervalo: Intervalo em segundos (default 300 = 5 minutos) """ print(f"🔄 Iniciando atualização automática a cada {intervalo} segundos...") print("📁 Os ficheiros serão sobrescritos (nomes fixos)") print("=" * 70) # Nomes fixos para os ficheiros CSV_OTS = os.path.join(desktop, "ordens_trabalho.csv") JSON_OTS = os.path.join(desktop, "workorders.json") HTML_OTS = os.path.join(desktop, "ultimas_ots_visual.html") CSV_PIS = os.path.join(desktop, "pedidos_intervencao.csv") JSON_PIS = os.path.join(desktop, "workrequests.json") HTML_PIS = os.path.join(desktop, "pedidos_intervencao_visual.html") JSON_PLANNING = os.path.join(desktop, "planning_tecnico.json") HTML_PLANNING = os.path.join(desktop, "planning_tecnico.html") # Ficheiros de stock CSV_STOCK = os.path.join(desktop, "stock.csv") JSON_STOCK = os.path.join(desktop, "stock.json") HTML_STOCK = os.path.join(desktop, "stock_visual.html") HTML_MOTOR = os.path.join(desktop, "motor_pesquisa_visual.html") # Ficheiro de log com timestamp da última atualização LOG_FILE = os.path.join(desktop, "ultima_atualizacao.txt") change_tracker = stock.ChangeTracker(desktop) # Compara já a primeira atualização com os JSON existentes, se houver. change_tracker.inicializar_de_ficheiro("workorders", JSON_OTS) change_tracker.inicializar_de_ficheiro("workrequests", JSON_PIS) change_tracker.inicializar_de_ficheiro("stock", JSON_STOCK) while True: try: inicio = datetime.now() print(f"\n📅 {inicio.strftime('%Y-%m-%d %H:%M:%S')} - Iniciando atualização...") # Verificar token if not api.token: print("🔑 Token não encontrado. A fazer login...") if not api.login_with_selenium(username, password, headless=False): print("❌ Falha no login. A tentar novamente em 60 segundos...") time.sleep(60) continue # ============================================================ # 1. BUSCAR ORDENS DE TRABALHO # ============================================================ print(f"\n📊 [1/4] A extrair as últimas {limite} ordens de trabalho...") workorders = api.get_all_workorders(limit=limite) if workorders: workorders = change_tracker.analisar("workorders", workorders) api.export_to_csv(workorders, CSV_OTS) with open(JSON_OTS, 'w', encoding='utf-8') as f: json.dump(workorders, f, ensure_ascii=False, indent=2) relatorio = api.gerar_relatorio_ultimas_ots(workorders, num_ultimas=None) if relatorio: api.gerar_html_visual(relatorio, HTML_OTS) print(f"✅ OTs atualizadas: {len(workorders)} registos") else: print("⚠️ Nenhuma OT obtida.") # ============================================================ # 1.5. SINCRONIZAR COMENTÁRIOS (só das OTs abertas) # ============================================================ print("\n💬 [1.5/4] A sincronizar comentários das OTs...") try: comentarios.sync_comentarios( api, limite=20000, # só sincroniza 2000 por ciclo (as mais relevantes) concorrencia=15, # 15 pedidos em paralelo apenas_abertas=True, # ignora encerradas/anuladas ) except Exception as e: print(f"⚠️ Erro ao sincronizar comentários: {e}") # ============================================================ # 1.6. REGENERAR MOTOR DE PESQUISA VISUAL # ============================================================ print("\n🔍 [1.6/4] A regenerar motor de pesquisa visual...") try: motor_pesquisa_visual.gerar_motor_visual( filename=HTML_MOTOR, refresh_interval=0, abrir=False, ) print("✅ Motor de pesquisa visual regenerado.") except Exception as e: print(f"⚠️ Erro ao regenerar motor visual: {e}") # ============================================================ # 2. BUSCAR PEDIDOS DE INTERVENÇÃO # ============================================================ print(f"\n📋 [2/4] A extrair os últimos {limite} pedidos de intervenção...") workrequests = api.get_all_workrequests(limit=limite) if workrequests: workrequests = change_tracker.analisar("workrequests", workrequests) api.export_workrequests_to_csv(workrequests, CSV_PIS) with open(JSON_PIS, 'w', encoding='utf-8') as f: json.dump(workrequests, f, ensure_ascii=False, indent=2) relatorio_pis = api.gerar_relatorio_ultimos_pis(workrequests) if relatorio_pis: api.gerar_html_pis(relatorio_pis, HTML_PIS) print(f"✅ PIs atualizados: {len(workrequests)} registos") else: print("⚠️ Nenhum PI obtido.") # ============================================================ # 3. BUSCAR PLANEAMENTO DE TÉCNICOS # ============================================================ print("\n📅 [3/4] A extrair planeamento de técnicos...") planning_data = api.get_planning_technician(days=5) if planning_data: if isinstance(planning_data, dict) and 'items' in planning_data: planning_items = planning_data['items'] elif isinstance(planning_data, list): planning_items = planning_data else: planning_items = [] with open(JSON_PLANNING, 'w', encoding='utf-8') as f: json.dump(planning_items, f, ensure_ascii=False, indent=2) api.gerar_html_planeamento(planning_items, HTML_PLANNING) print(f"✅ Planning atualizado: {len(planning_items)} OTs") else: print("⚠️ Nenhum planeamento obtido.") ########### APAGAR CACHE ApÓS X TEMPO PARA FORÇAR ATUALIZAÇÂO DE PEÇAS DO STOCK #### ########### ############ cache_file = os.path.join(desktop, "stock_cache.json") if os.path.exists(cache_file): idade = time.time() - os.path.getmtime(cache_file) if idade > 3600: # mais de 1h os.remove(cache_file) print(f"🗑️ Cache stock com {idade/3600:.0f} min — apagado para refresh.") # ============================================================ # 4. BUSCAR STOCK (NOVO) # ============================================================ print("\n📦 [4/4] A extrair stock de peças...") try: stock_api = stock.StockAPI() if not stock_api.load_token(): print("⚠️ Token não disponível para o stock.") else: # Verificar se o token é válido parts = stock_api.get_all_parts(limit=limite) if parts and not stock_api.token_expired: # Processar peças com stock parts_com_stock = stock_api.process_parts_with_stock( parts, limit=None, save_interval=50 ) # Verificar se token expirou durante o processamento if stock_api.token_expired: print("🔄 Token expirou durante o processamento do stock!") print("🔑 A renovar token com Selenium...") # Renovar token via Selenium (usando o método do Dimo.py) if api.login_with_selenium(username, password, headless=True, force=True): print("✅ Token renovado! A recarregar stock...") # Recarregar token no stock_api if stock_api.recarregar_token(): print("✅ Token recarregado no stock_api.") # Re-tentar processar (as peças em cache vão acelerar) print("🔄 A retomar processamento do stock...") parts_com_stock = stock_api.process_parts_with_stock( parts, limit=None, save_interval=50 ) if stock_api.token_expired: print("⚠️ Token expirou novamente. A guardar dados parciais...") else: print("✅ Stock processado com sucesso após renovação!") else: print("⚠️ Falha ao recarregar token no stock_api.") else: print("❌ Falha ao renovar token. Stock incompleto.") # Guardar ficheiros de stock (mesmo que parciais) if parts_com_stock: parts_com_stock = change_tracker.analisar("stock", parts_com_stock) stock_api.export_to_csv(parts_com_stock, CSV_STOCK) with open(JSON_STOCK, 'w', encoding='utf-8') as f: json.dump(parts_com_stock, f, ensure_ascii=False, indent=2) stock_api.gerar_html_stock( parts_com_stock, HTML_STOCK, refresh_interval=300 ) print(f"✅ Stock atualizado: {len(parts_com_stock)} peças") else: print("⚠️ Nenhuma peça processada.") else: if stock_api.token_expired: print("🔄 Token expirou na busca inicial de peças. A renovar...") if api.login_with_selenium(username, password, headless=True, force=True): print("✅ Token renovado! A retomar stock...") stock_api.recarregar_token() parts = stock_api.get_all_parts(limit=limite) if parts: parts_com_stock = stock_api.process_parts_with_stock(parts, limit=None) if parts_com_stock: parts_com_stock = change_tracker.analisar("stock", parts_com_stock) stock_api.export_to_csv(parts_com_stock, CSV_STOCK) with open(JSON_STOCK, 'w', encoding='utf-8') as f: json.dump(parts_com_stock, f, ensure_ascii=False, indent=2) stock_api.gerar_html_stock(parts_com_stock, HTML_STOCK) print(f"✅ Stock atualizado: {len(parts_com_stock)} peças") else: print("⚠️ Nenhuma peça obtida.") except Exception as e: print(f"❌ Erro ao obter stock: {e}") import traceback traceback.print_exc() # ============================================================ # GUARDAR TIMESTAMP # ============================================================ with open(LOG_FILE, 'w', encoding='utf-8') as f: f.write(f"Última atualização: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"Total de OTs: {len(workorders) if workorders else 0}\n") f.write(f"Total de PIs: {len(workrequests) if workrequests else 0}\n") f.write(f"Total de OTs no Planning: {len(planning_items) if planning_data else 0}\n") f.write(f"Total de Peças (stock): {len(parts_com_stock) if 'parts_com_stock' in locals() else 0}\n") fim = datetime.now() duracao = (fim - inicio).total_seconds() print("\n" + "=" * 70) print(f"✅ Atualização concluída em {duracao:.1f}s às {fim.strftime('%H:%M:%S')}") print(f"📁 Ficheiros atualizados (sobrescritos):") print(f" 📄 {CSV_OTS}") print(f" 📄 {HTML_OTS}") print(f" 📄 {HTML_PIS}") print(f" 📄 {HTML_PLANNING}") print(f" 📄 {HTML_STOCK}") print(f"⏳ Próxima atualização em {intervalo} segundos...") print("=" * 70) except Exception as e: print(f"❌ Erro na atualização automática: {e}") import traceback traceback.print_exc() # Aguardar o intervalo definido time.sleep(intervalo) # ============================================================================ # MAIN # ============================================================================ def main(): USERNAME = "francisco.amorim" PASSWORD = "FAmorim2026" DESKTOP = os.path.expanduser("~/Desktop") # Limite de registos LIMITE = 10000 # Intervalo de atualização (5 minutos = 300 segundos) INTERVALO = 30 api = DimoMaintAPI() print("=" * 70) print("🔑 DIMO MAINT - EXPORTADOR AUTOMÁTICO") print("=" * 70) print(f"📊 LIMITE DE REGISTOS: {LIMITE}") print(f"🔄 ATUALIZAÇÃO A CADA: {INTERVALO} segundos ({INTERVALO/60:.1f} minutos)") print("=" * 70) print() # Tentar login inicial if api.login_with_selenium(USERNAME, PASSWORD, headless=False, force=False): print("✅ Login efetuado com sucesso!") # Iniciar servidor proxy print("\n🚀 A iniciar servidor proxy (resolução CORS)...") proxy_server = iniciar_servidor_proxy(porta=8888) # Configurar proxy configurar_proxy(api, USERNAME, PASSWORD) print("✅ Proxy configurado com renovação automática de token!") # Menu print("\n🔍 Opções:") print("1. Executar uma vez e sair") print("2. Executar em modo automático") print("3. Sair") opcao = input("\nEscolha uma opção (1/2/3): ").strip() if opcao == "1": print("\n📊 A executar extração única...") auto_update(api, USERNAME, PASSWORD, DESKTOP, LIMITE, 1) elif opcao == "2": print("\n🔄 A iniciar modo automático...") print(f"⚠️ Para parar, pressione Ctrl+C no terminal") auto_update(api, USERNAME, PASSWORD, DESKTOP, LIMITE, INTERVALO) else: print("👋 A sair...") proxy_server.shutdown() else: print("❌ Login falhou. Verifique as credenciais.") print("\n🔍 Pressiona Enter para fechar...") try: input() except KeyboardInterrupt: print("\n👋 Até logo!") if __name__ == "__main__": # Variável global para o proxy api_instance = None main()