"""Extracao de compromissos, tarefas e projetos das conversas do WhatsApp com o Gemini.""" import re import unicodedata from datetime import date, datetime, timedelta from . import llm, preferencias, whatsapp from .banco import atualizar_item, gravar_item, itens_abertos, progresso, salvar_progresso from .config import DIAS_PROJETOS, DIAS_TAREFAS, FUSO, MODELO_EXTRACAO, MSGS_CONTEXTO, MSGS_POR_LOTE NULO_STR = {"type": ["string", "null"]} ESQUEMA = { "type": "object", "properties": { "itens": { "type": "array", "items": { "type": "object", "properties": { "acao": {"type": "string", "enum": ["novo", "atualizar", "concluir", "cancelar"]}, "item_id": {"type": ["integer", "null"]}, "tipo": {"type": "string", "enum": ["compromisso", "tarefa", "projeto"]}, "titulo": {"type": "string"}, "descricao": NULO_STR, "inicio": NULO_STR, "fim": NULO_STR, "prazo": NULO_STR, "projeto": NULO_STR, "pessoas": {"type": "array", "items": {"type": "string"}}, "confianca": {"type": "number"}, "evidencia": {"type": "string"}, "linha": {"type": ["integer", "null"]}, }, "required": ["acao", "tipo", "titulo", "confianca", "evidencia"], }, } }, "required": ["itens"], } SISTEMA = """Voce e o assistente pessoal do Joao (aparece como "Eu" nas conversas): estudante do ITA, diretor de \ consultoria estrategica na ITA Junior e pesquisador (IC ITAndroids). Voce le trechos das conversas de WhatsApp \ dele e identifica o que precisa entrar na agenda ou na lista de tarefas DELE. Tipos: - compromisso: algo com data (e de preferencia horario) em que o Joao vai estar: reuniao, aula extra, prova, \ call, evento, entrega presencial. Use "inicio"/"fim" no formato AAAA-MM-DDTHH:MM (ou AAAA-MM-DD se nao houver hora). - tarefa: algo que o Joao precisa FAZER: ele se comprometeu ("vou mandar", "deixa comigo"), alguem pediu a ele e \ ele nao recusou, ou ele foi designado. Use "prazo" (AAAA-MM-DD ou AAAA-MM-DDTHH:MM) quando houver; senao null. - projeto: iniciativa em andamento que dura semanas (ex.: IC, um projeto da ITA Junior, um campeonato), para a \ visao semanal. Descreva o estado atual e o proximo passo conhecido. Regras: - So o que e responsabilidade ou compromisso do Joao. Ignore tarefas de outras pessoas, conversa casual, \ memes, combinados que ja aconteceram e coisas que ele recusou. - Em GRUPOS, so extraia quando: (a) o proprio Joao ("Eu") disse que vai fazer/ir, (b) alguem se dirigiu \ diretamente a ele (pelo nome "Joao"/"Penna", respondendo a uma mensagem dele, ou designando-o), ou (c) e uma \ reuniao/entrega de um grupo de trabalho do qual ele claramente faz parte. Avisos, divulgacoes, convites gerais, \ oportunidades ("inscricoes abertas", "quem puder..."), pendencias de outros membros e eventos que ele nao \ confirmou NAO sao itens dele. - Em conversas diretas, pedidos feitos a ele e combinados entre os dois contam. - Resolva datas relativas ("amanha", "sexta", "semana que vem") a partir da data/hora da mensagem. - Linhas marcadas [CONTEXTO] ja foram analisadas antes: use so como contexto, nao extraia delas de novo. - Mensagens com mais de {dias_tarefas} dias (antes de {corte}): extraia apenas projetos em andamento e \ compromissos/prazos com data a partir de hoje ({hoje}). - Se o item ja existe na lista de itens abertos, use acao "atualizar" (mudou data, detalhe) ou "concluir" \ (ficou claro que foi feito) ou "cancelar" (foi desmarcado), com o item_id. Nao crie duplicatas. - Itens marcados "(Classroom)" ja sao acompanhados pelo Classroom: nao os crie de novo nem tente altera-los. - "confianca": 0.9+ quando explicito e com data; 0.6-0.8 quando provavel; abaixo de 0.5 quando duvidoso. - "evidencia": a frase curta que justifica o item, com o nome de quem disse. - "linha": o numero (n) da mensagem que originou ou atualizou o item. - Titulos curtos e acionaveis, em portugues ("Enviar relatorio de diagnostico para o cliente X"). - Se nao houver nada, devolva {{"itens": []}}.""" def _slug(texto: str) -> str: t = unicodedata.normalize("NFKD", texto.lower()).encode("ascii", "ignore").decode() return re.sub(r"[^a-z0-9]+", "-", t).strip("-")[:60] def _lista_abertos(con): """So os itens que a leitura do WhatsApp pode alterar; os do Classroom vao como referencia, sem id.""" linhas = [] for i in itens_abertos(con): if i["origem"] != "whatsapp": linhas.append(f"(Classroom) {i['titulo']} — prazo {i['prazo']}") continue quando = i["inicio"] or i["prazo"] or "sem data" linhas.append(f"#{i['id']} [{i['tipo']}] {i['titulo']} ({quando})" + (f" — projeto {i['projeto']}" if i["projeto"] else "")) return "\n".join(linhas[-250:]) or "(nenhum)" def _data(valor): if not valor: return None try: return datetime.fromisoformat(valor).date() if "T" in valor else date.fromisoformat(valor[:10]) except ValueError: return None def _aplicar(con, chat, chat_nome, it, msg_ts): hoje = datetime.now(FUSO).date() if it["acao"] in ("concluir", "cancelar", "atualizar") and it.get("item_id"): existe = con.execute("SELECT origem FROM itens WHERE id = ?", (it["item_id"],)).fetchone() if existe and existe["origem"] != "whatsapp": return "ignorado" # Classroom/Notion sao a fonte da verdade dos itens deles if existe: if it["acao"] == "atualizar": campos = {k: it[k] for k in ("titulo", "descricao", "inicio", "fim", "prazo", "projeto") if it.get(k)} atualizar_item(con, it["item_id"], **campos) else: atualizar_item(con, it["item_id"], status="concluido" if it["acao"] == "concluir" else "descartado") return "atualizado" # item novo: descarta o que ja passou quando = _data(it.get("inicio")) if it["tipo"] == "compromisso" else _data(it.get("prazo")) if it["tipo"] == "compromisso" and (quando is None or quando < hoje): return "ignorado" if it["tipo"] == "tarefa" and quando and quando < hoje - timedelta(days=DIAS_TAREFAS): return "ignorado" antiga = datetime.fromtimestamp(msg_ts, FUSO).date() < hoje - timedelta(days=DIAS_TAREFAS) if it["tipo"] == "tarefa" and not quando and antiga: return "ignorado" # tarefa sem prazo de mensagem antiga: provavelmente ja resolvida conf = float(it.get("confianca") or 0) if conf < 0.4: return "ignorado" gravar_item(con, { **{k: it.get(k) for k in ("tipo", "titulo", "descricao", "inicio", "fim", "prazo", "projeto", "pessoas")}, "status": "aberto" if conf >= 0.75 else "a_confirmar", "confianca": conf, "origem": "whatsapp", "origem_ref": f"{chat}:{_slug(it['titulo'])}", "evidencia": f"{chat_nome or 'conversa'}: {it.get('evidencia', '')}"[:500], "msg_ts": msg_ts, }) return "novo" def extrair_chat(con, chat, desde_ts, contagem): ultimo = progresso(con, f"wa:{chat}", desde_ts) linhas = whatsapp.linhas_conversa(con, chat, desde_ts) novas = [l for l in linhas if l["ts"] > ultimo] if not novas: return agora_ = datetime.now(FUSO) sistema = SISTEMA.format(dias_tarefas=DIAS_TAREFAS, hoje=agora_.strftime("%d/%m/%Y (%A)"), corte=(agora_ - timedelta(days=DIAS_TAREFAS)).strftime("%d/%m")) sistema += "\n\n" + preferencias.para_prompt(con) nome = novas[0]["chat_nome"] or "conversa" tipo_chat = "grupo" if novas[0]["em_grupo"] else "conversa direta" anteriores = [l for l in linhas if l["ts"] <= ultimo][-MSGS_CONTEXTO:] for i in range(0, len(novas), MSGS_POR_LOTE): lote = novas[i:i + MSGS_POR_LOTE] contexto = (anteriores + novas[:i])[-MSGS_CONTEXTO:] texto = (f"Agora: {agora_.strftime('%A %d/%m/%Y %H:%M')}\n{tipo_chat.capitalize()}: {nome}\n\n" f"Itens abertos:\n{_lista_abertos(con)}\n\nMensagens:\n" + "\n".join(f"[CONTEXTO] {l['texto']}" for l in contexto) + ("\n" if contexto else "") + "\n".join(f"({n}) {l['texto']}" for n, l in enumerate(lote))) resp = llm.gerar(MODELO_EXTRACAO, texto, sistema=sistema, esquema=ESQUEMA, con=con, etapa="extracao") for it in resp.get("itens", []): n = it.get("linha") msg_ts = lote[n]["ts"] if isinstance(n, int) and 0 <= n < len(lote) else lote[-1]["ts"] r = _aplicar(con, chat, nome, it, msg_ts) contagem[r] = contagem.get(r, 0) + 1 salvar_progresso(con, f"wa:{chat}", lote[-1]["ts"]) ESQUEMA_CONSOLIDAR = { "type": "object", "properties": { "duplicatas": {"type": "array", "items": {"type": "object", "properties": { "manter": {"type": "integer"}, "remover": {"type": "array", "items": {"type": "integer"}}}, "required": ["manter", "remover"]}}, "descartar": {"type": "array", "items": {"type": "object", "properties": { "id": {"type": "integer"}, "motivo": {"type": "string"}}, "required": ["id", "motivo"]}}, }, "required": ["duplicatas", "descartar"], } SISTEMA_CONSOLIDAR = """Voce revisa a lista de itens que um assistente extraiu das conversas de WhatsApp do Joao. 1. duplicatas: itens que sao a MESMA coisa (mesmo compromisso ou mesma tarefa, ainda que vindos de conversas \ diferentes ou com titulos diferentes). Mantenha o mais completo. 2. descartar: itens que NAO sao do Joao: convites e divulgacoes de festas/eventos que ele nao confirmou, \ oportunidades genericas ("inscreva-se", "summer job aberto"), avisos para a turma toda sem acao dele, \ pendencias de outras pessoas. Na duvida, NAO descarte. Use apenas os ids fornecidos.""" def consolidar(con, log=print): itens = [i for i in itens_abertos(con) if i["origem"] == "whatsapp"] if len(itens) < 2: return lista = "\n".join(f"#{i['id']} [{i['tipo']}] {i['titulo']} | {i['inicio'] or i['prazo'] or 'sem data'} | " f"{(i['evidencia'] or '')[:160]}" for i in itens) r = llm.gerar(MODELO_EXTRACAO, lista, sistema=SISTEMA_CONSOLIDAR, esquema=ESQUEMA_CONSOLIDAR, con=con, etapa="consolidar") validos = {i["id"] for i in itens} n_dup = n_desc = 0 for d in r.get("duplicatas", []): if d["manter"] not in validos: continue for x in d["remover"]: if x in validos and x != d["manter"]: atualizar_item(con, x, status="descartado", descricao=f"duplicata de #{d['manter']}") n_dup += 1 for d in r.get("descartar", []): if d["id"] in validos: atualizar_item(con, d["id"], status="descartado", descricao=f"descartado: {d['motivo'][:200]}") n_desc += 1 log(f"consolidacao: {n_dup} duplicatas, {n_desc} descartados") def extrair(con, dias=DIAS_PROJETOS, limite_chats=None, log=print): desde_ts = whatsapp.desde(dias) chats = whatsapp.chats_relevantes(desde_ts)[:limite_chats] log(f"conversas relevantes: {len(chats)}") n = whatsapp.processar_midias(con, chats, desde_ts) log(f"midias lidas: {n}") contagem = {} for k, chat in enumerate(chats, 1): try: extrair_chat(con, chat, desde_ts, contagem) except Exception as e: log(f"falha em {chat}: {type(e).__name__}: {str(e)[:200]}") if k % 20 == 0: log(f"{k}/{len(chats)} conversas | {contagem}") log(f"extracao concluida: {contagem}") if contagem.get("novo"): consolidar(con, log) return contagem