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 logging import mimetypes import io 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 from email.mime.text import MIMEText from dotenv import load_dotenv from docx import Document from docx.shared import Cm from docx.enum.text import WD_ALIGN_PARAGRAPH # Importações para XML from docx.oxml import parse_xml # 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) --- 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º' } 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: 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 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: 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 [] 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", "" # ==================================================================== # Função de Assinatura (HEADER ANCHORED -> TOP RIGHT PAGE EDGE) # ==================================================================== def add_signature_to_doc(doc_bytes: bytes, image_path_str: str, image_width_cm: float) -> Optional[bytes]: """ Adiciona a assinatura FLUTUANTE ancorada no CABEÇALHO. Posicionada no CANTO SUPERIOR DIREITO em relação à PÁGINA (Marca de Água). """ 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]) # --- XML CRÍTICO PARA POSICIONAMENTO NO TOPO DA PÁGINA --- # 1. relativeFrom="page" (Ignora margens, usa o papel) # 2. posOffset (Vertical) = Cerca de 1.5cm (540000 EMUs) do topo físico # 3. align="right" (RelativeFrom Margin) para alinhar à direita graphic_xml = f""" right 540000 """ run._r.append(parse_xml(graphic_xml)) # Insere no CABEÇALHO section = doc.sections[0] header = section.header # Cria parágrafo se não existir 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) 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 PT grave 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: # 1. Insere assinatura no DOCX (Header, Top Right Watermark) doc_assinado = add_signature_to_doc(fbytes, SIGN_IMAGE_PATH, SIGN_IMAGE_WIDTH_CM) if doc_assinado: final_name = f"ASSINADO_{fname}" final_anexo = (final_name, doc_assinado) send_email(EMAIL_APROVADO_PARA, f"[APROVADO] {disc}", "Segue anexo em formato Word (DOCX) assinado.", [final_anexo, att_rep], EMAIL_CC_GERAL, **email_args) send_email(sender, f"[APROVADO] {disc}", "Aprovado. Segue cópia do relatório.", [att_rep, (fname, fbytes)], **email_args) else: send_email(sender, f"[ERRO] {disc}", "Erro técnico na 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()