[{'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"]  # ObjectId string\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    # Prisma @default / @updatedAt do not apply to raw inserts — set BSON Date explicitly\n    # so the web app receives real Date fields (same instant for date / createdAt / updatedAt).\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    # Result is {"$oid":"..."} - extract the oid\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_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 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'}]