| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361 |
- import datetime as dt
- import functools
- from .config import parse_timezone, to_float_or_none
- from .filter import IPV4_RE, normalize_domain, resolve_field
- def parse_created_time(value, formats, timezone):
- if value is None:
- return None
- s = str(value).strip()
- if not s:
- return None
- for fmt in formats:
- try:
- parsed = dt.datetime.strptime(s, fmt)
- if parsed.tzinfo is None:
- parsed = parsed.replace(tzinfo=timezone)
- return parsed.astimezone(dt.timezone.utc)
- except Exception:
- continue
- try:
- iso_text = s.replace("Z", "+00:00")
- parsed = dt.datetime.fromisoformat(iso_text)
- if parsed.tzinfo is None:
- parsed = parsed.replace(tzinfo=timezone)
- return parsed.astimezone(dt.timezone.utc)
- except Exception:
- return None
- def build_lexicographic_descriptors(scoring_cfg, prefer_lower):
- out = []
- for item in scoring_cfg.get("lexicographic_fields", []):
- if isinstance(item, str):
- field_name = item.strip()
- order = "asc" if prefer_lower else "desc"
- else:
- field_name = str(item.get("field", "")).strip()
- order = str(item.get("order", "")).strip().lower()
- if not order:
- order = "asc" if prefer_lower else "desc"
- out.append({"field": field_name, "order": order})
- return out
- def parse_scored_records(records, field_map, record_mapping_cfg, scoring_cfg):
- if not scoring_cfg.get("enabled", False):
- return []
- strategy = str(scoring_cfg.get("strategy", "weighted_average")).strip()
- prefer_lower = bool(scoring_cfg.get("prefer_lower", False))
- timezone = parse_timezone(record_mapping_cfg.get("created_time_timezone", "UTC"))
- time_formats = [str(x) for x in record_mapping_cfg.get("created_time_formats", [])]
- weighted_fields = scoring_cfg.get("weighted_fields", []) if strategy == "weighted_average" else []
- lex_descriptors = build_lexicographic_descriptors(scoring_cfg, prefer_lower) if strategy == "lexicographic" else []
- needed_fields = set()
- for item in weighted_fields:
- needed_fields.add(str(item.get("field", "")).strip())
- for item in lex_descriptors:
- needed_fields.add(str(item.get("field", "")).strip())
- for item in scoring_cfg.get("tie_breakers", []):
- needed_fields.add(str(item.get("field", "")).strip())
- needed_fields.discard("domain")
- needed_fields.discard("created_at")
- out = []
- for record in records:
- domain = normalize_domain(resolve_field(record, "domain", field_map))
- if not domain:
- continue
- created_raw = resolve_field(record, "created_at", field_map)
- created_at = parse_created_time(created_raw, time_formats, timezone)
- field_values = {}
- for field_name in needed_fields:
- field_values[field_name] = resolve_field(record, field_name, field_map)
- score_value = None
- scores = []
- lex_values = []
- if strategy == "weighted_average":
- total = 0.0
- total_weight = 0.0
- missing = False
- for item in weighted_fields:
- field_name = str(item.get("field", "")).strip()
- weight = float(item.get("weight"))
- raw_v = resolve_field(record, field_name, field_map)
- val = to_float_or_none(raw_v)
- scores.append(val)
- if val is None:
- missing = True
- continue
- field_lower = item.get("prefer_lower", prefer_lower)
- if str(item.get("order", "")).lower() == "desc":
- field_lower = False
- elif str(item.get("order", "")).lower() == "asc":
- field_lower = True
- effective_val = val if (bool(field_lower) == prefer_lower) else -val
- total += effective_val * weight
- total_weight += weight
- if not missing and total_weight > 0:
- score_value = total / total_weight
- if strategy == "lexicographic":
- for item in lex_descriptors:
- field_name = item["field"]
- order = item["order"]
- raw_v = resolve_field(record, field_name, field_map)
- num_v = to_float_or_none(raw_v)
- v = num_v if num_v is not None else raw_v
- lex_values.append({"field": field_name, "value": v, "order": order})
- scores.append(v)
- out.append(
- {
- "domain": domain,
- "created_at": created_at,
- "created_raw": created_raw,
- "scores": scores,
- "score_value": score_value,
- "lex_values": lex_values,
- "field_values": field_values,
- "raw_record": record,
- }
- )
- return out
- def cmp_scalar(a, b, order):
- a_none = a is None
- b_none = b is None
- if a_none and b_none:
- return 0
- if a_none:
- return 1
- if b_none:
- return -1
- if isinstance(a, dt.datetime):
- a = a.timestamp()
- if isinstance(b, dt.datetime):
- b = b.timestamp()
- a_num = to_float_or_none(a)
- b_num = to_float_or_none(b)
- if a_num is not None and b_num is not None:
- if a_num < b_num:
- base = -1
- elif a_num > b_num:
- base = 1
- else:
- base = 0
- else:
- a_s = str(a).lower()
- b_s = str(b).lower()
- if a_s < b_s:
- base = -1
- elif a_s > b_s:
- base = 1
- else:
- base = 0
- return base if order == "asc" else -base
- def get_sort_field_value(record, field_name):
- if field_name == "domain":
- return record.get("domain")
- if field_name == "created_at":
- return record.get("created_at")
- return record.get("field_values", {}).get(field_name)
- def rank_scored_records(records, scoring_cfg):
- if not records:
- return []
- within_hours = float(scoring_cfg.get("within_hours", 0))
- strategy = str(scoring_cfg.get("strategy", "weighted_average")).strip()
- prefer_lower = bool(scoring_cfg.get("prefer_lower", False))
- tie_breakers = scoring_cfg.get("tie_breakers", [])
- if within_hours > 0:
- now = dt.datetime.now(dt.timezone.utc)
- cutoff = now - dt.timedelta(hours=within_hours)
- recent = [r for r in records if r.get("created_at") is not None and r["created_at"] >= cutoff]
- candidates = recent if recent else records
- else:
- candidates = records
- default_lex_order = "asc" if prefer_lower else "desc"
- def compare(a, b):
- if strategy == "weighted_average":
- order = "asc" if prefer_lower else "desc"
- c = cmp_scalar(a.get("score_value"), b.get("score_value"), order)
- if c != 0:
- return c
- elif strategy == "lexicographic":
- a_lex = a.get("lex_values", [])
- b_lex = b.get("lex_values", [])
- n = max(len(a_lex), len(b_lex))
- for i in range(n):
- av = a_lex[i]["value"] if i < len(a_lex) else None
- bv = b_lex[i]["value"] if i < len(b_lex) else None
- order = default_lex_order
- if i < len(a_lex) and a_lex[i].get("order"):
- order = a_lex[i]["order"]
- c = cmp_scalar(av, bv, order)
- if c != 0:
- return c
- for item in tie_breakers:
- field_name = str(item.get("field", "")).strip()
- order = str(item.get("order", "asc")).strip().lower()
- av = get_sort_field_value(a, field_name)
- bv = get_sort_field_value(b, field_name)
- c = cmp_scalar(av, bv, order)
- if c != 0:
- return c
- return cmp_scalar(a.get("domain"), b.get("domain"), "asc")
- return sorted(candidates, key=functools.cmp_to_key(compare))
- def choose_top_candidate_domains(filtered_domains, top_n, ranked_scored):
- if ranked_scored:
- domains_by_score = [x["domain"] for x in ranked_scored]
- return domains_by_score[:top_n]
- return filtered_domains[:top_n]
- def choose_domain(filtered_domains, top_n, ranked_scored, last_good="", switch_margin_pct=0.0):
- if ranked_scored:
- top_domains = choose_top_candidate_domains(filtered_domains, top_n, ranked_scored)
- if top_domains:
- best_domain = top_domains[0]
- if switch_margin_pct > 0 and last_good and best_domain != last_good:
- last_cand = next((r for r in ranked_scored if r.get("domain") == last_good), None)
- best_cand = next((r for r in ranked_scored if r.get("domain") == best_domain), None)
- if last_cand and best_cand:
- best_score = to_float_or_none(best_cand.get("score_value"))
- last_score = to_float_or_none(last_cand.get("score_value"))
- if best_score is not None and last_score is not None and last_score != 0:
- improvement_pct = abs((last_score - best_score) / last_score) * 100.0
- if improvement_pct < switch_margin_pct:
- top_reordered = [last_good] + [d for d in top_domains if d != last_good]
- return last_good, top_reordered[:top_n]
- return best_domain, top_domains
- if filtered_domains:
- return filtered_domains[0], filtered_domains[:top_n]
- return None, []
- def text_or_blank(value):
- if value is None:
- return ""
- return str(value).strip()
- def set_if_nonempty_text(obj, key, value):
- text = text_or_blank(value)
- if text:
- obj[key] = text
- def base_top_candidate(domain, source_type):
- candidate = {"domain": domain, "source_type": source_type}
- if IPV4_RE.match(domain):
- candidate["ip"] = domain
- return candidate
- def maybe_resolve_field(record, field_name, field_map):
- if not isinstance(record, dict):
- return None
- if field_name not in field_map:
- return None
- return resolve_field(record, field_name, field_map)
- def build_cfst_candidate(row):
- domain = row.get("domain", "")
- candidate = base_top_candidate(domain=domain, source_type="cfst_local")
- set_if_nonempty_text(candidate, "ip", row.get("ip") or domain)
- set_if_nonempty_text(candidate, "loss_rate", row.get("loss_rate"))
- set_if_nonempty_text(candidate, "avg_latency", row.get("avg_latency"))
- set_if_nonempty_text(candidate, "download_speed", row.get("download_speed"))
- set_if_nonempty_text(candidate, "region", row.get("region"))
- return candidate
- def build_api_candidate(domain, record, field_map, scored_record=None):
- candidate = base_top_candidate(domain=domain, source_type="api")
- if record:
- set_if_nonempty_text(candidate, "created_raw", maybe_resolve_field(record, "created_at", field_map))
- set_if_nonempty_text(candidate, "avg_latency", maybe_resolve_field(record, "avg_latency", field_map))
- set_if_nonempty_text(candidate, "loss_rate", maybe_resolve_field(record, "avg_pkg_lost_rate", field_map))
- set_if_nonempty_text(candidate, "download_speed", maybe_resolve_field(record, "download_speed", field_map))
- location_country = text_or_blank(maybe_resolve_field(record, "location_country", field_map))
- location_city = text_or_blank(maybe_resolve_field(record, "location_city", field_map))
- region = "/".join([x for x in [location_country, location_city] if x])
- set_if_nonempty_text(candidate, "region", region)
- if scored_record:
- if scored_record.get("score_value") is not None:
- candidate["score_value"] = scored_record.get("score_value")
- scores = list(scored_record.get("scores", []))
- if scores:
- candidate["scores"] = scores
- if "created_raw" not in candidate:
- set_if_nonempty_text(candidate, "created_raw", scored_record.get("created_raw"))
- return candidate
- def build_top_candidates(source_type, candidate_domains, cfst_rows=None, records=None, field_map=None, ranked_scored=None):
- if source_type == "cfst_local":
- row_map = {}
- for row in cfst_rows or []:
- domain = row.get("domain", "")
- if domain and domain not in row_map:
- row_map[domain] = row
- return [build_cfst_candidate(row_map.get(d, {"domain": d, "ip": d})) for d in candidate_domains]
- ranked_map = {}
- for item in ranked_scored or []:
- domain = item.get("domain", "")
- if domain and domain not in ranked_map:
- ranked_map[domain] = item
- record_map = {}
- for record in records or []:
- domain = normalize_domain(maybe_resolve_field(record, "domain", field_map or {}))
- if domain and domain not in record_map:
- record_map[domain] = record
- out = []
- for domain in candidate_domains:
- scored_record = ranked_map.get(domain)
- record = None
- if scored_record:
- record = scored_record.get("raw_record")
- if record is None:
- record = record_map.get(domain)
- out.append(build_api_candidate(domain, record, field_map or {}, scored_record=scored_record))
- return out
|