[{'Text': '    def gbs_add_statistic(\n        type: str,\n        operator_id: str,\n        period: str,\n        tech: str,\n        channel: str,\n        domain: str,\n        subscribers: int,\n        is_estimate: bool = False,\n        restated: bool = False,\n        notes: str = "",\n        created_by: str = "gbs-agent",\n        ctx=None,\n    ) -> str:\n        """Insert a GBS Statistics record with state: 1 (pending).\n\n        GBS (Global Broadband Statistics) data entry tool.\n        Creates a pending record that an admin will review and approve/publish.\n\n        Operator name and country are auto-filled from the operator record\n        to prevent mismatches. Duplicate detection prevents re-inserting\n        the same (type, tech, channel, domain) combo for the same period.\n\n        Args:\n            type:        broadband | mobile | iptv\n            operator_id: Operator ObjectId (from gbs_get_status output)\n            period:      Quarter string (e.g. "2025Q3")\n            tech:        Technology (must match operator\'s technologies list)\n            channel:     Infrastructure | Retail\n            domain:      Residential | Business | Total\n            subscribers: Subscriber count (must be >= 0)\n            is_estimate: Whether this value is estimated (default: False)\n            restated:    Whether this value was restated from a previous period\n            notes:       Optional notes about this data point\n            created_by:  Who created this record (email or agent name)\n\n        Returns:\n            Summary of the created record or an error message.\n        """\n        if type not in VALID_TYPES:\n            return f"Invalid type: {type}. Must be one of {VALID_TYPES}"\n        if channel not in VALID_CHANNELS:\n            return f"Invalid channel: {channel}. Must be one of {VALID_CHANNELS}"\n        if domain not in VALID_DOMAINS:\n            return f"Invalid domain: {domain}. Must be one of {VALID_DOMAINS}"\n        if not _valid_object_id(operator_id):\n            return "Invalid operator_id: must be 24 hex characters"\n        if not _valid_period(period):\n            return f"Invalid period: \'{period}\'. Must be format YYYYQN (e.g. \'2025Q3\')"\n        if subscribers < 0:\n            return f"Invalid subscribers: {subscribers}. Must be >= 0"\n\n        # Single mongosh call: lookup operator, validate tech, check dupes, insert\n        result_raw = mongo.insert_statistic_safe(\n            operator_id=operator_id,\n            type_val=type,\n            period=period,\n            tech_val=tech,\n            channel=channel,\n            domain=domain,\n            subscribers=subscribers,\n            created_by=created_by,\n            is_estimate=is_estimate,\n            restated=restated,\n            notes=notes or None,\n        )\n        result = json.loads(result_raw)\n\n        if not result.get("ok"):\n            return result.get("error", "Insert failed: unknown error")\n\n        inserted_id = result["insertedId"]\n        op_name = result.get("operator", "")\n        op_country = result.get("country", "")\n        tech_match = result.get("tech", tech)\n        admin_url = _gbs_admin_url(operator_id)\n        return (\n            f"Created statistic {inserted_id} (state: pending)\\n"\n            f"  {op_name} ({op_country}) | {period} | {type} | {tech_match} | "\n            f"{channel} | {domain} | subscribers={subscribers:,}\\n"\n            f"  Admin: {admin_url}"\n        )\n\n    def gbs_create_source(\n        operator_id: str,\n        year: int,\n        quarter: int,\n        type: str,\n        url: str = "",\n        file_url: Optional[str] = None,\n        ctx=None,\n    ) -> str:\n        """Insert a GBS Source record linking an operator\'s data to its origin.\n\n        GBS (Global Broadband Statistics) source attribution tool.\n        Associates a statistics record with the webpage, PDF, image, or file\n        where the subscriber numbers were found. Sources can optionally be\n        archived to S3 for permanent storage.\n\n        Args:\n            operator_id: Operator ObjectId (from gbs_get_status output)\n            year:        Year (e.g. 2025)\n            quarter:     Quarter 1-4\n            type:        Source type (e.g. "report", "earnings_call", "regulatory_filing")\n            url:         Optional web page link — stored as-is in MongoDB ``url``\n            file_url:    Optional file reference. Archived into ``pt-research-sources``;\n                         MongoDB ``fileUrl`` becomes the permanent HTTPS URL there.\n                - ``s3://bucket/key`` — same-account ``copy_object``\n                - ``http://`` or ``https://`` — download then upload to S3\n\n        Returns:\n            Summary of the created source record or an error message.\n        """\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: must be 24 hex characters"\n        if quarter < 1 or quarter > 4:\n            return f"Invalid quarter: {quarter}. Must be 1-4"\n        if year < 1990 or year > 2100:\n            return f"Invalid year: {year}. Must be 1990-2100"\n        if not type or not type.strip():\n            return "Invalid type: must not be empty"\n\n        web_url = url.strip() if url and url.strip() else None\n        has_file = bool(file_url and file_url.strip())\n        if not web_url and not has_file:\n            return "At least one of url or file_url must be provided"\n\n        if has_file:\n            raw_chk = file_url.strip()  # type: ignore[union-attr]\n            lo = raw_chk.lower()\n            if not (\n                lo.startswith("s3://")\n                or lo.startswith("http://")\n                or lo.startswith("https://")\n            ):\n                return (\n                    "file_url must be s3://bucket/key (in-account copy) or "\n                    "http(s)://... (download then archive)"\n                )\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        op_name = op.get("name", "")\n\n        final_file_url: Optional[str] = None\n        if has_file:\n            raw = file_url.strip()  # type: ignore[union-attr]\n            lower = raw.lower()\n            try:\n                if lower.startswith("s3://"):\n                    src_bucket, src_key = s3u.parse_s3_uri(raw)\n                    orig_name = src_key.rstrip("/").rsplit("/", 1)[-1] or "file"\n                    dest_key = s3u.build_sources_dest_key(\n                        operator_id=operator_id,\n                        year=year,\n                        quarter=quarter,\n                        original_filename=orig_name,\n                    )\n                    final_file_url = s3u.copy_to_sources_bucket(\n                        source_bucket=src_bucket,\n                        source_key=src_key,\n                        dest_key=dest_key,\n                    )\n                elif lower.startswith("http://") or lower.startswith("https://"):\n                    body, ct = s3u.download_from_url(raw)\n                    path = urlparse(raw).path or ""\n                    orig_name = path.rstrip("/").rsplit("/", 1)[-1] or "download"\n                    dest_key = s3u.build_sources_dest_key(\n                        operator_id=operator_id,\n                        year=year,\n                        quarter=quarter,\n                        original_filename=orig_name,\n                    )\n                    final_file_url = s3u.upload_to_sources_bucket(\n                        body=body,\n                        content_type=ct,\n                        dest_key=dest_key,\n                    )\n            except RuntimeError as e:\n                return str(e)\n\n        existing = mongo.find_sources(operator_id, year, quarter)\n        for s in existing:\n            if (s.get("type") == type\n                    and s.get("url") == web_url\n                    and s.get("fileUrl") == final_file_url):\n                eid = mongo._oid_from_doc(s)\n                return f"Duplicate: source already exists (id={eid}). Not inserted."\n\n        doc = {\n            "operatorId": operator_id,\n            "year": year,\n            "quarter": quarter,\n            "type": type,\n            "url": web_url,\n            "fileUrl": final_file_url,\n        }\n        inserted_id = mongo.insert_one_source(doc)\n        summary_url = web_url or final_file_url or "none"\n        return (\n            f"Created source {inserted_id}\\n"\n            f"  {op_name} | {year}Q{quarter} | type={type} | url={summary_url}"\n        )\n\nelse:\n    # Stub: tools unavailable when PT_RESEARCH_DATABASE_URI not set\n\n    def gbs_get_status(country: str = "", operator: str = "", period: str = "", ctx=None) -> str:\n        """GBS tools unavailable: PT_RESEARCH_DATABASE_URI not set."""\n        return "GBS tools require PT_RESEARCH_DATABASE_URI environment variable."\n\n    def gbs_add_statistic(**kwargs) -> str:\n        """GBS tools unavailable: PT_RESEARCH_DATABASE_URI not set."""\n        return "GBS tools require PT_RESEARCH_DATABASE_URI environment variable."\n\n    def gbs_create_source(**kwargs) -> str:\n        """GBS tools unavailable: PT_RESEARCH_DATABASE_URI not set."""\n        return "GBS tools require PT_RESEARCH_DATABASE_URI environment variable."\n'}]