from fba import fba
def check_instance(args: argparse.Namespace) -> int:
- print(f"DEBUG: args.domain='{args.domain}' - CALLED!")
+ # DEBUG: print(f"DEBUG: args.domain='{args.domain}' - CALLED!")
status = 0
if not validators.domain(args.domain):
print(f"WARNING: args.domain='{args.domain}' is not valid")
else:
print(f"INFO: args.domain='{args.domain}' is not known")
- print(f"DEBUG: status={status} - EXIT!")
+ # DEBUG: print(f"DEBUG: status={status} - EXIT!")
return status
def fetch_bkali(args: argparse.Namespace):
- print(f"DEBUG: args[]={type(args)} - CALLED!")
+ # DEBUG: print(f"DEBUG: args[]={type(args)} - CALLED!")
domains = list()
try:
fetched = fba.post_json_api("gql.api.bka.li", "/v1/graphql", json.dumps({
"query": "query domainlist {nodeinfo(order_by: {domain: asc}) {domain}}"
}))
- print(f"DEBUG: fetched({len(fetched)})[]='{type(fetched)}'")
+ # DEBUG: print(f"DEBUG: fetched({len(fetched)})[]='{type(fetched)}'")
if len(fetched) == 0:
raise Exception("WARNING: Returned no records")
elif not "data" in fetched:
raise Exception(f"WARNING: fetched()={len(fetched['data'])} does not contain element 'nodeinfo'")
for entry in fetched["data"]["nodeinfo"]:
- print(f"DEBUG: entry['{type(entry)}']='{entry}'")
+ # DEBUG: print(f"DEBUG: entry['{type(entry)}']='{entry}'")
if not "domain" in entry:
print(f"WARNING: entry does not contain 'domain' - SKIPPED!")
continue
print(f"WARNING: domain='{entry['domain']}' is not a valid domain - SKIPPED!")
continue
elif fba.is_blacklisted(entry["domain"]):
- print(f"DEBUG: domain='{entry['domain']}' is blacklisted - SKIPPED!")
+ # DEBUG: print(f"DEBUG: domain='{entry['domain']}' is blacklisted - SKIPPED!")
continue
elif fba.is_instance_registered(entry["domain"]):
- print(f"DEBUG: domain='{entry['domain']}' is already registered - SKIPPED!")
+ # DEBUG: print(f"DEBUG: domain='{entry['domain']}' is already registered - SKIPPED!")
continue
- print(f"DEBUG: Adding domain='{entry['domain']}' ...")
+ # DEBUG: print(f"DEBUG: Adding domain='{entry['domain']}' ...")
domains.append(entry["domain"])
except BaseException as e:
print(f"ERROR: Cannot fetch graphql,exception[{type(e)}]:'{str(e)}'")
sys.exit(255)
- print(f"DEBUG: domains()={len(domains)}")
+ # DEBUG: print(f"DEBUG: domains()={len(domains)}")
if len(domains) > 0:
boot.acquire_lock()
print(f"INFO: Fetching instances from domain='{domain}' ...")
fba.fetch_instances(domain, None, None, sys.argv[0])
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def fetch_blocks(args: argparse.Namespace):
- print(f"DEBUG: args[]={type(args)} - CALLED!")
+ # DEBUG: print(f"DEBUG: args[]={type(args)} - CALLED!")
if args.domain != None and args.domain != "":
if not validators.domain(args.domain):
print(f"WARNING: domain='{args.domain}' is not valid.")
rows = fba.cursor.fetchall()
print(f"INFO: Checking {len(rows)} entries ...")
for blocker, software, origin, nodeinfo_url in rows:
- print("DEBUG: BEFORE blocker,software,origin,nodeinfo_url:", blocker, software, origin, nodeinfo_url)
+ # DEBUG: print("DEBUG: BEFORE blocker,software,origin,nodeinfo_url:", blocker, software, origin, nodeinfo_url)
blockdict = []
blocker = fba.tidyup_domain(blocker)
- print("DEBUG: AFTER blocker,software:", blocker, software)
+ # DEBUG: print("DEBUG: AFTER blocker,software:", blocker, software)
if blocker == "":
print("WARNING: blocker is now empty!")
print(f"WARNING: blocker='{blocker}' is blacklisted now!")
continue
- print(f"DEBUG: blocker='{blocker}'")
+ # DEBUG: print(f"DEBUG: blocker='{blocker}'")
fba.update_last_blocked(blocker)
if software == "pleroma":
print(f"WARNING: json()={len(json['metadata'])} does not have key 'federation', blocker='{blocker}'")
continue
- print("DEBUG: Updating nodeinfo:", blocker)
+ # DEBUG: print("DEBUG: Updating nodeinfo:", blocker)
fba.update_last_nodeinfo(blocker)
federation = json["metadata"]["federation"]
if "enabled" in federation:
- print("DEBUG: Instance has no block list to analyze:", blocker)
+ # DEBUG: print("DEBUG: Instance has no block list to analyze:", blocker)
continue
if "mrf_simple" in federation:
{**federation["mrf_simple"],
**{"quarantined_instances": federation["quarantined_instances"]}}
).items():
- print("DEBUG: block_level, blocks():", block_level, len(blocks))
+ # DEBUG: print("DEBUG: block_level, blocks():", block_level, len(blocks))
block_level = fba.tidyup_domain(block_level)
- print("DEBUG: BEFORE block_level:", block_level)
+ # DEBUG: print("DEBUG: BEFORE block_level:", block_level)
if block_level == "":
print("WARNING: block_level is now empty!")
continue
- print(f"DEBUG: Checking {len(blocks)} entries from blocker='{blocker}',software='{software}',block_level='{block_level}' ...")
+ # DEBUG: print(f"DEBUG: Checking {len(blocks)} entries from blocker='{blocker}',software='{software}',block_level='{block_level}' ...")
for blocked in blocks:
- print("DEBUG: BEFORE blocked:", blocked)
+ # DEBUG: print("DEBUG: BEFORE blocked:", blocked)
blocked = fba.tidyup_domain(blocked)
- print("DEBUG: AFTER blocked:", blocked)
+ # DEBUG: print("DEBUG: AFTER blocked:", blocked)
if blocked == "":
print("WARNING: blocked is empty after fba.tidyup_domain():", blocker, block_level)
continue
elif fba.is_blacklisted(blocked):
- print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
+ # DEBUG: print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
continue
elif blocked.count("*") > 1:
# -ACK!-oma also started obscuring domains without hash
"SELECT domain, nodeinfo_url FROM instances WHERE domain LIKE ? ORDER BY rowid LIMIT 1", [blocked.replace("*", "_")]
)
searchres = fba.cursor.fetchone()
- print("DEBUG: searchres[]:", type(searchres))
+ # DEBUG: print("DEBUG: searchres[]:", type(searchres))
if searchres == None:
print(f"WARNING: Cannot deobsfucate blocked='{blocked}' - SKIPPED!")
blocked = searchres[0]
nodeinfo_url = searchres[1]
- print("DEBUG: Looked up domain:", blocked)
+ # DEBUG: print("DEBUG: Looked up domain:", blocked)
elif not validators.domain(blocked):
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
- print("DEBUG: Looking up instance by domain:", blocked)
+ # DEBUG: print("DEBUG: Looking up instance by domain:", blocked)
if not validators.domain(blocked):
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
elif not fba.is_instance_registered(blocked):
- print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., blocker='{blocker}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
+ # DEBUG: print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., blocker='{blocker}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
fba.add_instance(blocked, blocker, sys.argv[0], nodeinfo_url)
if not fba.is_instance_blocked(blocker, blocked, block_level):
- print("DEBUG: Blocking:", blocker, blocked, block_level)
+ # DEBUG: print("DEBUG: Blocking:", blocker, blocked, block_level)
fba.block_instance(blocker, blocked, "unknown", block_level)
if block_level == "reject":
- print("DEBUG: Adding to blockdict:", blocked)
+ # DEBUG: print("DEBUG: Adding to blockdict:", blocked)
blockdict.append(
{
"blocked": blocked,
"reason" : None
})
else:
- print(f"DEBUG: Updating block last seen for blocker='{blocker}',blocked='{blocked}' ...")
+ # DEBUG: print(f"DEBUG: Updating block last seen for blocker='{blocker}',blocked='{blocked}' ...")
fba.update_last_seen(blocker, blocked, block_level)
- print("DEBUG: Committing changes ...")
+ # DEBUG: print("DEBUG: Committing changes ...")
fba.connection.commit()
# Reasons
if "mrf_simple_info" in federation:
- print("DEBUG: Found mrf_simple_info:", blocker)
+ # DEBUG: print("DEBUG: Found mrf_simple_info:", blocker)
for block_level, info in (
{**federation["mrf_simple_info"],
**(federation["quarantined_instances_info"]
if "quarantined_instances_info" in federation
else {})}
).items():
- print("DEBUG: block_level, info.items():", block_level, len(info.items()))
+ # DEBUG: print("DEBUG: block_level, info.items():", block_level, len(info.items()))
block_level = fba.tidyup_domain(block_level)
- print("DEBUG: BEFORE block_level:", block_level)
+ # DEBUG: print("DEBUG: BEFORE block_level:", block_level)
if block_level == "":
print("WARNING: block_level is now empty!")
continue
- print(f"DEBUG: Checking {len(info.items())} entries from blocker='{blocker}',software='{software}',block_level='{block_level}' ...")
+ # DEBUG: print(f"DEBUG: Checking {len(info.items())} entries from blocker='{blocker}',software='{software}',block_level='{block_level}' ...")
for blocked, reason in info.items():
- print("DEBUG: BEFORE blocked:", blocked)
+ # DEBUG: print("DEBUG: BEFORE blocked:", blocked)
blocked = fba.tidyup_domain(blocked)
- print("DEBUG: AFTER blocked:", blocked)
+ # DEBUG: print("DEBUG: AFTER blocked:", blocked)
if blocked == "":
print("WARNING: blocked is empty after fba.tidyup_domain():", blocker, block_level)
continue
elif fba.is_blacklisted(blocked):
- print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
+ # DEBUG: print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
continue
elif blocked.count("*") > 1:
# same domain guess as above, but for reasons field
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
- print("DEBUG: Looking up instance by domain:", blocked)
+ # DEBUG: print("DEBUG: Looking up instance by domain:", blocked)
if not validators.domain(blocked):
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
elif not fba.is_instance_registered(blocked):
- print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., blocker='{blocker}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
+ # DEBUG: print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., blocker='{blocker}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
fba.add_instance(blocked, blocker, sys.argv[0], nodeinfo_url)
- print("DEBUG: Updating block reason:", blocker, blocked, reason["reason"])
+ # DEBUG: print("DEBUG: Updating block reason:", blocker, blocked, reason["reason"])
fba.update_block_reason(reason["reason"], blocker, blocked, block_level)
for entry in blockdict:
if entry["blocked"] == blocked:
- print("DEBUG: Updating entry reason:", blocked)
+ # DEBUG: print("DEBUG: Updating entry reason:", blocked)
entry["reason"] = reason["reason"]
fba.connection.commit()
}
# handling CSRF, I've saw at least one server requiring it to access the endpoint
- print("DEBUG: Fetching meta:", blocker)
+ # DEBUG: print("DEBUG: Fetching meta:", blocker)
meta = bs4.BeautifulSoup(
fba.get_response(blocker, "/", fba.headers, (config.get("connection_timeout"), config.get("read_timeout"))).text,
"html.parser",
)
try:
csrf = meta.find("meta", attrs={"name": "csrf-token"})["content"]
- print("DEBUG: Adding CSRF token:", blocker, csrf)
+ # DEBUG: print("DEBUG: Adding CSRF token:", blocker, csrf)
reqheaders = {**fba.api_headers, **{"X-CSRF-Token": csrf}}
except BaseException as e:
- print("DEBUG: No CSRF token found, using normal headers:", blocker, e)
+ # DEBUG: print("DEBUG: No CSRF token found, using normal headers:", blocker, e)
reqheaders = fba.api_headers
- print("DEBUG: Querying API domain_blocks:", blocker)
+ # DEBUG: print("DEBUG: Querying API domain_blocks:", blocker)
blocks = fba.get_response(blocker, "/api/v1/instance/domain_blocks", reqheaders, (config.get("connection_timeout"), config.get("read_timeout"))).json()
print(f"INFO: Checking {len(blocks)} entries from blocker='{blocker}',software='{software}' ...")
'reason': block['comment']
}
- print("DEBUG: severity,domain,hash,comment:", block['severity'], block['domain'], block['digest'], block['comment'])
+ # DEBUG: print("DEBUG: severity,domain,hash,comment:", block['severity'], block['domain'], block['digest'], block['comment'])
if block['severity'] == 'suspend':
- print(f"DEBUG: Adding entry='{entry}' with severity='{block['severity']}' ...")
+ # DEBUG: print(f"DEBUG: Adding entry='{entry}' with severity='{block['severity']}' ...")
json['reject'].append(entry)
elif block['severity'] == 'silence':
- print(f"DEBUG: Adding entry='{entry}' with severity='{block['severity']}' ...")
+ # DEBUG: print(f"DEBUG: Adding entry='{entry}' with severity='{block['severity']}' ...")
json['followers_only'].append(entry)
elif block['severity'] == 'reject_media':
- print(f"DEBUG: Adding entry='{entry}' with severity='{block['severity']}' ...")
+ # DEBUG: print(f"DEBUG: Adding entry='{entry}' with severity='{block['severity']}' ...")
json['media_removal'].append(entry)
elif block['severity'] == 'reject_reports':
- print(f"DEBUG: Adding entry='{entry}' with severity='{block['severity']}' ...")
+ # DEBUG: print(f"DEBUG: Adding entry='{entry}' with severity='{block['severity']}' ...")
json['report_removal'].append(entry)
else:
print("WARNING: Unknown severity:", block['severity'], block['domain'])
except BaseException as e:
- print(f"DEBUG: Failed, trying mastodon-specific fetches: blocker='{blocker}',exception[{type(e)}]={str(e)}")
+ # DEBUG: print(f"DEBUG: Failed, trying mastodon-specific fetches: blocker='{blocker}',exception[{type(e)}]={str(e)}")
json = fba.get_mastodon_blocks(blocker)
print(f"INFO: Checking {len(json.items())} entries from blocker='{blocker}',software='{software}' ...")
for block_level, blocks in json.items():
- print("DEBUG: blocker,block_level,blocks():", blocker, block_level, len(blocks))
+ # DEBUG: print("DEBUG: blocker,block_level,blocks():", blocker, block_level, len(blocks))
block_level = fba.tidyup_domain(block_level)
- print("DEBUG: AFTER-block_level:", block_level)
+ # DEBUG: print("DEBUG: AFTER-block_level:", block_level)
if block_level == "":
print("WARNING: block_level is empty, blocker:", blocker)
continue
- print(f"DEBUG: Checking {len(blocks)} entries from blocker='{blocker}',software='{software}',block_level='{block_level}' ...")
+ # DEBUG: print(f"DEBUG: Checking {len(blocks)} entries from blocker='{blocker}',software='{software}',block_level='{block_level}' ...")
for block in blocks:
blocked, blocked_hash, reason = block.values()
- print("DEBUG: blocked,hash,reason:", blocked, blocked_hash, reason)
+ # DEBUG: print("DEBUG: blocked,hash,reason:", blocked, blocked_hash, reason)
blocked = fba.tidyup_domain(blocked)
- print("DEBUG: AFTER-blocked:", blocked)
+ # DEBUG: print("DEBUG: AFTER-blocked:", blocked)
if blocked == "":
print("WARNING: blocked is empty:", blocker)
continue
elif fba.is_blacklisted(blocked):
- print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
+ # DEBUG: print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
continue
elif blocked.count("*") > 0:
# Doing the hash search for instance names as well to tidy up DB
print(f"WARNING: Cannot deobsfucate blocked='{blocked}',blocked_hash='{blocked_hash}' - SKIPPED!")
continue
- print("DEBUG: Updating domain: ", searchres[0])
+ # DEBUG: print("DEBUG: Updating domain: ", searchres[0])
blocked = searchres[0]
origin = searchres[1]
nodeinfo_url = searchres[2]
- print("DEBUG: Looking up instance by domain:", blocked)
+ # DEBUG: print("DEBUG: Looking up instance by domain:", blocked)
if not validators.domain(blocked):
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
elif not fba.is_instance_registered(blocked):
- print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., blocker='{blocker}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
+ # DEBUG: print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., blocker='{blocker}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
fba.add_instance(blocked, blocker, sys.argv[0], nodeinfo_url)
elif not validators.domain(blocked):
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
- print("DEBUG: Looking up instance by domain:", blocked)
+ # DEBUG: print("DEBUG: Looking up instance by domain:", blocked)
if not validators.domain(blocked):
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
elif not fba.is_instance_registered(blocked):
- print("DEBUG: Hash wasn't found, adding:", blocked, blocker)
+ # DEBUG: print("DEBUG: Hash wasn't found, adding:", blocked, blocker)
fba.add_instance(blocked, blocker, sys.argv[0], nodeinfo_url)
blocking = blocked if blocked.count("*") <= 1 else blocked_hash
- print(f"DEBUG: blocking='{blocking}',blocked='{blocked}',blocked_hash='{blocked_hash}'")
+ # DEBUG: print(f"DEBUG: blocking='{blocking}',blocked='{blocked}',blocked_hash='{blocked_hash}'")
if not fba.is_instance_blocked(blocker, blocked, block_level):
- print("DEBUG: Blocking:", blocker, blocked, block_level)
+ # DEBUG: print("DEBUG: Blocking:", blocker, blocked, block_level)
fba.block_instance(blocker, blocking, reason, block_level)
if block_level == "reject":
"reason" : reason
})
else:
- print(f"DEBUG: Updating block last seen and reason for blocker='{blocker}',blocking='{blocking}' ...")
+ # DEBUG: print(f"DEBUG: Updating block last seen and reason for blocker='{blocker}',blocking='{blocking}' ...")
fba.update_last_seen(blocker, blocking, block_level)
fba.update_block_reason(reason, blocker, blocking, block_level)
- print("DEBUG: Committing changes ...")
+ # DEBUG: print("DEBUG: Committing changes ...")
fba.connection.commit()
except Exception as e:
print(f"ERROR: blocker='{blocker}',software='{software}',exception[{type(e)}]:'{str(e)}'")
print(f"INFO: Checking {len(json.items())} entries from blocker='{blocker}',software='{software}' ...")
for block_level, blocks in json.items():
- print("DEBUG: blocker,block_level,blocks():", blocker, block_level, len(blocks))
+ # DEBUG: print("DEBUG: blocker,block_level,blocks():", blocker, block_level, len(blocks))
block_level = fba.tidyup_domain(block_level)
- print("DEBUG: AFTER-block_level:", block_level)
+ # DEBUG: print("DEBUG: AFTER-block_level:", block_level)
if block_level == "":
print("WARNING: block_level is empty, blocker:", blocker)
continue
- print(f"DEBUG: Checking {len(blocks)} entries from blocker='{blocker}',software='{software}',block_level='{block_level}' ...")
+ # DEBUG: print(f"DEBUG: Checking {len(blocks)} entries from blocker='{blocker}',software='{software}',block_level='{block_level}' ...")
for block in blocks:
blocked, reason = block.values()
- print("DEBUG: BEFORE blocked:", blocked)
+ # DEBUG: print("DEBUG: BEFORE blocked:", blocked)
blocked = fba.tidyup_domain(blocked)
- print("DEBUG: AFTER blocked:", blocked)
+ # DEBUG: print("DEBUG: AFTER blocked:", blocked)
if blocked == "":
print("WARNING: blocked is empty:", blocker)
continue
elif fba.is_blacklisted(blocked):
- print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
+ # DEBUG: print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
continue
elif blocked.count("*") > 0:
# Some friendica servers also obscure domains without hash
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
- print("DEBUG: Looking up instance by domain:", blocked)
+ # DEBUG: print("DEBUG: Looking up instance by domain:", blocked)
if not validators.domain(blocked):
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
elif not fba.is_instance_registered(blocked):
- print("DEBUG: Hash wasn't found, adding:", blocked, blocker)
+ # DEBUG: print("DEBUG: Hash wasn't found, adding:", blocked, blocker)
fba.add_instance(blocked, blocker, sys.argv[0], nodeinfo_url)
if not fba.is_instance_blocked(blocker, blocked, block_level):
"reason" : reason
})
else:
- print(f"DEBUG: Updating block last seen and reason for blocker='{blocker}',blocked='{blocked}' ...")
+ # DEBUG: print(f"DEBUG: Updating block last seen and reason for blocker='{blocker}',blocked='{blocked}' ...")
fba.update_last_seen(blocker, blocked, block_level)
fba.update_block_reason(reason, blocker, blocked, block_level)
- print("DEBUG: Committing changes ...")
+ # DEBUG: print("DEBUG: Committing changes ...")
fba.connection.commit()
except Exception as e:
print(f"ERROR: blocker='{blocker}',software='{software}',exception[{type(e)}]:'{str(e)}'")
print(f"INFO: Checking {len(federation)} entries from blocker='{blocker}',software='{software}' ...")
for peer in federation:
blocked = peer["domain"].lower()
- print("DEBUG: BEFORE blocked:", blocked)
+ # DEBUG: print("DEBUG: BEFORE blocked:", blocked)
blocked = fba.tidyup_domain(blocked)
- print("DEBUG: AFTER blocked:", blocked)
+ # DEBUG: print("DEBUG: AFTER blocked:", blocked)
if blocked == "":
print("WARNING: blocked is empty:", blocker)
continue
elif fba.is_blacklisted(blocked):
- print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
+ # DEBUG: print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
continue
elif blocked.count("*") > 0:
# GTS does not have hashes for obscured domains, so we have to guess it
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
- print("DEBUG: Looking up instance by domain:", blocked)
+ # DEBUG: print("DEBUG: Looking up instance by domain:", blocked)
if not validators.domain(blocked):
print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
continue
elif not fba.is_instance_registered(blocked):
- print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., blocker='{blocker}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
+ # DEBUG: print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., blocker='{blocker}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
fba.add_instance(blocked, blocker, sys.argv[0], nodeinfo_url)
if not fba.is_instance_blocked(blocker, blocked, "reject"):
- print(f"DEBUG: blocker='{blocker}' is blocking '{blocked}' for unknown reason at this point")
+ # DEBUG: print(f"DEBUG: blocker='{blocker}' is blocking '{blocked}' for unknown reason at this point")
fba.block_instance(blocker, blocked, "unknown", "reject")
blockdict.append({
"reason" : None
})
else:
- print(f"DEBUG: Updating block last seen for blocker='{blocker}',blocked='{blocked}' ...")
+ # DEBUG: print(f"DEBUG: Updating block last seen for blocker='{blocker}',blocked='{blocked}' ...")
fba.update_last_seen(blocker, blocked, "reject")
if "public_comment" in peer:
- print("DEBUG: Updating block reason:", blocker, blocked, peer["public_comment"])
+ # DEBUG: print("DEBUG: Updating block reason:", blocker, blocked, peer["public_comment"])
fba.update_block_reason(peer["public_comment"], blocker, blocked, "reject")
for entry in blockdict:
if entry["blocked"] == blocked:
- print(f"DEBUG: Setting block reason for blocked='{blocked}':'{peer['public_comment']}'")
+ # DEBUG: print(f"DEBUG: Setting block reason for blocked='{blocked}':'{peer['public_comment']}'")
entry["reason"] = peer["public_comment"]
- print("DEBUG: Committing changes ...")
+ # DEBUG: print("DEBUG: Committing changes ...")
fba.connection.commit()
except Exception as e:
print(f"ERROR: blocker='{blocker}',software='{software}',exception[{type(e)}]:'{str(e)}'")
blockdict = []
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def fetch_cs(args: argparse.Namespace):
- print(f"DEBUG: args[]={type(args)} - CALLED!")
+ # DEBUG: print(f"DEBUG: args[]={type(args)} - CALLED!")
domains = {
"silenced": list(),
"blocked": list(),
fba.get_response("meta.chaos.social", "/federation", fba.headers, (config.get("connection_timeout"), config.get("read_timeout"))).text,
"html.parser",
)
- print(f"DEBUG: doc()={len(doc)}[]={type(doc)}")
+ # DEBUG: print(f"DEBUG: doc()={len(doc)}[]={type(doc)}")
silenced = doc.find("h2", {"id": "silenced-instances"}).findNext("table")
- print(f"DEBUG: silenced[]={type(silenced)}")
+ # DEBUG: print(f"DEBUG: silenced[]={type(silenced)}")
domains["silenced"] = domains["silenced"] + find_domains(silenced)
blocked = doc.find("h2", {"id": "blocked-instances"}).findNext("table")
- print(f"DEBUG: blocked[]={type(blocked)}")
+ # DEBUG: print(f"DEBUG: blocked[]={type(blocked)}")
domains["blocked"] = domains["blocked"] + find_domains(blocked)
except BaseException as e:
print(f"ERROR: Cannot fetch from meta.chaos.social,exception[{type(e)}]:'{str(e)}'")
sys.exit(255)
- print(f"DEBUG: domains()={len(domains)}")
+ # DEBUG: print(f"DEBUG: domains()={len(domains)}")
if len(domains) > 0:
boot.acquire_lock()
print(f"INFO: Adding {len(domains)} new instances ...")
for block_level in domains:
- print(f"DEBUG: block_level='{block_level}'")
+ # DEBUG: print(f"DEBUG: block_level='{block_level}'")
for row in domains[block_level]:
- print(f"DEBUG: row='{row}'")
+ # DEBUG: print(f"DEBUG: row='{row}'")
if not fba.is_instance_registered(row["domain"]):
print(f"INFO: Fetching instances from domain='{row['domain']}' ...")
fba.fetch_instances(row["domain"], None, None, sys.argv[0])
if not fba.is_instance_blocked('chaos.social', row["domain"], block_level):
- print(f"DEBUG: domain='{row['domain']}',block_level='{block_level}' blocked by chaos.social, adding ...")
+ # DEBUG: print(f"DEBUG: domain='{row['domain']}',block_level='{block_level}' blocked by chaos.social, adding ...")
fba.block_instance('chaos.social', row["domain"], row["reason"], block_level)
- print("DEBUG: Committing changes ...")
+ # DEBUG: print("DEBUG: Committing changes ...")
fba.connection.commit()
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def fetch_fba_rss(args: argparse.Namespace):
- print(f"DEBUG: args[]={type(args)} - CALLED!")
+ # DEBUG: print(f"DEBUG: args[]={type(args)} - CALLED!")
domains = list()
try:
print(f"INFO: Fetch FBA-specific RSS args.feed='{args.feed}' ...")
response = fba.get_url(args.feed, fba.headers, config.get("connection_timeout"), config.get("read_timeout"))
- print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',response.text()={len(response.text)}")
+ # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',response.text()={len(response.text)}")
if response.ok and response.status_code < 300 and len(response.text) > 0:
- print(f"DEBUG: Parsing RSS feed ...")
+ # DEBUG: print(f"DEBUG: Parsing RSS feed ...")
rss = atoma.parse_rss_bytes(response.content)
- print(f"DEBUG: rss[]={type(rss)}")
+ # DEBUG: print(f"DEBUG: rss[]={type(rss)}")
for item in rss.items:
- print(f"DEBUG: item={item}")
+ # DEBUG: print(f"DEBUG: item={item}")
domain = item.link.split("=")[1]
if fba.is_blacklisted(domain):
- print(f"DEBUG: domain='{domain}' is blacklisted - SKIPPED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}' is blacklisted - SKIPPED!")
continue
elif domain in domains:
- print(f"DEBUG: domain='{domain}' is already added - SKIPPED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}' is already added - SKIPPED!")
continue
elif fba.is_instance_registered(domain):
- print(f"DEBUG: domain='{domain}' is already registered - SKIPPED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}' is already registered - SKIPPED!")
continue
- print(f"DEBUG: domain='{domain}'")
+ # DEBUG: print(f"DEBUG: domain='{domain}'")
domains.append(domain)
except BaseException as e:
print(f"ERROR: Cannot fetch feed='{feed}',exception[{type(e)}]:'{str(e)}'")
sys.exit(255)
- print(f"DEBUG: domains()={len(domains)}")
+ # DEBUG: print(f"DEBUG: domains()={len(domains)}")
if len(domains) > 0:
boot.acquire_lock()
print(f"INFO: Fetching instances from domain='{domain}' ...")
fba.fetch_instances(domain, None, None, sys.argv[0])
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def fetch_fbabot_atom(args: argparse.Namespace):
- print(f"DEBUG: args[]={type(args)} - CALLED!")
+ # DEBUG: print(f"DEBUG: args[]={type(args)} - CALLED!")
feed = "https://ryona.agency/users/fba/feed.atom"
domains = list()
print(f"INFO: Fetching ATOM feed='{feed}' from FBA bot account ...")
response = fba.get_url(feed, fba.headers, (config.get("connection_timeout"), config.get("read_timeout")))
- print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',response.text()={len(response.text)}")
+ # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',response.text()={len(response.text)}")
if response.ok and response.status_code < 300 and len(response.text) > 0:
- print(f"DEBUG: Parsing ATOM feed ...")
+ # DEBUG: print(f"DEBUG: Parsing ATOM feed ...")
atom = atoma.parse_atom_bytes(response.content)
- print(f"DEBUG: atom[]={type(atom)}")
+ # DEBUG: print(f"DEBUG: atom[]={type(atom)}")
for entry in atom.entries:
- print(f"DEBUG: entry[]={type(entry)}")
+ # DEBUG: print(f"DEBUG: entry[]={type(entry)}")
doc = bs4.BeautifulSoup(entry.content.value, "html.parser")
- print(f"DEBUG: doc[]={type(doc)}")
+ # DEBUG: print(f"DEBUG: doc[]={type(doc)}")
for element in doc.findAll("a"):
- print(f"DEBUG: element[{type(element)}]={element}")
+ # DEBUG: print(f"DEBUG: element[{type(element)}]={element}")
domain = fba.tidyup_domain(element["href"])
- print(f"DEBUG: domain='{domain}'")
+ # DEBUG: print(f"DEBUG: domain='{domain}'")
if fba.is_blacklisted(domain):
- print(f"DEBUG: domain='{domain}' is blacklisted - SKIPPED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}' is blacklisted - SKIPPED!")
continue
elif domain in domains:
- print(f"DEBUG: domain='{domain}' is already added - SKIPPED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}' is already added - SKIPPED!")
continue
elif fba.is_instance_registered(domain):
- print(f"DEBUG: domain='{domain}' is already registered - SKIPPED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}' is already registered - SKIPPED!")
continue
- print(f"DEBUG: domain='{domain}'")
+ # DEBUG: print(f"DEBUG: domain='{domain}'")
domains.append(domain)
except BaseException as e:
print(f"ERROR: Cannot fetch feed='{feed}',exception[{type(e)}]:'{str(e)}'")
sys.exit(255)
- print(f"DEBUG: domains()={len(domains)}")
+ # DEBUG: print(f"DEBUG: domains()={len(domains)}")
if len(domains) > 0:
boot.acquire_lock()
print(f"INFO: Fetching instances from domain='{domain}' ...")
fba.fetch_instances(domain, None, None, sys.argv[0])
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def fetch_instances(args: argparse.Namespace):
- print(f"DEBUG: args[]={type(args)} - CALLED!")
+ # DEBUG: print(f"DEBUG: args[]={type(args)} - CALLED!")
boot.acquire_lock()
# Initial fetch
fba.fetch_instances(args.domain, None, None, sys.argv[0])
if args.single:
- print(f"DEBUG: Not fetching more instances - EXIT!")
+ # DEBUG: print(f"DEBUG: Not fetching more instances - EXIT!")
return
# Loop through some instances
rows = fba.cursor.fetchall()
print(f"INFO: Checking {len(rows)} entries ...")
for row in rows:
- print("DEBUG: domain:", row[0])
+ # DEBUG: print("DEBUG: domain:", row[0])
if fba.is_blacklisted(row[0]):
print("WARNING: domain is blacklisted:", row[0])
continue
print(f"INFO: Fetching instances for instance '{row[0]}' ('{row[2]}') of origin='{row[1]}',nodeinfo_url='{row[3]}'")
fba.fetch_instances(row[0], row[1], row[2], sys.argv[0], row[3])
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
##### Other functions #####
def is_primitive(var: any) -> bool:
- # NOISY-DEBUG: print(f"DEBUG: var[]='{type(var)}' - CALLED!")
+ # NOISY-DEBUG: # DEBUG: print(f"DEBUG: var[]='{type(var)}' - CALLED!")
return type(var) in {int, str, float, bool} or var == None
def fetch_instances(domain: str, origin: str, software: str, script: str, path: str = None):
- print(f"DEBUG: domain={domain},origin={origin},software={software},path={path} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain={domain},origin={origin},software={software},path={path} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
if not is_instance_registered(domain):
- print("DEBUG: Adding new domain:", domain, origin)
+ # DEBUG: print("DEBUG: Adding new domain:", domain, origin)
add_instance(domain, origin, script, path)
- print("DEBUG: Fetching instances for domain:", domain, software)
+ # DEBUG: print("DEBUG: Fetching instances for domain:", domain, software)
peerlist = get_peers(domain, software)
if (peerlist is None):
print("ERROR: Cannot fetch peers:", domain)
return
elif instances.has_pending_instance_data(domain):
- print(f"DEBUG: domain='{domain}' has pending nodeinfo data, flushing ...")
+ # DEBUG: print(f"DEBUG: domain='{domain}' has pending nodeinfo data, flushing ...")
instances.update_instance_data(domain)
print(f"INFO: Checking {len(peerlist)} instances from {domain} ...")
# Skip "None" types as tidup() cannot parse them
continue
- print(f"DEBUG: instance='{instance}' - BEFORE")
+ # DEBUG: print(f"DEBUG: instance='{instance}' - BEFORE")
instance = tidyup_domain(instance)
- print(f"DEBUG: instance='{instance}' - AFTER")
+ # DEBUG: print(f"DEBUG: instance='{instance}' - AFTER")
if instance == "":
print("WARNING: Empty instance after tidyup_domain(), domain:", domain)
print(f"WARNING: Bad instance='{instance}' from domain='{domain}',origin='{origin}',software='{software}'")
continue
elif is_blacklisted(instance):
- print("DEBUG: instance is blacklisted:", instance)
+ # DEBUG: print("DEBUG: instance is blacklisted:", instance)
continue
- print("DEBUG: Handling instance:", instance)
+ # DEBUG: print("DEBUG: Handling instance:", instance)
try:
if not is_instance_registered(instance):
- print("DEBUG: Adding new instance:", instance, domain)
+ # DEBUG: print("DEBUG: Adding new instance:", instance, domain)
add_instance(instance, domain, sys.argv[0])
except BaseException as e:
print(f"ERROR: instance='{instance}',exception[{type(e)}]:'{str(e)}'")
continue
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def add_peers(rows: dict) -> list:
- print(f"DEBUG: rows()={len(rows)} - CALLED!")
+ # DEBUG: print(f"DEBUG: rows()={len(rows)} - CALLED!")
peers = list()
for element in ["linked", "allowed", "blocked"]:
- print(f"DEBUG: Checking element='{element}'")
+ # DEBUG: print(f"DEBUG: Checking element='{element}'")
if element in rows and rows[element] != None:
- print(f"DEBUG: Adding {len(rows[element])} peer(s) to peers list ...")
+ # DEBUG: print(f"DEBUG: Adding {len(rows[element])} peer(s) to peers list ...")
for peer in rows[element]:
- print(f"DEBUG: peer='{peer}' - BEFORE!")
+ # DEBUG: print(f"DEBUG: peer='{peer}' - BEFORE!")
peer = tidyup_domain(peer)
- print(f"DEBUG: peer='{peer}' - AFTER!")
+ # DEBUG: print(f"DEBUG: peer='{peer}' - AFTER!")
if is_blacklisted(peer):
- print(f"DEBUG: peer='{peer}' is blacklisted, skipped!")
+ # DEBUG: print(f"DEBUG: peer='{peer}' is blacklisted, skipped!")
continue
- print(f"DEBUG: Adding peer='{peer}' ...")
+ # DEBUG: print(f"DEBUG: Adding peer='{peer}' ...")
peers.append(peer)
- print(f"DEBUG: peers()={len(peers)} - EXIT!")
+ # DEBUG: print(f"DEBUG: peers()={len(peers)} - EXIT!")
return peers
def remove_version(software: str) -> str:
- print(f"DEBUG: software='{software}' - CALLED!")
+ # DEBUG: print(f"DEBUG: software='{software}' - CALLED!")
if not "." in software and " " not in software:
print(f"WARNING: software='{software}' does not contain a version number.")
return software
elif " - " in software:
temp = software.split(" - ")[0]
- print(f"DEBUG: software='{software}'")
+ # DEBUG: print(f"DEBUG: software='{software}'")
version = None
if " " in software:
version = temp.split(" ")[-1]
elif "-" in software:
version = temp.split("-")[-1]
else:
- print(f"DEBUG: Was not able to find common seperator, returning untouched software='{software}'")
+ # DEBUG: print(f"DEBUG: Was not able to find common seperator, returning untouched software='{software}'")
return software
matches = None
match = None
- print(f"DEBUG: Checking {len(patterns)} patterns ...")
+ # DEBUG: print(f"DEBUG: Checking {len(patterns)} patterns ...")
for pattern in patterns:
# Run match()
match = pattern.match(version)
- print(f"DEBUG: match[]={type(match)}")
+ # DEBUG: print(f"DEBUG: match[]={type(match)}")
if type(match) is re.Match:
break
- print(f"DEBUG: version[{type(version)}]='{version}',match='{match}'")
+ # DEBUG: print(f"DEBUG: version[{type(version)}]='{version}',match='{match}'")
if type(match) is not re.Match:
print(f"WARNING: version='{version}' does not match regex, leaving software='{software}' untouched.")
return software
- print(f"DEBUG: Found valid version number: '{version}', removing it ...")
+ # DEBUG: print(f"DEBUG: Found valid version number: '{version}', removing it ...")
end = len(temp) - len(version) - 1
- print(f"DEBUG: end[{type(end)}]={end}")
+ # DEBUG: print(f"DEBUG: end[{type(end)}]={end}")
software = temp[0:end].strip()
if " version" in software:
- print(f"DEBUG: software='{software}' contains word ' version'")
+ # DEBUG: print(f"DEBUG: software='{software}' contains word ' version'")
software = strip_until(software, " version")
- print(f"DEBUG: software='{software}' - EXIT!")
+ # DEBUG: print(f"DEBUG: software='{software}' - EXIT!")
return software
def strip_powered_by(software: str) -> str:
- print(f"DEBUG: software='{software}' - CALLED!")
+ # DEBUG: print(f"DEBUG: software='{software}' - CALLED!")
if software == "":
print(f"ERROR: Bad method call, 'software' is empty")
raise Exception("Parameter 'software' is empty")
return software
start = software.find("powered by ")
- print(f"DEBUG: start[{type(start)}]='{start}'")
+ # DEBUG: print(f"DEBUG: start[{type(start)}]='{start}'")
software = software[start + 11:].strip()
- print(f"DEBUG: software='{software}'")
+ # DEBUG: print(f"DEBUG: software='{software}'")
software = strip_until(software, " - ")
- print(f"DEBUG: software='{software}' - EXIT!")
+ # DEBUG: print(f"DEBUG: software='{software}' - EXIT!")
return software
def strip_hosted_on(software: str) -> str:
- print(f"DEBUG: software='{software}' - CALLED!")
+ # DEBUG: print(f"DEBUG: software='{software}' - CALLED!")
if software == "":
print(f"ERROR: Bad method call, 'software' is empty")
raise Exception("Parameter 'software' is empty")
return software
end = software.find("hosted on ")
- print(f"DEBUG: end[{type(end)}]='{end}'")
+ # DEBUG: print(f"DEBUG: end[{type(end)}]='{end}'")
software = software[0, start].strip()
- print(f"DEBUG: software='{software}'")
+ # DEBUG: print(f"DEBUG: software='{software}'")
software = strip_until(software, " - ")
- print(f"DEBUG: software='{software}' - EXIT!")
+ # DEBUG: print(f"DEBUG: software='{software}' - EXIT!")
return software
def strip_until(software: str, until: str) -> str:
- print(f"DEBUG: software='{software}',until='{until}' - CALLED!")
+ # DEBUG: print(f"DEBUG: software='{software}',until='{until}' - CALLED!")
if software == "":
print(f"ERROR: Bad method call, 'software' is empty")
raise Exception("Parameter 'software' is empty")
# Next, strip until part
end = software.find(until)
- print(f"DEBUG: end[{type(end)}]='{end}'")
+ # DEBUG: print(f"DEBUG: end[{type(end)}]='{end}'")
if end > 0:
software = software[0:end].strip()
- print(f"DEBUG: software='{software}' - EXIT!")
+ # DEBUG: print(f"DEBUG: software='{software}' - EXIT!")
return software
def is_blacklisted(domain: str) -> bool:
except:
pass
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def get_hash(domain: str) -> str:
if type(domain) != str:
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print("DEBUG: Updating last_blocked for domain", domain)
+ # DEBUG: print("DEBUG: Updating last_blocked for domain", domain)
instances.set("last_blocked", domain, time.time())
# Running pending updated
- print(f"DEBUG: Invoking instances.update_instance_data({domain}) ...")
+ # DEBUG: print(f"DEBUG: Invoking instances.update_instance_data({domain}) ...")
instances.update_instance_data(domain)
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def log_error(domain: str, response: requests.models.Response):
- print("DEBUG: domain,response[]:", domain, type(response))
+ # DEBUG: print("DEBUG: domain,response[]:", domain, type(response))
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
try:
- print("DEBUG: BEFORE response[]:", type(response))
+ # DEBUG: print("DEBUG: BEFORE response[]:", type(response))
if isinstance(response, BaseException) or isinstance(response, json.decoder.JSONDecodeError):
response = str(response)
- print("DEBUG: AFTER response[]:", type(response))
+ # DEBUG: print("DEBUG: AFTER response[]:", type(response))
if type(response) is str:
cursor.execute("INSERT INTO error_log (domain, error_code, error_message, created) VALUES (?, 999, ?, ?)",[
domain,
])
# Cleanup old entries
- print(f"DEBUG: Purging old records (distance: {config.get('error_log_cleanup')})")
+ # DEBUG: print(f"DEBUG: Purging old records (distance: {config.get('error_log_cleanup')})")
cursor.execute("DELETE FROM error_log WHERE created < ?", [time.time() - config.get("error_log_cleanup")])
except BaseException as e:
print(f"ERROR: failed SQL query: domain='{domain}',exception[{type(e)}]:'{str(e)}'")
sys.exit(255)
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def update_last_error(domain: str, response: requests.models.Response):
- print("DEBUG: domain,response[]:", domain, type(response))
+ # DEBUG: print("DEBUG: domain,response[]:", domain, type(response))
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print("DEBUG: BEFORE response[]:", type(response))
+ # DEBUG: print("DEBUG: BEFORE response[]:", type(response))
if isinstance(response, BaseException) or isinstance(response, json.decoder.JSONDecodeError):
response = f"{type}:str(response)"
- print("DEBUG: AFTER response[]:", type(response))
+ # DEBUG: print("DEBUG: AFTER response[]:", type(response))
if type(response) is str:
- print(f"DEBUG: Setting last_error_details='{response}'");
+ # DEBUG: print(f"DEBUG: Setting last_error_details='{response}'");
instances.set("last_status_code" , domain, 999)
instances.set("last_error_details", domain, response)
else:
- print(f"DEBUG: Setting last_error_details='{response.reason}'");
+ # DEBUG: print(f"DEBUG: Setting last_error_details='{response.reason}'");
instances.set("last_status_code" , domain, response.status_code)
instances.set("last_error_details", domain, response.reason)
# Running pending updated
- print(f"DEBUG: Invoking instances.update_instance_data({domain}) ...")
+ # DEBUG: print(f"DEBUG: Invoking instances.update_instance_data({domain}) ...")
instances.update_instance_data(domain)
log_error(domain, response)
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def update_last_instance_fetch(domain: str):
- print(f"DEBUG: domain={domain} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain={domain} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print("DEBUG: Updating last_instance_fetch for domain:", domain)
+ # DEBUG: print("DEBUG: Updating last_instance_fetch for domain:", domain)
instances.set("last_instance_fetch", domain, time.time())
# Running pending updated
- print(f"DEBUG: Invoking instances.update_instance_data({domain}) ...")
+ # DEBUG: print(f"DEBUG: Invoking instances.update_instance_data({domain}) ...")
instances.update_instance_data(domain)
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def update_last_nodeinfo(domain: str):
- print(f"DEBUG: domain={domain} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain={domain} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print("DEBUG: Updating last_nodeinfo for domain:", domain)
+ # DEBUG: print("DEBUG: Updating last_nodeinfo for domain:", domain)
instances.set("last_nodeinfo", domain, time.time())
instances.set("last_updated" , domain, time.time())
# Running pending updated
- print(f"DEBUG: Invoking instances.update_instance_data({domain}) ...")
+ # DEBUG: print(f"DEBUG: Invoking instances.update_instance_data({domain}) ...")
instances.update_instance_data(domain)
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def get_peers(domain: str, software: str) -> list:
- print(f"DEBUG: domain({len(domain)})={domain},software={software} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain({len(domain)})={domain},software={software} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
peers = list()
if software == "misskey":
- print(f"DEBUG: domain='{domain}' is misskey, sending API POST request ...")
+ # DEBUG: print(f"DEBUG: domain='{domain}' is misskey, sending API POST request ...")
offset = 0
step = config.get("misskey_limit")
# instances page-by-page, since that troonware doesn't support
# sending them all at once
while True:
- print(f"DEBUG: Fetching offset='{offset}' from '{domain}' ...")
+ # DEBUG: print(f"DEBUG: Fetching offset='{offset}' from '{domain}' ...")
if offset == 0:
fetched = post_json_api(domain, "/api/federation/instances", json.dumps({
"sort" : "+pubAt",
"Origin": domain
})
- print(f"DEBUG: fetched()={len(fetched)}")
+ # DEBUG: print(f"DEBUG: fetched()={len(fetched)}")
if len(fetched) == 0:
- print("DEBUG: Returned zero bytes, exiting loop:", domain)
+ # DEBUG: print("DEBUG: Returned zero bytes, exiting loop:", domain)
break
elif len(fetched) != config.get("misskey_limit"):
- print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_limit')}'")
+ # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_limit')}'")
offset = offset + (config.get("misskey_limit") - len(fetched))
else:
- print("DEBUG: Raising offset by step:", step)
+ # DEBUG: print("DEBUG: Raising offset by step:", step)
offset = offset + step
# Check records
- print(f"DEBUG: fetched({len(fetched)})[]={type(fetched)}")
+ # DEBUG: print(f"DEBUG: fetched({len(fetched)})[]={type(fetched)}")
if isinstance(fetched, dict) and "error" in fetched and "message" in fetched["error"]:
print(f"WARNING: post_json_api() returned error: {fetched['error']['message']}")
update_last_error(domain, fetched["error"]["message"])
already = 0
for row in fetched:
- print(f"DEBUG: row():{len(row)}")
+ # DEBUG: print(f"DEBUG: row():{len(row)}")
if not "host" in row:
print(f"WARNING: row()={len(row)} does not contain element 'host': {row},domain='{domain}'")
continue
print(f"WARNING: row[host][]={type(row['host'])} is not 'str'")
continue
elif is_blacklisted(row["host"]):
- print(f"DEBUG: row[host]='{row['host']}' is blacklisted. domain='{domain}'")
+ # DEBUG: print(f"DEBUG: row[host]='{row['host']}' is blacklisted. domain='{domain}'")
continue
elif row["host"] in peers:
- print(f"DEBUG: Not adding row[host]='{row['host']}', already found.")
+ # DEBUG: print(f"DEBUG: Not adding row[host]='{row['host']}', already found.")
already = already + 1
continue
- print(f"DEBUG: Adding peer: '{row['host']}'")
+ # DEBUG: print(f"DEBUG: Adding peer: '{row['host']}'")
peers.append(row["host"])
if already == len(fetched):
print(f"WARNING: Host returned same set of '{already}' instances, aborting loop!")
break
- print(f"DEBUG: Adding '{len(peers)}' for domain='{domain}'")
+ # DEBUG: print(f"DEBUG: Adding '{len(peers)}' for domain='{domain}'")
instances.set("total_peers", domain, len(peers))
- print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
+ # DEBUG: print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
update_last_instance_fetch(domain)
- print("DEBUG: Returning peers[]:", type(peers))
+ # DEBUG: print("DEBUG: Returning peers[]:", type(peers))
return peers
elif software == "lemmy":
- print(f"DEBUG: domain='{domain}' is Lemmy, fetching JSON ...")
+ # DEBUG: print(f"DEBUG: domain='{domain}' is Lemmy, fetching JSON ...")
try:
response = get_response(domain, "/api/v3/site", api_headers, (config.get("connection_timeout"), config.get("read_timeout")))
data = json_from_response(response)
- print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',data[]='{type(data)}'")
+ # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',data[]='{type(data)}'")
if not response.ok or response.status_code >= 400:
print("WARNING: Could not reach any JSON API:", domain)
update_last_error(domain, response)
elif response.ok and isinstance(data, list):
- print(f"DEBUG: domain='{domain}' returned a list: '{data}'")
+ # DEBUG: print(f"DEBUG: domain='{domain}' returned a list: '{data}'")
sys.exit(255)
elif "federated_instances" in data:
- print(f"DEBUG: Found federated_instances for domain='{domain}'")
+ # DEBUG: print(f"DEBUG: Found federated_instances for domain='{domain}'")
peers = peers + add_peers(data["federated_instances"])
- print("DEBUG: Added instance(s) to peers")
+ # DEBUG: print("DEBUG: Added instance(s) to peers")
else:
print("WARNING: JSON response does not contain 'federated_instances':", domain)
update_last_error(domain, response)
except BaseException as e:
print(f"WARNING: Exception during fetching JSON: domain='{domain}',exception[{type(e)}]:'{str(e)}'")
- print(f"DEBUG: Adding '{len(peers)}' for domain='{domain}'")
+ # DEBUG: print(f"DEBUG: Adding '{len(peers)}' for domain='{domain}'")
instances.set("total_peers", domain, len(peers))
- print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
+ # DEBUG: print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
update_last_instance_fetch(domain)
- print("DEBUG: Returning peers[]:", type(peers))
+ # DEBUG: print("DEBUG: Returning peers[]:", type(peers))
return peers
elif software == "peertube":
- print(f"DEBUG: domain='{domain}' is a PeerTube, fetching JSON ...")
+ # DEBUG: print(f"DEBUG: domain='{domain}' is a PeerTube, fetching JSON ...")
start = 0
for mode in ["followers", "following"]:
- print(f"DEBUG: domain='{domain}',mode='{mode}'")
+ # DEBUG: print(f"DEBUG: domain='{domain}',mode='{mode}'")
while True:
try:
response = get_response(domain, "/api/v1/server/{mode}?start={start}&count=100", headers, (config.get("connection_timeout"), config.get("read_timeout")))
data = json_from_response(response)
- print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',data[]='{type(data)}'")
+ # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',data[]='{type(data)}'")
if response.ok and isinstance(data, dict):
- print("DEBUG: Success, data:", len(data))
+ # DEBUG: print("DEBUG: Success, data:", len(data))
if "data" in data:
- print(f"DEBUG: Found {len(data['data'])} record(s).")
+ # DEBUG: print(f"DEBUG: Found {len(data['data'])} record(s).")
for record in data["data"]:
- print(f"DEBUG: record()={len(record)}")
+ # DEBUG: print(f"DEBUG: record()={len(record)}")
if mode in record and "host" in record[mode]:
- print(f"DEBUG: Found host={record[mode]['host']}, adding ...")
+ # DEBUG: print(f"DEBUG: Found host={record[mode]['host']}, adding ...")
peers.append(record[mode]["host"])
else:
print(f"WARNING: record from '{domain}' has no '{mode}' or 'host' record: {record}")
if len(data["data"]) < 100:
- print("DEBUG: Reached end of JSON response:", domain)
+ # DEBUG: print("DEBUG: Reached end of JSON response:", domain)
break
# Continue with next row
except BaseException as e:
print(f"WARNING: Exception during fetching JSON: domain='{domain}',exception[{type(e)}]:'{str(e)}'")
- print(f"DEBUG: Adding '{len(peers)}' for domain='{domain}'")
+ # DEBUG: print(f"DEBUG: Adding '{len(peers)}' for domain='{domain}'")
instances.set("total_peers", domain, len(peers))
- print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
+ # DEBUG: print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
update_last_instance_fetch(domain)
- print("DEBUG: Returning peers[]:", type(peers))
+ # DEBUG: print("DEBUG: Returning peers[]:", type(peers))
return peers
- print(f"DEBUG: Fetching get_peers_url='{get_peers_url}' from '{domain}' ...")
+ # DEBUG: print(f"DEBUG: Fetching get_peers_url='{get_peers_url}' from '{domain}' ...")
try:
response = get_response(domain, get_peers_url, api_headers, (config.get("connection_timeout"), config.get("read_timeout")))
data = json_from_response(response)
- print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},data[]='{type(data)}'")
+ # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},data[]='{type(data)}'")
if not response.ok or response.status_code >= 400:
- print(f"DEBUG: Was not able to fetch '{get_peers_url}', trying alternative ...")
+ # DEBUG: print(f"DEBUG: Was not able to fetch '{get_peers_url}', trying alternative ...")
response = get_response(domain, "/api/v3/site", api_headers, (config.get("connection_timeout"), config.get("read_timeout")))
data = json_from_response(response)
- print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},data[]='{type(data)}'")
+ # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},data[]='{type(data)}'")
if not response.ok or response.status_code >= 400:
print("WARNING: Could not reach any JSON API:", domain)
update_last_error(domain, response)
elif response.ok and isinstance(data, list):
- print(f"DEBUG: domain='{domain}' returned a list: '{data}'")
+ # DEBUG: print(f"DEBUG: domain='{domain}' returned a list: '{data}'")
sys.exit(255)
elif "federated_instances" in data:
- print(f"DEBUG: Found federated_instances for domain='{domain}'")
+ # DEBUG: print(f"DEBUG: Found federated_instances for domain='{domain}'")
peers = peers + add_peers(data["federated_instances"])
- print("DEBUG: Added instance(s) to peers")
+ # DEBUG: print("DEBUG: Added instance(s) to peers")
else:
print("WARNING: JSON response does not contain 'federated_instances':", domain)
update_last_error(domain, response)
else:
- print("DEBUG: Querying API was successful:", domain, len(data))
+ # DEBUG: print("DEBUG: Querying API was successful:", domain, len(data))
peers = data
except BaseException as e:
print("WARNING: Some error during get():", domain, e)
update_last_error(domain, e)
- print(f"DEBUG: Adding '{len(peers)}' for domain='{domain}'")
+ # DEBUG: print(f"DEBUG: Adding '{len(peers)}' for domain='{domain}'")
instances.set("total_peers", domain, len(peers))
- print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
+ # DEBUG: print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
update_last_instance_fetch(domain)
- print("DEBUG: Returning peers[]:", type(peers))
+ # DEBUG: print("DEBUG: Returning peers[]:", type(peers))
return peers
def post_json_api(domain: str, path: str, parameter: str, extra_headers: dict = {}) -> dict:
elif type(parameter) != str:
raise ValueError(f"parameter[]={type(parameter)} is not 'str'")
- print("DEBUG: Sending POST to domain,path,parameter:", domain, path, parameter, extra_headers)
+ # DEBUG: print("DEBUG: Sending POST to domain,path,parameter:", domain, path, parameter, extra_headers)
data = {}
try:
response = reqto.post(
)
data = json_from_response(response)
- print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},data[]='{type(data)}'")
+ # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},data[]='{type(data)}'")
if not response.ok or response.status_code >= 400:
print(f"WARNING: Cannot query JSON API: domain='{domain}',path='{path}',parameter()={len(parameter)},response.status_code='{response.status_code}',data[]='{type(data)}'")
update_last_error(domain, response)
except BaseException as e:
print(f"WARNING: Some error during post(): domain='{domain}',path='{path}',parameter()={len(parameter)},exception[{type(e)}]:'{str(e)}'")
- print(f"DEBUG: Returning data({len(data)})=[]:{type(data)}")
+ # DEBUG: print(f"DEBUG: Returning data({len(data)})=[]:{type(data)}")
return data
def fetch_nodeinfo(domain: str, path: str = None) -> list:
- print(f"DEBUG: domain='{domain}',path={path} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}',path={path} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
elif type(path) != str and path != None:
raise ValueError(f"Parameter path[]={type(path)} is not 'str'")
- print(f"DEBUG: Fetching nodeinfo from domain='{domain}' ...")
+ # DEBUG: print(f"DEBUG: Fetching nodeinfo from domain='{domain}' ...")
nodeinfo = fetch_wellknown_nodeinfo(domain)
- print(f"DEBUG: nodeinfo({len(nodeinfo)})={nodeinfo}")
+ # DEBUG: print(f"DEBUG: nodeinfo({len(nodeinfo)})={nodeinfo}")
if len(nodeinfo) > 0:
- print("DEBUG: nodeinfo()={len(nodeinfo))} - EXIT!")
+ # DEBUG: print("DEBUG: nodeinfo()={len(nodeinfo))} - EXIT!")
return nodeinfo
request_paths = [
data = {}
for request in request_paths:
if path != None and path != "" and path != f"https://{domain}{path}":
- print(f"DEBUG: path='{path}' does not match request='{request}' - SKIPPED!")
+ # DEBUG: print(f"DEBUG: path='{path}' does not match request='{request}' - SKIPPED!")
continue
try:
- print(f"DEBUG: Fetching request='{request}' from domain='{domain}' ...")
+ # DEBUG: print(f"DEBUG: Fetching request='{request}' from domain='{domain}' ...")
response = get_response(domain, request, api_headers, (config.get("nodeinfo_connection_timeout"), config.get("nodeinfo_read_timeout")))
data = json_from_response(response)
- print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},data[]='{type(data)}'")
+ # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},data[]='{type(data)}'")
if response.ok and isinstance(data, dict):
- print("DEBUG: Success:", request)
+ # DEBUG: print("DEBUG: Success:", request)
instances.set("detection_mode", domain, "STATIC_CHECK")
instances.set("nodeinfo_url" , domain, request)
break
continue
except BaseException as e:
- print("DEBUG: Cannot fetch API request:", request)
+ # DEBUG: print("DEBUG: Cannot fetch API request:", request)
update_last_error(domain, e)
pass
- print(f"DEBUG: data()={len(data)} - EXIT!")
+ # DEBUG: print(f"DEBUG: data()={len(data)} - EXIT!")
return data
def fetch_wellknown_nodeinfo(domain: str) -> list:
- print(f"DEBUG: domain={domain} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain={domain} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print("DEBUG: Fetching .well-known info for domain:", domain)
+ # DEBUG: print("DEBUG: Fetching .well-known info for domain:", domain)
data = {}
try:
response = get_response(domain, "/.well-known/nodeinfo", api_headers, (config.get("nodeinfo_connection_timeout"), config.get("nodeinfo_read_timeout")))
data = json_from_response(response)
- print("DEBUG: domain,response.ok,data[]:", domain, response.ok, type(data))
+ # DEBUG: print("DEBUG: domain,response.ok,data[]:", domain, response.ok, type(data))
if response.ok and isinstance(data, dict):
nodeinfo = data
- print("DEBUG: Found entries:", len(nodeinfo), domain)
+ # DEBUG: print("DEBUG: Found entries:", len(nodeinfo), domain)
if "links" in nodeinfo:
- print("DEBUG: Found links in nodeinfo():", len(nodeinfo["links"]))
+ # DEBUG: print("DEBUG: Found links in nodeinfo():", len(nodeinfo["links"]))
for link in nodeinfo["links"]:
- print("DEBUG: rel,href:", link["rel"], link["href"])
+ # DEBUG: print("DEBUG: rel,href:", link["rel"], link["href"])
if link["rel"] in nodeinfo_identifier:
- print("DEBUG: Fetching nodeinfo from:", link["href"])
+ # DEBUG: print("DEBUG: Fetching nodeinfo from:", link["href"])
response = get_url(link["href"])
data = json_from_response(response)
- print("DEBUG: href,response.ok,response.status_code:", link["href"], response.ok, response.status_code)
+ # DEBUG: print("DEBUG: href,response.ok,response.status_code:", link["href"], response.ok, response.status_code)
if response.ok and isinstance(data, dict):
- print("DEBUG: Found JSON nodeinfo():", len(data))
+ # DEBUG: print("DEBUG: Found JSON nodeinfo():", len(data))
instances.set("detection_mode", domain, "AUTO_DISCOVERY")
instances.set("nodeinfo_url" , domain, link["href"])
break
update_last_error(domain, e)
pass
- print("DEBUG: Returning data[]:", type(data))
+ # DEBUG: print("DEBUG: Returning data[]:", type(data))
return data
def fetch_generator_from_path(domain: str, path: str = "/") -> str:
- print(f"DEBUG: domain({len(domain)})={domain},path={path} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain({len(domain)})={domain},path={path} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
elif path == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print(f"DEBUG: domain='{domain}',path='{path}' - CALLED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}',path='{path}' - CALLED!")
software = None
try:
- print(f"DEBUG: Fetching path='{path}' from '{domain}' ...")
+ # DEBUG: print(f"DEBUG: Fetching path='{path}' from '{domain}' ...")
response = get_response(domain, path, headers, (config.get("connection_timeout"), config.get("read_timeout")))
- print("DEBUG: domain,response.ok,response.status_code,response.text[]:", domain, response.ok, response.status_code, type(response.text))
+ # DEBUG: print("DEBUG: domain,response.ok,response.status_code,response.text[]:", domain, response.ok, response.status_code, type(response.text))
if response.ok and response.status_code < 300 and len(response.text) > 0:
- print("DEBUG: Search for <meta name='generator'>:", domain)
+ # DEBUG: print("DEBUG: Search for <meta name='generator'>:", domain)
doc = bs4.BeautifulSoup(response.text, "html.parser")
- print("DEBUG: doc[]:", type(doc))
+ # DEBUG: print("DEBUG: doc[]:", type(doc))
generator = doc.find("meta", {"name": "generator"})
site_name = doc.find("meta", {"property": "og:site_name"})
- print(f"DEBUG: generator='{generator}',site_name='{site_name}'")
+ # DEBUG: print(f"DEBUG: generator='{generator}',site_name='{site_name}'")
if isinstance(generator, bs4.element.Tag):
- print("DEBUG: Found generator meta tag:", domain)
+ # DEBUG: print("DEBUG: Found generator meta tag:", domain)
software = tidyup_domain(generator.get("content"))
print(f"INFO: domain='{domain}' is generated by '{software}'")
instances.set("detection_mode", domain, "GENERATOR")
remove_pending_error(domain)
elif isinstance(site_name, bs4.element.Tag):
- print("DEBUG: Found property=og:site_name:", domain)
+ # DEBUG: print("DEBUG: Found property=og:site_name:", domain)
sofware = tidyup_domain(site_name.get("content"))
print(f"INFO: domain='{domain}' has og:site_name='{software}'")
instances.set("detection_mode", domain, "SITE_NAME")
remove_pending_error(domain)
except BaseException as e:
- print(f"DEBUG: Cannot fetch / from '{domain}':", e)
+ # DEBUG: print(f"DEBUG: Cannot fetch / from '{domain}':", e)
update_last_error(domain, e)
pass
- print(f"DEBUG: software[]={type(software)}")
+ # DEBUG: print(f"DEBUG: software[]={type(software)}")
if type(software) is str and software == "":
- print(f"DEBUG: Corrected empty string to None for software of domain='{domain}'")
+ # DEBUG: print(f"DEBUG: Corrected empty string to None for software of domain='{domain}'")
software = None
elif type(software) is str and ("." in software or " " in software):
- print(f"DEBUG: software='{software}' may contain a version number, domain='{domain}', removing it ...")
+ # DEBUG: print(f"DEBUG: software='{software}' may contain a version number, domain='{domain}', removing it ...")
software = remove_version(software)
- print(f"DEBUG: software[]={type(software)}")
+ # DEBUG: print(f"DEBUG: software[]={type(software)}")
if type(software) is str and " powered by " in software:
- print(f"DEBUG: software='{software}' has 'powered by' in it")
+ # DEBUG: print(f"DEBUG: software='{software}' has 'powered by' in it")
software = remove_version(strip_powered_by(software))
elif type(software) is str and " hosted on " in software:
- print(f"DEBUG: software='{software}' has 'hosted on' in it")
+ # DEBUG: print(f"DEBUG: software='{software}' has 'hosted on' in it")
software = remove_version(strip_hosted_on(software))
elif type(software) is str and " by " in software:
- print(f"DEBUG: software='{software}' has ' by ' in it")
+ # DEBUG: print(f"DEBUG: software='{software}' has ' by ' in it")
software = strip_until(software, " by ")
elif type(software) is str and " see " in software:
- print(f"DEBUG: software='{software}' has ' see ' in it")
+ # DEBUG: print(f"DEBUG: software='{software}' has ' see ' in it")
software = strip_until(software, " see ")
- print(f"DEBUG: software='{software}' - EXIT!")
+ # DEBUG: print(f"DEBUG: software='{software}' - EXIT!")
return software
def determine_software(domain: str, path: str = None) -> str:
- print(f"DEBUG: domain({len(domain)})={domain},path={path} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain({len(domain)})={domain},path={path} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
elif type(path) != str and path != None:
raise ValueError(f"Parameter path[]={type(path)} is not 'str'")
- print("DEBUG: Determining software for domain,path:", domain, path)
+ # DEBUG: print("DEBUG: Determining software for domain,path:", domain, path)
software = None
- print(f"DEBUG: Fetching nodeinfo from '{domain}' ...")
+ # DEBUG: print(f"DEBUG: Fetching nodeinfo from '{domain}' ...")
data = fetch_nodeinfo(domain, path)
- print("DEBUG: data[]:", type(data))
+ # DEBUG: print("DEBUG: data[]:", type(data))
if not isinstance(data, dict) or len(data) == 0:
- print("DEBUG: Could not determine software type:", domain)
+ # DEBUG: print("DEBUG: Could not determine software type:", domain)
return fetch_generator_from_path(domain)
- print("DEBUG: data():", len(data), data)
+ # DEBUG: print("DEBUG: data():", len(data), data)
if "status" in data and data["status"] == "error" and "message" in data:
print("WARNING: JSON response is an error:", data["message"])
update_last_error(domain, data["message"])
update_last_error(domain, data["message"])
return fetch_generator_from_path(domain)
elif "software" not in data or "name" not in data["software"]:
- print(f"DEBUG: JSON response from domain='{domain}' does not include [software][name], fetching / ...")
+ # DEBUG: print(f"DEBUG: JSON response from domain='{domain}' does not include [software][name], fetching / ...")
software = fetch_generator_from_path(domain)
- print(f"DEBUG: Generator for domain='{domain}' is: {software}, EXIT!")
+ # DEBUG: print(f"DEBUG: Generator for domain='{domain}' is: {software}, EXIT!")
return software
software = tidyup_domain(data["software"]["name"])
- print("DEBUG: sofware after tidyup_domain():", software)
+ # DEBUG: print("DEBUG: sofware after tidyup_domain():", software)
if software in ["akkoma", "rebased"]:
- print("DEBUG: Setting pleroma:", domain, software)
+ # DEBUG: print("DEBUG: Setting pleroma:", domain, software)
software = "pleroma"
elif software in ["hometown", "ecko"]:
- print("DEBUG: Setting mastodon:", domain, software)
+ # DEBUG: print("DEBUG: Setting mastodon:", domain, software)
software = "mastodon"
elif software in ["calckey", "groundpolis", "foundkey", "cherrypick", "meisskey"]:
- print("DEBUG: Setting misskey:", domain, software)
+ # DEBUG: print("DEBUG: Setting misskey:", domain, software)
software = "misskey"
elif software.find("/") > 0:
print("WARNING: Spliting of slash:", software)
print("WARNING: Spliting of pipe:", software)
software = tidyup_domain(software.split("|")[0]);
elif "powered by" in software:
- print(f"DEBUG: software='{software}' has 'powered by' in it")
+ # DEBUG: print(f"DEBUG: software='{software}' has 'powered by' in it")
software = strip_powered_by(software)
elif type(software) is str and " by " in software:
- print(f"DEBUG: software='{software}' has ' by ' in it")
+ # DEBUG: print(f"DEBUG: software='{software}' has ' by ' in it")
software = strip_until(software, " by ")
elif type(software) is str and " see " in software:
- print(f"DEBUG: software='{software}' has ' see ' in it")
+ # DEBUG: print(f"DEBUG: software='{software}' has ' see ' in it")
software = strip_until(software, " see ")
- print(f"DEBUG: software[]={type(software)}")
+ # DEBUG: print(f"DEBUG: software[]={type(software)}")
if software == "":
print("WARNING: tidyup_domain() left no software name behind:", domain)
software = None
- print(f"DEBUG: software[]={type(software)}")
+ # DEBUG: print(f"DEBUG: software[]={type(software)}")
if str(software) == "":
- print(f"DEBUG: software for '{domain}' was not detected, trying generator ...")
+ # DEBUG: print(f"DEBUG: software for '{domain}' was not detected, trying generator ...")
software = fetch_generator_from_path(domain)
elif len(str(software)) > 0 and ("." in software or " " in software):
- print(f"DEBUG: software='{software}' may contain a version number, domain='{domain}', removing it ...")
+ # DEBUG: print(f"DEBUG: software='{software}' may contain a version number, domain='{domain}', removing it ...")
software = remove_version(software)
- print(f"DEBUG: software[]={type(software)}")
+ # DEBUG: print(f"DEBUG: software[]={type(software)}")
if type(software) is str and "powered by" in software:
- print(f"DEBUG: software='{software}' has 'powered by' in it")
+ # DEBUG: print(f"DEBUG: software='{software}' has 'powered by' in it")
software = remove_version(strip_powered_by(software))
- print("DEBUG: Returning domain,software:", domain, software)
+ # DEBUG: print("DEBUG: Returning domain,software:", domain, software)
return software
def update_block_reason(reason: str, blocker: str, blocked: str, block_level: str):
- print(f"DEBUG: reason='{reason}',blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
+ # DEBUG: print(f"DEBUG: reason='{reason}',blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
if type(reason) != str and reason != None:
raise ValueError(f"Parameter reason[]='{type(reason)}' is not 'str'")
elif type(blocker) != str:
elif type(block_level) != str:
raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
- print("DEBUG: Updating block reason:", reason, blocker, blocked, block_level)
+ # DEBUG: print("DEBUG: Updating block reason:", reason, blocker, blocked, block_level)
try:
cursor.execute(
"UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND reason IN ('','unknown') LIMIT 1",
),
)
- print(f"DEBUG: cursor.rowcount={cursor.rowcount}")
+ # DEBUG: print(f"DEBUG: cursor.rowcount={cursor.rowcount}")
if cursor.rowcount == 0:
- print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',reason='{reason}' - EXIT!")
+ # DEBUG: print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',reason='{reason}' - EXIT!")
return
except BaseException as e:
print(f"ERROR: failed SQL query: reason='{reason}',blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',exception[{type(e)}]:'{str(e)}'")
sys.exit(255)
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def update_last_seen(blocker: str, blocked: str, block_level: str):
- print("DEBUG: Updating last_seen for:", blocker, blocked, block_level)
+ # DEBUG: print("DEBUG: Updating last_seen for:", blocker, blocked, block_level)
try:
cursor.execute(
"UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
)
)
- print(f"DEBUG: cursor.rowcount={cursor.rowcount}")
+ # DEBUG: print(f"DEBUG: cursor.rowcount={cursor.rowcount}")
if cursor.rowcount == 0:
- print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' - EXIT!")
+ # DEBUG: print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' - EXIT!")
return
except BaseException as e:
print(f"ERROR: failed SQL query: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',exception[{type(e)}]:'{str(e)}'")
sys.exit(255)
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def is_instance_blocked(blocker: str, blocked: str, block_level: str) -> bool:
- print(f"DEBUG: blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
+ # DEBUG: print(f"DEBUG: blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
if type(blocker) != str:
raise ValueError(f"Parameter blocker[]={type(blocker)} is not of type 'str'")
elif blocker == "":
is_blocked = cursor.fetchone() != None
- print(f"DEBUG: is_blocked='{is_blocked}' - EXIT!")
+ # DEBUG: print(f"DEBUG: is_blocked='{is_blocked}' - EXIT!")
return is_blocked
def block_instance(blocker: str, blocked: str, reason: str, block_level: str):
- print("DEBUG: blocker,blocked,reason,block_level:", blocker, blocked, reason, block_level)
+ # DEBUG: print("DEBUG: blocker,blocked,reason,block_level:", blocker, blocked, reason, block_level)
if type(blocker) != str:
raise ValueError(f"Parameter blocker[]={type(blocker)} is not 'str'")
elif blocker == "":
print(f"ERROR: failed SQL query: blocker='{blocker}',blocked='{blocked}',reason='{reason}',block_level='{block_level}',exception[{type(e)}]:'{str(e)}'")
sys.exit(255)
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def is_instance_registered(domain: str) -> bool:
- print(f"DEBUG: domain={domain} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain={domain} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print(f"DEBUG: domain='{domain}' - CALLED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
if not cache.key_exists("is_registered"):
- print(f"DEBUG: Cache for 'is_registered' not initialized, fetching all rows ...")
+ # DEBUG: print(f"DEBUG: Cache for 'is_registered' not initialized, fetching all rows ...")
try:
cursor.execute("SELECT domain FROM instances")
# Is cache found?
registered = cache.sub_key_exists("is_registered", domain)
- print(f"DEBUG: registered='{registered}' - EXIT!")
+ # DEBUG: print(f"DEBUG: registered='{registered}' - EXIT!")
return registered
def add_instance(domain: str, origin: str, originator: str, path: str = None):
- print(f"DEBUG: domain={domain},origin={origin},originator={originator},path={path} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain={domain},origin={origin},originator={originator},path={path} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
elif is_blacklisted(domain):
raise Exception(f"domain='{domain}' is blacklisted, but method invoked")
- print("DEBUG: domain,origin,originator,path:", domain, origin, originator, path)
+ # DEBUG: print("DEBUG: domain,origin,originator,path:", domain, origin, originator, path)
software = determine_software(domain, path)
- print("DEBUG: Determined software:", software)
+ # DEBUG: print("DEBUG: Determined software:", software)
print(f"INFO: Adding instance domain='{domain}' (origin='{origin}',software='{software}')")
try:
cache.set_sub_key("is_registered", domain, True)
if instances.has_pending_instance_data(domain):
- print(f"DEBUG: domain='{domain}' has pending nodeinfo being updated ...")
+ # DEBUG: print(f"DEBUG: domain='{domain}' has pending nodeinfo being updated ...")
instances.set("last_status_code" , domain, None)
instances.set("last_error_details", domain, None)
instances.update_instance_data(domain)
remove_pending_error(domain)
if domain in pending_errors:
- print("DEBUG: domain has pending error being updated:", domain)
+ # DEBUG: print("DEBUG: domain has pending error being updated:", domain)
update_last_error(domain, pending_errors[domain])
remove_pending_error(domain)
print(f"ERROR: failed SQL query: domain='{domain}',exception[{type(e)}]:'{str(e)}'")
sys.exit(255)
else:
- print("DEBUG: Updating nodeinfo for domain:", domain)
+ # DEBUG: print("DEBUG: Updating nodeinfo for domain:", domain)
update_last_nodeinfo(domain)
- print("DEBUG: EXIT!")
+ # DEBUG: print("DEBUG: EXIT!")
def send_bot_post(instance: str, blocks: dict):
- print(f"DEBUG: instance={instance},blocks()={len(blocks)} - CALLED!")
+ # DEBUG: print(f"DEBUG: instance={instance},blocks()={len(blocks)} - CALLED!")
message = instance + " has blocked the following instances:\n\n"
truncated = False
return True
def get_mastodon_blocks(domain: str) -> dict:
- print(f"DEBUG: domain={domain} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain={domain} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print("DEBUG: Fetching mastodon blocks from domain:", domain)
+ # DEBUG: print("DEBUG: Fetching mastodon blocks from domain:", domain)
blocks = {
"Suspended servers": [],
"Filtered media" : [],
header_text = tidyup_domain(header.text)
if header_text in language_mapping:
- print(f"DEBUG: header_text='{header_text}'")
+ # DEBUG: print(f"DEBUG: header_text='{header_text}'")
header_text = language_mapping[header_text]
if header_text in blocks or header_text.lower() in blocks:
}
)
- print("DEBUG: Returning blocks for domain:", domain)
+ # DEBUG: print("DEBUG: Returning blocks for domain:", domain)
return {
"reject" : blocks["Suspended servers"],
"media_removal" : blocks["Filtered media"],
}
def get_friendica_blocks(domain: str) -> dict:
- print(f"DEBUG: domain={domain} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain={domain} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print("DEBUG: Fetching friendica blocks from domain:", domain)
+ # DEBUG: print("DEBUG: Fetching friendica blocks from domain:", domain)
blocks = []
try:
# Prevents exceptions:
if blocklist is None:
- print("DEBUG: Instance has no block list:", domain)
+ # DEBUG: print("DEBUG: Instance has no block list:", domain)
return {}
for line in blocklist.find("table").find_all("tr")[1:]:
- print(f"DEBUG: line='{line}'")
+ # DEBUG: print(f"DEBUG: line='{line}'")
blocks.append({
"domain": tidyup_domain(line.find_all("td")[0].text),
"reason": tidyup_domain(line.find_all("td")[1].text)
})
- print("DEBUG: Returning blocks() for domain:", domain, len(blocks))
+ # DEBUG: print("DEBUG: Returning blocks() for domain:", domain, len(blocks))
return {
"reject": blocks
}
def get_misskey_blocks(domain: str) -> dict:
- print(f"DEBUG: domain={domain} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain={domain} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
raise ValueError(f"Parameter 'domain' cannot be empty")
- print("DEBUG: Fetching misskey blocks from domain:", domain)
+ # DEBUG: print("DEBUG: Fetching misskey blocks from domain:", domain)
blocks = {
"suspended": [],
"blocked" : []
# instances page-by-page, since that troonware doesn't support
# sending them all at once
try:
- print(f"DEBUG: Fetching offset='{offset}' from '{domain}' ...")
+ # DEBUG: print(f"DEBUG: Fetching offset='{offset}' from '{domain}' ...")
if offset == 0:
- print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
+ # DEBUG: print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
fetched = post_json_api(domain, "/api/federation/instances", json.dumps({
"sort" : "+pubAt",
"host" : None,
"Origin": domain
})
else:
- print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
+ # DEBUG: print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
fetched = post_json_api(domain, "/api/federation/instances", json.dumps({
"sort" : "+pubAt",
"host" : None,
"Origin": domain
})
- print("DEBUG: fetched():", len(fetched))
+ # DEBUG: print("DEBUG: fetched():", len(fetched))
if len(fetched) == 0:
- print("DEBUG: Returned zero bytes, exiting loop:", domain)
+ # DEBUG: print("DEBUG: Returned zero bytes, exiting loop:", domain)
break
elif len(fetched) != config.get("misskey_limit"):
- print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_limit')}'")
+ # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_limit')}'")
offset = offset + (config.get("misskey_limit") - len(fetched))
else:
- print("DEBUG: Raising offset by step:", step)
+ # DEBUG: print("DEBUG: Raising offset by step:", step)
offset = offset + step
for instance in fetched:
# same shit, different asshole ("blocked" aka full suspend)
try:
if offset == 0:
- print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
+ # DEBUG: print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
fetched = post_json_api(domain,"/api/federation/instances", json.dumps({
"sort" : "+pubAt",
"host" : None,
"Origin": domain
})
else:
- print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
+ # DEBUG: print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
fetched = post_json_api(domain,"/api/federation/instances", json.dumps({
"sort" : "+pubAt",
"host" : None,
"Origin": domain
})
- print("DEBUG: fetched():", len(fetched))
+ # DEBUG: print("DEBUG: fetched():", len(fetched))
if len(fetched) == 0:
- print("DEBUG: Returned zero bytes, exiting loop:", domain)
+ # DEBUG: print("DEBUG: Returned zero bytes, exiting loop:", domain)
break
elif len(fetched) != config.get("misskey_limit"):
- print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_limit')}'")
+ # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_limit')}'")
offset = offset + (config.get("misskey_limit") - len(fetched))
else:
- print("DEBUG: Raising offset by step:", step)
+ # DEBUG: print("DEBUG: Raising offset by step:", step)
offset = offset + step
for instance in fetched:
offset = 0
break
- print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
+ # DEBUG: print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
update_last_instance_fetch(domain)
- print("DEBUG: Returning for domain,blocked(),suspended():", domain, len(blocks["blocked"]), len(blocks["suspended"]))
+ # DEBUG: print("DEBUG: Returning for domain,blocked(),suspended():", domain, len(blocks["blocked"]), len(blocks["suspended"]))
return {
"reject" : blocks["blocked"],
"followers_only": blocks["suspended"]
}
def tidyup_reason(reason: str) -> str:
- print(f"DEBUG: reason='{reason}' - CALLED!")
+ # DEBUG: print(f"DEBUG: reason='{reason}' - CALLED!")
if type(reason) != str:
raise ValueError(f"Parameter reason[]={type(reason)} is not expected")
# Replace â with "
reason = re.sub("â", "\"", reason)
- #print(f"DEBUG: reason='{reason}' - EXIT!")
+ ## DEBUG: print(f"DEBUG: reason='{reason}' - EXIT!")
return reason
def tidyup_domain(domain: str) -> str:
- print(f"DEBUG: domain='{domain}' - CALLED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]={type(domain)} is not expected")
# No individual users in block lists
domain = re.sub("(.+)\@", "", domain)
- print(f"DEBUG: domain='{domain}' - EXIT!")
+ # DEBUG: print(f"DEBUG: domain='{domain}' - EXIT!")
return domain
def json_from_response(response: requests.models.Response) -> list:
- print(f"DEBUG: response[]={type(response)} - CALLED!")
+ # DEBUG: print(f"DEBUG: response[]={type(response)} - CALLED!")
if not isinstance(response, requests.models.Response):
raise ValueError(f"Parameter response[]='{type(response)}' is not type of 'Response'")
data = list()
if response.text.strip() != "":
- print(f"DEBUG: response.text()={len(response.text)} is not empty, invoking response.json() ...")
+ # DEBUG: print(f"DEBUG: response.text()={len(response.text)} is not empty, invoking response.json() ...")
try:
data = response.json()
except json.decoder.JSONDecodeError:
pass
- print(f"DEBUG: data[]={type(data)} - EXIT!")
+ # DEBUG: print(f"DEBUG: data[]={type(data)} - EXIT!")
return data
def get_response(domain: str, path: str, headers: dict, timeout: list) -> requests.models.Response:
- print(f"DEBUG: domain='{domain}',path='{path}',headers()={len(headers)},timeout={timeout} - CALLED!")
+ # DEBUG: print(f"DEBUG: domain='{domain}',path='{path}',headers()={len(headers)},timeout={timeout} - CALLED!")
if type(domain) != str:
raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
elif domain == "":
raise ValueError("Parameter 'path' cannot be empty")
try:
- print(f"DEBUG: Sending request to '{domain}{path}' ...")
+ # DEBUG: print(f"DEBUG: Sending request to '{domain}{path}' ...")
response = reqto.get(
f"https://{domain}{path}",
headers=headers,
timeout=timeout
);
except requests.exceptions.ConnectionError as e:
- print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(e)}]='{str(e)}'")
+ # DEBUG: print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(e)}]='{str(e)}'")
update_last_error(domain, e)
raise e
- print(f"DEBUG: response[]='{type(response)}' - EXXIT!")
+ # DEBUG: print(f"DEBUG: response[]='{type(response)}' - EXXIT!")
return response
def has_element(elements: list, key: str, value: any) -> bool:
- print(f"DEBUG: element()={len(element)},key='{key}',value[]='{type(value)}' - CALLED!")
+ # DEBUG: print(f"DEBUG: element()={len(element)},key='{key}',value[]='{type(value)}' - CALLED!")
if type(key) != str:
raise ValueError(f"Parameter key[]='{type(key)}' is not 'str'")
elif key == "":
raise ValueError("Parameter 'key' cannot be empty")
has = False
- print(f"DEBUG: Checking elements()={len(elements)} ...")
+ # DEBUG: print(f"DEBUG: Checking elements()={len(elements)} ...")
for element in elements:
- print(f"DEBUG: element[]='{type(element)}'")
+ # DEBUG: print(f"DEBUG: element[]='{type(element)}'")
if type(element) != dict:
raise ValueError(f"element[]='{type(element)}' is not 'dict'")
elif not key in element:
has = True
break
- print(f"DEBUG: has={has} - EXIT!")
+ # DEBUG: print(f"DEBUG: has={has} - EXIT!")
return has
def find_domains(tag: bs4.element.Tag) -> list:
- print(f"DEBUG: tag[]={type(tag)} - CALLED!")
+ # DEBUG: print(f"DEBUG: tag[]={type(tag)} - CALLED!")
if not isinstance(tag, bs4.element.Tag):
raise ValueError(f"Parameter tag[]={type(tag)} is not type of bs4.element.Tag")
elif not isinstance(tag, bs4.element.Tag):
domains = list()
for element in tag.select("tr"):
- print(f"DEBUG: element[]={type(element)}")
+ # DEBUG: print(f"DEBUG: element[]={type(element)}")
if not element.find("td"):
- print("DEBUG: Skipping element, no <td> found")
+ # DEBUG: print("DEBUG: Skipping element, no <td> found")
continue
domain = tidyup_domain(element.find("td").text)
reason = tidyup_reason(element.findAll("td")[1].text)
- print(f"DEBUG: domain='{domain}',reason='{reason}'")
+ # DEBUG: print(f"DEBUG: domain='{domain}',reason='{reason}'")
if is_blacklisted(domain):
print(f"WARNING: domain='{domain}' is blacklisted - skipped!")
continue
elif domain == "gab.com/.ai, develop.gab.com":
- print(f"DEBUG: Multiple domains detected in one row")
+ # DEBUG: print(f"DEBUG: Multiple domains detected in one row")
domains.append({
"domain": "gab.com",
"reason": reason,
print(f"WARNING: domain='{domain}' is not a valid domain - skipped!")
continue
- print(f"DEBUG: Adding domain='{domain}' ...")
+ # DEBUG: print(f"DEBUG: Adding domain='{domain}' ...")
domains.append({
"domain": domain,
"reason": reason,
})
- print(f"DEBUG: domains()={len(domains)} - EXIT!")
+ # DEBUG: print(f"DEBUG: domains()={len(domains)} - EXIT!")
return domains
def get_url(url: str) -> requests.models.Response:
- print(f"DEBUG: url='{url}' - CALLED!")
+ # DEBUG: print(f"DEBUG: url='{url}' - CALLED!")
if type(url) != str:
raise ValueError(f"Parameter url[]='{type(url)}' is not 'str'")
elif url == "":
raise ValueError("Parameter 'url' cannot be empty")
- print(f"DEBUG: Parsing url='{url}'")
+ # DEBUG: print(f"DEBUG: Parsing url='{url}'")
components = urllib.parse(url)
# Invoke other function, avoid trailing ?
else:
response = get_response(components.hostname, f"{components.path}")
- print(f"DEBUG: response[]='{type(response)}' - EXXIT!")
+ # DEBUG: print(f"DEBUG: response[]='{type(response)}' - EXXIT!")
return response
projects / fba.git / commitdiff