[{'Text': '"""GBS (Global Broadband Statistics) tools.\n\nGBS is a quarterly process: associates find operator subscriber numbers\n(web/PDF/image), enter them as pending records, admins approve/publish.\n\nCollections (pt-research-app Prisma schema):\n  Operator       — 1300+ operators, 175 UK. Fields: name, country, technologies\n  Statistics     — Subscriber counts. Fields: type, operatorId, period, tech,\n                   channel, domain, subscribers, state (1=pending, 2=approved, 3=published)\n  Sources        — Source URLs/files linked to operator/period\n  GlobalVariables— Tracks currentStatsPeriod (e.g. "2025Q3")\n\nValid enums: types=[broadband, mobile, iptv]\n             channels=[Infrastructure, Retail]\n             domains=[Residential, Business, Total]\n\nUses mongosh subprocess (not PyMongo). Requires PT_RESEARCH_DATABASE_URI.\n"""\n\nimport json\nimport re\nfrom typing import Optional\n\nfrom point_topic_mcp.core.utils import check_env_vars, dynamic_docstring\nfrom point_topic_mcp.core import mongodb_utils as mongo\nfrom dotenv import load_dotenv\n\nload_dotenv()\n\nADMIN_BASE = "https://pt-research-app.vercel.app"\n\n# Collections available for read_gbs read-only queries\n_GBS_COLLECTIONS = {\n    "Operator", "Statistics", "Sources", "GlobalVariables",\n    "operator", "statistics", "sources", "globalvariables", "globalVariables",\n}\n\n# Stage names that MODIFY data — rejected by read_gbs\n_GBS_WRITE_STAGES = {"$out", "$merge"}\n\n# Cheatsheet injected into read_gbs docstring — agent instructions, live state, and pipeline templates.\n_GBS_CHEATSHEET = """\nGBS CHEATSHEET\n==============\n\nAlways provide the admin URL after adding a statistic:\n  https://pt-research-app.vercel.app/operators/{operator_id}/statistics\n\nCurrent period: 2026Q1  |  Previous: 2025Q4  |  Tariffs: 2026Q2\n\nKey people:\n  veronica.speiser@point-topic.com — 177 recs since May (UK ops)\n  aliciamenendezbuick@gmail.com    — 56 recs (global sweep)\n  gbs-agent                        — automated\n\nCOLLECTION FIELDS:\n\nStatistics:\n  operatorId (ObjectId), period (YYYYQN), type (broadband|mobile|iptv),\n  tech (free text), channel (Infrastructure|Retail),\n  domain (Residential|Business|Total), subscribers (int),\n  state (1=pending, 2=approved, 3=published), createdBy (email)\n\nOperator:\n  name (string), country (string), technologies (string[]), isArchived (bool)\n\nGlobalVariables:\n  currentStatsPeriod (string), currentTariffsPeriod (string)\n\nPIPELINE TEMPLATES:\n\n0. Resolve current period:\n  [{"$match": {"isCurrent": true}}, {"$project": {"currentStatsPeriod": 1}}]\n\n1. Country coverage (replaces old get_status):\n  [{"$match": {"country": "United Kingdom", "isArchived": {"$ne": true}}},\n   {"$lookup": {"from": "Statistics", "let": {"oid": "$_id"},\n     "pipeline": [{"$match": {"$expr": {"$eq": ["$operatorId", "$$oid"]}}},\n       {"$match": {"period": "2026Q1", "type": "broadband"}}],\n     "as": "stats"}},\n   {"$addFields": {"statCount": {"$size": "$stats"}}},\n   {"$project": {"name": 1, "statCount": 1}}, {"$sort": {"statCount": -1}}]\n\n2. Operator drill-down (get OID first from Operator by name):\n  [{"$match": {"operatorId": {"$oid": "<OID>"}, "period": "2026Q1"}},\n   {"$project": {"tech": 1, "channel": 1, "domain": 1, "subscribers": 1, "state": 1}}]\n\n3. Gap analysis — techs from previous period missing in current:\n  [{"$match": {"operatorId": {"$oid": "<OID>"}, "period": "2025Q4"}},\n   {"$group": {"_id": null, "techs": {"$addToSet": "$tech"}}}]\n\n4. Recent activity:\n  [{"$match": {"type": "broadband", "createdAt": {"$gte": {"$date": "<DATE>"}}}},\n   {"$group": {"_id": "$createdBy", "count": {"$sum": 1}}}]\n\n5. Period progress (all operators with data in current period):\n  [{"$match": {"period": "2026Q1", "type": "broadband", "is_archived": {"$ne": true}}},\n   {"$group": {"_id": "$operator", "count": {"$sum": 1},\n     "techs": {"$addToSet": "$tech"}, "states": {"$addToSet": "$state"}}},\n   {"$sort": {"count": -1}}]\n\n6. Countries with data this quarter:\n  [{"$match": {"period": "2026Q1", "type": "broadband"}},\n   {"$group": {"_id": "$country", "count": {"$sum": 1}}},\n   {"$sort": {"count": -1}}]\n\n7. EJSON syntax: ObjectId={"$oid": "..."}, Date={"$date": "..."}\n\n8. Full operator history (all periods, compact):\n  [{"$match": {"name": "<NAME>", "country": "<COUNTRY>"}},\n   {"$lookup": {"from": "Statistics", "let": {"oid": "$_id"},\n     "pipeline": [{"$match": {"$expr": {"$eq": ["$operatorId", "$$oid"]}}},\n       {"$group": {"_id": "$period", "count": {"$sum": 1},\n         "techs": {"$addToSet": "$tech"}}}, {"$sort": {"_id": -1}}],\n     "as": "period_summary"}}]\n\nBT FAMILY:\n  BT Group (662ff7a3f9d5751ffbf300f6) — 4 active techs, only 1 record in 2026Q1\n  BT Wholesale Business — separate operator, fully entered\n\nRULES:\n- Never guess subscriber numbers — tool can only report what\'s entered\n- Use $lookup/$in not N+1 calls (each mongosh spawn = ~200ms)\n- Always check previous complete period for carry-forward gaps\n"""\n\nif check_env_vars("gbs_tools", ["PT_RESEARCH_DATABASE_URI"]):\n\n    VALID_TYPES = ["broadband", "mobile", "iptv"]\n    VALID_CHANNELS = ["Infrastructure", "Retail"]\n    VALID_DOMAINS = ["Residential", "Business", "Total"]\n\n    def _valid_object_id(oid: str) -> bool:\n        return bool(oid and len(oid) == 24 and all(c in "0123456789abcdefABCDEF" for c in oid))\n\n    def _valid_period(p: str) -> bool:\n        return bool(re.match(r"^\\d{4}Q[1-4]$", p))\n\n    def _admin_url(operator_id: str) -> str:\n        return f"{ADMIN_BASE}/operators/{operator_id}/statistics"\n\n    def _validate_pipeline(pipeline: list) -> None:\n        """Validate pipeline is read-only. Raises ValueError if not."""\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 _GBS_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    @dynamic_docstring([("{GBS_CHEATSHEET}", lambda: _GBS_CHEATSHEET)])\n    def read_gbs(pipeline_json: str = "", pipeline: list | None = None) -> str:\n        """Run a read-only MongoDB aggregation pipeline against GBS collections.\n\n        Primary collections:\n          - Operator        — 1300+ operators (fields: name, country, technologies)\n          - Statistics      — Subscriber counts (fields: type, operatorId, period, tech,\n                              channel, domain, subscribers, state)\n          - Sources         — Source URLs/files linked to operator/period\n          - GlobalVariables — System config (currentStatsPeriod, currentTariffsPeriod)\n\n        Provide pipeline_json (JSON string) or pipeline (Python list for internal use).\n        Supports EJSON syntax: {"$oid": "..."} for ObjectId, {"$date": "..."} for dates.\n\n        {GBS_CHEATSHEET}\n\n        Returns:\n            Formatted JSON string of results, or error message.\n        """\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        if not isinstance(pipeline, list):\n            return f"Error: Pipeline must be a JSON array, got {type(pipeline).__name__}"\n\n        try:\n            _validate_pipeline(pipeline)\n        except ValueError as e:\n            return f"Error: {e}"\n\n        # Auto-detect target collection from $match keys\n        collection = "Statistics"\n        first_stage = pipeline[0] if pipeline else {}\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 (\n                    "operatorId" in mkeys or "type" in mkeys or "period" in mkeys\n                    or "state" in mkeys or "subscribers" in mkeys\n                ):\n                    collection = "Statistics"\n                elif "year" in mkeys and "quarter" in mkeys:\n                    collection = "Sources"\n                elif mkeys & {"name", "country", "isArchived", "technologies", "asn", "_id"}:\n                    collection = "Operator"\n\n        if collection not in _GBS_COLLECTIONS:\n            return f"Error: Collection \'{collection}\' not allowed: {sorted(_GBS_COLLECTIONS)}"\n\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        try:\n            formatted = json.dumps(result, indent=2, default=str)\n        except Exception:\n            formatted = str(result)\n\n        MAX_CHARS = 50000\n        if len(formatted) > MAX_CHARS:\n            formatted = formatted[:MAX_CHARS] + (\n                f"\\n\\n... [truncated: {len(result)} total docs]"\n            )\n\n        return formatted\n\n    def add_statistic(type: str, operator_id: str, period: str, tech: str, channel: str,\n                      domain: str, subscribers: int, is_estimate: bool = False,\n                      restated: bool = False, notes: str = "", created_by: str = "gbs-agent",\n                      ctx=None) -> str:\n        """Insert a pending Statistics record (state=1). Auto-fills operator/country, checks dupes."""\n        if type not in VALID_TYPES:\n            return f"Invalid type: {type}. Must be {VALID_TYPES}"\n        if channel not in VALID_CHANNELS:\n            return f"Invalid channel: {channel}. Must be {VALID_CHANNELS}"\n        if domain not in VALID_DOMAINS:\n            return f"Invalid domain: {domain}. Must be {VALID_DOMAINS}"\n        if not _valid_object_id(operator_id):\n            return "Invalid operator_id: must be 24 hex chars"\n        if not _valid_period(period):\n            return f"Invalid period: \'{period}\'. Must be YYYYQN (e.g. \'2025Q3\')"\n        if subscribers < 0:\n            return f"Invalid subscribers: {subscribers}."\n\n        r = json.loads(mongo.insert_statistic_safe(operator_id, type, period, tech, channel,\n                       domain, subscribers, created_by, is_estimate, restated, notes or None))\n        if not r.get("ok"):\n            return r.get("error", "Insert failed")\n        return (f"Created statistic {r[\'insertedId\']} (pending)\\n  {r[\'operator\']} ({r[\'country\']}) | "\n                f"{period} | {type} | {r[\'tech\']} | {channel} | {domain} | subs={subscribers:,}\\n"\n                f"  Admin: {_admin_url(operator_id)}")\n\n    def create_source(operator_id: str, year: int, quarter: int, type: str,\n                      url: str = "", file_url: Optional[str] = None, ctx=None) -> str:\n        """Create a Source record linking data to its origin. Optional S3 archival."""\n        from urllib.parse import urlparse\n        from point_topic_mcp.core import s3_utils as s3u\n\n        if not _valid_object_id(operator_id):\n            return "Invalid operator_id"\n        if quarter < 1 or quarter > 4:\n            return f"Invalid quarter: {quarter}"\n        if year < 1990 or year > 2100:\n            return f"Invalid year: {year}"\n        web_url = url.strip() or None\n        if not web_url and not (file_url and file_url.strip()):\n            return "At least one of url or file_url required"\n        raw_file = file_url.strip().lower() if file_url and file_url.strip() else ""\n        if raw_file and not (raw_file.startswith("s3://") or raw_file.startswith("http")):\n            return "file_url must be s3://... or http(s)://..."\n\n        op = mongo.find_one_operator_by_id(operator_id)\n        if not op:\n            return f"Operator not found: {operator_id}"\n\n        final_file_url = None\n        if raw_file and file_url:\n            try:\n                if raw_file.startswith("s3://"):\n                    b, k = s3u.parse_s3_uri(file_url)\n                    fn = k.rstrip("/").rsplit("/", 1)[-1] or "file"\n                    dk = s3u.build_sources_dest_key(operator_id=operator_id, year=year, quarter=quarter, original_filename=fn)\n                    final_file_url = s3u.copy_to_sources_bucket(source_bucket=b, source_key=k, dest_key=dk)\n                else:\n                    body, ct = s3u.download_from_url(file_url)\n                    fn = urlparse(file_url).path.rstrip("/").rsplit("/", 1)[-1] or "download"\n                    dk = s3u.build_sources_dest_key(operator_id=operator_id, year=year, quarter=quarter, original_filename=fn)\n                    final_file_url = s3u.upload_to_sources_bucket(body, ct, dk)\n            except RuntimeError as e:\n                return str(e)\n\n        for s in mongo.find_sources(operator_id, year, quarter):\n            if s.get("type") == type and s.get("url") == web_url and s.get("fileUrl") == final_file_url:\n                return f"Duplicate source (id={mongo._oid_from_doc(s)})"\n\n        iid = mongo.insert_one_source({"operatorId": operator_id, "year": year, "quarter": quarter,\n                                        "type": type, "url": web_url, "fileUrl": final_file_url})\n        return f"Created source {iid}\\n  {op.get(\'name\',\'\')} | {year}Q{quarter} | type={type} | url={web_url or final_file_url or \'none\'}"\n'}]