# comentarios.py - Sync de comentários das OTs (paralelo + cache local) import os import json import time import asyncio import aiohttp from datetime import datetime from concurrent.futures import ThreadPoolExecutor BASE_URL = "https://prod.mydimomaintmx.cloud" TENANT = "SERRATEC" DESKTOP = os.path.expanduser("~/Desktop") COMENTARIOS_FILE = os.path.join(DESKTOP, "comentarios.json") COMENTARIOS_CACHE = os.path.join(DESKTOP, "comentarios_cache.json") def _carregar_json(path, default): if os.path.exists(path): try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except Exception: return default return default def _guardar_json(path, data): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) async def _fetch_ot(session, ot_id, token, semaphore, timeout=30): """Busca os detalhes completos de uma OT (incluindo comentários).""" url = f"{BASE_URL}/{TENANT}/api/workOrderReader/Get" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "tenant": TENANT, } params = {"id": ot_id} async with semaphore: for tentativa in range(3): try: async with session.get( url, headers=headers, params=params, timeout=aiohttp.ClientTimeout(total=timeout) ) as resp: if resp.status == 200: return ot_id, await resp.json() elif resp.status == 401: return ot_id, {"_error": "401_TOKEN_EXPIRADO"} elif resp.status == 429: await asyncio.sleep(2 ** tentativa) continue else: return ot_id, {"_error": f"HTTP_{resp.status}"} except (asyncio.TimeoutError, aiohttp.ClientError) as e: if tentativa == 2: return ot_id, {"_error": f"CONN_{type(e).__name__}"} await asyncio.sleep(1 + tentativa) return ot_id, {"_error": "MAX_RETRIES"} async def _fetch_all(ot_ids, token, concorrencia=15): """Busca todas as OTs em paralelo, com limite de concorrência.""" semaphore = asyncio.Semaphore(concorrencia) connector = aiohttp.TCPConnector(limit=concorrencia * 2, limit_per_host=concorrencia) async with aiohttp.ClientSession(connector=connector) as session: tarefas = [_fetch_ot(session, ot_id, token, semaphore) for ot_id in ot_ids] resultados = await asyncio.gather(*tarefas) return dict(resultados) def _extrair_comentarios(ot_detalhes): """Extrai apenas os campos que interessam para pesquisa.""" comentarios = [] for c in (ot_detalhes.get("commentList") or []): comentarios.append({ "author": (c.get("user") or {}).get("fullName", "Desconhecido"), "date": c.get("creationDateTime"), "text": c.get("comment") or c.get("commentHtml") or "", }) assinaturas = [] for s in (ot_detalhes.get("signatures") or []): assinaturas.append({ "signer": (s.get("signatory") or {}).get("fullName", "Desconhecido"), "date": s.get("signatureDateTime"), }) return { "code": ot_detalhes.get("code"), "name": ot_detalhes.get("name"), "state": (ot_detalhes.get("state") or {}).get("name"), "assetCode": (ot_detalhes.get("asset") or {}).get("code"), "assetName": (ot_detalhes.get("asset") or {}).get("name"), "beginDateTime": ot_detalhes.get("beginDateTime"), "endDateTime": ot_detalhes.get("endDateTime"), "lastModificationDateTime": ot_detalhes.get("lastModificationDateTime"), "technician": ot_detalhes.get("technician"), "remedy": (ot_detalhes.get("remedy") or {}).get("name") if isinstance(ot_detalhes.get("remedy"), dict) else ot_detalhes.get("remedy"), "comments": comentarios, "signatures": assinaturas, "synced_at": datetime.now().isoformat(), } def sync_comentarios(api, limite=None, concorrencia=15, apenas_abertas=True): """ Sincroniza comentários de todas as OTs (ou apenas das abertas). Args: api: instância de DimoMaintAPI (para aceder ao token) limite: máximo de OTs a sincronizar (None = todas) concorrencia: número de pedidos em paralelo apenas_abertas: se True, só sincroniza OTs não encerradas + recentes """ if not api.token: print("❌ Sem token para sincronizar comentários.") return # 1. Carregar lista de OTs workorders = _carregar_json(os.path.join(DESKTOP, "workorders.json"), []) if not workorders: print("⚠️ workorders.json vazio ou inexistente. Executa o sync principal primeiro.") return # 2. Carregar cache existente (comentários já sincronizados) cache = _carregar_json(COMENTARIOS_CACHE, {}) # 3. Determinar que OTs precisam de sync a_sincronizar = [] for wo in workorders: ot_id = wo.get("id") if not ot_id: continue # Filtro: apenas abertas + modificadas recentemente if apenas_abertas: estado = wo.get("stateId", "") # 'ada41e02...' = Encerrada | 'd67894ef...' = Anulada if estado in ("ada41e02-5e6a-4dc7-9f5c-e6b14340f1a6", "d67894ef-2d53-4e63-a840-71089d2cc851"): continue # Verificar se já está em cache e não mudou cached = cache.get(ot_id) if cached: last_mod = wo.get("lastModificationDateTime") or wo.get("beginDateTime") if cached.get("lastModificationDateTime") == last_mod: continue # nada mudou, salta a_sincronizar.append(ot_id) if limite: a_sincronizar = a_sincronizar[:limite] if not a_sincronizar: print("✅ Nada para sincronizar — cache está atualizada.") return print(f"📥 A sincronizar {len(a_sincronizar)} OTs (concorrência={concorrencia})...") t0 = time.time() # 4. Fetch em paralelo resultados = asyncio.run(_fetch_all(a_sincronizar, api.token, concorrencia)) # 5. Processar resultados novos, erros_token = 0, 0 for ot_id, detalhes in resultados.items(): if "_error" in detalhes: if detalhes["_error"] == "401_TOKEN_EXPIRADO": erros_token += 1 continue cache[ot_id] = _extrair_comentarios(detalhes) novos += 1 # 6. Guardar _guardar_json(COMENTARIOS_CACHE, cache) # 7. Gerar também um JSONL mais leve (só o que interessa para pesquisa) jsonl_path = os.path.join(DESKTOP, "comentarios.jsonl") with open(jsonl_path, "w", encoding="utf-8") as f: for ot_id, dados in cache.items(): f.write(json.dumps({"id": ot_id, **dados}, ensure_ascii=False) + "\n") dur = time.time() - t0 print(f"✅ Sync concluída em {dur:.1f}s") print(f" 📥 Novos/atualizados: {novos}") print(f" ❌ Erros de token: {erros_token}") print(f" 💾 Cache total: {len(cache)} OTs") print(f" 📄 {COMENTARIOS_CACHE}") print(f" 📄 {jsonl_path} ({os.path.getsize(jsonl_path)/1024/1024:.2f} MB)")