Gangtise Kb

# 冈底斯知识库 Access Key
# 在冈底斯平台的「账号设置 > API 访问」中获取
GANGTISE_ACCESS_KEY=your_gangtise_access_key_here

# 冈底斯知识库 Secret Key（请妥善保管，切勿泄露）
GANGTISE_SECRET_KEY=your_gangtise_secret_key_here

[project]
name = "gangtise-kb-mcp"
version = "1.0.0"
description = "冈特斯开放平台 MCP Server"
requires-python = ">=3.11"
dependencies = [
    "fastmcp>=2.0.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["."]

#!/usr/bin/env python3
"""
冈特斯开放平台 MCP Server。

直接复用 ../scripts/ 里已有的客户端和查询函数，不重复实现逻辑。
提供 5 个工具：
  - gangtise_kb_search     知识库语义搜索（研报/公告/纪要）
  - gangtise_forecast      公司盈利预测（EPS/PE/ROE 等）
  - gangtise_indicator     经济数据指标 AI 查询
  - gangtise_meetings      电话会议/路演纪要列表
  - gangtise_download_url  溯源下载 URL 获取
"""
import json
import sys
import time
from datetime import datetime
from pathlib import Path

# 复用已有 scripts，不动原始文件
SCRIPTS_DIR = Path(__file__).parent.parent / "scripts"
sys.path.insert(0, str(SCRIPTS_DIR))

from fastmcp import FastMCP
from _client import post, post_stream, BASE_URL
from query_kb import query_knowledge, RT_NAMES
from forecast import query_forecast, KEY_FIELDS, FIELD_NAMES
from indicator import query_indicator
from meeting_list import query_meetings, PROCESS_MAP, METHOD_MAP

mcp = FastMCP("Gangtise KB", version="1.0.0")


# ─── 工具 1：知识库搜索 ──────────────────────────────────

@mcp.tool()
def gangtise_kb_search(
    query: str,
    resource_types: str = "10,40",
    top: int = 10,
    days: int = 365,
) -> str:
    """
    冈特斯知识库语义搜索。搜索券商研报、公告、会议纪要、分析师观点等。

    参数:
      query: 查询关键词，如 "振华股份铬盐" 或 "新能源汽车行业展望"
      resource_types: 资源类型（逗号分隔）
                      10=券商研报 20=内部研报 40=分析师观点 50=公告
                      60=会议纪要 70=调研纪要 80=网络资源 90=产业公众号
                      默认 "10,40"（研报+分析师观点）
      top: 返回数量，最多 20，默认 10
      days: 搜索最近 N 天，默认 365
    返回:
      格式化的搜索结果，含标题、公司、时间、内容摘要、sourceId
    """
    types = [int(t.strip()) for t in resource_types.split(",") if t.strip()]
    result = query_knowledge(query, types, min(top, 20), days)

    if not result or result.get("code") != "000000":
        msg = result.get("msg", "无响应") if result else "无响应"
        return f"查询失败: {msg}"

    data = result.get("data", [])
    if not data:
        return "无结果。"

    lines = []
    for query_result in data:
        items = query_result.get("data", [])
        if not items:
            lines.append("无匹配结果。")
            continue

        for i, item in enumerate(items, 1):
            rt = item.get("resourceType")
            ts = item.get("time")
            date_str = datetime.fromtimestamp(ts / 1000).strftime("%Y-%m-%d") if ts else "N/A"
            source_id = item.get("sourceId", "")
            content = item.get("content", "")
            if len(content) > 600:
                content = content[:600] + "..."

            lines.append(f"\n[{i}] {item.get('title', 'N/A')}")
            lines.append(f"    公司: {item.get('company', 'N/A')} | 类型: {RT_NAMES.get(rt, rt)} | 日期: {date_str}")
            if source_id and rt != 40:
                lines.append(f"    SourceId: {source_id}")
            if content:
                lines.append(f"    摘要: {content}")

    return "\n".join(lines) if lines else "无结果。"


# ─── 工具 2：盈利预测 ────────────────────────────────────

@mcp.tool()
def gangtise_forecast(stock_code: str) -> str:
    """
    查询公司盈利预测（一致预期）。返回历年实际值和未来预测值。

    参数:
      stock_code: 股票代码，如 "603067.SH" 或 "000858.SZ"
    返回:
      含归母净利润、EPS、PE、ROE、营收等指标的多年预测表
    """
    result = query_forecast(stock_code)

    if not result or result.get("code") != "000000":
        msg = result.get("msg", "无响应") if result else "无响应"
        return f"查询失败: {msg}"

    items = result.get("data", [])
    if not items:
        return "无预测数据。"

    items.sort(key=lambda x: x.get("fincForeYear", ""))

    lines = [f"【{stock_code} 盈利预测】"]
    for item in items:
        year = item.get("fincForeYear", "N/A")
        # A=实际值 E=预测值
        suffix = "(实际)" if str(year).endswith("A") else "(预测)"
        lines.append(f"\n── {year} {suffix} ──")
        for field in KEY_FIELDS:
            val = item.get(field)
            if val is not None and val != "":
                lines.append(f"  {FIELD_NAMES.get(field, field)}: {val}")

    return "\n".join(lines)


# ─── 工具 3：经济数据指标 ────────────────────────────────

@mcp.tool()
def gangtise_indicator(text: str) -> str:
    """
    经济数据指标 AI 查询。用自然语言查询行业数据、宏观经济、公司经营指标。
    注意：非流式模式约等待 10-30 秒。

    参数:
      text: 自然语言查询，如 "铬盐行业2025年产能" 或 "2024年GDP增速"
    返回:
      AI 生成的数据分析回答
    """
    result = query_indicator(text, stream=False)

    if not result:
        return "查询失败或无响应。"

    if isinstance(result, str):
        return result

    if result.get("code") != "000000":
        return f"查询失败: {result.get('msg', '未知错误')}"

    choices = result.get("data", {}).get("choices", [])
    if not choices:
        return "无结果。"

    msg = choices[0].get("message", {})
    content = msg.get("content", "")
    return content if content else "无内容返回。"


# ─── 工具 4：电话会议列表 ────────────────────────────────

@mcp.tool()
def gangtise_meetings(
    stock: str | None = None,
    topic: str | None = None,
    days: int = 7,
    size: int = 10,
    page: int = 1,
) -> str:
    """
    查询电话会议、路演纪要列表。

    参数:
      stock: 股票代码过滤，如 "603067.SH"（可选）
      topic: 会议主题关键词过滤，如 "铬盐" 或 "新能源"（可选）
      days: 查询最近 N 天，默认 7
      size: 每页数量，默认 10
      page: 页码，默认 1
    返回:
      会议列表，含时间、主题、机构、摘要
    """
    stock_codes = [stock] if stock else None
    topics = [topic] if topic else None

    result = query_meetings(page, size, stock_codes, topics, days)

    if not result or result.get("code") != "000000":
        msg = result.get("msg", "无响应") if result else "无响应"
        return f"查询失败: {msg}"

    page_data = result.get("data", {})
    total = page_data.get("total", 0)
    items = page_data.get("list", [])

    if not items:
        return f"最近 {days} 天无会议记录。"

    lines = [f"共 {total} 条会议，当前显示 {len(items)} 条："]
    for item in items:
        cnfr_time = item.get("cnfrTime")
        time_str = datetime.fromtimestamp(cnfr_time / 1000).strftime("%m-%d %H:%M") if cnfr_time else "N/A"
        process = PROCESS_MAP.get(item.get("process"), str(item.get("process", "")))
        method = METHOD_MAP.get(item.get("cnfrCreateMethod"), "")

        lines.append(f"\n[{time_str}] {item.get('topic', 'N/A')}")
        lines.append(f"  机构: {item.get('partyName', 'N/A')} | 行业: {item.get('blockName', 'N/A')} | 状态: {process} | 类型: {method}")

        security = item.get("securityAbbr")
        if security:
            lines.append(f"  关联股票: {security} ({item.get('securityId', '')})")

        summary = item.get("summary", "")
        if summary:
            if len(summary) > 400:
                summary = summary[:400] + "..."
            lines.append(f"  摘要: {summary}")

    return "\n".join(lines)


# ─── 工具 5：溯源下载 URL ────────────────────────────────

@mcp.tool()
def gangtise_download_url(resource_type: int, source_id: str) -> str:
    """
    获取研报/纪要的溯源下载 URL（type 40 分析师观点不支持）。

    参数:
      resource_type: 资源类型（10=券商研报 20=内部研报 50=公告 60=会议纪要 70=调研纪要 80/90=网络资源）
      source_id: 来自 gangtise_kb_search 返回的 sourceId
    返回:
      下载链接或提示信息
    """
    if resource_type == 40:
        return "类型 40（分析师观点）不支持溯源下载。"

    url = f"{BASE_URL}/application/open-data/ai/resource/download?resourceType={resource_type}&sourceId={source_id}"
    return f"下载链接：{url}\n（注：部分券商研报有白名单限制，80/90 类型返回第三方 URL）"


if __name__ == "__main__":
    mcp.run()

#!/usr/bin/env python3
"""
冈特斯开放平台公共 HTTP 客户端。

职责：
  - Token 获取与文件缓存（1 小时 TTL）
  - 统一 HTTP 请求封装（POST/GET）
  - SSL context 复用
  - 统一错误处理
"""
import json
import os
import sys
import time
import urllib.request
import urllib.error
import ssl
from pathlib import Path

# 路径常量
SKILL_ROOT = Path(__file__).parent.parent
TOKEN_CACHE_FILE = SKILL_ROOT / ".token_cache"

# API 常量
BASE_URL = "https://open.gangtise.com"
LOGIN_ENDPOINT = "/application/auth/oauth/open/loginV2"
TOKEN_TTL = 3500  # 缓存有效期（秒），略小于服务端 3600 秒

# 全局 SSL context（复用，避免每次创建）
_ssl_ctx = ssl.create_default_context()
_ssl_ctx.check_hostname = False
_ssl_ctx.verify_mode = ssl.CERT_NONE


# ─── 凭证加载 ───────────────────────────────────────────

# 复用 configure.py 的凭证逻辑
sys.path.insert(0, str(Path(__file__).parent))
from configure import get_credentials, check_configured


def require_configured() -> None:
    """检查凭证是否可用，不可用则退出。"""
    if not check_configured():
        print("未配置凭证，请先设置环境变量或运行 python3 scripts/configure.py", file=sys.stderr)
        sys.exit(1)


# ─── Token 缓存 ─────────────────────────────────────────

def _read_cached_token() -> str | None:
    """从缓存文件读取未过期的 token。"""
    if not TOKEN_CACHE_FILE.exists():
        return None
    try:
        with open(TOKEN_CACHE_FILE, 'r') as f:
            cache = json.load(f)
        # 检查是否过期
        if time.time() - cache.get('ts', 0) < TOKEN_TTL:
            return cache.get('token')
    except Exception:
        pass
    return None


def _write_cached_token(token: str) -> None:
    """将 token 写入缓存文件。"""
    try:
        with open(TOKEN_CACHE_FILE, 'w') as f:
            json.dump({'token': token, 'ts': time.time()}, f)
        os.chmod(TOKEN_CACHE_FILE, 0o600)
    except Exception:
        pass  # 缓存写入失败不影响功能


def get_token() -> str | None:
    """
    获取 access token（带缓存）。

    优先读缓存，未命中则调用 loginV2 并缓存。
    返回值已包含 Bearer 前缀。
    """
    # 1. 尝试缓存
    cached = _read_cached_token()
    if cached:
        return cached

    # 2. 调用 loginV2
    ak, sk = get_credentials()
    if not ak or not sk:
        return None

    data = {"accessKey": ak, "secretAccessKey": sk}
    result = post(LOGIN_ENDPOINT, data, auth=False)
    if result and result.get('code') == '000000' and result.get('data'):
        token = result['data'].get('accessToken')
        if token:
            _write_cached_token(token)
            return token

    print("Token 获取失败", file=sys.stderr)
    return None


# ─── HTTP 请求 ───────────────────────────────────────────

def post(endpoint: str, data: dict, auth: bool = True, timeout: int = 30) -> dict | None:
    """
    发送 POST 请求到冈特斯 API。

    参数:
      endpoint: API 路径（如 /application/open-data/...）
      data: 请求体字典
      auth: 是否附加 Authorization header
      timeout: 超时秒数
    返回:
      解析后的 JSON dict，失败返回 None。
    """
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
    }
    if auth:
        token = get_token()
        if not token:
            return None
        headers['Authorization'] = token if token.startswith('Bearer ') else f'Bearer {token}'

    req = urllib.request.Request(
        f"{BASE_URL}{endpoint}",
        data=json.dumps(data).encode('utf-8'),
        headers=headers,
        method='POST'
    )

    try:
        with urllib.request.urlopen(req, context=_ssl_ctx, timeout=timeout) as response:
            return json.loads(response.read().decode('utf-8'))
    except urllib.error.HTTPError as e:
        _handle_http_error(e)
    except Exception as e:
        print(f"请求错误: {e}", file=sys.stderr)
    return None


def get(endpoint: str, params: dict | None = None, timeout: int = 30) -> urllib.response.addinfourl | None:
    """
    发送 GET 请求，返回原始 response 对象（用于文件下载等）。

    调用方负责读取和关闭 response。失败返回 None。
    """
    token = get_token()
    if not token:
        return None

    url = f"{BASE_URL}{endpoint}"
    if params:
        qs = '&'.join(f"{k}={v}" for k, v in params.items())
        url = f"{url}?{qs}"

    req = urllib.request.Request(
        url,
        headers={'Authorization': token if token.startswith('Bearer ') else f'Bearer {token}'},
        method='GET'
    )

    try:
        return urllib.request.urlopen(req, context=_ssl_ctx, timeout=timeout)
    except urllib.error.HTTPError as e:
        _handle_http_error(e)
    except Exception as e:
        print(f"请求错误: {e}", file=sys.stderr)
    return None


def post_stream(endpoint: str, data: dict, timeout: int = 120):
    """
    发送 POST 请求并返回原始 response 用于 SSE 流式读取。

    调用方负责逐行读取。失败返回 None。
    """
    token = get_token()
    if not token:
        return None

    headers = {
        'Content-Type': 'application/json',
        'Accept': 'text/event-stream',
        'Authorization': token if token.startswith('Bearer ') else f'Bearer {token}',
    }

    req = urllib.request.Request(
        f"{BASE_URL}{endpoint}",
        data=json.dumps(data).encode('utf-8'),
        headers=headers,
        method='POST'
    )

    try:
        return urllib.request.urlopen(req, context=_ssl_ctx, timeout=timeout)
    except urllib.error.HTTPError as e:
        _handle_http_error(e)
    except Exception as e:
        print(f"请求错误: {e}", file=sys.stderr)
    return None


# ─── 错误处理 ───────────────────────────────────────────

def _handle_http_error(e: urllib.error.HTTPError) -> None:
    """统一处理 HTTP 错误，尝试解析冈特斯业务错误码。"""
    try:
        body = e.read().decode('utf-8', errors='replace')
        err = json.loads(body)
        code = err.get('code', '')
        msg = err.get('msg', '')
        if code or msg:
            print(f"错误 [{code}]: {msg}", file=sys.stderr)
            return
    except Exception:
        pass
    print(f"HTTP {e.code}: {e.reason}", file=sys.stderr)

#!/usr/bin/env python3
"""
clean_md.py — Gangtise 纪要 HTML→Markdown 清洗脚本

用法:
  # 清洗单个文件（原地覆盖）
  python3 scripts/clean_md.py file.md

  # 清洗单个文件（输出到新文件）
  python3 scripts/clean_md.py file.md -o clean_file.md

  # 批量清洗目录下所有 .md / .txt
  python3 scripts/clean_md.py dir/

  # 从 stdin 读取，stdout 输出（管道模式）
  cat dirty.md | python3 scripts/clean_md.py -

清洗规则:
  1. 删除 <span class='meeting_summary_num' ...>N</span> 隐藏时间戳
  2. HTML 标签转 Markdown（h1→#, h2→##, strong→**, li→-, p→段落等）
  3. 清理 HTML 实体（&nbsp; &amp; &lt; &gt; &quot; &#xNNNN;）
  4. 修复多余空行（连续3+空行合并为2）
  5. 修复行尾多余空格
"""

import re
import sys
import argparse
from pathlib import Path


def clean_gangtise_html(text: str) -> str:
    """将 Gangtise 纪要的 HTML 混合文本转为干净 Markdown。"""

    # ── Step 0: 删除隐藏的时间戳 span ──
    # <span class='meeting_summary_num' style='display:none' data-time-start='...' data-time-end='...'>N</span>
    text = re.sub(
        r"<span\s+class=['\"]meeting_summary_num['\"][^>]*>.*?</span>",
        "",
        text,
        flags=re.DOTALL,
    )

    # ── Step 1: 块级标签转 Markdown ──

    # <h1>...</h1> → # ...
    text = re.sub(r"<h1[^>]*>(.*?)</h1>", r"\n# \1\n", text, flags=re.DOTALL)
    # <h2>...</h2> → ## ...
    text = re.sub(r"<h2[^>]*>(.*?)</h2>", r"\n## \1\n", text, flags=re.DOTALL)
    # <h3>...</h3> → ### ...
    text = re.sub(r"<h3[^>]*>(.*?)</h3>", r"\n### \1\n", text, flags=re.DOTALL)
    # <h4>...</h4> → #### ...
    text = re.sub(r"<h4[^>]*>(.*?)</h4>", r"\n#### \1\n", text, flags=re.DOTALL)

    # <strong>...</strong> / <b>...</b> → **...**
    text = re.sub(r"<(?:strong|b)[^>]*>(.*?)</(?:strong|b)>", r"**\1**", text, flags=re.DOTALL)
    # <em>...</em> / <i>...</i> → *...*
    text = re.sub(r"<(?:em|i)[^>]*>(.*?)</(?:em|i)>", r"*\1*", text, flags=re.DOTALL)

    # <li><p>...</p></li> → 先去掉内层 p
    text = re.sub(r"<li>\s*<p>(.*?)</p>\s*</li>", r"<li>\1</li>", text, flags=re.DOTALL)

    # <li>...</li> → - ...
    text = re.sub(r"<li[^>]*>(.*?)</li>", lambda m: "- " + m.group(1).strip(), text, flags=re.DOTALL)

    # <p>...</p> → 段落（前后加空行）
    text = re.sub(r"<p[^>]*>(.*?)</p>", r"\n\1\n", text, flags=re.DOTALL)

    # <br\s*/?>  → 换行
    text = re.sub(r"<br\s*/?>", "\n", text)

    # ── Step 2: 删除剩余 HTML 标签 ──
    # 先删 <ul>, </ul>, <ol>, </ol> 等容器标签（不影响内容）
    text = re.sub(r"</?(?:ul|ol|div|section|article|header|footer|nav|table|thead|tbody|tr|td|th|blockquote|figure|figcaption|details|summary|main|aside|span|a|img|code|pre|sup|sub)[^>]*>", "", text)
    # 兜底：删除所有残留的 HTML 标签
    text = re.sub(r"<[^>]+>", "", text)

    # ── Step 3: HTML 实体 ──
    text = text.replace("&nbsp;", " ")
    text = text.replace("&amp;", "&")
    text = text.replace("&lt;", "<")
    text = text.replace("&gt;", ">")
    text = text.replace("&quot;", '"')
    text = text.replace("&apos;", "'")
    text = text.replace("&#39;", "'")
    # 处理 &#xNNNN; 和 &#NNNN; 数字实体
    text = re.sub(r"&#x([0-9a-fA-F]+);", lambda m: chr(int(m.group(1), 16)), text)
    text = re.sub(r"&#(\d+);", lambda m: chr(int(m.group(1))), text)

    # ── Step 4: 清理格式 ──
    # 行尾多余空格（保留 Markdown 的两个空格换行）
    text = re.sub(r"[ \t]+$", "", text, flags=re.MULTILINE)
    # 连续3+空行 → 2空行
    text = re.sub(r"\n{3,}", "\n\n", text)
    # 标题前确保有空行
    text = re.sub(r"([^\n])\n(#{1,4} )", r"\1\n\n\2", text)
    # 列表项 "- " 前如果紧跟文本，加空行
    text = re.sub(r"([^\n-])\n(- )", r"\1\n\n\2", text)

    return text.strip() + "\n"


def process_file(path: Path, output: Path | None = None):
    """清洗单个文件。"""
    content = path.read_text(encoding="utf-8", errors="ignore")
    cleaned = clean_gangtise_html(content)
    dest = output or path
    dest.write_text(cleaned, encoding="utf-8")
    print(f"✅ {path} → {dest}  ({len(content)} → {len(cleaned)} bytes)")


def main():
    parser = argparse.ArgumentParser(description="Gangtise 纪要 HTML→Markdown 清洗")
    parser.add_argument("input", help="文件路径、目录路径、或 '-' 表示 stdin")
    parser.add_argument("-o", "--output", help="输出文件路径（单文件模式）", default=None)
    args = parser.parse_args()

    if args.input == "-":
        # stdin → stdout
        text = sys.stdin.read()
        sys.stdout.write(clean_gangtise_html(text))
        return

    p = Path(args.input)
    if p.is_file():
        process_file(p, Path(args.output) if args.output else None)
    elif p.is_dir():
        files = sorted(p.glob("*.md")) + sorted(p.glob("*.txt"))
        if not files:
            print(f"⚠️  {p} 下没有 .md / .txt 文件")
            return
        for f in files:
            process_file(f)
        print(f"\n共清洗 {len(files)} 个文件")
    else:
        print(f"❌ 路径不存在: {p}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Gangtise Knowledge Base Skill 配置与凭证管理。

凭证加载优先级：
  1. 环境变量 GANGTISE_ACCESS_KEY / GANGTISE_SECRET_KEY
  2. .env 文件（项目根目录）
  3. config.json（兼容旧配置）
"""
import json
import os
import sys
from pathlib import Path

SKILL_ROOT = Path(__file__).parent.parent
CONFIG_FILE = SKILL_ROOT / "config.json"
ENV_FILE = SKILL_ROOT / ".env"

# 环境变量名
ENV_ACCESS_KEY = "GANGTISE_ACCESS_KEY"
ENV_SECRET_KEY = "GANGTISE_SECRET_KEY"


def _load_dotenv() -> dict[str, str]:
    """从 .env 文件解析 KEY=VALUE 对，不引入额外依赖。"""
    result: dict[str, str] = {}
    if not ENV_FILE.exists():
        return result
    try:
        with open(ENV_FILE, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                # 跳过空行和注释
                if not line or line.startswith('#'):
                    continue
                if '=' not in line:
                    continue
                key, _, value = line.partition('=')
                key = key.strip()
                value = value.strip().strip('"').strip("'")
                result[key] = value
    except Exception:
        pass
    return result


def _load_config_json() -> dict:
    """从 config.json 加载旧格式配置。"""
    if CONFIG_FILE.exists():
        try:
            with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
                return json.load(f)
        except Exception:
            pass
    return {}


def get_credentials() -> tuple[str | None, str | None]:
    """
    获取 API 凭证，按优先级尝试：
      1. 环境变量
      2. .env 文件
      3. config.json
    返回 (access_key, secret_key)。
    """
    # 1. 环境变量
    ak = os.environ.get(ENV_ACCESS_KEY)
    sk = os.environ.get(ENV_SECRET_KEY)
    if ak and sk:
        return ak, sk

    # 2. .env 文件
    dotenv = _load_dotenv()
    ak = dotenv.get(ENV_ACCESS_KEY)
    sk = dotenv.get(ENV_SECRET_KEY)
    if ak and sk:
        return ak, sk

    # 3. config.json（兼容旧配置）
    config = _load_config_json()
    return config.get('ACCESS_KEY'), config.get('SECRET_KEY')


def check_configured() -> bool:
    """检查凭证是否可用。"""
    ak, sk = get_credentials()
    return bool(ak and sk)


def save_config(config: dict) -> bool:
    """保存配置到 config.json（仅供交互式 configure 使用）。"""
    try:
        with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
            json.dump(config, f, indent=2, ensure_ascii=False)
        os.chmod(CONFIG_FILE, 0o600)
        return True
    except Exception as e:
        print(f"Error saving config: {e}", file=sys.stderr)
        return False


def main():
    print("=" * 60)
    print("Gangtise Knowledge Base - Configuration Setup")
    print("=" * 60)
    print()

    # 先检查环境变量 / .env 是否已配置
    if check_configured():
        ak, _ = get_credentials()
        print(f"已检测到有效凭证（Access Key: {ak[:4]}...）")
        print("如需更新，请修改环境变量或 .env 文件。")
        return

    print("请输入 Gangtise API 凭证。")
    print("获取地址: https://open.gangtise.com")
    print()

    # 交互式输入，写入 .env 文件
    ak = input("Access Key: ").strip()
    sk = input("Secret Key: ").strip()

    if not ak or not sk:
        print("\nError: Both Access Key and Secret Key are required.", file=sys.stderr)
        sys.exit(1)

    # 写入 .env 文件
    try:
        with open(ENV_FILE, 'w', encoding='utf-8') as f:
            f.write(f"{ENV_ACCESS_KEY}={ak}\n")
            f.write(f"{ENV_SECRET_KEY}={sk}\n")
        os.chmod(ENV_FILE, 0o600)
        print(f"\n凭证已保存到: {ENV_FILE}")
        print("也可直接设置环境变量 GANGTISE_ACCESS_KEY / GANGTISE_SECRET_KEY。")
    except Exception as e:
        print(f"\n保存失败: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
通过 sourceId 溯源下载原始资源（研报 PDF、公告等）。

用法:
  python3 scripts/download_resource.py --type 10 --id SOURCE_ID
  python3 scripts/download_resource.py --type 10 --id SOURCE_ID --output report.pdf
"""
import json
import sys
import argparse
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

from _client import require_configured, get as http_get

DOWNLOAD_ENDPOINT = "/application/open-data/ai/resource/download"

# 溯源支持的资源类型（不支持 40 首席分析师观点）
SUPPORTED_TYPES = {10, 20, 50, 60, 70, 80, 90}


def download_resource(resource_type: int, source_id: str, output_path: str | None = None) -> str | None:
    """
    下载溯源资源。

    返回:
      - 第三方 URL 字符串（Content-Type 为 json 时）
      - 保存的文件路径（文件流时）
      - None（失败时）
    """
    if resource_type not in SUPPORTED_TYPES:
        print(f"错误: resourceType {resource_type} 不支持溯源（type 40 不可用）", file=sys.stderr)
        return None

    params = {"resourceType": resource_type, "sourceId": source_id}
    response = http_get(DOWNLOAD_ENDPOINT, params, timeout=60)
    if not response:
        return None

    try:
        content_type = response.headers.get('Content-Type', '')

        # 返回第三方 URL
        if 'application/json' in content_type:
            result = json.loads(response.read().decode('utf-8'))
            return result.get('url')

        # 返回文件流，保存到本地
        if output_path is None:
            cd = response.headers.get('Content-Disposition', '')
            if 'filename=' in cd:
                output_path = cd.split('filename=')[-1].strip('"\'')
            else:
                ext = '.pdf' if 'pdf' in content_type else '.bin'
                output_path = f"resource_{source_id}{ext}"

        with open(output_path, 'wb') as f:
            f.write(response.read())
        return output_path
    except Exception as e:
        print(f"下载错误: {e}", file=sys.stderr)
        return None
    finally:
        response.close()


def auto_clean_md(filepath: str) -> str | None:
    """如果文件是 .md 或 .txt，自动调用 clean_md 清洗 HTML→Markdown，原地覆盖。"""
    p = Path(filepath)
    if p.suffix.lower() not in ('.md', '.txt'):
        return None
    try:
        from clean_md import clean_gangtise_html
        content = p.read_text(encoding='utf-8', errors='ignore')
        cleaned = clean_gangtise_html(content)
        p.write_text(cleaned, encoding='utf-8')
        return str(p)
    except ImportError:
        # clean_md.py 不存在时静默跳过
        return None
    except Exception as e:
        print(f"⚠️  自动清洗失败: {e}", file=sys.stderr)
        return None


def main():
    parser = argparse.ArgumentParser(description='溯源下载原始资源')
    parser.add_argument('--type', type=int, required=True, help='资源类型 (10/20/50/60/70/80/90)')
    parser.add_argument('--id', required=True, help='sourceId')
    parser.add_argument('--output', help='输出文件路径（可选）')
    parser.add_argument('--no-clean', action='store_true', help='跳过自动 MD 清洗')

    args = parser.parse_args()
    require_configured()

    result = download_resource(args.type, args.id, args.output)
    if result:
        print(result)
        # 自动清洗
        if not args.no_clean and Path(result).is_file():
            cleaned = auto_clean_md(result)
            if cleaned:
                print(f"✅ 已自动清洗: {cleaned}", file=sys.stderr)
    else:
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
查询公司盈利预测数据。

用法:
  python3 scripts/forecast.py 600519.SH
  python3 scripts/forecast.py 000858.SZ --json
"""
import json
import sys
import argparse
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

from _client import require_configured, post

FORECAST_ENDPOINT = "/application/open-data/report/forecast/info"

# 字段中文映射
FIELD_NAMES = {
    'shnp': '归母净利润(百万)', 'shnpGr': '净利润增长率(%)', 'eps': '每股收益',
    'pe': '市盈率', 'bps': '每股净资产', 'pb': '市净率', 'peg': 'PEG',
    'roe': 'ROE(%)', 'ps': '市销率', 'or': '营业收入(百万)', 'rgr': '营收增长率(%)',
    'sgpm': '销售毛利率(%)', 'npmos': '净利率(%)', 'roic': 'ROIC(%)', 'coas': '总资产周转率(%)',
}

# 显示顺序
KEY_FIELDS = ['shnp', 'shnpGr', 'eps', 'pe', 'roe', 'bps', 'pb', 'or', 'rgr', 'sgpm', 'npmos', 'roic']


def query_forecast(stock_code: str) -> dict | None:
    """查询公司盈利预测。"""
    return post(FORECAST_ENDPOINT, {"stockCode": stock_code})


def format_results(result: dict) -> None:
    """格式化输出盈利预测。"""
    if not result or result.get('code') != '000000':
        print(f"查询失败: {result.get('msg', '未知错误') if result else '无响应'}")
        return

    items = result.get('data', [])
    if not items:
        print("无预测数据。")
        return

    items.sort(key=lambda x: x.get('fincForeYear', ''))

    for item in items:
        print(f"\n--- {item.get('fincForeYear', 'N/A')} ---")
        for field in KEY_FIELDS:
            val = item.get(field, '')
            if val:
                print(f"  {FIELD_NAMES.get(field, field)}: {val}")


def main():
    parser = argparse.ArgumentParser(description='查询公司盈利预测')
    parser.add_argument('stock_code', help='股票代码，如 600519.SH')
    parser.add_argument('--json', action='store_true', help='输出原始 JSON')

    args = parser.parse_args()
    require_configured()

    result = query_forecast(args.stock_code)
    if result:
        if args.json:
            print(json.dumps(result, indent=2, ensure_ascii=False))
        else:
            format_results(result)
    else:
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""获取冈特斯 access token（带缓存）。"""
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

from _client import require_configured, get_token

if __name__ == "__main__":
    require_configured()
    token = get_token()
    if token:
        print(token)
    else:
        sys.exit(1)

#!/usr/bin/env python3
"""
经济数据指标查询 Agent。

用法:
  python3 scripts/indicator.py "比亚迪仰望销量"
  python3 scripts/indicator.py "2024年GDP增速" --stream
  python3 scripts/indicator.py "光伏行业产能" --json
"""
import json
import sys
import argparse
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

from _client import require_configured, post, post_stream

INDICATOR_ENDPOINT = "/application/open-ai/ai/search/indicator"


def query_indicator(text: str, stream: bool = False) -> dict | str | None:
    """
    查询经济数据指标。
    stream=False 返回完整 dict；stream=True 逐行打印并返回拼接后的内容。
    """
    if not stream:
        return post(INDICATOR_ENDPOINT, {"text": text, "stream": False}, timeout=60)

    # 流式模式
    response = post_stream(INDICATOR_ENDPOINT, {"text": text, "stream": True})
    if not response:
        return None

    try:
        content_parts = []
        for raw_line in response:
            line = raw_line.decode('utf-8').strip()
            if not line.startswith('data:'):
                continue
            payload = line[5:].strip()
            if payload == '[DONE]':
                break
            try:
                chunk = json.loads(payload)
                delta = chunk.get('choices', [{}])[0].get('delta', {})
                c = delta.get('content', '')
                if c:
                    content_parts.append(c)
                    print(c, end='', flush=True)
            except json.JSONDecodeError:
                continue
        print()  # 换行
        return ''.join(content_parts)
    finally:
        response.close()


def format_result(result: dict) -> None:
    """格式化非流式结果。"""
    if not result or result.get('code') != '000000':
        print(f"查询失败: {result.get('msg', '未知错误') if result else '无响应'}")
        return

    choices = result.get('data', {}).get('choices', [])
    if not choices:
        print("无结果。")
        return

    msg = choices[0].get('message', {})
    content = msg.get('content', '')
    reasoning = msg.get('reasoning_content', '')

    if reasoning and reasoning != content:
        print(f"[推理过程]\n{reasoning}\n")
    if content:
        print(content)


def main():
    parser = argparse.ArgumentParser(description='经济数据指标查询')
    parser.add_argument('text', help='查询内容，如 "比亚迪仰望销量"')
    parser.add_argument('--stream', action='store_true', help='流式输出')
    parser.add_argument('--json', action='store_true', help='输出原始 JSON（仅非流式）')

    args = parser.parse_args()
    require_configured()

    if args.stream:
        result = query_indicator(args.text, stream=True)
        if result is None:
            sys.exit(1)
    else:
        result = query_indicator(args.text, stream=False)
        if result:
            if args.json:
                print(json.dumps(result, indent=2, ensure_ascii=False))
            else:
                format_result(result)
        else:
            sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
查询电话会议列表。

用法:
  python3 scripts/meeting_list.py
  python3 scripts/meeting_list.py --stock 600519.SH
  python3 scripts/meeting_list.py --topic 银行 --size 20
  python3 scripts/meeting_list.py --json
"""
import json
import sys
import argparse
import time
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

from _client import require_configured, post

MEETING_ENDPOINT = "/application/open-meeting/cnfr/getList"

# 会议进程映射
PROCESS_MAP = {1: '会议中', 2: '未开始', 3: '录制成功', 4: '录制失败', 7: '上传完成'}
# 会议类型映射
METHOD_MAP = {1: '电话会议', 2: '上传会议', 3: '腾讯会议', 4: '进门财经', 5: 'Zoom'}


def query_meetings(page_num: int = 1, page_size: int = 10, stock_codes: list | None = None,
                   topic: list | None = None, days_back: int = 7) -> dict | None:
    """查询会议列表。"""
    end_time = int(time.time() * 1000)
    start_time = end_time - (days_back * 24 * 60 * 60 * 1000)

    data = {
        "pageNum": page_num,
        "pageSize": page_size,
        "startTime": start_time,
        "endTime": end_time,
        "status": 17,  # 已发布
    }
    if stock_codes:
        data["securityIdList"] = stock_codes
    if topic:
        data["topicList"] = topic

    return post(MEETING_ENDPOINT, data)


def format_results(result: dict) -> None:
    """格式化输出会议列表。"""
    if not result or result.get('code') != '000000':
        print(f"查询失败: {result.get('msg', '未知错误') if result else '无响应'}")
        return

    page_data = result.get('data', {})
    total = page_data.get('total', 0)
    items = page_data.get('list', [])

    print(f"共 {total} 条会议，当前页 {len(items)} 条")

    for item in items:
        cnfr_time = item.get('cnfrTime')
        time_str = datetime.fromtimestamp(cnfr_time / 1000).strftime('%m-%d %H:%M') if cnfr_time else 'N/A'
        process = PROCESS_MAP.get(item.get('process'), str(item.get('process', '')))
        method = METHOD_MAP.get(item.get('cnfrCreateMethod'), str(item.get('cnfrCreateMethod', '')))

        print(f"\n--- [{time_str}] {item.get('topic', 'N/A')} ---")
        print(f"  机构: {item.get('partyName', 'N/A')} | 行业: {item.get('blockName', 'N/A')} | 状态: {process}")
        print(f"  类型: {method} | 类别: {item.get('categoryStmt', 'N/A')}")

        security = item.get('securityAbbr')
        if security:
            print(f"  关联股票: {security} ({item.get('securityId', '')})")

        summary = item.get('summary')
        if summary:
            if len(summary) > 300:
                summary = summary[:300] + "..."
            print(f"  摘要: {summary}")

        live_url = item.get('liveUrl')
        if live_url:
            print(f"  直播: {live_url}")


def main():
    parser = argparse.ArgumentParser(description='查询电话会议列表')
    parser.add_argument('--stock', help='股票代码，如 600519.SH')
    parser.add_argument('--topic', help='会议主题关键词')
    parser.add_argument('--days', type=int, default=7, help='查询天数（默认 7）')
    parser.add_argument('--page', type=int, default=1, help='页码（默认 1）')
    parser.add_argument('--size', type=int, default=10, help='每页数量（默认 10）')
    parser.add_argument('--json', action='store_true', help='输出原始 JSON')

    args = parser.parse_args()
    require_configured()

    stock_codes = [args.stock] if args.stock else None
    topics = [args.topic] if args.topic else None

    result = query_meetings(args.page, args.size, stock_codes, topics, args.days)
    if result:
        if args.json:
            print(json.dumps(result, indent=2, ensure_ascii=False))
        else:
            format_results(result)
    else:
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
知识库语义搜索。

用法:
  python3 scripts/query_kb.py "比亚迪最新消息"
  python3 scripts/query_kb.py "特斯拉" --type 10,40 --top 5 --days 180
  python3 scripts/query_kb.py "宁德时代" --json
"""
import json
import sys
import argparse
import time
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

from _client import require_configured, post, BASE_URL

KNOWLEDGE_ENDPOINT = "/application/open-data/ai/search/knowledge/batch"

# 资源类型映射
RT_NAMES = {
    10: '券商研报', 20: '内部研报', 40: '分析师观点', 50: '公告',
    60: '会议纪要', 70: '调研纪要', 80: '网络资源', 90: '产业公众号',
}


def query_knowledge(queries, resource_types=None, top=20, days_back=365, knowledge_name='system_knowledge_doc'):
    """查询知识库。"""
    end_time = int(time.time() * 1000)
    start_time = end_time - (days_back * 24 * 60 * 60 * 1000)

    data = {
        "queries": queries if isinstance(queries, list) else [queries],
        "resourceTypes": resource_types or [10, 40],
        "knowledgeNames": [knowledge_name],
        "startTime": start_time,
        "endTime": end_time,
        "top": min(top, 20),
    }
    return post(KNOWLEDGE_ENDPOINT, data, timeout=60)


def format_results(result):
    """格式化输出搜索结果。"""
    if not result or result.get('code') != '000000':
        print(f"查询失败: {result.get('msg', '未知错误') if result else '无响应'}")
        return

    data = result.get('data', [])
    if not data:
        print("无结果。")
        return

    for query_result in data:
        query = query_result.get('query', '')
        items = query_result.get('data', [])

        print(f"\n{'='*60}")
        print(f"查询: {query}")
        print(f"{'='*60}")

        if not items:
            print("  无匹配结果。")
            continue

        for i, item in enumerate(items, 1):
            rt = item.get('resourceType')
            print(f"\n--- 结果 {i} ---")
            print(f"标题: {item.get('title', 'N/A')}")
            print(f"公司: {item.get('company', 'N/A')}")
            print(f"类型: {RT_NAMES.get(rt, 'Other')} ({rt})")

            ts = item.get('time')
            if ts:
                print(f"时间: {datetime.fromtimestamp(ts / 1000).strftime('%Y-%m-%d')}")

            # 溯源信息（type 40 不支持下载）
            source_id = item.get('sourceId')
            if source_id:
                print(f"SourceId: {source_id}")
                if rt and rt != 40:
                    print(f"下载: {BASE_URL}/application/open-data/ai/resource/download?resourceType={rt}&sourceId={source_id}")

            content = item.get('content', '')
            if content:
                if len(content) > 800:
                    content = content[:800] + "..."
                print(f"内容: {content}")


def main():
    parser = argparse.ArgumentParser(description='知识库语义搜索')
    parser.add_argument('query', help='查询关键词')
    parser.add_argument('--type', default='10,40', help='资源类型（逗号分隔，默认 10,40）')
    parser.add_argument('--top', type=int, default=20, help='返回数量（默认 20，最大 20）')
    parser.add_argument('--days', type=int, default=365, help='搜索天数（默认 365）')
    parser.add_argument('--json', action='store_true', help='输出原始 JSON')

    args = parser.parse_args()
    require_configured()

    resource_types = [int(t.strip()) for t in args.type.split(',')]
    result = query_knowledge(args.query, resource_types, args.top, args.days)

    if result:
        if args.json:
            print(json.dumps(result, indent=2, ensure_ascii=False))
        else:
            format_results(result)
    else:
        sys.exit(1)


if __name__ == "__main__":
    main()

capture-gangtise-kb