import os
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
import subprocess
import tempfile
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.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, APIError
except ImportError:
OpenAI = None; APIError = None
try:
import language_tool_python
except ImportError:
language_tool_python = None
try:
from spellchecker import SpellChecker
except ImportError:
SpellChecker = None
# --- LISTA BRANCA ---
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()
# --- NOVA FUNÇÃO: CONVERTER .DOC PARA .DOCX ---
def converter_doc_para_docx(doc_bytes: bytes) -> Optional[bytes]:
"""
Usa o LibreOffice (headless) para converter .doc binário para .docx XML.
Requer 'libreoffice' instalado no sistema (sudo apt install libreoffice).
"""
with tempfile.TemporaryDirectory() as temp_dir:
temp_dir_path = Path(temp_dir)
input_path = temp_dir_path / "temp_input.doc"
# Salva o .doc em disco temporariamente
with open(input_path, "wb") as f:
f.write(doc_bytes)
try:
# Comando para converter usando LibreOffice
# --headless: sem interface gráfica
# --convert-to docx: formato de saída
# --outdir: pasta de saída
cmd = [
"libreoffice", "--headless", "--convert-to", "docx",
str(input_path), "--outdir", str(temp_dir_path)
]
# Tenta executar (silenciando output para não sujar logs)
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:
logging.error("Falha na conversão: ficheiro .docx não gerado.")
return None
except FileNotFoundError:
logging.error("ERRO: LibreOffice não encontrado. Instale com 'sudo apt install libreoffice'.")
return None
except Exception as e:
logging.error(f"Erro ao converter .doc: {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)
# ==============================
# Processamento Principal (Atualizado para .doc e .docx)
# ==============================
def process_inbox():
if not all([IMAP_HOST, IMAP_USER, IMAP_PASS]):
logging.error("Config IMAP incompleta.")
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 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"--- EMAIL: {subject} | De: {sender} ---")
docs_parts = []
# --- DETEÇÃO DE .DOC e .DOCX ---
for part in msg.walk():
raw_filename = part.get_filename()
if raw_filename:
fname = clean_header(raw_filename)
# Verifica extensão .docx OU .doc
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:
logging.info(f">> Anexo: {fname}")
payload = part.get_payload(decode=True)
if payload:
docs_parts.append((fname, payload))
if not docs_parts:
imap.store(num, '+FLAGS', '\\Seen')
continue
for fname, fbytes in docs_parts:
# --- CONVERSÃO AUTOMÁTICA DE .DOC PARA .DOCX ---
if fname.lower().endswith('.doc'):
logging.info(f"Detectado formato antigo (.doc). A converter para .docx com LibreOffice...")
converted_bytes = converter_doc_para_docx(fbytes)
if converted_bytes:
fbytes = converted_bytes
fname = fname + "x" # Renomeia de .doc para .docx
logging.info(f"Conversão sucesso: Agora é {fname}")
else:
logging.warning("Falha na conversão do .doc. Ignorando anexo.")
send_email(sender, f"[ERRO FORMATO] {fname}", "O ficheiro .doc não pôde ser convertido. Por favor envie em .docx.")
continue
# -----------------------------------------------
logging.info(f"A processar texto de {fname}...")
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:
# Se era .doc, o professor recebe de volta um .docx atualizado
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 (Formato atualizado para DOCX).", [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.")
except Exception as e: logging.error(f"Erro 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()