spunteam/streamlit-web-crawler
0
1import csv
2import time
3import os
4from visa_scraper import IndonesianVisaScraper
5from typing import List, Dict, Any
6
7# --- Configuration ---
8OUTPUT_CSV_FILE = 'indonesian_visa_data_all.csv'
9# Add a delay between requests to avoid overwhelming the server (in seconds)
10REQUEST_DELAY = 0.5
11
12# --- Test Mode Settings ---
13# Set TEST_MODE to True to run on a small sample.
14# Set it to False to run on all countries.
15TEST_MODE = True
16TEST_LIMIT = 5 # Number of countries to test if TEST_MODE is True
17
18def save_to_csv(data: List[Dict[str, Any]], filename: str):
19 """
20 Saves a list of dictionaries to a CSV file. Appends if the file exists,
21 otherwise creates a new file and writes the header.
22 """
23 if not data:
24 return
25
26 # Check if the file already exists to decide whether to write a header
27 file_exists = os.path.isfile(filename)
28
29 with open(filename, 'a', newline='', encoding='utf-8') as csvfile:
30 fieldnames = data[0].keys()
31 writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
32
33 if not file_exists:
34 writer.writeheader() # Write header only if the file is new
35
36 writer.writerows(data)
37
38def main():
39 """
40 Main function to orchestrate the scraping and saving process.
41 """
42 scraper = IndonesianVisaScraper()
43
44 # Define the CSV headers
45 csv_headers = [
46 'country', 'main_purpose', 'sub_activity_name', 'visa_name',
47 'visa_code', 'duration', 'stay_summary', 'cost_summary',
48 'is_multiple_entry', 'is_visa_on_arrival', 'is_guarantor_required',
49 'passport_validity', 'full_description', 'detailed_info_html',
50 'visa_id', 'sub_activity_id'
51 ]
52
53 # Create an empty file with headers first to ensure it's clean
54 with open(OUTPUT_CSV_FILE, 'w', newline='', encoding='utf-8') as f:
55 writer = csv.DictWriter(f, fieldnames=csv_headers)
56 writer.writeheader()
57
58 total_purposes = len(scraper.PARENT_ACTIVITY_MAPPING)
59
60 countries_to_scrape = list(scraper.COUNTRY_MAPPING.keys())
61
62 if TEST_MODE:
63 print("--- 🧪 TEST MODE ENABLED ---")
64 print(f"Running for the first {TEST_LIMIT} countries only.")
65 countries_to_scrape = countries_to_scrape[:TEST_LIMIT]
66 else:
67 print("--- 🚀 FULL SCRAPE MODE ---")
68
69 total_countries = len(countries_to_scrape)
70
71 print(f"Target file: {OUTPUT_CSV_FILE}")
72 print(f"Scraping for {total_countries} countries and {total_purposes} main purposes.")
73 print("-" * 50)
74
75 # 1. Iterate through each country
76 for i, country_name in enumerate(countries_to_scrape, 1):
77 country_id = scraper.get_country_id(country_name)
78
79 # 2. Iterate through each main purpose (parent activity)
80 for j, parent_activity_name in enumerate(scraper.PARENT_ACTIVITY_MAPPING.keys(), 1):
81 parent_activity_id = scraper.get_parent_activity_id(parent_activity_name)
82
83 print(f"({i}/{total_countries}) {country_name} | ({j}/{total_purposes}) {parent_activity_name}")
84
85 # 3. Get all sub-activities for the main purpose
86 time.sleep(REQUEST_DELAY) # Respectful delay
87 sub_activities = scraper.get_sub_activities(parent_activity_id)
88
89 if not sub_activities:
90 print(" -> No sub-activities found. Skipping.")
91 continue
92
93 # 4. Iterate through each sub-activity
94 for sub_activity in sub_activities:
95 sub_activity_id = sub_activity['id']
96 sub_activity_name = sub_activity['name']
97 print(f" -> Sub-activity: {sub_activity_name}")
98
99 # 5. Get available visa types for the sub-activity and country
100 time.sleep(REQUEST_DELAY) # Respectful delay
101 visa_types_data = scraper.get_visa_types(sub_activity_id, country_id)
102
103 rows_to_write = []
104
105 if not visa_types_data or not visa_types_data.get('data'):
106 message = "No specific visa found"
107 if visa_types_data and visa_types_data.get('status') == 'empty':
108 message = visa_types_data.get('message', "Guarantor likely required")
109
110 print(f" -> {message}")
111 # Add a row indicating why there's no visa data
112 row = {field: '' for field in csv_headers}
113 row.update({
114 'country': country_name,
115 'main_purpose': parent_activity_name,
116 'sub_activity_name': sub_activity_name,
117 'visa_name': message,
118 })
119 rows_to_write.append(row)
120 else:
121 visa_list = visa_types_data.get('data', [])
122 print(f" -> Found {len(visa_list)} potential visa type(s). Fetching details...")
123
124 # 6. Iterate through each visa type and get full details
125 for visa_type in visa_list:
126 visa_id = visa_type['id']
127 time.sleep(REQUEST_DELAY) # Respectful delay
128 details_response = scraper.get_visa_full_details(visa_id)
129
130 if details_response and details_response['success']:
131 details = details_response['data']
132
133 # 7. Prepare a row with all collected data
134 row = {
135 'country': country_name,
136 'main_purpose': parent_activity_name,
137 'sub_activity_name': sub_activity_name,
138 'visa_name': details.get('name', visa_type.get('name')),
139 'visa_code': details.get('code', 'N/A'),
140 'duration': details.get('duration_time', 'N/A'),
141 'stay_summary': visa_type.get('stay_summary', 'N/A'),
142 'cost_summary': visa_type.get('cost_summary', 'N/A'),
143 'is_multiple_entry': details.get('is_multiple_entry', False),
144 'is_visa_on_arrival': details.get('is_arrival', False),
145 'is_guarantor_required': details.get('is_guarantor', False),
146 'passport_validity': f"{details.get('passport_value', 'N/A')} {details.get('passport_unit', '')}".strip(),
147 'full_description': details.get('description', 'N/A'),
148 'detailed_info_html': details.get('info_html', 'N/A'),
149 'visa_id': visa_id,
150 'sub_activity_id': sub_activity_id,
151 }
152 rows_to_write.append(row)
153 print(f" - Fetched details for: {row['visa_name']}")
154 else:
155 print(f" - FAILED to fetch details for visa ID {visa_id}")
156
157 # 8. Append the collected rows to the CSV file
158 if rows_to_write:
159 save_to_csv(rows_to_write, OUTPUT_CSV_FILE)
160
161 print("-" * 50)
162 print(f"✅ Scraping complete! Data saved to {OUTPUT_CSV_FILE}")
163
164if __name__ == "__main__":
165 main()