vmess-domain-rotator/scripts/rotator/config.py

import datetime as dt
import json
import math
import os
import re
import urllib.parse


def build_url(base_url, params):
    if not params:
        return base_url
    parsed = urllib.parse.urlparse(base_url)
    current = urllib.parse.parse_qs(parsed.query)
    for k, v in params.items():
        current[k] = [str(v)]
    query = urllib.parse.urlencode(current, doseq=True)
    return urllib.parse.urlunparse(parsed._replace(query=query))


def resolve_path(base_dir, path_value):
    path_text = str(path_value or "").strip()
    if not path_text:
        return ""
    if os.path.isabs(path_text):
        return os.path.normpath(path_text)
    return os.path.normpath(os.path.join(base_dir, path_text))


def get_source_type(cfg):
    source_cfg = cfg.get("source", {})
    if isinstance(source_cfg, dict):
        source_type = str(source_cfg.get("type", "api")).strip().lower()
        if source_type:
            return source_type
    return "api"


def parse_timezone(tz_raw):
    if tz_raw is None:
        return dt.timezone.utc
    s = str(tz_raw).strip().upper()
    if s in {"", "UTC", "Z", "+00:00", "+0000"}:
        return dt.timezone.utc

    m = re.match(r"^([+-])(\d{2}):?(\d{2})$", s)
    if not m:
        raise ValueError(f"invalid created_time_timezone: {tz_raw}")

    sign = 1 if m.group(1) == "+" else -1
    hh = int(m.group(2))
    mm = int(m.group(3))
    if hh > 23 or mm > 59:
        raise ValueError(f"invalid created_time_timezone offset: {tz_raw}")
    return dt.timezone(sign * dt.timedelta(hours=hh, minutes=mm))


def to_float_or_none(value):
    try:
        f = float(value)
        if not math.isfinite(f):
            return None
        return f
    except Exception:
        return None


def build_output_settings(output_cfg, config_path_abs):
    # legacy keys (current_domain_file/current_domain_json/substore_vars_file)
    # were ambiguous and drifted between server/local configs. Refuse to start
    # so operators migrate to the unified key set documented in CLAUDE.md.
    legacy_key_map = {
        "current_domain_file": "selected_value_file",
        "current_domain_json": "selected_value_json",
        "substore_vars_file": "export_vars_file",
    }
    legacy_in_use = [k for k in legacy_key_map if k in output_cfg]
    if legacy_in_use:
        renames = ", ".join(
            "{} -> {}".format(k, legacy_key_map[k]) for k in legacy_in_use
        )
        raise ValueError(
            "deprecated output keys in config: {}. Rename them: {}.".format(
                ", ".join(legacy_in_use), renames
            )
        )

    runtime_dir_cfg = output_cfg.get("runtime_dir", "./runtime")
    runtime_dir = resolve_path(os.path.dirname(config_path_abs), runtime_dir_cfg)

    selected_text_name = output_cfg.get("selected_value_file", "current_domain.txt")
    selected_json_name = output_cfg.get("selected_value_json", "current_domain.json")
    state_name = output_cfg.get("state_file", "state.json")
    vars_name = output_cfg.get("export_vars_file", "substore_vars.json")

    return {
        "runtime_dir": runtime_dir,
        "selected_text_path": os.path.join(runtime_dir, selected_text_name),
        "selected_json_path": os.path.join(runtime_dir, selected_json_name),
        "state_path": os.path.join(runtime_dir, state_name),
        "vars_path": os.path.join(runtime_dir, vars_name),
        "selected_json_key": str(output_cfg.get("selected_value_json_key", "domain")).strip() or "domain",
        "state_last_good_key": str(output_cfg.get("state_last_good_key", "last_good_domain")).strip() or "last_good_domain",
        "vars_value_key": str(output_cfg.get("substore_value_key", "AUTO_DOMAIN")).strip() or "AUTO_DOMAIN",
    }


def print_output_settings(config_path_abs, cfg):
    output_cfg = cfg.get("output", {})
    settings = build_output_settings(output_cfg, config_path_abs)
    print(json.dumps(settings, ensure_ascii=True))


def validate_config(cfg):
    source_type = get_source_type(cfg)
    if source_type not in {"api", "cfst_local"}:
        raise ValueError("source.type must be 'api' or 'cfst_local'")

    output_cfg = cfg.get("output", {})
    if output_cfg and not isinstance(output_cfg, dict):
        raise ValueError("output must be an object")

    if source_type == "cfst_local":
        cfst_cfg = cfg.get("cfst_local")
        if not isinstance(cfst_cfg, dict):
            raise ValueError("cfst_local is required and must be an object when source.type=cfst_local")

        work_dir = str(cfst_cfg.get("work_dir", "")).strip()
        if not work_dir:
            raise ValueError("cfst_local.work_dir is required")

        binary = str(cfst_cfg.get("binary", "")).strip()
        if not binary:
            raise ValueError("cfst_local.binary is required")

        result_file = str(cfst_cfg.get("result_file", "")).strip()
        if not result_file:
            raise ValueError("cfst_local.result_file is required")

        run_args = cfst_cfg.get("run_args", [])
        if not isinstance(run_args, list):
            raise ValueError("cfst_local.run_args must be an array")

        columns_cfg = cfst_cfg.get("columns", {})
        if columns_cfg and not isinstance(columns_cfg, dict):
            raise ValueError("cfst_local.columns must be an object")
        return

    # source.type == "api" 校验
    api_cfg = cfg.get("api")
    if not isinstance(api_cfg, dict):
        raise ValueError("api is required and must be an object when source.type=api")
    api_url = str(api_cfg.get("url", "")).strip()
    if not api_url:
        raise ValueError("api.url is required when source.type=api")

    parser_cfg = cfg.get("parser")
    if parser_cfg is not None and not isinstance(parser_cfg, dict):
        raise ValueError("parser must be an object")

    selection_cfg = cfg.get("selection")
    if selection_cfg is not None and not isinstance(selection_cfg, dict):
        raise ValueError("selection must be an object")

    domain_filter_cfg = cfg.get("domain_filter")
    if domain_filter_cfg is not None and not isinstance(domain_filter_cfg, dict):
        raise ValueError("domain_filter must be an object")

    record_mapping = cfg.get("record_mapping")
    if not isinstance(record_mapping, dict):
        raise ValueError("record_mapping is required and must be an object")

    records_path = str(record_mapping.get("records_path", "")).strip()
    if not records_path:
        raise ValueError("record_mapping.records_path is required")

    field_map = record_mapping.get("field_map")
    if not isinstance(field_map, dict) or not field_map:
        raise ValueError("record_mapping.field_map is required and must be a non-empty object")

    for key, path in field_map.items():
        if not str(key).strip() or not str(path).strip():
            raise ValueError("record_mapping.field_map contains empty field name or path")

    for required in ["domain", "created_at"]:
        if required not in field_map:
            raise ValueError(f"record_mapping.field_map.{required} is required")

    created_time_formats = record_mapping.get("created_time_formats")
    if not isinstance(created_time_formats, list) or not created_time_formats:
        raise ValueError("record_mapping.created_time_formats is required and must be a non-empty array")

    for fmt in created_time_formats:
        if not str(fmt).strip():
            raise ValueError("record_mapping.created_time_formats contains empty format")

    parse_timezone(record_mapping.get("created_time_timezone", "UTC"))

    def ensure_field_registered(field_name, where):
        if field_name not in field_map:
            raise ValueError(f"{where}: field '{field_name}' is not in record_mapping.field_map")

    record_filter = cfg.get("record_filter", {})
    if record_filter.get("enabled", False):
        rules = record_filter.get("exclude_if_any", [])
        if not isinstance(rules, list):
            raise ValueError("record_filter.exclude_if_any must be an array")

        for i, rule in enumerate(rules):
            if not isinstance(rule, dict):
                raise ValueError(f"record_filter.exclude_if_any[{i}] must be an object")
            field_name = str(rule.get("field", "")).strip()
            if not field_name:
                raise ValueError(f"record_filter.exclude_if_any[{i}].field is required")
            ensure_field_registered(field_name, f"record_filter.exclude_if_any[{i}]")

            has_matcher = any(k in rule for k in ["contains", "equals", "regex"])
            if not has_matcher:
                raise ValueError(f"record_filter.exclude_if_any[{i}] must include one of contains/equals/regex")

    scoring = cfg.get("scoring", {})
    if scoring.get("enabled", False):
        strategy = str(scoring.get("strategy", "")).strip()
        if strategy not in {"weighted_average", "lexicographic"}:
            raise ValueError("scoring.strategy must be 'weighted_average' or 'lexicographic'")

        within_hours = to_float_or_none(scoring.get("within_hours", 0))
        if within_hours is None or within_hours < 0:
            raise ValueError("scoring.within_hours must be a non-negative number (0 = disabled)")

        if strategy == "weighted_average":
            weighted_fields = scoring.get("weighted_fields")
            if not isinstance(weighted_fields, list) or not weighted_fields:
                raise ValueError("scoring.weighted_fields is required for weighted_average strategy")

            for i, item in enumerate(weighted_fields):
                if not isinstance(item, dict):
                    raise ValueError(f"scoring.weighted_fields[{i}] must be an object")
                field_name = str(item.get("field", "")).strip()
                if not field_name:
                    raise ValueError(f"scoring.weighted_fields[{i}].field is required")
                ensure_field_registered(field_name, f"scoring.weighted_fields[{i}]")

                weight = to_float_or_none(item.get("weight"))
                if weight is None or weight <= 0:
                    raise ValueError(f"scoring.weighted_fields[{i}].weight must be > 0")

        if strategy == "lexicographic":
            lex_fields = scoring.get("lexicographic_fields")
            if not isinstance(lex_fields, list) or not lex_fields:
                raise ValueError("scoring.lexicographic_fields is required for lexicographic strategy")

            for i, item in enumerate(lex_fields):
                if isinstance(item, str):
                    field_name = item.strip()
                    order = ""
                elif isinstance(item, dict):
                    field_name = str(item.get("field", "")).strip()
                    order = str(item.get("order", "")).strip().lower()
                else:
                    raise ValueError(f"scoring.lexicographic_fields[{i}] must be string or object")

                if not field_name:
                    raise ValueError(f"scoring.lexicographic_fields[{i}] field is required")
                ensure_field_registered(field_name, f"scoring.lexicographic_fields[{i}]")
                if order and order not in {"asc", "desc"}:
                    raise ValueError(f"scoring.lexicographic_fields[{i}].order must be asc or desc")

        tie_breakers = scoring.get("tie_breakers", [])
        if tie_breakers is not None:
            if not isinstance(tie_breakers, list):
                raise ValueError("scoring.tie_breakers must be an array")
            for i, item in enumerate(tie_breakers):
                if not isinstance(item, dict):
                    raise ValueError(f"scoring.tie_breakers[{i}] must be an object")
                field_name = str(item.get("field", "")).strip()
                order = str(item.get("order", "")).strip().lower()
                if not field_name:
                    raise ValueError(f"scoring.tie_breakers[{i}].field is required")
                if order not in {"asc", "desc"}:
                    raise ValueError(f"scoring.tie_breakers[{i}].order must be asc or desc")
                ensure_field_registered(field_name, f"scoring.tie_breakers[{i}]")