from __future__ import annotations
|
|
import argparse
|
import json
|
import sys
|
from pathlib import Path
|
|
from .v1_bridge import V1Bridge, V1ContractError
|
from .telemetry import RunClock
|
from .workflow import _input_summary, run_ticker_pipeline
|
|
|
class CliInputError(ValueError):
|
pass
|
|
|
class JsonArgumentParser(argparse.ArgumentParser):
|
def error(self, message: str) -> None:
|
raise CliInputError(message)
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
parser = JsonArgumentParser(description="股票估值 V2 端到端协调层")
|
mode = parser.add_mutually_exclusive_group(required=True)
|
mode.add_argument("--input", type=Path, help="兼容 V1 标准估值快照")
|
mode.add_argument("--ticker", help="股票代码，例如 001270.SZ")
|
parser.add_argument("--as-of")
|
parser.add_argument("--output-dir", required=True, type=Path)
|
parser.add_argument("--cache-dir", type=Path)
|
parser.add_argument("--judgment", type=Path)
|
parser.add_argument("--fixture-dir", type=Path)
|
parser.add_argument("--task-start")
|
parser.add_argument("--force", action="store_true")
|
return parser
|
|
|
def _v1_mode(args: argparse.Namespace) -> int:
|
try:
|
bridge = V1Bridge.load()
|
status = bridge.run_input(args.input, args.output_dir, args.force)
|
except V1ContractError as exc:
|
print(
|
json.dumps(
|
{"status": "RUNTIME_ERROR", "error_code": "E_V1_CONTRACT", "error": str(exc)},
|
ensure_ascii=False,
|
indent=2,
|
),
|
file=sys.stderr,
|
)
|
return 3
|
except Exception as exc:
|
try:
|
bridge # type: ignore[used-before-assignment]
|
input_error = bridge.module.InputError # type: ignore[possibly-undefined]
|
except Exception:
|
input_error = ()
|
if input_error and isinstance(exc, input_error):
|
print(
|
json.dumps(
|
{"status": "INPUT_ERROR", "errors": exc.errors},
|
ensure_ascii=False,
|
indent=2,
|
),
|
file=sys.stderr,
|
)
|
return 2
|
if isinstance(exc, (OSError, json.JSONDecodeError)):
|
print(
|
json.dumps({"status": "RUNTIME_ERROR", "error": str(exc)}, ensure_ascii=False, indent=2),
|
file=sys.stderr,
|
)
|
return 3
|
raise
|
print(json.dumps(status, ensure_ascii=False, indent=2))
|
return 0
|
|
|
def main(argv: list[str] | None = None) -> int:
|
parser = build_parser()
|
try:
|
args = parser.parse_args(argv)
|
except CliInputError as exc:
|
print(json.dumps(_input_summary(RunClock(), exc, False), ensure_ascii=False, sort_keys=True, separators=(",", ":")))
|
return 2
|
if args.input is not None:
|
if any((args.as_of, args.cache_dir, args.judgment, args.fixture_dir, args.task_start)):
|
summary = _input_summary(RunClock(), CliInputError("--input 模式不接受 ticker 模式参数"), False)
|
print(json.dumps(summary, ensure_ascii=False, sort_keys=True, separators=(",", ":")))
|
return 2
|
return _v1_mode(args)
|
if not args.as_of or args.cache_dir is None:
|
summary = _input_summary(RunClock(), CliInputError("--ticker 模式必须提供 --as-of 和 --cache-dir"), False)
|
print(json.dumps(summary, ensure_ascii=False, sort_keys=True, separators=(",", ":")))
|
return 2
|
code, summary = run_ticker_pipeline(
|
ticker=args.ticker,
|
as_of=args.as_of,
|
output_dir=args.output_dir,
|
cache_dir=args.cache_dir,
|
judgment_path=args.judgment,
|
fixture_dir=args.fixture_dir,
|
task_start=args.task_start,
|
force=args.force,
|
)
|
print(json.dumps(summary, ensure_ascii=False, sort_keys=True, separators=(",", ":")))
|
return code
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
