guanghu-ice-heart/server-tools/light-arrival-navigation/route_recall.py

#!/usr/bin/env python3
import argparse
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
DEFAULT_MAP = ROOT / "routing/persona-contribution-map.json"


def recall(query, map_path=DEFAULT_MAP):
    data = json.loads(Path(map_path).read_text(encoding="utf-8"))
    needle = query.strip().casefold()
    if not needle:
        return {"status": "NO_TRUSTED_PATH", "query": query, "matches": []}
    terms = [part for part in needle.replace("/", " ").split() if part]
    matches = []
    for item in data["contributions"]:
        identifiers = [
            value
            for value in [item.get("id"), item.get("arrival_id"), *item.get("project_ids", [])]
            if isinstance(value, str) and value
        ]
        keywords = [value for value in item.get("keywords", []) if isinstance(value, str) and value]
        searchable = [
            *identifiers,
            item.get("arrival_name"),
            item.get("title"),
            *keywords,
        ]
        haystack = " ".join(value for value in searchable if isinstance(value, str) and value).casefold()
        score = 0
        reasons = []
        for identifier in identifiers:
            if identifier.casefold() in needle:
                score += 100
                reasons.append(identifier)
        for keyword in keywords:
            if keyword.casefold() in needle or keyword.casefold() == needle:
                score += 20
                reasons.append(keyword)
        for term in terms:
            if len(term) >= 2 and term in haystack:
                score += 5
        if needle in haystack:
            score += 10
        if score:
            matches.append({
                "score": score,
                "contribution_id": item["id"],
                "arrival": {"id": item["arrival_id"], "name": item["arrival_name"]},
                "title": item["title"],
                "matched_by": sorted(set(reasons)),
                "paths": item["canonical_paths"],
                "grants_execution_authority": False
            })
    matches.sort(key=lambda row: (-row["score"], row["contribution_id"]))
    return {"status": "FOUND_CONFIDENT_PATH" if matches else "NO_TRUSTED_PATH", "query": query, "matches": matches}


def main():
    parser = argparse.ArgumentParser(description="GLS-0231 read-only contribution route recall")
    parser.add_argument("query")
    parser.add_argument("--map", default=str(DEFAULT_MAP))
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()
    result = recall(args.query, args.map)
    if args.json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return
    print(result["status"])
    for match in result["matches"]:
        print(f'{match["contribution_id"]} · {match["arrival"]["name"]} · {match["title"]}')
        for path in match["paths"]:
            print(f"  - {path}")


if __name__ == "__main__":
    main()