# -*- 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