# -*- coding: utf-8 -*-
"""
沪深交易所公告与数据爬虫模块

数据源: sse.com.cn（上交所）、szse.cn（深交所）
提供 IPO 日历、上市公司列表、公告搜索等功能。
可作为巨潮资讯 (cninfo.com.cn) 的补充数据源。

主要函数:
    - get_ipo_calendar()                          — 获取近期 IPO 日历
    - get_listed_companies(market="sh")           — 获取上市公司列表
    - search_announcements(keyword, market="both") — 搜索公告（跨交易所）

API:
    深交所:
      - 上市公司列表: http://www.szse.cn/api/report/ShowReport/data (POST)
        SHOWTYPE=JSON, CATALOGID=1110, TABKEY=tab1
    上交所:
      - 股票列表: http://query.sse.com.cn/security/stock/getStockListData.do (GET)
        jsonCallBack=jsonpCallback, isPagination=false, sqlId=COMMON_SSE_GP_GPLB_C
        Referer: www.sse.com.cn

注意事项:
    - 上交所 API 需要 Referer: https://www.sse.com.cn/
    - 两个交易所均可能对请求频率有限制，使用 rate_limit 控制
    - 数据更新频率: 交易日盘后更新
"""

import json
import re
import logging
from typing import Optional, Dict, Any, List

try:
    from .http_utils import http_get, http_get_json, http_post, rate_limit, get_session
except ImportError:
    from http_utils import http_get, http_get_json, http_post, rate_limit, get_session

log = logging.getLogger(__name__)

# ==================== 常量 ====================

# 深交所 API
SZSE_BASE = "http://www.szse.cn"
SZSE_LISTED_COMPANIES_URL = "http://www.szse.cn/api/report/ShowReport/data"
SZSE_IPO_CALENDAR_URL = "http://www.szse.cn/api/report/ShowReport/data"
SZSE_ANNOUNCEMENT_URL = "http://www.szse.cn/api/disc/announcement/annList"

# 上交所 API
SSE_BASE = "https://www.sse.com.cn"
SSE_STOCK_LIST_URL = "http://query.sse.com.cn/security/stock/getStockListData.do"
SSE_IPO_CALENDAR_URL = "http://query.sse.com.cn/security/stock/getStockListData.do"

# 深交所报表系统目录/标签参数（提取为常量，便于交易所改版时集中维护）
SZSE_CATALOG_IPO = "1110"        # IPO 日历目录ID
SZSE_CATALOG_LISTED = "1110"     # 上市公司列表目录ID
SZSE_CATALOG_ANNOUNCEMENT = "1800_cxda"  # 公告查询目录ID
SZSE_TABKEY = "tab1"

# 请求头
SZSE_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0.0.0 Safari/537.36"
    ),
    "Accept": "application/json, text/javascript, */*; q=0.01",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    "Referer": "http://www.szse.cn/",
    "X-Requested-With": "XMLHttpRequest",
    "Content-Type": "application/json",
}

SSE_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0.0.0 Safari/537.36"
    ),
    "Accept": "application/json, text/javascript, */*; q=0.01",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    "Referer": "https://www.sse.com.cn/",
    "X-Requested-With": "XMLHttpRequest",
}


# ==================== 安全类型转换 ====================

def _safe_float(val, default=None):
    """安全转换为浮点数"""
    if val is None or val == "" or val == "-":
        return default
    try:
        return float(val)
    except (ValueError, TypeError):
        return default


def _safe_str(val, default=""):
    """安全转换为字符串"""
    if val is None:
        return default
    return str(val)


# ==================== IPO 日历 ====================


def _get_szse_ipo_calendar() -> List[Dict[str, Any]]:
    """
    获取深交所 IPO 日历（内部函数）。
    """
    params = {
        "SHOWTYPE": "JSON",
        "CATALOGID": SZSE_CATALOG_IPO,
        "TABKEY": SZSE_TABKEY,
        "PAGENO": "1",
        "random": "0.123456789",
    }

    rate_limit(url=SZSE_BASE)
    log.info("正在获取深交所 IPO 日历...")

    try:
        resp = http_post(
            SZSE_LISTED_COMPANIES_URL,
            data=params,
            headers=SZSE_HEADERS,
            timeout=30,
        )
        if resp is None:
            log.warning("深交所 IPO 日历请求失败")
            return []

        data = resp.json()
    except Exception as e:
        log.error(f"深交所 IPO 日历解析失败: {e}")
        return []

    # 解析返回数据
    items = data if isinstance(data, list) else data.get("data", [])
    if not items:
        return []

    result = []
    for item in items:
        result.append({
            "market": "sz",
            "stock_code": _safe_str(item.get("zqdm", "")),
            "stock_name": _safe_str(item.get("zqjc", "")),
            "ipo_date": _safe_str(item.get("ssrq", "")),
            "ipo_price": _safe_float(item.get("fxjg")),
            "ipo_pe": _safe_float(item.get("fxssrpe")),
            "status": _safe_str(item.get("status", "")),
        })

    log.info(f"深交所 IPO 日历: {len(result)} 条")
    return result


def _get_sse_ipo_calendar() -> List[Dict[str, Any]]:
    """
    获取上交所 IPO 日历（内部函数）。
    上交所股票查询接口也可查到待上市/新上市股票。
    """
    params = {
        "jsonCallBack": "jsonpCallback",
        "isPagination": "false",
        "sqlId": "COMMON_SSE_GP_GPLB_C",   # 股票列表通用查询
        "pageHelp.cacheSize": "1",
        "pageHelp.beginPage": "1",
        "pageHelp.pageSize": "200",
        "pageHelp.pageNo": "1",
        "pageHelp.endPage": "1",
    }

    rate_limit(url=SSE_BASE)
    log.info("正在获取上交所 IPO 日历...")

    resp = http_get(
        SSE_IPO_CALENDAR_URL,
        params=params,
        headers=SSE_HEADERS,
        timeout=30,
    )
    if resp is None:
        log.warning("上交所 IPO 日历请求失败")
        return []

    # 上交所返回 JSONP 格式，需要提取 JSON 部分
    text = resp.text
    json_match = re.search(r'jsonpCallback\((.*)\)', text, re.DOTALL)
    if not json_match:
        json_match = re.search(r'\{.*\}', text, re.DOTALL)
        if not json_match:
            log.error("上交所 IPO 数据 JSON 提取失败")
            return []

    try:
        data = json.loads(json_match.group(1) if json_match.lastindex else json_match.group(0))
    except json.JSONDecodeError as e:
        log.error(f"上交所 IPO 数据解析失败: {e}")
        return []

    items = data.get("result", []) or data.get("pageHelp", {}).get("data", [])
    if not items:
        return []

    result = []
    for item in items:
        result.append({
            "market": "sh",
            "stock_code": _safe_str(item.get("SECURITY_CODE_A", "")),
            "stock_name": _safe_str(item.get("SECURITY_ABBR_A", "")),
            "list_date": _safe_str(item.get("LISTING_DATE", "")),
            "total_shares": _safe_float(item.get("TOTAL_SHARES")),
            "info": _safe_str(item.get("REMARK", "")),
        })

    log.info(f"上交所 IPO 日历: {len(result)} 条")
    return result


def get_ipo_calendar(market: str = "both") -> List[Dict[str, Any]]:
    """
    获取近期 IPO 日历（新上市公司信息）。

    分别从深交所和上交所获取最近上市/待上市的公司数据。

    Args:
        market: 交易所选择，可选值:
            - "sh"   : 仅上交所
            - "sz"   : 仅深交所
            - "both" : 沪深两市（默认）

    Returns:
        IPO 日历列表，每项包含:
        - market: 交易所 (sh/sz)
        - stock_code: 股票代码
        - stock_name: 股票名称
        - ipo_date / list_date: 上市日期
        失败返回空列表
    """
    results = []

    if market in ("sz", "both"):
        try:
            results.extend(_get_szse_ipo_calendar())
        except Exception as e:
            log.error(f"深交所 IPO 日历获取异常: {e}")

    if market in ("sh", "both"):
        try:
            results.extend(_get_sse_ipo_calendar())
        except Exception as e:
            log.error(f"上交所 IPO 日历获取异常: {e}")

    log.info(f"IPO 日历汇总: {len(results)} 条（market={market}）")
    return results


# ==================== 上市公司列表 ====================


def _get_szse_listed_companies() -> List[Dict[str, Any]]:
    """
    获取深交所上市公司列表（内部函数）。
    使用深交所报表接口以 POST 请求获取。
    """
    params = {
        "SHOWTYPE": "JSON",
        "CATALOGID": SZSE_CATALOG_LISTED,
        "TABKEY": SZSE_TABKEY,
        "PAGENO": "1",
        "random": "0.123456789",
    }

    rate_limit(url=SZSE_BASE)
    log.info("正在获取深交所上市公司列表...")

    try:
        resp = http_post(
            SZSE_LISTED_COMPANIES_URL,
            data=params,
            headers=SZSE_HEADERS,
            timeout=30,
        )
        if resp is None:
            log.warning("深交所上市公司列表请求失败")
            return []

        data = resp.json()
    except Exception as e:
        log.error(f"深交所上市公司列表解析失败: {e}")
        return []

    items = data if isinstance(data, list) else data.get("data", [])
    if not items:
        return []

    companies = []
    for item in items:
        company = {
            "market": "sz",
            "stock_code": _safe_str(item.get("zqdm", "")),
            "stock_name": _safe_str(item.get("zqjc", "")),
            "company_name": _safe_str(item.get("gsjc", "")),
            "industry": _safe_str(item.get("ssrq", "")),  # 上市板块
            "list_date": _safe_str(item.get("ssrq", "")),
            "total_shares": _safe_float(item.get("zgb")),
            "area": _safe_str(item.get("area", "")),
        }
        companies.append(company)

    log.info(f"深交所上市公司: {len(companies)} 家")
    return companies


def _get_sse_listed_companies() -> List[Dict[str, Any]]:
    """
    获取上交所上市公司列表（内部函数）。
    """
    params = {
        "jsonCallBack": "jsonpCallback",
        "isPagination": "false",
        "sqlId": "COMMON_SSE_GP_GPLB_C",
        "pageHelp.cacheSize": "1",
        "pageHelp.beginPage": "1",
        "pageHelp.pageSize": "2000",   # 上交所约2000家公司
        "pageHelp.pageNo": "1",
        "pageHelp.endPage": "1",
    }

    rate_limit(url=SSE_BASE)
    log.info("正在获取上交所上市公司列表...")

    resp = http_get(
        SSE_STOCK_LIST_URL,
        params=params,
        headers=SSE_HEADERS,
        timeout=30,
    )
    if resp is None:
        log.warning("上交所上市公司列表请求失败")
        return []

    text = resp.text
    json_match = re.search(r'jsonpCallback\((.*)\)', text, re.DOTALL)
    if not json_match:
        json_match = re.search(r'\{.*\}', text, re.DOTALL)
        if not json_match:
            log.error("上交所股票列表 JSON 提取失败")
            return []

    try:
        data = json.loads(json_match.group(1) if json_match.lastindex else json_match.group(0))
    except json.JSONDecodeError as e:
        log.error(f"上交所股票列表解析失败: {e}")
        return []

    items = data.get("result", []) or data.get("pageHelp", {}).get("data", [])
    if not items:
        return []

    companies = []
    for item in items:
        company = {
            "market": "sh",
            "stock_code": _safe_str(item.get("SECURITY_CODE_A", "")),
            "stock_name": _safe_str(item.get("SECURITY_ABBR_A", "")),
            "company_name": _safe_str(item.get("COMPANY_ABBR", "")),
            "list_date": _safe_str(item.get("LISTING_DATE", "")),
            "total_shares": _safe_float(item.get("TOTAL_SHARES")),
            "remark": _safe_str(item.get("REMARK", "")),
        }
        companies.append(company)

    log.info(f"上交所上市公司: {len(companies)} 家")
    return companies


def get_listed_companies(market: str = "both") -> List[Dict[str, Any]]:
    """
    获取沪深上市公司列表。

    Args:
        market: 交易所选择，可选值:
            - "sh"   : 仅上交所
            - "sz"   : 仅深交所
            - "both" : 沪深两市（默认）

    Returns:
        上市公司列表，每项包含: market, stock_code, stock_name, list_date 等
        失败返回空列表
    """
    results = []

    if market in ("sz", "both"):
        try:
            results.extend(_get_szse_listed_companies())
        except Exception as e:
            log.error(f"深交所上市公司列表获取异常: {e}")

    if market in ("sh", "both"):
        try:
            results.extend(_get_sse_listed_companies())
        except Exception as e:
            log.error(f"上交所上市公司列表获取异常: {e}")

    log.info(f"上市公司列表汇总: {len(results)} 家（market={market}）")
    return results


# ==================== 公告搜索 ====================


def _search_szse_announcements(keyword: str,
                                max_results: int = 50) -> List[Dict[str, Any]]:
    """
    搜索深交所公告（内部函数）。
    """
    params = {
        "SHOWTYPE": "JSON",
        "CATALOGID": SZSE_CATALOG_ANNOUNCEMENT,
        "TABKEY": SZSE_TABKEY,
        "txtK": keyword,            # 关键词
        "txtM": "",                 # 证券代码（可选）
        "PAGENO": "1",
        "pageSize": str(min(max_results, 50)),
    }

    rate_limit(url=SZSE_BASE)
    log.info(f"正在搜索深交所公告: keyword={keyword}")

    try:
        resp = http_post(
            SZSE_ANNOUNCEMENT_URL,
            data=params,
            headers=SZSE_HEADERS,
            timeout=30,
        )
        if resp is None:
            log.warning(f"深交所公告搜索请求失败: {keyword}")
            return []

        data = resp.json()
    except Exception as e:
        log.error(f"深交所公告搜索解析失败: {e}")
        return []

    items = data if isinstance(data, list) else data.get("data", [])
    if not items:
        return []

    results = []
    for item in items:
        ann = {
            "market": "sz",
            "title": _safe_str(item.get("title", "")
                               or item.get("gsggbt", "")
                               or item.get("announcementTitle", "")),
            "stock_code": _safe_str(item.get("stockCode", "")
                                    or item.get("zqdm", "")),
            "stock_name": _safe_str(item.get("stockName", "")
                                    or item.get("zqjc", "")),
            "publish_date": _safe_str(item.get("publishDate", "")
                                      or item.get("ggrq", "")),
            "url": _safe_str(item.get("url", "")
                             or item.get("attachPath", "")),
            "category": _safe_str(item.get("category", "")
                                  or item.get("announcementType", "")),
        }
        results.append(ann)

    log.info(f"深交所公告搜索结果: {len(results)} 条")
    return results


def _search_sse_announcements(keyword: str,
                               max_results: int = 50) -> List[Dict[str, Any]]:
    """
    搜索上交所公告（内部函数）。
    上交所公告搜索通过披露查询接口。
    """
    url = "http://query.sse.com.cn/security/stock/queryCompanyBulletin.do"
    params = {
        "jsonCallBack": "jsonpCallback",
        "isPagination": "true",
        "keyWord": keyword,
        "pageHelp.pageSize": str(min(max_results, 50)),
        "pageHelp.pageNo": "1",
        "pageHelp.beginPage": "1",
        "pageHelp.endPage": "1",
        "pageHelp.cacheSize": "1",
        "productId": "",
        "securityType": "0101",     # A股
        "reportType2": "DQGG",      # 定期公告
        "reportType": "ALL",        # 所有类型
        "beginDate": "",
        "endDate": "",
    }

    rate_limit(url=SSE_BASE)
    log.info(f"正在搜索上交所公告: keyword={keyword}")

    resp = http_get(url, params=params, headers=SSE_HEADERS, timeout=30)
    if resp is None:
        log.warning(f"上交所公告搜索请求失败: {keyword}")
        return []

    text = resp.text
    json_match = re.search(r'jsonpCallback\((.*)\)', text, re.DOTALL)
    if not json_match:
        json_match = re.search(r'\{.*\}', text, re.DOTALL)
        if not json_match:
            log.error("上交所公告 JSON 提取失败")
            return []

    try:
        data = json.loads(json_match.group(1) if json_match.lastindex else json_match.group(0))
    except json.JSONDecodeError as e:
        log.error(f"上交所公告解析失败: {e}")
        return []

    items = data.get("result", []) or data.get("pageHelp", {}).get("data", [])
    if not items:
        return []

    results = []
    for item in items:
        ann = {
            "market": "sh",
            "title": _safe_str(item.get("title", "")
                               or item.get("bulletinTitle", "")),
            "stock_code": _safe_str(item.get("securityCode", "")
                                    or item.get("stockCode", "")),
            "stock_name": _safe_str(item.get("securityAbbr", "")
                                    or item.get("stockName", "")),
            "publish_date": _safe_str(item.get("publishDate", "")
                                      or item.get("bulletinDate", "")),
            "url": _safe_str(item.get("URL", "")
                             or item.get("bulletinUrl", "")),
            "category": _safe_str(item.get("bulletinType", "")
                                  or item.get("reportType", "")),
        }
        results.append(ann)

    log.info(f"上交所公告搜索结果: {len(results)} 条")
    return results


def search_announcements(keyword: str,
                          market: str = "both",
                          max_results: int = 100) -> List[Dict[str, Any]]:
    """
    搜索沪深交易所公告（跨交易所合并结果）。

    作为巨潮资讯 (cninfo.com.cn) 的补充数据源，直接从交易所
    官方接口获取公告数据。

    Args:
        keyword: 搜索关键词（如 "分红"、"增发"、"重组" 等）
        market: 交易所选择，可选值:
            - "sh"   : 仅上交所
            - "sz"   : 仅深交所
            - "both" : 沪深两市（默认）
        max_results: 每个交易所最大返回条数（默认 100）

    Returns:
        公告列表，每项包含: market, title, stock_code, stock_name,
        publish_date, url, category
        失败返回空列表
    """
    results = []

    per_market_max = max_results if market != "both" else max_results // 2

    if market in ("sz", "both"):
        try:
            results.extend(_search_szse_announcements(keyword, per_market_max))
        except Exception as e:
            log.error(f"深交所公告搜索异常: {e}")

    if market in ("sh", "both"):
        try:
            results.extend(_search_sse_announcements(keyword, per_market_max))
        except Exception as e:
            log.error(f"上交所公告搜索异常: {e}")

    # 按发布日期降序排列
    results.sort(key=lambda x: x.get("publish_date", ""), reverse=True)

    log.info(f"公告搜索汇总: '{keyword}' 共 {len(results)} 条（market={market}）")
    return results


# ==================== 便捷查询 ====================


def get_recent_ipo_list(days: int = 30) -> List[Dict[str, Any]]:
    """
    获取最近 N 天内 IPO 的公司列表。

    Args:
        days: 最近天数（默认 30 天）

    Returns:
        近期 IPO 公司列表
    """
    from datetime import datetime, timedelta
    cutoff = datetime.now() - timedelta(days=days)
    cutoff_str = cutoff.strftime("%Y-%m-%d")

    ipo_list = get_ipo_calendar(market="both")
    recent = []

    for ipo in ipo_list:
        date_str = ipo.get("ipo_date") or ipo.get("list_date", "")
        if date_str and date_str >= cutoff_str:
            recent.append(ipo)

    recent.sort(key=lambda x: x.get("ipo_date") or x.get("list_date", ""),
                reverse=True)
    log.info(f"近 {days} 天 IPO 公司: {len(recent)} 家")
    return recent


def find_company_announcements(stock_code: str,
                                keyword: str = "",
                                market: str = "both",
                                max_results: int = 30) -> List[Dict[str, Any]]:
    """
    查找指定公司的公告。
    将公司代码作为关键词的一部分进行搜索。

    Args:
        stock_code: 股票代码（如 "600519"）
        keyword: 额外搜索关键词（可选，如 "分红"）
        market: 交易所

    Returns:
        该公司相关公告列表
    """
    search_keyword = f"{stock_code} {keyword}".strip()
    return search_announcements(
        keyword=search_keyword,
        market=market,
        max_results=max_results,
    )


# ==================== 北交所（v4.7 新增） ====================

# 北交所官网接口
BSE_BASE = "https://www.bse.cn"
BSE_LISTED_COMPANIES_API = f"{BSE_BASE}/nqxxController/nqxxCnzq.do"
BSE_IPO_AUDIT_API = f"{BSE_BASE}/nqxxController/nqxxAuditList.do"

# 东方财富北交所板块（fs 参数），作为北交所数据的可靠降级源
EM_BSE_CLIST_API = "https://push2.eastmoney.com/api/qt/clist/get"


def _fetch_bse_native(params: Dict[str, Any], api_url: str) -> Optional[str]:
    """调用北交所官网 JSONP 接口，返回剥离回调后的原始文本。"""
    try:
        import urllib.parse
        body = urllib.parse.urlencode(params)
        rate_limit(url=BSE_BASE)
        full = f"{api_url}?{body}&callback=cb"
        resp = http_get(full, headers={
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/131.0.0.0",
            "Referer": f"{BSE_BASE}/nq/listedcompany.html",
        }, timeout=20)
        if resp is None:
            return None
        text = resp.text
        # 剥离 JSONP 回调: callback([...]) 或 null([...])
        m = re.search(r"\((.*)\)\s*;?\s*$", text, re.DOTALL)
        return m.group(1) if m else text
    except Exception as e:
        log.error(f"北交所原生接口请求失败: {e}")
        return None


def _fetch_em_bse_stocks(limit: int) -> List[Dict[str, Any]]:
    """通过东方财富 clist 接口获取北交所上市公司列表（降级源）。"""
    try:
        import urllib.parse
        params = {
            "pn": "1", "pz": str(min(limit, 200)), "po": "1", "np": "1",
            "fltt": "2", "invt": "2", "fid": "f26",
            "fs": "m:0+t:81+t:82+t:83",  # 北交所
            "fields": "f12,f14,f13,f26,f100,f2",
        }
        qs = urllib.parse.urlencode(params)
        rate_limit(url="push2.eastmoney.com")
        resp = http_get(f"{EM_BSE_CLIST_API}?{qs}", headers={
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/131.0.0.0",
            "Referer": "https://quote.eastmoney.com/",
        }, timeout=20)
        if resp is None:
            return []
        data = json.loads(resp.text)
        items = data.get("data", {}).get("diff", []) if isinstance(data.get("data"), dict) else []
        results = []
        for it in items:
            results.append({
                "code": _safe_str(it.get("f12", "")),
                "name": _safe_str(it.get("f14", "")),
                "short_code": _safe_str(it.get("f13", "")),
                "listing_date": _safe_str(it.get("f26", "")),
                "industry": _safe_str(it.get("f100", "")),
                "market": "bj",
            })
        return results
    except Exception as e:
        log.error(f"东方财富北交所列表请求失败: {e}")
        return []


def get_bse_stocks(limit: int = 200) -> List[Dict[str, Any]]:
    """获取北交所上市公司列表。

    优先调用北交所官网接口，失败时降级到东方财富北交所板块数据。

    Args:
        limit: 返回条数，默认 200，最大 500

    Returns:
        [
            {"code": "837592", "name": "华信永道", "short_code": "837592",
             "listing_date": "2023-02-16", "industry": "信息技术", "market": "bj"}, ...
        ]
    """
    # 1) 北交所原生接口
    raw = _fetch_bse_native(
        {"page": 0, "typejb": "T", "xxfcbj": 2, "xxzqdm": "",
         "sortfield": "xxzqdm", "sorttype": "asc", "pagesize": min(limit, 100)},
        BSE_LISTED_COMPANIES_API,
    )
    if raw:
        try:
            data = json.loads(raw)
            if isinstance(data, list) and data and isinstance(data[0], dict) and "msg" not in data[0]:
                results = []
                for it in data[:limit]:
                    results.append({
                        "code": _safe_str(it.get("xxzqdm", it.get("zqdm", ""))),
                        "name": _safe_str(it.get("jc", it.get("zqjc", ""))),
                        "short_code": _safe_str(it.get("zqdm", "")),
                        "listing_date": _safe_str(it.get("ssrq", "")),
                        "industry": _safe_str(it.get("hy", it.get("hymc", ""))),
                        "market": "bj",
                    })
                if results:
                    log.info(f"北交所原生接口获取 {len(results)} 条")
                    return results
        except Exception as e:
            log.warning(f"北交所原生接口解析失败，降级东方财富: {e}")

    # 2) 降级：东方财富北交所板块
    results = _fetch_em_bse_stocks(limit)
    log.info(f"北交所列表获取 {len(results)} 条（东方财富降级源）")
    return results


def get_bse_ipo(limit: int = 50) -> List[Dict[str, Any]]:
    """获取北交所 IPO 审核/发行信息。

    Args:
        limit: 返回条数，默认 50

    Returns:
        [
            {"code": "...", "name": "...", "status": "已注册",
             "audit_date": "2026-07-31", "market": "bj"}, ...
        ]
    """
    raw = _fetch_bse_native(
        {"page": 0, "xxfcbj": 2, "typejb": "T"},
        BSE_IPO_AUDIT_API,
    )
    if raw:
        try:
            data = json.loads(raw)
            if isinstance(data, list) and data and isinstance(data[0], dict) and "msg" not in data[0]:
                results = []
                for it in data[:limit]:
                    results.append({
                        "code": _safe_str(it.get("xxzqdm", it.get("zqdm", ""))),
                        "name": _safe_str(it.get("jc", it.get("zqjc", ""))),
                        "status": _safe_str(it.get("zt", it.get("status", ""))),
                        "audit_date": _safe_str(it.get("sxrq", it.get("ssrq", ""))),
                        "market": "bj",
                    })
                if results:
                    log.info(f"北交所 IPO 审核获取 {len(results)} 条")
                    return results
        except Exception as e:
            log.warning(f"北交所 IPO 审核解析失败: {e}")

    # 降级：东方财富 IPO 日历（北交所部分由 get_ipo_calendar 覆盖）
    ipo = get_ipo_calendar(market="both")
    results = [i for i in ipo if str(i.get("stock_code", "")).startswith(("8", "4"))]
    log.info(f"北交所 IPO 获取 {len(results)} 条（降级源）")
    return results[:limit]


# ==================== 测试入口 ====================

if __name__ == "__main__":
    print("=" * 60)
    print("沪深交易所数据爬虫 - 功能测试")
    print("=" * 60)

    # 测试1: 获取上市公司列表
    print("\n[测试1] 获取上交所上市公司列表（前10家）...")
    companies = get_listed_companies(market="sh")
    if companies:
        print(f"共获取 {len(companies)} 家上交所上市公司")
        for c in companies[:10]:
            print(f"  {c.get('market')} {c.get('stock_code')} {c.get('stock_name')} "
                  f"上市日:{c.get('list_date', 'N/A')}")
    else:
        print("获取失败（上交所接口可能需要特定 Referer）")

    # 测试2: 获取深交所上市公司列表
    print("\n[测试2] 获取深交所上市公司列表（前10家）...")
    sz_companies = get_listed_companies(market="sz")
    if sz_companies:
        print(f"共获取 {len(sz_companies)} 家深交所上市公司")
        for c in sz_companies[:10]:
            print(f"  {c.get('market')} {c.get('stock_code')} {c.get('stock_name')}")
    else:
        print("获取失败")

    # 测试3: 搜索公告
    print("\n[测试3] 搜索 '分红' 相关公告（深交所）...")
    anns = search_announcements(keyword="分红", market="sz", max_results=10)
    if anns:
        print(f"搜索到 {len(anns)} 条公告")
        for ann in anns[:5]:
            print(f"  [{ann.get('market')}] {ann.get('stock_code')} "
                  f"{ann.get('title', '')[:60]} "
                  f"日期:{ann.get('publish_date', '')}")
    else:
        print("搜索无结果或接口不可用")

    # 测试4: 搜索 '增发' 公告（上交所）
    print("\n[测试4] 搜索 '增发' 相关公告（上交所）...")
    anns_sse = search_announcements(keyword="增发", market="sh", max_results=10)
    if anns_sse:
        print(f"搜索到 {len(anns_sse)} 条公告")
        for ann in anns_sse[:5]:
            print(f"  [{ann.get('market')}] {ann.get('stock_code')} "
                  f"{ann.get('title', '')[:60]}")
    else:
        print("搜索无结果或接口不可用")

    # 测试5: 查找指定公司公告
    print("\n[测试5] 查找贵州茅台(600519)近期公告...")
    moutai_anns = find_company_announcements("600519", keyword="分红")
    if moutai_anns:
        for ann in moutai_anns[:5]:
            print(f"  [{ann.get('market')}] {ann.get('title', '')[:60]} "
                  f"日期:{ann.get('publish_date', '')}")
    else:
        print("未找到相关公告或接口不可用")

    print("\n" + "=" * 60)
    print("测试完成")
    print("提示: 交易所接口可能有反爬限制，如测试失败属正常现象。")
    print("      实际使用时建议在工作日盘后时段调用。")