"""
title: Flash Card
author: Fu-Jie
author_url: https://github.com/Fu-Jie
funding_url: https://github.com/Fu-Jie/awesome-openwebui
version: 0.2.1
icon_url: data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIyNCIgaGVpZ2h0PSIyNCIgdmlld0JveD0iMCAwIDI0IDI0IiBmaWxsPSJub25lIiBzdHJva2U9ImN1cnJlbnRDb2xvciIgc3Ryb2tlLXdpZHRoPSIyIiBzdHJva2UtbGluZWNhcD0icm91bmQiIHN0cm9rZS1saW5lam9pbj0icm91bmQiPjxwb2x5Z29uIHBvaW50cz0iMTIgMiAyIDcgMTIgMTIgMjIgNyAxMiAyIi8+PHBvbHlsaW5lIHBvaW50cz0iMiAxNyAxMiAyMiAyMiAxNyIvPjxwb2x5bGluZSBwb2ludHM9IjIgMTIgMTIgMTcgMjIgMTIiLz48L3N2Zz4=
description: Quickly generates beautiful flashcards from text, extracting key points and categories.
"""
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any, List
import json
import logging
from open_webui.utils.chat import generate_chat_completion
from open_webui.models.users import Users
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
HTML_WRAPPER_TEMPLATE = """
"""
class Action:
class Valves(BaseModel):
MODEL_ID: str = Field(
default="",
description="Model ID used for generating card content. If empty, uses the current model.",
)
MIN_TEXT_LENGTH: int = Field(
default=50,
description="Minimum text length required to generate a flashcard (characters).",
)
MAX_TEXT_LENGTH: int = Field(
default=2000,
description="Recommended maximum text length. For longer texts, deep analysis tools are recommended.",
)
LANGUAGE: str = Field(
default="en",
description="Target language for card content (e.g., 'en', 'zh').",
)
SHOW_STATUS: bool = Field(
default=True,
description="Whether to show status updates in the chat interface.",
)
CLEAR_PREVIOUS_HTML: bool = Field(
default=False,
description="Whether to force clear previous plugin results (if True, overwrites instead of merging).",
)
MESSAGE_COUNT: int = Field(
default=1,
description="Number of recent messages to use for generation. Set to 1 for just the last message, or higher for more context.",
)
def __init__(self):
self.valves = self.Valves()
async def action(
self,
body: dict,
__user__: Optional[Dict[str, Any]] = None,
__event_emitter__: Optional[Any] = None,
__request__: Optional[Any] = None,
) -> Optional[dict]:
logger.info(f"Action: {__name__} triggered")
if not __event_emitter__:
return body
# Get messages based on MESSAGE_COUNT
messages = body.get("messages", [])
if not messages:
return body
# Get last N messages based on MESSAGE_COUNT
message_count = min(self.valves.MESSAGE_COUNT, len(messages))
recent_messages = messages[-message_count:]
# Aggregate content from selected messages with labels
aggregated_parts = []
for i, msg in enumerate(recent_messages, 1):
text_content = self._extract_text_content(msg.get("content"))
if text_content:
role = msg.get("role", "unknown")
role_label = (
"User"
if role == "user"
else "Assistant" if role == "assistant" else role
)
aggregated_parts.append(f"[{role_label} Message {i}]\n{text_content}")
if not aggregated_parts:
return body
target_message = "\n\n---\n\n".join(aggregated_parts)
# Check text length
text_length = len(target_message)
if text_length < self.valves.MIN_TEXT_LENGTH:
await self._emit_notification(
__event_emitter__,
f"Text too short ({text_length} chars), recommended at least {self.valves.MIN_TEXT_LENGTH} chars.",
"warning",
)
return body
if text_length > self.valves.MAX_TEXT_LENGTH:
await self._emit_notification(
__event_emitter__,
f"Text quite long ({text_length} chars), consider using 'Deep Reading' for deep analysis.",
"info",
)
# Notify user that we are generating the card
await self._emit_notification(
__event_emitter__, "⚡ Generating Flash Card...", "info"
)
await self._emit_status(
__event_emitter__, "⚡ Flash Card: Starting generation...", done=False
)
try:
# 1. Extract information using LLM
await self._emit_status(
__event_emitter__,
"⚡ Flash Card: Calling AI model to analyze content...",
done=False,
)
user_id = __user__.get("id") if __user__ else "default"
user_obj = Users.get_user_by_id(user_id)
target_model = (
self.valves.MODEL_ID if self.valves.MODEL_ID else body.get("model")
)
system_prompt = f"""
You are a Flash Card Generation Expert, specializing in creating knowledge cards suitable for learning and memorization. Your task is to distill text into concise, easy-to-remember flashcards.
Please extract the following fields and return them in JSON format:
1. "title": Create a short, precise title (3-8 words), highlighting the core concept.
2. "summary": Summarize the core essence in one sentence (10-25 words), making it easy to understand and remember.
3. "key_points": List 3-5 key memory points (5-15 words each).
- Each point should be an independent knowledge unit.
- Use concise, conversational expression.
- Avoid long sentences.
4. "tags": List 2-4 classification tags (1-3 words each).
5. "category": Choose a main category (e.g., Concept, Skill, Fact, Method, etc.).
Target Language: {self.valves.LANGUAGE}
Important Principles:
- **Minimalism**: Refine each point to the extreme.
- **Memory First**: Content should be easy to memorize and recall.
- **Core Focus**: Extract only the most core knowledge points.
- **Conversational**: Use easy-to-understand language.
- Return ONLY the JSON object, do not include markdown formatting.
"""
prompt = f"Please refine the following text into a learning flashcard:\n\n{target_message}"
payload = {
"model": target_model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt},
],
"stream": False,
}
response = await generate_chat_completion(__request__, payload, user_obj)
content = response["choices"][0]["message"]["content"]
await self._emit_status(
__event_emitter__,
"⚡ Flash Card: AI analysis complete, parsing data...",
done=False,
)
# Parse JSON
try:
# simple cleanup in case of markdown code blocks
if "```json" in content:
content = content.split("```json")[1].split("```")[0].strip()
elif "```" in content:
content = content.split("```")[1].split("```")[0].strip()
card_data = json.loads(content)
except Exception as e:
logger.error(f"Failed to parse JSON: {e}, content: {content}")
await self._emit_status(
__event_emitter__, "❌ Flash Card: Data parsing failed", done=True
)
await self._emit_notification(
__event_emitter__,
"❌ Failed to generate card data, please try again.",
"error",
)
return body
# 2. Generate HTML components
await self._emit_status(
__event_emitter__, "⚡ Flash Card: Rendering card...", done=False
)
card_content, card_style = self.generate_html_card_components(card_data)
# 3. Append to message
# Extract existing HTML if any
existing_html_block = ""
match = re.search(
r"```html\s*([\s\S]*?)```",
body["messages"][-1]["content"],
)
if match:
existing_html_block = match.group(1)
if self.valves.CLEAR_PREVIOUS_HTML:
body["messages"][-1]["content"] = self._remove_existing_html(
body["messages"][-1]["content"]
)
final_html = self._merge_html(
"", card_content, card_style, "", self.valves.LANGUAGE
)
else:
if existing_html_block:
body["messages"][-1]["content"] = self._remove_existing_html(
body["messages"][-1]["content"]
)
final_html = self._merge_html(
existing_html_block,
card_content,
card_style,
"",
self.valves.LANGUAGE,
)
else:
final_html = self._merge_html(
"", card_content, card_style, "", self.valves.LANGUAGE
)
html_embed_tag = f"```html\n{final_html}\n```"
body["messages"][-1]["content"] += f"\n\n{html_embed_tag}"
await self._emit_status(
__event_emitter__, "✅ Flash Card: Generation complete!", done=True
)
await self._emit_notification(
__event_emitter__, "⚡ Flash Card generated successfully!", "success"
)
return body
except Exception as e:
logger.error(f"Error generating knowledge card: {e}")
await self._emit_status(
__event_emitter__, "❌ Flash Card: Generation failed", done=True
)
await self._emit_notification(
__event_emitter__,
f"❌ Error generating knowledge card: {str(e)}",
"error",
)
return body
async def _emit_status(self, emitter, description: str, done: bool = False):
"""Emits a status update event."""
if self.valves.SHOW_STATUS and emitter:
await emitter(
{"type": "status", "data": {"description": description, "done": done}}
)
async def _emit_notification(self, emitter, content: str, ntype: str = "info"):
"""Emits a notification event (info/success/warning/error)."""
if emitter:
await emitter(
{"type": "notification", "data": {"type": ntype, "content": content}}
)
def _remove_existing_html(self, content: str) -> str:
"""Removes existing plugin-generated HTML code blocks from the content."""
pattern = r"```html\s*[\s\S]*?```"
return re.sub(pattern, "", content).strip()
def _extract_text_content(self, content) -> str:
"""Extract text from message content, supporting multimodal message formats"""
if isinstance(content, str):
return content
elif isinstance(content, list):
# Multimodal message: [{"type": "text", "text": "..."}, {"type": "image_url", ...}]
text_parts = []
for item in content:
if isinstance(item, dict) and item.get("type") == "text":
text_parts.append(item.get("text", ""))
elif isinstance(item, str):
text_parts.append(item)
return "\n".join(text_parts)
return str(content) if content else ""
def _merge_html(
self,
existing_html_code: str,
new_content: str,
new_styles: str = "",
new_scripts: str = "",
user_language: str = "en-US",
) -> str:
"""
Merges new content into an existing HTML container, or creates a new one.
"""
if (
"" in existing_html_code
and "" in existing_html_code
):
base_html = existing_html_code
base_html = re.sub(r"^```html\s*", "", base_html)
base_html = re.sub(r"\s*```$", "", base_html)
else:
base_html = HTML_WRAPPER_TEMPLATE.replace("{user_language}", user_language)
wrapped_content = f'