crypto_monitor/lib/options/options_review_images_lib.py

"""期权复盘截图:独立命名空间,不与合约 journal 混用."""
from __future__ import annotations

import json
import os
import re
from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence

OPTIONS_REVIEW_UPLOAD_TFS: tuple[str, ...] = ("chart", "entry", "exit", "other")
OPTIONS_REVIEW_ALLOWED_EXT = frozenset({".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp"})
_DRAFT_ID_RE = re.compile(r"^[a-f0-9]{32}$")
_SLOT_FILE_RE = re.compile(
    r"^options_journal_([a-f0-9]{32})_(chart|entry|exit|other)\.(png|jpg|jpeg|webp|gif|bmp)$",
    re.I,
)


def normalize_options_review_draft_id(raw: Any) -> Optional[str]:
    s = str(raw or "").strip().lower()
    if _DRAFT_ID_RE.match(s):
        return s
    return None


def _safe_ext(filename: str) -> str:
    ext = os.path.splitext(str(filename or ""))[1].lower()
    return ext if ext in OPTIONS_REVIEW_ALLOWED_EXT else ".png"


def options_review_upload_dir(base_upload_folder: str) -> str:
    """独立子目录 static/images/options_journal."""
    base = os.path.abspath(base_upload_folder or "")
    path = os.path.join(base, "options_journal")
    os.makedirs(path, exist_ok=True)
    return path


def build_options_review_slot_filename(
    draft_id: str,
    tf: str,
    ext: str,
    *,
    secure_filename_fn: Callable[[str], str],
) -> str:
    ext = ext if ext.startswith(".") else f".{ext}"
    ext = _safe_ext(f"x{ext}")
    fname = secure_filename_fn(f"options_journal_{draft_id}_{tf}{ext}")
    return fname or ""


def is_valid_options_review_file(filename: str, draft_id: str, tf: str) -> bool:
    fn = os.path.basename(str(filename or "").strip())
    if not fn or fn != str(filename or "").strip():
        return False
    m = _SLOT_FILE_RE.match(fn)
    if not m:
        return False
    return m.group(1) == draft_id.lower() and m.group(2) == tf


def save_options_review_slot_file(
    file,
    draft_id: str,
    tf: str,
    upload_folder: str,
    *,
    secure_filename_fn: Callable[[str], str],
) -> Optional[Dict[str, str]]:
    if tf not in OPTIONS_REVIEW_UPLOAD_TFS or not draft_id or not upload_folder:
        return None
    if not file or not getattr(file, "filename", None):
        return None
    ext = _safe_ext(file.filename)
    fname = build_options_review_slot_filename(
        draft_id, tf, ext, secure_filename_fn=secure_filename_fn
    )
    if not fname:
        return None
    os.makedirs(upload_folder, exist_ok=True)
    path = os.path.join(upload_folder, fname)
    file.save(path)
    return {"tf": tf, "file": fname}


def parse_options_review_images_json(raw: Any) -> List[Dict[str, str]]:
    if not raw:
        return []
    if isinstance(raw, list):
        data = raw
    else:
        try:
            data = json.loads(str(raw))
        except (TypeError, ValueError, json.JSONDecodeError):
            return []
    if not isinstance(data, list):
        return []
    out: List[Dict[str, str]] = []
    for item in data:
        if not isinstance(item, dict):
            continue
        tf = str(item.get("tf") or "").strip()
        file = str(item.get("file") or "").strip()
        if file:
            out.append({"tf": tf, "file": file})
    return out


def images_json_dumps(items: Sequence[Mapping[str, str]]) -> Optional[str]:
    if not items:
        return None
    return json.dumps(list(items), ensure_ascii=False, separators=(",", ":"))


def options_review_image_paths(row: Any, upload_folder: str) -> List[str]:
    upload_folder = os.path.abspath(upload_folder or "")
    paths: List[str] = []
    seen: set[str] = set()

    def _add(name: Optional[str]) -> None:
        if not name:
            return
        p = os.path.abspath(os.path.join(upload_folder, str(name).strip()))
        if os.path.isfile(p) and p not in seen:
            seen.add(p)
            paths.append(p)

    try:
        keys = row.keys() if hasattr(row, "keys") else ()
    except Exception:
        keys = ()
    images = parse_options_review_images_json(
        row["images_json"] if "images_json" in keys else getattr(row, "images_json", None)
    )
    for item in images:
        _add(item.get("file"))
    if "image" in keys or hasattr(row, "image"):
        _add(row["image"] if "image" in keys else getattr(row, "image", None))
    return paths