Model/backend/magic_plan/magic_plan_client.py
2026-05-01 16:20:55 +00:00

51 lines
1.9 KiB
Python

import requests
from backend.magic_plan.models import (
MagicPlanPlan,
MagicPlanSummary,
MagicPlanXMLDetail,
MagicPlanXMLSummary,
)
from backend.magic_plan.xml_parser import parse_magicplan_xml
def _parse_xml_summary(data: dict[str, object]) -> MagicPlanXMLSummary:
return MagicPlanXMLSummary(
id=str(data.get("id", "")),
project_id=str(data.get("project_id", "")),
name=str(data.get("name", "")),
address=str(data.get("address", "")),
creation_date=str(data.get("creation_date", "")),
update_date=str(data.get("update_date", "")),
workgroup_id=str(data.get("workgroup_id", "")),
team_id=str(data.get("team_id", "")),
created_by=str(data.get("created_by", "")),
thumbnail_url=str(data.get("thumbnail_url", "")),
public_url=str(data.get("public_url", "")),
cloud_url=str(data.get("cloud_url", "")),
url_3d=str(data.get("3d_url", "")),
)
class MagicPlanClient:
BASE_URL = "https://cloud.magicplan.app/api/v2"
def __init__(self, api_key: str) -> None:
self._session = requests.Session()
self._session.headers.update({"Authorization": f"Bearer {api_key}"})
def get_plans(self, _project_id: str) -> list[MagicPlanSummary]:
raise NotImplementedError
def get_plan_xml(self, plan_id: str) -> MagicPlanXMLDetail:
resp = self._session.get(f"{self.BASE_URL}/plans/get/{plan_id}")
resp.raise_for_status()
data: dict[str, object] = resp.json()["data"]
plan_data = data["plan"] # type: ignore[index]
detail_data = data["plan_detail"] # type: ignore[index]
summary = _parse_xml_summary(plan_data) # type: ignore[arg-type]
plan_xml = parse_magicplan_xml(detail_data["magicplan_format_xml"]) # type: ignore[index]
return MagicPlanXMLDetail(summary=summary, plan_xml=plan_xml)
def get_plan(self, _plan_id: str) -> MagicPlanPlan:
raise NotImplementedError