Scripts of registering discovered locations, create and delete servic…

…es for bitbucket monorepo usecase

catalog-scripts/create_services_bitbucket.py

@@ -0,0 +1,72 @@
+import requests
+import random
+from requests.auth import HTTPBasicAuth
+import nltk
+
+def create_directory_and_yaml(repo_name, num_directories, yaml_filename, yaml_content_template):
+
+    username = ""  # Your Bitbucket username
+    app_password = ""  # Your Bitbucket app password
+    workspace = "" # Your Bitbucket workspace
+    prefix_path = "mock_rserver_root/configs/services/" # Your prefix_path
+    suffix_path = "/.ownership/" # Your suffix_path
+
+    base_url = f"https://api.bitbucket.org/2.0/repositories/{workspace}/test/src"
+
+    english_words = set(nltk.corpus.words.words())
+
+    for i in range(num_directories):
+        url = f"{base_url}"
+        files = []
+        directory_name = random.choice(list(english_words)).lower() + "-service"
+        directory_path = prefix_path + directory_name + suffix_path
+
+        updated_yaml_content = yaml_content_template.replace("<replace with directory_name>", directory_name)
+        updated_yaml_content = updated_yaml_content.replace("<replace with source-location>", f"url:https://bitbucket.org/{workspace}/{repository_name}/src/main/{directory_path}")
+
+        # Create YAML file
+        yaml_file_content = updated_yaml_content.strip()
+        yaml_file_path = directory_path + yaml_filename
+
+        yaml_file_data = {
+            "message": f"Create YAML file '{yaml_file_path}'",
+           f"/{yaml_file_path}": yaml_file_content,
+        }
+
+        response = requests.post(url, auth=HTTPBasicAuth(username, app_password), files=files, data=yaml_file_data)
+        if response.status_code == 201:
+            print(f"Created {directory_name}")
+
+    if response.status_code == 201:
+        print("Files created successfully!")
+    else:
+        print("Failed to upload file. Status code:", response.status_code)
+        print("Error message:", response.text)
+
+repository_name = ""
+num_directories = 50
+yaml_filename = ""
+
+# YAML content template with placeholder for directory name and source location
+yaml_content_template = """
+apiVersion: backstage.io/v1alpha1
+kind: Component
+metadata:
+  tags:
+    - java
+    - map-my-trip
+  name: <replace with directory_name>
+  annotations:
+    backstage.io/source-location: <replace with source-location>
+    backstage.io/techdocs-ref: dir:.
+    jira/project-key: IDP
+    backstage.io/kubernetes-label-selector: 'app=idp-ui'
+    backstage.io/kubernetes-namespace: '63feee14cbf66e3c798c4bdc'
+spec:
+  type: service
+  system: movie
+  lifecycle: experimental
+  owner: Harness_Account_All_Users
+"""
+
+create_directory_and_yaml(repository_name, num_directories, yaml_filename, yaml_content_template)

catalog-scripts/delete_services_bitbucket.py

@@ -0,0 +1,56 @@
+import requests
+import re
+from requests.auth import HTTPBasicAuth
+
+username = "" # Your Bitbucket username
+workspace = "" # Your Bitbucket workspace
+app_password = "" # Your Bitbucket app_password
+repository = "" # Your Bitbucket repo
+branch =  "" # Your Bitbucket branch
+prefix_path = "mock_rserver_root/configs/services/"
+catalog_path = r"mock_rserver_root/configs/services/.*?-service"
+url = f"https://api.bitbucket.org/2.0/repositories/{workspace}/{repository}/src"
+
+def extract_directories(response):
+    if "values" in response:
+        for value in response["values"]:
+            if "path" in value:
+                path = value["path"]
+                print(path)
+                match = re.search(r"mock_rserver_root/configs/services/(.*?-service)", path)
+                if match:
+                    directories.append(match.group(1))
+    return directories
+
+directories = []
+dir_url = f"https://api.bitbucket.org/2.0/repositories/{workspace}/{repository}/src/{branch}/{prefix_path}"
+response = requests.get(dir_url, auth=HTTPBasicAuth(username, app_password))
+if response.status_code != 200:
+    print(f"Failed to fetch the latest commit. Status code: {response.status_code}")
+    exit()
+
+if response.status_code == 200:
+    directories.extend(extract_directories(response.json()))
+
+    # Loop until there are no more "next" links
+    while "next" in response.json():
+        next_url = response.json()["next"]
+        response = requests.get(next_url, auth=HTTPBasicAuth(username, app_password))
+        if response.status_code == 200:
+            directories.extend(extract_directories(response.json()))
+        else:
+            print("Failed to fetch next page of directories")
+            break
+
+count = 0
+unique_directories = set(directories)
+for directory in unique_directories:
+
+	payload = {'files': f'/{prefix_path}/{directory}/.ownership/catalog-info.yaml'}
+	files=[]
+
+	response = requests.post(url,data=payload, files=files, auth=HTTPBasicAuth(username, app_password))
+	if response.status_code == 201:
+		count += 1
+
+print(f"{count} files deleted successfully")

catalog-scripts/register_discovered_locations_bitbucket.py

@@ -0,0 +1,84 @@
+import requests
+import re
+from requests.auth import HTTPBasicAuth
+from requests.packages.urllib3.util.retry import Retry
+from requests.adapters import HTTPAdapter
+
+username = "" # Your Bitbucket username
+workspace = "" # Your Bitbucket workspace
+app_password = "" # Your Bitbucket app_password
+repository = "" # Your Bitbucket repo
+branch = "" # Your Bitbucket branch
+prefix_path = "mock_rserver_root/configs/services/" # Your prefix path
+catalog_path = r"mock_rserver_root/configs/services/(.*?-service)" 
+
+account = "" 
+# Replace with your x-api-key. Refer https://developer.harness.io/docs/platform/automation/api/api-quickstart/#create-a-harness-api-key-and-token to generate one
+x_api_key = ""
+
+api_url = f"https://idp.harness.io/{account}/idp/api/catalog/locations"
+
+def extract_directories(response):
+    if "values" in response:
+        for value in response["values"]:
+            if "path" in value:
+                path = value["path"]
+                match = re.search(catalog_path, path)
+                if match:
+                    directories.append(match.group(1))
+    return directories
+
+directories = []
+dir_url = f"https://api.bitbucket.org/2.0/repositories/{workspace}/{repository}/src/{branch}/{prefix_path}"
+response = requests.get(dir_url, auth=HTTPBasicAuth(username, app_password))
+if response.status_code != 200:
+    print(f"Failed to fetch the latest commit. Status code: {response.status_code}")
+    exit()
+
+if response.status_code == 200:
+    directories.extend(extract_directories(response.json()))
+
+    # Loop until there are no more "next" links
+    while "next" in response.json():
+        next_url = response.json()["next"]
+        response = requests.get(next_url, auth=HTTPBasicAuth(username, app_password))
+        if response.status_code == 200:
+            directories.extend(extract_directories(response.json()))
+        else:
+            print("Failed to fetch next page of directories")
+            break
+
+count = 0
+unique_directories = set(directories)
+
+for directory in unique_directories:
+    api_payload = {
+            "target": f"https://bitbucket.org/{workspace}/{repository}/src/{prefix_path}/{directory}/.ownership/catalog-info.yaml",
+            "type": "url"
+    }
+    api_headers = {
+        # "Authorization": f"Bearer {bearer_token}",
+        "x-api-key": f"{x_api_key}",
+        "Content-Type": "application/json",
+        "Harness-Account": f"{account}"
+    }
+
+    retries = Retry(total=3, backoff_factor=1, status_forcelist=[401, 500, 502, 503, 504])
+    session = requests.Session()
+    session.mount("http://", HTTPAdapter(max_retries=retries))
+    session.mount("https://", HTTPAdapter(max_retries=retries))
+
+    try:
+        api_response = session.post(api_url, json=api_payload, headers=api_headers)
+        if api_response.status_code == 200 or api_response.status_code == 201:
+            print(f"Location registered for file: {directory}")
+            count += 1
+        elif api_response.status_code == 409:
+            print(f"Location already exists for file: {directory}. Refreshing it")
+            count += 1
+        else:
+            print(f"Failed to register location for file: {directory}. Status code: {api_response.status_code}")
+    except requests.exceptions.RequestException as e:
+                    print(f"Failed to make API call for file: {directory}. Error: {str(e)}")
+
+print(f"Registered/Refreshed {count} locations")