260 lines
11 KiB
Python
260 lines
11 KiB
Python
import httpx
|
|
import logging
|
|
from typing import Dict, Optional, Any
|
|
from pydantic import BaseModel, Field, validator
|
|
from .config import Config
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=getattr(logging, Config.LOG_LEVEL))
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class VisitorInfo(BaseModel):
|
|
"""Visitor information model"""
|
|
employeeId: str = Field(..., description="Employee ID", max_length=Config.MAX_EMPLOYEE_ID_LENGTH)
|
|
visitorName: str = Field(..., description="Visitor name", max_length=Config.MAX_NAME_LENGTH)
|
|
reason: str = Field(..., description="Visit reason", max_length=Config.MAX_REASON_LENGTH)
|
|
userId: str = Field(..., description="User ID", max_length=Config.MAX_USER_ID_LENGTH)
|
|
|
|
@validator('employeeId')
|
|
def validate_employee_id(cls, v):
|
|
if not v.strip():
|
|
raise ValueError("Employee ID cannot be empty")
|
|
return v.strip()
|
|
|
|
@validator('visitorName')
|
|
def validate_visitor_name(cls, v):
|
|
if not v.strip():
|
|
raise ValueError("Visitor name cannot be empty")
|
|
return v.strip()
|
|
|
|
@validator('reason')
|
|
def validate_reason(cls, v):
|
|
if not v.strip():
|
|
raise ValueError("Visit reason cannot be empty")
|
|
return v.strip()
|
|
|
|
@validator('userId')
|
|
def validate_user_id(cls, v):
|
|
if not v.strip():
|
|
raise ValueError("User ID cannot be empty")
|
|
return v.strip()
|
|
|
|
class VisitorService:
|
|
"""Visitor service class"""
|
|
|
|
def __init__(self):
|
|
self.base_url = Config.BASE_URL
|
|
self.timeout = Config.VISITOR_API_TIMEOUT
|
|
self.headers = Config.DEFAULT_HEADERS
|
|
|
|
async def query_visitor(self, user_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Query visitor reservation records by user ID
|
|
|
|
Args:
|
|
user_id: User ID
|
|
|
|
Returns:
|
|
Visitor reservation records dictionary
|
|
"""
|
|
try:
|
|
url = f"{self.base_url}{Config.VISITOR_QUERY_ENDPOINT}"
|
|
params = {"userId": user_id}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(url, params=params, headers=self.headers)
|
|
|
|
if response.status_code == 200:
|
|
response_data = response.json()
|
|
|
|
# Extract and filter relevant visitor information
|
|
if response_data.get("code") == 200 and "data" in response_data:
|
|
visitor_data = response_data["data"]
|
|
|
|
# Status mapping
|
|
status_map = {
|
|
0: "Under Review",
|
|
1: "Approved",
|
|
2: "Rejected",
|
|
3: "Unknown"
|
|
}
|
|
|
|
# Gender mapping
|
|
gender_map = {
|
|
"1": "Female",
|
|
"2": "Male",
|
|
"0": "Unknown"
|
|
}
|
|
|
|
# Filter only essential information for terminal display
|
|
filtered_data = {
|
|
"visitorName": visitor_data.get("visitorName"),
|
|
"visitTime": visitor_data.get("visitTime"),
|
|
"expireTime": visitor_data.get("expireTime"),
|
|
"receptionLocation": visitor_data.get("receptionLocation"),
|
|
"status": visitor_data.get("status"),
|
|
"visitorSex": visitor_data.get("visitorSex")
|
|
}
|
|
|
|
# Add text mappings for AI processing
|
|
status_value = filtered_data.get("status")
|
|
gender_value = str(filtered_data.get("visitorSex", ""))
|
|
|
|
enhanced_data = {
|
|
**filtered_data,
|
|
"status_text": status_map.get(status_value, f"Unknown Status({status_value})"),
|
|
"gender_text": gender_map.get(gender_value, f"Unknown({gender_value})")
|
|
}
|
|
|
|
logger.info(f"Successfully queried visitor reservation records: {user_id}")
|
|
return {
|
|
"success": True,
|
|
"data": enhanced_data,
|
|
"message": "Query successful"
|
|
}
|
|
else:
|
|
# Handle API error response
|
|
error_message = response_data.get("msg", "Unknown error")
|
|
logger.warning(f"API returned error: {error_message}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": error_message
|
|
}
|
|
else:
|
|
# Handle error response with specific format
|
|
try:
|
|
error_data = response.json()
|
|
error_message = error_data.get("msg", "Unknown error")
|
|
error_code = error_data.get("code", response.status_code)
|
|
logger.warning(f"API error response: {error_code} - {error_message}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": error_message
|
|
}
|
|
except:
|
|
# Fallback for non-JSON error responses
|
|
logger.error(f"Failed to query reservation records: {response.status_code} - {response.text}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": f"Query failed: HTTP {response.status_code}"
|
|
}
|
|
|
|
except httpx.TimeoutException:
|
|
logger.error(f"Query reservation records timeout: {user_id}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": "Request timeout, please try again later"
|
|
}
|
|
except httpx.RequestError as e:
|
|
logger.error(f"Query reservation records network error: {e}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Query reservation records exception: {e}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": f"System error: {str(e)}"
|
|
}
|
|
|
|
async def register_visitor(self, visitor_info: VisitorInfo) -> Dict[str, Any]:
|
|
"""
|
|
Register visitor information
|
|
|
|
Args:
|
|
visitor_info: Visitor information object
|
|
|
|
Returns:
|
|
Registration result dictionary
|
|
"""
|
|
try:
|
|
url = f"{self.base_url}{Config.VISITOR_REGISTER_ENDPOINT}"
|
|
payload = visitor_info.dict()
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(url, params=payload, headers=self.headers)
|
|
|
|
if response.status_code == 200:
|
|
result_data = response.json()
|
|
api_code = result_data.get("code")
|
|
api_message = result_data.get("msg", "")
|
|
|
|
status_map = {
|
|
0: "Under Review",
|
|
1: "Approved",
|
|
2: "Rejected",
|
|
3: "Unknown"
|
|
}
|
|
|
|
if api_code == 200:
|
|
# Registration successful
|
|
api_data = result_data.get("data")
|
|
status_text = status_map.get(api_data, f"Unknown Status({api_data})") if api_data is not None else "Unknown Status"
|
|
logger.info(f"Successfully registered visitor: {visitor_info.visitorName}")
|
|
return {
|
|
"success": True,
|
|
"data": result_data,
|
|
"message": api_message,
|
|
"status": status_text
|
|
}
|
|
elif api_code == 2:
|
|
# Duplicate registration
|
|
logger.warning(f"Visitor already registered: {visitor_info.visitorName}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": api_message
|
|
}
|
|
elif api_code == 1:
|
|
# Missing visitor name
|
|
logger.warning(f"Missing visitor name: {api_message}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": api_message
|
|
}
|
|
else:
|
|
# Other business errors
|
|
logger.warning(f"API business error for visitor {visitor_info.visitorName}: code={api_code}, message={api_message}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": api_message
|
|
}
|
|
else:
|
|
# HTTP error
|
|
logger.error(f"Visitor registration HTTP error: {response.status_code} - {response.text}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": f"HTTP error: {response.status_code}"
|
|
}
|
|
|
|
except httpx.TimeoutException:
|
|
logger.error(f"Visitor registration timeout: {visitor_info.visitorName}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": "Request timeout, please try again later"
|
|
}
|
|
except httpx.RequestError as e:
|
|
logger.error(f"Visitor registration network error: {e}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Visitor registration exception: {e}")
|
|
return {
|
|
"success": False,
|
|
"data": None,
|
|
"message": f"System error: {str(e)}"
|
|
} |