cppa-cursor-browser/api/composers.py at 7bbfe43d626d50b76beecdc75838000ba5772e79 · cppalliance/cppa-cursor-browser · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
"""
API routes for composers — mirrors:
  src/app/api/composers/route.ts       GET /api/composers
  src/app/api/composers/[id]/route.ts  GET /api/composers/<id>
"""

import json
import logging
import os
import sqlite3
from contextlib import closing

from flask import Blueprint, jsonify

from utils.workspace_path import resolve_workspace_path
from utils.path_helpers import to_epoch_ms

bp = Blueprint("composers", __name__)
_logger = logging.getLogger(__name__)


def _read_json_file(path: str):
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


@bp.route("/api/composers")
def list_composers():
    try:
        workspace_path = resolve_workspace_path()
        composers = []

        for name in os.listdir(workspace_path):
            full = os.path.join(workspace_path, name)
            if not os.path.isdir(full):
                continue

            db_path = os.path.join(full, "state.vscdb")
            wj_path = os.path.join(full, "workspace.json")
            if not os.path.isfile(db_path):
                continue

            workspace_folder = None
            try:
                wd = _read_json_file(wj_path)
                workspace_folder = wd.get("folder")
            except Exception:
                pass

            try:
                # closing() guarantees .close() on scope exit (issue #17).
                with closing(sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)) as conn:
                    row = conn.execute(
                        "SELECT value FROM ItemTable WHERE [key] = 'composer.composerData'"
                    ).fetchone()

                if row and row[0]:
                    data = json.loads(row[0])
                    all_composers = data.get("allComposers")
                    if isinstance(all_composers, list):
                        for c in all_composers:
                            c["conversation"] = c.get("conversation") or []
                            c["workspaceId"] = name
                            c["workspaceFolder"] = workspace_folder
                            composers.append(c)
            except Exception:
                pass

        composers.sort(key=lambda c: to_epoch_ms(c.get("lastUpdatedAt")), reverse=True)
        return jsonify(composers)

    except Exception:
        _logger.exception("Failed to get composers")
        return jsonify({"error": "Failed to get composers"}), 500


@bp.route("/api/composers/<composer_id>")
def get_composer(composer_id):
    try:
        workspace_path = resolve_workspace_path()

        # Search per-workspace databases
        for name in os.listdir(workspace_path):
            full = os.path.join(workspace_path, name)
            if not os.path.isdir(full):
                continue
            db_path = os.path.join(full, "state.vscdb")
            if not os.path.isfile(db_path):
                continue

            try:
                # closing() guarantees .close() on scope exit (issue #17).
                with closing(sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)) as conn:
                    row = conn.execute(
                        "SELECT value FROM ItemTable WHERE [key] = 'composer.composerData'"
                    ).fetchone()

                if row and row[0]:
                    data = json.loads(row[0])
                    for c in (data.get("allComposers") or []):
                        if c.get("composerId") == composer_id:
                            return jsonify(c)
            except Exception:
                pass

        # Fallback: global storage
        global_db_path = os.path.normpath(os.path.join(workspace_path, "..", "globalStorage", "state.vscdb"))
        if os.path.isfile(global_db_path):
            try:
                # closing() guarantees .close() on scope exit (issue #17).
                with closing(sqlite3.connect(f"file:{global_db_path}?mode=ro", uri=True)) as conn:
                    row = conn.execute(
                        "SELECT value FROM cursorDiskKV WHERE key = ?",
                        (f"composerData:{composer_id}",),
                    ).fetchone()

                if row and row[0]:
                    raw = row[0] if isinstance(row[0], str) else row[0].decode("utf-8")
                    composer = json.loads(raw)
                    composer.setdefault("conversation", [])
                    return jsonify(composer)
            except Exception:
                pass

        return jsonify({"error": "Composer not found"}), 404

    except Exception:
        _logger.exception("Failed to get composer")
        return jsonify({"error": "Failed to get composer"}), 500