import os # Define para não procurar GPU; deve vir antes de 'import spacy' ou 'torch' os.environ["CUDA_VISIBLE_DEVICES"] = "" import re import sys import json import time import email import imaplib import smtplib import tempfile import logging import mimetypes import io import subprocess import shutil import codecs # <--- NOVO: Para corrigir os acentos no relatório from pathlib import Path from typing import Tuple, List, Dict, Optional from email.header import decode_header, make_header from email.message import EmailMessage from email.utils import parseaddr, make_msgid from email.mime.text import MIMEText from email.mime.image import MIMEImage from dotenv import load_dotenv from docx import Document from docx.shared import Cm, Pt from docx.enum.text import WD_ALIGN_PARAGRAPH # Importações para a função de extração robusta from docx.oxml.text.paragraph import CT_P from docx.oxml.table import CT_Tbl from docx.table import Table from docx.text.paragraph import Paragraph # Importação correta para XML from docx.oxml import parse_xml from docx.oxml.ns import qn # Bloco OpenAI try: from openai import OpenAI, APIError except ImportError: OpenAI = None APIError = None # Bloco LanguageTool try: import language_tool_python except ImportError: language_tool_python = None # ============================== # Constantes e Configurações # ============================== PALAVRAS_PERSONALIZADAS = [ 'multicomputador', 'multiprocessador', 'Flynn', 'telemóvel', 'hardware', 'software', 'backend', 'frontend', 'framework', 'INSUTEC', 'EISI', 'Arquitetura', 'Computadores', 'SQL', 'script', 'Rouget', 'Ruano', 'BFS', 'DFS', 'A*', 'gestão', 'Epifania', 'Rodrigues', 'algoritmo.py', 'desempenho.py', 'Debruçe', 'lab.ed-consulting.ao', 'lab.insutec.ao', 'mnt', 'ed', 'consulting' ] logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] - %(message)s", handlers=[logging.FileHandler("app.log", encoding='utf-8'), logging.StreamHandler(sys.stdout)] ) load_dotenv() IMAP_HOST = os.getenv("IMAP_HOST", "").strip() IMAP_USER = os.getenv("IMAP_USER", "").strip() IMAP_PASS = os.getenv("IMAP_PASS", "").strip() IMAP_LABEL = os.getenv("IMAP_LABEL", "INBOX").strip() SMTP_HOST = os.getenv("SMTP_HOST", "").strip() SMTP_PORT = int(os.getenv("SMTP_PORT", "465")) SMTP_USER = os.getenv("SMTP_USER", "").strip() SMTP_PASS = os.getenv("SMTP_PASS", "").strip() USE_SMTP_SSL = os.getenv("USE_SMTP_SSL", "true").lower() == "true" OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "").strip() OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini").strip() DISCIPLINA_FALLBACK = os.getenv("DISCIPLINA", "Não Identificada").strip() ASSUNTO_PROVA_PALAVRA_CHAVE = os.getenv("ASSUNTO_PROVA_PREFIXO", "Enunciados").strip() MAX_ISSUES = int(os.getenv("MAX_ISSUES", "15")) EMAIL_APROVADO_PARA = os.getenv("EMAIL_APROVADO_PARA", "rouget.fundora@gmail.com").strip() EMAIL_CC_GERAL = os.getenv("EMAIL_CC_GERAL", "rouget.ruano@insutec.ao").strip() ASSINATURA = os.getenv("ASSINATURA", "Atenciosamente,\nEquipa de Validação").strip() SCRIPT_DIR = Path(__file__).resolve().parent SIGN_IMAGE_PATH_FROM_ENV = os.getenv("SIGN_IMAGE_PATH", "./img/assina.png").strip() SIGN_IMAGE_PATH_ABSOLUTE = SCRIPT_DIR / SIGN_IMAGE_PATH_FROM_ENV SIGN_IMAGE_PATH = str(SIGN_IMAGE_PATH_ABSOLUTE) SIGN_IMAGE_WIDTH_CM = float(os.getenv("SIGN_IMAGE_WIDTH_CM", "5")) REPLY_SENDER_ON_ERROR_FLAG = os.getenv("REPLY_SENDER_ON_ERROR", "true").lower() == "true" # =================================================================== # FUNÇÕES DE CARREGAMENTO DE MODELOS # =================================================================== try: logging.info("A iniciar LanguageTool (pode demorar um pouco na 1ª vez)...") # Carrega o corretor para Português de Portugal (e Angola) LANG_TOOL = language_tool_python.LanguageTool('pt-PT') logging.info("LanguageTool carregado com sucesso.") except Exception as e: logging.error(f"Falha ao carregar LanguageTool (Java instalado?): {e}") LANG_TOOL = None # Se falhar, continua sem bloquear o script # ============================== # Funções Utilitárias # ============================== def clean_header(raw_header: Optional[str]) -> str: if not raw_header: return "" try: return str(make_header(decode_header(raw_header))) except Exception: return raw_header def get_text_from_xml_element(element): if element is None: return "" try: text_nodes = element.xpath(".//*[local-name()='t']") if text_nodes: return " ".join(node.text for node in text_nodes if node.text) except Exception as e: logging.warning(f"Erro menor ao extrair nó de texto: {e}") return "" def extract_full_docx_text(bytes_content: bytes) -> str: full_text = [] try: with io.BytesIO(bytes_content) as file_stream: doc = Document(file_stream) for section in doc.sections: for header in (section.header, section.first_page_header, section.even_page_header, section.footer, section.first_page_footer, section.even_page_footer): if header is None: continue full_text.append(get_text_from_xml_element(header._element)) full_text.append("\n--- CONTEÚDO PRINCIPAL DO DOCUMENTO ---\n") for block in doc.element.body: full_text.append(get_text_from_xml_element(block)) raw_text = "\n".join(full_text) clean_text = re.sub(r'[\n\r\t]+', ' ', raw_text) return clean_text.strip() except Exception as e: logging.error(f"Falha CRÍTICA ao extrair texto do DOCX: {e}", exc_info=True) return "" # ============================== # Funções de Verificação # ============================== def spelling_issues_pt(tool, text: str, custom_words: List[str]) -> List[Dict]: if tool is None: return [] try: if hasattr(tool, 'disabled_words'): tool.disabled_words.update(custom_words) elif hasattr(tool, 'disable_spellchecking_for_words'): tool.disable_spellchecking_for_words(custom_words) matches = tool.check(text or "") issues = [] for m in matches: if m.category in {'TYPOGRAPHY', 'STYLE', 'WHITESPACE', 'PUNCTUATION', 'GRAMMAR', 'COLLOQUIALISMS', 'REDUNDANCY', 'CASING', 'SPELLING'}: if m.ruleId not in ('MORFOLOGIK_RULE_PT_PT', 'UPPERCASE_SPELLING'): continue # Cria um contexto visual para o erro ctx = (text[max(0, m.offset - 20):m.offset] + f"[[{text[m.offset:m.offset+m.errorLength]}]]" + text[m.offset+m.errorLength:m.offset+m.errorLength+20]) issues.append({ "message": m.message, "context": ctx, "rule": m.ruleId }) return issues except Exception: return [] def try_extract_discipline_regex(text: str) -> Optional[str]: if not text: return None header_sample = text[:3000] lookahead = r"Ano\s*Lec?tivo\s*:|Data\s*:|Curso\s*:|Docente\s*:|Nome\s*:|Prova\s*:|Duração\s*:|Nº\s*:|Turma\s*:|Ano\s*:" patterns = [ r"(?i)(?:Disciplina|Cadeira|Unidade Curricular|Curricular Unit)\s*[:\.]\s*(.+?)\s*(?=" + lookahead + r")", r"(?i)(?:Assunto)\s*[:\.]\s*(.+?)\s*(?=" + lookahead + r")" ] for pattern in patterns: match = re.search(pattern, header_sample) if match: return match.group(1).strip().rstrip(".,;") fallback = r"(?i)(?:Disciplina|Cadeira|Unidade Curricular|Curricular Unit)\s*[:\.]\s*([^\n\r]+)" match = re.search(fallback, header_sample) if match: return match.group(1).strip().rstrip(".,;") return None def evaluate_proof_content(text: str, fallback: str) -> Tuple[str, bool, str, str]: if not OPENAI_API_KEY: return fallback, False, "Validação IA desativada.", "" disciplina_detectada = try_extract_discipline_regex(text) client = OpenAI(api_key=OPENAI_API_KEY, timeout=30.0) instrucao = f"A disciplina identificada é '{disciplina_detectada}'." if disciplina_detectada else "Tente deduzir a disciplina." prompt = f""" Aja como um Coordenador Pedagógico. Analise o texto de uma prova. {instrucao} Responda APENAS em JSON: {{"disciplina_identificada": "Nome", "permissivel": true|false, "justificativa": "...", "sugestoes": "..."}} Critério: 'permissivel' deve ser false se o conteúdo for de uma matéria completamente errada. """ try: resp = client.chat.completions.create( model=OPENAI_MODEL, messages=[{"role": "system", "content": prompt}, {"role": "user", "content": text}], temperature=0.0, response_format={"type": "json_object"} ) data = json.loads(resp.choices[0].message.content or "{}") final_disc = disciplina_detectada if disciplina_detectada else data.get("disciplina_identificada", fallback) return final_disc, data.get("permissivel", False), data.get("justificativa", ""), data.get("sugestoes", "") except Exception as e: logging.error(f"Erro OpenAI: {e}") return fallback, False, "Erro IA", "" # ==================================================================== # Funções de Documento (ASSINATURA FLUTUANTE) # ==================================================================== def add_signature_to_doc(doc_bytes: bytes, image_path_str: str, image_width_cm: float) -> Optional[bytes]: image_path = Path(image_path_str) if not image_path.is_file(): logging.warning(f"Imagem de assinatura NÃO encontrada em {image_path_str}. Retornando doc original.") return doc_bytes try: doc_stream = io.BytesIO(doc_bytes) doc = Document(doc_stream) def insert_floating_image(paragraph, image_path_str, width_cm): run = paragraph.add_run() rId, image = paragraph.part.get_or_add_image(image_path_str) width_emu = int(width_cm * 360000) img_size = image.px_width, image.px_height height_emu = int(width_emu * img_size[1] / img_size[0]) graphic_xml = f""" right 0 """ run._r.append(parse_xml(graphic_xml)) section = doc.sections[0] footer = section.footer p = footer.add_paragraph() p.alignment = WD_ALIGN_PARAGRAPH.RIGHT insert_floating_image(p, str(image_path), image_width_cm) new_doc_stream = io.BytesIO() doc.save(new_doc_stream) return new_doc_stream.getvalue() except Exception as e: logging.error(f"Falha ao adicionar assinatura flutuante: {e}", exc_info=True) return None def convert_docx_to_pdf_linux(docx_bytes: bytes) -> bytes: soffice_cmd = shutil.which("libreoffice") or shutil.which("soffice") if not soffice_cmd: raise FileNotFoundError("LibreOffice não encontrado.") with tempfile.TemporaryDirectory() as temp_dir: temp_dir_path = Path(temp_dir) docx_path = temp_dir_path / "input.docx" with open(docx_path, "wb") as f: f.write(docx_bytes) cmd = [soffice_cmd, "--headless", "--convert-to", "pdf", "--outdir", temp_dir, str(docx_path)] subprocess.run(cmd, capture_output=True, text=True, timeout=45, check=True) pdf_path = temp_dir_path / "input.pdf" if not pdf_path.exists(): raise FileNotFoundError("PDF não criado.") with open(pdf_path, "rb") as f: return f.read() def send_email(to_addr: str, subject: str, body: str, attachments: List[Tuple[str, bytes]] = None, cc_addr: Optional[str] = None, assinatura: str = "", sign_image_path: str = "", sign_image_width_cm: float = 5.0): if not all([SMTP_HOST, SMTP_USER, SMTP_PASS]): return msg = EmailMessage() msg["From"] = SMTP_USER; msg["To"] = to_addr; msg["Subject"] = subject if cc_addr: msg["Cc"] = cc_addr assinatura_html = "" if assinatura: assinatura_html = f"

{assinatura.replace(chr(10), '
')}

" msg.set_content(f"{body}\n\n{assinatura}", 'plain', 'utf-8') msg.add_alternative(f"

{body.replace(chr(10), '
')}

{assinatura_html}", 'html', 'utf-8') if attachments: for fname, fbytes in attachments: try: # Deteta se é texto (como o relatório) e força UTF-8 maintype, subtype = (mimetypes.guess_type(fname)[0] or 'application/octet-stream').split('/') if fname.endswith('.txt'): # MIMEText lida melhor com charset part = MIMEText(fbytes.decode('utf-8-sig', errors='replace'), 'plain', 'utf-8') part.add_header('Content-Disposition', 'attachment', filename=fname) msg.attach(part) else: msg.add_attachment(fbytes, maintype=maintype, subtype=subtype, filename=fname) except Exception as e: logging.warning(f"Erro ao anexar {fname}: {e}") with (smtplib.SMTP_SSL if USE_SMTP_SSL else smtplib.SMTP)(SMTP_HOST, SMTP_PORT) as s: if not USE_SMTP_SSL: s.starttls() s.login(SMTP_USER, SMTP_PASS) s.send_message(msg) # ==================================================================== # Processamento Principal # ==================================================================== def process_inbox(): if not all([IMAP_HOST, IMAP_USER, IMAP_PASS]): return imap = imaplib.IMAP4_SSL(IMAP_HOST) imap.login(IMAP_USER, IMAP_PASS) imap.select(f'"{IMAP_LABEL}"') status, data = imap.search(None, '(UNSEEN)') if status != "OK" or not data[0]: return for num in data[0].split(): try: res, raw_full = imap.fetch(num, '(BODY.PEEK[])') if res != 'OK': continue msg = email.message_from_bytes(raw_full[0][1]) subject = clean_header(msg.get("Subject", "")) if ASSUNTO_PROVA_PALAVRA_CHAVE.lower() not in subject.lower(): continue logging.info(f"Processando: {subject}") sender_email = parseaddr(clean_header(msg.get("From", "")))[1] docx_parts = [] for p in msg.walk(): if p.get_content_type() == "application/vnd.openxmlformats-officedocument.wordprocessingml.document": fname = p.get_filename() or "anexo.docx" if not fname.startswith('~$'): payload = p.get_payload(decode=True) if payload: docx_parts.append((clean_header(fname), payload)) if not docx_parts: imap.store(num, '+FLAGS', '\\Seen') continue for fname, file_bytes in docx_parts: text = extract_full_docx_text(file_bytes) if not text: continue # --- Construção do Relatório (Ortografia + IA) --- issues = spelling_issues_pt(LANG_TOOL, text, PALAVRAS_PERSONALIZADAS) # Formata os erros ortográficos para o texto do e-mail/relatório if LANG_TOOL is None: ortografia_texto = "Estado: Desativada (Módulo não carregado)" elif not issues: ortografia_texto = "Estado: Aprovado (Nenhum erro relevante encontrado)" else: ortografia_texto = f"Estado: {len(issues)} erros encontrados.\n" for i, issue in enumerate(issues, 1): ortografia_texto += f" {i}. {issue['message']}\n No contexto: {issue['context']}\n" disciplina, is_ia_ok, justif, sugestoes = evaluate_proof_content(text, DISCIPLINA_FALLBACK) aprovado = is_ia_ok and (len(issues) <= MAX_ISSUES) # --- Conteúdo do Relatório Formatado --- report = ( f"RELATÓRIO DE VALIDAÇÃO\n" f"========================================\n" f"STATUS FINAL: {'APROVADO' if aprovado else 'REVISÃO NECESSÁRIA'}\n" f"Disciplina: {disciplina}\n" f"Ficheiro: {fname}\n" f"----------------------------------------\n" f"[1] ANÁLISE DE CONTEÚDO (IA)\n" f"Justificativa: {justif}\n" f"Sugestões: {sugestoes}\n" f"----------------------------------------\n" f"[2] ANÁLISE ORTOGRÁFICA\n" f"{ortografia_texto}\n" f"----------------------------------------\n" ) # Adiciona BOM (Byte Order Mark) para forçar UTF-8 no Windows att_report_bytes = codecs.BOM_UTF8 + report.encode('utf-8') att_report = ("relatorio.txt", att_report_bytes) email_args = {"assinatura": ASSINATURA, "sign_image_path": SIGN_IMAGE_PATH} if aprovado: doc_assinado = add_signature_to_doc(file_bytes, SIGN_IMAGE_PATH, SIGN_IMAGE_WIDTH_CM) if doc_assinado: try: pdf_bytes = convert_docx_to_pdf_linux(doc_assinado) final_anexo = (f"ASSINADO_{Path(fname).stem}.pdf", pdf_bytes) except: final_anexo = (f"ASSINADO_{fname}", doc_assinado) # Envia ao Coordenador: Prova assinada + Relatório send_email(EMAIL_APROVADO_PARA, f"[APROVADO] {disciplina}", "Segue anexo assinado e relatório de validação.", [final_anexo, att_report], EMAIL_CC_GERAL, **email_args) # Envia ao Professor: Relatório + Ficheiro Original send_email(sender_email, f"[APROVADO] {disciplina}", "Seu enunciado foi aprovado. Segue o relatório.", [att_report, (fname, file_bytes)], **email_args) else: send_email(sender_email, f"[ERRO] {disciplina}", "Aprovado, mas erro técnico ao assinar.", [att_report], **email_args) else: send_email(sender_email, f"[REVISÃO] {disciplina}", "Necessária revisão. Ver relatório anexo.", [att_report, (fname, file_bytes)], **email_args) imap.store(num, '+FLAGS', '\\Seen') except Exception as e: logging.error(f"Erro no email {num}: {e}") imap.close() imap.logout() def main(): try: process_inbox() except Exception as e: logging.critical(f"Fatal: {e}") if __name__ == "__main__": main()