hello-agents/Co-creation-projects/lcyting-StockSage-agent/backend/app/services/stock_file_cache.py

"""
股票数据文件缓存服务 — 每只股票的所有数据存储到本地文件

每次获取数据时优先从本地文件读取，未命中或过期才调用接口。
支持 grep 风格的文件内容搜索。
"""

from __future__ import annotations

import json
import logging
import os
import subprocess
import threading
from datetime import date, datetime
from pathlib import Path
from typing import Optional, Any

from app.config import settings

logger = logging.getLogger(__name__)

_cache_lock = threading.Lock()

# 缓存根目录
_STOCK_CACHE_ROOT = settings.DATA_DIR / "stock_cache"

# 各数据类型的文件名
_DATA_TYPE_FILES = {
    "quote": "quote.json",
    "financial": "financial.json",
    "profile": "profile.json",
    "holders": "holders.json",
    "sentiment": "sentiment.json",
    "news": "news.json",
}

# 当日缓存有效期（同一只股票同一天只调一次接口）
_TODAY = date.today().isoformat()


class StockFileCache:
    """股票数据文件缓存"""

    def __init__(self):
        _STOCK_CACHE_ROOT.mkdir(parents=True, exist_ok=True)
        self._index_file = _STOCK_CACHE_ROOT / "_index.json"
        self._index: dict = {}
        self._load_index()

    # ---- 索引管理 ----

    def _load_index(self):
        """加载主索引"""
        try:
            if self._index_file.exists():
                self._index = json.loads(self._index_file.read_text(encoding="utf-8"))
                logger.debug("文件缓存索引已加载: %d 条", len(self._index))
        except Exception:
            self._index = {}

    def _save_index(self):
        """保存主索引"""
        try:
            self._index_file.write_text(json.dumps(self._index, ensure_ascii=False, indent=2), encoding="utf-8")
        except Exception as e:
            logger.warning("保存缓存索引失败: %s", e)

    def _stock_dir(self, stock_code: str) -> Path:
        clean = stock_code.strip().upper()
        d = _STOCK_CACHE_ROOT / clean
        d.mkdir(parents=True, exist_ok=True)
        return d

    def _data_file(self, stock_code: str, data_type: str) -> Path:
        filename = _DATA_TYPE_FILES.get(data_type, f"{data_type}.json")
        return self._stock_dir(stock_code) / filename

    def _update_index(self, stock_code: str, data_type: str):
        code = stock_code.strip().upper()
        if code not in self._index:
            self._index[code] = {"data_types": [], "cached_at": datetime.now().isoformat()}
        if data_type not in self._index[code]["data_types"]:
            self._index[code]["data_types"].append(data_type)
        self._index[code]["cached_at"] = datetime.now().isoformat()

    # ---- 读写操作 ----

    def get(self, stock_code: str, data_type: str, max_age_hours: int = 24) -> Optional[dict]:
        """
        从文件缓存读取数据

        Args:
            stock_code: 股票代码
            data_type: 数据类型 (quote/financial/profile/holders/sentiment/news)
            max_age_hours: 最大有效时长（小时），超过则视为过期

        Returns:
            缓存数据字典，未命中或过期返回 None
        """
        filepath = self._data_file(stock_code, data_type)
        if not filepath.exists():
            return None

        # 检查文件时效
        mtime = datetime.fromtimestamp(filepath.stat().st_mtime)
        age_hours = (datetime.now() - mtime).total_seconds() / 3600
        file_date = mtime.strftime("%Y-%m-%d")

        # 当日数据直接返回（不限小时数）
        if file_date == _TODAY:
            pass
        elif age_hours > max_age_hours:
            logger.debug("缓存过期: %s/%s (%.1f小时前)", stock_code, data_type, age_hours)
            return None

        try:
            data = json.loads(filepath.read_text(encoding="utf-8"))
            logger.debug("文件缓存命中: %s/%s", stock_code, data_type)
            return data
        except Exception as e:
            logger.warning("读取缓存文件失败 %s: %s", filepath, e)
            return None

    def set(self, stock_code: str, data_type: str, data: dict) -> bool:
        """
        将数据写入文件缓存

        Args:
            stock_code: 股票代码
            data_type: 数据类型
            data: 数据字典

        Returns:
            是否写入成功
        """
        filepath = self._data_file(stock_code, data_type)
        try:
            wrapper = {
                "stock_code": stock_code,
                "data_type": data_type,
                "cached_at": datetime.now().isoformat(),
                "cache_date": _TODAY,
                "data": data,
            }
            filepath.write_text(json.dumps(wrapper, ensure_ascii=False, indent=2), encoding="utf-8")
            self._update_index(stock_code, data_type)
            self._save_index()
            logger.debug("文件缓存写入: %s/%s", stock_code, data_type)
            return True
        except Exception as e:
            logger.warning("写入缓存文件失败 %s: %s", filepath, e)
            return False

    def has(self, stock_code: str, data_type: str) -> bool:
        """检查是否存在有效缓存"""
        filepath = self._data_file(stock_code, data_type)
        if not filepath.exists():
            return False
        mtime = datetime.fromtimestamp(filepath.stat().st_mtime)
        return mtime.strftime("%Y-%m-%d") == _TODAY

    # ---- grep 风格搜索 ----

    def grep_search(self, keyword: str, data_type: Optional[str] = None) -> list[dict]:
        """
        在所有缓存文件中搜索关键词（类似 grep）

        Args:
            keyword: 搜索关键词
            data_type: 限定数据类型，None 为全部

        Returns:
            匹配结果列表 [{stock_code, data_type, file_path, line: str, ...}]
        """
        results = []
        keyword_lower = keyword.lower()

        # 先查索引快速定位候选股票
        candidates = []
        for code, info in self._index.items():
            if keyword_lower in code.lower():
                candidates.append(code)
                continue
            types = info.get("data_types", [])
            if data_type and data_type not in types:
                continue
            candidates.append(code)

        # 对候选股票目录做内容 grep
        for code in candidates:
            stock_dir = self._stock_dir(code)
            if not stock_dir.exists():
                continue

            for fname in stock_dir.glob("*.json"):
                dtype = fname.stem
                if data_type and dtype != data_type:
                    continue

                try:
                    content = fname.read_text(encoding="utf-8")
                    if keyword_lower in content.lower():
                        # 提取匹配行
                        lines = content.split("\n")
                        matched_lines = [l.strip() for l in lines if keyword_lower in l.lower()]
                        results.append({
                            "stock_code": code,
                            "data_type": dtype,
                            "file_path": str(fname),
                            "matched_lines": matched_lines[:10],
                            "match_count": len(matched_lines),
                            "cached_at": datetime.fromtimestamp(fname.stat().st_mtime).isoformat(),
                        })
                except Exception:
                    continue

        return results

    def get_stock_codes(self) -> list[str]:
        """获取所有已缓存的股票代码"""
        return list(self._index.keys())

    def get_stock_data_types(self, stock_code: str) -> list[str]:
        """获取某股票已缓存的数据类型"""
        info = self._index.get(stock_code.upper(), {})
        return info.get("data_types", [])

    def clear_stock_cache(self, stock_code: Optional[str] = None):
        """清除缓存"""
        if stock_code:
            stock_dir = self._stock_dir(stock_code)
            for f in stock_dir.glob("*.json"):
                try:
                    f.unlink()
                except Exception:
                    pass
            code = stock_code.upper()
            self._index.pop(code, None)
            self._save_index()
        else:
            for f in _STOCK_CACHE_ROOT.glob("**/*.json"):
                try:
                    f.unlink()
                except Exception:
                    pass
            self._index.clear()
            self._save_index()

    def get_stats(self) -> dict:
        """获取缓存统计"""
        total_files = sum(1 for _ in _STOCK_CACHE_ROOT.glob("**/*.json"))
        total_size = sum(f.stat().st_size for f in _STOCK_CACHE_ROOT.glob("**/*.json") if f.is_file())
        return {
            "stock_count": len(self._index),
            "total_files": total_files,
            "total_size_mb": round(total_size / 1024 / 1024, 2),
            "cache_root": str(_STOCK_CACHE_ROOT),
        }


_stock_cache_instance: Optional[StockFileCache] = None


def get_stock_file_cache() -> StockFileCache:
    """获取 StockFileCache 全局单例"""
    global _stock_cache_instance
    if _stock_cache_instance is None:
        with _cache_lock:
            if _stock_cache_instance is None:
                _stock_cache_instance = StockFileCache()
    return _stock_cache_instance