import asyncio import logging from datetime import date from quart import Blueprint, Response, jsonify, request from fabledassistant.auth import login_required, get_current_user_id from fabledassistant.config import Config from fabledassistant.services.assist import build_assist_messages from fabledassistant.services.generation_buffer import ( GenerationState, create_assist_buffer, get_assist_buffer, ) from fabledassistant.services.generation_task import run_assist_generation from fabledassistant.services.notes import ( convert_note_to_task, convert_task_to_note, create_note, delete_note, get_all_tags, get_backlinks, get_note, get_note_by_title, get_or_create_note_by_title, list_notes, update_note, ) from fabledassistant.services.settings import get_setting from fabledassistant.services.tag_suggestions import suggest_tags from fabledassistant.utils.tags import extract_tags logger = logging.getLogger(__name__) notes_bp = Blueprint("notes", __name__, url_prefix="/api/notes") @notes_bp.route("", methods=["GET"]) @login_required async def list_notes_route(): uid = get_current_user_id() q = request.args.get("q") tag = request.args.getlist("tag") sort = request.args.get("sort", "updated_at") order = request.args.get("order", "desc") limit = request.args.get("limit", 50, type=int) offset = request.args.get("offset", 0, type=int) # Default to non-task notes only; ?is_task=true for tasks, ?all=true for everything is_task: bool | None = False if request.args.get("all", "").lower() == "true": is_task = None elif request.args.get("is_task", "").lower() == "true": is_task = True notes, total = await list_notes( uid, q=q, tags=tag or None, is_task=is_task, sort=sort, order=order, limit=limit, offset=offset ) return jsonify({"notes": [n.to_dict() for n in notes], "total": total}) @notes_bp.route("", methods=["POST"]) @login_required async def create_note_route(): uid = get_current_user_id() data = await request.get_json() body = data.get("body", "") tags = extract_tags(body) # Optional task fields status = data.get("status") priority = data.get("priority") due_date = None if data.get("due_date"): due_date = date.fromisoformat(data["due_date"]) note = await create_note( uid, title=data.get("title", ""), body=body, tags=tags, parent_id=data.get("parent_id"), status=status, priority=priority, due_date=due_date, ) return jsonify(note.to_dict()), 201 @notes_bp.route("/tags", methods=["GET"]) @login_required async def list_tags_route(): uid = get_current_user_id() q = request.args.get("q") tags = await get_all_tags(uid, q=q) return jsonify({"tags": tags}) @notes_bp.route("/suggest-tags", methods=["POST"]) @login_required async def suggest_tags_route(): uid = get_current_user_id() data = await request.get_json() title = data.get("title", "") body = data.get("body", "") tags = await suggest_tags(uid, title, body) return jsonify({"suggested_tags": tags}) @notes_bp.route("//append-tag", methods=["POST"]) @login_required async def append_tag_route(note_id: int): uid = get_current_user_id() data = await request.get_json() tag = data.get("tag", "").strip() if not tag: return jsonify({"error": "tag is required"}), 400 note = await get_note(uid, note_id) if note is None: return jsonify({"error": "Note not found"}), 404 # Append #tag to body new_body = note.body.rstrip() + f"\n#{tag}" if note.body else f"#{tag}" tags = extract_tags(new_body) updated = await update_note(uid, note_id, body=new_body, tags=tags) return jsonify(updated.to_dict()) @notes_bp.route("/by-title", methods=["GET"]) @login_required async def get_note_by_title_route(): uid = get_current_user_id() title = request.args.get("title", "") if not title: return jsonify({"error": "title parameter is required"}), 400 note = await get_note_by_title(uid, title) if note is None: return jsonify({"error": "Note not found"}), 404 return jsonify(note.to_dict()) @notes_bp.route("/resolve-title", methods=["POST"]) @login_required async def resolve_title_route(): uid = get_current_user_id() data = await request.get_json() title = data.get("title", "").strip() if not title: return jsonify({"error": "title is required"}), 400 note = await get_or_create_note_by_title(uid, title) return jsonify(note.to_dict()) @notes_bp.route("/", methods=["GET"]) @login_required async def get_note_route(note_id: int): uid = get_current_user_id() note = await get_note(uid, note_id) if note is None: return jsonify({"error": "Note not found"}), 404 return jsonify(note.to_dict()) @notes_bp.route("/", methods=["PUT"]) @login_required async def update_note_route(note_id: int): uid = get_current_user_id() data = await request.get_json() fields = {} for key in ("title", "body", "parent_id", "status", "priority"): if key in data: fields[key] = data[key] if "due_date" in data: fields["due_date"] = ( date.fromisoformat(data["due_date"]) if data["due_date"] else None ) if "body" in fields: fields["tags"] = extract_tags(fields["body"]) note = await update_note(uid, note_id, **fields) if note is None: return jsonify({"error": "Note not found"}), 404 return jsonify(note.to_dict()) @notes_bp.route("/", methods=["DELETE"]) @login_required async def delete_note_route(note_id: int): uid = get_current_user_id() deleted = await delete_note(uid, note_id) if not deleted: return jsonify({"error": "Note not found"}), 404 return "", 204 @notes_bp.route("//convert-to-task", methods=["POST"]) @login_required async def convert_note_to_task_route(note_id: int): uid = get_current_user_id() try: note = await convert_note_to_task(uid, note_id) return jsonify(note.to_dict()), 200 except ValueError as e: return jsonify({"error": str(e)}), 404 @notes_bp.route("//convert-to-note", methods=["POST"]) @login_required async def convert_task_to_note_route(note_id: int): uid = get_current_user_id() try: note = await convert_task_to_note(uid, note_id) return jsonify(note.to_dict()), 200 except ValueError as e: return jsonify({"error": str(e)}), 404 @notes_bp.route("//backlinks", methods=["GET"]) @login_required async def get_backlinks_route(note_id: int): uid = get_current_user_id() links = await get_backlinks(uid, note_id) return jsonify({"backlinks": links}) @notes_bp.route("/assist", methods=["POST"]) @login_required async def assist_route(): """Launch a background assist generation task.""" uid = get_current_user_id() data = await request.get_json() body = data.get("body", "") target_section = data.get("target_section", "") instruction = data.get("instruction", "") if not target_section or not instruction: return jsonify({"error": "target_section and instruction are required"}), 400 # Reject if an assist generation is already running for this user existing = get_assist_buffer(uid) if existing and existing.state == GenerationState.RUNNING: return jsonify({"error": "Assist generation already running"}), 409 model = await get_setting(uid, "default_model", Config.OLLAMA_MODEL) messages = build_assist_messages(body, target_section, instruction) buf = create_assist_buffer(uid) asyncio.create_task(run_assist_generation(buf, messages, model)) return jsonify({"status": "started"}), 202 @notes_bp.route("/assist/stream", methods=["GET"]) @login_required async def assist_stream_route(): """SSE endpoint that tails the assist generation buffer.""" uid = get_current_user_id() buf = get_assist_buffer(uid) if buf is None: return jsonify({"error": "No active assist generation"}), 404 last_id_str = request.headers.get("Last-Event-ID") or request.args.get("last_event_id") last_id = int(last_id_str) if last_id_str is not None else -1 async def stream(): cursor = last_id while True: pending = buf.events_after(cursor) for event in pending: yield buf.format_sse(event) cursor = event.index if buf.state != GenerationState.RUNNING: break got_new = await buf.wait_for_event(cursor, timeout=15.0) if not got_new: yield ": keepalive\n\n" return Response( stream(), content_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", }, )