f8cd37ac8c
Includes: API (Express/TypeScript), Astro site, Python workers, document generators, FCC compliance tools, Canada CRTC formation, Ansible infrastructure, and deployment scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
97 lines
3.8 KiB
Python
97 lines
3.8 KiB
Python
"""SIP Navigator (Cataleya Orchid One) preset.
|
|
|
|
Orchid One publishes a REST CDR API in newer deployments, but older ones
|
|
only expose the web CDR-export page. The preset tries the API first
|
|
(requires ``api_key``); if the admin doesn't have API access, it falls
|
|
back to the Playwright web-scrape flow.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.request
|
|
from datetime import datetime, timedelta
|
|
from typing import Iterable, Optional
|
|
|
|
from .base import BasePreset, CredentialField, FetchedFile
|
|
from ._scrape_base import ScrapePreset
|
|
|
|
|
|
class SIPNavigatorPreset(ScrapePreset):
|
|
PRESET_SLUG = "sip_navigator"
|
|
LABEL = "SIP Navigator (Cataleya Orchid One)"
|
|
CDR_FORMAT = "generic_csv"
|
|
TRANSPORT_METHOD = "api" # preferred; scrape fallback if api_key absent
|
|
DEFAULT_CRON = "0 3 * * *"
|
|
|
|
CREDENTIAL_FIELDS = (
|
|
CredentialField("admin_url", "Orchid One admin URL", "text",
|
|
help="e.g. https://orchid.example.com"),
|
|
CredentialField("username", "Admin username", "text"),
|
|
CredentialField("password", "Admin password", "password", sensitive=True),
|
|
CredentialField("api_key", "Orchid One API key", "password",
|
|
required=False, sensitive=True,
|
|
help="Leave blank if the deployment doesn't expose the REST API."),
|
|
)
|
|
|
|
FORMAT_CONFIG = {
|
|
"start_time": "setup_time",
|
|
"caller_number": "calling_party",
|
|
"called_number": "called_party",
|
|
"duration_sec": "answer_duration",
|
|
"billed_amount": "charge_amount",
|
|
"call_id": "global_call_id",
|
|
"trunk_group": "ingress_peering",
|
|
"disposition": "release_reason",
|
|
"ts_format": "%Y-%m-%dT%H:%M:%S",
|
|
}
|
|
|
|
# ── API path (preferred) ────────────────────────────────────────
|
|
|
|
def validate(self, profile_config: dict, secrets: dict) -> tuple[bool, str]:
|
|
if secrets.get("api_key"):
|
|
try:
|
|
base = profile_config["admin_url"].rstrip("/")
|
|
req = urllib.request.Request(
|
|
f"{base}/api/v1/system/status",
|
|
headers={"Authorization": f"Bearer {secrets['api_key']}"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
return True, f"API reachable (HTTP {resp.status})"
|
|
except Exception as exc:
|
|
return False, f"Orchid API validate failed: {exc}"
|
|
# No API key — fall back to ScrapePreset.validate() (HEAD the admin URL)
|
|
return super().validate(profile_config, secrets)
|
|
|
|
def fetch(
|
|
self,
|
|
profile_config: dict,
|
|
secrets: dict,
|
|
since: Optional[datetime],
|
|
) -> Iterable[FetchedFile]:
|
|
if secrets.get("api_key"):
|
|
return self._fetch_api(profile_config, secrets, since)
|
|
# No API key → scrape (scaffolded, raises until recon finalized)
|
|
return super().fetch(profile_config, secrets, since)
|
|
|
|
def _fetch_api(
|
|
self,
|
|
cfg: dict,
|
|
secrets: dict,
|
|
since: Optional[datetime],
|
|
) -> Iterable[FetchedFile]:
|
|
base = cfg["admin_url"].rstrip("/")
|
|
since = since or (datetime.utcnow() - timedelta(days=1))
|
|
end = datetime.utcnow()
|
|
req = urllib.request.Request(
|
|
f"{base}/api/v1/cdrs?start={since:%Y-%m-%dT%H:%M:%SZ}"
|
|
f"&end={end:%Y-%m-%dT%H:%M:%SZ}&format=csv",
|
|
headers={"Authorization": f"Bearer {secrets['api_key']}",
|
|
"Accept": "text/csv"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
content = resp.read()
|
|
yield FetchedFile(
|
|
remote_path=f"sip_navigator_cdrs_{end:%Y%m%dT%H%M%SZ}.csv",
|
|
mtime=end, content=content, size_bytes=len(content),
|
|
)
|