[{'Text': '"""Raw GBS MongoDB query tool.\n\nProvides a single read-only tool that runs any aggregation pipeline against\nthe GBS collections (Operator, Statistics, Sources, GlobalVariables).\n\nThis is the escape hatch: when the structured get_status/add_statistic/create_source\ntools don\'t cover what you need, use this to explore, debug, and discover edge cases.\n\nSafety: only read operations allowed. Write operations ($out, $merge, etc.) are\nexplicitly rejected. Insert/update/delete handled by validated tools only.\n"""\n\nimport json\nimport re\n\nfrom point_topic_mcp.core.utils import check_env_vars\nfrom point_topic_mcp.core import mongodb_utils as mongo\nfrom dotenv import load_dotenv\n\nload_dotenv()\n\n# Collections available for read-only queries\nALLOWED_COLLECTIONS = {\n    "Operator",\n    "Statistics",\n    "Sources",\n    "GlobalVariables",\n    "operator",\n    "statistics",\n    "sources",\n    "globalvariables",\n    "globalVariables",\n}\n\n# Stage names that MODIFY data — reject these\n_WRITE_STAGES = {"$out", "$merge"}\n\n\nif check_env_vars("gbs_query_tools", ["PT_RESEARCH_DATABASE_URI"]):\n\n    def _validate_pipeline(pipeline: list) -> None:\n        """Validate pipeline is read-only and uses allowed stages.\n\n        Raises ValueError with clear message if validation fails.\n        """\n        if not isinstance(pipeline, list):\n            raise ValueError("Pipeline must be a JSON array of stage objects")\n\n        for i, stage in enumerate(pipeline):\n            if not isinstance(stage, dict):\n                raise ValueError(f"Stage {i} is not a JSON object: {stage}")\n            if not stage:\n                raise ValueError(f"Stage {i} is empty")\n            stage_name = next(iter(stage))\n            if stage_name in _WRITE_STAGES:\n                raise ValueError(\n                    f"Stage {i} uses \'{stage_name}\' which is write-only. "\n                    f"Only read-only aggregation pipelines are allowed."\n                )\n\n    def _find_used_collections(pipeline: list) -> set:\n        """Determine which collections the pipeline references."""\n        used = set()\n        for stage in pipeline:\n            if "$lookup" in stage:\n                from_val = stage["$lookup"].get("from", "")\n                # from could be a dynamic expression — use the literal string\n                if isinstance(from_val, str) and from_val:\n                    used.add(from_val)\n            if "$unionWith" in stage:\n                coll = stage["$unionWith"].get("coll", "")\n                if isinstance(coll, str) and coll:\n                    used.add(coll)\n            if "$facet" in stage:\n                for facet_name, facet_pipeline in stage["$facet"].items():\n                    used |= _find_used_collections(facet_pipeline)\n        return used\n\n    def read_gbs(pipeline_json: str = "", pipeline: list = None) -> str:\n        """Run a read-only MongoDB aggregation pipeline against GBS collections.\n\n        Primary collections (pipeline runs against these):\n          - Operator      — 1300+ operators worldwide (fields: name, country, technologies, asn)\n          - Statistics    — Subscriber counts (fields: type, operatorId, period, tech, channel,\n                            domain, subscribers, state)\n          - Sources       — Source URLs/files\n          - GlobalVariables — System config (currentStatsPeriod, currentTariffsPeriod)\n\n        Provide either:\n          pipeline_json: JSON string of an aggregation pipeline array\n          pipeline: Python list (used programmatically, not by agents)\n\n        Examples of useful pipelines:\n\n        1. Get BT Group stats across all periods:\n        ```json\n        [\n          {"$match": {"name": "BT Group", "country": "United Kingdom"}},\n          {"$lookup": {\n            "from": "Statistics",\n            "let": {"oid": "$_id"},\n            "pipeline": [\n              {"$match": {"$expr": {"$eq": ["$operatorId", "$$oid"]}}}\n            ],\n            "as": "stats"\n          }},\n          {"$project": {"name": 1, "stats.period": 1, "stats.tech": 1, "stats.subscribers": 1, "stats.state": 1}}\n        ]\n        ```\n\n        2. Get current GlobalVariables:\n        ```json\n        [\n          {"$match": {"isCurrent": true}},\n          {"$project": {"currentStatsPeriod": 1, "currentTariffsPeriod": 1, "updatedBy": 1}}\n        ]\n        ```\n\n        3. Count stats by state and period:\n        ```json\n        [\n          {"$group": {"_id": {"period": "$period", "state": "$state"}, "count": {"$sum": 1}}},\n          {"$sort": {"_id.period": -1, "_id.state": 1}}\n        ]\n        ```\n\n        Returns:\n            JSON string with pipeline results.\n            Error messages for invalid pipelines or empty data.\n        """\n        # Parse pipeline\n        if pipeline_json:\n            try:\n                pipeline = json.loads(pipeline_json)\n            except json.JSONDecodeError as e:\n                return f"Error: Invalid JSON pipeline — {e}"\n        elif pipeline is None:\n            return "Error: Provide pipeline_json (JSON string) or pipeline (list)"\n\n        # Validate read-only\n        try:\n            _validate_pipeline(pipeline)\n        except ValueError as e:\n            return f"Error: {e}"\n\n        # Determine which collection to run against\n        # Default to Statistics (most useful), detect from pipeline\n        collection = "Statistics"\n        first_stage = pipeline[0] if pipeline else {}\n        if "$match" in first_stage:\n            match = first_stage["$match"]\n            # If matching on name/country/technologies, likely going against Operator\n            if isinstance(match, dict):\n                match_keys = set(match.keys())\n                if match_keys & {"name", "country", "isArchived", "technologies", "asn", "_id"}:\n                    if "type" not in match_keys and "period" not in match_keys and "state" not in match_keys:\n                        collection = "Operator"\n\n        # If $lookup references a collection, also check first stage for collection hints\n        used_collections = _find_used_collections(pipeline)\n        # If pipeline looks like it targets GlobalVariables\n        if used_collections or collection == "Statistics":\n            pass  # auto-detect below\n\n        # Better auto-detect: if $match keys look like a specific collection\n        if "$match" in first_stage:\n            match = first_stage["$match"]\n            if isinstance(match, dict):\n                mkeys = set(match.keys())\n                if "isCurrent" in mkeys or "currentStatsPeriod" in mkeys:\n                    collection = "GlobalVariables"\n                elif "operatorId" in mkeys or "type" in mkeys or "period" in mkeys or "state" in mkeys or "subscribers" in mkeys:\n                    collection = "Statistics"\n                elif "year" in mkeys and "quarter" in mkeys:\n                    collection = "Sources"\n\n        # Ensure target collection is allowed\n        if collection not in ALLOWED_COLLECTIONS:\n            return f"Error: Collection \'{collection}\' not in allowed set: {sorted(ALLOWED_COLLECTIONS)}"\n\n        # Run the pipeline\n        try:\n            result = mongo.run_aggregation(pipeline, collection=collection, timeout=60)\n        except RuntimeError as e:\n            return f"Query error: {e}"\n        except json.JSONDecodeError as e:\n            return f"Result parsing error: {e}"\n        except Exception as e:\n            return f"Unexpected error: {e}"\n\n        if not result:\n            return "[] — no documents matched the query"\n\n        # Format as pretty JSON (limit output size for very large results)\n        try:\n            formatted = json.dumps(result, indent=2, default=str)\n        except Exception as e:\n            formatted = str(result)\n\n        # Truncate if too large (MCP message limits)\n        MAX_CHARS = 50000\n        if len(formatted) > MAX_CHARS:\n            formatted = formatted[:MAX_CHARS] + f"\\n\\n... [truncated: {len(result)} total docs, showing first {formatted[:MAX_CHARS].count(chr(10))} lines]"\n\n        return formatted\n'}]