add xiaozhi server

wangzhibo

2026-08-30 c8515170fd399b4d2dfaebb59128d4674e58915e

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
import json 
from copy import deepcopy 
from pathlib import Path 
from typing import Any 
 
from app.config import settings 
 
 
class CourseService: 
    def __init__(self, data_dir: Path | None = None) -> None: 
        self.data_dir = data_dir or settings.data_dir 
 
    def _read_json(self, path: Path) -> dict[str, Any]: 
        with path.open(encoding="utf-8") as f: 
            return json.load(f) 
 
    def static_url(self, server_rel_path: str, base_url: str) -> str: 
        rel = server_rel_path.replace("\\", "/").lstrip("/") 
        return f"{base_url.rstrip('/')}/static/{rel}" 
 
    def _resolve_media_url(self, path: str, base_url: str) -> str: 
        if path.startswith("http://") or path.startswith("https://"): 
            return path 
        return self.static_url(path, base_url) 
 
    def _resolve_asset_urls(self, asset: dict[str, Any], base_url: str) -> dict[str, Any]: 
        resolved = dict(asset) 
        if resolved.get("url"): 
            resolved["url"] = self._resolve_media_url(resolved["url"], base_url) 
        if resolved.get("cover_url"): 
            resolved["cover_url"] = self._resolve_media_url(resolved["cover_url"], base_url) 
        return resolved 
 
    def get_course_index(self) -> dict[str, Any]: 
        return self._read_json(self.data_dir / "courses" / "index.json") 
 
    def get_course(self, course_id: str) -> dict[str, Any]: 
        index = self.get_course_index() 
        for item in index.get("courses", []): 
            if item["course_id"] == course_id: 
                config_path = self.data_dir / item["config"] 
                return self._read_json(config_path) 
        raise FileNotFoundError(f"Course not found: {course_id}") 
 
    def get_tasklist(self) -> dict[str, Any]: 
        return self._read_json(self.data_dir / "tasklist.json") 
 
    def get_course_task(self, course_id: str) -> dict[str, Any]: 
        path = self.data_dir / "courses" / course_id / "courseTask.json" 
        if not path.exists(): 
            raise FileNotFoundError(f"Course task not found: {course_id}") 
        return self._read_json(path) 
 
    def build_course_task_sync_payload(self, course_id: str, base_url: str) -> dict[str, Any]: 
        task = deepcopy(self.get_course_task(course_id)) 
        assets = task.get("assets", []) 
        task["assets"] = [self._resolve_asset_urls(asset, base_url) for asset in assets] 
        return task 
 
    def build_base_course_sync_payloads(self, base_url: str) -> list[dict[str, Any]]: 
        index = self.get_course_index() 
        payloads: list[dict[str, Any]] = [] 
        for item in index.get("courses", []): 
            course_id = item["course_id"] 
            task_path = self.data_dir / "courses" / course_id / "courseTask.json" 
            if task_path.exists(): 
                payloads.append(self.build_course_task_sync_payload(course_id, base_url)) 
        return payloads 
 
    def build_manifest_assets(self, course_id: str, base_url: str) -> list[dict[str, Any]]: 
        task = self.get_course_task(course_id) 
        return [self._resolve_asset_urls(asset, base_url) for asset in task.get("assets", [])]