def fetch_instances(domain: str, origin: str, software: str, command: str, path: str = None):
global _DEPTH
logger.debug("domain='%s',origin='%s',software='%s',command='%s',path='%s',_DEPTH=%d - CALLED!", domain, origin, software, command, path, _DEPTH)
def fetch_instances(domain: str, origin: str, software: str, command: str, path: str = None):
global _DEPTH
logger.debug("domain='%s',origin='%s',software='%s',command='%s',path='%s',_DEPTH=%d - CALLED!", domain, origin, software, command, path, _DEPTH)
domain_helper.raise_on(domain)
if not isinstance(origin, str) and origin is not None:
domain_helper.raise_on(domain)
if not isinstance(origin, str) and origin is not None:
raise ValueError("Parameter 'command' is empty")
elif command in ["fetch_blocks", "fetch_cs", "fetch_bkali", "fetch_relays", "fetch_fedipact", "fetch_joinmobilizon", "fetch_joinmisskey", "fetch_joinfediverse"] and origin is None:
raise ValueError(f"Parameter command='{command}' but origin is None, please fix invoking this function.")
raise ValueError("Parameter 'command' is empty")
elif command in ["fetch_blocks", "fetch_cs", "fetch_bkali", "fetch_relays", "fetch_fedipact", "fetch_joinmobilizon", "fetch_joinmisskey", "fetch_joinfediverse"] and origin is None:
raise ValueError(f"Parameter command='{command}' but origin is None, please fix invoking this function.")
+ elif not isinstance(path, str) and path is not None:
+ raise ValueError(f"Parameter path[]='{type(path)}' is not of type 'str'")
+ elif _DEPTH > 0 and instances.is_recent(domain, "last_instance_fetch"):
+ raise ValueError(f"domain='{domain}' has recently been fetched but function was invoked")
+ elif software is None and not instances.is_recent(domain, "last_nodeinfo"):
try:
logger.debug("Software for domain='%s' is not set, determining ...", domain)
software = determine_software(domain, path)
try:
logger.debug("Software for domain='%s' is not set, determining ...", domain)
software = determine_software(domain, path)
instances.set_last_error(domain, exception)
logger.debug("Determined software='%s' for domain='%s'", software, domain)
instances.set_last_error(domain, exception)
logger.debug("Determined software='%s' for domain='%s'", software, domain)
elif not isinstance(software, str):
raise ValueError(f"Parameter software[]='{type(software)}' is not of type 'str'")
elif not isinstance(software, str):
raise ValueError(f"Parameter software[]='{type(software)}' is not of type 'str'")
logger.debug("Checking if domain='%s' is registered ...", domain)
if not instances.is_registered(domain):
logger.debug("Checking if domain='%s' is registered ...", domain)
if not instances.is_registered(domain):
if instances.has_pending(domain):
logger.debug("Flushing updates for domain='%s' ...", domain)
if instances.has_pending(domain):
logger.debug("Flushing updates for domain='%s' ...", domain)
logger.debug("instance='%s' - BEFORE!", instance)
instance = instance.encode("idna").decode("utf-8")
logger.debug("instance='%s' - BEFORE!", instance)
instance = instance.encode("idna").decode("utf-8")
logger.debug("Checking if domain='%s' has pending updates ...", domain)
if instances.has_pending(domain):
logger.debug("Flushing updates for domain='%s' ...", domain)
logger.debug("Checking if domain='%s' has pending updates ...", domain)
if instances.has_pending(domain):
logger.debug("Flushing updates for domain='%s' ...", domain)
logger.debug("instance='%s',origin='%s',_DEPTH=%d reached!", instance, origin, _DEPTH)
if _DEPTH <= config.get("max_crawl_depth") and len(peerlist) >= config.get("min_peers_length"):
logger.debug("instance='%s',origin='%s',_DEPTH=%d reached!", instance, origin, _DEPTH)
if _DEPTH <= config.get("max_crawl_depth") and len(peerlist) >= config.get("min_peers_length"):
logger.debug("Checking if domain='%s' has pending updates ...", domain)
if instances.has_pending(domain):
logger.debug("Flushing updates for domain='%s' ...", domain)
logger.debug("Checking if domain='%s' has pending updates ...", domain)
if instances.has_pending(domain):
logger.debug("Flushing updates for domain='%s' ...", domain)
-def determine_software(domain: str, path: str = None) -> str:
- logger.debug("domain='%s',path='%s' - CALLED!", domain, path)
+def determine_software(domain: str, path: str = None, nodeinfo_url: str = None) -> str:
+ logger.debug("domain='%s',path='%s',nodeinfo_url='%s' - CALLED!", domain, path, nodeinfo_url)
domain_helper.raise_on(domain)
if not isinstance(path, str) and path is not None:
raise ValueError(f"Parameter path[]='{type(path)}' is not of type 'str'")
domain_helper.raise_on(domain)
if not isinstance(path, str) and path is not None:
raise ValueError(f"Parameter path[]='{type(path)}' is not of type 'str'")
- logger.debug("Determining software for domain='%s',path='%s'", domain, path)
+ logger.debug("Fetching nodeinfo from domain='%s',path='%s',nodeinfo_url='%s' ...", domain, path, nodeinfo_url)
+ data = nodeinfo.fetch(domain, path, nodeinfo_url)
logger.debug("software='%s' has 'powered by' in it", software)
software = version.remove(software_helper.strip_powered_by(software))
logger.debug("software='%s' has 'powered by' in it", software)
software = version.remove(software_helper.strip_powered_by(software))