307 lines
12 KiB
Python
307 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Fuzzy match unmatched RADIUS users against Gaiia accounts.
|
|
Smarter matching: requires first-name correlation when matching on last name,
|
|
handles substring traps ("kim" in "kimberly"), etc.
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from difflib import SequenceMatcher
|
|
|
|
GAIIA_KEY = "gaiia_sk_live_955b27e3680747a69ad40b403a912155_b0d1be3c"
|
|
GAIIA_ENDPOINT = "https://api.gaiia.com/api/v1"
|
|
|
|
|
|
def fetch_all_accounts():
|
|
accounts = []
|
|
cursor = None
|
|
while True:
|
|
after_arg = f', after: "{cursor}"' if cursor else ""
|
|
query = f"""
|
|
query {{
|
|
accounts(first: 250{after_arg}) {{
|
|
nodes {{ id readableId name displayName }}
|
|
pageInfo {{ hasNextPage endCursor }}
|
|
}}
|
|
}}
|
|
"""
|
|
result = subprocess.run(
|
|
["curl", "-s", "-X", "POST", GAIIA_ENDPOINT,
|
|
"-H", "Content-Type: application/json",
|
|
"-H", f"X-Gaiia-Api-Key: {GAIIA_KEY}",
|
|
"-H", "x-timezone: America/Chicago",
|
|
"-d", json.dumps({"query": query})],
|
|
capture_output=True, text=True, timeout=30,
|
|
)
|
|
data = json.loads(result.stdout)
|
|
for n in data["data"]["accounts"]["nodes"]:
|
|
name = n.get("displayName") or n.get("name") or ""
|
|
if name:
|
|
accounts.append((n["id"], name.strip(), n["readableId"]))
|
|
page_info = data["data"]["accounts"]["pageInfo"]
|
|
if not page_info["hasNextPage"]:
|
|
break
|
|
cursor = page_info["endCursor"]
|
|
return accounts
|
|
|
|
|
|
def normalize(s):
|
|
return re.sub(r'[^\w\s]', '', s.lower()).strip()
|
|
|
|
|
|
def split_name(display_name):
|
|
"""Split into first name(s) and last name."""
|
|
parts = normalize(display_name).split()
|
|
if len(parts) == 1:
|
|
return parts[0], ""
|
|
# Filter out suffixes, LLC, numbers, etc.
|
|
content_parts = [p for p in parts if not re.match(r'^(#?\d+|llc|inc|ltd|shop|jr|sr)$', p)]
|
|
if len(content_parts) >= 2:
|
|
return " ".join(content_parts[:-1]), content_parts[-1]
|
|
return content_parts[0], ""
|
|
|
|
|
|
def first_name_match_score(username, last_name, first_name_str):
|
|
"""How well does the first name correlate with the username? Returns 0-1.
|
|
|
|
The first initial must appear as the FIRST character of the username
|
|
portion before the last name (natural "firstname+lastname" order).
|
|
This prevents 'a' from "Amy" spuriously matching inside "ryan".
|
|
"""
|
|
if not first_name_str:
|
|
return 0.5 # neutral
|
|
|
|
# Find where last name appears in username
|
|
last_idx = username.find(last_name) if last_name else -1
|
|
username_before_last = username[:last_idx] if last_idx > 0 else username
|
|
|
|
first_names = first_name_str.split()
|
|
best = 0.0
|
|
|
|
for fn in first_names:
|
|
if len(fn) < 2:
|
|
continue
|
|
fn_init = fn[0]
|
|
if fn in username:
|
|
return 1.0 # full first name in username
|
|
# First initial must be the FIRST character before the last name
|
|
if username_before_last and username_before_last[0] == fn_init:
|
|
best = max(best, 0.7)
|
|
# Fuzzy match
|
|
ratio = SequenceMatcher(None, fn, username).ratio()
|
|
best = max(best, ratio * 0.8)
|
|
|
|
return best
|
|
|
|
|
|
def match_unmatched(unmatched_users, accounts):
|
|
"""Fuzzy match with smarter scoring."""
|
|
# Manual overrides from user knowledge
|
|
MANUAL_MATCHES = {
|
|
# (username, readable_id) -> forces match to specific Gaiia account
|
|
"barbihardin": "James Hardin",
|
|
"daycor": ("Cathy Day", 755),
|
|
"tech": ("Graham McIntire", 1),
|
|
# Add more as discovered
|
|
}
|
|
|
|
# Build lookup by name
|
|
accounts_by_name = {normalize(name): (aid, name, rid) for aid, name, rid in accounts}
|
|
|
|
results = []
|
|
|
|
for user in unmatched_users:
|
|
username = user["username"].lower()
|
|
username_no_digits = re.sub(r'\d+$', '', username)
|
|
|
|
# Check manual overrides first
|
|
if username in MANUAL_MATCHES:
|
|
override = MANUAL_MATCHES[username]
|
|
# tuple form: (display_name, readable_id)
|
|
if isinstance(override, tuple):
|
|
target_name, target_rid = override
|
|
for aid, name, rid in accounts:
|
|
if normalize(name) == normalize(target_name) and rid == target_rid:
|
|
results.append({
|
|
**user,
|
|
"gaiia_id": aid, "gaiia_name": name,
|
|
"gaiia_readable_id": rid,
|
|
"match_type": "fuzzy(manual_override)",
|
|
})
|
|
break
|
|
else:
|
|
# Fallback: match by name only
|
|
norm_target = normalize(target_name)
|
|
if norm_target in accounts_by_name:
|
|
aid, name, rid = accounts_by_name[norm_target]
|
|
results.append({
|
|
**user,
|
|
"gaiia_id": aid, "gaiia_name": name,
|
|
"gaiia_readable_id": rid,
|
|
"match_type": "fuzzy(manual_override)",
|
|
})
|
|
continue
|
|
continue
|
|
|
|
# String form: match by name
|
|
target_name = override
|
|
norm_target = normalize(target_name)
|
|
if norm_target in accounts_by_name:
|
|
aid, name, rid = accounts_by_name[norm_target]
|
|
results.append({
|
|
**user,
|
|
"gaiia_id": aid,
|
|
"gaiia_name": name,
|
|
"gaiia_readable_id": rid,
|
|
"match_type": "fuzzy(manual_override)",
|
|
})
|
|
continue
|
|
|
|
best_match = None
|
|
best_score = 0.0
|
|
best_reason = ""
|
|
|
|
for acct_id, acct_name, readable_id in accounts:
|
|
first_names, last_name = split_name(acct_name)
|
|
aname = normalize(acct_name)
|
|
aname_parts = aname.split()
|
|
|
|
score = 0.0
|
|
reason = ""
|
|
|
|
# ---- Strategy 1: last name match + first name correlation ----
|
|
if last_name and len(last_name) >= 3:
|
|
# Check that last_name appears in username as a distinct substring
|
|
# (not "kim" inside "kimberly")
|
|
idx = username.find(last_name)
|
|
if idx >= 0:
|
|
# Check it's a reasonable boundary match
|
|
# Allow: at start, at end, or surrounded by non-alpha
|
|
before_ok = idx == 0 or not username[idx-1].isalpha()
|
|
after_ok = (idx + len(last_name) == len(username)) or not username[idx+len(last_name)].isalpha()
|
|
if not before_ok and not after_ok:
|
|
# "kim" inside "kimberly" - reject this last name match
|
|
pass
|
|
else:
|
|
# Check first name correlation
|
|
fn_score = first_name_match_score(username, last_name, first_names)
|
|
if fn_score >= 0.5:
|
|
score = 0.7 + (fn_score * 0.2)
|
|
reason = f"last+first: {last_name}+{first_names}(fn={fn_score:.2f})"
|
|
elif fn_score > 0:
|
|
score = 0.6 + fn_score * 0.1
|
|
reason = f"last+weakfirst: {last_name}+{first_names}(fn={fn_score:.2f})"
|
|
else:
|
|
# Last name match but no first name correlation - low confidence
|
|
score = 0.55
|
|
reason = f"lastonly_nofirst: {last_name}"
|
|
|
|
# ---- Strategy 2: Sequence matcher on nodigits variant ----
|
|
if score < 0.6 and username_no_digits != username:
|
|
seq_score = SequenceMatcher(None, username_no_digits, aname).ratio()
|
|
if seq_score > 0.75:
|
|
score = max(score, seq_score * 0.9)
|
|
reason = f"seq_nodigits: {seq_score:.2f}"
|
|
|
|
# ---- Strategy 3: Direct sequence match ----
|
|
if score < 0.6:
|
|
seq_score = SequenceMatcher(None, username, aname).ratio()
|
|
if seq_score > 0.75:
|
|
score = max(score, seq_score * 0.9)
|
|
reason = f"sequence: {seq_score:.2f}"
|
|
|
|
# ---- Strategy 4: All account name parts in username ----
|
|
if score < 0.6:
|
|
all_found = True
|
|
for part in aname_parts:
|
|
if len(part) >= 3 and part not in username:
|
|
all_found = False
|
|
break
|
|
if all_found and len(aname_parts) >= 2:
|
|
score = 0.85
|
|
reason = f"allparts: {aname}"
|
|
|
|
if score > best_score:
|
|
best_score = score
|
|
best_match = (acct_id, acct_name, readable_id, score)
|
|
best_reason = reason
|
|
|
|
# Threshold: 0.7+
|
|
if best_match and best_score >= 0.7:
|
|
results.append({
|
|
**user,
|
|
"gaiia_id": best_match[0],
|
|
"gaiia_name": best_match[1],
|
|
"gaiia_readable_id": best_match[2],
|
|
"match_type": f"fuzzy({best_reason})",
|
|
})
|
|
else:
|
|
results.append({
|
|
**user,
|
|
"gaiia_id": user.get("gaiia_id", ""),
|
|
"gaiia_name": user.get("gaiia_name", ""),
|
|
"gaiia_readable_id": user.get("gaiia_readable_id", ""),
|
|
"match_type": f"still_unmatched(best: {best_reason}, score: {round(best_score,3)})",
|
|
})
|
|
|
|
return results
|
|
|
|
|
|
def resolve_ambiguous(ambiguous_rows, accounts):
|
|
"""For ambiguous matches (duplicate Gaiia names), pick the lowest readable_id."""
|
|
results = []
|
|
for row in ambiguous_rows:
|
|
ambig_names = row["gaiia_name"].replace("AMBIGUOUS: ", "").split(", ")
|
|
candidates = [(a_id, name, r_id) for a_id, name, r_id in accounts if name in ambig_names]
|
|
if candidates:
|
|
candidates.sort(key=lambda x: x[2])
|
|
row["gaiia_id"] = candidates[0][0]
|
|
row["gaiia_name"] = candidates[0][1]
|
|
row["gaiia_readable_id"] = candidates[0][2]
|
|
row["match_type"] = "ambiguous_resolved(lowest_id)"
|
|
else:
|
|
row["match_type"] = "ambiguous_unresolved"
|
|
results.append(row)
|
|
return results
|
|
|
|
|
|
def main():
|
|
reader = csv.DictReader(sys.stdin)
|
|
all_rows = list(reader)
|
|
|
|
unmatched = [r for r in all_rows if r["match_type"] == "no_match"]
|
|
ambiguous = [r for r in all_rows if r["match_type"] == "ambiguous"]
|
|
|
|
print(f"Unmatched: {len(unmatched)}, Ambiguous: {len(ambiguous)}", file=sys.stderr)
|
|
|
|
print("Fetching Gaiia accounts...", file=sys.stderr)
|
|
accounts = fetch_all_accounts()
|
|
print(f"Fetched {len(accounts)} accounts", file=sys.stderr)
|
|
|
|
matched_results = match_unmatched(unmatched, accounts)
|
|
ambig_results = resolve_ambiguous(ambiguous, accounts)
|
|
|
|
all_results = matched_results + ambig_results
|
|
|
|
fieldnames = ["radius_id", "username", "password", "gaiia_id", "gaiia_name", "gaiia_readable_id", "match_type"]
|
|
writer = csv.DictWriter(sys.stdout, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
for r in all_results:
|
|
writer.writerow({k: r.get(k, "") for k in fieldnames})
|
|
|
|
newly_matched = sum(1 for r in matched_results if r["gaiia_id"])
|
|
still = sum(1 for r in matched_results if "still_unmatched" in r.get("match_type", ""))
|
|
print(f"\nNewly matched: {newly_matched}, Still unmatched: {still}", file=sys.stderr)
|
|
|
|
print("\nStill unmatched:", file=sys.stderr)
|
|
for r in matched_results:
|
|
if "still_unmatched" in r.get("match_type", ""):
|
|
print(f" {r['username']:30s} ({r['match_type']})", file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|