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, session: Optional[requests.Session] = None) -> 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:
        if session is not None:
            resp = session.post(url, headers=headers, json=payload, timeout=10)
        else:
            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, session: Optional[requests.Session] = None) -> requests.Response:
    for attempt in range(retries + 1):
        if session is not None:
            resp = session.get(url, headers=headers, timeout=10)
        else:
            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, session: Optional[requests.Session] = None) -> 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.
        """
        close_session = False
        if session is None:
            session = requests.Session(impersonate="chrome120")
            close_session = True

        try:
            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, session=session)
                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}",
                    session=session
                )

            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
            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, retries=0, session=session)
                if resp_cat.status_code == 200 and isinstance(resp_cat.json(), list) and resp_cat.json():
                    categories = resp_cat.json()
            except Exception as e:
                print(f"[INFO] Primary center visacategory query failed for {vac_code_fetched}: {e}")

            if not categories:
                for alt_center in centers:
                    alt_code = alt_center.get("isoCode") or alt_center.get("vacCode")
                    if not alt_code or alt_code == vac_code_fetched:
                        continue
                    try:
                        alt_url_cat = f"https://{domain}/master/visacategory/{target_api_iso3}/{origin_api_iso3}/{alt_code}/en-US"
                        resp_alt_cat = _vfs_get(alt_url_cat, headers=headers, retries=0, session=session)
                        if resp_alt_cat.status_code == 200 and isinstance(resp_alt_cat.json(), list) and resp_alt_cat.json():
                            categories = resp_alt_cat.json()
                            break
                    except Exception as e:
                        continue

            if not categories:
                categories = [{
                    "visaCategoryCode": "Short Stay Visa",
                    "code": "Short Stay Visa",
                    "name": "Short Stay Visa"
                }]

            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 类别",
                        session=session
                    )
                else:
                    selected_cat = _select_candidate_with_llm(
                        categories,
                        f"选择符合签证类型 '{visa_clean}' 的 Visa Category 类别",
                        session=session
                    )

            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, retries=0, session=session)
                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 子类别",
                            session=session
                        )
                    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
            }
        finally:
            if close_session:
                session.close()

    def fetch_free_config(self, origin, target, visa_type, vac_code=None, session: Optional[requests.Session] = 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, session=session)

        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