import os # Define para não procurar GPU 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 from pathlib import Path from typing import Tuple, List, Dict, Optional, Set 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 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 # ============================== # Configurações e Listas # ============================== 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() # --- LISTA BRANCA (TERMOS TÉCNICOS VÁLIDOS) --- # O script vai ignorar qualquer palavra que esteja aqui, maiúscula ou minúscula. WHITELIST_TECNICA = { # --- CORREÇÃO: Termos adicionados para evitar falsos positivos recentes --- 'smpp', 'osi', 'comumente', 'resolution', 'protocol', # TI / Programação 'software', 'hardware', 'mouse', 'keyboard', 'backend', 'frontend', 'fullstack', 'framework', 'api', 'rest', 'json', 'xml', 'html', 'css', 'sql', 'nosql', 'docker', 'kubernetes', 'linux', 'windows', 'macos', 'android', 'ios', 'ubuntu', 'python', 'java', 'javascript', 'typescript', 'php', 'ruby', 'c++', 'c#', 'server', 'client', 'host', 'localhost', 'ip', 'tcp', 'udp', 'http', 'https', 'ssh', 'ftp', 'smtp', 'imap', 'pop3', 'dhcp', 'dns', 'vpn', 'lan', 'wan', 'wlan', 'router', 'switch', 'firewall', 'proxy', 'gateway', 'bandwidth', 'throughput', 'bug', 'debug', 'deploy', 'build', 'script', 'code', 'git', 'github', 'commit', 'login', 'logout', 'signin', 'signup', 'password', 'username', 'email', 'url', 'browser', 'cookie', 'cache', 'token', 'auth', 'hash', 'algorithm', 'array', 'string', 'int', 'float', 'bool', 'class', 'object', 'void', 'null', 'nan', 'upload', 'download', 'online', 'offline', 'backup', 'database', 'data', 'pixel', 'resolution', 'bit', 'byte', 'kb', 'mb', 'gb', 'tb', 'hz', 'wifi', 'wi-fi', 'bluetooth', 'iot', 'ai', 'ml', 'bot', 'crypto', 'blockchain', 'interface', 'ui', 'ux', 'web', 'app', 'desktop', 'mobile', 'smart', 'icmp', 'arp', 'rip', 'ospf', 'bgp', 'mpls', 'qos', 'nat', 'mac', 'vlan', 'cpu', 'gpu', 'ram', 'rom', 'ssd', 'hdd', 'usb', 'hdmi', 'bios', 'uefi', 'driver', 'file', 'folder', 'directory', 'root', 'admin', 'sudo', 'install', # Matemática / Física / Química 'sen', 'cos', 'tan', 'cot', 'sec', 'csc', 'log', 'ln', 'lim', 'det', 'mmc', 'mdc', 'joule', 'newton', 'watt', 'volt', 'ampere', 'ohm', 'hertz', 'pascal', 'coulomb', 'farad', 'henry', 'tesla', 'weber', 'lumen', 'lux', 'mol', 'kpa', 'mpa', 'gpa', 'ph', 'ion', 'atom', 'molecule', 'dna', 'rna', # Contexto Angola / Institucional 'insutec', 'eisi', 'ert', 'isp', 'rouget', 'ruano', 'epifania', 'rodrigues', 'eleuterio', 'moma', 'angola', 'luanda', 'kz', 'ao', 'docente', 'discente', '1ª', '2ª', '3ª', '4ª', '1º', '2º', '3º', '4º' } 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" # =================================================================== # Carregamento LanguageTool # =================================================================== try: logging.info("A iniciar LanguageTool...") LANG_TOOL = language_tool_python.LanguageTool('pt-PT') logging.info("LanguageTool carregado.") except Exception as e: logging.error(f"Falha LanguageTool: {e}") LANG_TOOL = None # ============================== # 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: pass 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--- BODY ---\n") for block in doc.element.body: full_text.append(get_text_from_xml_element(block)) return re.sub(r'[\n\r\t]+', ' ', "\n".join(full_text)).strip() except Exception as e: logging.error(f"Erro extração DOCX: {e}") return "" # ============================== # LÓGICA INTELIGENTE DE FILTRAGEM # ============================== def is_technical_term(word: str) -> bool: """ Analisa a palavra e retorna True se for um termo técnico, código, símbolo ou inglês válido, que deva ser ignorado pelo corretor PT. """ w_lower = word.lower() # 1. Verifica na Whitelist (O(1) lookup) if w_lower in WHITELIST_TECNICA: return True # 2. Filtros de Estrutura (Código e Variáveis) # - Contém números (ex: H2O, 4G, IPv6) if any(char.isdigit() for char in word): return True # - Mistura de maiúsculas/minúsculas no meio (ex: camelCase, eBay, iPhone) # (Ignora se for apenas a primeira maiúscula -> Título) if not word.islower() and not word.isupper() and not word.istitle(): return True # - Palavras muito curtas (variáveis x, y, pi, id, db) if len(word) < 3: return True # 3. Heurística de Inglês (Sufixos Comuns) english_suffixes = ('ware', 'ing', 'tion', 'sion', 'ment', 'able', 'ible', 'base', 'file', 'size', 'type') if w_lower.endswith(english_suffixes): return True return False def spelling_issues_pt(tool, text: str) -> List[Dict]: if tool is None: return [] try: # Pré-carrega a whitelist no tool para eficiência if hasattr(tool, 'disabled_words'): tool.disabled_words.update(WHITELIST_TECNICA) matches = tool.check(text or "") issues = [] ALLOWED_RULES = {'MORFOLOGIK_RULE_PT_PT'} for m in matches: if m.ruleId not in ALLOWED_RULES: continue error_text = text[m.offset:m.offset+m.errorLength] if is_technical_term(error_text): continue ctx = (text[max(0, m.offset - 20):m.offset] + f"[[{error_text}]]" + text[m.offset+m.errorLength:m.offset+m.errorLength+20]) issues.append({ "message": "Possível erro ortográfico (PT).", "context": ctx, "rule": m.ruleId }) return issues except Exception: return [] # ============================== # IA e Regex # ============================== 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(".,;") return None def evaluate_proof_content(text: str, fallback: str) -> Tuple[str, bool, str, str]: if not OPENAI_API_KEY: return fallback, False, "Sem IA", "" disciplina = try_extract_discipline_regex(text) client = OpenAI(api_key=OPENAI_API_KEY, timeout=30.0) instrucao = f"Disciplina: '{disciplina}'." if disciplina else "Deduza a disciplina." prompt = f"""Aja como Coordenador. Analise a prova. {instrucao} Responda JSON: {{"disciplina_identificada": "Nome", "permissivel": true|false, "justificativa": "...", "sugestoes": "..."}} 'permissivel': false APENAS se a matéria estiver totalmente 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 "{}") return (disciplina or data.get("disciplina_identificada", fallback), data.get("permissivel", False), data.get("justificativa", ""), data.get("sugestoes", "")) except: return fallback, False, "Erro IA", "" # ==================================================================== # Assinatura e PDF # ==================================================================== 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(): 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: return None def convert_docx_to_pdf_linux(docx_bytes: bytes) -> bytes: soffice = shutil.which("libreoffice") or shutil.which("soffice") if not soffice: raise FileNotFoundError("LibreOffice missing") with tempfile.TemporaryDirectory() as temp_dir: path = Path(temp_dir) / "input.docx" with open(path, "wb") as f: f.write(docx_bytes) subprocess.run([soffice, "--headless", "--convert-to", "pdf", "--outdir", temp_dir, str(path)], capture_output=True, check=True, timeout=45) with open(path.with_suffix(".pdf"), "rb") as f: return f.read() def send_email(to, sub, body, atts=None, cc=None, assinatura="", sign_image_path="", sign_image_width_cm=5.0): if not all([SMTP_HOST, SMTP_USER, SMTP_PASS]): return msg = EmailMessage() msg["From"], msg["To"], msg["Subject"] = SMTP_USER, to, sub if cc: msg["Cc"] = cc html_sig = f"

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

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

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

{html_sig}", 'html', 'utf-8') if atts: for fname, fbytes in atts: try: mt, st = (mimetypes.guess_type(fname)[0] or 'application/octet-stream').split('/') if fname.endswith('.txt'): 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=mt, subtype=st, filename=fname) except: 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) 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 = imap.fetch(num, '(BODY.PEEK[])') if res != 'OK': continue msg = email.message_from_bytes(raw[0][1]) subject = clean_header(msg.get("Subject", "")) if ASSUNTO_PROVA_PALAVRA_CHAVE.lower() not in subject.lower(): continue sender = parseaddr(clean_header(msg.get("From", "")))[1] logging.info(f"Processando: {subject} de {sender}") 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, fbytes in docx_parts: text = extract_full_docx_text(fbytes) if not text: continue issues = spelling_issues_pt(LANG_TOOL, text) if LANG_TOOL is None: ort_txt = "Estado: Desativado" elif not issues: ort_txt = "Estado: Aprovado (Nenhum erro ortográfico detetado)" else: ort_txt = f"Estado: {len(issues)} erros prováveis.\n" + "\n".join([f"{i+1}. {e['context']}" for i, e in enumerate(issues)]) disc, ok_ia, just, sug = evaluate_proof_content(text, DISCIPLINA_FALLBACK) aprovado = ok_ia and (len(issues) <= MAX_ISSUES) report = f"VALIDAÇÃO: {'APROVADO' if aprovado else 'REVISÃO'}\nDisciplina: {disc}\n\n[IA]: {just}\n\n[Ortografia Inteligente]:\n{ort_txt}" att_rep = ("relatorio.txt", codecs.BOM_UTF8 + report.encode('utf-8')) email_args = {"assinatura": ASSINATURA, "sign_image_path": SIGN_IMAGE_PATH} if aprovado: doc_ass = add_signature_to_doc(fbytes, SIGN_IMAGE_PATH, SIGN_IMAGE_WIDTH_CM) if doc_ass: try: pdf = convert_docx_to_pdf_linux(doc_ass); final = (f"ASSINADO_{Path(fname).stem}.pdf", pdf) except: final = (f"ASSINADO_{fname}", doc_ass) send_email(EMAIL_APROVADO_PARA, f"[APROVADO] {disc}", "Segue anexo.", [final, att_rep], EMAIL_CC_GERAL, **email_args) send_email(sender, f"[APROVADO] {disc}", "Aprovado.", [att_rep, (fname, fbytes)], **email_args) else: send_email(sender, f"[ERRO] {disc}", "Erro assinatura.", [att_rep], **email_args) else: send_email(sender, f"[REVISÃO] {disc}", "Necessária revisão ortográfica.", [att_rep, (fname, fbytes)], **email_args) imap.store(num, '+FLAGS', '\\Seen') except Exception as e: logging.error(f"Erro: {e}") imap.close(); imap.logout() def main(): try: process_inbox() except Exception as e: logging.critical(f"Fatal: {e}") if __name__ == "__main__": main()