import re from urllib.parse import urlsplit, urlunsplit from tasks.models import TaskExternalLink HTTP_LINK_PATTERN = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE) TRAILING_LINK_PUNCTUATION = ".,;:!?)]}" def normalize_http_link(value): candidate = str(value or "").strip().rstrip(TRAILING_LINK_PUNCTUATION) if not candidate or len(candidate) > 2048: return None try: parsed = urlsplit(candidate) except ValueError: return None if parsed.scheme.lower() not in {"http", "https"} or not parsed.hostname: return None return urlunsplit( ( parsed.scheme.lower(), parsed.netloc.lower(), parsed.path, parsed.query, parsed.fragment, ) ) def http_link_identity(value): normalized = normalize_http_link(value) if normalized is None: return None parsed = urlsplit(normalized) path = "" if parsed.path == "/" else parsed.path return urlunsplit((parsed.scheme, parsed.netloc, path, parsed.query, parsed.fragment)) def extract_http_links(text): links = [] identities = set() for match in HTTP_LINK_PATTERN.finditer(text or ""): link = normalize_http_link(match.group(0)) identity = http_link_identity(link) if link is None or identity is None or identity in identities: continue identities.add(identity) links.append(link) return links def add_comment_links_to_task(*, task, text, created_by): existing_identities = { identity for url in task.external_links.values_list("url", flat=True) if (identity := http_link_identity(url)) is not None } created = [] for url in extract_http_links(text): identity = http_link_identity(url) if identity in existing_identities: continue link, was_created = TaskExternalLink.objects.get_or_create( task=task, url=url, defaults={"created_by": created_by}, ) existing_identities.add(identity) if was_created: created.append(link) return created