#!/usr/bin/env python3 """One-shot: move root shared modules into lib/ and rewrite imports.""" from __future__ import annotations import re import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent PACKAGE_FILES: dict[str, list[str]] = { "strategy": [ "strategy_config.py", "strategy_db.py", "strategy_exchange_base.py", "strategy_exchange_binance.py", "strategy_exchange_gate.py", "strategy_exchange_okx.py", "strategy_records_register.py", "strategy_register.py", "strategy_roll_lib.py", "strategy_roll_monitor_lib.py", "strategy_roll_ui_lib.py", "strategy_snapshot_lib.py", "strategy_trade_labels.py", "strategy_trend_exchange.py", "strategy_trend_lib.py", "strategy_trend_register.py", "strategy_ui.py", "strategy_wechat_notify.py", ], "key_monitor": [ "key_monitor_full_margin_lib.py", "key_monitor_lib.py", "key_monitor_schema_lib.py", "key_sl_tp_lib.py", "fib_key_monitor_lib.py", "false_breakout_key_monitor_lib.py", "trigger_entry_key_monitor_lib.py", ], "trade": [ "trade_result_lib.py", "trade_exchange_stats_lib.py", "trade_stats_calendar_lib.py", "order_monitor_display_lib.py", "position_sizing_lib.py", "account_risk_lib.py", "manual_sltp_lib.py", "time_close_lib.py", "daily_open_limit_lib.py", ], "hub": [ "hub_auth.py", "hub_bridge.py", "hub_calculator_lib.py", "hub_calculator_market_lib.py", "hub_entry_plan_lib.py", "hub_fund_history_lib.py", "hub_host_status_lib.py", "hub_kline_store.py", "hub_macro_calendar_lib.py", "hub_market_info_lib.py", "hub_ohlcv_lib.py", "hub_position_metrics.py", "hub_sso.py", "hub_symbol_archive_lib.py", "hub_trades_lib.py", "hub_volume_rank_lib.py", ], "ai": [ "ai_client.py", "ai_review_lib.py", ], "instance": [ "instance_embed_context_lib.py", "instance_embed_lib.py", "instance_nav_lib.py", "focus_chart_lib.py", "journal_chart_lib.py", ], "exchange": [ "gate_transfer_lib.py", "gate_position_history_lib.py", "okx_orders_lib.py", ], "common": [ "form_submit_lib.py", "history_window_lib.py", "wechat_notify_lib.py", "auto_transfer_daily_lib.py", ], } DIR_MOVES: list[tuple[str, str]] = [ ("strategy_templates", "lib/strategy/templates"), ("embed_templates", "lib/instance/templates"), ("static", "lib/common/static"), ] MODULE_TO_LIB: dict[str, str] = {} for pkg, files in PACKAGE_FILES.items(): for fname in files: MODULE_TO_LIB[fname[:-3]] = f"lib.{pkg}.{fname[:-3]}" IMPORT_FROM_RE = re.compile( r"^(\s*)from\s+(" + "|".join(re.escape(m) for m in sorted(MODULE_TO_LIB, key=len, reverse=True)) + r")\s+import\s+", re.MULTILINE, ) IMPORT_BARE_RE = re.compile( r"^(\s*)import\s+(" + "|".join(re.escape(m) for m in sorted(MODULE_TO_LIB, key=len, reverse=True)) + r")(\s|$)", re.MULTILINE, ) def git_mv(src: Path, dst: Path) -> None: dst.parent.mkdir(parents=True, exist_ok=True) if not src.exists(): if dst.exists(): return raise FileNotFoundError(src) subprocess.run(["git", "mv", str(src), str(dst)], cwd=ROOT, check=True) def move_files() -> None: (ROOT / "lib").mkdir(exist_ok=True) for pkg in PACKAGE_FILES: (ROOT / "lib" / pkg).mkdir(parents=True, exist_ok=True) init = ROOT / "lib" / pkg / "__init__.py" if not init.exists(): init.write_text('"""Shared library package."""\n', encoding="utf-8") lib_init = ROOT / "lib" / "__init__.py" if not lib_init.exists(): lib_init.write_text('"""crypto_monitor shared libraries."""\n', encoding="utf-8") paths_py = ROOT / "lib" / "paths.py" if not paths_py.exists(): paths_py.write_text( '''"""Repository path helpers for lib/ assets.""" from __future__ import annotations import os from pathlib import Path LIB_DIR = Path(__file__).resolve().parent REPO_ROOT = LIB_DIR.parent def strategy_templates_dir(repo_root: str | Path | None = None) -> str: root = Path(repo_root) if repo_root is not None else REPO_ROOT return str(root / "lib" / "strategy" / "templates") def embed_templates_dir(repo_root: str | Path | None = None) -> str: root = Path(repo_root) if repo_root is not None else REPO_ROOT return str(root / "lib" / "instance" / "templates") def common_static_dir(repo_root: str | Path | None = None) -> str: root = Path(repo_root) if repo_root is not None else REPO_ROOT return str(root / "lib" / "common" / "static") ''', encoding="utf-8", ) for pkg, files in PACKAGE_FILES.items(): for fname in files: git_mv(ROOT / fname, ROOT / "lib" / pkg / fname) for src_rel, dst_rel in DIR_MOVES: git_mv(ROOT / src_rel, ROOT / dst_rel) def rewrite_imports_in_text(text: str) -> str: def from_repl(m: re.Match) -> str: mod = m.group(2) return f"{m.group(1)}from {MODULE_TO_LIB[mod]} import " def bare_repl(m: re.Match) -> str: mod = m.group(2) return f"{m.group(1)}import {MODULE_TO_LIB[mod]}{m.group(3)}" text = IMPORT_FROM_RE.sub(from_repl, text) text = IMPORT_BARE_RE.sub(bare_repl, text) return text def patch_path_literals(text: str) -> str: replacements = [ ('os.path.join(repo_root, "strategy_templates")', 'strategy_templates_dir(repo_root)'), ('os.path.join(repo_root, "embed_templates")', 'embed_templates_dir(repo_root)'), ('os.path.join(os.path.dirname(BASE_DIR), "static")', 'common_static_dir(os.path.dirname(BASE_DIR))'), ('_REPO_ROOT / "static"', '_REPO_ROOT / "lib" / "common" / "static"'), ('ROOT / "strategy_templates"', 'ROOT / "lib" / "strategy" / "templates"'), ('ROOT / "embed_templates"', 'ROOT / "lib" / "instance" / "templates"'), ('ROOT / "static"', 'ROOT / "lib" / "common" / "static"'), ] for old, new in replacements: text = text.replace(old, new) return text def ensure_paths_import(text: str, filepath: Path) -> str: needs = [] if "strategy_templates_dir(" in text and "from lib.paths import" not in text: needs.append("strategy_templates_dir") if "embed_templates_dir(" in text and "from lib.paths import" not in text: needs.append("embed_templates_dir") if "common_static_dir(" in text and "from lib.paths import" not in text: needs.append("common_static_dir") if not needs: return text imp = f"from lib.paths import {', '.join(sorted(set(needs)))}\n" if text.startswith('"""') or text.startswith("'''"): end = text.find('"""', 3) if text.startswith('"""') else text.find("'''", 3) if end != -1: end += 3 return text[:end] + "\n\n" + imp + text[end + 1 :] if text.startswith("from __future__"): lines = text.splitlines(keepends=True) i = 0 while i < len(lines) and ( lines[i].startswith("from __future__") or lines[i].strip() == "" ): i += 1 return "".join(lines[:i]) + imp + "".join(lines[i:]) return imp + text def rewrite_all_py_files() -> None: skip = {ROOT / "scripts" / "migrate_to_lib.py"} for path in ROOT.rglob("*.py"): if path in skip or ".venv" in path.parts or "__pycache__" in path.parts: continue original = path.read_text(encoding="utf-8") updated = rewrite_imports_in_text(original) updated = patch_path_literals(updated) updated = ensure_paths_import(updated, path) if updated != original: path.write_text(updated, encoding="utf-8") def main() -> int: move_files() rewrite_all_py_files() print("Migration complete.") return 0 if __name__ == "__main__": sys.exit(main())