"""ActorConsole SDK for Python 3.10+. Standard library only; copy into your Actor."""
import asyncio
import json
import os
import time
import traceback
import urllib.request
import urllib.parse
import uuid
import re


class ActorConsole:
    def __init__(self, key=None, endpoint=None):
        self.key = key or os.getenv("ACTORCONSOLE_KEY", "")
        if not re.fullmatch(r"aci_[a-f0-9]{64}", self.key):
            raise ValueError("Set an actor-specific ACTORCONSOLE_KEY from SDK setup.")
        self.endpoint = (endpoint or os.getenv("ACTORCONSOLE_ENDPOINT", "https://actorconsole.com/api/telemetry")).rstrip("/")
        parsed = urllib.parse.urlparse(self.endpoint)
        if parsed.scheme != "https" and not (parsed.scheme == "http" and parsed.hostname in ("localhost", "127.0.0.1", "::1")):
            raise ValueError("ActorConsole endpoint must use HTTPS.")
        self.run_id = os.getenv("ACTOR_RUN_ID") or os.getenv("APIFY_ACTOR_RUN_ID") or "local-" + str(uuid.uuid4())
        self.user_id = os.getenv("APIFY_USER_ID")
        self.paid = os.getenv("APIFY_USER_IS_PAYING") == "1"
        self.started_at = time.monotonic()

    async def _request(self, path, payload):
        def send():
            request = urllib.request.Request(self.endpoint + "/" + path, data=json.dumps(payload).encode(), headers={"Authorization": "Bearer " + self.key, "Content-Type": "application/json"}, method="POST")
            # Refuse redirects so credentials cannot be forwarded to another origin.
            class NoRedirect(urllib.request.HTTPRedirectHandler):
                def redirect_request(self, req, fp, code, msg, headers, newurl):
                    return None
            with urllib.request.build_opener(NoRedirect).open(request, timeout=5) as response:
                result = json.load(response)
                if not result.get("ok"):
                    raise RuntimeError("ActorConsole rejected the request.")
                return result.get("data")
        return await asyncio.to_thread(send)

    async def _report(self, path, payload):
        try:
            await self._request(path, payload)
            return True
        except Exception:
            print("ActorConsole: " + path + " report could not be delivered. Check SDK setup.")
            return False

    async def check_gate(self):
        if os.getenv("APIFY_IS_AT_HOME") != "1":
            return {"admitted": True, "reason": "Local development"}
        if not self.user_id:
            raise RuntimeError("Apify did not provide a user ID. Admission cannot be checked.")
        result = await self._request("gate", {"runId": self.run_id, "userId": self.user_id, "paid": self.paid})
        if not isinstance(result, dict) or not isinstance(result.get("admitted"), bool):
            raise RuntimeError("Invalid ActorConsole admission response.")
        return result

    async def capture_error(self, error, context=None):
        return await self._report("errors", {"eventId": str(uuid.uuid4()), "runId": self.run_id, "name": type(error).__name__, "message": str(error)[:8000] or type(error).__name__, "stack": "".join(traceback.format_exception(type(error), error, error.__traceback__))[:16000], "context": context or {}})

    async def track_cost(self, source, amount_usd, event_id=None):
        return await self._report("costs", {"eventId": event_id or str(uuid.uuid4()), "runId": self.run_id, "source": source, "amountUsd": amount_usd})

    async def report_run(self, status):
        payload = {"runId": self.run_id, "paid": self.paid, "status": status, "durationSeconds": time.monotonic() - self.started_at}
        if self.user_id:
            payload["userId"] = self.user_id
        allocated = os.getenv("ACTOR_MEMORY_MBYTES") or os.getenv("APIFY_MEMORY_MBYTES")
        if allocated and allocated.isdigit() and int(allocated) > 0:
            payload["memoryAllocatedMb"] = int(allocated)
        try:
            import resource
            import sys
            peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
            payload["memoryPeakMb"] = peak / (1024 * 1024 if sys.platform == "darwin" else 1024)
        except ImportError:
            pass
        return await self._report("runs", payload)

    async def __aenter__(self):
        gate = await self.check_gate()
        if not gate["admitted"]:
            raise RuntimeError("Free-user run limit: " + gate["reason"])
        return self

    async def __aexit__(self, error_type, error, tb):
        if error is not None:
            await self.capture_error(error)
        await self.report_run("FAILED" if error is not None else "SUCCEEDED")
        return False