Code refactoring (#36)

* Add FloodWaitError import * Add .idea * code refactoring * code refactoring * code refactoring * code refactoring * Add ruff * code refactoring with ruff * Add Ruff GitHub Action for style checking * Add Ruff GitHub Action for style checking * Add Ruff GitHub Action for style checking * Add Ruff GitHub Action for style checking * Update ruff config * Add Ruff GitHub Action for style checking * code refactoring * Add Ruff GitHub Action for style checking * fix merge conflicts * remove ruff.yml * code review --------- Co-authored-by: ask0n <[email protected]>
IvanGlinkin · May 16, 2024 · d27e8f4 · d27e8f4
1 parent 290e4bd
commit d27e8f4
Show file tree

Hide file tree

Showing 7 changed files with 365 additions and 294 deletions.
diff --git a/.gitignore b/.gitignore
@@ -7,3 +7,4 @@ avatars
 __pycache__/
 .DS_Store
 reports-*
+.idea
diff --git a/GUI_template.py b/GUI_template.py
diff --git a/backend/banners.py b/backend/banners.py
@@ -1,6 +1,6 @@
 from .functions import get_location_details
 
-### Banner
+# Banner
 banner = """
  ██████╗██╗      ██████╗ ███████╗███████╗     ██████╗██╗██████╗  ██████╗██╗   ██╗██╗████████╗                      
 ██╔════╝██║     ██╔═══██╗██╔════╝██╔════╝    ██╔════╝██║██╔══██╗██╔════╝██║   ██║██║╚══██╔══╝                      
@@ -22,45 +22,57 @@
 [ ! ] https://t.me/EASM_HydrAttack                                                                                                               
 """
 
-def print_geo_coordinater(lat,lon):
+
+def print_geo_coordinater(lat, lon):
     print("[ * ] Harvesting information based on the next coordinates:")
     print("\t[ * * ] Latitude: ", lat)
     print("\t[ * * ] Longitude:", lon)
 
-def pring_city_by_geo(lat,lon):
+
+def print_city_by_geo(lat, lon):
     town, city, country = get_location_details(lat, lon)
     print(f"\t[ * * ] Country:   {country}\n\t[ * * ] City:\t   {city}\n\t[ * * ] Town:\t   {town}\n")
 
+
 def print_len_steps(steps, radius):
     print("[ * ] Overall steps to be performed:", steps, ", with overall diameter", radius * 2, "meters")
 
+
 def print_telegram_initialization():
     print("\n[ * ] Telegram client initialization...", end="")
 
+
 def print_successfully():
     print("successfully")
 
+
 def print_start_harvesting():
     print("\n[ * ] Start harvesting data:")
 
+
 def print_current_step(step, lat, lon):
     print(f"\t[ {step} ] Latitude {round(lat, 4)}, Longitude {round(lon, 4)}")
 
+
 def print_update_local_json():
     print("\t\t[ > ] Harvesting data finished")
     print("\t\t[ > ] Updating JSON file...", end="")
 
+
 def print_update_html():
     print("\t\t[ > ] Generating HTML file...", end="")
 
+
 def print_files_stored(report_json_directory, report_html_directory, filename):
     print("\n[ * ] Harvesting is finished and final files are generated:")
-    print("\t[ * * ] JSON:" , report_json_directory + filename + ".json")
-    print("\t[ * * ] HTML:" , report_html_directory + filename + ".html")
+    print("\t[ * * ] JSON:", report_json_directory + filename + ".json")
+    print("\t[ * * ] HTML:", report_html_directory + filename + ".html")
+
 
 def print_combined_data():
     print("\n[ * ] Combining all JSON files together and generating the global HTML map file")
 
+
 def finishing_application():
     print("\n[ * ] Everything has been executed successfully!")
     print("\t[ * * ] Enjoy your CCTV data!")
diff --git a/backend/combine_data.py b/backend/combine_data.py
@@ -1,13 +1,16 @@
 from .functions import combine_json_files
 from .json_into_html import generate_html_from_json
 
-def combine_data(report_json_directory="./reports-json/", report_html_directory="./reports-html/"):
 
+def combine_data(report_json_directory="./reports-json/", report_html_directory="./reports-html/"):
     combine_json_files(report_json_directory, f"{report_json_directory}_combined_data.json")
 
     # Generate the HTML file from JSON
-    generate_html_from_json(f"{report_json_directory}_combined_data.json", f"{report_html_directory}_combined_data.html")
+    generate_html_from_json(
+        f"{report_json_directory}_combined_data.json", f"{report_html_directory}_combined_data.html"
+    )
+
 
 # Call the function to execute the functionality
 if __name__ == "__main__":
-    combine_data()
+    combine_data()
diff --git a/backend/functions.py b/backend/functions.py
@@ -18,76 +18,82 @@ def generate_pattern(length):
     pattern.append((length - 3) // 2 - 1)
     return pattern
 
+
 def calculate_length(metres):
-    remainder = ((metres + 400) % 800)
+    remainder = (metres + 400) % 800
     if remainder == 0:
         return metres
     else:
-        return (metres - remainder + 800)
+        return metres - remainder + 800
+
 
 # Function to calculate coordinates based on steps and direction
 def calculate_coordinates(lat, lon, direction, distance):
     # Radius of the Earth in kilometers
-    R = 6378.1  # Earth radius in km
+    r = 6378.1  # Earth radius in km
     # Convert latitude and longitude from degrees to radians
     lat_rad = radians(lat)
-    lon_rad = radians(lon)
+
     # Calculate new latitude and longitude based on direction and distance
-    if direction == 'starting':
+    if direction == "starting":
         new_lon = lon
         new_lat = lat
-    elif direction == 'west':
-        new_lon = lon - (distance / (R * cos(lat_rad))) * (180 / pi)
+    elif direction == "west":
+        new_lon = lon - (distance / (r * cos(lat_rad))) * (180 / pi)
         new_lat = lat
-    elif direction == 'south':
-        new_lat = lat - (distance / R) * (180 / pi)
+    elif direction == "south":
+        new_lat = lat - (distance / r) * (180 / pi)
         new_lon = lon
-    elif direction == 'east':
-        new_lon = lon + (distance / (R * cos(lat_rad))) * (180 / pi)
+    elif direction == "east":
+        new_lon = lon + (distance / (r * cos(lat_rad))) * (180 / pi)
         new_lat = lat
-    elif direction == 'north':
-        new_lat = lat + (distance / R) * (180 / pi)
+    elif direction == "north":
+        new_lat = lat + (distance / r) * (180 / pi)
         new_lon = lon
     else:
         raise ValueError("Invalid direction")
     return new_lat, new_lon
 
+
 # Function to combine all json files into one
 def combine_json_files(folder_path, output_file):
     combined_data = {}
     for filename in listdir(folder_path):
         file_path = path.join(folder_path, filename)
-        if filename.endswith('.json') and filename != output_file:
-            with open(file_path, 'r') as file:
+        if filename.endswith(".json") and filename != output_file:
+            with open(file_path) as file:
                 data = load(file)
                 combined_data.update(data)  # Merge dictionaries
 
-    with open(output_file, 'w') as output:
+    with open(output_file, "w") as output:
         dump(combined_data, output, indent=4)
 
+
 # Get the city based on coordinates
 def get_location_details(latitude, longitude):
-    url = f"https://nominatim.openstreetmap.org/reverse?lat={latitude}&lon={longitude}&format=json"
-    headers = {
-    'User-Agent': 'CCTV Bot'
-	}
-    response = get(url, headers=headers)
-    data = response.json()
-    if 'address' in data:
-        town = data['address'].get('town', '')
-        city = data['address'].get('city', '')
-        country = data['address'].get('country', '')
-        return town, city, country
-    else:
-        return None, None, None
+    url = "https://nominatim.openstreetmap.org/reverse"
+    headers = {"User-Agent": "CCTV Bot"}
+    response = get(url, headers=headers, params={"lat": latitude, "lon": longitude, "format": "json"})
+
+    if response.status_code == 200 and "application/json" in response.headers.get("Content-Type"):
+        data: dict = response.json()
+        if "address" in data:
+            town = data["address"].get("town", "")
+            city = data["address"].get("city", "")
+            country = data["address"].get("country", "")
+            return town, city, country
+
+    return "", "", ""
+
 
 def countdown_timer(duration):
     for remaining in range(duration, 0, -1):
         print(f"\t\t[ > ] Sleeping for {remaining} seconds before processing the next coordinates...", end="\r")
         sleep(1)
     print(" " * 100, end="\r")  # Clear the line after countdown
 
-#Download avatars
+
+# Download avatars
 def download_avatar(user_id, username, user_url, output_folder):
     avatar_filename = path.join(output_folder, f"{user_id}-{username}.jpg")
     if username is None:
@@ -100,56 +106,98 @@ def download_avatar(user_id, username, user_url, output_folder):
     try:
         response = get(user_url)
         if response.status_code == 200:
-            soup = BeautifulSoup(response.text, 'html.parser')
+            soup = BeautifulSoup(response.text, "html.parser")
             meta_tag = soup.find("meta", property="og:image")
             if meta_tag and "cdn-telegram.org" in meta_tag["content"]:
                 image_url = meta_tag["content"]
                 response = get(image_url)
                 if response.status_code == 200:
-                    with open(avatar_filename, 'wb') as image_file:
+                    with open(avatar_filename, "wb") as image_file:
                         image_file.write(response.content)
                     print(f"Downloaded avatar for user {user_id}-{username} successfully")
                 else:
-                    print(f"Failed to download avatar for user {user_id}-{username}. Status code: {response.status_code}")
+                    print(
+                        f"Failed to download avatar for user {user_id}-{username}. Status code: {response.status_code}"
+                    )
             else:
                 print(f"No profile photo found for user {user_id}-{username}")
         else:
             print(f"Failed to fetch user page for user {user_id}-{username}. Status code: {response.status_code}")
     except Exception as e:
         print(f"Error downloading avatar for user {user_id}-{username}: {e}")
 
+
 def download_avatars(json_file, output_folder):
     print(f"Starting avatars download based on {json_file}...")
-    with open(json_file, 'r') as f:
+    with open(json_file) as f:
         data = load(f)
 
     with ThreadPoolExecutor(max_workers=5) as executor:
         for user_id, user_data in data.items():
             # Skip users without a photo_id
-            if user_data.get('photo_id') is None:
+            if user_data.get("photo_id") is None:
                 continue
-            username = user_data.get('username', '')
+            username = user_data.get("username", "")
             user_url = f"https://t.me/{username}"
             executor.submit(download_avatar, user_id, username, user_url, output_folder)
 
+
 def create_config(file_path):
     """Create a custom YAML configuration file by asking the user for input."""
 
     settings = {
-        'api_config': {
-            'phone': {'prompt': 'TG phone number: ', 'default': None, 'mandatory': True},
-            'api_id': {'prompt': 'TG api_id: ', 'default': None, 'mandatory': True, 'type': int},
-            'api_hash': {'prompt': 'TG api_hash:', 'default': None, 'mandatory': True},
+        "api_config": {
+            "phone": {
+                "prompt": "TG phone number: ",
+                "default": None,
+                "mandatory": True
+            },
+            "api_id": {
+                "prompt": "TG api_id: ",
+                "default": None,
+                "mandatory": True,
+                "type": int
+            },
+            "api_hash": {
+                "prompt": "TG api_hash:",
+                "default": None,
+                "mandatory": True
+            },
+        },
+        "location": {
+            "lat": {
+                "prompt": "Starting latitude [51.51404]: ",
+                "default": 51.51404,
+                "mandatory": False,
+                "type": float
+            },
+            "lon": {
+                "prompt": "Starting longitude [-0.15063]: ",
+                "default": -0.15063,
+                "mandatory": False,
+                "type": float,
+            },
+            "meters": {
+                "prompt": "Search radius(meters) [1200]: ",
+                "default": 1200,
+                "mandatory": False,
+                "type": int
+            },
         },
-        'location': {
-            'lat': {'prompt': 'Starting latitude [51.51404]: ', 'default': 51.51404, 'mandatory': False, 'type': float},
-            'lon': {'prompt': 'Starting longitude [-0.15063]: ', 'default': -0.15063, 'mandatory': False, 'type': float},
-            'meters': {'prompt': 'Search radius(meters) [1200]: ', 'default': 1200, 'mandatory': False, 'type': int},
+        "misc": {
+            "timesleep": {
+                "prompt": "Waiting time between locations(sec) [30]: ",
+                "default": 30,
+                "mandatory": False,
+                "type": int,
+            },
+            "speed_kmh": {
+                "prompt": "Moving speed between locations(km/h) [50]: ",
+                "default": 50,
+                "mandatory": False,
+                "type": int,
+            },
         },
-        'misc': {
-            'timesleep': {'prompt': 'Waiting time between locations(sec) [30]: ', 'default': 30, 'mandatory': False, 'type': int},
-            'speed_kmh': {'prompt': 'Moving speed between locations(km/h) [50]: ', 'default': 50, 'mandatory': False, 'type': int},
-        }
     }
 
     # Create a dictionary to store the configurations
@@ -158,33 +206,34 @@ def create_config(file_path):
     print("\nCheck README.md and prepare required values at https://my.telegram.org/auth")
     # Iterate over the settings dictionary to prompt user for each value
     for group, group_settings in settings.items():
-         for setting, details in group_settings.items():
+        for setting, details in group_settings.items():
             while True:  # Loop until valid input is given or the program exits
-                user_input = input(details['prompt'])
-                if details['mandatory'] and not user_input.strip():  # Check if the input is mandatory and empty
+                user_input = input(details["prompt"])
+                if details["mandatory"] and not user_input.strip():  # Check if the input is mandatory and empty
                     print(f"Value {setting} is mandatory. Exiting program.")
                     sys.exit()  # Exit the program if input is mandatory and not provided
                 elif user_input.strip():
-                    value = details['type'](user_input) if 'type' in details else user_input
+                    value = details["type"](user_input) if "type" in details else user_input
                     config_data[group][setting] = value
                     break
-                elif details['default'] is not None:  # Use default if available and input is not mandatory
-                    config_data[group][setting] = details['default']
+                elif details["default"] is not None:  # Use default if available and input is not mandatory
+                    config_data[group][setting] = details["default"]
                     break
                 else:
                     print(f"No input provided for {setting}, and no default available. Exiting program.")
                     sys.exit()  # Exit if no default is available and input is not provided
 
     # Write the collected data to a YAML file
-    with open(file_path, 'w') as file:
+    with open(file_path, "w") as file:
         yaml.safe_dump(config_data, file)
     print(f"Config file created at {file_path}")
 
+
 def load_config(file_path):
     """Load the YAML configuration file, creating it if it does not exist."""
     if not path.exists(file_path):
         print(f"No config file found at {file_path}. Creating initial configuration...")
         create_config(file_path)
 
-    with open(file_path, 'r') as file:
+    with open(file_path) as file:
         return yaml.safe_load(file)