"""Fontes estruturadas: Classroom (ITA), Notion (Projetos) e Google Agenda (pessoal).""" import os from datetime import datetime, timedelta, timezone from googleapiclient.discovery import build from .banco import agora, gravar_item from .config import FUSO, agenda_agente_id, carregar_env from .google_auth import credenciais ENTREGUE = {"TURNED_IN", "RETURNED"} # ---------- Classroom ---------- def _prazo_classroom(w): d, h = w.get("dueDate"), w.get("dueTime") or {} if not d: return None utc = datetime(d["year"], d["month"], d["day"], h.get("hours", 23), h.get("minutes", 59), tzinfo=timezone.utc) return utc.astimezone(FUSO).strftime("%Y-%m-%dT%H:%M") def sincronizar_classroom(con, log=print): """Cada atividade com prazo vira uma tarefa; entregue -> concluida. Sem LLM: e dado estruturado.""" cr = build("classroom", "v1", credentials=credenciais("ita"), cache_discovery=False) limite = (datetime.now(FUSO) - timedelta(days=14)).strftime("%Y-%m-%d") n = 0 for curso in cr.courses().list(courseStates=["ACTIVE"], pageSize=100).execute().get("courses", []): cid = curso["id"] trabalhos = cr.courses().courseWork().list(courseId=cid, pageSize=100).execute().get("courseWork", []) entregas = {s["courseWorkId"]: s["state"] for s in cr.courses().courseWork().studentSubmissions().list( courseId=cid, courseWorkId="-", pageSize=300).execute().get("studentSubmissions", [])} for w in trabalhos: prazo = _prazo_classroom(w) if not prazo or prazo[:10] < limite: continue status = "concluido" if entregas.get(w["id"]) in ENTREGUE else "aberto" atual = con.execute("SELECT status FROM itens WHERE origem = 'classroom' AND origem_ref = ?", (f"{cid}/{w['id']}",)).fetchone() if status == "aberto" and atual and atual["status"] in ("concluido", "descartado"): status = atual["status"] # o usuario fechou pelo grupo (ex.: prova presencial, sem entrega) gravar_item(con, { "tipo": "tarefa", "titulo": f"{w['title']} ({curso['name']})", "descricao": (w.get("description") or "")[:800], "prazo": prazo, "status": status, "confianca": 1.0, "projeto": curso["name"], "origem": "classroom", "origem_ref": f"{cid}/{w['id']}", "evidencia": w.get("alternateLink"), }) n += 1 log(f"classroom: {n} atividades com prazo") return n # ---------- Notion ---------- def _texto_blocos(notion, bloco_id, nivel=0, limite=4000): partes = [] for b in notion.blocks.children.list(block_id=bloco_id, page_size=100)["results"]: dados = b.get(b["type"], {}) texto = "".join(t.get("plain_text", "") for t in dados.get("rich_text", [])) if b["type"] == "to_do": texto = ("[x] " if dados.get("checked") else "[ ] ") + texto elif b["type"].startswith("heading"): texto = f"## {texto}" elif b["type"] == "child_database": texto = f"[banco: {dados.get('title')}]" if texto: partes.append(" " * nivel + texto) if b.get("has_children") and nivel < 2 and b["type"] not in ("child_page", "child_database"): partes.append(_texto_blocos(notion, b["id"], nivel + 1, limite)) if sum(len(p) for p in partes) > limite: break return "\n".join(p for p in partes if p)[:limite] def _titulo(o): for p in o.get("properties", {}).values(): if p.get("type") == "title": return "".join(t["plain_text"] for t in p["title"]) return "".join(t["plain_text"] for t in o.get("title", [])) or "?" def _props(o): """Propriedades simples (status, datas, selects) para o resumo semanal.""" out = [] for nome, p in o.get("properties", {}).items(): t, v = p.get("type"), p.get(p.get("type")) if t in ("select", "status") and v: out.append(f"{nome}: {v['name']}") elif t == "multi_select" and v: out.append(f"{nome}: {', '.join(x['name'] for x in v)}") elif t == "date" and v: out.append(f"{nome}: {v['start']}" + (f" a {v['end']}" if v.get("end") else "")) elif t == "checkbox": out.append(f"{nome}: {'sim' if v else 'nao'}") return "; ".join(out) def sincronizar_notion(con, log=print): """Guarda um retrato de cada pagina visivel (projetos e sub-bancos como o Cronograma).""" from notion_client import Client carregar_env() notion = Client(auth=os.environ["NOTION_TOKEN"]) n = 0 for o in notion.search(page_size=100)["results"]: if o["object"] != "page": continue anterior = con.execute("SELECT editado_em FROM notion WHERE pagina_id = ?", (o["id"],)).fetchone() if anterior and anterior["editado_em"] == o["last_edited_time"]: continue conteudo = "\n".join(x for x in (_props(o), _texto_blocos(notion, o["id"])) if x) con.execute("INSERT OR REPLACE INTO notion VALUES (?, ?, ?, ?, ?)", (o["id"], _titulo(o), conteudo, o["last_edited_time"], agora())) n += 1 con.commit() log(f"notion: {n} paginas atualizadas") return n # ---------- Google Agenda ---------- def agenda(): return build("calendar", "v3", credentials=credenciais("pessoal"), cache_discovery=False) def eventos(inicio: datetime, fim: datetime, incluir_agente=False): """Eventos de todas as agendas visiveis da conta pessoal (exceto a Agente, salvo pedido).""" cal = agenda() agente_id = agenda_agente_id() saida = [] for c in cal.calendarList().list().execute()["items"]: if c.get("selected") is False or (c["id"] == agente_id and not incluir_agente): continue r = cal.events().list(calendarId=c["id"], timeMin=inicio.isoformat(), timeMax=fim.isoformat(), singleEvents=True, orderBy="startTime", maxResults=250).execute() for e in r.get("items", []): if e.get("status") == "cancelled": continue saida.append({"agenda": c.get("summary"), "titulo": e.get("summary", "(sem titulo)"), "inicio": e["start"].get("dateTime") or e["start"].get("date"), "fim": e["end"].get("dateTime") or e["end"].get("date"), "da_agente": c["id"] == agente_id}) unicos = {(e["titulo"], e["inicio"]): e for e in saida} # o mesmo evento pode estar em duas agendas return sorted(unicos.values(), key=lambda e: e["inicio"])