import os # Desativa GPU para evitar erros de compatibilidade com bibliotecas de IA os.environ["CUDA_VISIBLE_DEVICES"] = "" import re import sys import json import email import imaplib import smtplib import logging import mimetypes import io import codecs import subprocess import tempfile from pathlib import Path from typing import Tuple, List, Dict, Optional from email.header import decode_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.oxml import parse_xml # ============================== # Configurações e Logs # ============================== 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() # --- CARREGAMENTO DE IA E FERRAMENTAS --- try: from openai import OpenAI except ImportError: OpenAI = None try: import language_tool_python except ImportError: language_tool_python = None try: from spellchecker import SpellChecker except ImportError: SpellChecker = None # --- LISTA BRANCA (TERMOS TÉCNICOS) --- WHITELIST_TECNICA = { 'smpp', 'osi', 'comumente', 'resolution', 'protocol', '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', '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', 'insutec', 'eisi', 'ert', 'isp', 'rouget', 'ruano', 'epifania', 'rodrigues', 'eleuterio', 'moma', 'angola', 'luanda', 'kz', 'ao', 'docente', 'discente', '1ª', '2ª', '3ª', '4ª', '1º', '2º', '3º', '4º' } # Configurações do .env IMAP_HOST = os.getenv("IMAP_HOST", "").strip().replace('"', '') IMAP_USER = os.getenv("IMAP_USER", "").strip().replace('"', '') IMAP_PASS = os.getenv("IMAP_PASS", "").strip().replace('"', '') IMAP_LABEL = os.getenv("IMAP_LABEL", "INBOX").strip() SMTP_HOST = os.getenv("SMTP_HOST", "").strip().replace('"', '') SMTP_PORT = int(os.getenv("SMTP_PORT", "465")) SMTP_USER = os.getenv("SMTP_USER", "").strip().replace('"', '') SMTP_PASS = os.getenv("SMTP_PASS", "").strip().replace('"', '') 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", "Enunciado").strip().replace('"', '') 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")) # ============================== # Inicialização Verificadores # ============================== LANG_TOOL_PT = None try: logging.info("A iniciar LanguageTool (PT)...") LANG_TOOL_PT = language_tool_python.LanguageTool('pt-PT') logging.info("LanguageTool PT carregado.") except Exception as e: logging.error(f"Falha LanguageTool: {e}") ENGLISH_CHECKER_FILTER = None if SpellChecker: try: ENGLISH_CHECKER_FILTER = SpellChecker(language='en') except Exception: pass # ============================== # Funções Utilitárias # ============================== def clean_header(raw_header: Optional[str]) -> str: if not raw_header: return "" try: decoded_fragments = decode_header(raw_header) header_str = "" for bytes_content, encoding in decoded_fragments: if isinstance(bytes_content, bytes): enc = encoding if encoding else "utf-8" try: header_str += bytes_content.decode(enc, errors="ignore") except LookupError: header_str += bytes_content.decode("cp1252", errors="ignore") else: header_str += str(bytes_content) return header_str.replace('\n', '').replace('\r', '').strip() except Exception as e: return str(raw_header) def verificar_assunto(assunto_email: str, palavra_chave: str) -> bool: if not assunto_email: return False return palavra_chave.lower() in assunto_email.lower() def converter_doc_para_docx(doc_bytes: bytes) -> Optional[bytes]: """Converte .doc binário para .docx XML usando LibreOffice.""" with tempfile.TemporaryDirectory() as temp_dir: temp_dir_path = Path(temp_dir) input_path = temp_dir_path / "temp_input.doc" with open(input_path, "wb") as f: f.write(doc_bytes) try: cmd = ["libreoffice", "--headless", "--convert-to", "docx", str(input_path), "--outdir", str(temp_dir_path)] subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) output_path = temp_dir_path / "temp_input.docx" if output_path.exists(): with open(output_path, "rb") as f: return f.read() else: return None except Exception as e: logging.error(f"Erro conversão .doc (LibreOffice instalado?): {e}") return None 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: 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 "" def is_technical_term(word: str) -> bool: w_lower = word.lower() if w_lower in WHITELIST_TECNICA: return True if any(char.isdigit() for char in word): return True if not word.islower() and not word.isupper() and not word.istitle(): return True if len(word) < 3: return True return False def check_spelling_mixed(tool_pt, text: str, allow_english: bool = True) -> List[Dict]: if tool_pt is None: return [] try: if hasattr(tool_pt, 'disabled_words'): tool_pt.disabled_words.update(WHITELIST_TECNICA) matches = tool_pt.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] clean_error_text = error_text.strip('.,;:?!()[]{}"\'') if is_technical_term(clean_error_text): continue if allow_english and ENGLISH_CHECKER_FILTER: if (clean_error_text in ENGLISH_CHECKER_FILTER) or (clean_error_text.lower() in ENGLISH_CHECKER_FILTER): continue 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": "Possível erro ortográfico.", "context": ctx, "rule": m.ruleId}) return issues except Exception as e: 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(".,;") 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": "..."}}""" 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", "" 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_in_header(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"""right540000""" run._r.append(parse_xml(graphic_xml)) section = doc.sections[0] header = section.header if len(header.paragraphs) == 0: p = header.add_paragraph() else: p = header.paragraphs[0] insert_floating_image_in_header(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 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) # ============================== # Lógica Principal # ============================== def process_inbox(): if not all([IMAP_HOST, IMAP_USER, IMAP_PASS]): logging.error("Configurações de IMAP incompletas.") return try: imap = imaplib.IMAP4_SSL(IMAP_HOST) imap.login(IMAP_USER, IMAP_PASS) imap.select(f'"{IMAP_LABEL}"') except Exception as e: logging.error(f"Erro ao conectar IMAP: {e}") return logging.info("A procurar emails NÃO LIDOS (UNSEEN)...") status, data = imap.search(None, '(UNSEEN)') if status != "OK" or not data[0]: logging.info("Nenhum email novo.") imap.close(); imap.logout() 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 not verificar_assunto(subject, ASSUNTO_PROVA_PALAVRA_CHAVE): continue sender = parseaddr(clean_header(msg.get("From", "")))[1] logging.info(f"Processando email de: {sender} | Assunto: {subject}") docs_parts = [] for part in msg.walk(): raw_filename = part.get_filename() if raw_filename: fname = clean_header(raw_filename) # Aceita .doc e .docx, ignora temporários (~$) is_docx = fname.lower().endswith('.docx') is_old_doc = fname.lower().endswith('.doc') is_temp = fname.startswith('~$') if (is_docx or is_old_doc) and not is_temp: payload = part.get_payload(decode=True) if payload: docs_parts.append((fname, payload)) if not docs_parts: # Marca como lido para não travar imap.store(num, '+FLAGS', '\\Seen') continue for fname, fbytes in docs_parts: # Conversão de .doc para .docx if fname.lower().endswith('.doc'): logging.info(f"Convertendo '{fname}' para formato .docx...") converted = converter_doc_para_docx(fbytes) if converted: fbytes = converted fname = fname + "x" else: logging.error(f"Falha ao converter {fname}.") send_email(sender, f"[ERRO] {fname}", "Formato .doc inválido ou corrompido.") continue text = extract_full_docx_text(fbytes) if not text: continue disc_preliminar = try_extract_discipline_regex(text) is_english = True if disc_preliminar and any(k in disc_preliminar.lower() for k in ['inglês', 'english']): is_english = True issues = check_spelling_mixed(LANG_TOOL_PT, text, allow_english=is_english) ort_txt = f"Estado: {len(issues)} erros prováveis.\n" + "\n".join([f"{i+1}. {e['context']}" for i, e in enumerate(issues)]) if issues else "Aprovado (Ortografia)" 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]:\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_assinado = add_signature_to_doc(fbytes, SIGN_IMAGE_PATH, SIGN_IMAGE_WIDTH_CM) if doc_assinado: final_name = f"ASSINADO_{fname}" send_email(EMAIL_APROVADO_PARA, f"[APROVADO] {disc}", "Segue anexo assinado.", [(final_name, doc_assinado), att_rep], cc=EMAIL_CC_GERAL, **email_args) send_email(sender, f"[APROVADO] {disc}", "Aprovado. Segue cópia assinada.", [att_rep, (fname, fbytes)], cc=EMAIL_CC_GERAL, **email_args) else: send_email(sender, f"[ERRO] {disc}", "Erro na assinatura.", [att_rep], cc=EMAIL_CC_GERAL, **email_args) else: send_email(sender, f"[REVISÃO] {disc}", "Necessária revisão.", [att_rep, (fname, fbytes)], cc=EMAIL_CC_GERAL, **email_args) imap.store(num, '+FLAGS', '\\Seen') logging.info("Email processado com sucesso.") except Exception as e: logging.error(f"Erro crítico ID {num}: {e}") imap.close(); imap.logout() def main(): try: process_inbox() except Exception as e: logging.critical(f"Fatal: {e}") if __name__ == "__main__": main()