dotfiles

ptman (6627B)

---

 #!/usr/bin/env python3
 import os
 import sys
 import json
 import subprocess
 import urllib.request
 import urllib.error
 import time
 import re
 from concurrent.futures import ThreadPoolExecutor
 
 API_URL = "https://translate.disroot.org/translate"
 TARGET_DIR = os.path.expanduser("~/.local/share/man/pt")
 CHUNK_LIMIT = 3000
 TRANSLATE_WORKERS = 3  # nº de chunks traduzidos em paralelo
 REQUEST_DELAY = 1.0    # respiro entre disparos, por worker
 
 
 def clean_backspaces(text):
     """Remove sequências de overstrike (char + backspace) usadas para negrito/sublinhado."""
     while '\x08' in text:
         text = re.sub(r'.\x08', '', text)
     return text
 
 
 def split_long_line(line, limit):
     """Quebra uma única linha maior que o limite em pedaços menores."""
     return [line[i:i + limit] for i in range(0, len(line), limit)] or [line]
 
 
 def translate_chunk(text):
     """Envia um bloco de texto para a API de tradução, com retentativas."""
     payload = {"q": text, "source": "auto", "target": "pt", "api_key": ""}
     req = urllib.request.Request(
         API_URL, data=json.dumps(payload).encode('utf-8'),
         headers={'Content-Type': 'application/json'}, method='POST'
     )
     for attempt in range(5):
         try:
             with urllib.request.urlopen(req, timeout=30) as response:
                 res_data = json.loads(response.read().decode('utf-8'))
                 return res_data.get("translatedText", "")
         except urllib.error.HTTPError as e:
             if e.code == 429:
                 time.sleep(3 * (attempt + 1))
                 continue
             return None  # erro definitivo do servidor
         except Exception:
             time.sleep(2)
             continue
     return None
 
 
 def build_chunks(text):
     """Divide o texto em blocos respeitando CHUNK_LIMIT."""
     lines = text.splitlines()
     chunks, current_chunk, current_length = [], [], 0
 
     for line in lines:
         if len(line) > CHUNK_LIMIT:
             if current_chunk:
                 chunks.append("\n".join(current_chunk))
                 current_chunk, current_length = [], 0
             chunks.extend(split_long_line(line, CHUNK_LIMIT))
             continue
 
         if current_length + len(line) + 1 > CHUNK_LIMIT:
             chunks.append("\n".join(current_chunk))
             current_chunk, current_length = [line], len(line)
         else:
             current_chunk.append(line)
             current_length += len(line) + 1
 
     if current_chunk:
         chunks.append("\n".join(current_chunk))
 
     return chunks
 
 
 def translate_text(text, status_prefix=""):
     """Traduz um texto inteiro, fatiando em chunks e mostrando progresso em paralelo."""
     chunks = build_chunks(text)
     total_chunks = len([c for c in chunks if c.strip()])
     if total_chunks == 0:
         return text
 
     results = [None] * len(chunks)
     done_count = [0]
 
     def worker(idx_chunk):
         idx, chunk = idx_chunk
         if not chunk.strip():
             results[idx] = chunk
             return
         translated = translate_chunk(chunk)
         results[idx] = translated if translated is not None else chunk
         done_count[0] += 1
         if status_prefix:
             pct = int((done_count[0] / total_chunks) * 100)
             print(
                 f"\r{status_prefix} -> Traduzindo... {done_count[0]}/{total_chunks} partes ({pct}%)",
                 end="", flush=True
             )
         time.sleep(REQUEST_DELAY)
 
     with ThreadPoolExecutor(max_workers=TRANSLATE_WORKERS) as pool:
         list(pool.map(worker, enumerate(chunks)))
 
     if status_prefix:
         print(f"\r{status_prefix} -> Concluído!{' ' * 25}", flush=True)
 
     return "\n".join(results)
 
 
 def resolve_man_path(args):
     """Resolve o caminho físico do manual via `man -w`."""
     try:
         output = subprocess.check_output(
             ["man", "-w"] + args, stderr=subprocess.DEVNULL
         ).decode('utf-8').strip()
     except subprocess.CalledProcessError:
         return None
 
     candidates = [p for p in output.splitlines() if p.strip()]
     return candidates[0] if candidates else None
 
 
 def process_manual(args):
     """Busca, traduz (se necessário) e exibe o manual especificado."""
     src_path = resolve_man_path(args)
     if not src_path:
         print(f"Erro: Manual não encontrado para: {' '.join(args)}")
         return False
 
     parts = src_path.split('/')
     if len(parts) < 2:
         print("Erro: Caminho do manual inválido.")
         return False
 
     section_dir, filename = parts[-2], parts[-1]
     clean_filename = filename.replace('.gz', '')
     dest_path = os.path.join(TARGET_DIR, section_dir, f"{clean_filename}.txt")
 
     # Retorna a versão em cache, se existir
     if os.path.exists(dest_path):
         subprocess.run(["vim", dest_path])
         return True
 
     print(f"-> Origem localizada: {src_path}")
     package_origin = "Desconhecido"
     
     try:
         apk_out = subprocess.check_output(
             ["apk", "info", "-W", src_path], stderr=subprocess.DEVNULL
         ).decode('utf-8').strip()
         if "is owned by" in apk_out:
             package_origin = apk_out.split("is owned by")[-1].strip()
     except Exception:
         pass
         
     print(f"-> Pacote: [{package_origin}]")
     print("-> Traduzindo sob demanda...")
 
     try:
         raw_text = subprocess.check_output(
             ["mandoc", "-Tutf8", src_path], stderr=subprocess.DEVNULL
         ).decode('utf-8', errors='ignore')
         raw_text = clean_backspaces(raw_text)
 
         if not raw_text.strip():
             return False
 
         translated = translate_text(raw_text, f"-> {clean_filename}")
 
         os.makedirs(os.path.dirname(dest_path), exist_ok=True)
         with open(dest_path, "w", encoding="utf-8") as f:
             f.write(translated)
 
         subprocess.run(["vim", dest_path])
         return True
 
     except Exception as e:
         print(f"\nFalha na tradução: {e}")
         print("Abrindo fallback em inglês...\n")
         time.sleep(1.5)
         subprocess.run(["man"] + args)
         return False
 
 
 def main():
     args = sys.argv[1:]
 
     if not args:
         print("Uso:")
         print("  ptman <cmd>        (Ex: ptman git)")
         print("  ptman <sec> <cmd>  (Ex: ptman 5 crontab)")
         print("  ptman clean        (Apaga o cache atual)")
         sys.exit(1)
 
     cmd = args[0].lower()
 
     if cmd == "clean":
         if os.path.exists(TARGET_DIR):
             subprocess.run(["rm", "-rf", TARGET_DIR])
             print("Cache limpo.")
         else:
             print("Cache já estava vazio.")
         sys.exit(0)
 
     process_manual(args)
 
 
 if __name__ == "__main__":
     main()