import os
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
from pathlib import Path
from typing import Tuple, List, Dict, Optional
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
# --- MODIFICAÇÃO: Importar Pt ---
from docx.shared import Cm, Pt
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.enum.text import WD_TAB_ALIGNMENT
try:
from openai import OpenAI, APIError
except ImportError:
OpenAI = None
APIError = None
try:
import language_tool_python
except ImportError:
language_tool_python = None
# ==============================
# Constantes e Configurações
# ==============================
PALAVRAS_PERSONALIZADAS = [
'multicomputador', 'multiprocessador', 'Flynn', 'telemóvel', 'hardware',
'software', 'backend', 'frontend', 'framework', 'INSUTEC', 'EISI',
'Arquitetura', 'Computadores', 'SQL', 'script'
]
# A lista abaixo existe mas será IGNORADA na validação principal
LISTA_PALAVRAS_INAPROPRIADAS = [
'merda', 'porra', 'caralho', 'foder', 'puta', 'cona',
'grosseira', 'estúpido', 'idiota'
]
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()
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()
# --- CORREÇÃO DE CAMINHO ABSOLUTO (PARA ASSINATURA) ---
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"
# ==============================
# 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 extract_full_docx_text(bytes_content: bytes) -> str:
full_text = []
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
for table in header.tables:
for row in table.rows:
for cell in row.cells:
full_text.append(cell.text)
for paragraph in header.paragraphs:
full_text.append(paragraph.text)
full_text.append("\n--- CONTEÚDO PRINCIPAL DO DOCUMENTO ---\n")
for paragraph in doc.paragraphs:
full_text.append(paragraph.text)
for table in doc.tables:
for row in table.rows:
for cell in row.cells:
full_text.append(cell.text)
return "\n".join(full_text).strip()
# ====================================================================
# FUNÇÃO MODIFICADA (spelling_issues_pt)
# ====================================================================
def spelling_issues_pt(text: str, custom_words: List[str]) -> List[Dict]:
if language_tool_python is None: return []
try:
tool = language_tool_python.LanguageTool('pt-PT')
# Manter a lista de palavras personalizadas ainda é útil
if hasattr(tool, 'disabled_words'): tool.disabled_words.update(custom_words)
elif hasattr(tool, 'disable_spellchecking_for_words'): tool.disable_spellchecking_for_words(custom_words)
matches = tool.check(text or "")
# Ignorar regras de "palavra desconhecida" (jargão técnico, nomes, etc.)
RULES_TO_IGNORE = {
'MORFOLOGIK_RULE_PT_PT', # Regra principal para "Palavra desconhecida / Erro ortográfico"
'UPPERCASE_SPELLING' # Regra para "Palavra em maiúsculas desconhecida" (ex: BFS)
}
issues = []
for m in matches:
# Se o erro for de um tipo que queremos ignorar, saltamos para o próximo.
if m.ruleId in RULES_TO_IGNORE:
continue
context = (text[max(0, m.offset - 30):m.offset] + f"⟦{text[m.offset:m.offset+m.errorLength]}⟧" + text[m.offset+m.errorLength:m.offset+m.errorLength+30])
issues.append({
"message": m.message,
"context": context,
"rule": m.ruleId,
"category": m.category
})
return issues
except Exception as e:
logging.warning(f"LanguageTool indisponível: {e}."); return []
# ====================================================================
# FIM DA FUNÇÃO MODIFICADA
# ====================================================================
def check_for_profanity(text: str, word_list: List[str]) -> List[str]:
text_lower = text.lower()
found_words = set()
for word in word_list:
try:
pattern = r'\b' + re.escape(word.lower()) + r'\b'
if re.search(pattern, text_lower):
found_words.add(word)
except re.error as e:
logging.warning(f"Erro de regex ao procurar por '{word}': {e}")
return list(found_words)
# ==============================
# Função de Extração Regex
# ==============================
def try_extract_discipline_regex(text: str) -> Optional[str]:
if not text: return None
header_sample = text[:2000]
patterns = [
r"(?i)(?:Disciplina|Cadeira|Unidade Curricular|Curricular Unit)\s*[:\.]\s*([^\n\r]+)",
r"(?i)(?:Assunto)\s*[:\.]\s*([^\n\r]+)"
]
for pattern in patterns:
match = re.search(pattern, header_sample)
if match:
clean_name = match.group(1).strip().rstrip(".,;")
logging.info(f"Disciplina detectada via REGEX: '{clean_name}'")
return clean_name
return None
# ====================================================================
# FUNÇÃO IA (evaluate_proof_content) - PROFANIDADE IGNORADA
# ====================================================================
def evaluate_proof_content(text: str, fallback: str) -> Tuple[str, bool, str, str]:
if not OPENAI_API_KEY:
return fallback, False, "Validação OpenAI desativada (API_KEY ausente ou inválida no ficheiro .env).", ""
disciplina_detectada = try_extract_discipline_regex(text)
client = OpenAI(api_key=OPENAI_API_KEY)
if disciplina_detectada:
instrucao_disciplina = f"A disciplina foi identificada no cabeçalho via REGEX como: '{disciplina_detectada}'. CONSIDERE ISTO COMO VERDADE ABSOLUTA."
tarefa_1_desc = f"1. **Identificar a Disciplina:** O sistema já identificou que a disciplina é '{disciplina_detectada}'. Apenas confirme."
tarefa_2_desc = f"2. **Análise de Relevância (TAREFA CRÍTICA):** Analise se as perguntas são pertinentes para a disciplina '{disciplina_detectada}'. Se encontrar perguntas de uma matéria COMPLETAMENTE DIFERENTE, 'permissivel' deve ser false."
else:
instrucao_disciplina = "A disciplina não está explícita no formato padrão. Tente deduzi-la pelo cabeçalho."
tarefa_1_desc = "1. **Identificar a Disciplina:** Leia o texto e encontre o nome EXATO da disciplina no cabeçalho."
tarefa_2_desc = "2. **Análise de Relevância (TAREFA CRÍTICA):** Analise CADA PERGUNTA. Verifique se as competências testadas são relevantes para a disciplina identificada."
prompt = f"""
Aja como um Coordenador de Curso e especialista em pedagogia universitária em Angola.
Analise o seguinte texto extraído de um documento Word (cabeçalho e corpo principal).
**Contexto Linguístico:** O documento está em Português de Angola.
{instrucao_disciplina}
Siga estas tarefas por ordem de PRIORIDADE:
{tarefa_1_desc}
{tarefa_2_desc}
---
**Critérios Pedagógicos (Afetam 'sugestoes')**
---
3. **Clareza e Nível Cognitivo:** Verifique se as perguntas são claras e usam verbos de ação adequados.
4. **Gestão de Contexto:** Se houver erros de copiar/colar no cabeçalho (como texto duplicado), mas as perguntas forem tecnicamente válidas para a disciplina, aceite como permissível, mas note o erro nas sugestões.
5. **Decisão Final:**
* "permissivel": `true` (Se a relevância da disciplina estiver correta).
* "permissivel": `false` (Se a relevância falhar, ou seja, perguntas de outra matéria).
Responda APENAS em JSON: {{"disciplina_identificada": "O nome EXATO", "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 "{}")
final_disciplina = disciplina_detectada if disciplina_detectada else data.get("disciplina_identificada", fallback)
return (
final_disciplina,
data.get("permissivel", False),
data.get("justificativa", "Análise automática."),
data.get("sugestoes", "Nenhuma sugestão.")
)
except Exception as e:
logging.error(f"Falha na API da OpenAI: {e}");
return fallback, False, "Erro na comunicação com a IA.", ""
# ====================================================================
# Lógica do RODAPÉ (Corrigida com Limpeza de Parágrafos)
# ====================================================================
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():
logging.warning(f"Imagem de assinatura NÃO encontrada em {image_path_str}. A enviar documento original.")
return None
try:
doc_stream = io.BytesIO(doc_bytes)
doc = Document(doc_stream)
# --- INÍCIO DA CORREÇÃO DEFINITIVA: Remover parágrafos em branco no final ---
# Iteramos de trás para a frente no corpo do documento
while doc.paragraphs:
last_paragraph = doc.paragraphs[-1]
# Verificamos se o parágrafo não tem texto E não tem imagens/desenhos
# O ._p.xpath('.//w:drawing') verifica se há elementos de imagem/desenho
if not last_paragraph.text.strip() and not last_paragraph._p.xpath('.//w:drawing'):
# Remove o elemento XML do parágrafo
p_element = last_paragraph._element
p_element.getparent().remove(p_element)
else:
# Encontramos um parágrafo com conteúdo, por isso paramos
break
# --- FIM DA CORREÇÃO DEFINITIVA ---
# --- Bloco do Rodapé (Mantém o fix anterior para o espaçamento) ---
section = doc.sections[0]
footer = section.first_page_footer if section.different_first_page_header_footer else section.footer
ftr_element = footer._element
for child in list(ftr_element):
ftr_element.remove(child)
paragraph = footer.add_paragraph()
paragraph.alignment = WD_ALIGN_PARAGRAPH.RIGHT
# Define o espaçamento do parágrafo do rodapé como zero
fmt = paragraph.paragraph_format
fmt.space_before = Pt(0)
fmt.space_after = Pt(0)
run = paragraph.add_run()
run.add_picture(str(image_path), width=Cm(image_width_cm))
new_doc_stream = io.BytesIO()
doc.save(new_doc_stream)
return new_doc_stream.getvalue()
except Exception as e:
logging.error(f"Falha ao adicionar assinatura: {e}", exc_info=True)
return None
# ====================================================================
# FIM DA FUNÇÃO MODIFICADA
# ====================================================================
# ====================================================================
# Conversão para PDF via LibreOffice (Linux)
# ====================================================================
def convert_docx_to_pdf_linux(docx_bytes: bytes) -> bytes:
soffice_cmd = shutil.which("libreoffice") or shutil.which("soffice")
if not soffice_cmd: raise FileNotFoundError("Executável 'libreoffice' ou 'soffice' não encontrado.")
with tempfile.TemporaryDirectory() as temp_dir:
temp_dir_path = Path(temp_dir)
docx_path = temp_dir_path / "input.docx"
with open(docx_path, "wb") as f: f.write(docx_bytes)
cmd = [soffice_cmd, "--headless", "--convert-to", "pdf", "--outdir", temp_dir, str(docx_path)]
try: subprocess.run(cmd, capture_output=True, text=True, timeout=30, check=True)
except subprocess.CalledProcessError as e: logging.error(f"Falha LibreOffice: {e.stderr}"); raise
pdf_path = temp_dir_path / "input.pdf"
if not pdf_path.exists(): raise FileNotFoundError(f"PDF não criado em {pdf_path}")
with open(pdf_path, "rb") as f: return f.read()
# ====================================================================
# Função de E-mail (Corrigida)
# ====================================================================
def send_email(to_addr: str, subject: str, body: str, attachments: List[Tuple[str, bytes]] = None, cc_addr: Optional[str] = None, assinatura: str = "", sign_image_path: str = "", sign_image_width_cm: float = 5.0):
if not all([SMTP_HOST, SMTP_USER, SMTP_PASS]): raise ConnectionError("Configurações de SMTP ausentes.")
msg = EmailMessage()
msg["From"] = SMTP_USER; msg["To"] = to_addr; msg["Subject"] = subject
if cc_addr: msg["Cc"] = cc_addr
assinatura_html = ""
image_cid_str_with_brackets = None
if assinatura:
assinatura_html = f"
{assinatura.replace(chr(10), '
')}
"
image_path_obj = Path(sign_image_path)
if assinatura_html and image_path_obj.is_file():
try:
image_cid_str_with_brackets = make_msgid(domain="insutec.ao")
image_cid_for_html = image_cid_str_with_brackets.strip('<>')
width_px = int(sign_image_width_cm * 37.795)
assinatura_html += f'
'
except Exception: image_cid_str_with_brackets = None
msg.set_content(f"{body}\n\n{assinatura}", 'plain', 'utf-8')
msg.add_alternative(f"{body.replace(chr(10), '
')}
{assinatura_html}", 'html', 'utf-8')
if image_cid_str_with_brackets and image_path_obj.is_file():
try:
img_data = image_path_obj.read_bytes()
img_part = MIMEImage(img_data)
img_part.add_header('Content-ID', image_cid_str_with_brackets)
img_part.add_header('Content-Disposition', 'inline', filename=image_path_obj.name)
msg.attach(img_part)
except Exception: pass
if attachments:
for fname, fbytes in attachments:
try:
if fname.lower().endswith('.txt'):
txt_part = MIMEText(fbytes.decode('utf-8'), 'plain', 'utf-8')
txt_part.add_header('Content-Disposition', 'attachment', filename=fname)
msg.attach(txt_part)
else:
maintype, subtype = (mimetypes.guess_type(fname)[0] or 'application/octet-stream').split('/')
msg.add_attachment(fbytes, maintype=maintype, subtype=subtype, filename=fname)
except Exception: pass
try:
smtp_class = smtplib.SMTP_SSL if USE_SMTP_SSL else smtplib.SMTP
with smtp_class(SMTP_HOST, SMTP_PORT) as s:
# --- LINHA CORRIGIDA ---
# O nome da variável é USE_SMTP_SSL, não USE_SSL
if not USE_SMTP_SSL: s.starttls()
# -----------------------
s.login(SMTP_USER, SMTP_PASS)
s.send_message(msg)
except Exception as e: logging.error(f"Erro envio e-mail: {e}"); raise
# ====================================================================
# FIM DA FUNÇÃO MODIFICADA
# ====================================================================
# ====================================================================
# Função principal
# ====================================================================
def process_inbox():
if not all([IMAP_HOST, IMAP_USER, IMAP_PASS, EMAIL_APROVADO_PARA, EMAIL_CC_GERAL]):
logging.error("Configurações de IMAP/E-mail ausentes."); return
imap = None
try:
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
email_ids = data[0].split()
logging.info(f"Encontrados {len(email_ids)} e-mails não lidos.")
for num in email_ids:
try:
res, raw_full = imap.fetch(num, '(BODY.PEEK[])')
if res != 'OK': continue
msg = email.message_from_bytes(raw_full[0][1])
subject = clean_header(msg.get("Subject", ""))
if ASSUNTO_PROVA_PALAVRA_CHAVE.lower() not in subject.lower(): continue
logging.info(f"Processando enunciado: '{subject}'")
from_header = clean_header(msg.get("From", ""))
sender_name, sender_email = parseaddr(from_header)
if not sender_email: sender_email = from_header
docx_parts = [(clean_header(p.get_filename()), p.get_payload(decode=True))
for p in msg.walk() if p.get_filename() and p.get_filename().lower().endswith(".docx")]
if not docx_parts:
imap.store(num, '+FLAGS', '\\Seen'); continue
for fname, file_bytes in docx_parts:
text = extract_full_docx_text(file_bytes)
if not text: continue
# 1. Verificação de Profanidade (DESATIVADA/IGNORADA)
found_profanities = []
is_permissible_profanidade = True
# 2. Verificação Ortográfica (FILTRADA)
issues = spelling_issues_pt(text, PALAVRAS_PERSONALIZADAS)
is_permissible_ortografia = len(issues) <= MAX_ISSUES
# 3. Verificação de Conteúdo (IA + Regex)
disciplina, is_permissible_ia, justification, suggestions = evaluate_proof_content(text, DISCIPLINA_FALLBACK)
final_is_permissible = is_permissible_ia and is_permissible_ortografia and is_permissible_profanidade
# --- MODIFICAÇÃO: Criar detalhes dos erros ---
detalhes_ortografia_str = ""
if issues:
detalhes_ortografia_str = "\n\n--- Detalhes dos Erros Ortográficos/Gramaticais ---\n"
for i, erro in enumerate(issues, 1):
msg = erro.get('message', 'Erro desconhecido')
ctx = erro.get('context', 'Sem contexto')
detalhes_ortografia_str += f"{i}. {msg}\n Contexto: ...{ctx.strip()}...\n"
# --- FIM DA MODIFICAÇÃO ---
# --- MODIFICAÇÃO: Atualizar 'report_content' ---
report_content = (
f"RELATÓRIO DE VALIDAÇÃO AUTOMÁTICA\n"
f"====================================\n"
f"- Remetente: {from_header}\n"
f"- Ficheiro: {fname}\n"
f"- Disciplina: {disciplina}\n"
f"---\n"
f"STATUS: {'ACEITÁVEL' if final_is_permissible else 'REQUER REVISÃO'}\n"
f"- Profanidade: Nenhuma detetada (Verificação Desativada)\n"
f"- Ortografia: {len(issues)} erros relevantes (Max: {MAX_ISSUES})\n"
f"- IA/Conteúdo: {'Aprovado' if is_permissible_ia else 'Rejeitado'}\n"
f"- Justificativa IA: {justification}\n"
f"- Sugestões: {suggestions}\n"
f"{detalhes_ortografia_str}"
)
report_attachment = ("relatorio_validacao.txt", report_content.encode('utf-8'))
original_doc_attachment = (fname, file_bytes)
common_email_args = {"assinatura": ASSINATURA, "sign_image_path": SIGN_IMAGE_PATH, "sign_image_width_cm": SIGN_IMAGE_WIDTH_CM}
if final_is_permissible:
modified_doc_bytes = add_signature_to_doc(file_bytes, SIGN_IMAGE_PATH, SIGN_IMAGE_WIDTH_CM)
if not modified_doc_bytes:
send_email(sender_email, f"[ERRO SISTEMA] Enunciado: {disciplina}", "Aprovado, mas falha técnica ao inserir assinatura.", [report_attachment], EMAIL_CC_GERAL, **common_email_args)
continue
try:
pdf_bytes = convert_docx_to_pdf_linux(modified_doc_bytes)
doc_final = (f"ASSINADO_{Path(fname).with_suffix('.pdf').name}", pdf_bytes)
except Exception:
doc_final = (f"ASSINADO_{fname}", modified_doc_bytes)
send_email(EMAIL_APROVADO_PARA, f"[APROVADO] Enunciado: {disciplina}", f"Enunciado de {sender_name} aprovado e assinado.", [doc_final], EMAIL_CC_GERAL, **common_email_args)
send_email(sender_email, f"[APROVADO] O seu enunciado: {disciplina}", "O seu enunciado foi aprovado e encaminhado.", [report_attachment, original_doc_attachment], EMAIL_CC_GERAL, **common_email_args)
# --- MODIFICAÇÃO DE LÓGICA: Sempre enviar relatório de erro ---
else:
send_email(sender_email, f"[REVISÃO NECESSÁRIA] Enunciado: {disciplina}", "O enunciado requer revisão. Ver relatório anexo.", [report_attachment, original_doc_attachment], EMAIL_CC_GERAL, **common_email_args)
imap.store(num, '+FLAGS', '\\Seen')
except Exception as e:
logging.error(f"Erro email ID {num}: {e}")
try: imap.store(num, '+FLAGS', '\\Seen')
except: pass
finally:
if imap:
try: imap.close(); imap.logout()
except: pass
def main():
try: process_inbox()
except Exception as e: logging.critical(f"Erro fatal: {e}", exc_info=True); sys.exit(1)
if __name__ == "__main__":
logging.info("Serviço iniciado.")
main()
logging.info("Serviço finalizado.")