import os
# Define para não procurar GPU (evita erros em alguns servidores sem 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 (manipulação avançada do Word)
from docx.oxml import parse_xml
# Bloco OpenAI
try:
from openai import OpenAI, APIError
except ImportError:
OpenAI = None
APIError = None
# Bloco LanguageTool (Verificação Gramatical)
try:
import language_tool_python
except ImportError:
language_tool_python = None
# Bloco SpellChecker (Filtro auxiliar de Inglês)
try:
from spellchecker import SpellChecker
except ImportError:
SpellChecker = 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 E SIGLAS) ---
# Palavras que devem ser sempre aceites, independentemente do idioma
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 (PT)
# ===================================================================
LANG_TOOL_PT = None
try:
logging.info("A iniciar LanguageTool (PT)...")
# Carrega dicionário Português
LANG_TOOL_PT = language_tool_python.LanguageTool('pt-PT')
logging.info("LanguageTool PT carregado.")
except Exception as e:
logging.error(f"Falha LanguageTool: {e}")
# Inicializa verificador de Inglês (pyspellchecker)
# Usado como SEGUNDA OPINIÃO: Se falhar no PT, checa se existe no EN.
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: 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:
"""Verifica heurísticas e lista branca fixa."""
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]:
"""
Verifica ortografia assumindo que o texto base é PT.
Se allow_english=True, palavras rejeitadas em PT são verificadas no dicionário de Inglês.
Se existirem em Inglês, o erro é ignorado (Texto Híbrido).
"""
if tool_pt is None: return []
try:
# Adiciona whitelist técnica ao PT para performance
if hasattr(tool_pt, 'disabled_words'):
tool_pt.disabled_words.update(WHITELIST_TECNICA)
matches = tool_pt.check(text or "")
issues = []
# Regra principal de "Palavra desconhecida" em PT
ALLOWED_RULES = {'MORFOLOGIK_RULE_PT_PT'}
for m in matches:
# Se não for erro de dicionário (ex: erro gramatical complexo), ignoramos para não ser chato
if m.ruleId not in ALLOWED_RULES:
continue
error_text = text[m.offset:m.offset+m.errorLength]
clean_error_text = error_text.strip('.,;:?!()[]{}"\'')
# 1. Filtro Técnico Rápido (Geral)
if is_technical_term(clean_error_text): continue
# 2. Filtro de INGLÊS (Interseção)
# Se a prova permite inglês, verificamos se a "palavra errada" existe em inglês.
if allow_english and ENGLISH_CHECKER_FILTER:
# Se a palavra existe em inglês, NÃO é um erro.
if (clean_error_text in ENGLISH_CHECKER_FILTER) or (clean_error_text.lower() in ENGLISH_CHECKER_FILTER):
continue
# Se chegou aqui: Não é PT, Não é Técnico, Não é Inglês => É Erro.
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 (Nem PT nem EN).",
"context": ctx,
"rule": m.ruleId
})
return issues
except Exception as e:
logging.error(f"Erro na verificação ortográfica: {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": "..."}}
'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
# ====================================================================
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"""
right
540000
"""
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)
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
# 1. Tenta identificar disciplina
disc_preliminar = try_extract_discipline_regex(text)
# Se for disciplina de Inglês, ativamos o "allow_english"
is_english_subject = False
if disc_preliminar and any(keyword in disc_preliminar.lower() for keyword in ['inglês', 'ingles', 'english', 'lingua inglesa']):
is_english_subject = True
logging.info(f"Disciplina '{disc_preliminar}': Modo Híbrido (PT+EN) ativado.")
else:
# NOTA: Mantemos True também para disciplinas técnicas (Informática) que usam muitos termos EN
# Se quiser ser rigoroso nas outras disciplinas, mude para False aqui.
is_english_subject = True # Deixamos True por padrão para tolerância geral
# 2. Correção Ortográfica Híbrida
# Usa PT como base. Se falhar, checa EN. Aceita se existir em QUALQUER um.
issues = check_spelling_mixed(LANG_TOOL_PT, text, allow_english=is_english_subject)
if LANG_TOOL_PT is None: ort_txt = "Estado: Desativado"
elif not issues: ort_txt = f"Estado: Aprovado (Nenhum erro 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)])
# 3. Avaliação IA
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}
# NOTA: EMAIL_CC_GERAL adicionado em TODOS os envios para o Coordenador saber de tudo
if aprovado:
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)
# Email Final Aprovado (CC para Coordenador)
send_email(EMAIL_APROVADO_PARA, f"[APROVADO] {disc}", "Segue anexo assinado.", [final_anexo, att_rep], cc=EMAIL_CC_GERAL, **email_args)
# Confirmação para Professor (CC para Coordenador)
send_email(sender, f"[APROVADO] {disc}", "Aprovado. Segue cópia do relatório.", [att_rep, (fname, fbytes)], cc=EMAIL_CC_GERAL, **email_args)
else:
send_email(sender, f"[ERRO] {disc}", "Erro técnico na assinatura.", [att_rep], cc=EMAIL_CC_GERAL, **email_args)
else:
# REPROVADO: Envia para professor (CC para Coordenador)
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')
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()