mirror of
https://github.com/open-webui/open-webui.git
synced 2026-08-02 06:00:49 -05:00
58 lines
1.4 KiB
Python
58 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
from urllib.parse import urlparse
|
|
|
|
import validators
|
|
from open_webui.retrieval.web.utils import resolve_hostname
|
|
from open_webui.utils.misc import get_allow_block_lists, is_host_allowed
|
|
from pydantic import BaseModel
|
|
|
|
|
|
def get_filtered_results(results, filter_list):
|
|
if not filter_list:
|
|
return results
|
|
|
|
allow_list, block_list = get_allow_block_lists(filter_list)
|
|
resolve_ips = False
|
|
for entry in allow_list + block_list:
|
|
try:
|
|
ipaddress.ip_address(entry)
|
|
except ValueError:
|
|
continue
|
|
resolve_ips = True
|
|
break
|
|
|
|
filtered_results = []
|
|
|
|
for result in results:
|
|
url = result.get('url') or result.get('link', '') or result.get('href', '')
|
|
if not validators.url(url):
|
|
continue
|
|
|
|
domain = urlparse(url).hostname
|
|
if not domain:
|
|
continue
|
|
|
|
hostnames = [domain]
|
|
|
|
if resolve_ips:
|
|
try:
|
|
ipv4_addresses, ipv6_addresses = resolve_hostname(domain)
|
|
hostnames.extend(ipv4_addresses)
|
|
hostnames.extend(ipv6_addresses)
|
|
except Exception:
|
|
pass
|
|
|
|
if is_host_allowed(hostnames, filter_list):
|
|
filtered_results.append(result)
|
|
continue
|
|
|
|
return filtered_results
|
|
|
|
|
|
class SearchResult(BaseModel):
|
|
link: str
|
|
title: str | None
|
|
snippet: str | None
|