IntelligenceHubWebService.py

# IntelligenceHubWebService.py

import os
import re
import json
import logging
import datetime
import time
import dateutil
import threading
import traceback
from functools import wraps
from typing import List, Tuple, Any, Dict, Optional
from dateutil import parser as date_parser
from flask import Flask, request, jsonify, session, redirect, url_for, render_template, abort, send_file, \
    make_response, Response

from GlobalConfig import *
from MyPythonUtility.DictTools import DictPrinter
from prompts_v2x import ANALYSIS_PROMPT_TABLE as PROMPT_TABLE_V2
from Tools.CommonPost import common_post
from Tools.RequestTracer import RequestTracer
from Tools.DateTimeUtility import get_aware_time, ensure_timezone_aware, time_str_to_datetime
from ServiceComponent.UserManager import UserManager
from MyPythonUtility.ArbitraryRPC import RPCService
from ServiceComponent.RSSPublisher import RSSPublisher, FeedItem
from ServiceComponent.PostManager import generate_html_from_markdown
from ServiceComponent.IntelligenceDistributionPageRender import get_intelligence_statistics_page
from ServiceComponent.IntelligenceHubDefines_v2 import APPENDIX_VECTOR_SCORE, APPENDIX_TOTAL_SCORE
from ServiceComponent.RateStatisticsPageRender import get_statistics_page
from ServiceComponent.IntelligenceVectorDBEngine import IntelligenceVectorDBEngine
from IntelligenceHub import CollectedData, IntelligenceHub, ProcessedData, APPENDIX_TIME_ARCHIVED
from Tools.PerformanceLogger import get_performance_logger, PerformanceLogger
from Tools.RateLimiter import SlidingWindowRateLimiter, TimedSemaphore

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

from distutils.util import strtobool


VECTOR_MAX_TOP_N = 50
VECTOR_DEFAULT_SCORE_THRESHOLD = 0.6

# 未登录用户默认限制（可在构造时通过 public_search_limits 覆盖）
DEFAULT_PUBLIC_SEARCH_LIMITS = {
    "mongo_max_per_page": 20,
    "mongo_max_page": 5,
    "mongo_default_window_days": 30,
    "vector_max_per_page": 10,
    "vector_max_top_n": 20,
    "vector_max_page": 2,
    "vector_allow_fulltext": False,
    "vector_allow_similar": False,
    "vector_min_score_threshold": 0.6,
    "vector_default_window_days": 30,
    "requests_per_minute": 10,
    "max_concurrent_vector": 5,
}


def to_bool(value, default=False):
    """Safely convert various types to boolean.

    Handles:
    - Native bool values
    - Strings: 'true', 'false', 'yes', 'no', 'on', 'off', '1', '0'
    - Integers: 1 = True, 0 = False
    - None: returns default

    Args:
        value: Value to convert
        default: Default if conversion fails

    Returns:
        Converted boolean value
    """
    if value is None:
        return default

    if isinstance(value, bool):
        return value

    if isinstance(value, int):
        return bool(value)

    try:
        # Handle string representations
        return bool(strtobool(str(value).strip().lower()))
    except (ValueError, AttributeError):
        return default


# def exclude_raw_data(result: List[dict]):
#     summary_result = []
#     for data in result:
#         # In v2, we extract those fields to ArchivedDataExtraFields
#         _uuid = data.pop('UUID', None)
#         appendix = data.pop('APPENDIX', None)
#         informant = data.pop('INFORMANT', None)
#
#         # Compatible with v1 analysis result
#         if 'TAXONOMY' not in data:
#             data['TAXONOMY'] = 'N/A'
#
#         clean_data = ProcessedData.model_validate(data).model_dump(exclude_unset=True, exclude_none=True)
#
#         if _uuid: clean_data['UUID'] = _uuid
#         if appendix: clean_data['APPENDIX'] = appendix
#         if informant: clean_data['INFORMANT'] = informant
#
#         summary_result.append(clean_data)
#     return summary_result


# 获取模型中定义的所有合法字段名集合
VALID_FIELDS = set(ProcessedData.model_fields.keys())

def exclude_raw_data(result: List[dict]):
    summary_result = []
    for data in result:
        _uuid = data.pop('UUID', None)
        appendix = data.pop('APPENDIX', None)
        informant = data.pop('INFORMANT', None)

        if 'TAXONOMY' not in data:
            data['TAXONOMY'] = 'N/A'

        # 核心清洗逻辑：只保留定义在 VALID_FIELDS 中，并且值不为 None 的数据
        # 这一步完全等价于 model_dump(exclude_unset=True, exclude_none=True) 的清洗效果，且0报错风险
        clean_data = {
            k: v for k, v in data.items()
            if k in VALID_FIELDS and v is not None
        }

        # 补回前面抽出的特殊字段
        if _uuid: clean_data['UUID'] = _uuid
        if appendix: clean_data['APPENDIX'] = appendix
        if informant: clean_data['INFORMANT'] = informant

        summary_result.append(clean_data)
    return summary_result


def normalize_prompt_version(raw: str) -> int:
    """
    从 raw 中提取第一个连续数字段作为版本号（int）。
    e.g. 'v20' -> 20, 'prompt_v20-beta' -> 20, '20' -> 20
    """
    if raw is None:
        raise ValueError("prompt_version is required")

    m = re.search(r'(\d+)', str(raw))
    if not m:
        raise ValueError(f"invalid prompt_version: {raw}")

    return int(m.group(1))


def post_collected_intelligence(url: str, data: CollectedData, timeout=10) -> dict:
    """
    Post collected intelligence to IntelligenceHub (/collect).
    :param url: IntelligenceHub url (without '/collect' path).
    :param data: Collector data.
    :param timeout: Timeout in second
    :return: Requests response or {'status': 'error', 'reason': 'error description'}
    """
    if not isinstance(data, CollectedData):
        return {'status': 'error', 'reason': 'Data must be CollectedData format.'}
    return common_post(f'{url}/collect', data.model_dump(exclude_unset=True), timeout)


def post_processed_intelligence(url: str, data: ProcessedData, timeout=10) -> dict:
    """
    Post processed data to IntelligenceHub (/processed).
    :param url: IntelligenceHub url (without '/processed' path).
    :param data: Processed data.
    :param timeout: Timeout in second
    :return: Requests response or {'status': 'error', 'reason': 'error description'}
    """
    if not isinstance(data, ProcessedData):
        return {'status': 'error', 'reason': 'Data must be ProcessedData format.'}
    return common_post(f'{url}/processed', data.model_dump(exclude_unset=True), timeout)


class WebServiceAccessManager:
    def __init__(self,
                 rpc_api_tokens: List[str],
                 collector_tokens: List[str],
                 processor_tokens: List[str],
                 user_manager: UserManager,
                 deny_on_empty_config: bool = False):
        self.rpc_api_tokens = rpc_api_tokens
        self.collector_tokens = collector_tokens
        self.processor_tokens = processor_tokens
        self.user_manager = user_manager
        self.deny_on_empty_config = deny_on_empty_config

    def check_rpc_api_token(self, token: str) -> bool:
        return (not self.deny_on_empty_config) if not self.rpc_api_tokens else (token in self.rpc_api_tokens)

    def check_collector_token(self, token: str) -> bool:
        return (not self.deny_on_empty_config) if not self.rpc_api_tokens else (token in self.collector_tokens)

    def check_processor_token(self, token: str) -> bool:
        return (not self.deny_on_empty_config) if not self.rpc_api_tokens else (token in self.processor_tokens)

    def check_user_credential(self, username: str, password: str, client_ip) -> int or None:
        if self.user_manager:
            result, _ = self.user_manager.authenticate(username, password, client_ip)
            return result
        else:
            return 1 if not self.deny_on_empty_config else None

    @staticmethod
    def login_required(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if 'logged_in' not in session or not session['logged_in']:
                return redirect(url_for('login'))
            return f(*args, **kwargs)
        return decorated_function


class IntelligenceHubWebService:
    def __init__(self, *,
                 intelligence_hub: IntelligenceHub,
                 access_manager: WebServiceAccessManager,
                 rss_publisher: RSSPublisher,
                 public_search_limits: Optional[Dict[str, Any]] = None,
                 monitor_api=None):

        # ---------------- Parameters ----------------

        self.intelligence_hub = intelligence_hub
        self.cluster_cache_ttl_sec = 15.0
        self._cluster_response_cache = {}
        self._cluster_response_cache_lock = threading.RLock()
        self.access_manager = access_manager
        self.rss_publisher = rss_publisher
        self.wsgi_app = None
        self.monitor_api = monitor_api

        # ---------------- Public Search Limits ----------------

        self.public_search_limits = dict(DEFAULT_PUBLIC_SEARCH_LIMITS)
        if public_search_limits:
            self.public_search_limits.update(public_search_limits)

        self._vector_search_rate_limiter = SlidingWindowRateLimiter(
            window_sec=60.0,
            max_requests=self.public_search_limits.get("requests_per_minute", 10)
        )
        self._vector_search_concurrency = TimedSemaphore(
            self.public_search_limits.get("max_concurrent_vector", 5)
        )
        self._perf_logger: PerformanceLogger = get_performance_logger()

        # ---------------- RPC Service ----------------

        self.rpc_service = RPCService(
            rpc_stub=self.intelligence_hub,
            token_checker=self.access_manager.check_rpc_api_token,
            error_handler=self.handle_error
        )

        # ------------- Other Components -------------

        self.request_tracer = None
        threading.Timer(30.0, self.dump_request_connection_periodically).start()

    def set_monitor_api(self, monitor_api):
        """Set the MonitorAPI instance so dashboard can reuse system/process/thread stats."""
        self.monitor_api = monitor_api

    @staticmethod
    def _serialize_for_json(obj):
        """Recursively convert Enum (and other non-JSON-serializable) values to JSON-safe types."""
        from enum import Enum
        if isinstance(obj, Enum):
            return obj.value
        if isinstance(obj, dict):
            return {k: IntelligenceHubWebService._serialize_for_json(v) for k, v in obj.items()}
        if isinstance(obj, (list, tuple)):
            return [IntelligenceHubWebService._serialize_for_json(v) for v in obj]
        if isinstance(obj, datetime.datetime):
            return obj.isoformat()
        return obj

    # ---------------------------------------------------- Routers -----------------------------------------------------

    def register_routers(self, app: Flask):

        self.wsgi_app = app
        self.request_tracer = RequestTracer(app)

        # --------------------------------------------------- Config --------------------------------------------------

        class CustomJSONEncoder(json.JSONEncoder):
            def default(self, obj):
                if isinstance(obj, datetime.datetime):
                    return obj.strftime("%Y-%m-%d %H:%M:%S")
                # TODO: Add more data type support.
                return super().default(obj)

        app.json_encoder = CustomJSONEncoder

        # -------------------------------------------------- Security --------------------------------------------------

        @app.before_request
        def refresh_session():
            session.modified = True

        @app.route('/login', methods=['GET', 'POST'])
        def login():
            if request.method == 'POST':
                client_ip = (request.headers.get('X-Forwarded-For', '').split(',')[0].strip() or
                             request.headers.get('X-Real-IP', '').strip() or
                             request.remote_addr)

                username = request.form['username']
                password = request.form['password']

                user_id = self.access_manager.check_user_credential(username, password, client_ip)

                if user_id:
                    session['logged_in'] = True
                    session['user_id'] = user_id
                    session['username'] = username
                    session['login_ip'] = client_ip
                    session.permanent = True
                    return redirect(url_for('show_post', article='index'))
                else:
                    logger.info(f"Login fail - IP: {client_ip}, Username: {username}")
                    return "Invalid credentials", 401
            return render_template('login.html')

        @app.route('/logout')
        @WebServiceAccessManager.login_required
        def logout():
            session.clear()
            return redirect(url_for('login'))

        @app.route('/auth_check', methods=['GET'])
        def auth_check():
            ok = session.get("logged_in") is True and session.get("user_id") is not None
            status = 204 if ok else 401
            resp = make_response(("", status))
            resp.headers["Cache-Control"] = "no-store"
            return resp

        # ---------------------------------------------- Post and Article ----------------------------------------------

        @app.route('/')
        def index():
            return redirect(url_for('show_post', article='index')) \
                if session.get('logged_in') \
                else self.get_rendered_md_post('index_public') or abort(404)

        @app.route('/post/<path:article>')
        @WebServiceAccessManager.login_required
        def show_post(article):
            """
            Render a Markdown article as HTML with caching mechanism.

            Args:
                article: URL path of the Markdown file (relative to 'posts' directory)

            Returns:
                Rendered HTML template or 404 error
            """
            return self.get_rendered_md_post(article) or abort(404)

        # -------------------------------------------- API and Open Service --------------------------------------------

        @app.route('/api', methods=['POST'])
        @WebServiceAccessManager.login_required
        def rpc_api():
            try:
                response = self.rpc_service.handle_flask_request(request)
                return response
            except Exception as e:
                    print('/api Error', e)
                    print(traceback.format_exc())
                    response = ''
            return response

        @app.route('/collect', methods=['POST'])
        def collect_api():
            try:
                data = dict(request.json)
                if not data.get('UUID', ''):
                    raise ValueError('Invalid UUID.')

                collector_token = data.get('token', '')
                if self.access_manager.check_collector_token(collector_token):
                    result = self.intelligence_hub.submit_collected_data(data)
                    response = 'queued' if result else 'error',
                else:
                    response = 'invalid token'
                    logger.warning(f'Post intelligence with invalid token: {collector_token}.')

                return jsonify({
                    'resp': response,
                    'uuid': data.get('UUID', '')
                })
            except Exception as e:
                logger.error(f'collect_api() fail: {str(e)}')
                return jsonify({'resp': 'error', 'uuid': ''})

        @app.get("/api/prompts/<prompt_version>")
        def get_prompt(prompt_version):
            if not prompt_version:
                abort(400, "prompt_version required")

            try:
                version_digit = normalize_prompt_version(prompt_version)
            except ValueError as e:
                abort(400, str(e))

            prompt_text = PROMPT_TABLE_V2.get(version_digit)
            if prompt_text is None:
                prompt_text = f"[Prompt v{version_digit}] Not configured."

            resp = Response(prompt_text, content_type="text/plain; charset=utf-8")
            resp.headers["X-Prompt-Version-Normalized"] = str(version_digit)
            return resp

        @app.route('/api/intelligence/<string:intelligence_uuid>', methods=['GET'])
        def intelligence_viewer_json(intelligence_uuid: str):
            try:
                intelligence = self.intelligence_hub.get_intelligence(intelligence_uuid)
                if not intelligence:
                    return jsonify({"error": "Intelligence not found"}), 404

                return jsonify({
                    "success": True,
                    "data": intelligence
                }), 200
            except Exception as e:
                print(str(e))
                traceback.print_exc()
                return jsonify({"error": "Server error"}), 500

        @app.route('/manual_rate', methods=['POST'])
        def submit_rating():
            try:
                data = request.get_json()
                _uuid = data.get('uuid')
                ratings = data.get('ratings')

                self.intelligence_hub.submit_intelligence_manual_rating(_uuid, ratings)

                return jsonify({'status': 'success', 'message': 'Ratings saved'})
            except Exception as e:
                return jsonify({'status': 'error', 'message': str(e)}), 500

        # ---------------------------------------------------- Pages ---------------------------------------------------

        # @app.route('/rssfeed.xml', methods=['GET'])
        # def rssfeed_api():
        #     try:
        #         count = request.args.get('count', default=100, type=int)
        #         threshold = request.args.get('threshold', default=6, type=int)
        #
        #         intelligences, _ = self.intelligence_hub.query_intelligence(
        #             threshold = threshold, skip = 0, limit = count)
        #
        #         try:
        #             rss_items = self._articles_to_rss_items(intelligences)
        #             feed_xml = self.rss_publisher.generate_feed(
        #                 'IIS',
        #                 '/intelligence',
        #                 'IIS Processed Intelligence',
        #                 rss_items)
        #             return feed_xml
        #         except Exception as e:
        #             logger.error(f"Rss Feed API error: {str(e)}", stack_info=True)
        #             return 'Error'
        #     except Exception as e:
        #         logger.error(f'rssfeed_api() error: {str(e)}', stack_info=True)
        #         return 'Error'

        @app.route('/intelligences', methods=['GET'])
        def intelligences_view():
            return render_template('intelligence_list.html')

        @app.route('/recommendations', methods=['GET'])
        def intelligences_recommendations_page():
            # recommendations = self.intelligence_hub.get_recommendations()
            # return default_article_list_render(
            #     recommendations, offset=0, count=len(recommendations), total_count=len(recommendations))
            return ''

        @app.route('/intelligences/clusters', methods=['GET'])
        def intelligences_clusters_view():
            return render_template('intelligence_cluster_list.html')

        @app.route('/intelligences/search', methods=['GET'])
        def intelligences_search_page():
            is_public = not session.get('logged_in', False)
            return render_template(
                'intelligence_search.html',
                public_mode=is_public,
                public_limits=self.public_search_limits if is_public else {}
            )

        @app.route('/intelligence/graph/view', methods=['GET'])
        @WebServiceAccessManager.login_required
        def intelligence_graph_page():
            # 如果用户通过 URL 参数带了 UUID (例如从详情页跳转过来)，直接传给前端
            seed_uuid = request.args.get('uuid', '')
            return render_template('intelligence_graph.html', seed_uuid=seed_uuid)

        # ----------------------------------------------------------------------------------------

        def _get_client_ip() -> str:
            """从请求头中提取真实客户端 IP。"""
            return (
                request.headers.get('X-Forwarded-For', '').split(',')[0].strip()
                or request.headers.get('X-Real-IP', '').strip()
                or request.remote_addr
                or 'unknown'
            )

        @app.route('/intelligences/query', methods=['GET', 'POST'])
        def intelligences_query_api():
            client_ip = _get_client_ip()
            is_logged_in = bool(session.get('logged_in', False))
            is_public = not is_logged_in
            perf_extra = {
                'client_ip': client_ip,
                'is_public': is_public,
                'path': '/intelligences/query',
            }

            try:
                # 1. 获取参数
                params = _get_combined_params()
                perf_extra['search_mode'] = params.get('search_mode', 'mongo')
                perf_extra['page'] = params.get('page', 1)
                perf_extra['per_page'] = params.get('per_page', 10)

                # 2. 应用未登录用户限制
                if is_public:
                    limit_err = _apply_public_search_limits(params, client_ip)
                    if limit_err:
                        self._perf_logger.record(
                            'search_query_rejected',
                            status='rejected',
                            error=limit_err,
                            extra=perf_extra,
                        )
                        return jsonify({'error': limit_err}), 429

                # 3. 全局向量搜索并发控制（仅对未登录用户生效）
                sem_acquired = False
                if is_public and params['search_mode'].startswith('vector'):
                    if not self._vector_search_concurrency.acquire(timeout=0):
                        self._perf_logger.record(
                            'search_query_rejected',
                            status='rejected',
                            error='Too many concurrent vector searches, please retry later.',
                            extra=perf_extra,
                        )
                        return jsonify({
                            'error': '服务器当前向量搜索压力过大，请稍后再试。',
                            'retry_after': 5,
                        }), 503
                    sem_acquired = True

                try:
                    # 4. 执行业务逻辑并记录性能
                    with self._perf_logger.timed('search_query', **perf_extra):
                        data = _perform_search_logic(params)

                    perf_extra['result_count'] = len(data.get('results', []))
                    perf_extra['total'] = data.get('total', 0)
                    return jsonify(data)

                finally:
                    if sem_acquired:
                        self._vector_search_concurrency.release()

            except Exception as e:
                logger.exception("intelligences_query_api error")
                self._perf_logger.record(
                    'search_query',
                    status='error',
                    error=str(e),
                    extra=perf_extra,
                )
                return jsonify({'error': str(e)}), 500

        def _apply_public_search_limits(p: Dict[str, Any], client_ip: str) -> Optional[str]:
            """
            对未登录用户应用搜索限制。返回错误信息或 None。

            游客模式统一仅允许按 Archive Time（归档时间）搜索，忽略 Publish Time，
            避免 Publish Time 与 Archive Time 双重过滤导致结果为空。
            """
            limits = self.public_search_limits
            mode = p.get('search_mode', 'mongo')

            # 速率限制（仅向量搜索）
            if mode.startswith('vector'):
                if not self._vector_search_rate_limiter.is_allowed(client_ip):
                    return (
                        f"未登录用户向量搜索过于频繁，"
                        f"每分钟最多 {limits.get('requests_per_minute', 10)} 次，请稍后再试。"
                    )

            # 游客模式禁止使用 Publish Time，统一使用 Archive Time
            p['start_time'] = ''
            p['end_time'] = ''

            # Mongo 模式限制
            if mode == 'mongo':
                max_per_page = limits.get('mongo_max_per_page', 20)
                if p.get('per_page', 10) > max_per_page:
                    p['per_page'] = max_per_page

                max_page = limits.get('mongo_max_page', 5)
                if p.get('page', 1) > max_page:
                    return f"未登录用户普通搜索仅支持前 {max_page} 页，请登录后查看更多。"

                # 未登录用户强制最近 N 天归档数据
                if not p.get('archive_start_time') or not p.get('archive_end_time'):
                    window_days = limits.get('mongo_default_window_days', 30)
                    now = get_aware_time()
                    start = now - datetime.timedelta(days=window_days)
                    p['archive_start_time'] = start.strftime('%Y-%m-%dT%H:%M:%S')
                    p['archive_end_time'] = now.strftime('%Y-%m-%dT%H:%M:%S')

            # 向量模式限制
            if mode.startswith('vector'):
                # 禁止相似推荐
                if mode == 'vector_similar' and not limits.get('vector_allow_similar', False):
                    return "未登录用户暂不支持相似推荐，请登录后使用。"

                # 禁止全文库
                if p.get('in_fulltext') and not limits.get('vector_allow_fulltext', False):
                    p['in_fulltext'] = False
                    p['in_summary'] = True

                # 每页/最大召回限制
                max_per_page = limits.get('vector_max_per_page', 10)
                max_top_n = limits.get('vector_max_top_n', 20)
                max_page = limits.get('vector_max_page', 2)

                if p.get('per_page', 10) > max_per_page:
                    p['per_page'] = max_per_page
                if p.get('page', 1) > max_page:
                    return f"未登录用户向量搜索仅支持前 {max_page} 页，请登录后查看更多。"

                # 强制最小相似度阈值
                min_threshold = limits.get('vector_min_score_threshold', 0.6)
                if p.get('score_threshold_min', 0) < min_threshold:
                    p['score_threshold_min'] = min_threshold

                # 限制召回总量
                requested_top_n = min(p['page'] * p['per_page'], VECTOR_MAX_TOP_N)
                p['_effective_top_n'] = min(requested_top_n, max_top_n)

                # 未登录用户强制最近 N 天归档数据
                if not p.get('archive_start_time') or not p.get('archive_end_time'):
                    window_days = limits.get('vector_default_window_days', 30)
                    now = get_aware_time()
                    start = now - datetime.timedelta(days=window_days)
                    p['archive_start_time'] = start.strftime('%Y-%m-%dT%H:%M:%S')
                    p['archive_end_time'] = now.strftime('%Y-%m-%dT%H:%M:%S')

            return None

        def _get_combined_params() -> Dict[str, Any]:
            """
            统一参数解析与清洗器。
            优先级：URL Query (GET) > JSON Body > Form Data。

            Returns:
                Dict[str, Any]: 包含清洗和类型转换后的参数字典。

            Param Details:
                -------------------- 模式选择 (Mode) --------------------
                search_mode (str): 搜索策略，默认为 'mongo'。
                    - 'mongo': [普通搜索] 仅使用 MongoDB 字段精准/模糊筛选。
                    - 'vector_text': [文本向量] 根据 'keywords' 进行自然语言语义搜索。
                    - 'vector_similar': [相似推荐] 根据 'reference' (UUID) 寻找内容相似的文章。

                --------------------- 通用 (General) ---------------------
                page (int): 页码，默认为 1。
                per_page (int): 每页数量，默认为 10 (最大限制 100)。
                keywords (str): 搜索文本/关键词。
                    - Mongo模式: 视底层实现用于文本匹配。
                    - Vector模式: 作为语义搜索的 Embedding 输入。
                start_time (str): 时间下限 (ISO 8601 格式字符串)。
                end_time (str): 时间上限 (ISO 8601 格式字符串)。
                    - Mongo模式: 对应数据库中的 period 或 archived_time。
                    - Vector模式: 对应向量库 metadata 中的 timestamp。
                threshold (float): 通用评分/过滤阈值，默认为 0。

                --------------------- Mongo 专属 ------------------------
                peoples (List[str]): 人物实体列表 (逗号分隔自动转列表)。
                locations (List[str]): 地点实体列表。
                organizations (List[str]): 组织实体列表。
                    * 逻辑说明: 单字段内为 OR 关系 (包含任一即匹配)，字段间为 AND 关系。

                --------------------- Vector 专属 -----------------------
                in_summary (bool): 是否在摘要库中召回，默认为 True。
                in_fulltext (bool): 是否在全文库中召回，默认为 False。
                score_threshold (float): 向量相似度截断阈值，默认为 0.5。
                reference (str): 目标文章 UUID。仅在 'vector_similar' 模式下必填。
                """
            combined = {}

            # 1. 基础数据源获取
            # 如果是 POST，尝试获取 Body
            if request.method == 'POST':
                json_data = request.get_json(silent=True)
                if json_data:
                    combined.update(json_data)
                else:
                    combined.update(request.form)

            # 2. URL 参数覆盖 (优先级最高，保证分享链接的有效性)
            # request.args 是 ImmutableMultiDict，转为 dict
            combined.update(request.args.to_dict())

            def _split(v: Any) -> List[str]:
                if not v: return []
                if isinstance(v, list): return v
                return [x.strip() for x in v.split(',') if x.strip()]

            # 3. 参数构造
            # 兼容旧版 threshold / score_threshold
            threshold_min = float(combined.get('threshold_min', combined.get('threshold', 0)))
            threshold_max = float(combined.get('threshold_max', 10))
            score_threshold_min = float(combined.get('score_threshold_min', combined.get('score_threshold', VECTOR_DEFAULT_SCORE_THRESHOLD)))
            score_threshold_max = float(combined.get('score_threshold_max', 1.0))

            params = {
                'search_mode': combined.get('search_mode', 'mongo'),
                'page': int(combined.get('page', 1)),
                'per_page': int(combined.get('per_page', 10)),
                # 保持原始字符串，在具体逻辑中再转 datetime，避免在此处 crash
                'start_time': combined.get('start_time', ''),
                'end_time': combined.get('end_time', ''),
                'archive_start_time': combined.get('archive_start_time', ''),
                'archive_end_time': combined.get('archive_end_time', ''),
                'keywords': combined.get('keywords', '').strip(),  # 去除首尾空格

                # Mongo 筛选字段
                'peoples': _split(combined.get('peoples', '')),
                'locations': _split(combined.get('locations', '')),
                'organizations': _split(combined.get('organizations', '')),
                'geography': _split(combined.get('geography', '')),
                'informant_domains': _split(combined.get('informant_domains', '')),
                'threshold_min': threshold_min,
                'threshold_max': threshold_max,

                # Vector 字段
                'in_summary': to_bool(combined.get('in_summary'), default=True),
                'in_fulltext': to_bool(combined.get('in_fulltext'), default=False),
                'score_threshold_min': score_threshold_min,
                'score_threshold_max': score_threshold_max,
                'reference': combined.get('reference', ''),
            }

            # 限制每页最大数量，防止恶意攻击
            if params['per_page'] > 100:
                params['per_page'] = 100

            return params

        def _perform_search_logic(params: Dict[str, Any]) -> Dict[str, Any]:
            mode = params['search_mode']

            if mode.startswith('vector'):
                results, total = _do_vector_search(params)
            else:
                results, total = _do_mongo_search(params)

            if not results:
                return {'results': [], 'total': 0}
            else:
                summary_result = exclude_raw_data(results)
                return {'results': summary_result, 'total': total}

        def _do_mongo_search(p: dict) -> Tuple[List[dict], int]:
            """走 Mongo 过滤"""
            query = {}
            if p['start_time'] and p['end_time']:
                query['period'] = (
                    datetime.datetime.fromisoformat(p['start_time']),
                    datetime.datetime.fromisoformat(p['end_time'])
                )
            if p['archive_start_time'] and p['archive_end_time']:
                query['archive_period'] = (
                    datetime.datetime.fromisoformat(p['archive_start_time']),
                    datetime.datetime.fromisoformat(p['archive_end_time'])
                )
            for field in ('locations', 'peoples', 'organizations', 'geography', 'keywords', 'informant_domains'):
                if p[field]:
                    query[field] = p[field]

            if p.get('threshold_min', 0) > 0:
                query['threshold'] = p['threshold_min']
            if p.get('threshold_max', 10) < 10:
                query['threshold_max'] = p['threshold_max']

            skip = (p['page'] - 1) * p['per_page']
            return self.intelligence_hub.query_intelligence(
                skip=skip, limit=p['per_page'], **query)

        def _do_vector_search(p: dict) -> Tuple[List[dict], int]:
            """走向量召回 + 内存分页"""

            text = ''
            if p['search_mode'] == 'vector_text':
                text = p.get('keywords', '')
            elif p['search_mode'] == 'vector_similar':
                if ref_uuids := p.get('reference', ''):
                    intelligence = self.intelligence_hub.get_intelligence(ref_uuids)
                    if intelligence:
                        text = IntelligenceVectorDBEngine.build_search_text(intelligence, 'summary')
            if not text:
                return [], 0

            top_n = p.get('_effective_top_n') or min(p['page'] * p['per_page'], VECTOR_MAX_TOP_N)
            vector_kwargs = {
                'text': text,
                'in_summary': p['in_summary'],
                'in_fulltext': p['in_fulltext'],
                'top_n': top_n,
                'score_threshold': p.get('score_threshold_min', VECTOR_DEFAULT_SCORE_THRESHOLD),
                'score_threshold_max': p.get('score_threshold_max', 1.0),
            }

            if p['start_time'] and p['end_time']:
                vector_kwargs['event_period'] = (
                    datetime.datetime.fromisoformat(p['start_time']),
                    datetime.datetime.fromisoformat(p['end_time'])
                )
            if p['archive_start_time'] and p['archive_end_time']:
                vector_kwargs['archive_period'] = (
                    datetime.datetime.fromisoformat(p['archive_start_time']),
                    datetime.datetime.fromisoformat(p['archive_end_time'])
                )

            raw: List[Tuple[str, float, dict]] = self.intelligence_hub.vector_search_intelligence(**vector_kwargs)

            # 分页
            start = (p['page'] - 1) * p['per_page']
            end = start + p['per_page']
            page_items = raw[start:end]

            uuids = []
            score_map = {}
            for doc_id, score, _ in page_items:
                uuids.append(doc_id)
                score_map[doc_id] = score
            articles = self.intelligence_hub.get_intelligence(uuids)

            for article in articles:
                doc_id = article.get('UUID')
                article['APPENDIX'][APPENDIX_VECTOR_SCORE] = score_map.get(doc_id, 0.0)

            articles.sort(key=lambda x: x['APPENDIX'][APPENDIX_VECTOR_SCORE], reverse=True)

            return articles, len(raw)

        # --------------------------------------------------------------------------------------------

        @app.get("/api/clusters/latest")
        def api_clusters_latest():
            try:
                limit = int(request.args.get("limit", 200))
                client_sort_by = request.args.get("sort_by", "time")
                desc = request.args.get("desc", "1") in ("1", "true", "True", "yes", "Y", "y")
                source = request.args.get("source", "online").lower().strip()
                refresh = request.args.get("refresh", "0") in ("1", "true", "True", "yes", "Y", "y")
                cache_key = ("clusters_latest", source, limit, client_sort_by, desc)

                hub = self.intelligence_hub
                agg = getattr(hub, "aggregation_engine_summary", None)
                if not agg:
                    return jsonify({"error": "Aggregation engine not configured"}), 501

                if not refresh:
                    with self._cluster_response_cache_lock:
                        now = time.time()
                        for key, item in list(self._cluster_response_cache.items()):
                            if now - item["created_at"] > self.cluster_cache_ttl_sec:
                                self._cluster_response_cache.pop(key, None)
                        cached = self._cluster_response_cache.get(cache_key)
                        if cached and now - cached["created_at"] <= self.cluster_cache_ttl_sec:
                            payload = dict(cached["payload"])
                            payload["cache"] = {
                                "hit": True,
                                "ttl_sec": self.cluster_cache_ttl_sec,
                                "created_at": cached["created_at"],
                            }
                            return jsonify(payload), 200

                # 建立一个闭包透传 fetcher 给 engine
                def doc_fetcher(uuids):
                    return hub.get_intelligence(uuids, light_weight=True)

                result = agg.build_rich_clusters_latest(
                    doc_fetcher=doc_fetcher,
                    doc_cleaner=exclude_raw_data,
                    limit=limit,
                    descending=desc,
                    source=source
                )
                result["cache"] = {
                    "hit": False,
                    "ttl_sec": self.cluster_cache_ttl_sec,
                    "created_at": time.time(),
                }
                with self._cluster_response_cache_lock:
                    self._cluster_response_cache[cache_key] = {
                        "created_at": result["cache"]["created_at"],
                        "payload": result,
                    }

                return jsonify(result), 200

            except Exception as e:
                logger.exception("api_clusters_latest error")
                return jsonify({"error": str(e)}), 500

        @app.get("/api/clusters/<cluster_id>/members")
        def api_cluster_members(cluster_id: str):
            try:
                limit = int(request.args.get("limit", 100))
                offset = int(request.args.get("offset", 0))
                sort_by = request.args.get("sort_by", "time")
                desc = request.args.get("desc", "1") in ("1", "true", "True")
                source = request.args.get("source", "online").lower().strip()
                refresh = request.args.get("refresh", "0") in ("1", "true", "True", "yes", "Y", "y")
                cache_key = ("cluster_members", source, cluster_id, offset, limit, sort_by, desc)

                hub = self.intelligence_hub
                agg = getattr(hub, "aggregation_engine_summary", None)
                if not agg:
                    return jsonify({"error": "Aggregation engine not configured"}), 501

                if not refresh:
                    with self._cluster_response_cache_lock:
                        now = time.time()
                        for key, item in list(self._cluster_response_cache.items()):
                            if now - item["created_at"] > self.cluster_cache_ttl_sec:
                                self._cluster_response_cache.pop(key, None)
                        cached = self._cluster_response_cache.get(cache_key)
                        if cached and now - cached["created_at"] <= self.cluster_cache_ttl_sec:
                            payload = dict(cached["payload"])
                            payload["cache"] = {
                                "hit": True,
                                "ttl_sec": self.cluster_cache_ttl_sec,
                                "created_at": cached["created_at"],
                            }
                            return jsonify(payload), 200

                def doc_fetcher(uuids):
                    return hub.get_intelligence(uuids, light_weight=True)

                result = agg.build_rich_cluster_members(
                    cluster_id=cluster_id,
                    doc_fetcher=doc_fetcher,
                    doc_cleaner=exclude_raw_data,
                    offset=offset,
                    limit=limit,
                    sort_by=sort_by,
                    descending=desc,
                    source=source
                )
                result["source"] = source
                result["cache"] = {
                    "hit": False,
                    "ttl_sec": self.cluster_cache_ttl_sec,
                    "created_at": time.time(),
                }
                with self._cluster_response_cache_lock:
                    self._cluster_response_cache[cache_key] = {
                        "created_at": result["cache"]["created_at"],
                        "payload": result,
                    }

                return jsonify(result), 200

            except ValueError as ve:
                return jsonify({"error": str(ve)}), 404
            except Exception as e:
                logger.exception("api_cluster_members error")
                return jsonify({"error": str(e)}), 500

        @app.route('/intelligence/<string:intelligence_uuid>', methods=['GET'])
        def intelligence_viewer_api(intelligence_uuid: str):
            return render_template('intelligence_detail.html', uuid=intelligence_uuid)

        # ---------------------------------------------- Management Pages ----------------------------------------------

        @app.route('/statistics/score_distribution.html', methods=['GET'])
        @WebServiceAccessManager.login_required
        def score_distribution_page():
            return get_statistics_page('/statistics/score_distribution')

        @app.route('/statistics/intelligence_statistics.html', methods=['GET'])
        @WebServiceAccessManager.login_required
        def intelligence_dashboard_page():
            return get_intelligence_statistics_page()

        @app.route('/maintenance/export_mongodb.html', methods=['GET'])
        @WebServiceAccessManager.login_required
        def export_mongodb_page():
            return render_template('export_mongodb.html')

    # ----------------------------------------------- Management Service -----------------------------------------------

        @app.route('/statistics/score_distribution', methods=['GET', 'POST'])
        @WebServiceAccessManager.login_required
        def get_score_distribution():
            try:
                # Get query parameters
                start_time_str = request.args.get('start_time')
                end_time_str = request.args.get('end_time')

                if not start_time_str or not end_time_str:
                    return jsonify({
                        "error": "Both start_time and end_time parameters are required"
                    }), 400

                # Convert to datetime objects
                # start_time = datetime.datetime.fromisoformat(start_time_str.replace('Z', '+00:00'))
                # end_time = datetime.datetime.fromisoformat(end_time_str.replace('Z', '+00:00'))

                start_time = ensure_timezone_aware(time_str_to_datetime(start_time_str))
                end_time = ensure_timezone_aware(time_str_to_datetime(end_time_str))

                stat_engine = self.intelligence_hub.get_statistics_engine()
                score_distribution = stat_engine.get_score_distribution(start_time, end_time)

                # Convert to array format for charting
                chart_data = [
                    {"score": score, "count": count}
                    for score, count in score_distribution.items()
                ]

                return jsonify({
                    "success": True,
                    "time_range": {
                        "start": start_time_str,
                        "end": end_time_str
                    },
                    "distribution": score_distribution,
                    "chart_data": chart_data,
                    "total_records": sum(score_distribution.values())
                })

            except ValueError:
                return jsonify({
                    "error": "Invalid time format. Please use ISO format (e.g., '2024-01-01T00:00:00Z')"
                }), 400
            except Exception as e:
                logger.error(f"Error processing request: {str(e)}")
                return jsonify({
                    "error": "Internal server error",
                    "message": str(e)
                }), 500

        @app.route('/statistics/intelligence_distribution/hourly', methods=['GET'])
        @WebServiceAccessManager.login_required
        def get_hourly_stats():
            """Get record counts grouped by hour for the specified time range"""
            start_time, end_time = self.get_time_range_params()

            stat_engine = self.intelligence_hub.get_statistics_engine()
            result = stat_engine.get_hourly_stats(start_time, end_time)

            return jsonify(result)

        @app.route('/statistics/intelligence_distribution/daily', methods=['GET'])
        @WebServiceAccessManager.login_required
        def get_daily_stats():
            """Get record counts grouped by day for the specified time range"""
            start_time, end_time = self.get_time_range_params()

            stat_engine = self.intelligence_hub.get_statistics_engine()
            result = stat_engine.get_daily_stats(start_time, end_time)

            return jsonify(result)

        @app.route('/statistics/intelligence_distribution/weekly', methods=['GET'])
        @WebServiceAccessManager.login_required
        def get_weekly_stats():
            """Get record counts grouped by week for the specified time range"""
            start_time, end_time = self.get_time_range_params()

            stat_engine = self.intelligence_hub.get_statistics_engine()
            result = stat_engine.get_weekly_stats(start_time, end_time)

            return jsonify(result)

        @app.route('/statistics/intelligence_distribution/monthly', methods=['GET'])
        @WebServiceAccessManager.login_required
        def get_monthly_stats():
            """Get record counts grouped by month for the specified time range"""
            start_time, end_time = self.get_time_range_params()

            stat_engine = self.intelligence_hub.get_statistics_engine()
            result = stat_engine.get_monthly_stats(start_time, end_time)

            return jsonify(result)

        @app.route('/statistics/intelligence_distribution/summary', methods=['GET'])
        @WebServiceAccessManager.login_required
        def get_stats_summary():
            """Get overall statistics for the specified time range"""
            start_time, end_time = self.get_time_range_params()

            stat_engine = self.intelligence_hub.get_statistics_engine()
            total_count, informant_stats = stat_engine.get_stats_summary(start_time, end_time)

            return jsonify({
                "total_count": total_count,
                "time_range": {
                    "start": start_time,
                    "end": end_time
                },
                "top_informants": informant_stats
            })

        @app.route('/statistics/intelligence_dashboard_data', methods=['GET'])
        @WebServiceAccessManager.login_required
        def get_intelligence_dashboard_data():
            """
            Return aggregated data for the IIS system dashboard.
            Includes hub state, AI client health, submission flow stats and system metrics.
            """
            try:
                now = get_aware_time()
                start_time = now - datetime.timedelta(hours=24)

                stat_engine = self.intelligence_hub.get_statistics_engine()
                hourly_distribution = stat_engine.get_hourly_stats(start_time, now)
                total_count, informant_stats = stat_engine.get_stats_summary(start_time, now)

                hub_stats = self.intelligence_hub.statistics
                ai_client_stats = {}
                if self.intelligence_hub.ai_client_manager:
                    ai_client_stats = self._serialize_for_json(
                        self.intelligence_hub.ai_client_manager.get_client_stats()
                    )
                    # get_client_stats() summary counts use Enum comparisons that may not match
                    # strings; recalculate from serialized client details for accurate dashboard.
                    clients = ai_client_stats.get('clients', [])
                    summary = ai_client_stats.setdefault('summary', {})
                    summary['available'] = sum(
                        1 for c in clients
                        if c.get('state', {}).get('status', '').lower() == 'available' and not c.get('state', {}).get('is_busy')
                    )
                    summary['busy'] = sum(1 for c in clients if c.get('state', {}).get('is_busy'))

                system_stats = {}
                process_stats = {}
                thread_stats = {}
                self_pid = os.getpid()
                if self.monitor_api and self.monitor_api.monitor:
                    try:
                        system_stats = self.monitor_api.monitor.get_system_stats()
                        process_stats = self.monitor_api.monitor.get_process_stats(self_pid) or {}
                        thread_stats = self.monitor_api.monitor.get_thread_stats(self_pid) or {}
                    except Exception as e:
                        logger.warning(f"Dashboard failed to read monitor stats: {e}")

                # Count active Python threads in this process for quick diagnostics
                active_python_threads = threading.active_count()

                return jsonify({
                    "timestamp": now.isoformat(),
                    "timestamp_formatted": now.strftime('%Y-%m-%d %H:%M:%S'),
                    "hub": hub_stats,
                    "ai_clients": ai_client_stats,
                    "hourly_distribution": hourly_distribution,
                    "summary": {
                        "total_count": total_count,
                        "time_range": {"start": start_time.isoformat(), "end": now.isoformat()},
                        "top_informants": informant_stats
                    },
                    "system": system_stats,
                    "process": process_stats,
                    "threads": thread_stats,
                    "active_python_threads": active_python_threads,
                })
            except Exception as e:
                logger.error(f"Error generating dashboard data: {e}", exc_info=True)
                return jsonify({"error": str(e)}), 500

        # --------------------------- Entity Frequency Statistics ---------------------------

        @app.route('/api/sources', methods=['GET'])
        def api_sources():
            """获取所有数据源域名列表及文章数量。"""
            try:
                engine = self.intelligence_hub.archive_db_query_engine
                domains = engine.get_source_domains(limit=500)
                return jsonify({"domains": domains})
            except Exception as e:
                logger.error(f"api_sources error: {e}", exc_info=True)
                return jsonify({"error": str(e)}), 500

        @app.route('/statistics/entity_frequency/page', methods=['GET'])
        def entity_frequency_page():
            """实体出现频率统计页面。未登录时进入 Public 模式，固定显示最近30天。"""
            is_public = not session.get('logged_in', False)
            return render_template('entity_frequency.html', public_mode=is_public)

        @app.route('/statistics/entity_frequency', methods=['GET'])
        def entity_frequency_api():
            """
            查询实体出现频率统计。
            未登录时固定返回最近30天数据，不可自定义时间范围。
            参数:
                entity_type: LOCATION | GEOGRAPHY | PEOPLE | ORGANIZATION
                start_time: ISO 格式开始时间（登录后可自定义）
                end_time: ISO 格式结束时间（登录后可自定义）
                granularity: day | week | month（默认 day）
                top_n: 默认 20，最大 50
                bottom_threshold: 默认 5
            """
            is_public = not session.get('logged_in', False)

            entity_type = request.args.get('entity_type', 'LOCATION')
            granularity = request.args.get('granularity', 'day')
            top_n = request.args.get('top_n', '20')
            bottom_threshold = request.args.get('bottom_threshold', '5')

            valid_types = {'LOCATION', 'GEOGRAPHY', 'PEOPLE', 'ORGANIZATION'}
            if entity_type not in valid_types:
                return jsonify({"error": f"Invalid entity_type. Must be one of {valid_types}"}), 400

            valid_granularities = {'day', 'week', 'month'}
            if granularity not in valid_granularities:
                return jsonify({"error": f"Invalid granularity. Must be one of {valid_granularities}"}), 400

            try:
                if is_public:
                    # Public 模式：固定近30天
                    end_time = get_aware_time()
                    start_time = end_time - datetime.timedelta(days=30)
                else:
                    start_str = request.args.get('start_time')
                    end_str = request.args.get('end_time')
                    if start_str:
                        start_time = dateutil.parser.parse(start_str)
                    else:
                        start_time = get_aware_time() - datetime.timedelta(days=30)

                    if end_str:
                        end_time = dateutil.parser.parse(end_str)
                    else:
                        end_time = get_aware_time()

                top_n = min(int(top_n), 50)
                bottom_threshold = int(bottom_threshold)
            except Exception as e:
                return jsonify({"error": f"Parameter parse error: {e}"}), 400

            # 时间范围最大 12 个月
            if (end_time - start_time).days > 366:
                return jsonify({"error": "Time range exceeds maximum limit of 12 months (366 days)."}), 400

            engine = self.intelligence_hub.entity_frequency_engine
            if not engine:
                return jsonify({"error": "EntityFrequencyEngine not initialized"}), 503

            try:
                result = engine.query_frequency(
                    entity_type=entity_type,
                    start_time=start_time,
                    end_time=end_time,
                    granularity=granularity,
                    top_n=top_n,
                    bottom_threshold=bottom_threshold,
                    allow_build=not is_public,
                )
                return jsonify(result)
            except Exception as e:
                logger.error(f"entity_frequency_api error: {e}", exc_info=True)
                return jsonify({"error": str(e)}), 500

        @app.route('/statistics/entity_frequency/trend', methods=['GET'])
        def entity_frequency_trend():
            """
            查询单个实体的趋势数据。
            未登录时固定返回最近30天数据，不可自定义时间范围。
            参数:
                entity_type, entity_name, start_time, end_time, granularity
            """
            is_public = not session.get('logged_in', False)

            entity_type = request.args.get('entity_type', '')
            entity_name = request.args.get('entity_name', '')
            granularity = request.args.get('granularity', 'day')

            if not entity_name:
                return jsonify({"error": "entity_name is required"}), 400

            valid_types = {'LOCATION', 'GEOGRAPHY', 'PEOPLE', 'ORGANIZATION'}
            if entity_type not in valid_types:
                return jsonify({"error": f"Invalid entity_type"}), 400

            try:
                if is_public:
                    end_time = get_aware_time()
                    start_time = end_time - datetime.timedelta(days=30)
                else:
                    start_str = request.args.get('start_time')
                    end_str = request.args.get('end_time')
                    start_time = dateutil.parser.parse(start_str) if start_str else get_aware_time() - datetime.timedelta(days=30)
                    end_time = dateutil.parser.parse(end_str) if end_str else get_aware_time()
            except Exception as e:
                return jsonify({"error": f"Parameter parse error: {e}"}), 400

            engine = self.intelligence_hub.entity_frequency_engine
            if not engine:
                return jsonify({"error": "EntityFrequencyEngine not initialized"}), 503

            try:
                result = engine.get_entity_trend(
                    entity_type=entity_type,
                    entity_name=entity_name,
                    start_time=start_time,
                    end_time=end_time,
                    granularity=granularity,
                    allow_build=not is_public,
                )
                return jsonify(result)
            except Exception as e:
                logger.error(f"entity_frequency_trend error: {e}", exc_info=True)
                return jsonify({"error": str(e)}), 500

        @app.route('/statistics/entity_frequency/build_cache/cancel', methods=['POST'])
        @WebServiceAccessManager.login_required
        def entity_frequency_build_cache_cancel():
            """取消正在进行的缓存构建任务。"""
            engine = self.intelligence_hub.entity_frequency_engine
            if not engine:
                return jsonify({"error": "EntityFrequencyEngine not initialized"}), 503
            engine.cancel_build()
            return jsonify({"status": "cancelled"})

        @app.route('/statistics/entity_frequency/build_cache', methods=['POST'])
        @WebServiceAccessManager.login_required
        def entity_frequency_build_cache():
            """
            手动触发实体频率缓存构建，返回 NDJSON 流式进度。
            """
            data = request.get_json() or {}
            start_str = data.get('start_time')
            end_str = data.get('end_time')

            engine = self.intelligence_hub.entity_frequency_engine
            if not engine:
                return jsonify({"error": "EntityFrequencyEngine not initialized"}), 503

            try:
                if start_str and end_str:
                    start_time = dateutil.parser.parse(start_str)
                    end_time = dateutil.parser.parse(end_str)
                else:
                    now = datetime.datetime.now(datetime.timezone.utc)
                    end_time = now.replace(hour=0, minute=0, second=0, microsecond=0)
                    start_time = end_time - datetime.timedelta(days=1)

                if (end_time - start_time).days > 366:
                    return jsonify({"error": "Time range exceeds maximum limit of 12 months (366 days)."}), 400
            except Exception as e:
                return jsonify({"error": f"Parameter parse error: {e}"}), 400

            def generate():
                try:
                    for progress in engine.build_cache_with_progress(start_time, end_time):
                        yield json.dumps(progress) + "\n"
                        if progress.get("status") == "cancelled":
                            return
                except Exception as e:
                    logger.error(f"build_cache stream error: {e}", exc_info=True)
                    yield json.dumps({"status": "error", "message": str(e)}) + "\n"

            return Response(generate(), mimetype='application/x-ndjson')

        @app.route('/api/debug/trigger_aggregation', methods=['POST'])
        @WebServiceAccessManager.login_required
        def debug_trigger_aggregation():
            data = request.get_json() or {}
            start_str = data.get('start_time')
            end_str = data.get('end_time')

            time_range = None
            if start_str and end_str:
                start_ts = dateutil.parser.parse(start_str).timestamp()
                end_ts = dateutil.parser.parse(end_str).timestamp()
                time_range = (start_ts, end_ts)

            try:
                agg_engine = self.intelligence_hub.aggregation_engine_summary
                if not agg_engine:
                    return jsonify({"error": "Engine not ready"}), 503

                # 新增传入 fetcher 以便后台监控线程使用
                def doc_fetcher(uuids):
                    return self.intelligence_hub.get_intelligence(uuids, light_weight=True)

                job_id = agg_engine.trigger_offline(time_range=time_range, doc_fetcher=doc_fetcher)
                return jsonify({
                    "status": "success",
                    "job_id": job_id,
                    "mode": "explicit_time" if time_range else "auto_discovery"
                })
            except Exception as e:
                return jsonify({"error": str(e)}), 500

        # --------------------------- Intelligence Graph ---------------------------

        @app.route('/intelligence/graph/build', methods=['POST'])
        @WebServiceAccessManager.login_required
        def trigger_graph_build():
            """
            触发一次异步的情报脉络图谱推演
            Body JSON: {"seed_uuid": "xxx", "max_depth": 3, "window_days": 7}
            """
            graph_engine = self.intelligence_hub.dynamic_graph_engine

            if graph_engine is None:
                return jsonify({
                    "error": "Service Unavailable",
                    "message": "Graph engine is not ready yet.",
                    "code": "ENGINE_INITIALIZING"
                }), 503

            data = request.get_json() or {}
            seed_uuid = data.get('seed_uuid')

            if not seed_uuid:
                return jsonify({"error": "Missing seed_uuid"}), 400

            max_depth = int(data.get('max_depth', 3))
            window_days = int(data.get('window_days', 7))

            # 触发异步推演，毫秒级返回 thread_id
            try:
                thread_id = graph_engine.trigger_graph_build(
                    seed_uuid=seed_uuid,
                    max_depth=max_depth,
                    window_days=window_days
                )
                return jsonify({
                    "status": "success",
                    "thread_id": thread_id,
                    "message": "Graph building task submitted."
                }), 202
            except Exception as e:
                return jsonify({"error": str(e)}), 500

        @app.route('/intelligence/graph/status/<thread_id>', methods=['GET'])
        @WebServiceAccessManager.login_required
        def get_graph_status(thread_id):
            """
            前端轮询接口，获取推演状态与最终数据
            """
            graph_engine = self.intelligence_hub.dynamic_graph_engine

            if graph_engine is None:
                return jsonify({
                    "error": "Service Unavailable",
                    "message": "Graph engine is not ready yet.",
                    "code": "ENGINE_INITIALIZING"
                }), 503

            snapshot = graph_engine.get_snapshot(thread_id)

            if not snapshot:
                return jsonify({"error": "Thread ID not found"}), 404

            # 隐藏内部的 seed_uuid 等非必要字段，只吐出前端需要的核心状态和数据
            return jsonify({
                "thread_id": snapshot.get("thread_id"),
                "status": snapshot.get("status"),  # PROCESSING, ANALYZING, COMPLETED, FAILED
                "created_at": snapshot.get("created_at"),
                "nodes": snapshot.get("nodes", []),
                "edges": snapshot.get("edges", []),
                "llm_summary": snapshot.get("llm_summary"),
                "llm_evaluation_score": snapshot.get("llm_evaluation_score")
            })

        @app.route('/maintenance/export_mongodb', methods=['POST'])
        @WebServiceAccessManager.login_required
        def export_mongodb():
            """
            Handle export request.
            Supports:
            1. Mode: 'range' (specific start/end) or 'all' (entire db)
            2. Split: 'month', 'week', 'year', or None
            3. Target: 'archive', 'cache', or 'all'
            """
            try:
                # 1. Get parameters
                data = request.get_json()
                if not data:
                    return jsonify({'status': 'error', 'message': 'No JSON data received'}), 400

                mode = data.get('mode', 'range')  # 'range' or 'all'
                target = data.get('target', 'archive')
                split_by = data.get('splitBy') # 'month', 'week', 'year' or None (empty string)

                # Normalize split_by to None if it's an empty string or 'none'
                if not split_by or split_by == 'none':
                    split_by = None

                # 2. Prepare Date Range (Only if mode is 'range')
                start_dt = None
                end_dt = None

                if mode == 'range':
                    start_date_str = data.get('startDate')
                    end_date_str = data.get('endDate')
                    if not start_date_str or not end_date_str:
                        return jsonify({'status': 'error', 'message': 'startDate and endDate are required for range mode'}), 400
                    try:
                        start_dt = date_parser.parse(start_date_str)
                        end_dt = date_parser.parse(end_date_str)
                    except ValueError as e:
                        return jsonify({'status': 'error', 'message': f'Invalid date format: {str(e)}'}), 400

                generated_files = []
                messages = []

                # 3. Define Export Logic Helper
                def run_export(db_instance, sub_dir, time_field, prefix):
                    """Helper to execute export on a specific DB instance"""
                    if not db_instance:
                        messages.append(f"Skipped {prefix}: Database instance not initialized.")
                        return

                    full_dir = os.path.join(EXPORT_PATH, sub_dir)

                    # Dispatch based on mode
                    if mode == 'all':
                        # Call export_all
                        files = db_instance.export_all(
                            directory=full_dir,
                            split_by=split_by,
                            time_field=time_field,
                            add_timestamp=True
                        )
                        if files:
                            generated_files.extend(files)
                            messages.append(f"Exported all {prefix} data ({len(files)} files)")
                        else:
                            messages.append(f"No data found in {prefix}")
                    else:
                        # Call export_by_time_range
                        output_file = db_instance.export_by_time_range(
                            start_dt=start_dt,
                            end_dt=end_dt,
                            directory=full_dir,
                            time_field=time_field,
                            file_prefix=prefix,
                            add_timestamp=True
                        )
                        if output_file:
                            generated_files.append(output_file)
                            messages.append(f"Exported {prefix} range to {os.path.basename(output_file)}")
                        else:
                            messages.append(f"No data found for {prefix} in range")

                # 4. Execute Exports based on Target

                # --- Export Archive DB ---
                if target in ['archive', 'all']:
                    time_field = f"APPENDIX.{APPENDIX_TIME_ARCHIVED}"
                    run_export(
                        db_instance=self.intelligence_hub.mongo_db_archive,
                        sub_dir='mongo_db_archive',
                        time_field=time_field,
                        prefix='intelligence_archived'
                    )

                # --- Export Cache DB ---
                if target in ['cache', 'all']:
                    cache_time_field = '__TIME_GOT__'
                    run_export(
                        db_instance=self.intelligence_hub.mongo_db_cache,
                        sub_dir='mongo_db_cache',
                        time_field=cache_time_field,
                        prefix='intelligence_cache'
                    )

                # 5. Return Result
                if generated_files:
                    return jsonify({
                        'status': 'success',
                        'message': '; '.join(messages),
                        'files': generated_files,
                        'count': len(generated_files)
                    })
                else:
                    return jsonify({
                        'status': 'warning',
                        'message': '; '.join(messages)
                    }), 200

            except Exception as e:
                logging.error(f"Export failed: {e}", exc_info=True)
                return jsonify({
                    'status': 'error',
                    'message': f'Server error: {str(e)}'
                }), 500

        @app.route('/download/<filename>')
        @WebServiceAccessManager.login_required
        def download_file(filename):
            """Download exported file"""
            try:
                if 'intelligence_archived' in filename:
                    # MongoDB export
                    file_dir = 'exports'
                else:
                    file_dir = 'download'

                return send_file(
                    os.path.join(file_dir, filename),
                    as_attachment=True
                )
            except FileNotFoundError:
                return jsonify({
                    'status': 'error',
                    'message': 'File not found'
                }), 404

    # ------------------------------------------------------------------------------------------------------------------

    def handle_error(self, error: str):
        print(f'Handle error in IntelligenceHubWebService: {error}')

    def _articles_to_rss_items(self, articles: dict | List[dict]) -> List[FeedItem]:
        # Using a fixed default time, combined with a unique UUID,
        # prevents RSS readers from mistakenly identifying article duplicates.
        default_date = datetime.datetime(1970, 1, 1)

        if not isinstance(articles, list):
            articles = [articles]
        try:
            rss_items = []
            for doc in articles:
                if 'EVENT_BRIEF' in doc and 'UUID' in doc:
                    rss_item = FeedItem(
                        guid=doc['UUID'],
                        title=doc.get('EVENT_TITLE', doc['EVENT_BRIEF']),
                        link=f"/intelligence/{doc['UUID']}",
                        description=doc['EVENT_BRIEF'],
                        pub_date=doc.get('APPENDIX', {}).get(APPENDIX_TIME_ARCHIVED, default_date))
                    rss_items.append(rss_item)
                else:
                    logger.warning(f'Warning: archived data field missing.')
            return rss_items
        except Exception as e:
            logger.error(f"Article to rss items failed: {str(e)}")
            return []

    def get_rendered_md_post(self, post_name: str) -> str:
        try:
            # Sanitize input and construct safe file path
            safe_article = post_name.replace('..', '').strip('/')  # Prevent directory traversal
            md_file_path = os.path.join('posts', f"{safe_article}.md")

            # Generate HTML from Markdown
            rendered_html_path = generate_html_from_markdown(md_file_path)

            if rendered_html_path:
                # Extract relative template path (remove 'templates/' prefix)
                template_path = os.path.relpath(
                    rendered_html_path,
                    start='templates'
                ).replace('\\', '/')  # Windows compatibility

                return render_template(template_path)
            else:
                return ''
        except Exception as e:
            logger.error(f'Invalid post: {post_name}')
            return ''

    def get_time_range_params(self):
        """
        Extract and validate time range parameters from request
        Returns start_time and end_time as datetime objects
        """
        start_str = request.args.get('start')
        end_str = request.args.get('end')

        if start_str:
            start_time = dateutil.parser.parse(start_str)
        else:
            # Default to 24 hours ago if no start time provided
            start_time = get_aware_time() - datetime.timedelta(hours=24)

        if end_str:
            end_time = dateutil.parser.parse(end_str)
        else:
            # Default to current time if no end time provided
            end_time = get_aware_time()

        return start_time, end_time

    def dump_request_connection_periodically(self):
        self.request_tracer.dump_long_running_requests()
        threading.Timer(30.0, self.dump_request_connection_periodically).start()