Initial AM516 Feishu bot package

capabilities/am516-delivery-prediction/scripts/delivery_prediction_query_template.py

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import ssl
+import urllib.error
+import urllib.parse
+import urllib.request
+
+import truststore
+
+
+truststore.inject_into_ssl()
+
+
+DEFAULT_BASE_URL = os.getenv("ZEROERR_AGENT_BASE_URL", "https://api.zeroerr-agent.com").rstrip("/")
+DEFAULT_ENDPOINT = "/api/SaleAgent/DeliveryPrediction/WithTransit"
+APPROVED_SKIP_TLS_HOSTNAME = "api.zeroerr-agent.com"
+API_KEY_ENV = "ZEROERR_AGENT_API_KEY"
+PROJECT_ROOT = Path(__file__).resolve().parents[3]
+LOCAL_API_KEY_FILE = PROJECT_ROOT / "configs" / "delivery_api_key.local"
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Query delivery prediction by product specification and quantity.")
+    parser.add_argument("--product", required=True, help="Product specification, for example eRobxxx[...]")
+    parser.add_argument("--quantity", required=True, type=int, help="Requested quantity, positive integer")
+    parser.add_argument(
+        "--skip-tls-verify",
+        action="store_true",
+        help="Temporarily disable TLS certificate verification for the approved API host",
+    )
+    return parser.parse_args()
+
+
+def emit_error(
+    message: str,
+    *,
+    status_code: int | None = None,
+    body: str | None = None,
+    details: dict[str, str] | None = None,
+) -> None:
+    payload: dict[str, object] = {
+        "ok": False,
+        "error": message,
+    }
+    if status_code is not None:
+        payload["status_code"] = status_code
+    if body:
+        payload["body_preview"] = body[:2000]
+    if details:
+        payload["details"] = details
+    print(json.dumps(payload, ensure_ascii=False, indent=2))
+
+
+def get_api_key() -> str:
+    api_key = os.getenv(API_KEY_ENV, "").strip()
+    if api_key:
+        return api_key
+
+    try:
+        return LOCAL_API_KEY_FILE.read_text(encoding="utf-8").strip()
+    except FileNotFoundError:
+        return ""
+
+
+def main() -> int:
+    args = parse_args()
+    product = args.product.strip()
+    if not product or "\n" in product or "\r" in product:
+        emit_error("product must be a non-empty single-line string")
+        return 2
+
+    if args.quantity <= 0:
+        emit_error("quantity must be a positive integer")
+        return 2
+
+    api_key = get_api_key()
+    if not api_key:
+        emit_error(f"missing API key: set {API_KEY_ENV} or configure local key file")
+        return 2
+
+    query = urllib.parse.urlencode(
+        {
+            "product_specification": product,
+            "quantity": args.quantity,
+        }
+    )
+    url = f"{DEFAULT_BASE_URL}{DEFAULT_ENDPOINT}?{query}"
+    target_hostname = urllib.parse.urlsplit(url).hostname
+    if args.skip_tls_verify and target_hostname != APPROVED_SKIP_TLS_HOSTNAME:
+        emit_error(
+            "TLS verification bypass is not permitted for this target host",
+            details={
+                "required_hostname": APPROVED_SKIP_TLS_HOSTNAME,
+                "actual_hostname": target_hostname or "missing",
+            },
+        )
+        return 2
+
+    headers = {
+        "X-API-Key": api_key,
+        "User-Agent": (
+            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+            "AppleWebKit/537.36 (KHTML, like Gecko) "
+            "Chrome/120.0.0.0 Safari/537.36"
+        ),
+    }
+
+    request = urllib.request.Request(url, headers=headers, method="GET")
+    context = ssl._create_unverified_context() if args.skip_tls_verify else None
+    try:
+        with urllib.request.urlopen(request, timeout=30, context=context) as response:
+            raw_body = response.read().decode("utf-8")
+    except urllib.error.HTTPError as exc:
+        body = exc.read().decode("utf-8", errors="replace")
+        emit_error(f"HTTP {exc.code} {exc.reason}", status_code=exc.code, body=body)
+        return 1
+    except urllib.error.URLError as exc:
+        reason = str(exc.reason)
+        if isinstance(exc.reason, ssl.SSLCertVerificationError) or "CERTIFICATE_VERIFY_FAILED" in reason:
+            emit_error(
+                "TLS certificate verification failed",
+                details={
+                    "cause": "The local Python certificate trust configuration could not validate the server certificate chain.",
+                    "approved_retry": "After explicit AR51 approval, use --skip-tls-verify once on a trusted network.",
+                    "long_term_action": "Repair the Python CA trust, approved company proxy certificate, or server certificate chain.",
+                },
+            )
+            return 1
+        emit_error(f"request failed: {exc.reason}")
+        return 1
+
+    try:
+        result = json.loads(raw_body)
+    except json.JSONDecodeError as exc:
+        emit_error(f"response is not valid JSON: {exc}", body=raw_body)
+        return 1
+
+    data = result.get("data")
+    if isinstance(data, dict):
+        summary = data.get("交期汇总", {})
+        if isinstance(summary, dict) and "总交期" in summary:
+            print(f"总交期: {summary['总交期']} 天")
+
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())