Delete EasyFinancialAgent/chat_direct copy.py

EasyFinancialAgent/chat_direct copy.py

@@ -1,291 +0,0 @@
-"""
-Financial AI Assistant - Direct Method Library (不依赖 HTTP)
-直接导入并调用 easy_financial_mcp.py 中的函数
-支持本地和 HF Space 部署
-"""
-
-import sys
-from pathlib import Path
-
-# 添加服务模块路径
-PROJECT_ROOT = Path(__file__).parent.parent.absolute()
-sys.path.insert(0, str(PROJECT_ROOT))
-
-# 直接导入 MCP 中定义的函数
-try:
-    from EasyFinancialAgent.easy_financial_mcp import (
-        search_company as _search_company,
-        get_company_info as _get_company_info,
-        get_company_filings as _get_company_filings,
-        get_financial_data as _get_financial_data,
-        extract_financial_metrics as _extract_financial_metrics,
-        get_latest_financial_data as _get_latest_financial_data,
-        advanced_search_company as _advanced_search_company
-    )
-    MCP_DIRECT_AVAILABLE = True
-    print("[FinancialAI] ✓ Direct MCP functions imported successfully")
-except ImportError as e:
-    MCP_DIRECT_AVAILABLE = False
-    print(f"[FinancialAI] ✗ Failed to import MCP functions: {e}")
-    # 定义占位符函数
-    def _advanced_search_company(x):
-        return {"error": "MCP not available"}
-    def _get_company_info(x):
-        return {"error": "MCP not available"}
-    def _get_company_filings(x):
-        return {"error": "MCP not available"}
-    def _get_latest_financial_data(x):
-        return {"error": "MCP not available"}
-    def _extract_financial_metrics(x, y=3):
-        return {"error": "MCP not available"}
-
-
-# ============================================================
-# 便捷方法 - 公司搜索相关
-# ============================================================
-
-def search_company_direct(company_input):
-    """
-    搜索公司信息（直接调用）
-    
-    Args:
-        company_input: 公司名称或代码
-    
-    Returns:
-        搜索结果
-    
-    Example:
-        result = search_company_direct("Apple")
-    """
-    if not MCP_DIRECT_AVAILABLE:
-        return {"error": "MCP functions not available"}
-    
-    try:
-        return _advanced_search_company(company_input)
-    except Exception as e:
-        return {"error": str(e)}
-
-
-def get_company_info_direct(cik):
-    """
-    获取公司详细信息（直接调用）
-    
-    Args:
-        cik: 公司 CIK 代码
-    
-    Returns:
-        公司信息
-    
-    Example:
-        result = get_company_info_direct("0000320193")
-    """
-    if not MCP_DIRECT_AVAILABLE:
-        return {"error": "MCP functions not available"}
-    
-    try:
-        return _get_company_info(cik)
-    except Exception as e:
-        return {"error": str(e)}
-
-
-def get_company_filings_direct(cik):
-    """
-    获取公司 SEC 文件列表（直接调用）
-    
-    Args:
-        cik: 公司 CIK 代码
-    
-    Returns:
-        文件列表
-    
-    Example:
-        result = get_company_filings_direct("0000320193")
-    """
-    if not MCP_DIRECT_AVAILABLE:
-        return {"error": "MCP functions not available"}
-    
-    try:
-        return _get_company_filings(cik)
-    except Exception as e:
-        return {"error": str(e)}
-
-
-# ============================================================
-# 便捷方法 - 财务数据相关
-# ============================================================
-
-def get_latest_financial_data_direct(cik):
-    """
-    获取公司最新财务数据（直接调用）
-    
-    Args:
-        cik: 公司 CIK 代码
-    
-    Returns:
-        最新财务数据
-    
-    Example:
-        result = get_latest_financial_data_direct("0000320193")
-    """
-    if not MCP_DIRECT_AVAILABLE:
-        return {"error": "MCP functions not available"}
-    
-    try:
-        return _get_latest_financial_data(cik)
-    except Exception as e:
-        return {"error": str(e)}
-
-
-def extract_financial_metrics_direct(cik, years=5):
-    """
-    提取多年财务指标趋势（直接调用）
-    
-    Args:
-        cik: 公司 CIK 代码
-        years: 年数（默认 3 年）
-    
-    Returns:
-        财务指标数据
-    
-    Example:
-        result = extract_financial_metrics_direct("0000320193", years=5)
-    """
-    if not MCP_DIRECT_AVAILABLE:
-        return {"error": "MCP functions not available"}
-    
-    try:
-        return _extract_financial_metrics(cik, years)
-    except Exception as e:
-        return {"error": str(e)}
-
-
-# ============================================================
-# 高级方法 - 综合查询
-# ============================================================
-
-def query_company_direct(company_input, get_filings=True, get_metrics=True):
-    """
-    综合查询公司信息（直接调用）
-    包括搜索、基本信息、文件列表和财务指标
-    
-    Args:
-        company_input: 公司名称或代码
-        get_filings: 是否获取文件列表
-        get_metrics: 是否获取财务指标
-    
-    Returns:
-        综合结果字典，包含 search, company_info, filings, metrics
-    
-    Example:
-        result = query_company_direct("Apple", get_filings=True, get_metrics=True)
-    """
-    from datetime import datetime
-    
-    result = {
-        "timestamp": datetime.now().isoformat(),
-        "query_input": company_input,
-        "status": "success",
-        "data": {
-            "company_search": None,
-            "company_info": None,
-            "filings": None,
-            "metrics": None
-        },
-        "errors": []
-    }
-    
-    if not MCP_DIRECT_AVAILABLE:
-        result["status"] = "error"
-        result["errors"].append("MCP functions not available")
-        return result
-    
-    try:
-        # 1. 搜索公司
-        search_result = search_company_direct(company_input)
-        if "error" in search_result:
-            result["errors"].append(f"Search error: {search_result['error']}")
-            result["status"] = "error"
-            return result
-        
-        result["data"]["company_search"] = search_result
-        
-        # 从搜索结果提取 CIK
-        cik = None
-        if isinstance(search_result, dict):
-            cik = search_result.get("cik")
-        elif isinstance(search_result, (list, tuple)) and len(search_result) > 0:
-            # 从列表中获取第一个元素
-            try:
-                first_item = search_result[0] if isinstance(search_result, (list, tuple)) else None
-                if isinstance(first_item, dict):
-                    cik = first_item.get("cik")
-            except (IndexError, TypeError):
-                pass
-        
-        if not cik:
-            result["errors"].append("Could not extract CIK from search result")
-            result["status"] = "error"
-            return result
-        
-        # 2. 获取公司信息
-        company_info = get_company_info_direct(cik)
-        if "error" not in company_info:
-            result["data"]["company_info"] = company_info
-        else:
-            result["errors"].append(f"Failed to get company info: {company_info.get('error')}")
-        
-        # 3. 获取文件列表
-        if get_filings:
-            filings = get_company_filings_direct(cik)
-            if "error" not in filings:
-                result["data"]["filings"] = filings
-            else:
-                result["errors"].append(f"Failed to get filings: {filings.get('error')}")
-        
-        # 4. 获取财务指标
-        if get_metrics:
-            metrics = extract_financial_metrics_direct(cik, years=3)
-            if "error" not in metrics:
-                result["data"]["metrics"] = metrics
-            else:
-                result["errors"].append(f"Failed to get metrics: {metrics.get('error')}")
-        
-    except Exception as e:
-        result["status"] = "error"
-        result["errors"].append(f"Exception: {str(e)}")
-        import traceback
-        result["errors"].append(traceback.format_exc())
-    
-    return result
-
-
-# ============================================================
-# 测试函数
-# ============================================================
-
-if __name__ == "__main__":
-    print("\n" + "="*60)
-    print("Financial AI Assistant - Direct Method Test")
-    print("="*60)
-    
-    # 测试 1: 公司搜索
-    print("\n1. 搜索公司 (Apple)...")
-    result = search_company_direct("Apple")
-    print(f"   结果: {result}")
-    
-    # 测试 2: 公司摘要
-    print("\n2. 获取 Tesla 完整信息...")
-    summary = query_company_direct("Tesla", get_filings=False, get_metrics=False)
-    print(f"   状态: {summary['status']}")
-    print(f"   数据: {summary['data']}")
-    print(f"   错误: {summary['errors']}")
-    
-    # 测试 3: 财务指标
-    print("\n3. 获取 Microsoft 财务指标...")
-    if summary['data']['company_search']:
-        cik = summary['data']['company_search'].get('cik')
-        if cik:
-            metrics = extract_financial_metrics_direct(cik, years=3)
-            print(f"   指标: {metrics}")
-    
-    print("\n" + "="*60)