Spaces:
Runtime error
Runtime error
Refactor VideoTemplateEditor component: remove AI editing features and related state management, streamline template loading, and update styles for improved layout. Enhance CSS for responsive design and visual consistency across various components, including dropdowns, sync stats, and quiz elements. Remove deprecated premium badge styles and adjust layout for chat components.
66de9c8 | """Reports: a written document (or an interactive one, with visuals) made from a notebook's posts or chats. | |
| Every report is stored as `blocks`, even a plain document (a few prose blocks), so the same page can show, extend and | |
| share either kind. Each block is {"id", "type", ...}: `prose` has Markdown `text`. A report cites nothing: it carries the | |
| list of the posts it was made from (`sources`) and the instructions it was written to (`instructions`), which the page shows | |
| together in a drop-down. (`citations` stays in the content, empty, for the older reports that have them.)""" | |
| import logging | |
| import re | |
| import time | |
| import uuid | |
| from dataclasses import dataclass, field | |
| from sqlalchemy import select | |
| from sqlalchemy.orm import Session | |
| from app.config import settings | |
| from app.corpus import Corpus | |
| from app.infographics.themes import theme_id | |
| from app.llm import run | |
| from app.llm.provider import report_lm, report_section_lm | |
| from app.llm.signatures import ( | |
| PlanInteractiveReport, | |
| WriteReport, | |
| WriteReportSection, | |
| WriteThemedReport, | |
| WriteThemedSection, | |
| ) | |
| from app.models import Artifact, Document, Job, Notebook | |
| from app.pipeline import report_blueprint, report_evidence, report_themes | |
| from app.pipeline.flashcards import make_cards | |
| from app.pipeline.generate import NothingToDo, mind_map_content | |
| from app.pipeline.material import Material, load_material | |
| from app.pipeline.quiz import make_questions | |
| from app.services.artifacts import start_artifact | |
| from app.services.jobs import update_job | |
| from app.services.renderer import PermanentJobError | |
| from app.services.report_templates import default_instructions | |
| EMBED_KINDS = ("mind_map", "flashcards", "quiz", "infographic") # every visual a reader can add with one click | |
| MAX_PLANNED_VISUALS = 4 # suggestions an interactive report's plan can make | |
| MAX_SECTIONS = 8 | |
| SAME_LANGUAGE = "the same language as the sources" # no language picked: write as the material is written | |
| BLOCK_CARDS, BLOCK_QUESTIONS = 8, 5 # a visual inside a report is a taste of the full thing | |
| log = logging.getLogger(__name__) | |
| MARKER = re.compile(r"\s*\[\d+\](?:\[\d+\])*") | |
| QUIZ_HEADING = re.compile(r"\b(quiz|test yourself|check your understanding|practice questions|review questions|self[- ]?check)\b", re.I) | |
| FLASHCARD_HEADING = re.compile(r"\bflash ?cards?\b", re.I) | |
| def visual_kind(heading: str) -> str | None: | |
| """The visual a section heading really is ("Quick quiz" is a quiz), or None for an ordinary section.""" | |
| if QUIZ_HEADING.search(heading): | |
| return "quiz" | |
| if FLASHCARD_HEADING.search(heading): | |
| return "flashcards" | |
| return None | |
| MOVED_BRIEF = {"quiz": "Check understanding of the key ideas in this report.", | |
| "flashcards": "Learn the key terms and facts from this report."} | |
| def merge_suggestions(*groups: list[dict]) -> list[dict]: | |
| """Suggestions from several places, one per section and kind, renumbered.""" | |
| seen, out = set(), [] | |
| for g in groups: | |
| for s in g: | |
| key = (s["after_block_id"], s["kind"]) | |
| if key not in seen: | |
| seen.add(key) | |
| out.append({**s, "id": f"s{len(out) + 1}"}) | |
| return out[:MAX_PLANNED_VISUALS + 2] | |
| FALLBACK_BRIEFS = { | |
| "infographic": ("A one page picture of the main ideas of this report.", "A quick picture of the whole report."), | |
| "flashcards": (MOVED_BRIEF["flashcards"], "Cards to learn the key terms and facts."), | |
| "quiz": (MOVED_BRIEF["quiz"], "A short quiz to check understanding."), | |
| "mind_map": ("How the main ideas of this report connect.", "See how the ideas fit together."), | |
| } | |
| MIN_SUGGESTIONS = 3 # a report always offers at least this many visuals (fewer only when fewer kinds are allowed) | |
| def fill_suggestions(suggestions: list[dict], blocks: list[dict], kinds: list[str]) -> list[dict]: | |
| """An interactive report always offers something to add (a document never does). Whatever the plan or the writer suggested is | |
| kept; then an infographic is added if there is none, and flashcards, a quiz and a mind map in turn until there are | |
| MIN_SUGGESTIONS, each after a different section. Plain code, so it never depends on the model remembering to suggest.""" | |
| prose = [b for b in blocks if b.get("type") == "prose"] | |
| if not prose: | |
| return suggestions | |
| inner = prose[1:-1] or prose # not the introduction or the summary, when there are others | |
| place = {"infographic": inner[-1]["id"], "flashcards": inner[0]["id"], "quiz": inner[len(inner) // 2]["id"], | |
| "mind_map": prose[0]["id"]} | |
| out = list(suggestions) | |
| have = {s["kind"] for s in out} | |
| for kind in ("infographic", "flashcards", "quiz", "mind_map"): | |
| if kind in kinds and kind not in have and (kind == "infographic" or len(out) < MIN_SUGGESTIONS): | |
| brief, why = FALLBACK_BRIEFS[kind] | |
| out.append({"id": "", "after_block_id": place[kind], "kind": kind, "brief": brief, "why": why}) | |
| have.add(kind) | |
| log.info("report: no %s suggested by the plan, offering one after %s", kind, place[kind]) | |
| return out | |
| def split_prose(markdown: str) -> list[dict]: | |
| """Markdown cut into prose blocks at each "## " heading, so a block is one section and can be followed by a visual.""" | |
| parts = re.split(r"(?m)^(?=## )", markdown.strip()) | |
| blocks = [] | |
| for part in parts: | |
| if part.strip(): | |
| blocks.append({"type": "prose", "text": part.strip()}) | |
| return blocks | |
| def number_blocks(blocks: list[dict]) -> list[dict]: | |
| for i, b in enumerate(blocks, start=1): | |
| b["id"] = f"b{i}" | |
| return blocks | |
| def source_rows(material: Material) -> list[dict]: | |
| """The posts a report was made from, for its Sources list: titles and the posts' own web addresses.""" | |
| return source_rows_for(material.docs) | |
| def source_rows_for(docs: list[Document]) -> list[dict]: | |
| return [{"title": d.title, "url": d.url, "document_id": str(d.id)} for d in docs] | |
| def write_document(db: Session, workspace_id: uuid.UUID, job: Job | None, material: Material, instructions: str, | |
| language: str) -> tuple[str, str]: | |
| """(title, Markdown) for a written report. Nothing is cited; a marker the model wrote anyway is removed.""" | |
| out = run.predict(WriteReport, db=db, workspace_id=workspace_id, job=job, lm=report_lm(), title=material.title, | |
| material=material.text, request=instructions, language=language) | |
| markdown = (out.get("markdown") or "").strip() | |
| if not markdown: | |
| raise NothingToDo("The report came back empty. Try again, or describe it differently.") | |
| return (out.get("report_title") or material.title).strip(), MARKER.sub("", markdown) | |
| def write_interactive(db: Session, workspace_id: uuid.UUID, job: Job, material: Material, instructions: str, | |
| language: str) -> tuple[str, list[dict], list[dict]] | None: | |
| """(title, blocks, suggestions) for an interactive report: a plan of sections, each written on its | |
| own (in parallel). The plan also says where a visual would fit; those spots come back as suggestions, and nothing | |
| is built until the reader clicks Add. None when planning fails, so the caller can write a plain document.""" | |
| kinds = [k for k in EMBED_KINDS if k != "mind_map" or not material.is_chat] | |
| update_job(db, job, progress=0.25, message="Planning the report") | |
| try: | |
| plan = run.predict(PlanInteractiveReport, db=db, workspace_id=workspace_id, job=job, lm=report_lm(), | |
| title=material.title, material=material.text, request=instructions, | |
| allowed_kinds=kinds, language=language) | |
| except Exception: | |
| log.warning("report plan failed", exc_info=True) | |
| return None | |
| sections = [x for x in plan.get("sections") or [] if (x.get("heading") or "").strip()][:MAX_SECTIONS] | |
| if len(sections) < 2: | |
| return None | |
| report_title = (plan.get("report_title") or material.title).strip() | |
| # A planned section that is really a quiz or flashcards is not written as text: it becomes a suggestion. | |
| kept_secs: list[dict] = [] | |
| moved: list[tuple[int, str, dict]] = [] | |
| for x in sections: | |
| kind = visual_kind(x["heading"]) | |
| if kind and kind in kinds and kept_secs: | |
| moved.append((len(kept_secs) - 1, kind, x)) | |
| else: | |
| kept_secs.append(x) | |
| sections = kept_secs | |
| update_job(db, job, progress=0.4, message=f"Writing {len(sections)} sections") | |
| written = run.predict_many( | |
| WriteReportSection, | |
| [{"title": material.title, "report_title": report_title, "heading": x["heading"].strip(), | |
| "brief": (x.get("brief") or "").strip(), "request": instructions, "material": material.text, | |
| "language": language} for x in sections], | |
| db=db, workspace_id=workspace_id, lm=report_lm()) | |
| blocks: list[dict] = [] | |
| suggestions: list[dict] = [] | |
| used_kinds: set[str] = set() | |
| anchor: dict[int, str] = {} # planned section number -> the block it became | |
| for n, (sec, out) in enumerate(zip(sections, written, strict=True)): | |
| body = MARKER.sub("", ((out or {}).get("markdown") or "").strip()) | |
| if not body: | |
| continue | |
| block = {"type": "prose", "text": f"## {sec['heading'].strip()}\n\n{body}"} | |
| blocks.append(block) | |
| number_blocks(blocks) | |
| anchor[n] = blocks[-1]["id"] | |
| embed = sec.get("embed") | |
| if embed and len(suggestions) < MAX_PLANNED_VISUALS and embed.get("kind") in kinds \ | |
| and embed["kind"] not in used_kinds and (embed.get("brief") or "").strip(): | |
| used_kinds.add(embed["kind"]) | |
| suggestions.append({"id": f"s{len(suggestions) + 1}", "after_block_id": blocks[-1]["id"], | |
| "kind": embed["kind"], "brief": embed["brief"].strip()[:300], | |
| "why": (embed.get("why") or "").strip()[:300]}) | |
| if not any(b["type"] == "prose" for b in blocks): | |
| return None | |
| for idx, kind, sec in moved: | |
| after = anchor.get(idx) or blocks[0]["id"] | |
| suggestions.append({"id": "", "after_block_id": after, "kind": kind, | |
| "brief": (sec.get("brief") or MOVED_BRIEF[kind]).strip()[:300], | |
| "why": "Offered here for you to add, not written into the text."}) | |
| return report_title, blocks, merge_suggestions(fill_suggestions(suggestions, blocks, kinds)) | |
| # A report built around themes (report_themes.py, report_blueprint.py, report_evidence.py) | |
| EVIDENCE_TOTAL = 100_000 # characters of evidence in all for a report written in one piece (about 25k tokens) | |
| DIGEST_CHARS = 300 # of each written section, for the introduction and the summary that are written after them | |
| class Themed: | |
| title: str | |
| blocks: list[dict] | |
| suggestions: list[dict] | |
| source: dict | |
| sources: list[dict] | |
| plan: dict | |
| failed: list[str] = field(default_factory=list) | |
| class ThemedRun: | |
| """What the steps of a themed report share.""" | |
| db: Session | |
| workspace_id: uuid.UUID | |
| job: Job | |
| docs: list[Document] | |
| corpus: Corpus | |
| bp: report_blueprint.Blueprint | |
| themes: list[report_themes.Theme] | |
| instructions: str | |
| language: str | |
| title: str | |
| def by_id(self) -> dict[str, report_themes.Theme]: | |
| return {t.id: t for t in self.themes} | |
| def names(self, s: report_blueprint.Section) -> list[str]: | |
| by_id = self.by_id | |
| return [by_id[i].name for i in s.theme_ids if i in by_id] | |
| def sections_info(self) -> list[dict]: | |
| return [{"heading": s.heading, "role": s.role, "depth": s.depth, "theme_ids": s.theme_ids, "must_cover": s.must_cover} | |
| for s in self.bp.sections] | |
| def finish(self, title: str, blocks: list[dict], suggestions: list[dict], used_posts: list[str], | |
| failed: list[str]) -> Themed: | |
| """The Sources list is every post the report was made from: the ones read as written and the ones summarised.""" | |
| used = [d for d in self.docs if d.path in set(used_posts)] or self.docs | |
| log.info("report sources: job=%s posts_total=%d posts_used=%d", self.job.id, len(self.docs), len(used)) | |
| plan = {"planner": "themes", "storyline": self.bp.storyline, "posts_total": len(self.docs), "posts_used": len(used), | |
| "themes": [report_themes.theme_summary(t) for t in self.themes], "sections": self.sections_info(), | |
| "connections": self.bp.connections} | |
| return Themed(title=title, blocks=blocks, suggestions=suggestions, plan=plan, failed=failed, | |
| source={"kind": "posts", "document_ids": [str(d.id) for d in self.docs], "count": len(self.docs), | |
| "read": len(used), "total": len(self.docs)}, | |
| sources=source_rows_for(used)) | |
| def themed_docs(db: Session, artifact: Artifact, params: dict) -> list[Document]: | |
| """The posts a themed report would read, or [] when the report is made the usual way: chats, a small selection, or the | |
| legacy planner.""" | |
| if settings.report_planner != "themes" or params.get("chat_ids"): | |
| log.info("report route: legacy artifact=%s reason=%s", artifact.id, | |
| "chats picked" if params.get("chat_ids") else f"REPORT_PLANNER={settings.report_planner}") | |
| return [] | |
| docs = report_themes.select_docs(db, artifact.workspace_id, artifact.notebook_id, params) | |
| if len(docs) <= report_themes.SMALL_SELECTION: | |
| log.info("report route: legacy artifact=%s reason=small selection (%d posts, themes need more than %d)", artifact.id, | |
| len(docs), report_themes.SMALL_SELECTION) | |
| return [] | |
| return docs | |
| def themed_title(db: Session, artifact: Artifact, params: dict, docs: list[Document]) -> str: | |
| """What the report is about, named the way load_material names its source.""" | |
| nb = db.get(Notebook, artifact.notebook_id) if artifact.notebook_id else None | |
| if nb and not params.get("document_ids"): | |
| return nb.title | |
| return docs[0].title + (f" + {len(docs) - 1} more" if len(docs) > 1 else "") | |
| def _digest(markdown: str) -> str: | |
| """The start of a written section, for the sections written after it to know what it says.""" | |
| text = MARKER.sub("", re.sub(r"(?m)^#+\s*", "", markdown)) | |
| return re.sub(r"\s+", " ", text).strip()[:DIGEST_CHARS] | |
| def _section_header(n: int, s: report_blueprint.Section, names: list[str]) -> str: | |
| return (f"SECTION {n}: {s.heading} (role: {s.role}, depth: {s.depth})\nSays: {s.brief}\n" | |
| f"Must cover: {'; '.join(s.must_cover) or '(see what it says)'}\nThemes: {', '.join(names) or '(the whole report)'}") | |
| def _section_input(ctx: ThemedRun, s: report_blueprint.Section, evidence: list[str]) -> dict: | |
| """Everything WriteThemedSection is given. What every section shares comes first in the signature.""" | |
| must = list(s.must_cover) | |
| if s.role == "summary" and ctx.names(s): | |
| must.append(f"Restate the main points about: {', '.join(ctx.names(s))}") | |
| return {"storyline": ctx.bp.storyline, "outline": report_blueprint.outline(ctx.bp), | |
| "connections": report_blueprint.connection_lines(ctx.bp, ctx.themes), "request": ctx.instructions, | |
| "language": ctx.language, "heading": s.heading, "role": s.role, "depth": s.depth, "brief": s.brief, | |
| "must_cover": must, "covered_elsewhere": s.covered_elsewhere, | |
| "bridges": report_blueprint.connection_lines(ctx.bp, ctx.themes, s.heading), "evidence": evidence} | |
| def _write_sections(ctx: ThemedRun, inputs: list[dict]) -> list[dict | None]: | |
| """WriteThemedSection for each input, together; one that failed or came back empty gets one more try (another temperature).""" | |
| kw = {"db": ctx.db, "workspace_id": ctx.workspace_id} | |
| started = time.monotonic() | |
| outs = run.predict_many(WriteThemedSection, inputs, lm=report_section_lm(), **kw) | |
| again = [k for k, o in enumerate(outs) if not ((o or {}).get("markdown") or "").strip()] | |
| log.info("report sections written: asked=%d ok=%d failed=%d seconds=%.1f", len(inputs), len(inputs) - len(again), len(again), | |
| time.monotonic() - started) | |
| if again: | |
| log.warning("report sections failed, trying again: %s", [inputs[k]["heading"] for k in again]) | |
| redo = run.predict_many(WriteThemedSection, [inputs[k] for k in again], lm=report_section_lm(retry=True), **kw) | |
| for k, o in zip(again, redo, strict=True): | |
| outs[k] = o | |
| gone = [inputs[k]["heading"] for k in again if not ((outs[k] or {}).get("markdown") or "").strip()] | |
| log.info("report sections retried: asked=%d recovered=%d still_failed=%s", len(again), len(again) - len(gone), gone) | |
| return outs | |
| def _themed_document(ctx: ThemedRun) -> Themed | None: | |
| """The whole report in one piece: the plan and every section's evidence go to one writer, so one voice writes it all.""" | |
| budgets = report_evidence.scale_budgets(ctx.bp.sections, EVIDENCE_TOTAL) | |
| parts: list[str] = [] | |
| used: list[str] = [] | |
| for n, s in enumerate(ctx.bp.sections, start=1): | |
| head = _section_header(n, s, ctx.names(s)) | |
| if s.role in ("body", "synthesis"): | |
| ev = report_evidence.build(ctx.corpus, s, ctx.by_id, budgets.get(s.heading)) | |
| used += ev.posts | |
| if ev.blocks: | |
| head += "\nEvidence:\n" + "\n\n".join(ev.blocks) | |
| parts.append(head) | |
| report_themes.progress(ctx.db, ctx.job, 0.5, "Writing the report") | |
| log.info("report document: writing in one piece sections=%d evidence_chars=%d (budget %d) posts_used=%d", len(parts), | |
| sum(len(p) for p in parts), EVIDENCE_TOTAL, len(set(used))) | |
| started = time.monotonic() | |
| out = run.predict(WriteThemedReport, db=ctx.db, workspace_id=ctx.workspace_id, job=ctx.job, lm=report_lm(), | |
| title=ctx.title, storyline=ctx.bp.storyline, request=ctx.instructions, language=ctx.language, | |
| connections=report_blueprint.connection_lines(ctx.bp, ctx.themes), sections=parts) | |
| markdown = MARKER.sub("", (out.get("markdown") or "").strip()) | |
| if not markdown: | |
| log.warning("report document came back empty, reading the posts instead") | |
| return None | |
| log.info("report document written: chars=%d seconds=%.1f", len(markdown), time.monotonic() - started) | |
| return ctx.finish((out.get("report_title") or ctx.bp.title).strip(), number_blocks(split_prose(markdown)), [], used, []) | |
| def _themed_interactive(ctx: ThemedRun, moved: list[tuple[report_blueprint.Section, str, report_blueprint.Section]]) -> Themed | None: | |
| """Wave 1: the body sections together, each given the storyline, the outline, what the others cover and only its own | |
| evidence. Wave 2: the introduction and the summary, written from what wave 1 produced.""" | |
| sections = ctx.bp.sections | |
| body = [(n, s) for n, s in enumerate(sections) if s.role in ("body", "synthesis")] | |
| frame = [(n, s) for n, s in enumerate(sections) if s.role in ("intro", "summary")] | |
| used: list[str] = [] | |
| inputs = [] | |
| for _, s in body: | |
| ev = report_evidence.build(ctx.corpus, s, ctx.by_id) | |
| used += ev.posts | |
| inputs.append(_section_input(ctx, s, ev.blocks)) | |
| report_themes.progress(ctx.db, ctx.job, 0.45, f"Writing {len(body)} sections") | |
| log.info("report wave 1: body sections=%d evidence_chars=%s", len(body), [sum(len(b) for b in i["evidence"]) for i in inputs]) | |
| written: dict[int, dict] = {} | |
| for (n, _), out in zip(body, _write_sections(ctx, inputs), strict=True): | |
| if ((out or {}).get("markdown") or "").strip(): | |
| written[n] = out | |
| if not written: | |
| log.warning("report wave 1 wrote nothing, reading the posts instead") | |
| return None | |
| digests = [f"{s.heading}: {_digest(written[n]['markdown'])}" for n, s in body if n in written] | |
| report_themes.progress(ctx.db, ctx.job, 0.8, "Writing the introduction and summary") | |
| log.info("report wave 2: intro/summary sections=%d digests=%d", len(frame), len(digests)) | |
| for (n, _), out in zip(frame, _write_sections(ctx, [_section_input(ctx, s, digests) for _, s in frame]), strict=True): | |
| if ((out or {}).get("markdown") or "").strip(): | |
| written[n] = out | |
| blocks: list[dict] = [] | |
| suggestions: list[dict] = [] | |
| anchor: dict[int, str] = {} | |
| used_kinds: set[str] = set() | |
| for n, s in enumerate(sections): | |
| out = written.get(n) | |
| if not out: | |
| continue | |
| blocks.append({"type": "prose", "text": f"## {s.heading}\n\n{MARKER.sub('', out['markdown'].strip())}"}) | |
| number_blocks(blocks) | |
| anchor[id(s)] = blocks[-1]["id"] | |
| embed = s.embed | |
| if embed and len(suggestions) < MAX_PLANNED_VISUALS and embed.get("kind") in EMBED_KINDS \ | |
| and embed["kind"] not in used_kinds and (embed.get("brief") or "").strip(): | |
| used_kinds.add(embed["kind"]) | |
| suggestions.append({"id": f"s{len(suggestions) + 1}", "after_block_id": blocks[-1]["id"], "kind": embed["kind"], | |
| "brief": embed["brief"].strip()[:300], "why": (embed.get("why") or "").strip()[:300]}) | |
| for after, kind, sec in moved: | |
| suggestions.append({"id": "", "after_block_id": anchor.get(id(after)) or blocks[0]["id"], "kind": kind, | |
| "brief": (sec.brief or MOVED_BRIEF[kind]).strip()[:300], | |
| "why": "Offered here for you to add, not written into the text."}) | |
| failed = [s.heading for n, s in enumerate(sections) if n not in written] | |
| if failed: | |
| log.warning("report sections left out after a retry: %s", failed) | |
| log.info("report interactive written: blocks=%d suggestions=%d moved_to_suggestions=%d failed=%d", len(blocks), | |
| len(suggestions), len(moved), len(failed)) | |
| suggestions = fill_suggestions(suggestions, blocks, list(EMBED_KINDS)) | |
| return ctx.finish(ctx.bp.title, blocks, merge_suggestions(suggestions), used, failed) | |
| def write_themed(db: Session, workspace_id: uuid.UUID, job: Job, docs: list[Document], title: str, instructions: str, | |
| language: str, interactive: bool) -> Themed | None: | |
| """The report from the posts' themes: group their stored ideas into weighted themes, plan the whole report, build each | |
| section's evidence from several posts, then write. A document is written in one piece from the plan; an interactive report | |
| section by section, each knowing the storyline, the outline and what the others cover, with the introduction and the | |
| summary written last from what the sections say. None when any step cannot be done (too few ideas, no usable plan, nothing | |
| written), so the caller reads the posts as a report always did. Never makes a report fail.""" | |
| started = time.monotonic() | |
| log.info("themed report starting: job=%s workspace=%s format=%s posts=%d language=%r", job.id, workspace_id, | |
| "interactive" if interactive else "document", len(docs), language) | |
| try: | |
| report_themes.progress(db, job, 0.05, f"Reading {len(docs)} posts") | |
| report_themes.ensure_ideas(db, job, workspace_id, docs) | |
| share = report_themes.fresh_share(docs) | |
| if share < report_themes.MIN_FRESH_SHARE: | |
| log.warning("themed report not possible, reading the posts instead: job=%s only %.0f%% of %d posts have ideas " | |
| "(need %.0f%%)", job.id, share * 100, len(docs), report_themes.MIN_FRESH_SHARE * 100) | |
| return None | |
| corpus = Corpus(workspace_id) | |
| corpus.sync() | |
| ideas = report_themes.collect_ideas(docs, corpus) | |
| report_themes.progress(db, job, 0.2, f"Finding the themes in {len(docs)} posts") | |
| themes, relations = report_themes.group_themes(db, job, workspace_id, title, instructions, ideas) | |
| report_themes.progress(db, job, 0.3, "Planning the report") | |
| kinds = list(EMBED_KINDS) if interactive else [] | |
| bp = report_blueprint.plan_blueprint(db, job, workspace_id, title, instructions, themes, relations, kinds, language) | |
| log.info("themed report planned: job=%s themes=%d sections=%d seconds=%.1f", job.id, len(themes), len(bp.sections), | |
| time.monotonic() - started) | |
| moved: list[tuple[report_blueprint.Section, str, report_blueprint.Section]] = [] | |
| if interactive: # a planned section that is really a quiz or flashcards is a suggestion, not text | |
| kept: list[report_blueprint.Section] = [] | |
| for s in bp.sections: | |
| kind = visual_kind(s.heading) | |
| if kind and kept: | |
| log.info("report section %r is a %s, so it becomes a suggestion after %r", s.heading, kind, kept[-1].heading) | |
| moved.append((kept[-1], kind, s)) | |
| if kept[-1].role == "body": # what the section was about is still said, in the section before it | |
| kept[-1].theme_ids += [i for i in s.theme_ids if i not in kept[-1].theme_ids] | |
| else: | |
| kept.append(s) | |
| bp.sections = kept | |
| if len(bp.sections) < 2: | |
| log.warning("themed report not possible, reading the posts instead: job=%s fewer than two sections left", job.id) | |
| return None | |
| ctx = ThemedRun(db, workspace_id, job, docs, corpus, bp, themes, instructions, language, title) | |
| themed = _themed_interactive(ctx, moved) if interactive else _themed_document(ctx) | |
| if themed: | |
| log.info("themed report done: job=%s posts_total=%d posts_used=%d blocks=%d suggestions=%d failed=%d seconds=%.1f", | |
| job.id, len(docs), themed.plan["posts_used"], len(themed.blocks), len(themed.suggestions), | |
| len(themed.failed), time.monotonic() - started) | |
| else: | |
| log.warning("themed report wrote nothing, reading the posts instead: job=%s", job.id) | |
| return themed | |
| except (report_themes.ThemesUnavailable, report_blueprint.BlueprintUnavailable) as exc: | |
| log.warning("themed report not possible, reading the posts instead: job=%s reason=%s", job.id, exc) | |
| except Exception: | |
| db.rollback() | |
| log.warning("themed report failed, reading the posts instead: job=%s", job.id, exc_info=True) | |
| return None | |
| def build_report(db: Session, job: Job, artifact: Artifact) -> dict: | |
| params = job.params | |
| topic = (params.get("topic") or "").strip() | |
| language = (params.get("language") or "").strip() or SAME_LANGUAGE | |
| report_format = "interactive" if params.get("report_format") == "interactive" else "document" | |
| # Posts with ideas stored are read through their themes, however many are picked; chats and small selections as always. | |
| docs = themed_docs(db, artifact, params) | |
| material = None if docs else load_material(db, artifact.workspace_id, artifact.notebook_id, params, job=job, what="report") | |
| about = topic or (themed_title(db, artifact, params, docs) if docs else material.title) | |
| instructions = (params.get("instructions") or "").strip() or default_instructions( | |
| report_format, params.get("template_id"), about) | |
| themed = write_themed(db, artifact.workspace_id, job, docs, themed_title(db, artifact, params, docs), instructions, language, | |
| report_format == "interactive") if docs else None | |
| log.info("report: job=%s artifact=%s format=%s way=%s", job.id, artifact.id, report_format, | |
| "themes" if themed else "legacy (posts read as text)") | |
| if themed: | |
| extra = {"plan": themed.plan, **({"failed_sections": themed.failed} if themed.failed else {})} | |
| title, blocks, suggestions = themed.title, themed.blocks, themed.suggestions | |
| source, sources = themed.source, themed.sources | |
| else: | |
| material = material or load_material(db, artifact.workspace_id, artifact.notebook_id, params, job=job, what="report") | |
| extra = {} | |
| made = write_interactive(db, artifact.workspace_id, job, material, instructions, language) \ | |
| if report_format == "interactive" else None | |
| if made: | |
| title, blocks, suggestions = made | |
| else: | |
| # A document, or an interactive report whose plan came back unusable: write it as one piece. Never a failure. | |
| report_format = "document" | |
| update_job(db, job, progress=0.3, message="Writing the report") | |
| title, markdown = write_document(db, artifact.workspace_id, job, material, instructions, language) | |
| blocks = number_blocks(split_prose(markdown)) | |
| suggestions = [] # visuals are offered in interactive reports only, never in a document | |
| source, sources = material.source, source_rows(material) | |
| artifact.content_json = { | |
| "title": title, "format": report_format, "template_id": params.get("template_id"), | |
| "instructions": instructions, "language": language, "topic": topic, "source": source, | |
| "blocks": blocks, "suggestions": suggestions, "suggestions_offered": True, "citations": [], "sources": sources, **extra, | |
| } | |
| artifact.status = "ready" | |
| db.commit() | |
| return {"artifact_id": str(artifact.id)} | |
| # Adding a visual to a finished report | |
| def _source_params(source: dict) -> dict: | |
| """The job params that load a report's own source again.""" | |
| return {"chat_ids": source.get("chat_ids") or [], "document_ids": source.get("document_ids") or []} | |
| def copy_block(kind: str, existing: Artifact) -> dict: | |
| """A visual's content taken from an artifact the user already made (a Mind Constellation, a quiz or flashcards).""" | |
| c = existing.content_json or {} | |
| if kind == "mind_map" and c.get("root"): | |
| return {"title": c.get("title"), "root": c["root"], "node_count": c.get("node_count"), | |
| "post_count": c.get("post_count")} | |
| if kind == "quiz" and c.get("questions"): | |
| return {"title": c.get("title"), "questions": c["questions"]} | |
| if kind == "flashcards" and c.get("cards"): | |
| return {"title": c.get("title"), "cards": c["cards"]} | |
| if kind == "infographic" and c.get("content"): # the picture itself is shared, not copied | |
| return {"title": c.get("title"), "artifact_id": str(existing.id), "theme": c.get("theme")} | |
| raise NothingToDo("That one has nothing to add.") | |
| def make_block(db: Session, job: Job, report: Artifact, kind: str, brief: str, theme: str | None = None) -> dict: | |
| """A new visual about `brief`, built from the report's own sources.""" | |
| source = (report.content_json or {})["source"] | |
| if kind == "infographic": | |
| # A picture is drawn by the renderer, so it is its own infographic (listed in the Library, counted in the | |
| # plan) that the block points at; the report page shows it once it is ready. | |
| title = f"Infographic: {brief[:60] or report.content_json.get('title', '')}" | |
| child, _ = start_artifact(db, report.workspace_id, "infographic", title=title, | |
| notebook_id=report.notebook_id, | |
| params={**_source_params(source), "theme": theme_id(theme), "instructions": brief[:600]}) | |
| return {"title": f"Infographic: {brief[:60]}", "artifact_id": str(child.id), "theme": theme_id(theme)} | |
| material = load_material(db, report.workspace_id, report.notebook_id, _source_params(source), job=job, what="report") | |
| language = (report.content_json or {}).get("language") or SAME_LANGUAGE | |
| update_job(db, job, progress=0.4, message="Building it") | |
| if kind == "mind_map": | |
| if material.is_chat: | |
| raise NothingToDo("A Mind Constellation needs posts, not chats.") | |
| c = mind_map_content(db, job, report.workspace_id, material.title, material.docs, brief) | |
| return {"title": c["title"], "root": c["root"], "node_count": c["node_count"], "post_count": c["post_count"]} | |
| if kind == "flashcards": | |
| cards = make_cards(db, report.workspace_id, material, topic=brief, count=BLOCK_CARDS, language=language, job=job) | |
| if not cards: | |
| raise NothingToDo("The sources did not give enough to make cards from.") | |
| return {"title": f"Flashcards: {brief[:60]}", "cards": cards} | |
| questions = make_questions(db, report.workspace_id, material, topic=brief, difficulty="medium", | |
| types=["multiple_choice", "multiple_select", "fill_blank"], count=BLOCK_QUESTIONS, | |
| language=language, job=job) | |
| if not questions: | |
| raise NothingToDo("The sources did not give enough to ask about.") | |
| return {"title": f"Test yourself: {brief[:60]}", "questions": questions} | |
| def insert_block(content: dict, block: dict, after: str | None, suggestion_id: str | None) -> dict: | |
| """The report's content with the block placed after `after` (or last), the suggestion it answers removed, and | |
| nothing else changed. Block ids stay unique.""" | |
| blocks = list(content.get("blocks") or []) | |
| used = {b["id"] for b in blocks} | |
| n = len(blocks) + 1 | |
| while f"b{n}" in used: | |
| n += 1 | |
| block = {**block, "id": f"b{n}", "after": after} | |
| at = next((i + 1 for i, b in enumerate(blocks) if b["id"] == after), len(blocks)) | |
| # Keep the block with the visuals already hanging off the same section, in the order they were added. | |
| while at < len(blocks) and blocks[at]["type"] != "prose" and blocks[at].get("after") == after: | |
| at += 1 | |
| blocks.insert(at, block) | |
| suggestions = [s for s in content.get("suggestions") or [] if s["id"] != suggestion_id] | |
| return {**content, "blocks": blocks, "suggestions": suggestions} | |
| def handle_report_block(db: Session, job: Job) -> dict: | |
| """Worker job: add one visual (new, or copied from an existing artifact) to a report.""" | |
| report = db.get(Artifact, uuid.UUID(job.params["artifact_id"])) | |
| if not report or report.type != "report": | |
| return {} | |
| p = job.params | |
| kind = p.get("kind") | |
| if kind not in EMBED_KINDS: | |
| raise PermanentJobError("Unknown kind of visual.") | |
| try: | |
| if p.get("existing_artifact_id"): | |
| existing = db.get(Artifact, uuid.UUID(p["existing_artifact_id"])) | |
| if not existing or existing.workspace_id != report.workspace_id or existing.type != kind \ | |
| or existing.status != "ready": | |
| raise NothingToDo("That one is not available.") | |
| block = copy_block(kind, existing) | |
| else: | |
| block = make_block(db, job, report, kind, (p.get("brief") or "").strip(), p.get("theme")) | |
| keep_visual(db, report, {"type": kind, **block}) # a new one also lands in the notebook's Studio list | |
| except NothingToDo as exc: | |
| raise PermanentJobError(str(exc)) from exc | |
| report.content_json = insert_block(report.content_json or {}, {"type": kind, **block}, p.get("after_block_id"), | |
| p.get("suggestion_id")) | |
| db.commit() | |
| return {"artifact_id": str(report.id)} | |
| def keep_visual(db: Session, report: Artifact, block: dict) -> Artifact | None: | |
| """A mind map, flashcards or quiz made inside a report would exist only as that block. It is also saved as an | |
| artifact of its own in the notebook, so it shows in the Studio list and "use one you already made" offers it | |
| again after the block is removed. Content that is already saved is not saved twice.""" | |
| kind = block.get("type") | |
| source = (report.content_json or {}).get("source") or {} | |
| if kind == "flashcards" and block.get("cards"): | |
| key, content = "cards", {"cards": block["cards"], "card_count": len(block["cards"])} | |
| elif kind == "quiz" and block.get("questions"): | |
| key, content = "questions", {"questions": block["questions"], "question_count": len(block["questions"])} | |
| elif kind == "mind_map" and block.get("root"): | |
| key, content = "root", {"root": block["root"], "node_count": block.get("node_count"), | |
| "post_count": block.get("post_count"), | |
| "document_ids": source.get("document_ids") or []} | |
| else: | |
| return None # an infographic already is its own artifact | |
| same = db.scalars(select(Artifact).where(Artifact.workspace_id == report.workspace_id, Artifact.type == kind, | |
| Artifact.notebook_id == report.notebook_id)) | |
| if any((a.content_json or {}).get(key) == content[key] for a in same): | |
| return None | |
| saved = Artifact(workspace_id=report.workspace_id, notebook_id=report.notebook_id, type=kind, status="ready", | |
| content_json={"title": block.get("title") or kind, "source": source, **content}) | |
| db.add(saved) | |
| db.commit() | |
| return saved | |
| def remove_block(content: dict, block_id: str) -> dict | None: | |
| """The content without that visual (prose cannot be removed), with the offer to add it again restored, or None | |
| when there is no such visual.""" | |
| blocks = list(content.get("blocks") or []) | |
| removed = next((b for b in blocks if b["id"] == block_id and b["type"] != "prose"), None) | |
| if not removed: | |
| return None | |
| keep = [b for b in blocks if b["id"] != block_id] | |
| suggestions = list(content.get("suggestions") or []) | |
| kind = removed["type"] | |
| if kind in FALLBACK_BRIEFS and not any(s["kind"] == kind for s in suggestions): | |
| brief, why = FALLBACK_BRIEFS[kind] | |
| suggestions.append({"id": "", "after_block_id": removed.get("after"), "kind": kind, "brief": brief, "why": why}) | |
| return {**content, "blocks": keep, "suggestions": suggestions} | |