#!/usr/bin/env python3
"""
Query Alibaba Cloud SLS logs with the official Python SDK.

Examples:
  python3 control/scripts/sls_query.py query \
    --console-url 'https://sls.console.aliyun.com/lognext/project/ydysskz-log-online-liunx/logsearch/ydysskz-log-online-liunx?slsRegion=cn-beijing' \
    --from '2026-05-14 08:20:00' \
    --to '2026-05-14 12:55:00' \
    --query '("260514104022771" OR "217326")'

  python3 control/scripts/sls_query.py print-triage \
    --console-url 'https://sls.console.aliyun.com/lognext/project/ydysskz-log-online-liunx/logsearch/ydysskz-log-online-liunx?slsRegion=cn-beijing' \
    --from '2026-05-14 08:20:00' \
    --to '2026-05-14 12:55:00' \
    --orders '260514129988475,260514085346987,260514102304111,260514104022771' \
    --order-ids '222128,216754,217325,217326'

Credentials are read from environment variables or config/local/aliyun-sls.env:
  ALIYUN_ACCESS_KEY_ID
  ALIYUN_ACCESS_KEY_SECRET
  ALIYUN_SECURITY_TOKEN optional
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import warnings
from datetime import datetime
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple
from urllib.parse import parse_qs, urlparse
from zoneinfo import ZoneInfo

warnings.filterwarnings("ignore", message="urllib3 v2 only supports OpenSSL.*")
try:
    from urllib3.exceptions import NotOpenSSLWarning

    warnings.filterwarnings("ignore", category=NotOpenSSLWarning)
except Exception:
    pass


DEFAULT_FIELDS = [
    "__time__",
    "time",
    "request_start_time",
    "trace_id",
    "business",
    "uri",
    "url",
    "request_method",
    "messages",
    "params",
    "file",
    "line",
]

SECRET_KEY_RE = re.compile(
    r'("(?:sign|pwd|password|token|access[_-]?key|secret|key)"\s*:\s*")([^"]+)(")',
    re.IGNORECASE,
)
PHONE_RE = re.compile(r"(?<!\d)(1[3-9]\d)(\d{4})(\d{4})(?!\d)")
LONG_TOKEN_RE = re.compile(r"(?<![A-Za-z0-9])([A-Za-z0-9_-]{32,})(?![A-Za-z0-9])")


def load_env_file(path: Path) -> None:
    if not path.exists():
        return
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        if key and key not in os.environ:
            os.environ[key] = value


def parse_console_url(console_url: str) -> Tuple[Optional[str], Optional[str], Optional[str]]:
    if not console_url:
        return None, None, None
    parsed = urlparse(console_url)
    parts = [p for p in parsed.path.split("/") if p]
    project = None
    logstore = None
    region = parse_qs(parsed.query).get("slsRegion", [None])[0]
    for index, part in enumerate(parts):
        if part == "project" and index + 1 < len(parts):
            project = parts[index + 1]
        if part == "logsearch" and index + 1 < len(parts):
            logstore = parts[index + 1]
    return project, logstore, region


def parse_time(value: str, tz_name: str) -> int:
    value = value.strip()
    if re.fullmatch(r"\d{10}", value):
        return int(value)
    tz = ZoneInfo(tz_name)
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            dt = datetime.strptime(value, fmt).replace(tzinfo=tz)
            return int(dt.timestamp())
        except ValueError:
            continue
    raise ValueError(f"Unsupported time format: {value}")


def require_sdk():
    try:
        from aliyun.log import GetLogsRequest, LogClient  # type: ignore
    except ModuleNotFoundError as exc:
        raise SystemExit(
            "Missing SDK: aliyun-log-python-sdk\n"
            "Install it with: python3 -m pip install --user aliyun-log-python-sdk"
        ) from exc
    return LogClient, GetLogsRequest


def build_endpoint(region: str, endpoint: str = "") -> str:
    if endpoint:
        return endpoint
    return f"{region}.log.aliyuncs.com"


def make_client(endpoint: str):
    access_key_id = os.environ.get("ALIYUN_ACCESS_KEY_ID") or os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID")
    access_key_secret = os.environ.get("ALIYUN_ACCESS_KEY_SECRET") or os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    security_token = os.environ.get("ALIYUN_SECURITY_TOKEN") or os.environ.get("ALIBABA_CLOUD_SECURITY_TOKEN")
    if not access_key_id or not access_key_secret:
        raise SystemExit(
            "Missing credentials. Set ALIYUN_ACCESS_KEY_ID and ALIYUN_ACCESS_KEY_SECRET "
            "or put them in config/local/aliyun-sls.env."
        )
    LogClient, _ = require_sdk()
    if security_token:
        return LogClient(endpoint, access_key_id, access_key_secret, security_token)
    return LogClient(endpoint, access_key_id, access_key_secret)


def sanitize_text(value: str) -> str:
    value = SECRET_KEY_RE.sub(r"\1***\3", value)
    value = PHONE_RE.sub(r"\1****\3", value)
    value = LONG_TOKEN_RE.sub(lambda m: m.group(1)[:6] + "***" + m.group(1)[-4:], value)
    return value


def sanitize_value(value):
    if isinstance(value, str):
        return sanitize_text(value)
    if isinstance(value, dict):
        return {k: sanitize_value("***" if k.lower() in {"sign", "pwd", "password", "token", "secret", "key"} else v) for k, v in value.items()}
    if isinstance(value, list):
        return [sanitize_value(v) for v in value]
    return value


def log_item_to_dict(log_item) -> Dict[str, str]:
    if hasattr(log_item, "get_contents"):
        contents = log_item.get_contents()
    else:
        contents = getattr(log_item, "contents", {})
    result = dict(contents or {})
    if hasattr(log_item, "get_time"):
        result.setdefault("__time__", str(log_item.get_time()))
    return result


def compact_log(row: Dict[str, str], fields: Iterable[str], max_field_len: int) -> Dict[str, str]:
    compact: Dict[str, str] = {}
    for field in fields:
        if field not in row:
            continue
        value = sanitize_value(row[field])
        if not isinstance(value, str):
            value = json.dumps(value, ensure_ascii=False)
        if max_field_len > 0 and len(value) > max_field_len:
            value = value[:max_field_len] + "...<truncated>"
        compact[field] = value
    if not compact:
        for key, value in row.items():
            if key.startswith("__"):
                continue
            text = sanitize_text(str(value))
            compact[key] = text[:max_field_len] + "...<truncated>" if max_field_len > 0 and len(text) > max_field_len else text
            if len(compact) >= 8:
                break
    return compact


def get_logs(client, project: str, logstore: str, from_ts: int, to_ts: int, query: str, line: int, offset: int, reverse: bool):
    _, GetLogsRequest = require_sdk()
    request = GetLogsRequest(project, logstore, from_ts, to_ts, "", query, line, offset, reverse)
    response = client.get_logs(request)
    logs = [log_item_to_dict(item) for item in response.get_logs()]
    completed = response.is_completed() if hasattr(response, "is_completed") else None
    return logs, completed


def print_logs(title: str, logs: List[Dict[str, str]], completed: Optional[bool], args) -> None:
    print(f"\n===== {title} =====")
    print(f"completed={completed} count={len(logs)}")
    fields = args.fields.split(",") if args.fields else DEFAULT_FIELDS
    for index, row in enumerate(logs, 1):
        compact = compact_log(row, fields, args.max_field_len)
        print(f"--- log #{index} ---")
        print(json.dumps(compact, ensure_ascii=False, sort_keys=False))


def resolve_target(args) -> Tuple[str, str, str]:
    url_project, url_logstore, url_region = parse_console_url(args.console_url or "")
    project = args.project or url_project or os.environ.get("SLS_PROJECT")
    logstore = args.logstore or url_logstore or os.environ.get("SLS_LOGSTORE")
    region = args.region or url_region or os.environ.get("SLS_REGION")
    if not project or not logstore or not region:
        raise SystemExit("Missing project/logstore/region. Pass --console-url or set explicit options.")
    return project, logstore, region


def split_csv(value: str) -> List[str]:
    return [item.strip() for item in (value or "").split(",") if item.strip()]


def quote_terms(terms: List[str]) -> str:
    escaped = [term.replace('"', '\\"') for term in terms if term]
    return "(" + " OR ".join(f'"{term}"' for term in escaped) + ")"


def run_query(args) -> None:
    load_env_file(Path(args.env_file))
    project, logstore, region = resolve_target(args)
    endpoint = build_endpoint(region, args.endpoint or os.environ.get("ALIYUN_LOG_ENDPOINT", ""))
    from_ts = parse_time(args.from_time, args.tz)
    to_ts = parse_time(args.to_time, args.tz)
    client = make_client(endpoint)
    print(f"target region={region} endpoint={endpoint} project={project} logstore={logstore} from={from_ts} to={to_ts}", file=sys.stderr)
    logs, completed = get_logs(client, project, logstore, from_ts, to_ts, args.query, args.line, args.offset, args.reverse)
    if args.jsonl:
        for row in logs:
            print(json.dumps(sanitize_value(row), ensure_ascii=False))
    else:
        print_logs(args.query, logs, completed, args)


def run_print_triage(args) -> None:
    load_env_file(Path(args.env_file))
    project, logstore, region = resolve_target(args)
    endpoint = build_endpoint(region, args.endpoint or os.environ.get("ALIYUN_LOG_ENDPOINT", ""))
    from_ts = parse_time(args.from_time, args.tz)
    to_ts = parse_time(args.to_time, args.tz)
    client = make_client(endpoint)
    identifiers = split_csv(args.orders) + split_csv(args.order_ids)
    if not identifiers:
        raise SystemExit("Pass --orders and/or --order-ids.")

    id_query = quote_terms(identifiers)
    queries = [
        ("all identifiers", id_query),
        ("active print trigger", f'{id_query} AND ("打印订单小票" OR "ai_api_start" OR "ai_api_end" OR "api/mealOrder/orderPrint")'),
        ("store orderPrint and Xpyun", f'{id_query} AND ("[MealOrder][orderPrint]" OR "Xpyun" OR "printReceipt" OR "job has been down and deleted")'),
        ("retry cron", f'{id_query} AND ("MealOrderPrintRetry" OR "meal_order_print_retry" OR "retry order_id")'),
    ]
    print(f"target region={region} endpoint={endpoint} project={project} logstore={logstore} from={from_ts} to={to_ts}", file=sys.stderr)
    for title, query in queries:
        logs, completed = get_logs(client, project, logstore, from_ts, to_ts, query, args.line, 0, args.reverse)
        print_logs(title + " :: " + query, logs, completed, args)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Query Alibaba Cloud SLS logs with Python SDK.")
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--console-url", default="", help="SLS console logsearch URL.")
    common.add_argument("--region", default="", help="SLS region, for example cn-beijing.")
    common.add_argument("--project", default="", help="SLS project.")
    common.add_argument("--logstore", default="", help="SLS logstore.")
    common.add_argument("--endpoint", default="", help="SLS endpoint. Defaults to <region>.log.aliyuncs.com.")
    common.add_argument("--env-file", default="config/local/aliyun-sls.env", help="Local env file for credentials.")
    common.add_argument("--from", dest="from_time", required=True, help="Start time, local time or Unix seconds.")
    common.add_argument("--to", dest="to_time", required=True, help="End time, local time or Unix seconds.")
    common.add_argument("--tz", default="Asia/Shanghai", help="Timezone for local time parsing.")
    common.add_argument("--line", type=int, default=100, help="Max log lines per query.")
    common.add_argument("--offset", type=int, default=0, help="Query offset.")
    common.add_argument("--reverse", action="store_true", help="Query newest first.")
    common.add_argument("--fields", default=",".join(DEFAULT_FIELDS), help="Comma separated output fields.")
    common.add_argument("--max-field-len", type=int, default=1200, help="Truncate long field values.")
    subparsers = parser.add_subparsers(dest="command", required=True)

    query = subparsers.add_parser("query", parents=[common], help="Run one SLS query.")
    query.add_argument("--query", required=True, help="SLS query string.")
    query.add_argument("--jsonl", action="store_true", help="Print sanitized raw logs as JSONL.")
    query.set_defaults(func=run_query)

    triage = subparsers.add_parser("print-triage", parents=[common], help="Run canned queries for meal order receipt printing.")
    triage.add_argument("--orders", default="", help="Comma separated order_no list.")
    triage.add_argument("--order-ids", default="", help="Comma separated order id list.")
    triage.set_defaults(func=run_print_triage)
    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()
    args.func(args)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())