0.4 秒赛博替身演武场源码精讲

"""本机模拟只验证代码流程；不能证明真实播音、动作执行或安全。"""

import ast
import contextlib
import importlib
import io
import json
import os
import sqlite3
import sys
import tempfile
import types
import unittest
import zipfile
from pathlib import Path
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[1]
WORKSPACE = ROOT.parent
DEV_KIT = ROOT / "primehub_dev_kit"
sys.path.insert(0, str(DEV_KIT))
extract_token = importlib.import_module("extract_token")
primehub_publish = importlib.import_module("primehub_publish")
SOURCE = (WORKSPACE / "Example/energy_station/src/main.py").read_text(encoding="utf-8")
UNIVERSAL_SOURCE = (WORKSPACE / "Example/universal_skill/src/main.py").read_text(encoding="utf-8")


class Clock(types.ModuleType):
    def __init__(self):
        super().__init__("time")
        self.now = 0.0

    def monotonic(self):
        return self.now

    def sleep(self, seconds):
        self.now += seconds
        if self.now > 600:
            raise AssertionError("模拟超过600秒，疑似无界等待")


class Robot:
    def __init__(self, clock, model, mode="normal", form="quadruped"):
        self.clock = clock
        self.model = model
        self.mode = mode
        self.form = form
        self.started = None
        self.duration = 0.0
        self.last_animation_end = 0.0
        self.lines = []
        self.animations = []
        self.lights = []

    def play_tts(self, text):
        if self.started is not None and self.clock.now < self.started + self.duration:
            raise AssertionError("上一句尚未播完就发出了下一句")
        if self.clock.now < self.last_animation_end:
            raise AssertionError("未等动作保持时长结束就开始下一句")
        self.lines.append(text)
        self.started = self.clock.now + 0.5
        self.duration = max(len(text) * 0.12, 1.0)

    def get_audio_play_state(self):
        if self.mode == "invalid":
            return {"state": None}
        elapsed = self.clock.now - self.started
        playing = 0 <= elapsed < self.duration
        if self.mode == "never_start":
            playing = False
        elif self.mode == "never_finish":
            playing = elapsed >= 0
        elif self.mode == "jitter" and 1.0 < elapsed < 1.3:
            playing = False
        return {"pkg_name": "tts", "state": {"value": 1 if playing else 2}}

    def play_animation(self, animation_id, interrupt):
        if self.clock.now < self.started + self.duration:
            raise AssertionError("台词未播完就触发了动作")
        if self.model == "q1":
            allowed = {3001, 3008, 3007}
        elif self.form == "quadruped":
            allowed = {1, 7, 26, 43}
        else:
            allowed = {26, 43}
        if animation_id not in allowed or interrupt:
            raise AssertionError("动作白名单或不打断约束被破坏")
        self.animations.append(animation_id)
        self.last_animation_end = self.clock.now + (6.5 if self.model == "q1" else 16.0)
        return self.mode != "reject_animation"

    def play_light(self, color, mode):
        self.lights.append((color, mode))

    def get_t1_robot_form(self):
        if self.model != "t1":
            raise AssertionError("Q1 不应调用 T1 专用接口")
        return self.form


def perform(model, mode="normal", form="quadruped", source_code=SOURCE):
    clock = Clock()
    robot = Robot(clock, model, mode, form)
    output = io.StringIO()
    script = source_code.replace('ROBOT_MODEL = "q1"', 'ROBOT_MODEL = "%s"' % model)
    with patch.dict(sys.modules, {"time": clock}), contextlib.redirect_stdout(output):
        exec(compile(script, "skill/src/main.py", "exec"), {"robot": robot})
    return robot, output.getvalue(), clock.now


class SkillTests(unittest.TestCase):
    def test_both_models_and_t1_forms_finish_twice(self):
        for model, form in (("q1", "quadruped"), ("t1", "quadruped"), ("t1", "biped")):
            for run in range(2):
                with self.subTest(model=model, form=form, run=run):
                    robot, output, elapsed = perform(model, form=form)
                    self.assertEqual(len(robot.lines), 5)
                    self.assertEqual(len(robot.animations), 3)
                    self.assertIn("ENERGY_STATION_DONE steps=3", output)
                    self.assertLess(elapsed, 120)

    def test_universal_skill_both_models_and_forms(self):
        for model, form in (("q1", "quadruped"), ("t1", "quadruped"), ("t1", "biped")):
            with self.subTest(model=model, form=form):
                robot, output, elapsed = perform(model, form=form, source_code=UNIVERSAL_SOURCE)
                self.assertEqual(len(robot.lines), 5)
                self.assertEqual(len(robot.animations), 3)
                self.assertIn("UNIVERSAL_SKILL_DONE steps=3", output)
                self.assertLess(elapsed, 120)

    def test_universal_skill_network_fallback_when_offline(self):
        online_script = UNIVERSAL_SOURCE.replace("ENABLE_NET_QUOTE = False", "ENABLE_NET_QUOTE = True")
        with patch("urllib.request.urlopen", side_effect=OSError("network unreachable")):
            robot, output, _ = perform("q1", source_code=online_script)
        self.assertIn("NET_FALLBACK", output)
        self.assertIn("UNIVERSAL_SKILL_DONE steps=3", output)
        self.assertEqual(len(robot.lines), 5)

    def test_short_idle_glitch_does_not_cut_speech(self):
        for model in ("q1", "t1"):
            with self.subTest(model=model):
                robot, output, _ = perform(model, "jitter")
                self.assertEqual(len(robot.lines), 5)
                self.assertIn("ENERGY_STATION_DONE", output)

    def test_audio_failures_stop_instead_of_claiming_success(self):
        for mode, error in (("never_start", "TTS_NOT_STARTED"),
                            ("never_finish", "TTS_NOT_FINISHED"),
                            ("invalid", "AUDIO_STATE_INVALID")):
            with self.subTest(mode=mode), self.assertRaisesRegex(RuntimeError, error):
                perform("q1", mode)

    def test_rejected_animation_stops(self):
        with self.assertRaisesRegex(RuntimeError, "ANIMATION_REJECTED"):
            perform("q1", "reject_animation")

    def test_unknown_t1_form_stops(self):
        with self.assertRaisesRegex(RuntimeError, "T1_FORM_UNKNOWN"):
            perform("t1", form="unknown")

    def test_python_310_and_stdlib_contract(self):
        for name, code, expected_imports in (
            ("energy_station", SOURCE, ["time"]),
            ("universal_skill", UNIVERSAL_SOURCE, ["json", "random", "time", "urllib.request"]),
        ):
            with self.subTest(skill=name):
                tree = ast.parse(code, feature_version=(3, 10))
                imports = sorted(node.names[0].name for node in ast.walk(tree) if isinstance(node, ast.Import))
                self.assertEqual(imports, expected_imports)
                self.assertFalse(any(isinstance(node, ast.ImportFrom) for node in ast.walk(tree)))


class ToolkitTests(unittest.TestCase):
    def test_anonymous_search_does_not_read_token(self):
        with patch.object(sys, "argv", ["primehub_publish.py", "search", "点歌"]), \
                patch.object(primehub_publish, "search", return_value=[]) as search, \
                patch.object(Path, "read_text", side_effect=AssertionError("不应读凭证")):
            primehub_publish.main()
        search.assert_called_once_with("点歌")

    def test_search_api_error_is_not_reported_as_empty_success(self):
        with patch.object(primehub_publish, "call", return_value={"code": 9999, "msg": "synthetic"}):
            with self.assertRaisesRegex(RuntimeError, "9999"):
                primehub_publish.search("点歌")

    def test_my_skills_queries_authenticated_account_endpoint(self):
        response = {"code": 1000, "data": {"records": [{"id": 1}]}}
        with patch.object(primehub_publish, "call", return_value=response) as call:
            self.assertEqual(primehub_publish.my_skills("synthetic-token"), [{"id": 1}])
        call.assert_called_once_with(
            "skillCenter/queryPageByUserCode", {"page": 1, "rows": 50}, "synthetic-token"
        )

    def test_my_skills_rejects_expired_login(self):
        with patch.object(primehub_publish, "call", return_value={"code": 1002, "msg": "synthetic"}):
            with self.assertRaisesRegex(RuntimeError, "1002"):
                primehub_publish.my_skills("synthetic-token")

    def test_shared_token_path(self):
        self.assertEqual(primehub_publish.TOKEN_FILE, extract_token.DEFAULT_TOKEN_FILE)
        self.assertTrue(Path(extract_token.DEFAULT_TOKEN_FILE).is_absolute())

    def test_msix_firefox_profile_discovery(self):
        with tempfile.TemporaryDirectory(prefix="qiyuan_msix_test_") as folder:
            database = (
                Path(folder) / "Packages/Mozilla.MozillaFirefox_synthetic/LocalCache"
                / "Roaming/Mozilla/Firefox/Profiles/default.synthetic/storage/default"
                / "https+++primehub.primebot.com/ls/data.sqlite"
            )
            database.parent.mkdir(parents=True)
            database.touch()
            with patch.dict(os.environ, {
                "LOCALAPPDATA": folder, "APPDATA": folder, "PRIMEHUB_FIREFOX_PROFILE": ""
            }):
                self.assertIn(database, extract_token.firefox_databases())

    def test_publish_accepts_json_argument_and_metadata_file(self):
        with tempfile.TemporaryDirectory(prefix="qiyuan_metadata_test_") as folder:
            metadata_path = Path(folder) / "publish.json"
            metadata_path.write_text('{"skillName":"synthetic"}', encoding="utf-8")
            cases = [
                ["publish", "synthetic.rpk", '{"skillName":"synthetic"}'],
                ["publish", "synthetic.rpk", "--metadata-file", str(metadata_path)],
            ]
            for arguments in cases:
                with self.subTest(arguments=arguments), \
                        patch.object(sys, "argv", ["primehub_publish.py"] + arguments), \
                        patch.object(Path, "read_text", autospec=True) as read_text, \
                        patch.object(primehub_publish, "upload", return_value="synthetic-url") as upload, \
                        patch.object(primehub_publish, "call", return_value={"code": 1000}) as call, \
                        contextlib.redirect_stdout(io.StringIO()):
                    read_text.side_effect = lambda path, **kwargs: (
                        '{"skillName":"synthetic"}' if path == metadata_path else "synthetic-token"
                    )
                    primehub_publish.main()
                    upload.assert_called_once_with("synthetic.rpk", "synthetic-token")
                    call.assert_called_once_with(
                        "skillCenter/publishSkill",
                        {"skillName": "synthetic", "packageUrl": "synthetic-url"},
                        "synthetic-token",
                    )

    def test_sqlite_backup_reads_wal_and_extraction_writes_selected_path(self):
        with tempfile.TemporaryDirectory(prefix="qiyuan_token_test_") as folder:
            profile = Path(folder) / "profile[synthetic]"
            database = profile / "storage/default/https+++primehub.primebot.com/ls/data.sqlite"
            database.parent.mkdir(parents=True)
            synthetic = b'"synthetic.test.token"'
            compressed = bytes([len(synthetic), (len(synthetic) - 1) << 2]) + synthetic
            connection = sqlite3.connect(database)
            try:
                connection.execute("PRAGMA journal_mode=WAL")
                connection.execute("CREATE TABLE data (key TEXT, value BLOB)")
                connection.execute("INSERT INTO data VALUES (?, ?)", ("skillhub_token", compressed))
                connection.commit()
                output_path = Path(folder) / "output/ph_token.txt"
                with patch.dict(os.environ, {"PRIMEHUB_FIREFOX_PROFILE": str(profile), "PRIMEHUB_TOKEN": ""}):
                    self.assertEqual(extract_token.firefox_databases(), [database])
                    self.assertEqual(extract_token.read_storage(database)["skillhub_token"], compressed)
                    with patch.object(extract_token, "firefox_databases", return_value=[database]), \
                            patch.object(sys, "argv", ["extract_token.py", str(output_path)]), \
                            contextlib.redirect_stdout(io.StringIO()) as output:
                        extract_token.main()
                self.assertEqual(output_path.read_text(encoding="utf-8"), "synthetic.test.token")
                self.assertNotIn("synthetic.test.token", output.getvalue())
            finally:
                connection.close()
            database.unlink()  # Windows 下也应已释放读取连接。

    def test_chromium_leveldb_and_direct_token_extraction(self):
        fake_jwt = "eyJhbGciOiJIUzI1NiJ9.eyJleHAiOjE5OTk5OTk5OTl9.signature_part_12345"
        with tempfile.TemporaryDirectory(prefix="qiyuan_chrome_test_") as folder:
            ldb_dir = Path(folder) / "Default/Local Storage/leveldb"
            ldb_dir.mkdir(parents=True)
            # 1. 测试 UTF-8 编码的 LevelDB .log 记录
            record_utf8 = b"_https://primehub.primebot.com\x00\x01skillhub_token\x01\"" + fake_jwt.encode("ascii") + b"\""
            (ldb_dir / "000003.log").write_bytes(record_utf8)
            output_path = Path(folder) / "out_chrome.txt"
            with patch.dict(os.environ, {"PRIMEHUB_CHROME_PROFILE": folder, "PRIMEHUB_TOKEN": ""}), \
                    patch.object(extract_token, "firefox_databases", return_value=[]), \
                    patch.object(sys, "argv", ["extract_token.py", "--browser", "chrome", str(output_path)]), \
                    contextlib.redirect_stdout(io.StringIO()) as out:
                extract_token.main()
            self.assertEqual(output_path.read_text(encoding="utf-8"), fake_jwt)
            self.assertNotIn(fake_jwt, out.getvalue())

            # 2. 测试 UTF-16LE 编码的 LevelDB .ldb 记录
            (ldb_dir / "000003.log").unlink()
            record_utf16 = ("_https://primehub.primebot.com\x00skillhub_token\x00" + fake_jwt).encode("utf-16le")
            (ldb_dir / "000005.ldb").write_bytes(record_utf16)
            output_utf16 = Path(folder) / "out_edge.txt"
            with patch.dict(os.environ, {"PRIMEHUB_CHROME_PROFILE": folder, "PRIMEHUB_TOKEN": ""}), \
                    patch.object(extract_token, "firefox_databases", return_value=[]), \
                    patch.object(sys, "argv", ["extract_token.py", "--browser", "edge", str(output_utf16)]), \
                    contextlib.redirect_stdout(io.StringIO()):
                extract_token.main()
            self.assertEqual(output_utf16.read_text(encoding="utf-8"), fake_jwt)

            # 3. 测试命令行 --token 直接传入
            output_direct = Path(folder) / "out_direct.txt"
            with patch.object(sys, "argv", ["extract_token.py", "--token", f'"{fake_jwt}"', str(output_direct)]), \
                    contextlib.redirect_stdout(io.StringIO()):
                extract_token.main()
            self.assertEqual(output_direct.read_text(encoding="utf-8"), fake_jwt)

    def test_missing_firefox_has_actionable_failure(self):
        with patch.dict(os.environ, {"PRIMEHUB_TOKEN": ""}), \
                patch.object(extract_token, "firefox_databases", return_value=[]), \
                patch.object(extract_token, "extract_from_chromium", return_value=None), \
                patch.object(sys, "argv", ["extract_token.py"]), \
                self.assertRaisesRegex(SystemExit, "Firefox"):
            extract_token.main()


class ArtifactTests(unittest.TestCase):
    def test_packages_and_metadata_match_each_model(self):
        for folder, pkg_prefix, pub_prefix in (
            ("energy_station", "cans518_energy_station", "publish"),
            ("universal_skill", "cans518_universal_skill", "publish_universal"),
        ):
            for model, model_id, model_code in (("q1", 100, "robot_q"), ("t1", 101, "robot_t")):
                with self.subTest(skill=folder, model=model):
                    package = ROOT / "dist" / ("%s_%s.rpk" % (folder, model))
                    self.assertLess(package.stat().st_size, 200 * 1024 * 1024)
                    with zipfile.ZipFile(package) as archive:
                        self.assertIsNone(archive.testzip())
                        self.assertEqual(set(archive.namelist()), {"cfg/pkg.yaml", "src/main.py", "readme.md"})
                        config = archive.read("cfg/pkg.yaml").decode("utf-8")
                        script = archive.read("src/main.py").decode("utf-8")
                    self.assertIn("robot_model: " + model, config)
                    self.assertIn("package_name: %s_%s_20261024" % (pkg_prefix, model), config)
                    self.assertIn('ROBOT_MODEL = "%s"' % model, script)
                    ast.parse(script, feature_version=(3, 10))
                    metadata = json.loads((ROOT / "dist" / ("%s_%s.json" % (pub_prefix, model))).read_text(encoding="utf-8"))
                    self.assertEqual(metadata["robotModelIds"], [model_id])
                    self.assertEqual(metadata["robotModelCodeList"], [model_code])
                    self.assertEqual(metadata["version"], "0.1.0")


if __name__ == "__main__":
    unittest.main(verbosity=2)