[{'Text': '        Insert a Statistics (GBS) record with state: 1 (pending).\n\n        Operator name and country are auto-filled from the operator record to prevent mismatches.\n\n        Args:\n            type: broadband | mobile | iptv\n            operator_id: Operator ObjectId (from list_operators or get_gbs_status)\n            period: Quarter string (e.g. "2025Q1")\n            tech: Technology (must match operator\'s technologies)\n            channel: Infrastructure | Retail\n            domain: Residential | Business | Total\n            subscribers: Subscriber count (must be >= 0)\n            is_estimate: Whether value is estimated\n            restated: Whether value was restated\n            notes: Optional notes\n            created_by: Creator identifier (e.g. user email)\n\n        Returns:\n            Inserted record summary or 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 f"Invalid operator_id: must be 24 hex chars"\n        if not _valid_period(period):\n            return f"Invalid period: \'{period}\'. Must be format YYYYQN (e.g. \'2025Q1\')"\n        if subscribers < 0:\n            return f"Invalid subscribers: {subscribers}. Must be >= 0"\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        op_country = op.get("country", "")\n        techs = op.get("technologies", [])\n        tech_match = next((t for t in techs if t.lower() == tech.lower()), None)\n        if not tech_match:\n            return f"Tech \'{tech}\' not valid for operator \'{op_name}\'. Valid: {techs}"\n\n        # Check for duplicate (same operator, period, type, tech, channel, domain)\n        existing = mongo.find_statistics(operator_id, period)\n        for s in existing:\n            if (s.get("type") == type and s.get("tech") == tech_match\n                    and s.get("channel") == channel and s.get("domain") == domain):\n                eid = mongo._oid_from_doc(s)\n                return (f"Duplicate: statistic already exists for "\n                        f"{type}/{tech_match}/{channel}/{domain} in {period} "\n                        f"(id={eid}, state={s.get(\'state\', \'?\')}). Not inserted.")\n\n        doc = {\n            "type": type,\n            "country": op_country,\n            "operator": op_name,\n            "operatorId": operator_id,\n            "period": period,\n            "tech": tech_match,\n            "channel": channel,\n            "domain": domain,\n            "subscribers": subscribers,\n            "is_estimate": is_estimate,\n            "restated": restated,\n            "notes": notes or None,\n            "createdBy": created_by,\n        }\n        inserted_id = mongo.insert_one_statistic(doc)\n        return (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    def 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        """\n        Insert a Source record for an operator/period.\n\n        Args:\n            operator_id: Operator ObjectId (from list_operators or get_gbs_status)\n            year: Year (e.g. 2025)\n            quarter: Quarter 1-4\n            type: Source type (e.g. "report", "earnings_call", "tariff", "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``; MongoDB\n                      ``fileUrl`` is always the permanent HTTPS URL there.\n                - ``s3://bucket/key`` — same-account ``copy_object`` (preferred for uploads).\n                - ``http://`` or ``https://`` — download (e.g. presigned GET or public URL) then upload.\n\n        Returns:\n            Inserted record summary or 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 f"Invalid operator_id: must be 24 hex chars"\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 (f"Created source {inserted_id}\\n"\n                f"  {op_name} | {year}Q{quarter} | type={type} | url={summary_url}")\n\n\ndef _run_test_flow():\n    """Run GBS tools with real MongoDB. Use: uv run python -m point_topic_mcp.tools.gbs_tools"""\n    import re\n    import sys\n\n    mod_name = "point_topic_mcp.tools.gbs_tools"\n    gbs = sys.modules.get(mod_name) or sys.modules.get("__main__")\n    if not gbs or not hasattr(gbs, "list_operators"):\n        print("SKIP: GBS tools not loaded (PT_RESEARCH_DATABASE_URI not set)")\n        return\n\n    print("=== GBS Tools Test Flow ===\\n")\n\n    print("1. list_operators(country=\'United Kingdom\'):")\n    try:\n        out = gbs.list_operators(country="United Kingdom")\n        print(out)\n    except Exception as e:\n        print(f"   FAIL: {e}")\n        return\n\n    # Extract first operator name and id from output\n    op_id = None\n    op_name = ""\n    country = "United Kingdom"\n    for line in out.split("\\n"):\n        if "|" in line and "id=" in line:\n            parts = line.split("|")\n            op_name = parts[0].strip()\n            m = re.search(r"id=([a-f0-9]{24})", line)\n            if m:\n                op_id = m.group(1)\n            break\n\n    if not op_id:\n        print("   No UK operators found - cannot continue test")\n        return\n\n    # Get the operator\'s first tech for a valid insert\n    op_tech = ""\n    for line in out.split("\\n"):\n        if op_id in line and "techs=[" in line:\n            m = re.search(r"techs=\\[\'([^\']+)\'", line)\n            if m:\n                op_tech = m.group(1)\n            break\n\n    print(f"\\n2. get_gbs_status(\'{country}\', \'{op_name}\', \'2025Q1\'):")\n    try:\n        status = gbs.get_gbs_status(country, op_name, "2025Q1")\n        print(status[:800] + ("..." if len(status) > 800 else ""))\n    except Exception as e:\n        print(f"   FAIL: {e}")\n\n    print("\\n3. add_statistic() validation:")\n    try:\n        result = gbs.add_statistic(\n            type="invalid",\n            operator_id="000000000000000000000001",\n            period="2025Q1",\n            tech="FTTP",\n            channel="Wrong",\n            domain="Residential",\n            subscribers=1000,\n        )\n        assert "Invalid" in result\n        print("   OK: rejected invalid type")\n    except Exception as e:\n        print(f"   FAIL: {e}")\n\n    print(f"\\n4. add_statistic() real insert for {op_name} (tech={op_tech}):")\n    try:\n        result = gbs.add_statistic(\n            type="broadband",\n            operator_id=op_id,\n            period="2099Q1",\n            tech=op_tech,\n            channel="Infrastructure",\n            domain="Residential",\n            subscribers=99999,\n            notes="__gbs_tools_test_flow__",\n            created_by="gbs-test-script",\n        )\n        print(f"   {result}")\n    except Exception as e:\n        print(f"   FAIL: {e}")\n\n    print(f"\\n5. create_source() real insert for {op_name}:")\n    try:\n        result = gbs.create_source(\n            operator_id=op_id,\n            year=2099,\n            quarter=1,\n            type="test",\n            url="https://example.com/gbs-test",\n        )\n        print(f"   {result}")\n    except Exception as e:\n        print(f"   FAIL: {e}")\n\n    print("\\n=== Test flow complete ===")\n    print("NOTE: test records inserted with period 2099Q1 / year 2099 — delete from dev when done")\n\n\nif __name__ == "__main__":\n    _run_test_flow()\n'}]