import os import re import sys import json import time import email import imaplib import smtplib import tempfile import logging 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 from email.mime.text import MIMEText from dotenv import load_dotenv from docx import Document from docx.shared import Cm try: from openai import OpenAI, APIError except ImportError: OpenAI = None APIError = None 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' ] 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")) DESTINO = os.getenv("DESTINO", "").strip() # ============================== # 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 # ALTERADO: Função para ler o documento completo, incluindo cabeçalhos e tabelas def extract_full_docx_text(bytes_content: bytes) -> str: """Extrai texto dos cabeçalhos, tabelas (no cabeçalho e corpo) e parágrafos.""" full_text = [] with tempfile.NamedTemporaryFile(suffix=".docx", delete=True) as tmp: tmp.write(bytes_content) tmp.flush() doc = Document(tmp.name) # 1. Extrair texto de todos os cabeçalhos for section in doc.sections: for header in (section.header, section.first_page_header, section.even_page_header): if header is None: continue for table in header.tables: for row in table.rows: for cell in row.cells: full_text.append(cell.text) for paragraph in header.paragraphs: full_text.append(paragraph.text) full_text.append("\n--- CONTEÚDO PRINCIPAL DO DOCUMENTO ---\n") # 2. Extrair texto do corpo principal for paragraph in doc.paragraphs: full_text.append(paragraph.text) for table in doc.tables: for row in table.rows: for cell in row.cells: full_text.append(cell.text) return "\n".join(full_text).strip() def spelling_issues_pt(text: str, custom_words: List[str]) -> List[Dict]: if language_tool_python is None: return [] try: tool = language_tool_python.LanguageTool('pt-BR') 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: context = (text[max(0, m.offset - 30):m.offset] + f"⟦{text[m.offset:m.offset+m.errorLength]}⟧" + text[m.offset+m.errorLength:m.offset+m.errorLength+30]) issues.append({"message": m.message, "context": context}) return issues except Exception as e: logging.warning(f"LanguageTool indisponível: {e}."); return [] # ALTERADO: Prompt da IA para dar prioridade ao cabeçalho def evaluate_proof_content(text: str, fallback: str) -> Tuple[str, bool, str, str]: if not OPENAI_API_KEY: return fallback, False, "Validação OpenAI desativada (API_KEY ausente ou inválida no ficheiro .env).", "" client = OpenAI(api_key=OPENAI_API_KEY) prompt = f""" Aja como um Coordenador de Curso. Analise o seguinte texto extraído de um documento Word, que contém primeiro o texto do cabeçalho, seguido do corpo principal. Siga estas tarefas por ordem de PRIORIDADE: 1. **Identificar a Disciplina (TAREFA CRÍTICA):** Leia o texto e encontre o nome EXATO da disciplina. Dê prioridade absoluta ao texto que aparece antes de '--- CONTEÚDO PRINCIPAL DO DOCUMENTO ---'. Procure por etiquetas como "Disciplina:", "Unidade Curricular:", "Curso:". NÃO deduza ou infira a disciplina a partir do conteúdo das perguntas se um nome explícito estiver presente. 2. **Análise de Conteúdo:** Usando o nome da disciplina que identificou, avalie se o conteúdo é relevante e adequado. 3. **Decisão e Sugestões:** Decida se o enunciado é "Permissível" ou "Requer Revisão" e forneça sugestões. Responda APENAS em JSON: {{"disciplina_identificada": "O nome EXATO encontrado", "permissivel": true|false, "justificativa": "análise", "sugestoes": "sugestões"}} """ try: resp = client.chat.completions.create(model=OPENAI_MODEL, messages=[{"role": "user", "content": prompt}, {"role": "assistant", "content": text}], temperature=0.1, response_format={"type": "json_object"}) data = json.loads(resp.choices[0].message.content or "{}") return (data.get("disciplina_identificada", fallback), data.get("permissivel", False), data.get("justificativa", "N/A"), data.get("sugestoes", "N/A")) except Exception as e: logging.error(f"Falha na API da OpenAI: {e}"); return fallback, False, "Erro na comunicação com a IA.", "" def send_email(to_addr: str, subject: str, body: str, attachments: List[Tuple[str, bytes]] = None): if not all([SMTP_HOST, SMTP_USER, SMTP_PASS]): raise ConnectionError("Configurações de SMTP ausentes.") msg = EmailMessage(); msg["From"], msg["To"], msg["Subject"] = SMTP_USER, to_addr, subject msg.set_content(body, 'plain', 'utf-8') if attachments: for fname, fbytes in attachments: if fname.lower().endswith('.txt'): report_part = MIMEText(fbytes.decode('utf-8'), 'plain', 'utf-8') report_part.add_header('Content-Disposition', 'attachment', filename=fname) msg.attach(report_part) else: msg.add_attachment(fbytes, maintype='application', subtype='octet-stream', filename=fname) try: smtp_class = smtplib.SMTP_SSL if USE_SMTP_SSL else smtplib.SMTP with smtp_class(SMTP_HOST, SMTP_PORT) as s: if not USE_SMTP_SSL: s.starttls() s.login(SMTP_USER, SMTP_PASS); s.send_message(msg) except Exception as e: logging.error(f"Falha ao enviar e-mail para {to_addr}: {e}"); raise def process_inbox(): if not all([IMAP_HOST, IMAP_USER, IMAP_PASS, DESTINO]): logging.error("Configurações de IMAP ou DESTINO ausentes."); return imap = None try: 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]: logging.info("Nenhum e-mail não lido encontrado."); return email_ids = data[0].split() logging.info(f"Encontrados {len(email_ids)} e-mails não lidos para verificar.") for num in email_ids: try: # Usa BODY.PEEK[] para ler sem marcar como lido 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(): logging.info(f"Ignorando e-mail (assunto '{subject}' não é de enunciado). Permanece não lido.") continue logging.info(f"E-mail de enunciado encontrado: '{subject}'. A processar...") from_header = clean_header(msg.get("From", "")) docx_parts = [] for part in msg.walk(): filename = clean_header(part.get_filename()) if filename and filename.lower().endswith(".docx"): docx_parts.append((filename, part.get_payload(decode=True))) if not docx_parts: logging.warning(f"E-mail de enunciado de '{from_header}' não continha anexo .docx.") continue for fname, file_bytes in docx_parts: # ALTERADO: Usa a nova função de extração text = extract_full_docx_text(file_bytes) if not text: continue issues = spelling_issues_pt(text, PALAVRAS_PERSONALIZADAS) disciplina, is_permissible, justification, suggestions = evaluate_proof_content(text, DISCIPLINA_FALLBACK) report_content = f"RELATÓRIO DE VALIDAÇÃO AUTOMÁTICA\n====================================\n- Remetente: {from_header}\n- Ficheiro: {fname}\n- Disciplina Identificada (IA): {disciplina}\n---\nANÁLISE ORTOGRÁFICA\n- Problemas: {len(issues)} (Tolerância: {MAX_ISSUES})\n- Status: {'Aceitável' if len(issues) <= MAX_ISSUES else 'Requer Atenção'}\n---\nANÁLISE DE CONTEÚDO (IA)\n- Decisão: {'Permissível' if is_permissible else 'Requer Revisão'}\n- Justificativa: {justification}\n- Sugestões: {suggestions}" email_body = f"Prezado(a) Coordenador(a),\n\nUm novo enunciado de prova foi analisado pelo sistema.\n- Disciplina: {disciplina}\n- Decisão da IA: {'Permissível' if is_permissible else 'Requer Revisão'}\n\nO enunciado original e o relatório detalhado estão em anexo." attachments = [(fname, file_bytes), ("relatorio_validacao.txt", report_content.encode('utf-8'))] send_email(DESTINO, f"[{'PERMISSÍVEL' if is_permissible else 'REVISÃO'}] Enunciado: {disciplina}", email_body, attachments) logging.info(f"Relatório sobre o e-mail de '{from_header}' enviado para {DESTINO}.") # Apenas depois de processar com sucesso, marca como lido imap.store(num, '+FLAGS', '\\Seen') logging.info(f"E-mail original de '{from_header}' foi processado e marcado como lido.") except Exception as e: logging.error(f"Erro ao processar e-mail ID {num.decode()}: {e}", exc_info=True) finally: if imap: try: imap.close(); imap.logout() except Exception: pass def main(): try: process_inbox() except Exception as e: logging.critical(f"Erro fatal na execução principal: {e}", exc_info=True) sys.exit(1) if __name__ == "__main__": logging.info("Iniciando serviço de validação de enunciados.") main() logging.info("Serviço de validação finalizado.")