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
import unicodedata # <--- NOVA IMPORTAÇÃO IMPORTANTE
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()
# Nota: A palavra chave do ENV ainda é carregada, mas a lógica regex terá prioridade
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)
# --- NOVA LÓGICA DE DETECÇÃO DE ASSUNTO ---
def verificar_assunto_inteligente(assunto_email: str) -> bool:
"""
Verifica se o assunto contém variações da palavra Enunciado
(plural, singular, maiúsculas, com ou sem acento).
"""
if not assunto_email:
return False
# 1. Normalizar: Remove acentos e converte para minúsculas
# Ex: "Enunciado 1ª Frequência" vira "enunciado 1a frequencia"
texto_normalizado = unicodedata.normalize('NFKD', assunto_email).encode('ASCII', 'ignore').decode('utf-8').lower()
# 2. Regex: Procura por "enunciad" seguido de 'o', 'a' ou 'os/as'
# Isso cobre: enunciado, enunciados, enunciada (erro comum), etc.
padrao = r"enunciad[oa]s?"
if re.search(padrao, texto_normalizado):
return True
return False
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", ""))
# --- MUDANÇA AQUI: Usa a verificação inteligente ---
if not verificar_assunto_inteligente(subject):
# Se quiser logar o que ignorou (opcional):
# logging.info(f"Ignorado (assunto inválido): {subject}")
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()