webapp/app/visa_free_config_fetchers/vfs_fetcher.py

# -*- coding: utf-8 -*-
import os
import re
import time
import json
from typing import List, Dict, Any, Optional
from curl_cffi import requests
from app.visa_free_config_fetchers.base_fetcher import BaseFetcher

COUNTRY_INFO = {
    "ie": {"iso3": "irl", "name": "Ireland"},
    "irl": {"iso3": "irl", "name": "Ireland"},
    "gb": {"iso3": "gbr", "name": "United Kingdom"},
    "uk": {"iso3": "gb", "name": "United Kingdom"},
    "gbr": {"iso3": "gbr", "name": "United Kingdom"},
    "cn": {"iso3": "chn", "name": "China"},
    "chn": {"iso3": "chn", "name": "China"},
    "sg": {"iso3": "sgp", "name": "Singapore"},
    "sgp": {"iso3": "sgp", "name": "Singapore"},
    "au": {"iso3": "aus", "name": "Australia"},
    "aus": {"iso3": "aus", "name": "Australia"},
    "nl": {"iso3": "nld", "name": "Netherlands"},
    "nld": {"iso3": "nld", "name": "Netherlands"},
    "fr": {"iso3": "fra", "name": "France"},
    "fra": {"iso3": "fra", "name": "France"},
    "it": {"iso3": "ita", "name": "Italy"},
    "ita": {"iso3": "ita", "name": "Italy"},
    "at": {"iso3": "aut", "name": "Austria"},
    "aut": {"iso3": "aut", "name": "Austria"},
    "dk": {"iso3": "dnk", "name": "Denmark"},
    "dnk": {"iso3": "dnk", "name": "Denmark"},
    "fi": {"iso3": "fin", "name": "Finland"},
    "fin": {"iso3": "fin", "name": "Finland"},
    "hu": {"iso3": "hun", "name": "Hungary"},
    "hun": {"iso3": "hun", "name": "Hungary"},
    "is": {"iso3": "isl", "name": "Iceland"},
    "isl": {"iso3": "isl", "name": "Iceland"},
    "no": {"iso3": "nor", "name": "Norway"},
    "nor": {"iso3": "nor", "name": "Norway"},
    "es": {"iso3": "esp", "name": "Spain"},
    "esp": {"iso3": "esp", "name": "Spain"},
    "de": {"iso3": "deu", "name": "Germany"},
    "deu": {"iso3": "deu", "name": "Germany"},
    "gr": {"iso3": "grc", "name": "Greece"},
    "grc": {"iso3": "grc", "name": "Greece"},
    "be": {"iso3": "bel", "name": "Belgium"},
    "bel": {"iso3": "bel", "name": "Belgium"},
    "pl": {"iso3": "pol", "name": "Poland"},
    "pol": {"iso3": "pol", "name": "Poland"}
}

ISO2_TO_ISO3_API = {
    "gb": "gbr",
    "uk": "gbr",
    "cn": "chn",
    "ie": "irl",
    "sg": "sgp",
    "au": "aus",
    "nl": "nld",
    "fr": "fra",
    "it": "ita",
    "at": "aut",
    "dk": "dnk",
    "fi": "fin",
    "hu": "hun",
    "is": "isl",
    "no": "nor",
    "es": "esp",
    "de": "deu",
    "gr": "grc",
    "be": "bel",
    "pl": "pol"
}

def _select_candidate_with_llm(candidates: List[Dict[str, Any]], prompt_context: str) -> Dict[str, Any]:
    if not candidates:
        raise RuntimeError("Empty candidate list passed to LLM selection")
    if len(candidates) == 1:
        return candidates[0]

    api_key = os.environ.get("QWEN_API_KEY") or os.environ.get("DASHSCOPE_API_KEY") or "sk-893e895724c6403d81374e515ffaf427"
    url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"

    formatted_candidates = "\n".join([f"{i}: {json.dumps(c, ensure_ascii=False)}" for i, c in enumerate(candidates)])
    prompt = (
        f"上下文要求：{prompt_context}\n"
        f"候选列表：\n{formatted_candidates}\n"
        f"请判断哪个候选最符合要求。仅输出该候选的索引数字（0-indexed），不要有任何其他文字。"
    )

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "qwen-plus",
        "messages": [
            {"role": "system", "content": "你是一个配置选择助手。按指令仅输出数字索引。"},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.0
    }

    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=10)
        if resp.status_code == 200:
            content = resp.json()["choices"][0]["message"]["content"].strip()
            match = re.search(r"\d+", content)
            if match:
                idx = int(match.group())
                if 0 <= idx < len(candidates):
                    return candidates[idx]
    except Exception as e:
        print(f"[WARN] LLM candidate selection failed: {e}")

    return candidates[0]

def _vfs_get(url: str, headers: dict, retries: int = 2) -> requests.Response:
    for attempt in range(retries + 1):
        resp = requests.get(url, headers=headers, impersonate="chrome120", timeout=10)
        if resp.status_code == 200 or attempt == retries:
            return resp
        time.sleep(0.5)
    return resp

class VFSFetcher(BaseFetcher):
    def __init__(self):
        super(VFSFetcher, self).__init__("vfs")

    def fetch_external_vac_info(self, origin: str, target: str, visa_type: str, vac_code: str) -> Dict[str, Any]:
        """
        Call external VFS master API using curl-cffi when vac_info is missing from metadata.
        Raises RuntimeError if data fetching or matching fails.
        """
        origin_clean = origin.lower().strip()
        target_clean = target.lower().strip()
        vac_clean = vac_code.lower().strip()
        visa_clean = visa_type.lower().strip()

        origin_info = COUNTRY_INFO.get(origin_clean, {"iso3": origin_clean})
        target_info = COUNTRY_INFO.get(target_clean, {"iso3": target_clean})
        origin_iso3 = origin_info["iso3"]
        target_iso3 = target_info["iso3"]

        origin_api_iso3 = ISO2_TO_ISO3_API.get(origin_iso3, origin_iso3)
        target_api_iso3 = ISO2_TO_ISO3_API.get(target_iso3, target_iso3)

        domain = "lift-apicn.vfsglobal.com" if origin_api_iso3 in ["chn", "cn"] else "lift-api.vfsglobal.com"
        headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "en-US,en;q=0.9",
            "route": f"{origin_api_iso3}/en/{target_api_iso3}",
            "origin": "https://visa.vfsglobal.com",
            "referer": "https://visa.vfsglobal.com/"
        }

        # 1. Fetch Centers
        url_center = f"https://{domain}/master/center/{target_api_iso3}/{origin_api_iso3}/en-US"
        try:
            resp_center = _vfs_get(url_center, headers=headers)
            if resp_center.status_code != 200:
                raise RuntimeError(f"VFS master center API returned HTTP {resp_center.status_code}")
            centers = resp_center.json()
        except Exception as e:
            raise RuntimeError(f"Failed to fetch VFS centers for {origin_api_iso3}->{target_api_iso3}: {e}")

        if not isinstance(centers, list) or not centers:
            raise RuntimeError(f"No centers returned for {origin_api_iso3}->{target_api_iso3}")

        # Match center candidate
        matched_centers = []
        for c in centers:
            iso_c = str(c.get("isoCode") or "").lower()
            vac_c = str(c.get("vacCode") or "").lower()
            name_c = str(c.get("centerName") or "").lower()
            city_c = str(c.get("city") or "").lower()
            if vac_clean in iso_c or vac_clean in vac_c or vac_clean in name_c or vac_clean in city_c:
                matched_centers.append(c)

        if not matched_centers:
            matched_centers = centers

        if len(matched_centers) == 1:
            selected_center = matched_centers[0]
        else:
            selected_center = _select_candidate_with_llm(
                matched_centers,
                f"选择办理 {origin_api_iso3} 到 {target_api_iso3} 的 VAC 签证中心，目标中心编号/城市: {vac_clean}"
            )

        vac_code_fetched = selected_center.get("isoCode") or selected_center.get("vacCode") or vac_code.upper()
        center_name_fetched = selected_center.get("centerName", "")
        address_fetched = selected_center.get("address", "")

        # 2. Fetch Visa Categories
        url_cat = f"https://{domain}/master/visacategory/{target_api_iso3}/{origin_api_iso3}/{vac_code_fetched}/en-US"
        try:
            resp_cat = _vfs_get(url_cat, headers=headers)
            if resp_cat.status_code != 200:
                raise RuntimeError(f"VFS visa category API returned HTTP {resp_cat.status_code}, resp_cat.text is {resp_cat.text}")
            categories = resp_cat.json()
        except Exception as e:
            raise RuntimeError(f"Failed to fetch VFS visa category for center {vac_code_fetched}: {e}")

        if not isinstance(categories, list) or not categories:
            raise RuntimeError(f"No visa categories returned for center {vac_code_fetched}")

        if len(categories) == 1:
            selected_cat = categories[0]
        else:
            matched_cats = [
                cat for cat in categories
                if visa_clean in str(cat.get("name", "")).lower() or visa_clean in str(cat.get("code", "")).lower()
            ]
            if len(matched_cats) == 1:
                selected_cat = matched_cats[0]
            elif len(matched_cats) > 1:
                selected_cat = _select_candidate_with_llm(
                    matched_cats,
                    f"选择符合签证类型 '{visa_clean}' 的 Visa Category 类别"
                )
            else:
                selected_cat = _select_candidate_with_llm(
                    categories,
                    f"选择符合签证类型 '{visa_clean}' 的 Visa Category 类别"
                )

        cat_code_fetched = selected_cat.get("visaCategoryCode") or selected_cat.get("code", "")
        cat_name_fetched = selected_cat.get("name", "")

        # 3. Fetch Visa Subcategories (Note: unauthenticated subvisacategory API may return HTTP 500 on VFS backend)
        url_sub = f"https://{domain}/master/subvisacategory/{target_api_iso3}/{origin_api_iso3}/{vac_code_fetched}/{cat_code_fetched}/en-US"
        subcat_code_fetched = "To"
        subcat_name_fetched = "Tourist"
        try:
            resp_sub = _vfs_get(url_sub, headers=headers)
            if resp_sub.status_code == 200 and isinstance(resp_sub.json(), list) and resp_sub.json():
                subcategories = resp_sub.json()
                if len(subcategories) == 1:
                    selected_sub = subcategories[0]
                else:
                    selected_sub = _select_candidate_with_llm(
                        subcategories,
                        f"选择符合签证类型 '{visa_clean}' 的 Subvisa Category 子类别"
                    )
                subcat_code_fetched = selected_sub.get("subCategoryCode") or selected_sub.get("code", "To")
                subcat_name_fetched = selected_sub.get("subCategoryName") or selected_sub.get("name", "Tourist")
        except Exception as e:
            print(f"[INFO] Subvisacategory query unauthenticated fallback: {e}")

        return {
            "routing_key": f"slot.{vac_clean}.{target_clean}.{visa_clean}",
            "vac_code": vac_code_fetched,
            "center_name": center_name_fetched,
            "address": address_fetched,
            "category_code": cat_code_fetched,
            "category_name": cat_name_fetched,
            "subcategory_code": subcat_code_fetched,
            "subcategory_name": subcat_name_fetched
        }

    def fetch_free_config(self, origin, target, visa_type, vac_code=None):
        origin_clean = origin.lower().strip()
        target_clean = target.lower().strip()
        vac_clean = (vac_code or "").lower().strip()
        visa_clean = visa_type.lower().strip()

        origin_info = COUNTRY_INFO.get(origin_clean, {"iso3": origin_clean, "name": origin.capitalize()})
        target_info = COUNTRY_INFO.get(target_clean, {"iso3": target_clean, "name": target.capitalize()})

        free_config = {
            "mission_code": target_info["iso3"],
            "mission_name": target_info["name"],
            "country_code": origin_info["iso3"],
            "country_name": origin_info["name"],
            "culture_code": "en-US",
            "language": "en",
            "apt_configs": {}
        }

        vacs = self.metadata.get("vacs", {})
        vac_key = "{}_{}_{}_{}".format(origin_clean, target_clean, vac_clean, visa_clean)
        vac_info = vacs.get(vac_key, {})

        if not vac_info and vac_clean:
            for k, v in vacs.items():
                if k.startswith("{}_{}_{}".format(origin_clean, target_clean, vac_clean)):
                    vac_info = v
                    break

        # If vac_info NOT found in metadata, call external tool / API online!
        if not vac_info and vac_clean:
            vac_info = self.fetch_external_vac_info(origin_clean, target_clean, visa_clean, vac_clean)

        if vac_info or vac_clean:
            routing_key = vac_info.get("routing_key", "slot.{}.{}.{}".format(vac_clean, target_clean, visa_clean))
            free_config["apt_configs"][routing_key] = {
                "center_name": vac_info.get("center_name", ""),
                "address": vac_info.get("address", ""),
                "vac_code": vac_info.get("vac_code", ""),
                "category_name": vac_info.get("category_name", ""),
                "category_code": vac_info.get("category_code", ""),
                "subcategory_name": vac_info.get("subcategory_name", ""),
                "subcategory_code": vac_info.get("subcategory_code", "")
            }

        return free_config