app/multimodal.py

from __future__ import annotations

from pathlib import Path

from app.config import RAW_DIR


def latest_file(paths: list[Path]) -> Path | None:
    if not paths:
        return None
    return max(paths, key=lambda path: path.stat().st_mtime)


def latest_chart_image(ticker: str | None) -> Path | None:
    if not ticker:
        return None
    image_dir = RAW_DIR / "images" / ticker.upper()
    if not image_dir.exists():
        return None
    candidates = [
        path
        for path in image_dir.glob("*")
        if path.is_file()
        and path.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}
        and ("chart" in path.stem.lower() or "tradingview" in path.stem.lower())
    ]
    return latest_file(candidates)


def ticker_tables(ticker: str | None, limit: int = 5) -> list[Path]:
    if not ticker:
        return []
    csv_dir = RAW_DIR / "csv" / ticker.upper()
    if not csv_dir.exists():
        return []
    candidates = [
        path
        for path in csv_dir.glob("*.csv")
        if path.is_file() and has_meaningful_csv_content(path)
    ]
    return sorted(candidates, key=lambda path: path.stat().st_mtime, reverse=True)[:limit]


def has_meaningful_csv_content(path: Path) -> bool:
    try:
        rows = path.read_text(encoding="utf-8-sig", errors="ignore").splitlines()
    except OSError:
        return False
    non_empty_rows = [row for row in rows if row.strip()]
    if len(non_empty_rows) < 2:
        return False
    return any(row.count(",") >= 2 for row in non_empty_rows[:3])


def artifact_label(path: Path) -> str:
    stem = path.stem.lower()
    if "stock_overview_timeseries" in stem:
        return "Bảng giá và giao dịch"
    if "financial_document" in stem or "financial_documents" in stem:
        return "Báo cáo tài chính"
    if "analysis_report" in stem:
        return "Báo cáo phân tích"
    if "ticker_news" in stem or "news_events" in stem:
        return "Tin tức và sự kiện"
    if "chart" in stem or "tradingview" in stem:
        return "Biểu đồ TradingView"
    return path.name


def ticker_pdfs(ticker: str | None, limit: int = 5) -> list[Path]:
    if not ticker:
        return []
    pdf_dir = RAW_DIR / "pdf" / ticker.upper()
    if not pdf_dir.exists():
        return []
    candidates = [path for path in pdf_dir.glob("*.pdf") if path.is_file()]
    return sorted(candidates, key=lambda path: path.stat().st_mtime, reverse=True)[:limit]


def multimodal_artifacts(ticker: str | None) -> dict[str, object]:
    return {
        "chart": latest_chart_image(ticker),
        "tables": ticker_tables(ticker),
        "pdfs": ticker_pdfs(ticker),
    }