diff --git a/deploy/sip_probe.py b/deploy/sip_probe.py new file mode 100644 index 0000000..51bc2ac --- /dev/null +++ b/deploy/sip_probe.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +"""Run one explicit, non-retrying SIP probe through the current ECS host.""" + +from __future__ import annotations + +import argparse +import base64 +import json +import os +import re +import sys +import time +from pathlib import Path +from typing import Any + +try: + from .aliyun_host import AliyunCLI, CloudError, plan +except ImportError: + from aliyun_host import AliyunCLI, CloudError, plan + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_CONFIG = ROOT / ".local/agent-call-persistent/aliyun-real.json" +DEFAULT_OUTPUT_DIR = ROOT / ".local/agent-call-persistent" +ANSI = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") +SIP_STATUS = re.compile(r"SIP/2\.0\s+([1-6]\d{2})(?:\s+([^\r\n]+))?") +TERMINAL = {"Finished", "Success", "Failed", "Stopped", "PartialFailed"} + + +def validate_number(number: str) -> str: + if not re.fullmatch(r"[0-9]{3,32}", number): + raise ValueError( + "number must contain only 3-32 digits; the original value is preserved" + ) + return number + + +def clean(text: str) -> str: + return ANSI.sub("", text) + + +def parse_int(value: str) -> int | None: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def remote_command(number: str, target: str, wait_seconds: int) -> str: + pattern = ( + f"{target}|INVITE sip:|ACK sip:|BYE sip:|" + r"SIP/2.0 (100|180|183|200|4[0-9][0-9]|5[0-9][0-9]|6[0-9][0-9])|" + r"Reason:|m=audio|a=rtpmap|Called|Executing.*outbound|" + r"Everyone is busy|Hangup|Spawn extension" + ) + return f"""set +e +docker exec agent-call-asterisk asterisk -rx 'pjsip set logger on' >/dev/null +start=$(date -u +%Y-%m-%dT%H:%M:%SZ) +echo "CALL_START $start number={number} target={target} codec=PCMA logger=on" +docker exec agent-call-asterisk asterisk -rx 'channel originate Local/{target}@outbound application Wait {wait_seconds}' +rc=$? +echo "ORIGINATE_RC $rc" +sleep {wait_seconds + 2} +echo "CALL_END $(date -u +%Y-%m-%dT%H:%M:%SZ)" +echo '== channels ==' +docker exec agent-call-asterisk asterisk -rx 'core show channels count' +echo '== filtered SIP/Asterisk evidence ==' +docker logs --since "$start" agent-call-asterisk 2>&1 | grep -E '{pattern}' || true +""" + + +def output_text(row: dict[str, Any]) -> str: + value = row.get("Output", "") + if not isinstance(value, str): + return "" + if "CALL_START" in value: + return value + try: + decoded = base64.b64decode(value, validate=True).decode("utf-8", "replace") + except (ValueError, UnicodeDecodeError): + return value + return decoded if "CALL_START" in decoded else value + + +def summarize(text: str, target: str) -> dict[str, Any]: + text = clean(text) + marker = text.find(f"INVITE sip:{target}") + relevant = text[marker:] if marker >= 0 else text + responses = [] + for match in SIP_STATUS.finditer(relevant): + code = parse_int(match.group(1)) + if code is not None: + responses.append({"code": code, "phrase": (match.group(2) or "").strip()}) + final = next((item for item in responses if item["code"] >= 200), None) + reason_match = re.search(r"Reason:\s*([^\r\n]+)", relevant, re.IGNORECASE) + origin_match = re.search(r"ORIGINATE_RC\s+(-?\d+)", text) + channels_match = re.search(r"(\d+) active channels", text) + if final is None: + result = "provisional" if responses else "no_response" + else: + result = "connected" if final["code"] < 300 else "rejected" + return { + "result": result, + "origin_rc": parse_int(origin_match.group(1)) if origin_match else None, + "responses": responses, + "final_response": final, + "reason": reason_match.group(1).strip() if reason_match else None, + "active_channels": parse_int(channels_match.group(1)) + if channels_match + else None, + } + + +def first_invocation_result(data: dict[str, Any]) -> dict[str, Any] | None: + rows = ( + data.get("Invocation", {}) + .get("InvocationResults", {}) + .get("InvocationResult", []) + ) + if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict): + return None + return rows[0] + + +def save_evidence(path: Path, evidence: dict[str, Any]) -> None: + if path.is_symlink(): + raise ValueError("output path must not be a symlink") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(evidence, ensure_ascii=False, indent=2), encoding="utf-8" + ) + os.chmod(path, 0o600) + + +def run(args: argparse.Namespace) -> tuple[int, Path, dict[str, Any]]: + number = validate_number(args.number) + config_path = Path(args.config).expanduser() + try: + cfg = json.loads(config_path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ValueError(f"config file is unavailable: {config_path}") from exc + api = AliyunCLI(cfg.get("profile")) + current = plan(cfg, api) + if current.get("action") != "reuse" or current.get("instance_status") != "Running": + raise CloudError( + "fixed-EIP host is not a running reusable instance; no call sent" + ) + instance_id = current.get("instance_id") + if not isinstance(instance_id, str) or not instance_id: + raise CloudError("reusable instance ID is unavailable; no call sent") + target = "7089" + number + command_name = ( + f"agent-call-sip-probe-{number}-{time.strftime('%Y%m%d%H%M%S', time.gmtime())}" + ) + run_result = api( + "ecs", + "RunCommand", + **{ + "InstanceId.1": instance_id, + "Type": "RunShellScript", + "Name": command_name, + "CommandContent": remote_command(number, target, args.wait_seconds), + "ContentEncoding": "PlainText", + "Timeout": max(30, args.wait_seconds + 12), + "RepeatMode": "Once", + "KeepCommand": "false", + }, + ) + command_id = run_result.get("CommandId") + invoke_id = run_result.get("InvokeId") + if not isinstance(command_id, str) or not isinstance(invoke_id, str): + raise CloudError( + "RunCommand returned no command/invocation ID; inspect cloud state before retry" + ) + + result: dict[str, Any] | None = None + row: dict[str, Any] | None = None + deadline = time.monotonic() + args.timeout_seconds + while time.monotonic() < deadline: + result = api( + "ecs", + "DescribeInvocationResults", + InvokeId=invoke_id, + ContentEncoding="PlainText", + MaxResults=10, + ) + candidate = first_invocation_result(result) + if candidate is not None: + row = candidate + status = candidate.get("InvokeRecordStatus") or candidate.get( + "InvocationStatus" + ) + if status in TERMINAL: + break + time.sleep(args.poll_seconds) + if row is None: + raise CloudError( + f"invocation {invoke_id} did not finish before timeout; do not redial automatically" + ) + status = row.get("InvokeRecordStatus") or row.get("InvocationStatus") + if status not in TERMINAL: + raise CloudError( + f"invocation {invoke_id} did not finish before timeout; do not redial automatically" + ) + + raw_output = output_text(row) + probe = summarize(raw_output, target) + stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) + output_path = ( + Path(args.output) + if args.output + else DEFAULT_OUTPUT_DIR / f"sip-probe-{number}-{stamp}.json" + ) + evidence = { + "instance_id": instance_id, + "public_ip": current.get("public_ip"), + "number": number, + "target": target, + "dial_prefix": "7089", + "codec": "PCMA", + "logger": "on", + "command_id": command_id, + "invoke_id": invoke_id, + "invocation_status": row.get("InvocationStatus"), + "invoke_record_status": row.get("InvokeRecordStatus"), + "exit_code": row.get("ExitCode"), + "probe": probe, + "raw_output": clean(raw_output), + } + save_evidence(output_path, evidence) + summary = {key: value for key, value in evidence.items() if key != "raw_output"} + summary["evidence_file"] = str(output_path) + return ( + 0 if row.get("ExitCode") == 0 and probe["result"] != "no_response" else 1, + output_path, + summary, + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--number", required=True, help="original callee number; one call, no retry" + ) + parser.add_argument("--config", default=str(DEFAULT_CONFIG)) + parser.add_argument("--output", help="0600 evidence JSON path") + parser.add_argument("--wait-seconds", type=int, default=5) + parser.add_argument("--poll-seconds", type=float, default=1.0) + parser.add_argument("--timeout-seconds", type=int, default=60) + args = parser.parse_args() + if not 3 <= args.wait_seconds <= 30: + parser.error("--wait-seconds must be 3-30") + if args.poll_seconds <= 0 or args.timeout_seconds <= 0: + parser.error("poll and timeout must be positive") + try: + status, _, summary = run(args) + except (CloudError, OSError, ValueError, json.JSONDecodeError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + print(json.dumps(summary, ensure_ascii=False, indent=2)) + return status + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/真实SIP_Asterisk快速部署_v1.0.md b/docs/真实SIP_Asterisk快速部署_v1.0.md index d3007fa..58b16b5 100644 --- a/docs/真实SIP_Asterisk快速部署_v1.0.md +++ b/docs/真实SIP_Asterisk快速部署_v1.0.md @@ -115,14 +115,13 @@ aliyun ecs RunCommand --RegionId cn-beijing \ --KeepCommand false ``` -确认供应商已经冻结目标号码、前缀、From 域、PAI 和编解码后，一次只测试一个号码；不自动重试： +确认供应商已经冻结目标号码、前缀、From 域、PAI 和编解码后，一次只测试一个号码；不自动重试。使用一键脚本会先只读核对固定 EIP/运行实例，默认等待 5 秒并保存 0600 证据： ```bash -docker exec agent-call-asterisk asterisk -rx \ - 'channel originate Local/<7089+原始被叫>@outbound application Wait 30' +python3 deploy/sip_probe.py --number 18601013734 ``` -记录完整 SIP 状态、`Reason`、RTP/录音结果和 `call_id`。`Avail`/OPTIONS 只证明探活，不证明外呼接通。 +脚本会自动构造 `7089` + 原始号码、开启 SIP logger、发起一次外呼并提取首个最终 SIP 响应；`--wait-seconds 3` 可缩短等待，`--output` 可指定证据文件。`Avail`/OPTIONS 只证明探活，不证明外呼接通。 ## 5. 更新与回收 diff --git a/tests/test_sip_probe.py b/tests/test_sip_probe.py new file mode 100644 index 0000000..c0370f2 --- /dev/null +++ b/tests/test_sip_probe.py @@ -0,0 +1,44 @@ +import base64 +import unittest + +from deploy import sip_probe + + +class SipProbeTests(unittest.TestCase): + def test_summarize_uses_first_final_response(self): + output = """CALL_START +INVITE sip:708913851814181@provider SIP/2.0 +SIP/2.0 100 trying +SIP/2.0 404 Not Found +Reason: Q.850;cause=16;text=\"NORMAL_CLEARING\" +SIP/2.0 200 OK +0 active channels +""" + result = sip_probe.summarize(output, "708913851814181") + self.assertEqual(result["result"], "rejected") + self.assertEqual(result["final_response"]["code"], 404) + self.assertEqual(result["reason"], 'Q.850;cause=16;text="NORMAL_CLEARING"') + self.assertEqual(result["active_channels"], 0) + + def test_remote_command_is_short_and_keeps_logger_on(self): + command = sip_probe.remote_command("13851814181", "708913851814181", 5) + self.assertIn("pjsip set logger on", command) + self.assertIn("Wait 5", command) + self.assertIn("sleep 7", command) + self.assertNotIn("pjsip set logger off", command) + + def test_output_text_accepts_plain_and_base64(self): + plain = "CALL_START\nSIP/2.0 404 Not Found\n" + encoded = base64.b64encode(plain.encode()).decode() + self.assertEqual(sip_probe.output_text({"Output": plain}), plain) + self.assertEqual(sip_probe.output_text({"Output": encoded}), plain) + + def test_number_validation_preserves_original_digits(self): + number = "001380051234567" + self.assertEqual(sip_probe.validate_number(number), number) + with self.assertRaises(ValueError): + sip_probe.validate_number("+861380051234567") + + +if __name__ == "__main__": + unittest.main()