68 lines
2.9 KiB
Python
68 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
class ProjectService:
|
|
def __init__(self, workspace_dir: Path) -> None:
|
|
self.workspace_dir = workspace_dir
|
|
self.projects_dir = workspace_dir / "projects"
|
|
self.index_dir = workspace_dir / "library" / "index"
|
|
|
|
def add_items(self, project_id: str, item_keys: list[str]) -> dict[str, object]:
|
|
selected_items = self._read_selected_items(project_id)
|
|
selected_items = sorted(set(selected_items + item_keys))
|
|
self._write_json(self._project_dir(project_id) / "selected-items.json", selected_items)
|
|
return self._rebuild_project_index(project_id, selected_items)
|
|
|
|
def remove_item(self, project_id: str, item_key: str) -> dict[str, object]:
|
|
selected_items = [key for key in self._read_selected_items(project_id) if key != item_key]
|
|
self._write_json(self._project_dir(project_id) / "selected-items.json", selected_items)
|
|
return self._rebuild_project_index(project_id, selected_items)
|
|
|
|
def get_project_view(self, project_id: str) -> dict[str, object]:
|
|
index_path = self._project_dir(project_id) / "project-index.json"
|
|
if not index_path.exists():
|
|
selected_items = self._read_selected_items(project_id)
|
|
return self._rebuild_project_index(project_id, selected_items)
|
|
return json.loads(index_path.read_text(encoding="utf-8"))
|
|
|
|
def _rebuild_project_index(self, project_id: str, selected_items: list[str]) -> dict[str, object]:
|
|
cards_index = self._read_json(self.index_dir / "cards.json")
|
|
collections_index = self._read_json(self.index_dir / "collections.json")
|
|
project_cards = [cards_index[item_key] for item_key in selected_items if item_key in cards_index]
|
|
project_collections = [
|
|
payload
|
|
for payload in collections_index.values()
|
|
if set(payload.get("item_keys", [])) & set(selected_items)
|
|
]
|
|
payload = {
|
|
"project_id": project_id,
|
|
"selected_items": selected_items,
|
|
"cards": project_cards,
|
|
"collections": project_collections,
|
|
}
|
|
self._write_json(self._project_dir(project_id) / "project-index.json", payload)
|
|
return payload
|
|
|
|
def _read_selected_items(self, project_id: str) -> list[str]:
|
|
path = self._project_dir(project_id) / "selected-items.json"
|
|
if not path.exists():
|
|
return []
|
|
return [str(value) for value in json.loads(path.read_text(encoding="utf-8"))]
|
|
|
|
def _project_dir(self, project_id: str) -> Path:
|
|
return self.projects_dir / project_id
|
|
|
|
@staticmethod
|
|
def _read_json(path: Path) -> dict[str, object]:
|
|
if not path.exists():
|
|
return {}
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
@staticmethod
|
|
def _write_json(path: Path, payload: object) -> None:
|
|
path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
|