strepsiades
139 lines
4.2 KiB
Python
139 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from sync_state_machine.config import (
|
|
JsonlDataSourceConfig,
|
|
DataSourceConfig,
|
|
PersistConfig,
|
|
LoggingConfig,
|
|
build_config,
|
|
list_preset_names,
|
|
load_named_preset_overrides,
|
|
load_overrides_from_file,
|
|
)
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
DEFAULT_NODE_TYPES: List[str] = [
|
|
"company",
|
|
"supplier",
|
|
"user",
|
|
"project",
|
|
"contract",
|
|
"contract_change",
|
|
"construction_task",
|
|
"construction_task_detail",
|
|
"material",
|
|
"material_detail",
|
|
"contract_settlement",
|
|
"contract_settlement_detail",
|
|
"personnel",
|
|
"personnel_detail",
|
|
"preparation",
|
|
"production",
|
|
"lar",
|
|
"units",
|
|
]
|
|
|
|
|
|
class PipelineUiConfig(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", validate_assignment=True)
|
|
|
|
mode: Literal["jsonl_to_jsonl", "jsonl_to_api"] = "jsonl_to_jsonl"
|
|
|
|
node_types: List[str] = Field(default_factory=lambda: list(DEFAULT_NODE_TYPES))
|
|
|
|
local_datasource: DataSourceConfig = Field(
|
|
default_factory=lambda: JsonlDataSourceConfig(
|
|
type="jsonl",
|
|
jsonl_dir=str(PROJECT_ROOT / "filtered_datasource" / "datasource" / "ecm-2025-12-02"),
|
|
)
|
|
)
|
|
remote_datasource: DataSourceConfig = Field(
|
|
default_factory=lambda: JsonlDataSourceConfig(
|
|
type="jsonl",
|
|
jsonl_dir=str(PROJECT_ROOT / "remote_datasource" / "datasource"),
|
|
)
|
|
)
|
|
|
|
persist: PersistConfig = Field(
|
|
default_factory=lambda: PersistConfig(
|
|
db_path=str(PROJECT_ROOT / "test_results" / "sync_state_machine" / "ui_pipeline.db"),
|
|
enable=True,
|
|
wipe_on_start=False,
|
|
)
|
|
)
|
|
logging: LoggingConfig = Field(default_factory=LoggingConfig)
|
|
|
|
strategies: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
|
|
|
|
|
|
def default_ui_config() -> PipelineUiConfig:
|
|
return preset_ui_config("simple_sm")
|
|
|
|
|
|
def list_ui_presets() -> List[str]:
|
|
return list_preset_names()
|
|
|
|
|
|
def preset_ui_config(preset_name: str) -> PipelineUiConfig:
|
|
overrides, _, _ = load_named_preset_overrides(PROJECT_ROOT, preset_name)
|
|
run_cfg = build_config(project_root=PROJECT_ROOT, overrides=overrides)
|
|
mode: Literal["jsonl_to_jsonl", "jsonl_to_api"] = (
|
|
"jsonl_to_api" if getattr(run_cfg.remote_datasource, "type", "jsonl") == "api" else "jsonl_to_jsonl"
|
|
)
|
|
|
|
strategies: Dict[str, Dict[str, Any]] = {}
|
|
for runtime in run_cfg.strategies:
|
|
item: Dict[str, Any] = {
|
|
"config": runtime.config.model_dump(),
|
|
}
|
|
strategies[runtime.node_type] = item
|
|
|
|
return PipelineUiConfig(
|
|
mode=mode,
|
|
node_types=list(run_cfg.node_types),
|
|
local_datasource=run_cfg.local_datasource.model_copy(deep=True),
|
|
remote_datasource=run_cfg.remote_datasource.model_copy(deep=True),
|
|
persist=run_cfg.persist.model_copy(deep=True),
|
|
logging=run_cfg.logging.model_copy(deep=True),
|
|
strategies=strategies,
|
|
)
|
|
|
|
|
|
def file_ui_config(config_file: str) -> PipelineUiConfig:
|
|
path = Path(config_file)
|
|
if not path.is_absolute():
|
|
path = PROJECT_ROOT / path
|
|
path = path.resolve()
|
|
|
|
if not path.exists() or not path.is_file():
|
|
raise FileNotFoundError(f"Config file not found: {path}")
|
|
|
|
overrides = load_overrides_from_file(path, PROJECT_ROOT)
|
|
run_cfg = build_config(project_root=PROJECT_ROOT, overrides=overrides)
|
|
mode: Literal["jsonl_to_jsonl", "jsonl_to_api"] = (
|
|
"jsonl_to_api" if getattr(run_cfg.remote_datasource, "type", "jsonl") == "api" else "jsonl_to_jsonl"
|
|
)
|
|
|
|
strategies: Dict[str, Dict[str, Any]] = {}
|
|
for runtime in run_cfg.strategies:
|
|
item: Dict[str, Any] = {
|
|
"config": runtime.config.model_dump(),
|
|
}
|
|
strategies[runtime.node_type] = item
|
|
|
|
return PipelineUiConfig(
|
|
mode=mode,
|
|
node_types=list(run_cfg.node_types),
|
|
local_datasource=run_cfg.local_datasource.model_copy(deep=True),
|
|
remote_datasource=run_cfg.remote_datasource.model_copy(deep=True),
|
|
persist=run_cfg.persist.model_copy(deep=True),
|
|
logging=run_cfg.logging.model_copy(deep=True),
|
|
strategies=strategies,
|
|
)
|