xiaolipro
7c29005cb6
All checks were successful
Build and Deploy Front / deploy (push) Successful in 7m17s
253 lines
8.2 KiB
Python
253 lines
8.2 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import yaml
|
||
|
||
from xhs_bridge.human_behavior import (
|
||
HumanBehaviorConfig,
|
||
parse_human_behavior_config,
|
||
write_human_behavior_json,
|
||
)
|
||
from xhs_bridge.human_guard import HumanGuardConfig, parse_human_guard_config
|
||
|
||
def application_root() -> Path:
|
||
"""OMS Agent 桌面端：engine/ 目录；PyInstaller 打包后为运行时根目录。"""
|
||
if getattr(sys, "frozen", False):
|
||
from xhs_bridge.pack_runtime import get_runtime_root
|
||
|
||
return get_runtime_root()
|
||
return Path(__file__).resolve().parent.parent
|
||
|
||
|
||
def _is_frozen() -> bool:
|
||
return bool(getattr(sys, "frozen", False))
|
||
|
||
|
||
PROJECT_ROOT = application_root()
|
||
DEFAULT_CONFIG_PATH = PROJECT_ROOT / "config.yaml"
|
||
DEFAULT_ENGINE_DIR = PROJECT_ROOT
|
||
|
||
|
||
@dataclass
|
||
class SkillsConfig:
|
||
"""内嵌 xiaohongshu-skills 引擎配置。"""
|
||
|
||
project_dir: Path = field(default_factory=lambda: DEFAULT_ENGINE_DIR)
|
||
python: str = "" # 空则自动探测项目根目录 .venv 或当前解释器
|
||
bridge_port: int = 9333
|
||
startup_timeout_sec: int = 45
|
||
|
||
@property
|
||
def bridge_ws_url(self) -> str:
|
||
return f"ws://127.0.0.1:{self.bridge_port}"
|
||
|
||
def python_path(self) -> Path:
|
||
if self.python:
|
||
return Path(os.path.expanduser(self.python)).resolve()
|
||
for folder in ("python", ".venv"):
|
||
candidate = self.project_dir / folder / "Scripts" / "python.exe"
|
||
if candidate.is_file():
|
||
return candidate
|
||
if not _is_frozen():
|
||
return Path(sys.executable).resolve()
|
||
raise FileNotFoundError("打包版应使用内置 skills_cli / XHSBridge，无需 Python。")
|
||
|
||
def bridge_launcher(self, *, port: int) -> list[str]:
|
||
if _is_frozen():
|
||
from xhs_bridge.pack_runtime import bundled_engine_dir
|
||
|
||
exe = bundled_engine_dir() / "XHSBridge.exe"
|
||
if exe.is_file():
|
||
return [str(exe), "--port", str(port)]
|
||
else:
|
||
bundled = self.project_dir / "engine" / "XHSBridge.exe"
|
||
if bundled.is_file():
|
||
return [str(bundled), "--port", str(port)]
|
||
return [str(self.python_path()), str(self.bridge_server_script()), "--port", str(port)]
|
||
|
||
def cli_launcher(self, argv: list[str]) -> list[str]:
|
||
if _is_frozen():
|
||
from xhs_bridge.pack_runtime import bundled_engine_dir
|
||
|
||
exe = bundled_engine_dir() / "skills_cli.exe"
|
||
if exe.is_file():
|
||
return [str(exe), *argv]
|
||
else:
|
||
bundled = self.project_dir / "engine" / "skills_cli.exe"
|
||
if bundled.is_file():
|
||
return [str(bundled), *argv]
|
||
return [str(self.python_path()), str(self.cli_script()), *argv]
|
||
|
||
def scripts_dir(self) -> Path:
|
||
return self.project_dir / "scripts"
|
||
|
||
def cli_script(self) -> Path:
|
||
return self.scripts_dir() / "cli.py"
|
||
|
||
def bridge_server_script(self) -> Path:
|
||
return self.scripts_dir() / "bridge_server.py"
|
||
|
||
def extension_dir(self) -> Path:
|
||
return self.project_dir / "extension"
|
||
|
||
|
||
@dataclass
|
||
class LicenseConfig:
|
||
provider: str = "mock"
|
||
api_base_url: str = "https://api.example.com"
|
||
cache_path: str = "~/.xhs_bridge/license.json"
|
||
|
||
def resolved_cache_path(self) -> Path:
|
||
return Path(os.path.expanduser(self.cache_path))
|
||
|
||
|
||
@dataclass
|
||
class OmsConfig:
|
||
"""OMS 物流查价 API（POST /order/price-check）。"""
|
||
|
||
api_base_url: str = "https://api.test.ideship.com/admin"
|
||
timeout_sec: int = 30
|
||
auth_token: str = ""
|
||
contact_name: str = ""
|
||
contact_phone: str = ""
|
||
wechat_qrcode_path: str = ""
|
||
|
||
|
||
@dataclass
|
||
class AppConfig:
|
||
bin_dir: Path = PROJECT_ROOT
|
||
skills: SkillsConfig = field(default_factory=SkillsConfig)
|
||
license: LicenseConfig = field(default_factory=LicenseConfig)
|
||
oms: OmsConfig = field(default_factory=OmsConfig)
|
||
human_guard: HumanGuardConfig = field(default_factory=HumanGuardConfig)
|
||
human_behavior: HumanBehaviorConfig = field(default_factory=HumanBehaviorConfig)
|
||
state_dir: Path = field(default_factory=lambda: PROJECT_ROOT / ".xhs_bridge")
|
||
|
||
def human_guard_state_path(self) -> Path:
|
||
return self.state_dir / "human_guard_state.json"
|
||
|
||
def bridge_pid_file(self) -> Path:
|
||
return self.state_dir / "bridge.pid"
|
||
|
||
|
||
def _expand_path(value: str | Path, *, base: Path | None = None) -> Path:
|
||
raw = Path(os.path.expanduser(str(value)))
|
||
if raw.is_absolute():
|
||
return raw.resolve()
|
||
anchor = (base or Path.cwd()).resolve()
|
||
return (anchor / raw).resolve()
|
||
|
||
|
||
def load_config(path: Path | None = None) -> AppConfig:
|
||
cfg_path = path or DEFAULT_CONFIG_PATH
|
||
data: dict[str, Any] = {}
|
||
if cfg_path.is_file():
|
||
with cfg_path.open("r", encoding="utf-8") as f:
|
||
data = yaml.safe_load(f) or {}
|
||
|
||
root = application_root()
|
||
bin_dir = _expand_path(data.get("bin_dir", root), base=root)
|
||
skills_raw = data.get("skills") or {}
|
||
license_raw = data.get("license") or {}
|
||
oms_raw = data.get("oms") or {}
|
||
human_guard = parse_human_guard_config(data.get("human_guard"))
|
||
human_behavior = parse_human_behavior_config(data.get("human_behavior"))
|
||
|
||
if _is_frozen():
|
||
project_dir = root.resolve()
|
||
else:
|
||
project_dir = _expand_path(skills_raw.get("project_dir", root), base=root)
|
||
skills = SkillsConfig(
|
||
project_dir=project_dir,
|
||
python=str(skills_raw.get("python") or ""),
|
||
bridge_port=int(skills_raw.get("bridge_port", 9333)),
|
||
startup_timeout_sec=int(skills_raw.get("startup_timeout_sec", 45)),
|
||
)
|
||
|
||
license_cfg = LicenseConfig(
|
||
provider=license_raw.get("provider", "mock"),
|
||
api_base_url=license_raw.get("api_base_url", LicenseConfig.api_base_url),
|
||
cache_path=license_raw.get("cache_path", LicenseConfig.cache_path),
|
||
)
|
||
|
||
oms_cfg = OmsConfig(
|
||
api_base_url=str(oms_raw.get("api_base_url", OmsConfig.api_base_url)),
|
||
timeout_sec=int(oms_raw.get("timeout_sec", OmsConfig.timeout_sec)),
|
||
auth_token=str(oms_raw.get("auth_token", "") or ""),
|
||
contact_name=str(oms_raw.get("contact_name", "") or ""),
|
||
contact_phone=str(oms_raw.get("contact_phone", "") or ""),
|
||
wechat_qrcode_path=str(oms_raw.get("wechat_qrcode_path", "") or ""),
|
||
)
|
||
|
||
state_dir = bin_dir / ".xhs_bridge"
|
||
state_dir.mkdir(parents=True, exist_ok=True)
|
||
write_human_behavior_json(human_behavior, state_dir)
|
||
|
||
return AppConfig(
|
||
bin_dir=bin_dir,
|
||
skills=skills,
|
||
license=license_cfg,
|
||
oms=oms_cfg,
|
||
human_guard=human_guard,
|
||
human_behavior=human_behavior,
|
||
state_dir=state_dir,
|
||
)
|
||
|
||
|
||
def config_file_path(path: Path | None = None) -> Path:
|
||
return path or DEFAULT_CONFIG_PATH
|
||
|
||
|
||
def load_config_yaml(path: Path | None = None) -> dict[str, Any]:
|
||
cfg_path = config_file_path(path)
|
||
if not cfg_path.is_file():
|
||
return {}
|
||
with cfg_path.open("r", encoding="utf-8") as f:
|
||
return yaml.safe_load(f) or {}
|
||
|
||
|
||
def save_human_guard_config(cfg: HumanGuardConfig, path: Path | None = None) -> Path:
|
||
from xhs_bridge.human_guard import human_guard_config_to_dict
|
||
|
||
cfg_path = config_file_path(path)
|
||
data = load_config_yaml(cfg_path)
|
||
data["human_guard"] = human_guard_config_to_dict(cfg)
|
||
with cfg_path.open("w", encoding="utf-8") as f:
|
||
yaml.dump(
|
||
data,
|
||
f,
|
||
allow_unicode=True,
|
||
default_flow_style=False,
|
||
sort_keys=False,
|
||
)
|
||
return cfg_path
|
||
|
||
|
||
def save_human_behavior_config(
|
||
cfg: HumanBehaviorConfig,
|
||
state_dir: Path,
|
||
path: Path | None = None,
|
||
) -> Path:
|
||
from xhs_bridge.human_behavior import human_behavior_config_to_dict
|
||
|
||
cfg_path = config_file_path(path)
|
||
data = load_config_yaml(cfg_path)
|
||
if "human_guard" in data:
|
||
del data["human_guard"]
|
||
data["human_behavior"] = human_behavior_config_to_dict(cfg)
|
||
with cfg_path.open("w", encoding="utf-8") as f:
|
||
yaml.dump(
|
||
data,
|
||
f,
|
||
allow_unicode=True,
|
||
default_flow_style=False,
|
||
sort_keys=False,
|
||
)
|
||
write_human_behavior_json(cfg, state_dir)
|
||
return cfg_path
|