crypto_monitor/scripts/migrate_to_lib.py

#!/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())