[{'Text': 'Edited point-topic-mcp/src/point_topic_mcp/tools/gbs_tools.py:\n\n```diff\n@@ -1,35 +1,24 @@\n-"""GBS (Global Broadband Statistics) tools for the PT Research App.\n+"""GBS (Global Broadband Statistics) tools.\n \n-GBS is a quarterly process: associates search the web for updated broadband\n-statistics from operators (most are required by law to report quarterly).\n-They find PDFs, images, or web data, extract subscriber numbers, and enter\n-them as **pending** records. Admins review and approve/publish.\n+GBS is a quarterly process: associates find operator subscriber numbers\n+(web/PDF/image), enter them as pending records, admins approve/publish.\n \n-MongoDB Collections (pt-research-app Prisma schema):\n-  Operator       — 1300+ operators globally (214 UK). Fields: name, country, technologies\n-  Statistics     — Subscriber counts. Fields: type, operatorId, period, tech, channel, domain,\n-                   subscribers, state (1=pending, 2=approved, 3=published)\n-  Sources        — Archival records linking statistics to source URLs/files\n-  GlobalVariables— Tracks currentStatsPeriod (e.g. "2025Q3") and currentTariffsPeriod\n+Collections (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 \n-Live data snapshot (2025Q3 — verified against production MongoDB):\n-  - UK operators: 175 total, 55 with data (31.4%), 120 without\n-  - Global records in 2025Q3: 1854 across 690 operators\n-  - Valid types:    broadband, mobile, iptv\n-  - Valid channels: Infrastructure, Retail\n-  - Valid domains:  Residential, Business, Total\n+Valid enums: types=[broadband, mobile, iptv]\n+             channels=[Infrastructure, Retail]\n+             domains=[Residential, Business, Total]\n \n-Tool workflow (agent-driven, no hardcoded flow):\n-  1. gbs_get_status(country) → overview of what\'s covered this quarter\n-  2. gbs_get_status(country, operator) → drill into a specific operator\'s records + gaps\n-  3. gbs_add_statistic(...) → insert a pending record when numbers are found\n-  4. gbs_create_source(...) → link source URLs/files to operator/period\n-\n-Uses mongosh subprocess (not PyMongo) for reliable SSL.\n-Requires: PT_RESEARCH_DATABASE_URI env var, mongosh on PATH.\n+Uses mongosh subprocess (not PyMongo). Requires PT_RESEARCH_DATABASE_URI.\n """\n \n import json\n+import re\n from collections import defaultdict\n from typing import Optional\n \n@@ -48,23 +37,16 @@\n     VALID_DOMAINS = ["Residential", "Business", "Total"]\n \n     def _valid_object_id(oid: str) -> bool:\n-        """Check if string is valid 24-char hex ObjectId."""\n-        if not oid or len(oid) != 24:\n-            return False\n-        return all(c in "0123456789abcdefABCDEF" for c in oid)\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-        """Check period matches format like \'2025Q1\'."""\n-        import re\n         return bool(re.match(r"^\\d{4}Q[1-4]$", p))\n \n-    def _gbs_admin_url(operator_id: str) -> str:\n-        """Admin link to review/publish this operator\'s statistics."""\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, period: str | None) -> dict:\n-        """Single mongosh call: resolve current period + run $lookup aggregation."""\n-        # Build the pipeline conditionally\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@@ -73,15 +55,11 @@\n             {"$match": match},\n             {"$lookup": {\n                 "from": "Statistics",\n-                "let": {"oid": "$_id", "tp": "__PERIOD_PLACEHOLDER__"},\n-                "pipeline": [\n-                    {"$match": {"$expr": {\n-                        "$and": [\n-                            {"$eq": ["$operatorId", "$$oid"]},\n-                            {"$eq": ["$period", "$$tp"]}\n-                        ]\n-                    }}}\n-                ],\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@@ -90,409 +68,180 @@\n         ]\n \n         period_expr = f\'"{period}"\' if period else "null"\n-        pipeline_json = json.dumps(pipeline)\n+        pipe_json = json.dumps(pipeline)\n \n         js = f"""\n         const gv = db.GlobalVariables.findOne({{isCurrent: true}});\n-        const currentPeriod = gv ? gv.currentStatsPeriod : null;\n-        const targetPeriod = {period_expr} || currentPeriod;\n-\n-        const pipeline = JSON.parse({json.dumps(pipeline_json)}.replace(\n-            \'"__PERIOD_PLACEHOLDER__"\', JSON.stringify(targetPeriod)\n-        ));\n-\n-        const docs = db.Operator.aggregate(pipeline).toArray();\n-        EJSON.stringify({{currentPeriod, targetPeriod, docs}});\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-        if not parsed:\n-            return {"error": "No data returned from MongoDB"}\n-        return parsed\n+        return parsed if parsed else {"error": "No data returned"}\n \n-    def gbs_get_status(\n-        country: str = "",\n-        operator: str = "",\n-        period: str = "",\n-        ctx=None,\n-    ) -> str:\n-        """GBS status at any zoom level: country overview → operator detail → period drill-down.\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-        GBS (Global Broadband Statistics) is a quarterly reporting process.\n-        Use this tool to understand what data exists, what\'s missing, and\n-        what needs to be entered for any operator/period combination.\n-\n-        Zoom levels:\n-          country only       → coverage overview: how many operators have data,\n-                               which ones are complete, which are missing entirely\n-          + operator         → that operator\'s records and gaps for the current period\n-          + period           → specific quarter (YYYYQN format, e.g. "2025Q3")\n-                               defaults to current period from GlobalVariables\n-\n-        Args:\n-            country:  Country name (exact match, e.g. "United Kingdom")\n-            operator: Operator name (exact match, case-sensitive).\n-                      Leave empty for country-level overview.\n-            period:   Quarter string (e.g. "2025Q3", "2024Q4").\n-                      Leave empty to use the current period.\n-\n-        Returns:\n-            Formatted status report. For country level: total operators,\n-            coverage split, lists of who has data and who doesn\'t.\n-            For operator level: records table, gap analysis, state breakdown, admin URL.\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\'). Use gbs_list_countries() to discover available countries."\n+            return "Error: country is required (e.g. \'United Kingdom\')"\n \n-        op_arg = operator.strip() if operator else None\n-        period_arg = period.strip() if period else None\n-\n-        data = _run_status_query(country, op_arg, period_arg)\n+        data = _run_status_query(country.strip() or None, 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("currentPeriod", "?")\n-        tp = data.get("targetPeriod", cp)\n-        docs = data.get("docs", [])\n+        cp, tp, docs = data.get("cp"), data.get("tp", data.get("cp")), data.get("docs", [])\n \n-        if op_arg:\n-            # ── Operator level ──\n+        if operator.strip():\n             if not docs:\n-                return (\n-                    f"Operator \'{operator}\' not found in \'{country}\'.\\n"\n-                    f"Check exact spelling (case-sensitive). "\n-                    f"Use gbs_get_status(country=\'{country}\') to see all operators."\n-                )\n-            return _format_operator_status(docs[0], tp, country)\n+                return f"Operator \'{operator}\' not found in \'{country}\'."\n+            name = docs[0]["name"]\n+            oid = docs[0].get("_id", {}).get("$oid", "?")\n+            techs = docs[0].get("techs", [])\n+            stats = docs[0].get("stats", [])\n+            return _fmt_operator(name, oid, techs, stats, tp, country)\n \n-        # ── Country level ──\n-        return _format_country_status(docs, cp, country)\n+        return _fmt_country(docs, cp, country)\n \n-    def _format_country_status(docs: list, period: str, country: str) -> str:\n-        """Format country-level roll-up."""\n-        with_data = [d for d in docs if d.get("statCount", 0) > 0]\n-        without = [d for d in docs if d.get("statCount", 0) == 0]\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-\n-        lines = [\n-            f"GBS STATUS: {country} — period {period}",\n-            f"Operators: {total} total | {len(with_data)} with data | {len(without)} without data",\n-        ]\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(with_data) / total, 1)}%")\n+            lines.append(f"Coverage: {round(100 * len(wd) / total, 1)}%")\n         lines.append("")\n-\n-        if with_data:\n-            lines.append("OPERATORS WITH DATA (by record count):")\n-            for d in sorted(with_data, key=lambda x: -x["statCount"])[:50]:\n-                record_label = "record" if d["statCount"] == 1 else "records"\n-                lines.append(f"  {d[\'name\'][:40]:40s} {d[\'statCount\']} {record_label}")\n-            lines.append("")\n-\n-        if without:\n-            shown = without[:30]\n-            lines.append(f"OPERATORS WITHOUT DATA ({len(shown)} of {len(without)} shown):")\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(without) > 30:\n-                lines.append(f"  ... and {len(without) - 30} more")\n-            lines.append("")\n-\n-        lines.append(f"Admin: {ADMIN_BASE}/admin")\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 _format_operator_status(doc: dict, period: str, country: str) -> str:\n-        """Format single-operator drill-down."""\n-        name = doc["name"]\n-        oid = doc.get("_id", {}).get("$oid", "?")\n-        techs = doc.get("techs", [])\n-        stats = doc.get("stats", [])\n-\n-        by_state: dict[int, int] = defaultdict(int)\n-        by_type: dict[str, int] = defaultdict(int)\n-        records: list[str] = []\n-        combos: set[tuple] = set()\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-            st = s.get("state", 0)\n-            by_state[st] += 1\n+            by_state[s.get("state", 0)] += 1\n             by_type[s.get("type", "?")] += 1\n-            combo = (s.get("type"), s.get("tech"), s.get("channel"), s.get("domain"))\n-            combos.add(combo)\n-            state_label = {1: "pending", 2: "approved", 3: "published"}.get(st, str(st))\n-            subs = s.get("subscribers", 0)\n-            records.append(\n-                f"  {s.get(\'type\', \'?\'):10s} {s.get(\'tech\', \'?\'):18s} "\n-                f"{s.get(\'channel\', \'?\'):15s} {s.get(\'domain\', \'?\'):12s} "\n-                f"{subs:>10,}  {state_label}"\n-            )\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-        # Gap analysis: only check techs that are actually used by this operator\n-        # (from their stats or from their profile\'s first 3 technologies)\n-        techs_in_use = sorted(set(s.get("tech") for s in stats)) or techs[:3]\n-        if not techs_in_use:\n-            techs_in_use = ["(unknown)"]\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-        gaps: list[str] = []\n-        for t in VALID_TYPES:\n-            for tech in techs_in_use:\n-                for ch in VALID_CHANNELS:\n-                    for dom in VALID_DOMAINS:\n-                        if (t, tech, ch, dom) not in combos:\n-                            gaps.append(f"  {t}/{tech}/{ch}/{dom}")\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-        state_names = {1: "pending", 2: "approved", 3: "published"}\n-        state_fmt = ", ".join(\n-            f"{state_names.get(k, str(k))}: {v}" for k, v in sorted(by_state.items())\n-        ) or "none"\n-\n-        type_fmt = ", ".join(f"{k}: {v}" for k, v in sorted(by_type.items())) or "none"\n-\n-        lines = [\n-            f"GBS STATUS: {name} ({country}) — {period}",\n-            "",\n-            f"ID:        {oid}",\n-            f"Techs:     {\', \'.join(techs[:10])}{\'...\' if len(techs) > 10 else \'\'}",\n-            f"Records:   {len(stats)} | Filled: {len(combos)} combos | Gaps: {len(gaps)}",\n-            f"States:    {state_fmt}",\n-            f"By type:   {type_fmt}",\n-            f"Admin:     {_gbs_admin_url(oid)}",\n-            "",\n-        ]\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-\n+            lines.append("EXISTING RECORDS:"), lines.extend(records), lines.append("")\n         if gaps:\n-            lines.append(f"GAPS ({len(gaps)}):")\n-            lines.extend(gaps[:20])\n+            lines.append(f"GAPS ({len(gaps)}):"), lines.extend(gaps[:20])\n             if len(gaps) > 20:\n-                lines.append(f"  ... and {len(gaps) - 20} more")\n-            lines.append("")\n+                lines.append(f"  ... +{len(gaps) - 20} more")\n \n         return "\\n".join(lines)\n \n-    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+    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 one of {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 one of {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 one of {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 characters"\n+            return "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. \'2025Q3\')"\n+            return f"Invalid period: \'{period}\'. Must be YYYYQN (e.g. \'2025Q3\')"\n         if subscribers < 0:\n-            return f"Invalid subscribers: {subscribers}. Must be >= 0"\n+            return f"Invalid subscribers: {subscribers}."\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+        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-        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+    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: must be 24 hex characters"\n+            return "Invalid operator_id"\n         if quarter < 1 or quarter > 4:\n-            return f"Invalid quarter: {quarter}. Must be 1-4"\n+            return f"Invalid quarter: {quarter}"\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+            return f"Invalid year: {year}"\n+        if not (web_url := (url.strip() or None)):\n+            if not file_url or not 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-        op_name = op.get("name", "")\n-\n-        final_file_url: Optional[str] = None\n-        if has_file:\n+        final_file_url = None\n+        if raw_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+                if raw.startswith("s3://"):\n+                    b, k = s3u.parse_s3_uri(raw)\n+                    dk = s3u.build_sources_dest_key(operator_id, year, quarter, k.rstrip("/").rsplit("/", 1)[-1] or "file")\n+                    final_file_url = s3u.copy_to_sources_bucket(b, k, dk)\n+                else:\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+                    fn = urlparse(raw).path.rstrip("/").rsplit("/", 1)[-1] or "download"\n+                    dk = s3u.build_sources_dest_key(operator_id, year, quarter, fn)\n+                    final_file_url = s3u.upload_to_sources_bucket(body, ct, dk)\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+        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-        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+        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 \n else:\n-    # Stub: tools unavailable when PT_RESEARCH_DATABASE_URI not set\n+    _STUB_MSG = "GBS tools require PT_RESEARCH_DATABASE_URI environment variable."\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+    def get_status(**kw): return _STUB_MSG\n+    def add_statistic(**kw): return _STUB_MSG\n+    def create_source(**kw): return _STUB_MSG\n\n```'}]