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 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' ] LISTA_PALAVRAS_INAPROPRIADAS = [ 'merda', 'porra', 'caralho', 'foder', 'puta', 'cona', 'grosseira', 'estúpido', 'idiota' ] 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 # =================================================================== LANG_TOOL = None # ============================== # Funções Utilitárias (Extração XML) # ============================== 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 pass 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]: """ Adiciona a assinatura no RODAPÉ de TODAS as páginas, mas como uma imagem FLUTUANTE (Floating) e 'Behind Text'. """ 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]) # XML com w:drawing encapsulando wp:anchor (CORRIGIDO) 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: maintype, subtype = (mimetypes.guess_type(fname)[0] or 'application/octet-stream').split('/') msg.add_attachment(fbytes, maintype=maintype, subtype=subtype, filename=fname) except Exception: pass 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 issues = spelling_issues_pt(LANG_TOOL, text, PALAVRAS_PERSONALIZADAS) disciplina, is_ia_ok, justif, sugestoes = evaluate_proof_content(text, DISCIPLINA_FALLBACK) aprovado = is_ia_ok and (len(issues) <= MAX_ISSUES) report = f"VALIDAÇÃO: {'APROVADO' if aprovado else 'REVISÃO'}\nDisciplina: {disciplina}\nJustificativa: {justif}" att_report = ("relatorio.txt", report.encode('utf-8')) 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) # --- CORREÇÃO AQUI: Agora envia o relatório (att_report) também para o Coordenador --- send_email(EMAIL_APROVADO_PARA, f"[APROVADO] {disciplina}", "Segue anexo assinado e relatório.", [final_anexo, att_report], EMAIL_CC_GERAL, **email_args) send_email(sender_email, f"[APROVADO] {disciplina}", "Seu enunciado foi aprovado.", [att_report, (fname, file_bytes)], **email_args) else: send_email(sender_email, f"[ERRO] {disciplina}", "Erro técnico ao assinar.", [att_report], **email_args) else: send_email(sender_email, f"[REVISÃO] {disciplina}", "Necessária revisão.", [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()