Add URL checks to responses (#250)

Co-authored-by: UltralyticsAssistant <[email protected]>

action.yml

@@ -91,6 +91,9 @@ runs:
 
     - name: Install Dependencies
       # Note tomli required for codespell with pyproject.toml
+      # For debug:
+      #   python -m pip install --upgrade pip wheel
+      #   pip install -q git+https://github.com/ultralytics/actions@main codespell tomli
       if: (github.event_name == 'pull_request' || github.event_name == 'pull_request_target') && github.event.action != 'closed'
       run: |
         pip install -q ultralytics-actions codespell tomli

actions/__init__.py

@@ -22,4 +22,4 @@
 #     ├── test_summarize_pr.py
 #     └── ...
 
-__version__ = "0.0.6"
+__version__ = "0.0.7"

actions/utils/common_utils.py

@@ -1,8 +1,92 @@
 # Ultralytics Actions 🚀, AGPL-3.0 license https://ultralytics.com/license
 
 import re
+import socket
+import time
+import urllib
+from concurrent.futures import ThreadPoolExecutor
 
 
 def remove_html_comments(body: str) -> str:
     """Removes HTML comments from a string using regex pattern matching."""
     return re.sub(r"<!--.*?-->", "", body, flags=re.DOTALL).strip()
+
+
+def clean_url(url):
+    """Remove extra characters from URL strings."""
+    for _ in range(3):
+        url = str(url).strip('"').strip("'").rstrip(".,:;!?`\\").replace(".git@main", "").replace("git+", "")
+    return url
+
+
+def is_url(url, check=True, max_attempts=3, timeout=2):
+    """Check if string is URL and check if URL exists."""
+    allow_list = (
+        "localhost",
+        "127.0.0",
+        ":5000",
+        ":3000",
+        ":8000",
+        ":8080",
+        ":6006",
+        "MODEL_ID",
+        "API_KEY",
+        "url",
+        "example",
+        "mailto:",
+    )
+    try:
+        # Check allow list
+        if any(x in url for x in allow_list):
+            return True
+
+        # Check structure
+        result = urllib.parse.urlparse(url)
+        if not all([result.scheme, result.netloc]):
+            return False
+
+        # Check response
+        if check:
+            for attempt in range(max_attempts):
+                try:
+                    req = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "Chrome/120.0.0.0"})
+                    with urllib.request.urlopen(req, timeout=timeout) as response:
+                        return response.getcode() < 400
+                except (urllib.error.URLError, socket.timeout):
+                    if attempt == max_attempts - 1:  # last attempt
+                        return False
+                    time.sleep(2**attempt)  # exponential backoff
+            return False
+        return True
+    except Exception:
+        return False
+
+
+def check_links_in_string(text, verbose=True, return_bad=False):
+    """Process a given text, find unique URLs within it, and check for any 404 errors."""
+    pattern = (
+        r"\[([^\]]+)\]\(([^)]+)\)"  # Matches Markdown links [text](url)
+        r"|"
+        r"("  # Start capturing group for plaintext URLs
+        r"(?:https?://)?"  # Optional http:// or https://
+        r"(?:www\.)?"  # Optional www.
+        r"[\w.-]+"  # Domain name and subdomains
+        r"\.[a-zA-Z]{2,}"  # TLD
+        r"(?:/[^\s\"')\]]*)?"  # Optional path
+        r")"
+    )
+    all_urls = []
+    for md_text, md_url, plain_url in re.findall(pattern, text):
+        url = md_url or plain_url
+        if url and urllib.parse.urlparse(url).scheme:
+            all_urls.append(url)
+
+    urls = set(map(clean_url, all_urls))  # remove extra characters and make unique
+    with ThreadPoolExecutor(max_workers=16) as executor:  # multi-thread
+        bad_urls = [url for url, valid in zip(urls, executor.map(lambda x: not is_url(x, check=True), urls)) if valid]
+
+    passing = not bad_urls
+    if verbose and not passing:
+        print(f"WARNING ⚠️ errors found in URLs {bad_urls}")
+
+    return (passing, bad_urls) if return_bad else passing

actions/utils/openai_utils.py

@@ -1,24 +1,45 @@
 # Ultralytics Actions 🚀, AGPL-3.0 license https://ultralytics.com/license
 
 import os
+import random
+from typing import Dict, List
 
 import requests
 
+from actions.utils.common_utils import check_links_in_string
+
 OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o")
 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
 
 
-def get_completion(messages: list) -> str:
+def get_completion(
+    messages: List[Dict[str, str]],
+    check_links: bool = True,
+    remove: List[str] = (" @giscus[bot]",),  # strings to remove from response
+) -> str:
     """Generates a completion using OpenAI's API based on input messages."""
     assert OPENAI_API_KEY, "OpenAI API key is required."
     url = "https://api.openai.com/v1/chat/completions"
     headers = {"Authorization": f"Bearer {OPENAI_API_KEY}", "Content-Type": "application/json"}
-    data = {"model": OPENAI_MODEL, "messages": messages}
 
-    r = requests.post(url, headers=headers, json=data)
-    r.raise_for_status()
-    content = r.json()["choices"][0]["message"]["content"].strip()
-    remove = [" @giscus[bot]"]
-    for x in remove:
-        content = content.replace(x, "")
+    content = ""
+    max_retries = 2
+    for attempt in range(max_retries + 2):  # attempt = [0, 1, 2, 3], 2 random retries before asking for no links
+        data = {"model": OPENAI_MODEL, "messages": messages, "seed": random.randint(1, 1000000)}
+
+        r = requests.post(url, headers=headers, json=data)
+        r.raise_for_status()
+        content = r.json()["choices"][0]["message"]["content"].strip()
+        for x in remove:
+            content = content.replace(x, "")
+        if not check_links or check_links_in_string(content):  # if no checks or checks are passing return response
+            return content
+
+        if attempt < max_retries:
+            print(f"Attempt {attempt + 1}: Found bad URLs. Retrying with a new random seed.")
+        else:
+            print("Max retries reached. Updating prompt to exclude links.")
+            messages.append({"role": "user", "content": "Please provide a response without any URLs or links in it."})
+            check_links = False  # automatically accept the last message
+
     return content