[{'Text': '"""MongoDB utils using mongosh (shell) for robust connectivity.\n\nUses mongosh subprocess instead of PyMongo - avoids Python SSL issues that\nfail in some cloud/EC2 environments. mongosh has its own SSL stack and\ntlsAllowInvalidCertificates works reliably.\n\nRequires: mongosh installed on the system.\n  - Ubuntu/Debian: apt install mongodb-mongosh\n  - Amazon Linux/RHEL: yum install mongodb-mongosh\n  - macOS: brew install mongosh\n"""\n\nimport json\nimport os\nimport subprocess\nfrom typing import Any\n\nfrom dotenv import load_dotenv\n\nload_dotenv()\n\n\n# Collections (pt-research-app Prisma schema)\nOPERATOR_COLLECTION = "Operator"\nSTATISTICS_COLLECTION = "Statistics"\nSOURCES_COLLECTION = "Sources"\nGLOBAL_VARIABLES_COLLECTION = "GlobalVariables"\n\n\ndef _js_str(s: str | None) -> str:\n    """Escape string for safe use in mongosh JS. Returns JSON-formatted string or null."""\n    if s is None:\n        return "null"\n    return json.dumps(str(s))\n\n\ndef _uri_with_tls_option() -> str:\n    uri = os.getenv("PT_RESEARCH_DATABASE_URI")\n    if not uri:\n        raise RuntimeError("PT_RESEARCH_DATABASE_URI is not set")\n    if os.getenv("PT_RESEARCH_TLS_SKIP_VERIFY", "").strip() in ("1", "true", "yes"):\n        sep = "&" if "?" in uri else "?"\n        uri = f"{uri}{sep}tlsAllowInvalidCertificates=true"\n    return uri\n\n\ndef _run_mongosh(eval_js: str, timeout: int = 15) -> str:\n    """Run mongosh --eval and return stdout. Raises on failure."""\n    uri = _uri_with_tls_option()\n    try:\n        result = subprocess.run(\n            ["mongosh", uri, "--eval", eval_js, "--quiet"],\n            capture_output=True,\n            text=True,\n            timeout=timeout,\n        )\n    except FileNotFoundError:\n        raise RuntimeError(\n            "mongosh not found. GBS tools require mongosh. "\n            "Install with: point-topic-mcp-install-mongosh  (or sudo point-topic-mcp-install-mongosh on Linux)"\n        ) from None\n    if result.returncode != 0:\n        raise RuntimeError(\n            f"mongosh failed: {result.stderr or result.stdout or result.returncode}"\n        )\n    return result.stdout.strip()\n\n\ndef _oid_from_doc(doc: dict) -> str:\n    """Extract ObjectId string from EJSON doc (handles {_id: {$oid: "..."}})."""\n    if not doc:\n        return ""\n    id_val = doc.get("_id")\n    if isinstance(id_val, dict) and "$oid" in id_val:\n        return id_val["$oid"]\n    if isinstance(id_val, str):\n        return id_val\n    return str(id_val)\n\n\ndef find_operators(country: str = "") -> list[dict[str, Any]]:\n    """List operators. country="" for all."""\n    query = f\'{{country: {_js_str(country)}}}\' if country else "{}"\n    # Use EJSON.stringify for proper ObjectId handling\n    js = f"EJSON.stringify(db.{OPERATOR_COLLECTION}.find({query}).toArray())"\n    out = _run_mongosh(js)\n    data = json.loads(out) if out else []\n    return data if isinstance(data, list) else [data]\n\n\ndef find_one_operator(name: str, country: str) -> dict | None:\n    """Find operator by name and country."""\n    query = f\'{{name: {_js_str(name)}, country: {_js_str(country)}}}\'\n    js = f"EJSON.stringify(db.{OPERATOR_COLLECTION}.findOne({query}))"\n    out = _run_mongosh(js)\n    data = json.loads(out) if out else None\n    return data if data and data.get("_id") else None\n\n\ndef find_one_operator_by_id(oid_str: str) -> dict | None:\n    """Find operator by ObjectId string."""\n    # mongosh: ObjectId("hex")\n    query = f\'ObjectId("{oid_str}")\'\n    js = f"EJSON.stringify(db.{OPERATOR_COLLECTION}.findOne({{_id: {query}}}))"\n    out = _run_mongosh(js)\n    data = json.loads(out) if out else None\n    return data if data and data.get("_id") else None\n\n\ndef find_statistics(operator_id_oid: str, period: str) -> list[dict]:\n    """Find statistics for operator and period."""\n    period_js = _js_str(period)\n    js = f"""EJSON.stringify(\n      db.{STATISTICS_COLLECTION}.find({{\n        operatorId: ObjectId("{operator_id_oid}"),\n        period: {period_js}\n      }}).toArray()\n    )"""\n    out = _run_mongosh(js)\n    data = json.loads(out) if out else []\n    return data if isinstance(data, list) else [data]\n\n\ndef find_global_variables_current() -> dict | None:\n    """Get current GlobalVariables record."""\n    js = f"EJSON.stringify(db.{GLOBAL_VARIABLES_COLLECTION}.findOne({{isCurrent: true}}))"\n    out = _run_mongosh(js)\n    data = json.loads(out) if out else None\n    return data if data else None\n\n\ndef insert_one_statistic(doc: dict) -> str:\n    """Insert one Statistics document. Returns inserted ObjectId string."""\n    oid = doc["operatorId"]\n    op = doc["operator"].replace("\\\\", "\\\\\\\\").replace(\'"\', \'\\\\"\')\n    tech = doc["tech"].replace("\\\\", "\\\\\\\\").replace(\'"\', \'\\\\"\')\n    notes = (doc.get("notes") or "").replace("\\\\", "\\\\\\\\").replace(\'"\', \'\\\\"\').replace("\\n", "\\\\n")\n    created_by = (doc.get("createdBy") or "gbs-agent").replace("\\\\", "\\\\\\\\").replace(\'"\', \'\\\\"\')\n    js = f"""\n    const now = new Date();\n    const result = db.{STATISTICS_COLLECTION}.insertOne({{\n      type: "{doc["type"]}",\n      country: "{doc["country"]}",\n      operator: "{op}",\n      operatorId: ObjectId("{oid}"),\n      period: "{doc["period"]}",\n      channel: "{doc["channel"]}",\n      domain: "{doc["domain"]}",\n      state: 1,\n      tech: "{tech}",\n      subscribers: {doc["subscribers"]},\n      is_estimate: {str(doc.get("is_estimate", False)).lower()},\n      restated: {str(doc.get("restated", False)).lower()},\n      notes: {f\'"{notes}"\' if doc.get("notes") else "null"},\n      is_archived: false,\n      createdBy: "{created_by}",\n      updatedBy: "{created_by}",\n      date: now,\n      createdAt: now,\n      updatedAt: now\n    }});\n    EJSON.stringify(result.insertedId);\n    """\n    out = _run_mongosh(js)\n    data = json.loads(out) if out else {}\n    if isinstance(data, dict) and "$oid" in data:\n        return data["$oid"]\n    return str(data)\n\n\ndef insert_statistic_safe(\n    operator_id: str,\n    type_val: str,\n    period: str,\n    tech_val: str,\n    channel: str,\n    domain: str,\n    subscribers: int,\n    created_by: str = "gbs-agent",\n    is_estimate: bool = False,\n    restated: bool = False,\n    notes: str | None = None,\n) -> str:\n    """Combined operator lookup + duplicate check + insert in ONE mongosh call.\n\n    Returns JSON with either:\n      {{"ok": true, "insertedId": "..."}}\n      {{"ok": false, "error": "message ..."}}\n\n    Eliminates the 3-call cascade that caused timeouts.\n    """\n    op_esc = lambda s: (s or "").replace("\\\\", "\\\\\\\\").replace(\'"\', \'\\\\"\').replace("\\n", "\\\\n")\n    type_esc = op_esc(type_val)\n    tech_esc = op_esc(tech_val)\n    notes_esc = op_esc(notes) if notes else None\n    by_esc = op_esc(created_by)\n\n    js = f"""\n    const op = db.{OPERATOR_COLLECTION}.findOne({{_id: ObjectId("{operator_id}")}});\n    if (!op) {{ print(EJSON.stringify({{ok: false, error: "Operator not found: {operator_id}"}})); exit(0); }}\n\n    const techMatch = (op.technologies || []).find(\n      t => t.toLowerCase() === "{tech_esc}".toLowerCase()\n    );\n    if (!techMatch) {{\n      print(EJSON.stringify({{ok: false, error: "Tech \'{tech_esc}\' not valid for \'" + op.name + "\'. Valid: " + JSON.stringify(op.technologies)}}));\n      exit(0);\n    }}\n\n    const existing = db.{STATISTICS_COLLECTION}.find({{operatorId: ObjectId("{operator_id}"), period: "{period}"}}).toArray();\n    const dup = existing.find(s =>\n      s.type === "{type_esc}" && s.tech === techMatch && s.channel === "{channel}" && s.domain === "{domain}"\n    );\n    if (dup) {{\n      print(EJSON.stringify({{ok: false, error: "Duplicate: statistic already exists for {type_esc}/" + techMatch + "/{channel}/{domain} in {period} (id=" + dup._id + ", state=" + dup.state + ")"}}));\n      exit(0);\n    }}\n\n    const now = new Date();\n    const result = db.{STATISTICS_COLLECTION}.insertOne({{\n      type: "{type_esc}",\n      country: op.country,\n      operator: op.name,\n      operatorId: ObjectId("{operator_id}"),\n      period: "{period}",\n      channel: "{channel}",\n      domain: "{domain}",\n      state: 1,\n      tech: techMatch,\n      subscribers: {subscribers},\n      is_estimate: {str(is_estimate).lower()},\n      restated: {str(restated).lower()},\n      notes: {"null" if notes_esc is None else f\'"{notes_esc}"\'},\n      is_archived: false,\n      createdBy: "{by_esc}",\n      updatedBy: "{by_esc}",\n      date: now,\n      createdAt: now,\n      updatedAt: now\n    }});\n    print(EJSON.stringify({{ok: true, insertedId: result.insertedId, operator: op.name, country: op.country, tech: techMatch}}));\n    """\n    out = _run_mongosh(js, timeout=30)\n    data = json.loads(out) if out else {"ok": False, "error": "No output from mongosh"}\n    return json.dumps(data)\n\n\ndef insert_one_source(doc: dict) -> str:\n    """Insert one Source document. Returns inserted ObjectId string."""\n    oid = doc["operatorId"]\n    url = (doc.get("url") or "").replace("\\\\", "\\\\\\\\").replace(\'"\', \'\\\\"\')\n    file_url = (doc.get("fileUrl") or "").replace("\\\\", "\\\\\\\\").replace(\'"\', \'\\\\"\')\n    type_val = doc["type"].replace("\\\\", "\\\\\\\\").replace(\'"\', \'\\\\"\')\n    # Sources requires createdAt + updatedAt (Prisma); raw inserts must set BSON Date.\n    js = f"""\n    const now = new Date();\n    const result = db.{SOURCES_COLLECTION}.insertOne({{\n      operatorId: ObjectId("{oid}"),\n      year: {doc["year"]},\n      quarter: {doc["quarter"]},\n      type: "{type_val}",\n      url: {f\'"{url}"\' if doc.get("url") else "null"},\n      fileUrl: {f\'"{file_url}"\' if doc.get("fileUrl") else "null"},\n      createdAt: now,\n      updatedAt: now\n    }});\n    EJSON.stringify(result.insertedId);\n    """\n    out = _run_mongosh(js)\n    data = json.loads(out) if out else {}\n    if isinstance(data, dict) and "$oid" in data:\n        return data["$oid"]\n    return str(data)\n\n\ndef find_sources(operator_id_oid: str, year: int, quarter: int) -> list[dict]:\n    """Find sources for operator, year, and quarter."""\n    js = f"""EJSON.stringify(\n      db.{SOURCES_COLLECTION}.find({{\n        operatorId: ObjectId("{operator_id_oid}"),\n        year: {year},\n        quarter: {quarter}\n      }}).toArray()\n    )"""\n    out = _run_mongosh(js)\n    data = json.loads(out) if out else []\n    return data if isinstance(data, list) else [data]\n\n\ndef run_aggregation(pipeline: list, collection: str = "Operator", timeout: int = 60) -> list:\n    """Run an aggregation pipeline and return results as list of dicts.\n\n     Efficient batch query, key for replacing N+1 individual calls.\n    """\n    js = f"EJSON.stringify(db.{collection}.aggregate({json.dumps(pipeline)}).toArray())"\n    out = _run_mongosh(js, timeout=timeout)\n    return json.loads(out) if out else []\n\n\ndef mongosh_available() -> bool:\n    """Check if mongosh is installed and can connect."""\n    try:\n        _run_mongosh("db.runCommand({ping:1})")\n        return True\n    except Exception:\n        return False\n'}]