删除 gp/stock.py

gp/stock.py

@@ -1,187 +0,0 @@
-#!/usr/bin/env python3
-from __future__ import annotations
-import io
-import json
-import os
-import re
-import sys
-import traceback
-import urllib.error
-import urllib.request
-
-# 将 stdout 编码设置为 UTF-8，确保 emoji 等字符能正常输出
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")
-else:
-    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
-sys.stderr = sys.stdout
-
-QUOTE_API_URL = "https://api.mairuiapi.com/hsrl/ssjy/{}/{}"
-DEFAULT_LICENCE = "3E81CB37-4BCE-4DAF-B0AC-AEA069A58973"
-FALLBACK_TEXT = "股票查询失败，请稍后再试。"
-
-
-def normalize_stock_code(code: str) -> str:
-    """验证并标准化股票代码（仅保留6位纯数字）"""
-    code = code.strip()
-    # 去掉可能的前缀（sh/SH/sz/SZ/bj/BJ）
-    code = re.sub(r"^(sh|sz|bj)", "", code, flags=re.IGNORECASE)
-    if re.match(r"^\d{6}$", code):
-        return code
-    # 如果传入了完整代码但包含其他字符，尝试提取数字
-    digits = re.findall(r"\d", code)
-    if len(digits) >= 6:
-        return "".join(digits[:6])
-    return ""
-
-
-def get_licence() -> str:
-    """获取 licence，优先从环境变量读取"""
-    return os.environ.get("STOCK_LICENCE", "").strip() or DEFAULT_LICENCE
-
-
-def fetch_stock_quote(stock_code: str) -> dict | None:
-    """获取股票行情数据并解析为字典"""
-    licence = get_licence()
-    url = QUOTE_API_URL.format(stock_code, licence)
-    try:
-        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
-        with urllib.request.urlopen(req, timeout=10) as response:
-            payload = json.load(response)
-    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError):
-        return None
-
-    # 处理 JSON 数组返回格式 [{}, ...]
-    if isinstance(payload, list) and len(payload) > 0:
-        payload = payload[0]
-
-    if not isinstance(payload, dict):
-        return None
-
-    # 检查接口返回是否包含有效数据
-    if "p" not in payload:
-        return None
-
-    return payload
-
-
-def format_stock_message(data: dict, stock_code: str = "") -> str:
-    """格式化股票信息为可读文本"""
-    p = data.get("p", 0) or 0
-    pc = data.get("pc", 0) or 0
-    ud = data.get("ud", 0) or 0
-
-    up_icon = "📈" if float(pc) >= 0 else "📉"
-    title = f"【{stock_code}】" if stock_code else "【查询结果】"
-
-    lines = [
-        title,
-        f"{'=' * 20}",
-        f"{up_icon} 当前价：{_fmt(p)}",
-        f"📊 涨跌幅：{_fmt(pc)}%",
-        f"📊 涨跌额：{_fmt(ud)}",
-        f"⬆️  最高：{_fmt(data.get('h'))}",
-        f"⬇️  最低：{_fmt(data.get('l'))}",
-        f"🔓 今开：{_fmt(data.get('o'))}",
-        f"🔒 昨收：{_fmt(data.get('yc'))}",
-        f"📈 成交量：{_fmt(data.get('v'))}手",
-        f"💰 成交额：{_fmt(data.get('cje'))}",
-        f"🔄 换手率：{_fmt(data.get('hs'))}%",
-        f"📐 振幅：{_fmt(data.get('zf'))}%",
-        f"📊 市盈率：{_fmt(data.get('pe'))}",
-        f"📊 市净率：{_fmt(data.get('sjl'))}",
-        f"📊 量比：{_fmt(data.get('lb'))}",
-        f"⚡ 涨速：{_fmt(data.get('zs'))}%",
-        f"⏱ 五分钟涨跌：{_fmt(data.get('fm'))}%",
-        f"📅 60日涨跌：{_fmt(data.get('zdf60'))}%",
-        f"📅 年初至今：{_fmt(data.get('zdfnc'))}%",
-        f"📊 总市值：{_fmt(data.get('sz'))}",
-        f"📊 流通市值：{_fmt(data.get('lt'))}",
-        f"{'=' * 20}",
-        f"更新时间：{data.get('t', '未知')}",
-    ]
-    return "\n".join(lines)
-
-
-def _fmt(val) -> str:
-    """统一格式化数值，金融数据保留2位小数"""
-    if val is None:
-        return "0"
-    if isinstance(val, (int, float)):
-        return f"{val:.2f}"
-    return str(val)
-
-
-def send_text(text: str) -> bool:
-    """发送文本消息到微信机器人"""
-    robot_port = os.environ.get("ROBOT_WECHAT_CLIENT_PORT", "").strip()
-    to_wxid = os.environ.get("ROBOT_FROM_WX_ID", "").strip()
-    if not robot_port or not to_wxid:
-        return False
-
-    api_url = (
-        f"http://127.0.0.1:{robot_port}/api/v1/robot/message/send/text"
-    )
-    body = json.dumps(
-        {
-            "to_wxid": to_wxid,
-            "content": text,
-        }
-    ).encode("utf-8")
-
-    request = urllib.request.Request(
-        api_url,
-        data=body,
-        headers={"Content-Type": "application/json"},
-        method="POST",
-    )
-
-    try:
-        with urllib.request.urlopen(request, timeout=10) as response:
-            if 200 <= response.status < 300:
-                return True
-            payload = json.load(response)
-    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError):
-        return False
-
-    code = payload.get("code")
-    return code == 200 or code == 0
-
-
-def main() -> int:
-    args = sys.argv[1:]
-    if not args:
-        sys.stdout.write("请提供股票代码，例如：python3 stock.py 600519\n")
-        return 0
-
-    raw_code = args[0]
-    stock_code = normalize_stock_code(raw_code)
-    if not stock_code:
-        sys.stdout.write(FALLBACK_TEXT)
-        sys.stdout.write("\n")
-        return 0
-
-    data = fetch_stock_quote(stock_code)
-    if not data:
-        sys.stdout.write(FALLBACK_TEXT)
-        sys.stdout.write("\n")
-        return 0
-
-    message = format_stock_message(data, stock_code)
-    if send_text(message):
-        return 0
-
-    # 发送失败，输出到stdout让宿主机器人捕获
-    sys.stdout.write(message)
-    sys.stdout.write("\n")
-    return 0
-
-
-if __name__ == "__main__":
-    try:
-        raise SystemExit(main())
-    except SystemExit:
-        raise
-    except Exception:
-        traceback.print_exc(file=sys.stdout)
-        raise SystemExit(1)