[{'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 collections import defaultdict\nfrom typing import Optional\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\nADMIN_BASE = "https://pt-research-app.vercel.app"\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 _run_status_query(country: str, operator: str | None = None, period: str | None = None) -> dict:\n        """Single mongosh call: resolve current period + $lookup aggregation."""\n        match: dict = {"country": country, "isArchived": {"$ne": True}}\n        if operator:\n            match["name"] = operator\n\n        pipeline = [\n            {"$match": match},\n            {"$lookup": {\n                "from": "Statistics",\n                "let": {"oid": "$_id", "tp": "__PERIOD__"},\n                "pipeline": [{"$match": {"$expr": {"$and": [\n                    {"$eq": ["$operatorId", "$$oid"]},\n                    {"$eq": ["$period", "$$tp"]}\n                ]}}}],\n                "as": "stats"\n            }},\n            {"$addFields": {"statCount": {"$size": "$stats"}}},\n            {"$project": {"name": 1, "techs": "$technologies", "statCount": 1, "stats": 1}},\n            {"$sort": {"name": 1}},\n        ]\n\n        period_expr = f\'"{period}"\' if period else "null"\n        pipe_json = json.dumps(pipeline)\n\n        js = f"""\n        const gv = db.GlobalVariables.findOne({{isCurrent: true}});\n        const cp = gv ? gv.currentStatsPeriod : null;\n        const tp = {period_expr} || cp;\n        const pipe = JSON.parse({json.dumps(pipe_json)}.replace(\'"__PERIOD__"\', JSON.stringify(tp)));\n        const docs = db.Operator.aggregate(pipe).toArray();\n        EJSON.stringify({{cp, tp, docs}});\n        """\n        result = mongo._run_mongosh(js, timeout=60)\n        parsed = json.loads(result) if result else {}\n        return parsed if parsed else {"error": "No data returned"}\n\n    def get_status(country: str = "", operator: str = "", period: str = "", ctx=None) -> str:\n        """GBS status at any level: country coverage → operator detail → period drill-down.\n\n        country only → coverage overview (total/with/without data, top operators)\n        + operator   → that operator\'s records, gaps, state breakdown\n        + period     → specific quarter (default: current from GlobalVariables)\n        """\n        if not country:\n            return "Error: country is required (e.g. \'United Kingdom\')"\n\n        data = _run_status_query(country, operator.strip() or None, period.strip() or None)\n        if data.get("error"):\n            return f"Query failed: {data[\'error\']}"\n\n        cp = data.get("cp", "?") or "?"\n        tp = data.get("tp", cp) or cp\n        docs = data.get("docs", [])\n\n        if operator.strip():\n            if not docs:\n                return f"Operator \'{operator}\' not found in \'{country}\'."\n            d = docs[0]\n            return _fmt_operator(d["name"], d.get("_id", {}).get("$oid", "?"), d.get("techs", []), d.get("stats", []), tp, country)\n\n        return _fmt_country(docs, cp, country)\n\n    def _fmt_country(docs, period, country):\n        wd = [d for d in docs if d.get("statCount", 0) > 0]\n        wo = [d for d in docs if d.get("statCount", 0) == 0]\n        total = len(docs)\n        lines = [f"GBS STATUS: {country} — period {period}",\n                 f"Operators: {total} total | {len(wd)} with data | {len(wo)} without data"]\n        if total > 0:\n            lines.append(f"Coverage: {round(100 * len(wd) / total, 1)}%")\n        lines.append("")\n        if wd:\n            lines.append("OPERATORS WITH DATA:")\n            for d in sorted(wd, key=lambda x: -x["statCount"])[:50]:\n                lines.append(f"  {d[\'name\'][:40]:40s} {d[\'statCount\']} rec")\n        if wo:\n            shown = wo[:30]\n            lines.append(f"\\nOPERATORS WITHOUT DATA ({len(shown)} of {len(wo)}):")\n            lines.append(f"  {\', \'.join(d[\'name\'] for d in shown)}")\n            if len(wo) > 30:\n                lines.append(f"  ... +{len(wo) - 30} more")\n        lines.append(f"\\nAdmin: {ADMIN_BASE}/admin")\n        return "\\n".join(lines)\n\n    def _fmt_operator(name, oid, techs, stats, period, country):\n        by_state, by_type, records, combos = defaultdict(int), defaultdict(int), [], set()\n        state_names = {1: "pending", 2: "approved", 3: "published"}\n\n        for s in stats:\n            by_state[s.get("state", 0)] += 1\n            by_type[s.get("type", "?")] += 1\n            combos.add((s.get("type"), s.get("tech"), s.get("channel"), s.get("domain")))\n            records.append(f"  {s.get(\'type\',\'?\'):10s} {s.get(\'tech\',\'?\'):18s} {s.get(\'channel\',\'?\'):15s} "\n                           f"{s.get(\'domain\',\'?\'):12s} {s.get(\'subscribers\',0):>10,}  "\n                           f"{state_names.get(s.get(\'state\',0), \'?\')}")\n\n        tus = sorted(set(s.get("tech") for s in stats)) or techs[:3] or ["?"]\n        gaps = [f"  {t}/{tech}/{ch}/{dom}" for t in VALID_TYPES\n                for tech in tus for ch in VALID_CHANNELS for dom in VALID_DOMAINS\n                if (t, tech, ch, dom) not in combos]\n\n        sf = ", ".join(f"{state_names.get(k, k)}: {v}" for k, v in sorted(by_state.items())) or "none"\n        tf = ", ".join(f"{k}: {v}" for k, v in sorted(by_type.items())) or "none"\n        tstr = ", ".join(techs[:10]) + ("..." if len(techs) > 10 else "")\n\n        lines = [f"GBS STATUS: {name} ({country}) — {period}", "",\n                 f"ID:        {oid}", f"Techs:     {tstr}",\n                 f"Records:   {len(stats)} | Filled: {len(combos)} combos | Gaps: {len(gaps)}",\n                 f"States:    {sf}", f"By type:   {tf}",\n                 f"Admin:     {_admin_url(oid)}", ""]\n        if records:\n            lines.append("EXISTING RECORDS:")\n            lines.extend(records)\n            lines.append("")\n        if gaps:\n            lines.append(f"GAPS ({len(gaps)}):")\n            lines.extend(gaps[:20])\n            if len(gaps) > 20:\n                lines.append(f"  ... +{len(gaps) - 20} more")\n\n        return "\\n".join(lines)\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'}]