Spaces:
Build error
Build error
File size: 5,042 Bytes
f188029 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | """
Canonical Pydantic models for the visa service automation pipeline.
Aligned with SPUN's Visa Guideline Template.
"""
from datetime import datetime
from enum import StrEnum
from typing import Dict, List, Optional
from pydantic import BaseModel, Field
class RequirementType(StrEnum):
VISA_FREE = "VISA_FREE"
VISA_REQUIRED = "VISA_REQUIRED"
EVISA = "EVISA"
VOA = "VOA" # Visa on Arrival
ETA = "ETA" # Electronic Travel Authorization
NOT_ADMITTED = "NOT_ADMITTED"
class MatchStatus(StrEnum):
NEW = "NEW"
EXISTS_IDENTICAL = "EXISTS_IDENTICAL"
EXISTS_OUTDATED = "EXISTS_OUTDATED"
# --- Eligibility ---
class VisaEligibility(BaseModel):
requirement_type: RequirementType
max_stay_days: Optional[int] = None
source: str
confidence: float = Field(ge=0, le=1)
# --- Document Requirements ---
class DocumentRequirement(BaseModel):
name: str = Field(..., description="e.g. 'Paspor', 'Foto', 'Hotel booking'")
is_required: bool = Field(..., description="True = wajib (*), False = optional")
needs_physical: bool = Field(default=False, description="'butuh fisik'")
format: str = Field(default="digital", description="'digital', 'fisik', 'e-statement'")
notes: Optional[str] = None
class PersonaDocuments(BaseModel):
"""Documents conditional on persona (occupation, marital, sponsorship, age)."""
occupation: Dict[str, List[DocumentRequirement]] = Field(
default_factory=dict,
description="Employee, Entrepreneur, Self-Employed, Students, Retired, Not working",
)
marital_status: Dict[str, List[DocumentRequirement]] = Field(
default_factory=dict,
description="Single, Married, Divorced, Widowed",
)
sponsorship: Dict[str, List[DocumentRequirement]] = Field(
default_factory=dict,
description="Self-sponsored, Individual-sponsored, Institution-sponsored",
)
age_group: Dict[str, List[DocumentRequirement]] = Field(
default_factory=dict,
description="Adult, Children",
)
# --- Form Questions ---
class FormQuestion(BaseModel):
question: str
applies_to: Optional[str] = Field(
None, description="None = general, or persona category"
)
# --- Visa Guideline (maps 1:1 to SPUN's template) ---
class VisaGuideline(BaseModel):
"""Maps 1:1 to SPUN's visa guideline template."""
# Header
visa_name: str
visa_code: Optional[str] = None
# Application info
application_method: str = Field(
..., description="Hybrid / Offline / Online"
)
application_location: Optional[str] = None
client_nationality: str
options: List[str] = Field(
default_factory=list, description="Regular / Express"
)
# Content
description: str = ""
eligibility_requirements: List[str] = Field(default_factory=list)
# Documents
general_documents: List[DocumentRequirement] = Field(default_factory=list)
persona_documents: PersonaDocuments = Field(default_factory=PersonaDocuments)
# Validity
validity_period: Optional[str] = None
stay_period: Optional[str] = None
# Processing
processing_time_min_days: Optional[int] = None
processing_time_max_days: Optional[int] = None
# Contact
embassy_contact: Optional[str] = None
# Pricing
government_fee: Optional[str] = None
cogs: Optional[str] = None
publish_rate: Optional[str] = None
# Process steps
application_steps: List[str] = Field(default_factory=list)
preferred_supply_partner: Optional[str] = None
# Sources
source_urls: List[str] = Field(default_factory=list)
# Form questions
general_form_questions: List[str] = Field(default_factory=list)
persona_form_questions: Dict[str, List[FormQuestion]] = Field(
default_factory=dict
)
# --- E-Visa Assessment ---
class EVisaAssessment(BaseModel):
is_evisa: bool
portal_url: Optional[str] = None
portal_accessible: bool = False
feasibility_score: float = Field(0.0, ge=0, le=1)
blockers: List[str] = Field(default_factory=list)
# --- Service Candidate (pipeline output) ---
class ServiceCandidate(BaseModel):
guideline: VisaGuideline
evisa_assessment: Optional[EVisaAssessment] = None
match_status: MatchStatus = MatchStatus.NEW
existing_service_id: Optional[int] = None
confidence_score: float = Field(0.0, ge=0, le=1)
scraped_at: datetime = Field(default_factory=datetime.utcnow)
# --- Pipeline Request/Response ---
class PipelineRequest(BaseModel):
"""Input to the pipeline orchestrator."""
nationality: str
destination: str
residency: Optional[str] = None
purpose: Optional[str] = None
class PipelineResult(BaseModel):
"""Output from the pipeline for a single nationality/destination pair."""
request: PipelineRequest
eligibility: Optional[VisaEligibility] = None
candidates: List[ServiceCandidate] = Field(default_factory=list)
errors: List[str] = Field(default_factory=list)
|