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>
74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
"""Grandstream UCM preset — web-UI CDR download via the local API.
|
|
|
|
Grandstream UCM62xx/63xx exposes a REST-style admin API at
|
|
``https://<host>/api`` with cookie-based auth. The CDR export endpoint
|
|
returns CSV; we reuse it via HTTPS transport.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Iterable, Optional
|
|
|
|
from .base import BasePreset, CredentialField, FetchedFile
|
|
from ..cdr_transports.https_transport import HTTPSTransport
|
|
|
|
|
|
class GrandstreamPreset(BasePreset):
|
|
PRESET_SLUG = "grandstream"
|
|
LABEL = "Grandstream UCM (62xx / 63xx)"
|
|
CDR_FORMAT = "generic_csv"
|
|
TRANSPORT_METHOD = "api"
|
|
DEFAULT_CRON = "0 2 * * *"
|
|
|
|
CREDENTIAL_FIELDS = (
|
|
CredentialField("host", "UCM host / IP", "text"),
|
|
CredentialField("port", "HTTPS port", "number", required=False,
|
|
help="Default 8089 for Grandstream admin API."),
|
|
CredentialField("username", "Admin username", "text"),
|
|
CredentialField("password", "Admin password", "password", sensitive=True),
|
|
)
|
|
|
|
FORMAT_CONFIG = {
|
|
"start_time": "AcctStartTime",
|
|
"caller_number": "src",
|
|
"called_number": "dst",
|
|
"duration_sec": "billsec",
|
|
"call_id": "uniqueid",
|
|
"trunk_group": "channel",
|
|
"disposition": "disposition",
|
|
"ts_format": "%Y-%m-%d %H:%M:%S",
|
|
}
|
|
|
|
def _transport(self, cfg: dict, secrets: dict) -> HTTPSTransport:
|
|
return HTTPSTransport(
|
|
host=cfg["host"],
|
|
port=int(cfg.get("port") or 8089),
|
|
username=cfg.get("username"),
|
|
password=secrets.get("password"),
|
|
remote_glob="cgi-bin/api/cdrapi?format=csv",
|
|
extra={"scheme": "https"},
|
|
)
|
|
|
|
def validate(self, profile_config: dict, secrets: dict) -> tuple[bool, str]:
|
|
try:
|
|
return self._transport(profile_config, secrets).validate()
|
|
except Exception as exc:
|
|
return False, f"{exc}"
|
|
|
|
def fetch(
|
|
self,
|
|
profile_config: dict,
|
|
secrets: dict,
|
|
since: Optional[datetime],
|
|
) -> Iterable[FetchedFile]:
|
|
# Grandstream's CDR endpoint returns the full export as one CSV
|
|
# per request — dedup is handled later via natural_key_hash.
|
|
transport = self._transport(profile_config, secrets)
|
|
content = transport.fetch(transport.remote_glob)
|
|
yield FetchedFile(
|
|
remote_path=f"grandstream_cdr_{datetime.utcnow():%Y%m%dT%H%M%SZ}.csv",
|
|
mtime=datetime.utcnow(),
|
|
content=content,
|
|
size_bytes=len(content),
|
|
)
|